feat(channels): bound session lifetime with sessionRotation

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

packages/channels/base/src/AcpBridge.test.ts

@@ -113,6 +113,7 @@ type TestableAcpBridge = AcpBridge & {
   };
   knownSessionIds: Set<string>;
   sessionBindingTokens: Map<string, object | undefined>;
+  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<typeof vi.fn>;
+      loadSession: ReturnType<typeof vi.fn>;
+    };
+    // 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<typeof vi.fn>;
+    };
+    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',

packages/channels/base/src/AcpBridge.ts

@@ -98,6 +98,9 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
       timeout: ReturnType<typeof setTimeout>;
     }
   >();
+  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<string> {
     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<string> {
     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<T>(run: () => Promise<T>): Promise<T> {
+    return new Promise<T>((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<RequestPermissionResponse> {