From ea33aac64868e932efaaf8e518bd7ea928ce018b Mon Sep 17 00:00:00 2001 From: jesse-merhi <79823012+jesse-merhi@users.noreply.github.com> Date: Mon, 22 Jun 2026 22:15:42 +1000 Subject: [PATCH] fix(gateway): persist webchat dispatch failures --- src/gateway/server-methods/chat.ts | 73 ++++++++++++++++++- .../server.chat.gateway-server-chat.test.ts | 67 +++++++++++++++++ 2 files changed, 136 insertions(+), 4 deletions(-) diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 7b3072b7ee22..f64d09377852 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -154,6 +154,7 @@ import { ADMIN_SCOPE } from "../method-scopes.js"; import { chatAbortMarkerTimestampMs, type ChatRunTiming } from "../server-chat-state.js"; import { getMaxChatHistoryMessagesBytes, MAX_PAYLOAD_BYTES } from "../server-constants.js"; import { resolveSessionHistoryTailReadOptions } from "../session-history-state.js"; +import { persistGatewaySessionLifecycleEvent } from "../session-lifecycle-state.js"; import { readSessionTranscriptIndex } from "../session-transcript-index.fs.js"; import { capArrayByJsonBytes, @@ -3746,6 +3747,14 @@ export const chatHandlers: GatewayRequestHandlers = { const deliveredReplies: Array<{ payload: ReplyPayload; kind: "block" | "final" }> = []; let appendedWebchatAgentMedia = false; let agentRunStarted = false; + let pendingDispatchLifecycleError: + | { + endedAt: number; + error: string; + sessionId: string; + startedAt: number; + } + | undefined; const userTurnRecorder: UserTurnTranscriptRecorder = createUserTurnTranscriptRecorder({ input: baseUserTurnInput, resolveInput: () => userTurnInputPromise, @@ -4957,6 +4966,7 @@ export const chatHandlers: GatewayRequestHandlers = { ); }) .catch(async (err: unknown) => { + const errorMessage = String(err); const emitAfterError = userTurnRecorder.hasPersisted() || userTurnRecorder.isBlocked() ? Promise.resolve() @@ -4966,7 +4976,19 @@ export const chatHandlers: GatewayRequestHandlers = { `webchat user transcript update failed after error: ${formatForLog(transcriptErr)}`, ); }); - const error = errorShape(ErrorCodes.UNAVAILABLE, String(err)); + if ( + !agentRunStarted && + !activeRunAbort.controller.signal.aborted && + !context.chatAbortedRuns.has(clientRunId) + ) { + pendingDispatchLifecycleError = { + endedAt: Date.now(), + error: errorMessage, + sessionId: activeRunAbort.entry?.sessionId ?? backingSessionId ?? clientRunId, + startedAt: activeRunAbort.entry?.startedAtMs ?? now, + }; + } + const error = errorShape(ErrorCodes.UNAVAILABLE, errorMessage); setGatewayDedupeEntry({ dedupe: context.dedupe, key: `chat:${clientRunId}`, @@ -4976,7 +4998,7 @@ export const chatHandlers: GatewayRequestHandlers = { payload: { runId: clientRunId, status: "error" as const, - summary: String(err), + summary: errorMessage, }, error, }, @@ -4986,14 +5008,57 @@ export const chatHandlers: GatewayRequestHandlers = { runId: clientRunId, sessionKey, agentId, - errorMessage: String(err), + errorMessage, }); }) - .finally(() => { + .finally(async () => { activeRunAbort.cleanup(); clearAgentRunContext(clientRunId, lifecycleGeneration); clearActiveChatSendDedupeRun(context.dedupe, activeChatSendDedupeKey, clientRunId); context.removeChatRun(clientRunId, clientRunId, sessionKey); + if (!pendingDispatchLifecycleError) { + return; + } + const hasActiveRun = hasTrackedActiveSessionRun({ + context, + requestedKey: rawSessionKey, + canonicalKey: sessionKey, + ...(sessionKey === "global" && agentId ? { agentId } : {}), + defaultAgentId: resolveDefaultAgentId(cfg), + }); + if (hasActiveRun) { + return; + } + const persisted = await persistGatewaySessionLifecycleEvent({ + sessionKey, + ...(sessionKey === "global" && agentId ? { agentId } : {}), + event: { + runId: clientRunId, + sessionId: pendingDispatchLifecycleError.sessionId, + lifecycleGeneration, + ts: pendingDispatchLifecycleError.endedAt, + data: { + phase: "error", + startedAt: pendingDispatchLifecycleError.startedAt, + endedAt: pendingDispatchLifecycleError.endedAt, + error: pendingDispatchLifecycleError.error, + }, + }, + }) + .then(() => true) + .catch((persistErr) => { + context.logGateway.warn( + `webchat session lifecycle persist failed after error: ${formatForLog(persistErr)}`, + ); + return false; + }); + if (persisted) { + emitSessionsChanged(context, { + sessionKey, + ...(agentId ? { agentId } : {}), + reason: "chat.dispatch-error", + }); + } }); } catch (err) { activeRunAbort.cleanup({ force: true }); diff --git a/src/gateway/server.chat.gateway-server-chat.test.ts b/src/gateway/server.chat.gateway-server-chat.test.ts index 4fd028286a2f..2a896816f365 100644 --- a/src/gateway/server.chat.gateway-server-chat.test.ts +++ b/src/gateway/server.chat.gateway-server-chat.test.ts @@ -692,6 +692,73 @@ describe("gateway server chat", () => { }); }); + test("marks a running webchat session failed when dispatch rejects before a reply", async () => { + await withMainSessionStore(async (dir) => { + await writeSessionStore({ + entries: { + main: { + sessionId: "sess-main", + sessionFile: path.join(dir, "sess-main.jsonl"), + updatedAt: 1_000, + status: "running", + startedAt: 900, + }, + }, + }); + const subscribeRes = await rpcReq(ws, "sessions.subscribe", {}); + expect(subscribeRes.ok).toBe(true); + dispatchInboundMessageMock.mockRejectedValueOnce(new Error("provider rejected request")); + + const errorPromise = onceMessage( + ws, + (o) => + o.type === "event" && + o.event === "chat" && + o.payload?.state === "error" && + o.payload?.runId === "idem-dispatch-error-1", + 8_000, + ); + const sessionChangedPromise = onceMessage( + ws, + (o) => + o.type === "event" && + o.event === "sessions.changed" && + o.payload?.reason === "chat.dispatch-error" && + o.payload?.sessionKey === "agent:main:main", + 8_000, + ); + const res = await rpcReq(ws, "chat.send", { + sessionKey: "main", + message: "run: pwd", + idempotencyKey: "idem-dispatch-error-1", + }); + expect(res.ok).toBe(true); + await errorPromise; + const sessionChanged = await sessionChangedPromise; + expectRecordFields(sessionChanged.payload, { + sessionId: "sess-main", + status: "failed", + hasActiveRun: false, + }); + + const sessionsRes = await rpcReq<{ sessions?: unknown[] }>(ws, "sessions.list", {}); + expect(sessionsRes.ok).toBe(true); + const session = sessionsRes.payload?.sessions?.find( + (row): row is Record => + Boolean(row) && + typeof row === "object" && + (row as { key?: unknown }).key === "agent:main:main", + ); + const actualSession = expectRecordFields(session, { + status: "failed", + hasActiveRun: false, + }); + expect(typeof actualSession.startedAt).toBe("number"); + expect(typeof actualSession.endedAt).toBe("number"); + expect(typeof actualSession.runtimeMs).toBe("number"); + }); + }); + test("chat.history hides assistant NO_REPLY-only entries", async () => { const historyMessages = await loadChatHistoryWithMessages(buildNoReplyHistoryFixture()); const textValues = collectHistoryTextValues(historyMessages);