diff --git a/docs/users/features/channels/overview.md b/docs/users/features/channels/overview.md index 56a03a50f98..ab9d0a9abbc 100644 --- a/docs/users/features/channels/overview.md +++ b/docs/users/features/channels/overview.md @@ -62,6 +62,7 @@ Channels are configured under the `channels` key in `settings.json`. Each channe | `senderPolicy` | No | Who can talk to the bot: `allowlist` (default), `open`, or `pairing` | | `allowedUsers` | No | List of user IDs allowed to use the bot (used by `allowlist` and `pairing` policies) | | `sessionScope` | No | How sessions are scoped: `user` (default), `thread`, or `single` | +| `sessionRotation` | No | Bounds after which a route starts a fresh session: `{ "maxTurns": N, "maxAgeHours": N }`. Unset means a session is reused forever. See [Session rotation](#session-rotation) | | `cwd` | No | Working directory for the agent. Defaults to the current directory | | `approvalMode` | No | Tool approval mode for channel sessions. Unattended webhook tasks require `yolo`; the setting applies to every session on the channel | | `instructions` | No | Custom instructions prepended to the first message of each session | @@ -91,6 +92,32 @@ Controls how conversation sessions are managed: - **`thread`** — One session per thread/topic. Useful for group chats with threads. - **`single`** — One shared session for all users. Everyone shares the same conversation. +### Session Rotation + +By default a route keeps the same session forever, so a long-lived route — a busy group thread, a `single`-scope channel — accumulates context without bound. Once it grows past the model's context window every later message on that route fails, while the rest of the channel keeps working. `sessionRotation` puts a ceiling on that: when the current session is past its bound, the next message starts a fresh one instead. + +```json +{ + "channels": { + "my-bot": { + "type": "dingtalk", + "sessionScope": "thread", + "sessionRotation": { + "maxTurns": 200, + "maxAgeHours": 24 + } + } + } +} +``` + +- **`maxTurns`** — Rotate once this many messages have started a turn on the current session. Messages that settle without one (a `!` shell command, a dropped loop firing) do not count. +- **`maxAgeHours`** — Rotate once the current session is older than this. + +Set either, both, or neither; whichever bound is hit first rotates. `maxTurns` must be a positive integer and `maxAgeHours` a positive number. Omitting `sessionRotation` keeps the previous behavior of never rotating. In `collect` dispatch mode, messages buffered while a turn runs are coalesced into one turn and count once against `maxTurns`. + +Rotation is a context reset, not a cleanup: the new session starts empty, so the bot no longer remembers the earlier conversation on that route. The channel posts a short notice in the chat or thread whose message triggered the rotation, and the daemon logs the rotated route. With `sessionScope: single`, only the chat whose message triggered the rotation is notified; other chats sharing the session see the reset without a notice. Counters are stored alongside the routes and survive a daemon restart. Sessions that were already routed before you enabled rotation start their clock at the first message after the upgrade. A route that still has a turn running or queued rotates on the next message after it settles instead of mid-turn; under sustained traffic, where every message arrives while a turn is still running or queued, the bound waits for the first pause in traffic. Each message routed during that window extends it, so a continuously saturated route — a webhook receiving events faster than turns complete, or an overrunning loop — rotates only once traffic stops. + ### Channel Memory Channel memory stores durable context for one chat or thread. Entries have stable diff --git a/packages/channels/base/src/AcpBridge.test.ts b/packages/channels/base/src/AcpBridge.test.ts index 4387a551a84..21db0269078 100644 --- a/packages/channels/base/src/AcpBridge.test.ts +++ b/packages/channels/base/src/AcpBridge.test.ts @@ -113,6 +113,7 @@ type TestableAcpBridge = AcpBridge & { }; knownSessionIds: Set; sessionBindingTokens: Map; + pendingSessionRequests: Set<{ reject: (error: Error) => void }>; channelLoopMcpServer: unknown; channelLoopToolHandlers: ChannelLoopToolHandler[]; channelLoopMcpRegistered: boolean; @@ -845,6 +846,87 @@ describe('AcpBridge', () => { expect(bridge.resolveChannelLoopToolHandler('s-1')).toBe(handler); }); + it('rejects in-flight session requests when the ACP child exits', async () => { + const bridge = new AcpBridge({ + cliEntryPath: '/tmp/qwen', + cwd: '/tmp', + }); + + await bridge.start(); + const proc = child.instances[0]!; + const connection = child.connections[0] as unknown as { + newSession: ReturnType; + loadSession: ReturnType; + }; + // The SDK never settles requests still awaiting a response once the + // stream ends, so neither call can finish on its own. + connection.newSession = vi.fn(() => new Promise(() => {})); + connection.loadSession = vi.fn(() => new Promise(() => {})); + + const pendingNew = bridge.newSession('/tmp'); + const pendingLoad = bridge.loadSession('s-1', '/tmp'); + + proc.emit('exit', 1, null); + + const reason = + 'ACP agent process exited while a session request was in flight'; + await expect(pendingNew).rejects.toThrow(reason); + await expect(pendingLoad).rejects.toThrow(reason); + expect( + (bridge as unknown as TestableAcpBridge).pendingSessionRequests.size, + ).toBe(0); + }); + + it('rejects in-flight session requests when the bridge stops', async () => { + const bridge = new AcpBridge({ + cliEntryPath: '/tmp/qwen', + cwd: '/tmp', + }); + + await bridge.start(); + const connection = child.connections[0] as unknown as { + newSession: ReturnType; + }; + connection.newSession = vi.fn(() => new Promise(() => {})); + + const pending = bridge.newSession('/tmp'); + bridge.stop(); + + await expect(pending).rejects.toThrow( + 'ACP agent process exited while a session request was in flight', + ); + }); + + it('drops a settled session request from the pending set', async () => { + const bridge = new AcpBridge({ + cliEntryPath: '/tmp/qwen', + cwd: '/tmp', + }) as unknown as TestableAcpBridge; + bridge.child = { killed: false, exitCode: null }; + bridge.connection = { + extMethod: vi.fn(), + newSession: vi.fn().mockResolvedValue({ sessionId: 's-1' }), + }; + + await expect(bridge.newSession('/tmp')).resolves.toBe('s-1'); + expect(bridge.pendingSessionRequests.size).toBe(0); + }); + + it('drops a failed session request from the pending set', async () => { + const bridge = new AcpBridge({ + cliEntryPath: '/tmp/qwen', + cwd: '/tmp', + }) as unknown as TestableAcpBridge; + bridge.child = { killed: false, exitCode: null }; + bridge.connection = { + extMethod: vi.fn(), + newSession: vi.fn().mockRejectedValue(new Error('spawn failed')), + }; + + await expect(bridge.newSession('/tmp')).rejects.toThrow('spawn failed'); + expect(bridge.pendingSessionRequests.size).toBe(0); + }); + it('kills the ACP child when it reports a large event loop stall', async () => { const bridge = new AcpBridge({ cliEntryPath: '/tmp/qwen', diff --git a/packages/channels/base/src/AcpBridge.ts b/packages/channels/base/src/AcpBridge.ts index 3eccc018f20..69a82f12228 100644 --- a/packages/channels/base/src/AcpBridge.ts +++ b/packages/channels/base/src/AcpBridge.ts @@ -98,6 +98,9 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge { timeout: ReturnType; } >(); + private readonly pendingSessionRequests = new Set<{ + reject: (error: Error) => void; + }>(); constructor(options: AcpBridgeOptions) { super(); @@ -149,6 +152,7 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge { ); // Do not emit sessionDied here: a full ACP process exit is handled by // channel start crash recovery, which reloads the persisted sessions. + this.rejectPendingSessionRequests(); this.resolvePendingPermissions(); this.knownSessionIds.clear(); this.sessionBindingTokens.clear(); @@ -232,11 +236,14 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge { bindingToken?: object, ): Promise { const conn = this.ensureConnection(); - await this.registerChannelLoopMcpServer(); - const response = await conn.newSession({ cwd, mcpServers: [] }); - this.knownSessionIds.add(response.sessionId); - this.sessionBindingTokens.set(response.sessionId, bindingToken); - return response.sessionId; + const sessionId = await this.settleOnChildExit(async () => { + await this.registerChannelLoopMcpServer(); + const response = await conn.newSession({ cwd, mcpServers: [] }); + return response.sessionId; + }); + this.knownSessionIds.add(sessionId); + this.sessionBindingTokens.set(sessionId, bindingToken); + return sessionId; } async loadSession( @@ -246,11 +253,13 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge { bindingToken?: object, ): Promise { const conn = this.ensureConnection(); - await this.registerChannelLoopMcpServer(); - await conn.loadSession({ - sessionId, - cwd, - mcpServers: [], + await this.settleOnChildExit(async () => { + await this.registerChannelLoopMcpServer(); + await conn.loadSession({ + sessionId, + cwd, + mcpServers: [], + }); }); this.knownSessionIds.add(sessionId); this.sessionBindingTokens.set(sessionId, bindingToken); @@ -361,6 +370,7 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge { } stop(): void { + this.rejectPendingSessionRequests(); this.resolvePendingPermissions(); this.knownSessionIds.clear(); this.sessionBindingTokens.clear(); @@ -462,6 +472,40 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge { return this.connection; } + /** + * The ACP SDK never settles requests still awaiting a response once the + * stream ends, so a child death mid-request would hang the caller forever; + * a restore's persist suspension, in particular, would never lift. Reject + * those requests when the child exits instead. + */ + private settleOnChildExit(run: () => Promise): Promise { + return new Promise((resolve, reject) => { + const pending = { reject }; + this.pendingSessionRequests.add(pending); + run().then( + (result) => { + this.pendingSessionRequests.delete(pending); + resolve(result); + }, + (error: unknown) => { + this.pendingSessionRequests.delete(pending); + reject(error); + }, + ); + }); + } + + private rejectPendingSessionRequests(): void { + for (const pending of this.pendingSessionRequests) { + pending.reject( + new Error( + 'ACP agent process exited while a session request was in flight', + ), + ); + } + this.pendingSessionRequests.clear(); + } + private requestPermission( request: RequestPermissionRequest, ): Promise { diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index e2c3e829266..cb3c857768d 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -2090,6 +2090,9 @@ describe('ChannelBase', () => { chatId: 'chat1', })), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router } as unknown as ChannelBaseOptions); @@ -8618,6 +8621,9 @@ describe('ChannelBase', () => { getTarget: vi.fn().mockReturnValue({ chatId: 'chat1' }), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router, @@ -8651,6 +8657,9 @@ describe('ChannelBase', () => { getTarget: vi.fn().mockReturnValue(target), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router, @@ -8684,6 +8693,9 @@ describe('ChannelBase', () => { getTarget: vi.fn().mockReturnValue(target), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router, @@ -8716,6 +8728,9 @@ describe('ChannelBase', () => { getTarget: vi.fn().mockReturnValue(target), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router, @@ -8760,6 +8775,9 @@ describe('ChannelBase', () => { getTarget: vi.fn().mockReturnValue(target), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router, @@ -8804,6 +8822,9 @@ describe('ChannelBase', () => { getTarget: vi.fn(), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router } as unknown as ChannelBaseOptions); @@ -8827,6 +8848,9 @@ describe('ChannelBase', () => { getTarget: vi.fn(), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router } as unknown as ChannelBaseOptions); const newBridge = createBridge(); @@ -8868,6 +8892,9 @@ describe('ChannelBase', () => { getTarget: vi.fn().mockReturnValue({ chatId: 'chat1' }), handleSessionDied: vi.fn(), setBridge: vi.fn(), + setChannelRotation: vi.fn(), + setSessionActivityChecker: vi.fn(), + onSessionRotated: vi.fn().mockReturnValue(() => {}), }; const ch = createChannel({}, { router, @@ -12342,6 +12369,560 @@ describe('ChannelBase', () => { }); }); + describe('session rotation', () => { + it('registers rotation bounds on a supplied router', () => { + const router = new SessionRouter(bridge, '/tmp'); + const spy = vi.spyOn(router, 'setChannelRotation'); + + createChannel({ sessionRotation: { maxTurns: 100 } }, { router }); + + expect(spy).toHaveBeenCalledWith('test-chan', { maxTurns: 100 }); + }); + + it('registers rotation bounds on a self-created router', () => { + const spy = vi.spyOn(SessionRouter.prototype, 'setChannelRotation'); + + createChannel({ sessionRotation: { maxAgeHours: 24 } }); + + expect(spy).toHaveBeenCalledWith('test-chan', { maxAgeHours: 24 }); + spy.mockRestore(); + }); + + it('announces rotation and discards the retired session', async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 1 } }); + await ch.handleInbound(envelope({ text: 'first' })); + const retiredId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + ch.sent = []; + + await ch.handleInbound(envelope({ text: 'second' })); + + const secondPrompt = (bridge.prompt as ReturnType).mock + .calls[1]!; + expect(secondPrompt[0]).not.toBe(retiredId); + expect(ch.sent.some((m) => m.text.includes('rotated'))).toBe(true); + expect(bridge.discardSession).toHaveBeenCalledWith(retiredId); + }); + + it("announces rotation in the rotated route's thread", async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 1 } }); + const threadMessages: Array<{ + chatId: string; + threadId: string | undefined; + text: string; + }> = []; + vi.spyOn(ch as never, 'sendThreadMessage').mockImplementation( + async (chatId: string, threadId: string | undefined, text: string) => { + threadMessages.push({ chatId, threadId, text }); + }, + ); + + await ch.handleInbound(envelope({ text: 'first', threadId: 'topic-1' })); + await ch.handleInbound(envelope({ text: 'second', threadId: 'topic-1' })); + + expect( + threadMessages.some( + (m) => m.threadId === 'topic-1' && m.text.includes('rotated'), + ), + ).toBe(true); + }); + + it('purges pending permissions of the rotated session', async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 1 } }); + await ch.handleInbound(envelope({ text: 'first' })); + const retiredId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + (bridge as unknown as EventEmitter).emit('permissionRequest', { + requestId: 'req-rotated', + sessionId: retiredId, + request: { + toolCall: { + toolCallId: 'tool-req-rotated', + kind: 'shell', + title: 'Run req-rotated', + }, + options: [ + { + optionId: 'proceed_once', + kind: 'allow_once', + name: 'Allow once', + }, + ], + }, + }); + await vi.waitFor(() => + expect( + ch.sent.some((m) => m.text.includes('Permission required')), + ).toBe(true), + ); + + // Rotate: the retirement purges the stale permission, so a late + // approval must not reach the bridge against the discarded session. + await ch.handleInbound(envelope({ text: 'second' })); + await ch.handleInbound(envelope({ text: '/approve req-rotated' })); + + expect( + (bridge as unknown as { respondToPermission: ReturnType }) + .respondToPermission, + ).not.toHaveBeenCalled(); + expect(ch.sent.at(-1)?.text).toBe( + 'No pending permission request with that id for this chat.', + ); + }); + + it('defers rotation while the outgoing turn is still running', async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 1 } }); + let settleFirst!: (value: string) => void; + (bridge.prompt as ReturnType).mockReturnValueOnce( + new Promise((resolve) => { + settleFirst = resolve; + }), + ); + + const firstTurn = ch.handleInbound(envelope({ text: 'first' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // Rotation is due, but the session is mid-turn: the second message + // queues behind it instead of retiring it. + const secondTurn = ch.handleInbound(envelope({ text: 'second' })); + // Let the second message resolve and queue behind the running turn; + // a rotation would have created a second session by now. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.newSession).toHaveBeenCalledTimes(1); + expect(bridge.discardSession).not.toHaveBeenCalled(); + + settleFirst('done'); + await firstTurn; + await secondTurn; + + // The deferred message reused the outgoing session ... + expect( + (bridge.prompt as ReturnType).mock.calls[1]![0], + ).toBe(sessionId); + // ... and the next message rotates it. + await ch.handleInbound(envelope({ text: 'third' })); + expect(bridge.discardSession).toHaveBeenCalledWith(sessionId); + expect( + (bridge.prompt as ReturnType).mock.calls[2]![0], + ).not.toBe(sessionId); + }); + + it('does not rotate a session out from under a message between resolve and its turn', async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 1 } }); + await ch.handleInbound(envelope({ text: 'first' })); + const retiredId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + let releaseShell!: (result: { output: string; exitCode: number }) => void; + const shellCommand = vi.fn().mockReturnValue( + new Promise<{ output: string; exitCode: number }>((resolve) => { + releaseShell = resolve; + }), + ); + (bridge as unknown as Record)['shellCommand'] = + shellCommand; + + // The bang message resolves the session (rotating the spent one) and + // stays mid-shell-command: it holds the session without a tracked turn. + const bangTurn = ch.handleInbound(envelope({ text: '!hang' })); + await vi.waitFor(() => expect(shellCommand).toHaveBeenCalledTimes(1)); + const shelledSessionId = shellCommand.mock.calls[0]![0] as string; + expect(shelledSessionId).not.toBe(retiredId); + + // A message at the bound must not retire the shelled session. + await ch.handleInbound(envelope({ text: 'second' })); + expect( + (bridge.prompt as ReturnType).mock.calls[1]![0], + ).toBe(shelledSessionId); + expect(bridge.discardSession).not.toHaveBeenCalledWith(shelledSessionId); + + releaseShell({ output: 'ok', exitCode: 0 }); + await bangTurn; + + // The lease must be released once the shell settles: at the bound the + // next message must rotate the shelled session. + await ch.handleInbound(envelope({ text: 'after-shell' })); + expect(bridge.discardSession).toHaveBeenCalledWith(shelledSessionId); + }); + + it('reclaims the queue and generation of a rotated session', async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 1 } }); + await ch.handleInbound(envelope({ text: 'first' })); + const retiredId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + const maps = ch as unknown as { + sessionQueues: Map; + sessionGenerations: Map; + }; + expect(maps.sessionQueues.has(retiredId)).toBe(true); + maps.sessionGenerations.set(retiredId, 1); + + await ch.handleInbound(envelope({ text: 'second' })); + + // Rotation retires the ID permanently: without reclamation both + // entries leak for the gateway's lifetime. + expect(maps.sessionQueues.has(retiredId)).toBe(false); + expect(maps.sessionGenerations.has(retiredId)).toBe(false); + }); + + it('does not count shell commands against maxTurns', async () => { + const ch = createChannel({ sessionRotation: { maxTurns: 2 } }); + const shellCommand = vi.fn().mockResolvedValue({ + output: 'ok', + exitCode: 0, + }); + (bridge as unknown as Record)['shellCommand'] = + shellCommand; + + await ch.handleInbound(envelope({ text: 'first' })); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // The shell command routes but starts no turn: it must give its + // resolve-time count back like the buffered and loop-drop paths. + await ch.handleInbound(envelope({ text: '!status' })); + expect(shellCommand).toHaveBeenCalledWith(sessionId, 'status'); + + await ch.handleInbound(envelope({ text: 'second' })); + expect( + (bridge.prompt as ReturnType).mock.calls[1]![0], + ).toBe(sessionId); + + await ch.handleInbound(envelope({ text: 'third' })); + expect( + (bridge.prompt as ReturnType).mock.calls[2]![0], + ).not.toBe(sessionId); + }); + + it('collect: buffered messages count once against maxTurns', async () => { + let settleFirst!: (value: string) => void; + let callCount = 0; + (bridge.prompt as ReturnType).mockImplementation(() => { + callCount++; + if (callCount === 1) { + return new Promise((resolve) => { + settleFirst = resolve; + }); + } + return Promise.resolve('response'); + }); + + const ch = createChannel({ + groupPolicy: 'open', + groups: { '*': { dispatchMode: 'collect' } }, + sessionRotation: { maxTurns: 3 }, + }); + + const groupMsg = (text: string): Envelope => + envelope({ + isGroup: true, + isMentioned: true, + chatId: 'g1', + text, + }); + + const first = ch.handleInbound(groupMsg('first')); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // These buffer behind the running turn and must not consume the bound: + // the drain coalesces them into a single counted turn. + await ch.handleInbound(groupMsg('second')); + await ch.handleInbound(groupMsg('third')); + + settleFirst('done'); + await first; + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(2)); + + // The session has carried two turns (first + coalesced), not four: + // the third real turn still reuses it, and only the next rotates. + await ch.handleInbound(groupMsg('fourth')); + expect( + (bridge.prompt as ReturnType).mock.calls[2]![0], + ).toBe(sessionId); + await ch.handleInbound(groupMsg('fifth')); + expect( + (bridge.prompt as ReturnType).mock.calls[3]![0], + ).not.toBe(sessionId); + }); + + it('rotation on one channel does not notify another sharing the router', async () => { + const router = new SessionRouter(bridge, '/tmp'); + const channelA = new TestChannel( + 'chan-a', + defaultConfig({ sessionRotation: { maxTurns: 1 } }), + bridge, + { router }, + ); + const channelB = new TestChannel('chan-b', defaultConfig(), bridge, { + router, + }); + + await channelA.handleInbound(envelope({ text: 'first' })); + channelA.sent = []; + await channelA.handleInbound(envelope({ text: 'second' })); + + expect(channelA.sent.some((m) => m.text.includes('rotated'))).toBe(true); + expect(channelB.sent).toEqual([]); + }); + + it('keeps the queue of a dead session so a lazy revival cannot split the chain', async () => { + const router = new SessionRouter(bridge, '/tmp', 'user', undefined, { + recoveryMode: 'lazy', + }); + (bridge.loadSession as ReturnType).mockImplementation( + (id: string) => id, + ); + const ch = createChannel({ dispatchMode: 'followup' }, { router }); + + let inFlight = 0; + let maxInFlight = 0; + let firstCall = true; + let settleFirst!: (value: string) => void; + (bridge.prompt as ReturnType).mockImplementation(() => { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + if (firstCall) { + firstCall = false; + return new Promise((resolve) => { + settleFirst = (value: string) => { + inFlight--; + resolve(value); + }; + }); + } + inFlight--; + return Promise.resolve('done'); + }); + + const firstTurn = ch.handleInbound(envelope({ text: 'first' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // A second message queues behind the running turn. + const secondTurn = ch.handleInbound(envelope({ text: 'second' })); + await new Promise((resolve) => setTimeout(resolve, 20)); + + // Stream-drop death: lazy recovery keeps the route, and the queue must + // survive too — the queued turn still holds the captured chain. + ch.onSessionDied(sessionId); + expect([ + ...( + ch as unknown as { sessionQueues: Map } + ).sessionQueues.keys(), + ]).toContain(sessionId); + + // The next message revives the same session ID; it must chain behind + // the stale queued turn instead of seeding a concurrent chain. + const thirdTurn = ch.handleInbound(envelope({ text: 'third' })); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + + settleFirst('done'); + await firstTurn; + await secondTurn; + await thirdTurn; + + expect(bridge.prompt).toHaveBeenCalledTimes(3); + expect(maxInFlight).toBe(1); + expect( + (bridge.prompt as ReturnType).mock.calls.map( + (call) => call[0], + ), + ).toEqual([sessionId, sessionId, sessionId]); + }); + + it('defers rotation while a loop turn is still running', async () => { + const ch = createChannel({ + sessionRotation: { maxTurns: 1 }, + dispatchMode: 'followup', + }); + ch.proactiveSupported = true; + let settleLoop!: (value: string) => void; + (bridge.prompt as ReturnType).mockReturnValueOnce( + new Promise((resolve) => { + settleLoop = resolve; + }), + ); + + const job: ChannelLoop = { + id: 'loop-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'user1', + chatId: 'chat1', + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'summary', + recurring: true, + enabled: true, + createdBy: 'User 1', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }; + + const loopRun = ch.runLoopPrompt(job); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // An inbound message on the loop's route is at the bound but must + // queue behind the running loop turn instead of retiring the session. + const inbound = ch.handleInbound(envelope({ text: 'while looping' })); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.newSession).toHaveBeenCalledTimes(1); + expect(bridge.discardSession).not.toHaveBeenCalled(); + + settleLoop('done'); + await loopRun; + await inbound; + + // The deferred message reused the loop's session ... + expect( + (bridge.prompt as ReturnType).mock.calls[1]![0], + ).toBe(sessionId); + // ... and the next message rotates it. + await ch.handleInbound(envelope({ text: 'after' })); + expect(bridge.discardSession).toHaveBeenCalledWith(sessionId); + }); + + it('does not consume the bound when a queued loop firing is dropped', async () => { + const ch = createChannel({ + sessionRotation: { maxTurns: 2 }, + dispatchMode: 'followup', + }); + ch.proactiveSupported = true; + let settleFirst!: (value: string) => void; + (bridge.prompt as ReturnType).mockReturnValueOnce( + new Promise((resolve) => { + settleFirst = resolve; + }), + ); + + const firstTurn = ch.handleInbound(envelope({ text: 'first' })); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // The firing queues behind the busy session; the loop is disabled + // while it waits, so the in-closure recheck drops it before any + // prompt runs on the session. + const job: ChannelLoop = { + id: 'loop-1', + channelName: 'test-chan', + target: { + channelName: 'test-chan', + senderId: 'user1', + chatId: 'chat1', + }, + cwd: '/tmp', + cron: '0 9 * * *', + prompt: 'post summary', + label: 'summary', + recurring: true, + enabled: true, + createdBy: 'User 1', + createdAt: '2026-06-30T01:00:00.000Z', + consecutiveFailures: 0, + runCount: 0, + }; + const loopRun = ch.runLoopPrompt(job, { + shouldContinue: async () => false, + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.newSession).toHaveBeenCalledTimes(1); + + settleFirst('done'); + await firstTurn; + await expect(loopRun).rejects.toThrow(/no longer enabled/); + + // The skipped firing gave its resolve-time count back: with maxTurns 2 + // the next real message still reuses the session, and only the one + // after it rotates. + await ch.handleInbound(envelope({ text: 'second' })); + expect( + (bridge.prompt as ReturnType).mock.calls[1]![0], + ).toBe(sessionId); + await ch.handleInbound(envelope({ text: 'third' })); + expect( + (bridge.prompt as ReturnType).mock.calls[2]![0], + ).not.toBe(sessionId); + expect(bridge.discardSession).toHaveBeenCalledWith(sessionId); + }); + + it('defers rotation while a webhook turn is still running', async () => { + const webhooks: ChannelWebhookConfig = { + sources: { + 'github-ci': { + targets: { + default: { + chatId: 'chat1', + senderId: 'user1', + }, + }, + }, + }, + }; + const ch = createChannel({ + approvalMode: 'yolo', + webhooks, + dispatchMode: 'followup', + sessionRotation: { maxTurns: 1 }, + }); + ch.proactiveSupported = true; + + let settleWebhook!: (value: string) => void; + (bridge.prompt as ReturnType).mockReturnValueOnce( + new Promise((resolve) => { + settleWebhook = resolve; + }), + ); + + const task: ChannelWebhookTask = { + channelName: 'test-chan', + source: 'github-ci', + eventType: 'ci_failed', + targetRef: 'default', + title: 'CI failed', + payload: { branch: 'main' }, + }; + + const webhookRun = ch.runWebhookTask(task); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(1)); + const sessionId = (bridge.prompt as ReturnType).mock + .calls[0]![0] as string; + + // An inbound message on the webhook's route is at the bound but must + // queue behind the running webhook turn instead of retiring the + // session out from under it. + const inbound = ch.handleInbound(envelope({ text: 'while webhook' })); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.newSession).toHaveBeenCalledTimes(1); + expect(bridge.discardSession).not.toHaveBeenCalled(); + + settleWebhook('done'); + await webhookRun; + await inbound; + + // The deferred message reused the webhook's session ... + expect( + (bridge.prompt as ReturnType).mock.calls[1]![0], + ).toBe(sessionId); + // ... and the next message rotates it. + await ch.handleInbound(envelope({ text: 'after' })); + expect(bridge.discardSession).toHaveBeenCalledWith(sessionId); + }); + }); + describe('session routing', () => { it('creates new session on first message', async () => { const ch = createChannel(); diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index c47dc5b0a75..af9db08dfa7 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -404,6 +404,8 @@ export abstract class ChannelBase { private commands: Map = new Map(); /** Per-session promise chain to serialize prompt + send (followup mode). */ private sessionQueues: Map> = new Map(); + /** Sessions with a turn running or queued; rotation defers while non-zero. */ + private sessionPendingTurns = new Map(); private readonly registerBridgeEvents: boolean; private readonly bridgeRecovery?: () => Promise | undefined; /** @@ -854,6 +856,16 @@ export abstract class ChannelBase { this.router = options?.router || new SessionRouter(bridge, config.cwd, config.sessionScope); + // Registration is idempotent and name-keyed, so the channel owning the + // config is the single owner of this invariant — gateway callers pass the + // same parsed config and need not mirror it. + this.router.setChannelRotation(this.name, config.sessionRotation); + this.router.setSessionActivityChecker(this.name, (sessionId) => + this.hasPendingTurns(sessionId), + ); + this.router.onSessionRotated((sessionId, target) => { + this.handleSessionRotated(sessionId, target); + }); this.registerSharedCommands(); if (this.loopController) { @@ -1522,6 +1534,9 @@ export abstract class ChannelBase { ); } if (options.shouldContinue && !(await options.shouldContinue())) { + // The firing was routed and counted but never prompted: give the + // count back so a dropped firing cannot consume the session's bound. + this.router.uncountTurn(this.name, sessionId); throw new ChannelLoopSkippedError( 'loop dropped because it is no longer enabled', ); @@ -1768,6 +1783,7 @@ export abstract class ChannelBase { sessionId, current.then(() => undefined).catch(() => {}), ); + this.trackSessionTurn(sessionId, current); return current; } @@ -2046,6 +2062,7 @@ export abstract class ChannelBase { sessionId, current.then(() => undefined).catch(() => undefined), ); + this.trackSessionTurn(sessionId, current); return await current; } @@ -2288,11 +2305,71 @@ export abstract class ChannelBase { onSessionDied(sessionId: string): void { this.router.handleSessionDied(sessionId); + this.purgeSessionState(sessionId); + } + + private purgeSessionState(sessionId: string): void { this.instructedSessions.delete(sessionId); this.unattendedMemorySessions.delete(sessionId); + // sessionQueues is deliberately NOT purged: a queued turn may still hold + // the captured chain, and deleting the entry would let the next message + // (which lazy recovery can re-attach to this same session ID) seed a + // fresh chain and run concurrently with the stale queued turn. /clear is + // the only path that may delete it, after the chain drains. this.removePendingPermissionsForSession(sessionId); } + /** + * The router retired a session by rotation: purge the per-session state + * a death would clean up, and tell the chat its context is starting fresh — + * rotation is automatic, so participants get no other signal. + */ + private handleSessionRotated( + sessionId: string, + target: SessionTarget | undefined, + ): void { + if (target?.channelName !== this.name) return; + this.purgeSessionState(sessionId); + // Rotation retires the ID permanently and defers until no turn is + // running or queued, so it reclaims what the death path must keep: a + // dead ID can be re-attached by lazy recovery with a queued turn still + // holding the chain, a rotated one cannot. + this.sessionQueues.delete(sessionId); + this.sessionGenerations.delete(sessionId); + void this.sendThreadMessage( + target.chatId, + target.threadId, + 'This conversation reached its configured limit and was rotated; starting a fresh session.', + ).catch((err: unknown) => { + process.stderr.write( + `[${this.name}] failed to announce session rotation in chat ${sanitizeLogText(target.chatId, 64)}: ${this.lifecycleError(err)}\n`, + ); + }); + } + + private hasPendingTurns(sessionId: string): boolean { + return (this.sessionPendingTurns.get(sessionId) ?? 0) > 0; + } + + private trackSessionTurn(sessionId: string, turn: Promise): void { + // The turn is now registered: release resolve()'s routing lease and let + // the pending-turn count carry the rotation deferral from here. + this.router.releaseRoutingLease(sessionId); + this.sessionPendingTurns.set( + sessionId, + (this.sessionPendingTurns.get(sessionId) ?? 0) + 1, + ); + const finish = (): void => { + const remaining = (this.sessionPendingTurns.get(sessionId) ?? 1) - 1; + if (remaining <= 0) { + this.sessionPendingTurns.delete(sessionId); + } else { + this.sessionPendingTurns.set(sessionId, remaining); + } + }; + void turn.then(finish, finish); + } + private attachBridgeEvents(bridge: ChannelAgentBridge): void { bridge.on('toolCall', this.bridgeToolCallListener); bridge.on('backgroundResponse', this.bridgeBackgroundResponseListener); @@ -5149,6 +5226,9 @@ export abstract class ChannelBase { const cmd = bangText.slice(1).trim(); const bridgeShellCommand = this.bridge.shellCommand; if (cmd && bridgeShellCommand) { + // No turn will start for this message, but the shell command still + // runs on the resolved session: hold resolve()'s routing lease until + // it settles so a rotation cannot discard the session mid-command. try { const result = await bridgeShellCommand(sessionId, cmd); const longestRun = Math.max( @@ -5177,6 +5257,11 @@ export abstract class ChannelBase { envelope.threadId, `Shell command failed: ${error instanceof Error ? error.message : String(error)}`, ); + } finally { + // No turn started for this message: give the resolve-time count + // back like the other no-turn paths, then release the lease. + this.router.uncountTurn(this.name, sessionId); + this.router.releaseRoutingLease(sessionId); } return; } @@ -5336,6 +5421,12 @@ export abstract class ChannelBase { `[${this.name}] onPromptBuffered threw for session ${sessionId}: ${err instanceof Error ? err.message : err}\n`, ); } + // The buffered message becomes part of the coalesced drain turn, not + // a turn of its own: undo the resolve-time count (the drain counts + // the coalesced message) and release the routing lease (no turn + // will register for this routing). + this.router.uncountTurn(this.name, sessionId); + this.router.releaseRoutingLease(sessionId); return; } case 'steer': { @@ -5849,6 +5940,7 @@ export abstract class ChannelBase { sessionId, current.catch(() => {}), ); + this.trackSessionTurn(sessionId, current); await current; } diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index 4900bc864e6..3fec1a2179b 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -16,8 +16,10 @@ import { type DaemonChannelSessionClient, } from './DaemonChannelBridge.js'; import type { ChannelAgentBridge } from './ChannelAgentBridge.js'; +import type { SessionTarget } from './types.js'; const mockRenameSync = vi.hoisted(() => vi.fn()); +const mockWriteFileSync = vi.hoisted(() => vi.fn()); vi.mock('node:fs', async (importOriginal) => { const actual = await importOriginal(); @@ -27,6 +29,10 @@ vi.mock('node:fs', async (importOriginal) => { mockRenameSync(from, to); return actual.renameSync(from, to); }, + writeFileSync: (...args: Parameters) => { + mockWriteFileSync(...args); + return actual.writeFileSync(...args); + }, }; }); @@ -41,6 +47,7 @@ function mockBridge(): ChannelAgentBridge { availableCommands: [], prompt: vi.fn().mockResolvedValue(''), cancelSession: vi.fn().mockResolvedValue(undefined), + discardSession: vi.fn().mockResolvedValue(undefined), }; } @@ -61,6 +68,21 @@ function writePersistedSession(persistPath: string, key = 'key1'): void { ); } +function rotationCounters(router: SessionRouter): { + toTurns: Map; + toStartedAt: Map; +} { + return router as unknown as { + toTurns: Map; + toStartedAt: Map; + }; +} + +function routingLeases(router: SessionRouter): Map { + return (router as unknown as { sessionRoutingLeases: Map }) + .sessionRoutingLeases; +} + function invalidationMetadataSize(router: SessionRouter): number { const state = router as unknown as { routeGenerations?: Map; @@ -109,6 +131,7 @@ describe('SessionRouter', () => { beforeEach(() => { sessionCounter = 0; mockRenameSync.mockClear(); + mockWriteFileSync.mockClear(); bridge = mockBridge(); tempDirs = []; }); @@ -1270,6 +1293,1599 @@ describe('SessionRouter', () => { }); }); + describe('session rotation', () => { + /** + * Emulate the channel contract: every message resolve() hands out + * releases its routing lease once settled (its turn was enqueued, or it + * exited without one). Rotation defers while a lease is held. + */ + async function routed( + router: SessionRouter, + ...args: Parameters + ): Promise { + const sessionId = await router.resolve(...args); + router.releaseRoutingLease(sessionId); + return sessionId; + } + + it('keeps reusing the session when no rotation is configured', async () => { + const router = new SessionRouter(bridge, '/tmp'); + const first = await routed(router, 'ch', 'alice', 'chat1'); + for (let index = 0; index < 20; index++) { + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + } + }); + + it('starts a new session once maxTurns is reached', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 3 }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + + const rotated = await routed(router, 'ch', 'alice', 'chat1'); + expect(rotated).not.toBe(first); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(rotated); + }); + + it('rotates only the route that hit the bound', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 2 }); + + const busy = await routed(router, 'ch', 'alice', 'chat1'); + const quiet = await routed(router, 'ch', 'bob', 'chat2'); + await routed(router, 'ch', 'alice', 'chat1'); + + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(busy); + expect(await routed(router, 'ch', 'bob', 'chat2')).toBe(quiet); + }); + + it('does not rotate a channel that has no bound configured', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('bounded', { maxTurns: 2 }); + + const unbounded = await routed(router, 'other', 'alice', 'chat1'); + for (let index = 0; index < 5; index++) { + expect(await routed(router, 'other', 'alice', 'chat1')).toBe(unbounded); + } + }); + + it('starts a new session once maxAgeHours has elapsed', async () => { + vi.useFakeTimers(); + try { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxAgeHours: 2 }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + vi.advanceTimersByTime(90 * 60 * 1000); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + + vi.advanceTimersByTime(31 * 60 * 1000); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + } finally { + vi.useRealTimers(); + } + }); + + it('ignores non-positive bounds instead of rotating every message', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 0, maxAgeHours: -1 }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + }); + + it('persists turn counts so a restart cannot reset the bound', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + const first = await routed(router, 'ch', 'alice', 'chat1'); + await routed(router, 'ch', 'alice', 'chat1'); + + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1'].turns).toBe(2); + + const revived = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + revived.setChannelRotation('ch', { maxTurns: 3 }); + revived.restoreRoutes(); + + expect(await routed(revived, 'ch', 'alice', 'chat1')).toBe(first); + expect(await routed(revived, 'ch', 'alice', 'chat1')).not.toBe(first); + }); + + it('accepts route stores written before rotation existed', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + router.setChannelRotation('ch', { maxAgeHours: 1 }); + + expect(router.restoreRoutes()).toEqual({ restored: 1, dropped: 0 }); + // No recorded start: the clock starts now rather than rotating on sight. + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe('old-session'); + }); + + it('persists the stamped start of a pre-rotation route across restarts', async () => { + vi.useFakeTimers(); + try { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writePersistedSession(persistPath, 'ch:alice:chat1'); + + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + router.setChannelRotation('ch', { maxAgeHours: 1 }); + router.restoreRoutes(); + + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe( + 'old-session', + ); + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(typeof persisted['ch:alice:chat1'].startedAt).toBe('number'); + + // The stamp write happens once; later messages stay write-free. + mockWriteFileSync.mockClear(); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe( + 'old-session', + ); + expect(mockWriteFileSync).not.toHaveBeenCalled(); + + // A restart restores the stamped clock instead of re-arming it. + vi.advanceTimersByTime(61 * 60 * 1000); + const revived = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + revived.setChannelRotation('ch', { maxAgeHours: 1 }); + revived.restoreRoutes(); + + expect(await routed(revived, 'ch', 'alice', 'chat1')).not.toBe( + 'old-session', + ); + } finally { + vi.useRealTimers(); + } + }); + + it('discards the retired session when rotating', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + const first = await routed(router, 'ch', 'alice', 'chat1'); + + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + await drainMicrotasks(); + + expect(bridge.discardSession).toHaveBeenCalledWith(first); + }); + + it('notifies rotation listeners with the retired session and target', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + const rotated: Array<{ + sessionId: string; + target: SessionTarget | undefined; + }> = []; + router.onSessionRotated((sessionId, target) => { + rotated.push({ sessionId, target }); + }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + await routed(router, 'ch', 'alice', 'chat1'); + + expect(rotated).toEqual([ + { + sessionId: first, + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + threadId: undefined, + isGroup: undefined, + }, + }, + ]); + }); + + it('unsubscribes rotation listeners', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + const listener = vi.fn(); + const unsubscribe = router.onSessionRotated(listener); + unsubscribe(); + + await routed(router, 'ch', 'alice', 'chat1'); + await routed(router, 'ch', 'alice', 'chat1'); + + expect(listener).not.toHaveBeenCalled(); + }); + + it('keeps rotating when a rotation listener throws', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + const seen: string[] = []; + router.onSessionRotated(() => { + throw new Error('listener boom'); + }); + router.onSessionRotated((sessionId) => { + seen.push(sessionId); + }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + // The throwing listener must not skip the discard, silence the rest, + // or fail the message after the retirement already landed. + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + await drainMicrotasks(); + + expect(seen).toEqual([first]); + expect(bridge.discardSession).toHaveBeenCalledWith(first); + }); + + it('defers rotation while the outgoing session is still active', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + let busy = true; + router.setSessionActivityChecker('ch', () => busy); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + // At the bound but mid-turn: the route stays and nothing is discarded. + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + expect(bridge.discardSession).not.toHaveBeenCalled(); + + busy = false; + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + await drainMicrotasks(); + expect(bridge.discardSession).toHaveBeenCalledWith(first); + }); + + it('defers rotation while a routed message has not settled yet', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + + // The second message rotates, and its routing lease stays held while + // the message is still between resolve() and turn registration. + const second = await router.resolve('ch', 'alice', 'chat1'); + expect(second).not.toBe(first); + + // A third message must not rotate the successor out from under the + // second: the second resolved it and is about to prompt it. + expect(await router.resolve('ch', 'alice', 'chat1')).toBe(second); + expect(bridge.discardSession).toHaveBeenCalledTimes(1); + expect(bridge.discardSession).toHaveBeenCalledWith(first); + + // Release one of the two leases: with one still outstanding the + // rotation must keep deferring. + router.releaseRoutingLease(second); + const held = await router.resolve('ch', 'alice', 'chat1'); + expect(held).toBe(second); + + router.releaseRoutingLease(held); + router.releaseRoutingLease(second); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(second); + await drainMicrotasks(); + expect(bridge.discardSession).toHaveBeenCalledWith(second); + }); + + it('does not rotate a route while its reload is in flight', async () => { + vi.useFakeTimers(); + try { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + startedAt: Date.now(), + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + router.setChannelRotation('ch', { maxAgeHours: 1 }); + let releaseLoad!: (sessionId: string) => void; + (bridge.loadSession as ReturnType).mockReturnValue( + new Promise((resolve) => { + releaseLoad = resolve; + }), + ); + router.restoreRoutes(); + + // The first resolve starts the reload; once the age bound passes + // mid-reload, a second resolve must wait for the reload instead of + // rotating (which would invalidate the concurrent message). + const first = router.resolve('ch', 'alice', 'chat1'); + await vi.advanceTimersByTimeAsync(2 * 60 * 60 * 1000); + const second = router.resolve('ch', 'alice', 'chat1'); + releaseLoad('reloaded-session'); + + expect(await first).toBe('reloaded-session'); + expect(await second).toBe('reloaded-session'); + expect(bridge.discardSession).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it('stops deferring rotation once the activity checker is unregistered', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + router.setSessionActivityChecker('ch', () => true); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + // The checker reports the session as active forever, so the bound + // never fires while it is registered. + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + + router.setSessionActivityChecker('ch', undefined); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + }); + + it('restores rotation counters in eager recovery', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 2, + startedAt: Date.now(), + }, + }), + ); + // Default (eager) recovery: channel start uses restoreSessions(). + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + + await router.restoreSessions(); + + // Carried turns (2) plus this message reach the bound: next rotates. + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe('old-session'); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe( + 'old-session', + ); + }); + + it('rotates an at-bound session handed back by an in-flight restore', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 3, + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + let releaseRestore!: (sessionId: string) => void; + (bridge.loadSession as ReturnType).mockReturnValue( + new Promise((resolve) => { + releaseRestore = resolve; + }), + ); + + // The message parks on the restore reservation; the restored session + // comes back already at its bound and must be rotated, not reused. + const restoring = router.restoreSessions(); + const waiter = router.resolve('ch', 'alice', 'chat1'); + releaseRestore('old-session'); + await restoring; + + const sessionId = await waiter; + router.releaseRoutingLease(sessionId); + expect(sessionId).not.toBe('old-session'); + await drainMicrotasks(); + expect(bridge.discardSession).toHaveBeenCalledWith('old-session'); + + // The waiter was the successor's first turn: the remaining bound of + // turns reuses it, and only then does it rotate. + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(sessionId); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(sessionId); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(sessionId); + }); + + it('routes every waiter when an at-bound restore rotates under them', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 3, + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + let releaseRestore!: (sessionId: string) => void; + (bridge.loadSession as ReturnType).mockReturnValue( + new Promise((resolve) => { + releaseRestore = resolve; + }), + ); + + // Two messages park on the same restore reservation. The restored + // session is already at its bound, so the first waiter rotates it — + // destroying the route token the second waiter's reservation still + // references. The second must follow the successor, not fail with + // the invalidation. + const restoring = router.restoreSessions(); + const first = router.resolve('ch', 'alice', 'chat1'); + const second = router.resolve('ch', 'alice', 'chat1'); + releaseRestore('old-session'); + await restoring; + + const firstId = await first; + const secondId = await second; + router.releaseRoutingLease(firstId); + router.releaseRoutingLease(secondId); + expect(firstId).not.toBe('old-session'); + expect(secondId).toBe(firstId); + expect(bridge.newSession).toHaveBeenCalledTimes(1); + await drainMicrotasks(); + expect(bridge.discardSession).toHaveBeenCalledWith('old-session'); + + // The retried message's turn is carried and counts against the bound: + // one more reuse reaches it, and only then does the route rotate. + expect(rotationCounters(router).toTurns.get(firstId)).toBe(2); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(firstId); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(firstId); + }); + + it('never persists a truncated store while an eager restore is mid-flight', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 3, + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + const restoring = router.restoreSessions(); + await Promise.resolve(); + // The message parks on alice's reservation and is released while + // bob's key is still unrestored; its rotation then persists. + const waiter = router.resolve('ch', 'alice', 'chat1'); + loadResolvers[0]!('old-alice'); + + // The waiter settles (rotating the at-bound session and persisting) + // before bob's restore is released: this is the truncation window. + const sessionId = await waiter; + router.releaseRoutingLease(sessionId); + expect(sessionId).not.toBe('old-alice'); + + loadResolvers[1]!('old-bob'); + await restoring; + + // Every write must hold the whole store: a truncated prefix becoming + // durable silently loses bob's route on the next restart. + for (const call of mockWriteFileSync.mock.calls) { + const written = JSON.parse(String(call[1])) as Record< + string, + { sessionId: string } + >; + expect(written['ch:bob:chat2']?.sessionId).toBe('old-bob'); + } + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:bob:chat2'].sessionId).toBe('old-bob'); + expect(persisted['ch:alice:chat1'].sessionId).toBe(sessionId); + }); + + it('keeps the store whole when a second restore overlaps the first', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + 'ch:carol:chat3': { + sessionId: 'old-carol', + target: { + channelName: 'ch', + senderId: 'carol', + chatId: 'chat3', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + // Drop the setup write: only router writes count below. + mockWriteFileSync.mockClear(); + + // A reconnect READY fires a second restore while the cold-start one + // still runs: the first has restored alice already, and the second's + // reservation pass drops alice back out of the store before the first + // finishes — the truncation window. + const first = router.restoreSessions(); + await drainMicrotasks(); + loadResolvers[0]!('old-alice'); + await drainMicrotasks(); + const second = router.restoreSessions(); + await drainMicrotasks(); + expect(loadResolvers).toHaveLength(3); + + // A new route created mid-overlap persists only through the flush of + // the LAST restore to finish: the earlier finisher must not consume + // the request or lift the suspension. + const dave = await router.resolve('ch', 'dave', 'chat4'); + router.releaseRoutingLease(dave); + + loadResolvers[1]!('old-bob'); + await drainMicrotasks(); + loadResolvers[3]!('old-carol'); + await drainMicrotasks(); + loadResolvers[2]!('old-alice'); + await drainMicrotasks(); + loadResolvers[4]!('old-bob'); + await drainMicrotasks(); + loadResolvers[5]!('old-carol'); + + await expect(first).resolves.toEqual({ restored: 3, failed: 0 }); + await expect(second).resolves.toEqual({ restored: 3, failed: 0 }); + + // Every write holds the whole store: an earlier finisher flushing the + // later restore's partial prefix is the truncation this guards. + expect(mockWriteFileSync).toHaveBeenCalled(); + for (const call of mockWriteFileSync.mock.calls) { + const written = JSON.parse(String(call[1])) as Record< + string, + { sessionId: string } + >; + expect(written['ch:alice:chat1']?.sessionId).toBe('old-alice'); + expect(written['ch:bob:chat2']?.sessionId).toBe('old-bob'); + expect(written['ch:carol:chat3']?.sessionId).toBe('old-carol'); + expect(written['ch:dave:chat4']).toBeDefined(); + } + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(Object.keys(persisted)).toHaveLength(4); + }); + + it('keeps live rotation counters when a second restore overlaps the first', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 2, + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 4 }); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + const first = router.restoreSessions(); + await drainMicrotasks(); + loadResolvers[0]!('old-alice'); + await drainMicrotasks(); + + // A message routes onto the restored session while bob still loads: + // its turn and lease must survive the second restore's reservation + // pass instead of being rewound to the stale on-disk snapshot. + expect(await router.resolve('ch', 'alice', 'chat1')).toBe('old-alice'); + + const second = router.restoreSessions(); + await drainMicrotasks(); + expect(loadResolvers).toHaveLength(3); + + loadResolvers[2]!('old-alice'); + await drainMicrotasks(); + loadResolvers[1]!('old-bob'); + await drainMicrotasks(); + loadResolvers[3]!('old-bob'); + + await expect(first).resolves.toEqual({ restored: 2, failed: 0 }); + await expect(second).resolves.toEqual({ restored: 2, failed: 0 }); + + expect(rotationCounters(router).toTurns.get('old-alice')).toBe(3); + expect(routingLeases(router).get('old-alice')).toBe(1); + router.releaseRoutingLease('old-alice'); + + // The last finisher's flush persisted the live counter, not the + // snapshot's stale turns: 2. + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1'].sessionId).toBe('old-alice'); + expect(persisted['ch:alice:chat1'].turns).toBe(3); + expect(persisted['ch:bob:chat2'].sessionId).toBe('old-bob'); + }); + + it('drains a routing lease released while a restore holds the route wiped', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 2, + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + const first = router.restoreSessions(); + await drainMicrotasks(); + loadResolvers[0]!('old-alice'); + await expect(first).resolves.toEqual({ restored: 1, failed: 0 }); + + // A message routes onto the restored session and holds its lease + // across an unsettled shell command when a reconnect restore wipes + // the route. + expect(await router.resolve('ch', 'alice', 'chat1')).toBe('old-alice'); + + const second = router.restoreSessions(); + await drainMicrotasks(); + + // The command settles in the wipe window: its release must net + // against the carry instead of being swallowed by the wipe and + // re-added as a phantom lease that defers rotation forever. + router.releaseRoutingLease('old-alice'); + + loadResolvers[1]!('old-alice'); + await expect(second).resolves.toEqual({ restored: 1, failed: 0 }); + + expect(routingLeases(router).has('old-alice')).toBe(false); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe( + 'old-alice', + ); + }); + + it('nets a turn uncounted mid-restore instead of rewinding to the snapshot', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 1, + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + const first = router.restoreSessions(); + await drainMicrotasks(); + loadResolvers[0]!('old-alice'); + await expect(first).resolves.toEqual({ restored: 1, failed: 0 }); + + // A message resolves the restored session, then buffers (collect + // mode) and settles without a turn while the next restore wipes it. + expect(await router.resolve('ch', 'alice', 'chat1')).toBe('old-alice'); + + const second = router.restoreSessions(); + await drainMicrotasks(); + router.uncountTurn('ch', 'old-alice'); + router.releaseRoutingLease('old-alice'); + + loadResolvers[1]!('old-alice'); + await expect(second).resolves.toEqual({ restored: 1, failed: 0 }); + + // The give-back survived the wipe window: one counted message, not + // the snapshot's pre-decrement count re-counted by the drain. + expect(rotationCounters(router).toTurns.get('old-alice')).toBe(1); + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1'].turns).toBe(1); + }); + + it('does not resurrect a route cleared before an overlapping restore', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + const first = router.restoreSessions(); + await drainMicrotasks(); + loadResolvers[0]!('old-alice'); + await drainMicrotasks(); + + // /clear lands while the first restore still loads bob; the removal + // is suspended with everything else. A reconnect READY then fires a + // second restore against the stale pre-clear snapshot. + router.removeSession('ch', 'alice', 'chat1'); + + const second = router.restoreSessions(); + await drainMicrotasks(); + expect(loadResolvers).toHaveLength(3); + + loadResolvers[1]!('old-bob'); + await drainMicrotasks(); + loadResolvers[2]!('old-bob'); + + await expect(first).resolves.toEqual({ restored: 2, failed: 0 }); + await expect(second).resolves.toEqual({ restored: 1, failed: 0 }); + + expect(router.getSession('ch', 'alice', 'chat1')).toBeUndefined(); + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1']).toBeUndefined(); + expect(persisted['ch:bob:chat2'].sessionId).toBe('old-bob'); + }); + + it('does not resurrect a rotated route when a second restore overlaps', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 3, + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + + const restoring = router.restoreSessions(); + await drainMicrotasks(); + const waiter = router.resolve('ch', 'alice', 'chat1'); + loadResolvers[0]!('old-alice'); + + // The restored session is at its bound: the waiter rotates it + // mid-restore, retiring the key while persistence stays suspended. + const successor = await waiter; + router.releaseRoutingLease(successor); + expect(successor).not.toBe('old-alice'); + + // A reconnect READY fires a second restore against the stale + // snapshot that still lists the rotated route. + const second = router.restoreSessions(); + await drainMicrotasks(); + expect(loadResolvers).toHaveLength(3); + + loadResolvers[1]!('old-bob'); + await drainMicrotasks(); + loadResolvers[2]!('old-bob'); + + await expect(restoring).resolves.toEqual({ restored: 2, failed: 0 }); + await expect(second).resolves.toEqual({ restored: 1, failed: 0 }); + + expect(router.getSession('ch', 'alice', 'chat1')).toBe(successor); + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1'].sessionId).toBe(successor); + expect(persisted['ch:bob:chat2'].sessionId).toBe('old-bob'); + }); + + it('flushes the whole store when an earlier finisher dropped a key', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + const loads: Array<{ + resolve: (sessionId: string) => void; + reject: (error: Error) => void; + }> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve, reject) => { + loads.push({ resolve, reject }); + }), + ); + mockWriteFileSync.mockClear(); + + const first = router.restoreSessions(); + await drainMicrotasks(); + loads[0]!.resolve('old-alice'); + await drainMicrotasks(); + + // The overlap's reservation pass wipes the alice route the first + // restore already brought back; its own alice load then fails. + const second = router.restoreSessions(); + await drainMicrotasks(); + expect(loads).toHaveLength(3); + + loads[2]!.reject(new Error('session file gone')); + await drainMicrotasks(); + loads[3]!.resolve('old-bob'); + await drainMicrotasks(); + loads[1]!.resolve('old-bob'); + + await expect(first).resolves.toEqual({ restored: 2, failed: 0 }); + await expect(second).resolves.toEqual({ restored: 1, failed: 1 }); + + // The earlier finisher's drop must reach disk through the last + // finisher's flush even though the last one dropped nothing itself: + // memory no longer maps alice, so the store must not keep her. + expect(mockWriteFileSync).toHaveBeenCalledTimes(1); + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1']).toBeUndefined(); + expect(persisted['ch:bob:chat2'].sessionId).toBe('old-bob'); + }); + + it('defers a mid-restore rotation persist to the restore end flush', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-alice', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 3, + }, + 'ch:bob:chat2': { + sessionId: 'old-bob', + target: { + channelName: 'ch', + senderId: 'bob', + chatId: 'chat2', + }, + cwd: '/tmp', + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + const loadResolvers: Array<(sessionId: string) => void> = []; + (bridge.loadSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + loadResolvers.push(resolve); + }), + ); + mockWriteFileSync.mockClear(); + + const restoring = router.restoreSessions(); + await drainMicrotasks(); + const waiter = router.resolve('ch', 'alice', 'chat1'); + loadResolvers[0]!('old-alice'); + + // The restored session is already at its bound: the waiter rotates it + // mid-restore. The retirement persist is suspended with everything + // else while bob still loads; a crash in that window re-fires the + // rotation once on the next start, per the rotateRoute contract. + const successor = await waiter; + router.releaseRoutingLease(successor); + expect(successor).not.toBe('old-alice'); + expect(bridge.discardSession).toHaveBeenCalledWith('old-alice'); + expect(mockWriteFileSync).not.toHaveBeenCalled(); + + loadResolvers[1]!('old-bob'); + await restoring; + + // Durability arrives with the end flush: the store is whole and the + // retirement is kept. + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1'].sessionId).toBe(successor); + expect(persisted['ch:alice:chat1'].turns).toBe(1); + expect(persisted['ch:bob:chat2'].sessionId).toBe('old-bob'); + }); + + it('rejects a parked waiter once the invalidation retry budget runs out', async () => { + const router = new SessionRouter(bridge, '/tmp'); + const creations: Array<(sessionId: string) => void> = []; + (bridge.newSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + creations.push(resolve); + }), + ); + + // The waiter parks on someone else's in-flight creation; each dispose + // invalidates it, and the next resolve recreates the route, leaving a + // successor for the retry. The creators of invalidated operations + // reject outright (creation carries no retry), so swallow theirs. + const creator1 = router.resolve('ch', 'alice', 'chat1'); + void creator1.catch(() => undefined); + const waiter = router.resolve('ch', 'alice', 'chat1'); + await drainMicrotasks(); + + router.dispose(); + const creator2 = router.resolve('ch', 'alice', 'chat1'); + void creator2.catch(() => undefined); + creations[0]!('session-a'); + await drainMicrotasks(); + + router.dispose(); + const creator3 = router.resolve('ch', 'alice', 'chat1'); + void creator3.catch(() => undefined); + creations[1]!('session-b'); + await drainMicrotasks(); + + router.dispose(); + const creator4 = router.resolve('ch', 'alice', 'chat1'); + void creator4.catch(() => undefined); + creations[2]!('session-c'); + await drainMicrotasks(); + + router.dispose(); + const creator5 = router.resolve('ch', 'alice', 'chat1'); + creations[3]!('session-d'); + await drainMicrotasks(); + + // The waiter survived three invalidations and rejects on the fourth + // instead of parking on yet another successor. + await expect(waiter).rejects.toThrow( + 'Session route operation was invalidated', + ); + expect(bridge.discardSession).toHaveBeenCalledWith( + 'session-a', + expect.anything(), + ); + expect(bridge.discardSession).toHaveBeenCalledWith( + 'session-b', + expect.anything(), + ); + expect(bridge.discardSession).toHaveBeenCalledWith( + 'session-c', + expect.anything(), + ); + + // The churn ends with the last creation: it still routes. + creations[4]!('session-e'); + expect(await creator5).toBe('session-e'); + }); + + it('routes a waiter that outlives consecutive invalidations', async () => { + const router = new SessionRouter(bridge, '/tmp'); + const creations: Array<(sessionId: string) => void> = []; + (bridge.newSession as ReturnType).mockImplementation( + () => + new Promise((resolve) => { + creations.push(resolve); + }), + ); + + // The waiter parks on someone else's in-flight creation; two disposes + // invalidate it back to back, and it must route the third successor. + const creator1 = router.resolve('ch', 'alice', 'chat1'); + void creator1.catch(() => undefined); + const waiter = router.resolve('ch', 'alice', 'chat1'); + await drainMicrotasks(); + + router.dispose(); + const creator2 = router.resolve('ch', 'alice', 'chat1'); + void creator2.catch(() => undefined); + creations[0]!('session-a'); + await drainMicrotasks(); + + router.dispose(); + const creator3 = router.resolve('ch', 'alice', 'chat1'); + creations[1]!('session-b'); + await drainMicrotasks(); + + creations[2]!('session-c'); + + expect(await waiter).toBe('session-c'); + expect(await creator3).toBe('session-c'); + // The stale results of the invalidated creations were discarded. + expect(bridge.discardSession).toHaveBeenCalledWith( + 'session-a', + expect.anything(), + ); + expect(bridge.discardSession).toHaveBeenCalledWith( + 'session-b', + expect.anything(), + ); + }); + + it('clears a bound when the channel re-registers without one', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 2 }); + const first = await routed(router, 'ch', 'alice', 'chat1'); + await routed(router, 'ch', 'alice', 'chat1'); // turns: 2, at the bound + + router.setChannelRotation('ch', undefined); + for (let index = 0; index < 3; index++) { + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + } + + router.setChannelRotation('ch', { maxTurns: 0 }); // normalized away + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + }); + + it('carries counters over an ID-changing reload', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 2, + startedAt: Date.now(), + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + router.setChannelRotation('ch', { maxTurns: 3 }); + (bridge.loadSession as ReturnType).mockResolvedValue( + 'reloaded-session', + ); + router.restoreRoutes(); + + // The ID-changing reload persists once: countTurn's write subsumes the + // carry-over, so maxTurns channels must not pay two whole-store writes. + mockWriteFileSync.mockClear(); + const carried = await routed(router, 'ch', 'alice', 'chat1'); + expect(carried).toBe('reloaded-session'); + expect(mockWriteFileSync).toHaveBeenCalledTimes(1); + // Carried turns (2) plus this message reach the bound: next rotates. + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(carried); + }); + + it('carries the start stamp over an ID-changing reload', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const stamp = Date.now() - 30 * 60 * 1000; + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 1, + startedAt: stamp, + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + router.setChannelRotation('ch', { maxAgeHours: 24 }); + (bridge.loadSession as ReturnType).mockResolvedValue( + 'reloaded-session', + ); + router.restoreRoutes(); + + mockWriteFileSync.mockClear(); + await router.resolve('ch', 'alice', 'chat1'); + + // Without the carry-over the reload would re-stamp 'now', silently + // re-arming the age clock on every restart that mints a fresh ID. + expect(rotationCounters(router).toStartedAt.get('reloaded-session')).toBe( + stamp, + ); + // Age-only channels keep the carry-over persist: countTurn writes + // nothing for them, so this is the only write of the new ID + stamp. + expect(mockWriteFileSync).toHaveBeenCalledTimes(1); + }); + + it('counts messages that waited on an in-flight creation', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 2 }); + let releaseCreation!: (sessionId: string) => void; + (bridge.newSession as ReturnType).mockReturnValueOnce( + new Promise((resolve) => { + releaseCreation = resolve; + }), + ); + + const creator = router.resolve('ch', 'alice', 'chat1'); + const waiter = router.resolve('ch', 'alice', 'chat1'); + releaseCreation('busy-session'); + + expect(await creator).toBe('busy-session'); + router.releaseRoutingLease('busy-session'); + expect(await waiter).toBe('busy-session'); + // The waiter takes its own lease before returning: without it, a + // message arriving before its turn registers could rotate the session + // out from under it. + expect(routingLeases(router).get('busy-session')).toBe(1); + router.releaseRoutingLease('busy-session'); + + // Creator and waiter both counted: the next message hits the bound. + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe( + 'busy-session', + ); + }); + + it('persists once per routed message instead of stacking writes', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 2 }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + expect(mockWriteFileSync).toHaveBeenCalledTimes(1); // creation = turn 1 + mockWriteFileSync.mockClear(); + + await routed(router, 'ch', 'alice', 'chat1'); + expect(mockWriteFileSync).toHaveBeenCalledTimes(1); // counter update + mockWriteFileSync.mockClear(); + + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + // Rotation persists the retirement so a failed successor creation + // cannot re-fire it after a restart; the successor creation persists + // again. + expect(mockWriteFileSync).toHaveBeenCalledTimes(2); + + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1'].turns).toBe(1); + }); + + it('keeps a retirement durable when the successor creation fails', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 1 }); + let rotations = 0; + router.onSessionRotated(() => { + rotations++; + }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + (bridge.newSession as ReturnType).mockRejectedValueOnce( + new Error('at capacity'), + ); + + // The rotation retires and persists BEFORE the successor is created; + // the creation failure must surface without un-retiring the route. + await expect(router.resolve('ch', 'alice', 'chat1')).rejects.toThrow( + 'at capacity', + ); + const persisted = JSON.parse(readFileSync(persistPath, 'utf-8')); + expect(persisted['ch:alice:chat1']).toBeUndefined(); + + // A restart therefore cannot restore the stale at-bound route and + // re-fire the rotation (a second notice + discard for one rotation). + const revived = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + revived.setChannelRotation('ch', { maxTurns: 1 }); + expect(revived.restoreRoutes()).toEqual({ restored: 0, dropped: 0 }); + + expect(await routed(revived, 'ch', 'alice', 'chat1')).not.toBe(first); + expect(rotations).toBe(1); + await drainMicrotasks(); + expect(bridge.discardSession).toHaveBeenCalledTimes(1); + expect(bridge.discardSession).toHaveBeenCalledWith(first); + }); + + it('does not write per message when only maxAgeHours is configured', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxAgeHours: 24 }); + + await routed(router, 'ch', 'alice', 'chat1'); + expect(mockWriteFileSync).toHaveBeenCalledTimes(1); + mockWriteFileSync.mockClear(); + + await routed(router, 'ch', 'alice', 'chat1'); + expect(mockWriteFileSync).not.toHaveBeenCalled(); + }); + + it('drops persisted entries with impossible rotation counters', () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + const entry = { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + }; + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { ...entry, turns: -3 }, + 'ch:bob:chat2': { ...entry, sessionId: 's2', turns: 0.5 }, + 'ch:carol:chat3': { ...entry, sessionId: 's3', startedAt: 0 }, + 'ch:dave:chat4': { ...entry, sessionId: 's4', turns: 2 }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath, { + recoveryMode: 'lazy', + }); + + expect(router.restoreRoutes()).toEqual({ restored: 1, dropped: 3 }); + }); + + it('drops counters when a session is removed by ID', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 5, maxAgeHours: 24 }); + const sessionId = await routed(router, 'ch', 'alice', 'chat1'); + + router.removeSessionId(sessionId); + + expect(rotationCounters(router).toTurns.size).toBe(0); + expect(rotationCounters(router).toStartedAt.size).toBe(0); + }); + + it('drops counters when a route is removed by key', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 5, maxAgeHours: 24 }); + await routed(router, 'ch', 'alice', 'chat1'); + + router.removeSession('ch', 'alice', 'chat1'); + + expect(rotationCounters(router).toTurns.size).toBe(0); + expect(rotationCounters(router).toStartedAt.size).toBe(0); + }); + + it('clears counters on dispose', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 5, maxAgeHours: 24 }); + await routed(router, 'ch', 'alice', 'chat1'); + + router.dispose(); + + expect(rotationCounters(router).toTurns.size).toBe(0); + expect(rotationCounters(router).toStartedAt.size).toBe(0); + }); + + it('gives the seed count back when the first message never prompts', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 2 }); + + const first = await router.resolve('ch', 'alice', 'chat1'); + router.releaseRoutingLease(first); + // The first firing is dropped before it prompts (a skipped loop + // firing): the uncount must give the seed count back, or the session + // would rotate one turn early for the rest of its life. + router.uncountTurn('ch', first); + expect(rotationCounters(router).toTurns.has(first)).toBe(false); + + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + }); + + it('reclaims routing leases when a session is removed by ID', async () => { + const router = new SessionRouter(bridge, '/tmp'); + const sessionId = await router.resolve('ch', 'alice', 'chat1'); + expect(routingLeases(router).get(sessionId)).toBe(1); + + router.removeSessionId(sessionId); + + expect(routingLeases(router).has(sessionId)).toBe(false); + }); + + it('enforces both bounds when they are configured together', async () => { + vi.useFakeTimers(); + try { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 10, maxAgeHours: 1 }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(first); + + // Only 2 of 10 turns used, but the age bound passes: the age + // rotation must still fire on a channel that also sets maxTurns. + vi.advanceTimersByTime(61 * 60 * 1000); + const aged = await routed(router, 'ch', 'alice', 'chat1'); + expect(aged).not.toBe(first); + + // Symmetrically, the turns bound fires while the successor is + // still age-fresh. + router.setChannelRotation('ch', { maxTurns: 2, maxAgeHours: 24 }); + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe(aged); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(aged); + } finally { + vi.useRealTimers(); + } + }); + + it('carries rotation counters over an ID-changing eager restore', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'routes.json'); + writeFileSync( + persistPath, + JSON.stringify({ + 'ch:alice:chat1': { + sessionId: 'old-session', + target: { + channelName: 'ch', + senderId: 'alice', + chatId: 'chat1', + }, + cwd: '/tmp', + turns: 2, + }, + }), + ); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + router.setChannelRotation('ch', { maxTurns: 3 }); + (bridge.loadSession as ReturnType).mockResolvedValue( + 'replacement-session', + ); + + await router.restoreSessions(); + + // The counters follow the id loadSession resolved to, not the + // persisted id: carried turns (2) plus this message reach the bound. + expect(await routed(router, 'ch', 'alice', 'chat1')).toBe( + 'replacement-session', + ); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe( + 'replacement-session', + ); + expect(rotationCounters(router).toTurns.has('old-session')).toBe(false); + }); + + it('keeps rotating when discarding the retired session throws', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + ( + bridge.discardSession as ReturnType + ).mockImplementationOnce(() => { + throw new Error('connection down'); + }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + await drainMicrotasks(); + expect(bridge.discardSession).toHaveBeenCalledWith(first); + }); + + it('attaches a catch handler to the fire-and-forget rotation discard', async () => { + const router = new SessionRouter(bridge, '/tmp'); + router.setChannelRotation('ch', { maxTurns: 1 }); + const attachCatch = vi.fn(); + (bridge.discardSession as ReturnType).mockReturnValueOnce({ + catch: attachCatch, + }); + + const first = await routed(router, 'ch', 'alice', 'chat1'); + expect(await routed(router, 'ch', 'alice', 'chat1')).not.toBe(first); + + // Without the catch handler a rejected close RPC becomes an unhandled + // rejection, taking the channel process down after the retirement + // already landed. + expect(attachCatch).toHaveBeenCalledOnce(); + }); + }); describe('clearAll', () => { it('clears all in-memory state', async () => { const router = new SessionRouter(bridge, '/tmp'); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index 32132b94ea2..415a57f2faa 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -10,7 +10,11 @@ import { } from 'node:fs'; import { dirname, join } from 'node:path'; import process from 'node:process'; -import type { SessionScope, SessionTarget } from './types.js'; +import type { + SessionRotationConfig, + SessionScope, + SessionTarget, +} from './types.js'; import type { ChannelAgentBridge } from './ChannelAgentBridge.js'; import { sanitizeLogText } from './sanitize.js'; @@ -18,6 +22,10 @@ interface PersistedEntry { sessionId: string; target: SessionTarget; cwd: string; + /** Messages routed to this session so far. Absent on pre-rotation stores. */ + turns?: number; + /** Epoch ms the session was first routed to. Absent on pre-rotation stores. */ + startedAt?: number; } interface SessionReservation { @@ -49,6 +57,9 @@ export class SessionRouter { private toSession: Map = new Map(); // routing key → session ID private toTarget: Map = new Map(); // session ID → target private toCwd: Map = new Map(); // session ID → cwd + private toTurns: Map = new Map(); // session ID → messages routed + private toStartedAt: Map = new Map(); // session ID → epoch ms + private sessionRoutingLeases: Map = new Map(); // session ID → routed-but-unsettled messages private creatingSessions: Map = new Map(); private sessionLoadWindows: Set = new Set(); private readonly liveSessionIds = new Set(); @@ -60,7 +71,29 @@ export class SessionRouter { private defaultScope: SessionScope; private channelScopes: Map = new Map(); private channelApprovalModes: Map = new Map(); + private channelRotations: Map = new Map(); + private sessionActivityCheckers = new Map< + string, + (sessionId: string) => boolean + >(); + private rotationListeners = new Set< + (sessionId: string, target: SessionTarget | undefined) => void + >(); private persistPath: string | undefined; + private persistSuspendDepth = 0; + private persistRequestedWhileSuspended = false; + // Rotation-state writes (countTurn, uncountTurn, leaseSession, + // releaseRoutingLease) that land while a restore's reservation pass holds + // the session wiped, keyed by the wiped session ID. The restore's carry + // nets them against the captured baseline instead of overwriting them. + private readonly rotationDeltas = new Map< + string, + { turns: number; leases: number } + >(); + // Routing keys removed while persistence is suspended: the removal only + // reaches disk at the last restore's flush, and any restore reading the + // stale snapshot until then must skip them instead of resurrecting them. + private readonly suspendedDeletionKeys = new Set(); private readonly recoveryMode: SessionRecoveryMode; constructor( @@ -98,6 +131,221 @@ export class SessionRouter { } } + /** Set session auto-rotation bounds for a specific channel. */ + setChannelRotation( + channelName: string, + rotation: SessionRotationConfig | undefined, + ): void { + const maxTurns = isValidTurnCount(rotation?.maxTurns) + ? rotation.maxTurns + : undefined; + const maxAgeHours = normalizeRotationBound(rotation?.maxAgeHours); + if (maxTurns === undefined && maxAgeHours === undefined) { + this.channelRotations.delete(channelName); + return; + } + this.channelRotations.set(channelName, { + ...(maxTurns !== undefined ? { maxTurns } : {}), + ...(maxAgeHours !== undefined ? { maxAgeHours } : {}), + }); + } + + /** + * Whether `sessionId` has outgrown its channel's rotation bounds. Checked + * before a message reuses the session, so the bound is a ceiling on what the + * session carries into a turn rather than what it is left holding after one. + */ + private shouldRotate(channelName: string, sessionId: string): boolean { + const rotation = this.channelRotations.get(channelName); + if (!rotation) return false; + + const { maxTurns, maxAgeHours } = rotation; + if ( + maxTurns !== undefined && + (this.toTurns.get(sessionId) ?? 0) >= maxTurns + ) { + return true; + } + if (maxAgeHours !== undefined) { + const startedAt = this.toStartedAt.get(sessionId); + // A session with no recorded start (restored from a pre-rotation store) + // gets its clock started here rather than rotating immediately. The + // stamp is persisted: age-only bounds write nothing per message, so a + // memory-only stamp would let restarts re-arm the clock indefinitely. + if (startedAt === undefined) { + this.toStartedAt.set(sessionId, Date.now()); + this.persist(); + return false; + } + if (Date.now() - startedAt >= maxAgeHours * 60 * 60 * 1000) return true; + } + return false; + } + + /** + * Register a checker reporting whether sessions on `channelName` still have + * a turn running or queued. Rotation defers while one does, so a route is + * never retired out from under an in-flight turn. + */ + setSessionActivityChecker( + channelName: string, + checker: ((sessionId: string) => boolean) | undefined, + ): void { + if (checker) { + this.sessionActivityCheckers.set(channelName, checker); + } else { + this.sessionActivityCheckers.delete(channelName); + } + } + + /** + * Subscribe to rotation retirements. Fires after the route is dropped and + * before the retired session is discarded, so owners can purge per-session + * state and notify the chat. Returns an unsubscribe function. + */ + onSessionRotated( + listener: (sessionId: string, target: SessionTarget | undefined) => void, + ): () => void { + this.rotationListeners.add(listener); + return () => { + this.rotationListeners.delete(listener); + }; + } + + /** + * Retire a route so the next resolve starts a fresh session on it. The + * outgoing session goes through the same retirement machinery as the other + * paths: owners purge their per-session state, then the bridge discards it — + * otherwise nothing would ever reclaim a rotated session. + */ + private rotateRoute( + key: string, + sessionId: string, + channelName: string, + target: SessionTarget | undefined, + ): void { + this.invalidateRouteOperation(key); + this.deleteByKey(key); + this.tombstoneSuspendedKey(key); + // Persist the retirement now: if the successor creation fails before its + // own persist, a restart must not restore the stale route and re-fire the + // whole rotation (a second notice and discard for what was one rotation). + // Mid-restore this write is suspended like any other and only becomes + // durable with the restore's end flush, so a crash in that window + // re-fires the rotation once on the next start. + this.persist(); + process.stderr.write( + `[SessionRouter] Rotated session for key ${sanitizeLogText(key, 256)} on ${sanitizeLogText(channelName, 128)}: ` + + `${sanitizeLogText(sessionId, 128)} reached its configured limit; ` + + `starting a new session.\n`, + ); + for (const listener of this.rotationListeners) { + try { + listener(sessionId, target); + } catch (error) { + process.stderr.write( + `[SessionRouter] Rotation listener error for session ${sanitizeLogText(sessionId, 128)}: ` + + `${sanitizeLogText(error instanceof Error ? error.message : String(error), 512)}\n`, + ); + } + } + this.discardRotatedSession(sessionId); + } + + private discardRotatedSession(sessionId: string): void { + if ([...this.toSession.values()].includes(sessionId)) return; + try { + void this.bridge.discardSession?.(sessionId).catch(() => undefined); + } catch { + // Best-effort cleanup must not fail the incoming message. + } + } + + private isSessionActive(channelName: string, sessionId: string): boolean { + const checker = this.sessionActivityCheckers.get(channelName); + return checker ? checker(sessionId) : false; + } + + private leaseSession(sessionId: string): void { + const delta = this.rotationDeltas.get(sessionId); + if (delta) { + delta.leases += 1; + return; + } + this.sessionRoutingLeases.set( + sessionId, + (this.sessionRoutingLeases.get(sessionId) ?? 0) + 1, + ); + } + + private hasRoutingLease(sessionId: string): boolean { + return (this.sessionRoutingLeases.get(sessionId) ?? 0) > 0; + } + + /** + * Release the routing lease resolve() took when handing out `sessionId`. + * Call exactly once per successful resolve: when the routed message is + * enqueued as a turn, or when it settles without one (buffered, handled as + * a shell command). Rotation defers while a lease is held, so a session is + * never retired out from under a message that resolved it but has not yet + * registered its turn. + */ + releaseRoutingLease(sessionId: string): void { + const delta = this.rotationDeltas.get(sessionId); + if (delta) { + delta.leases -= 1; + return; + } + const leases = this.sessionRoutingLeases.get(sessionId) ?? 0; + if (leases <= 1) { + this.sessionRoutingLeases.delete(sessionId); + } else { + this.sessionRoutingLeases.set(sessionId, leases - 1); + } + } + + /** + * Counting turns costs a persist per message, and only the turns bound ever + * reads the counter, so channels without maxTurns opt out of both. + */ + private countTurn(channelName: string, sessionId: string): void { + const rotation = this.channelRotations.get(channelName); + if (rotation?.maxTurns === undefined) return; + const delta = this.rotationDeltas.get(sessionId); + if (delta) { + delta.turns += 1; + this.persist(); + return; + } + this.toTurns.set(sessionId, (this.toTurns.get(sessionId) ?? 0) + 1); + this.persist(); + } + + /** + * Reverse countTurn for a routed message that settled without starting a + * turn (buffered in collect mode, dropped loop firing, shell command): + * the bound counts turns actually started. Zero is an absent counter, so + * a session whose only count was given back rotates exactly on schedule. + */ + uncountTurn(channelName: string, sessionId: string): void { + const rotation = this.channelRotations.get(channelName); + if (rotation?.maxTurns === undefined) return; + const delta = this.rotationDeltas.get(sessionId); + if (delta) { + delta.turns -= 1; + this.persist(); + return; + } + const turns = this.toTurns.get(sessionId); + if (turns === undefined) return; + if (turns <= 1) { + this.toTurns.delete(sessionId); + } else { + this.toTurns.set(sessionId, turns - 1); + } + this.persist(); + } + private routingKey( channelName: string, senderId: string, @@ -152,9 +400,40 @@ export class SessionRouter { }; let failedWaits = 0; for (;;) { - const existing = this.toSession.get(key); + let existing = this.toSession.get(key); + // Checked before both the live-reuse and the lazy-reload path so a route + // cannot dodge its bound by having been evicted from memory. Skipped + // while a creation is in flight on this key: invalidating it would fail + // the concurrent message instead of rotating it, and the bound is just as + // well enforced on the next one. Deferred while the outgoing session + // still has a turn running or queued, or while a routed message has not + // settled yet (a resolve whose turn is registered only in a later + // microtask): retiring it then would discard a session a concurrent + // message is about to prompt, or auto-cancel its pending approvals and + // drop its late output — so the bound is enforced on the next message + // instead. Each deferred message enqueues a turn of its own, so a route + // whose messages never pause defers until the first idle gap; that + // limit is documented in the Session Rotation docs. + if ( + existing && + !this.creatingSessions.has(key) && + !this.isSessionActive(channelName, existing) && + !this.hasRoutingLease(existing) && + this.shouldRotate(channelName, existing) + ) { + this.rotateRoute(key, existing, channelName, { + channelName: input.channelName, + senderId: input.senderId, + chatId: input.chatId, + threadId: input.threadId, + isGroup: input.isGroup, + }); + existing = undefined; + } if (existing && this.isLive(existing)) { this.promoteTargetToGroup(existing, isGroup); + this.countTurn(channelName, existing); + this.leaseSession(existing); return existing; } @@ -168,10 +447,29 @@ export class SessionRouter { this.scheduleDiscardInvalidatedSession(sessionId, creating); throw error; } + // A restore reservation can hand back a session whose persisted + // counters are already at the bound. No lease or count is taken + // yet and the reservation is already cleared, so re-entering the + // loop routes this message through the rotation gate above. + if (this.shouldRotate(channelName, sessionId)) { + continue; + } this.promoteTargetToGroup(sessionId, isGroup); + this.countTurn(channelName, sessionId); + this.leaseSession(sessionId); return sessionId; } catch (error) { if (creating.invalidationError) { + // A rotation or reload can hand the key to a successor operation + // while this waiter is parked; route against the successor + // instead of failing the message. Without a successor the + // invalidation was a deliberate rejection (e.g. removeSession), + // which stays terminal. + if (this.creatingSessions.has(key) || this.toSession.has(key)) { + failedWaits++; + if (failedWaits > 3) throw creating.invalidationError; + continue; + } throw creating.invalidationError; } if (this.creatingSessions.get(key) === creating) { @@ -208,6 +506,7 @@ export class SessionRouter { throw error; } this.promoteTargetToGroup(sessionId, isGroup); + this.leaseSession(sessionId); return sessionId; } finally { if (this.creatingSessions.get(key) === operation) { @@ -259,6 +558,7 @@ export class SessionRouter { isGroup: input.isGroup, }); this.toCwd.set(sessionId, input.cwd); + this.seedRotationCounters(input.channelName, sessionId); this.liveSessionIds.add(sessionId); this.persist(); return sessionId; @@ -309,12 +609,29 @@ export class SessionRouter { } if (loadedSessionId !== savedSessionId) { const target = this.toTarget.get(savedSessionId); + // The reload is the same conversation under a new ID, so its age and + // turn count carry over — otherwise reloading would reset the bound. + const turns = this.toTurns.get(savedSessionId); + const startedAt = this.toStartedAt.get(savedSessionId); this.deleteByKey(key); this.toSession.set(key, loadedSessionId); if (target) this.toTarget.set(loadedSessionId, target); this.toCwd.set(loadedSessionId, savedCwd); - this.persist(); + if (turns !== undefined) this.toTurns.set(loadedSessionId, turns); + if (startedAt !== undefined) { + this.toStartedAt.set(loadedSessionId, startedAt); + } + // Age-only channels get no countTurn persist below, so this is + // their only writer of the new ID and carried counters; maxTurns + // channels persist again in countTurn immediately below, and this + // write would be fully subsumed by it. + if ( + this.channelRotations.get(input.channelName)?.maxTurns === undefined + ) { + this.persist(); + } } + this.countTurn(input.channelName, loadedSessionId); this.liveSessionIds.add(loadedSessionId); return loadedSessionId; } catch (loadError) { @@ -344,6 +661,7 @@ export class SessionRouter { isGroup: input.isGroup, }); this.toCwd.set(replacement, input.cwd); + this.seedRotationCounters(input.channelName, replacement); this.liveSessionIds.add(replacement); this.persist(); process.stderr.write( @@ -421,6 +739,7 @@ export class SessionRouter { this.invalidateRouteOperation(key); const sessionId = this.deleteByKey(key); if (sessionId) removedIds.push(sessionId); + this.tombstoneSuspendedKey(key); } else if (scope === 'single') { return removedIds; } else { @@ -434,6 +753,7 @@ export class SessionRouter { this.invalidateRouteOperation(k); const sessionId = this.deleteByKey(k); if (sessionId) removedIds.push(sessionId); + this.tombstoneSuspendedKey(k); } } for (const [key, operation] of [...this.creatingSessions]) { @@ -442,6 +762,7 @@ export class SessionRouter { operation.target.senderId === senderId ) { this.invalidateRouteOperation(key); + this.tombstoneSuspendedKey(key); } } } @@ -456,6 +777,7 @@ export class SessionRouter { if (mappedSessionId === sessionId) { this.invalidateRouteOperation(key); this.toSession.delete(key); + this.tombstoneSuspendedKey(key); removed = true; } } @@ -465,6 +787,9 @@ export class SessionRouter { if (this.toCwd.delete(sessionId)) { removed = true; } + this.toTurns.delete(sessionId); + this.toStartedAt.delete(sessionId); + this.sessionRoutingLeases.delete(sessionId); this.liveSessionIds.delete(sessionId); if (!removed && this.sessionLoadWindows.size > 0) { for (const loadWindow of this.sessionLoadWindows) { @@ -495,6 +820,10 @@ export class SessionRouter { this.toSession.delete(key); this.toTarget.delete(sessionId); this.toCwd.delete(sessionId); + this.toTurns.delete(sessionId); + this.toStartedAt.delete(sessionId); + this.sessionRoutingLeases.delete(sessionId); + this.rotationDeltas.delete(sessionId); this.liveSessionIds.delete(sessionId); return sessionId; } @@ -538,6 +867,7 @@ export class SessionRouter { this.toSession.set(key, entry.sessionId); this.toTarget.set(entry.sessionId, entry.target); this.toCwd.set(entry.sessionId, entry.cwd); + this.restoreRotationState(entry.sessionId, entry); restored++; } if (persisted.dropped > 0) this.persist(); @@ -547,7 +877,11 @@ export class SessionRouter { /** * Restore session mappings from a previous bridge. * Called after bridge restart — attempts loadSession for each saved mapping. - * Failed loads are dropped (new session on next message). + * Failed loads are dropped (new session on next message). Overlapping + * restores (e.g. a reconnect READY mid cold-start restore) are safe: + * persistence stays suspended until the last one finishes, and rotation + * state already live in memory survives the later restore's reservation + * pass. */ async restoreSessions(): Promise<{ restored: number; @@ -563,89 +897,159 @@ export class SessionRouter { let changed = persisted.dropped > 0; const reservations = new Map< string, - { reservation: SessionReservation; operation: SessionOperation } + { + reservation: SessionReservation; + operation: SessionOperation; + liveSessionId?: string; + liveRotation?: { + turns?: number; + startedAt?: number; + leases?: number; + }; + } >(); for (const key of persisted.droppedKeys) { this.deleteByKey(key); } - // Reserve every persisted key up front so inbound messages during restart - // wait for restore instead of returning stale IDs or creating duplicates. - for (const key of Object.keys(entries)) { - this.deleteByKey(key); - const reservation = this.createSessionReservation(); - reservation.promise.catch(() => undefined); - const operation = this.createSessionOperation( - key, - entries[key]!.target, - () => reservation.promise, - ); - operation.promise.catch(() => undefined); - this.creatingSessions.set(key, operation); - reservations.set(key, { reservation, operation }); - } - - const loadWindow = this.beginSessionLoad(); + // Released waiters route (and persist) while this loop still restores + // later keys; suspend persistence so a store holding only the restored + // prefix cannot become durable. Restores can overlap (a reconnect READY + // can fire while a cold-start restore still runs), so suspension is a + // depth and only the last restore to finish flushes the whole store. + this.persistSuspendDepth++; + let completed = false; try { - for (const [key, entry] of Object.entries(entries)) { - const reserved = reservations.get(key); - if (!reserved) continue; - const { reservation, operation } = reserved; - try { - this.assertOperationCurrent(operation); - const options = this.sessionOptions(entry.target.channelName); - const sessionId = await this.bridge.loadSession( - entry.sessionId, - entry.cwd, - options, - operation, - ); + // Reserve every persisted key up front so inbound messages during restart + // wait for restore instead of returning stale IDs or creating duplicates. + for (const key of Object.keys(entries)) { + // A route removed while persistence stays suspended only reaches + // disk at the last restore's flush; until then this restore reads + // the stale pre-deletion snapshot and must not resurrect the key. + if (this.suspendedDeletionKeys.has(key)) continue; + const liveSessionId = this.toSession.get(key); + // A route already back in memory can have routed messages newer than + // the persisted snapshot (persists stay suspended across the restore + // window); carry its rotation state across the wipe so the reload + // below cannot rewind it. + const liveRotation = liveSessionId + ? { + turns: this.toTurns.get(liveSessionId), + startedAt: this.toStartedAt.get(liveSessionId), + leases: this.sessionRoutingLeases.get(liveSessionId), + } + : undefined; + this.deleteByKey(key); + if (liveSessionId) { + this.rotationDeltas.set(liveSessionId, { turns: 0, leases: 0 }); + } + const reservation = this.createSessionReservation(); + reservation.promise.catch(() => undefined); + const operation = this.createSessionOperation( + key, + entries[key]!.target, + () => reservation.promise, + ); + operation.promise.catch(() => undefined); + this.creatingSessions.set(key, operation); + reservations.set(key, { + reservation, + operation, + liveSessionId, + liveRotation, + }); + } + + const loadWindow = this.beginSessionLoad(); + try { + for (const [key, entry] of Object.entries(entries)) { + const reserved = reservations.get(key); + if (!reserved) continue; + const { reservation, operation } = reserved; try { this.assertOperationCurrent(operation); - } catch (error) { - this.scheduleDiscardInvalidatedSession(sessionId, operation); - throw error; - } - if (typeof sessionId !== 'string' || sessionId.length === 0) { - throw new Error('Invalid restored session ID'); - } - if (loadWindow.delete(sessionId)) { - throw new Error('Restored session died before routing completed'); - } - this.toSession.set(key, sessionId); - this.toTarget.set(sessionId, entry.target); - this.toCwd.set(sessionId, entry.cwd); - reservation.resolve(sessionId); - if (sessionId !== entry.sessionId) { + const options = this.sessionOptions(entry.target.channelName); + const sessionId = await this.bridge.loadSession( + entry.sessionId, + entry.cwd, + options, + operation, + ); + try { + this.assertOperationCurrent(operation); + } catch (error) { + this.scheduleDiscardInvalidatedSession(sessionId, operation); + throw error; + } + if (typeof sessionId !== 'string' || sessionId.length === 0) { + throw new Error('Invalid restored session ID'); + } + if (loadWindow.delete(sessionId)) { + throw new Error('Restored session died before routing completed'); + } + this.toSession.set(key, sessionId); + this.toTarget.set(sessionId, entry.target); + this.toCwd.set(sessionId, entry.cwd); + this.restoreRotationState(sessionId, entry); + if (reserved.liveRotation && reserved.liveSessionId) { + this.carryLiveRotationState( + sessionId, + reserved.liveRotation, + reserved.liveSessionId, + ); + } + reservation.resolve(sessionId); + if (sessionId !== entry.sessionId) { + changed = true; + } + restored++; + } catch (err) { + const reason = err instanceof Error ? err.message : String(err); + process.stderr.write( + `[SessionRouter] Failed to restore session ${sanitizeLogText(entry.sessionId, 128)} for key ${sanitizeLogText(key, 256)}: ${sanitizeLogText(reason, 512)}\n`, + ); + reservation.reject( + new Error('Session restore failed', { cause: err }), + ); + if (reserved.liveSessionId) { + this.rotationDeltas.delete(reserved.liveSessionId); + } + // The drop must reach disk even when an overlapping restore + // flushes last: the last finisher only reads its own `changed`. + this.persistRequestedWhileSuspended = true; + // Session can't be loaded — will create fresh on next message + failed++; changed = true; + } finally { + if (this.creatingSessions.get(key) === operation) { + this.creatingSessions.delete(key); + } + this.releaseRouteToken(key, operation); } - restored++; - } catch (err) { - const reason = err instanceof Error ? err.message : String(err); - process.stderr.write( - `[SessionRouter] Failed to restore session ${sanitizeLogText(entry.sessionId, 128)} for key ${sanitizeLogText(key, 256)}: ${sanitizeLogText(reason, 512)}\n`, - ); - reservation.reject( - new Error('Session restore failed', { cause: err }), - ); - // Session can't be loaded — will create fresh on next message - failed++; - changed = true; - } finally { - if (this.creatingSessions.get(key) === operation) { - this.creatingSessions.delete(key); - } - this.releaseRouteToken(key, operation); } + } finally { + this.endSessionLoad(loadWindow); } + completed = true; } finally { - this.endSessionLoad(loadWindow); - } - - // Update persist file to only include successfully restored sessions - if (changed && restoreGeneration === this.lifecycleGeneration) { - this.persist(); + this.persistSuspendDepth--; + // Update persist file to only include successfully restored sessions. + // A persist requested while suspended counts: the mid-restore write + // was skipped to keep a truncated store from becoming durable. An + // earlier finisher leaves the suspension and the pending flush to the + // restore still running — flushing sooner would persist the other + // restore's partial prefix. + if ( + completed && + this.persistSuspendDepth === 0 && + (changed || this.persistRequestedWhileSuspended) && + restoreGeneration === this.lifecycleGeneration + ) { + this.persistRequestedWhileSuspended = false; + this.suspendedDeletionKeys.clear(); + this.persist(); + } } return { restored, failed }; @@ -659,10 +1063,15 @@ export class SessionRouter { this.toSession.clear(); this.toTarget.clear(); this.toCwd.clear(); + this.toTurns.clear(); + this.toStartedAt.clear(); + this.sessionRoutingLeases.clear(); + this.rotationDeltas.clear(); this.creatingSessions.clear(); this.sessionLoadWindows.clear(); this.liveSessionIds.clear(); this.routeTokens.clear(); + this.suspendedDeletionKeys.clear(); } /** Clear in-memory state and delete persist file. Used on clean shutdown. */ @@ -728,6 +1137,65 @@ export class SessionRouter { return { entries, dropped: droppedKeys.length, droppedKeys }; } + /** + * Seed rotation counters for a freshly created session. The creating + * message is turn one, so a turns-bound channel starts at 1 and the + * creation persist doubles as its count — no second write per new session. + */ + private seedRotationCounters(channelName: string, sessionId: string): void { + const rotation = this.channelRotations.get(channelName); + if (!rotation) return; + if (rotation.maxTurns !== undefined) this.toTurns.set(sessionId, 1); + if (rotation.maxAgeHours !== undefined) { + this.toStartedAt.set(sessionId, Date.now()); + } + } + + /** Carry a persisted entry's rotation counters back into memory. */ + private restoreRotationState(sessionId: string, entry: PersistedEntry): void { + if (entry.turns !== undefined) this.toTurns.set(sessionId, entry.turns); + if (entry.startedAt !== undefined) { + this.toStartedAt.set(sessionId, entry.startedAt); + } + } + + /** + * Re-apply the rotation state a route had live in memory when a restore + * reserved it: routed messages made it newer than the persisted snapshot + * the restore reads, so it wins over restoreRotationState's seed. Writes + * that landed between the reservation's wipe and this carry waited in + * rotationDeltas; net them so a mid-restore release or uncount cannot + * resurface as a phantom lease or count. Zero turns stay absent, matching + * uncountTurn's representation. + */ + private carryLiveRotationState( + sessionId: string, + liveRotation: { + turns?: number; + startedAt?: number; + leases?: number; + }, + wipedSessionId: string, + ): void { + const delta = this.rotationDeltas.get(wipedSessionId); + this.rotationDeltas.delete(wipedSessionId); + const turns = (liveRotation.turns ?? 0) + (delta?.turns ?? 0); + if (turns > 0) { + this.toTurns.set(sessionId, turns); + } else { + this.toTurns.delete(sessionId); + } + if (liveRotation.startedAt !== undefined) { + this.toStartedAt.set(sessionId, liveRotation.startedAt); + } + const leases = (liveRotation.leases ?? 0) + (delta?.leases ?? 0); + if (leases > 0) { + this.sessionRoutingLeases.set(sessionId, leases); + } else { + this.sessionRoutingLeases.delete(sessionId); + } + } + private isPersistedEntry(value: unknown): value is PersistedEntry { if (typeof value !== 'object' || value === null) return false; const entry = value as Record; @@ -739,6 +1207,9 @@ export class SessionRouter { entry['sessionId'].length > 0 && typeof entry['cwd'] === 'string' && entry['cwd'].length > 0 && + (entry['turns'] === undefined || isValidTurnCount(entry['turns'])) && + (entry['startedAt'] === undefined || + isValidRotationBound(entry['startedAt'])) && typeof typedTarget['channelName'] === 'string' && typeof typedTarget['senderId'] === 'string' && typeof typedTarget['chatId'] === 'string' && @@ -749,17 +1220,39 @@ export class SessionRouter { ); } + /** + * Record a route removed while persistence is suspended mid-restore, and + * ask the last restore's flush to write the removal out. Without the + * record an overlapping restore reading the stale snapshot would re-add + * the route and the flush would persist it again. + */ + private tombstoneSuspendedKey(key: string): void { + if (this.persistSuspendDepth === 0) return; + this.suspendedDeletionKeys.add(key); + this.persist(); + } + private persist(): void { if (!this.persistPath) return; + if (this.persistSuspendDepth > 0) { + // Mid-restore the store holds only the restored prefix; let + // restoreSessions() flush once the store is whole again. + this.persistRequestedWhileSuspended = true; + return; + } const data: Record = {}; for (const [key, sessionId] of this.toSession) { const target = this.toTarget.get(sessionId); if (!target) continue; + const turns = this.toTurns.get(sessionId); + const startedAt = this.toStartedAt.get(sessionId); data[key] = { sessionId, target, cwd: this.toCwd.get(sessionId) ?? this.defaultCwd, + ...(turns !== undefined ? { turns } : {}), + ...(startedAt !== undefined ? { startedAt } : {}), }; } @@ -943,3 +1436,28 @@ export class SessionRouter { this.sessionLoadWindows.delete(loadWindow); } } + +/** + * The single definition of a valid rotation bound, shared by parse-time + * validation (fail loudly) and the router (defensively drop). A bound is only + * meaningful when positive and finite; anything else (0, negative, NaN, a stray + * string from a hand-edited config) is not a bound. + */ +export function isValidRotationBound(value: unknown): value is number { + return typeof value === 'number' && Number.isFinite(value) && value > 0; +} + +/** + * The single definition of a valid turn-count value, shared by parse-time + * validation, the settings store, and the persisted-entry gate: turns are + * whole messages, so a bound (or a restored counter) must be a positive + * integer. Fractional values would rotate from the second message on (0 < x + * <= 1) or silently act as their ceiling (x > 1). + */ +export function isValidTurnCount(value: unknown): value is number { + return typeof value === 'number' && Number.isInteger(value) && value > 0; +} + +function normalizeRotationBound(value: unknown): number | undefined { + return isValidRotationBound(value) ? value : undefined; +} diff --git a/packages/channels/base/src/index.ts b/packages/channels/base/src/index.ts index 8bdf47f9bd5..3b4531f1b9d 100644 --- a/packages/channels/base/src/index.ts +++ b/packages/channels/base/src/index.ts @@ -89,7 +89,11 @@ export { DmGate } from './DmGate.js'; export type { DmCheckResult } from './DmGate.js'; export { SenderGate } from './SenderGate.js'; export type { SenderCheckResult } from './SenderGate.js'; -export { SessionRouter } from './SessionRouter.js'; +export { + SessionRouter, + isValidRotationBound, + isValidTurnCount, +} from './SessionRouter.js'; export { sanitizeSenderName, sanitizePromptText, @@ -145,6 +149,7 @@ export type { ObservedChannelContactGraph, SanitizedToolCallEvent, SenderPolicy, + SessionRotationConfig, SessionScope, SessionTarget, UserInputPresentationResult, diff --git a/packages/channels/base/src/types.ts b/packages/channels/base/src/types.ts index e2537ce0f54..54bd626f9bc 100644 --- a/packages/channels/base/src/types.ts +++ b/packages/channels/base/src/types.ts @@ -52,6 +52,20 @@ export interface BlockStreamingCoalesceConfig { idleMs?: number; } +/** + * Bounds on how long a single routed session keeps accumulating context before + * the router starts a fresh one. Without a bound a long-lived route (a busy + * group thread, say) grows monotonically until it hits provider context limits, + * at which point every later message on that route fails while the rest of the + * channel stays healthy. + */ +export interface SessionRotationConfig { + /** Route to a new session once this many messages have used the current one. */ + maxTurns?: number; + /** Route to a new session once the current one is older than this many hours. */ + maxAgeHours?: number; +} + export interface ChannelConfig { type: ChannelType; token: string; @@ -60,6 +74,8 @@ export interface ChannelConfig { senderPolicy: SenderPolicy; allowedUsers: string[]; sessionScope: SessionScope; + /** Auto-rotation bounds for sessions on this channel. Unset means never rotate. */ + sessionRotation?: SessionRotationConfig; cwd: string; approvalMode?: string; instructions?: string; diff --git a/packages/channels/telegram/src/TelegramAdapter.test.ts b/packages/channels/telegram/src/TelegramAdapter.test.ts index 8fba92ede13..62bf1eb4c4d 100644 --- a/packages/channels/telegram/src/TelegramAdapter.test.ts +++ b/packages/channels/telegram/src/TelegramAdapter.test.ts @@ -101,7 +101,12 @@ function createChannel( { ...config, ...configOverrides }, {} as ChannelAgentBridge, { - router: router as never, + router: { + setChannelRotation: () => {}, + setSessionActivityChecker: () => {}, + onSessionRotated: () => () => {}, + ...(router as Record), + } as never, }, ); } @@ -164,6 +169,19 @@ describe('TelegramChannel', () => { expect(channel.supportsProactiveSend()).toBe(true); }); + it('wires configured session rotation into the router', () => { + const setChannelRotation = vi.fn(); + createChannel( + { sessionRotation: { maxTurns: 5, maxAgeHours: 2 } }, + { setChannelRotation }, + ); + + expect(setChannelRotation).toHaveBeenCalledWith('telegram', { + maxTurns: 5, + maxAgeHours: 2, + }); + }); + it('clears active typing intervals on disconnect', () => { const clearIntervalSpy = vi.spyOn(global, 'clearInterval'); const channel = createChannel(); diff --git a/packages/cli/src/commands/channel/config-utils.test.ts b/packages/cli/src/commands/channel/config-utils.test.ts index 3ffa4cbd1eb..b6b5bfdbbe4 100644 --- a/packages/cli/src/commands/channel/config-utils.test.ts +++ b/packages/cli/src/commands/channel/config-utils.test.ts @@ -99,6 +99,97 @@ describe('parseChannelConfig', () => { ); }); + it('parses sessionRotation bounds', async () => { + const result = await parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: { maxTurns: 200, maxAgeHours: 24 }, + }); + + expect(result['sessionRotation']).toEqual({ + maxTurns: 200, + maxAgeHours: 24, + }); + }); + + it('leaves sessionRotation unset when omitted', async () => { + const result = await parseChannelConfig('bot', { + type: 'telegram', + token: 't', + }); + + expect(result['sessionRotation']).toBeUndefined(); + }); + + it('throws when a sessionRotation bound is not a positive number', async () => { + await expect( + parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: { maxTurns: 0 }, + }), + ).rejects.toThrow('"sessionRotation.maxTurns" must be a positive integer'); + await expect( + parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: { maxAgeHours: -1 }, + }), + ).rejects.toThrow( + '"sessionRotation.maxAgeHours" must be a positive number', + ); + }); + + it('throws when maxTurns is fractional', async () => { + await expect( + parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: { maxTurns: 0.5 }, + }), + ).rejects.toThrow('"sessionRotation.maxTurns" must be a positive integer'); + }); + + it('accepts a fractional maxAgeHours', async () => { + const result = await parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: { maxAgeHours: 0.5 }, + }); + + expect(result['sessionRotation']).toEqual({ maxAgeHours: 0.5 }); + }); + + it('throws on unknown sessionRotation keys instead of dropping them', async () => { + await expect( + parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: { maxTurn: 200 }, + }), + ).rejects.toThrow('"sessionRotation.maxTurn"'); + }); + + it('treats an explicit null sessionRotation as unset', async () => { + const result = await parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: null, + }); + + expect(result['sessionRotation']).toBeUndefined(); + }); + + it('throws when sessionRotation is not an object', async () => { + await expect( + parseChannelConfig('bot', { + type: 'telegram', + token: 't', + sessionRotation: 'daily', + }), + ).rejects.toThrow(/sessionRotation/); + }); + it('throws when plugin-required fields are missing', async () => { await expect( parseChannelConfig('bot', { type: 'telegram' }), diff --git a/packages/cli/src/commands/channel/config-utils.ts b/packages/cli/src/commands/channel/config-utils.ts index 9ffe908120b..ad8ded2ca26 100644 --- a/packages/cli/src/commands/channel/config-utils.ts +++ b/packages/cli/src/commands/channel/config-utils.ts @@ -4,6 +4,10 @@ import type { ChannelWebhookSourceConfig, ChannelWebhookTargetConfig, } from '@qwen-code/channel-base'; +import { + isValidRotationBound, + isValidTurnCount, +} from '@qwen-code/channel-base'; import { resolveChannelCwd } from './channel-cwd.js'; import { getPlugin, supportedTypes } from './channel-registry.js'; @@ -177,6 +181,43 @@ function optionalBooleanField( return value; } +function parseSessionRotationConfig( + channelName: string, + rawConfig: Record, +): ChannelConfig['sessionRotation'] { + const raw = rawConfig['sessionRotation']; + if (raw === undefined || raw === null) return undefined; + const parsed = requireObjectField(channelName, 'sessionRotation', raw); + + // Reject unknown keys loudly: a typo'd bound (say "maxTurn") must not + // silently disable rotation — the exact failure mode rotation prevents. + for (const key of Object.keys(parsed)) { + if (key !== 'maxTurns' && key !== 'maxAgeHours') { + throw new Error( + `Channel "${channelName}" field "sessionRotation.${key}" is not a valid sessionRotation key.`, + ); + } + } + + const maxTurns = parsed['maxTurns']; + if (maxTurns !== undefined && !isValidTurnCount(maxTurns)) { + throw new Error( + `Channel "${channelName}" field "sessionRotation.maxTurns" must be a positive integer.`, + ); + } + const maxAgeHours = parsed['maxAgeHours']; + if (maxAgeHours !== undefined && !isValidRotationBound(maxAgeHours)) { + throw new Error( + `Channel "${channelName}" field "sessionRotation.maxAgeHours" must be a positive number.`, + ); + } + if (maxTurns === undefined && maxAgeHours === undefined) return undefined; + return { + ...(maxTurns !== undefined ? { maxTurns } : {}), + ...(maxAgeHours !== undefined ? { maxAgeHours } : {}), + }; +} + function requireObjectField( channelName: string, path: string, @@ -466,6 +507,7 @@ export async function parseChannelConfig( (rawConfig['sessionScope'] as ChannelConfig['sessionScope']) || plugin?.defaultSessionScope || 'user', + sessionRotation: parseSessionRotationConfig(name, rawConfig), cwd: resolveChannelCwd(rawConfig['cwd'] as string | undefined, defaultCwd), approvalMode: parseApprovalModeConfig(name, rawConfig), instructions: rawConfig['instructions'] as string | undefined, diff --git a/packages/cli/src/serve/channel-settings-store.test.ts b/packages/cli/src/serve/channel-settings-store.test.ts index 32764641e17..871d1d6aadb 100644 --- a/packages/cli/src/serve/channel-settings-store.test.ts +++ b/packages/cli/src/serve/channel-settings-store.test.ts @@ -452,6 +452,72 @@ describe('WorkspaceChannelSettingsStore', () => { clientSecret: { operation: 'replace', value: 'secret' } as const, }, }, + { + label: 'sessionRotation with a non-positive bound', + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: { maxTurns: 0 }, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }, + { + label: 'sessionRotation with an unknown bound', + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: { maxTurns: 10, maxMessages: 5 }, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }, + { + label: 'sessionRotation with a fractional maxTurns', + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: { maxTurns: 2.5 }, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }, + { + label: 'sessionRotation not an object', + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: 'daily', + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }, + { + label: 'sessionRotation with a zero maxAgeHours', + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: { maxAgeHours: 0 }, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }, + { + label: 'sessionRotation with a non-numeric maxAgeHours', + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: { maxAgeHours: 'daily' }, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }, ])('rejects $label without writing', async ({ config, secrets }) => { const store = new WorkspaceChannelSettingsStore(workspace); const before = fs.readFileSync(settingsPath, 'utf8'); @@ -485,6 +551,7 @@ describe('WorkspaceChannelSettingsStore', () => { groupHistoryLimit: 25, blockStreaming: 'on', identity: { id: 'ops', displayName: 'Ops' }, + sessionRotation: { maxTurns: 200, maxAgeHours: 24 }, }, secrets: { clientSecret: { @@ -507,6 +574,45 @@ describe('WorkspaceChannelSettingsStore', () => { groupHistoryLimit: 25, blockStreaming: 'on', identity: { id: 'ops', displayName: 'Ops' }, + sessionRotation: { maxTurns: 200, maxAgeHours: 24 }, + }); + }); + + it('accepts an explicit null sessionRotation as unset', async () => { + const store = new WorkspaceChannelSettingsStore(workspace); + + const next = await store.upsert('bot', { + expectedRevision: store.snapshot().revision, + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: null, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }); + + expect(next.channels['bot']!['sessionRotation']).toBeNull(); + }); + + it('accepts a fractional maxAgeHours like the config parser', async () => { + const store = new WorkspaceChannelSettingsStore(workspace); + + const next = await store.upsert('bot', { + expectedRevision: store.snapshot().revision, + config: { + type: 'management-validation-test', + clientId: 'client-id', + sessionRotation: { maxAgeHours: 0.5 }, + }, + secrets: { + clientSecret: { operation: 'replace', value: 'secret' } as const, + }, + }); + + expect(next.channels['bot']!['sessionRotation']).toEqual({ + maxAgeHours: 0.5, }); }); diff --git a/packages/cli/src/serve/channel-settings-store.ts b/packages/cli/src/serve/channel-settings-store.ts index 6d80ea1afe2..514c332eac7 100644 --- a/packages/cli/src/serve/channel-settings-store.ts +++ b/packages/cli/src/serve/channel-settings-store.ts @@ -7,6 +7,10 @@ import { createHash } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { ChannelConfigFieldDescriptor } from '@qwen-code/channel-base'; +import { + isValidRotationBound, + isValidTurnCount, +} from '@qwen-code/channel-base'; import { getPlugin, UNSAFE_OBJECT_KEYS, @@ -191,6 +195,26 @@ function assertSharedField(key: string, value: unknown): boolean { assertNumberRecord(key, value, new Set(['idleMs'])); return true; } + if (key === 'sessionRotation') { + // The config parser treats an explicit null as "unset"; agree here so a + // hand-cleared settings.json survives a later full-replace upsert. + if (value === null) return true; + if (!isRecord(value)) { + throw invalidConfig(`Channel field "${key}" must be an object.`); + } + for (const [nestedKey, nestedValue] of Object.entries(value)) { + const validBound = + nestedKey === 'maxTurns' + ? isValidTurnCount(nestedValue) + : nestedKey === 'maxAgeHours' + ? isValidRotationBound(nestedValue) + : false; + if (!validBound) { + throw invalidConfig(`Channel field "${key}.${nestedKey}" is invalid.`); + } + } + return true; + } if (key === 'memoryScope') { if (!isRecord(value)) { throw invalidConfig(`Channel field "${key}" must be an object.`);