fix(gateway): preserve chat retry guard after terminal state

scotthuang · scotthuang · commit 8f8dbab281a9 · 2026-05-29T04:27:12.000+08:00
diff --git a/src/gateway/chat-abort.ts b/src/gateway/chat-abort.ts
@@ -15,6 +15,12 @@ export type ChatAbortControllerEntry = {
   providerId?: string;
   authProviderId?: string;
   abortStopReason?: string;
+  /**
+   * Controls only the sessions.list active-run projection. Terminal lifecycle
+   * clears this before chat.send settles, while the entry stays as the retry
+   * idempotency guard until normal cleanup removes it.
+   */
+  projectSessionActive?: boolean;
   /**
    * Which RPC owns this registration. Absent (undefined) is treated as
    * `"chat-send"` so pre-existing callers that constructed entries without
diff --git a/src/gateway/server-chat.agent-events.test.ts b/src/gateway/server-chat.agent-events.test.ts
@@ -44,6 +44,7 @@ import {
   createSessionEventSubscriberRegistry,
   createSessionMessageSubscriberRegistry,
   createToolEventRecipientRegistry,
+  type AgentEventHandlerOptions,
 } from "./server-chat.js";
 import { loadGatewaySessionRow } from "./server-chat.load-gateway-session-row.runtime.js";
 import { loadSessionEntry } from "./session-utils.js";
@@ -79,14 +80,16 @@ describe("agent event handler", () => {
     resolveSessionKeyForRun?: (runId: string) => string | undefined;
     lifecycleErrorRetryGraceMs?: number;
     isChatSendRunActive?: (runId: string) => boolean;
+    clearTrackedActiveRun?: AgentEventHandlerOptions["clearTrackedActiveRun"];
   }) {
     const nowSpy =
       params?.now === undefined ? undefined : vi.spyOn(Date, "now").mockReturnValue(params.now);
     const broadcast = vi.fn();
     const broadcastToConnIds = vi.fn();
     const nodeSendToSession = vi.fn();
     const clearAgentRunContext = vi.fn();
-    const clearTrackedActiveRun = vi.fn();
+    const clearTrackedActiveRun =
+      vi.fn<NonNullable<AgentEventHandlerOptions["clearTrackedActiveRun"]>>();
     const agentRunSeq = new Map<string, number>();
     const chatRunState = createChatRunState();
     const toolEventRecipients = createToolEventRecipientRegistry();
@@ -107,7 +110,7 @@ describe("agent event handler", () => {
       loadGatewaySessionRowForSnapshot: loadGatewaySessionRow,
       lifecycleErrorRetryGraceMs: params?.lifecycleErrorRetryGraceMs,
       isChatSendRunActive: params?.isChatSendRunActive,
-      clearTrackedActiveRun,
+      clearTrackedActiveRun: params?.clearTrackedActiveRun ?? clearTrackedActiveRun,
     });
 
     return {
@@ -1895,6 +1898,43 @@ describe("agent event handler", () => {
     );
   });
 
+  it("keeps chat send retry guards while hiding terminal session projection", () => {
+    const trackedActiveRuns = new Map<
+      string,
+      { sessionKey: string; projectSessionActive?: boolean }
+    >([["client-run", { sessionKey: "session-finished" }]]);
+    const { chatRunState, handler } = createHarness({
+      clearTrackedActiveRun: ({ runId, clientRunId, sessionKey }) => {
+        for (const candidateRunId of new Set([runId, clientRunId])) {
+          const entry = trackedActiveRuns.get(candidateRunId);
+          if (entry?.sessionKey === sessionKey) {
+            entry.projectSessionActive = false;
+          }
+        }
+      },
+    });
+    chatRunState.registry.add("provider-run", {
+      sessionKey: "session-finished",
+      clientRunId: "client-run",
+    });
+
+    handler({
+      runId: "provider-run",
+      seq: 2,
+      stream: "lifecycle",
+      ts: 1_800,
+      data: {
+        phase: "end",
+        startedAt: 900,
+        endedAt: 1_700,
+      },
+    });
+
+    const retryGuard = trackedActiveRuns.get("client-run");
+    expect(retryGuard).toBeDefined();
+    expect(retryGuard?.projectSessionActive).toBe(false);
+  });
+
   it("keeps aborted chat run markers through terminal lifecycle cleanup", () => {
     const { broadcast, chatRunState, handler } = createHarness();
     chatRunState.registry.add("run-aborted", {
diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts
@@ -666,7 +666,11 @@ function collectTrackedActiveSessionRunKeys(
     return keys;
   }
   for (const active of context.chatAbortControllers.values()) {
-    if (typeof active.sessionKey === "string" && active.sessionKey.trim()) {
+    if (
+      active.projectSessionActive !== false &&
+      typeof active.sessionKey === "string" &&
+      active.sessionKey.trim()
+    ) {
       keys.add(active.sessionKey);
     }
   }
diff --git a/src/gateway/server-runtime-subscriptions.ts b/src/gateway/server-runtime-subscriptions.ts
@@ -49,7 +49,7 @@ export function startGatewayEventSubscriptions(params: {
           for (const candidateRunId of new Set([runId, clientRunId])) {
             const entry = params.chatAbortControllers.get(candidateRunId);
             if (entry?.sessionKey === sessionKey) {
-              params.chatAbortControllers.delete(candidateRunId);
+              entry.projectSessionActive = false;
             }
           }
         },
diff --git a/src/gateway/server.sessions.list-changed.test.ts b/src/gateway/server.sessions.list-changed.test.ts
@@ -254,6 +254,42 @@ test("sessions.list marks sessions with active abortable runs", async () => {
   expect(session.hasActiveRun).toBe(true);
 });
 
+test("sessions.list ignores terminal abortable runs kept for retry guards", async () => {
+  await createSessionStoreDir();
+  await writeSessionStore({
+    entries: {
+      main: sessionStoreEntry("sess-main"),
+    },
+  });
+
+  const respond = vi.fn();
+  const sessionsHandlers = await getSessionsHandlers();
+  const { getRuntimeConfig } = await getGatewayConfigModule();
+  await sessionsHandlers["sessions.list"]({
+    req: {
+      type: "req",
+      id: "req-sessions-list-terminal-run",
+      method: "sessions.list",
+      params: {},
+    },
+    params: {},
+    respond,
+    client: null,
+    isWebchatConnect: () => false,
+    context: {
+      getRuntimeConfig,
+      loadGatewayModelCatalog: async () => [],
+      chatAbortControllers: new Map([
+        ["run-1", { sessionKey: "agent:main:main", projectSessionActive: false }],
+      ]),
+    } as never,
+  });
+
+  const payload = expectRespondPayload(respond);
+  const session = findSession(payload, "agent:main:main");
+  expect(session.hasActiveRun).toBe(false);
+});
+
 test("sessions.list yields before responding during bulk transcript hydration", async () => {
   const { dir } = await createSessionStoreDir();
   const entries: Record<string, ReturnType<typeof sessionStoreEntry>> = {};

Original file line number	Diff line number	Diff line change
`@@ -666,7 +666,11 @@ function collectTrackedActiveSessionRunKeys(`
`666`	`666`	`return keys;`
`667`	`667`	`}`
`668`	`668`	`for (const active of context.chatAbortControllers.values()) {`
`669`		`- if (typeof active.sessionKey === "string" && active.sessionKey.trim()) {`
	`669`	`+ if (`
	`670`	`+ active.projectSessionActive !== false &&`
	`671`	`+ typeof active.sessionKey === "string" &&`
	`672`	`+ active.sessionKey.trim()`
	`673`	`+ ) {`
`670`	`674`	`keys.add(active.sessionKey);`
`671`	`675`	`}`
`672`	`676`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ export function startGatewayEventSubscriptions(params: {`
`49`	`49`	`for (const candidateRunId of new Set([runId, clientRunId])) {`
`50`	`50`	`const entry = params.chatAbortControllers.get(candidateRunId);`
`51`	`51`	`if (entry?.sessionKey === sessionKey) {`
`52`		`- params.chatAbortControllers.delete(candidateRunId);`
	`52`	`+ entry.projectSessionActive = false;`
`53`	`53`	`}`
`54`	`54`	`}`
`55`	`55`	`},`