diff --git a/src/gateway/server-methods/chat-send-dispatch-errors.test.ts b/src/gateway/server-methods/chat-send-dispatch-errors.test.ts new file mode 100644 index 000000000000..3f436d26cd42 --- /dev/null +++ b/src/gateway/server-methods/chat-send-dispatch-errors.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it, vi } from "vitest"; +import { createChatSendDispatchErrorLifecycle } from "./chat-send-dispatch-errors.js"; + +describe("createChatSendDispatchErrorLifecycle", () => { + it("terminalizes an admitted queued followup as successful despite later dispatch failure", async () => { + const broadcast = vi.fn(); + const cleanupAdmittedRun = vi.fn(); + const removeChatRun = vi.fn(); + const warn = vi.fn(); + const dedupe = new Map(); + const lifecycle = createChatSendDispatchErrorLifecycle({ + admission: { + activeRunAbort: { + cleanup: vi.fn(), + controller: new AbortController(), + entry: undefined, + registered: true, + } as never, + cleanupAdmittedRun, + lifecycleGeneration: 1, + restartSafeAdmission: undefined, + }, + context: { + agentRunSeq: new Map(), + broadcast, + chatAbortedRuns: new Set(), + dedupe, + getRuntimeConfig: () => ({}), + logGateway: { warn }, + nodeSendToSession: vi.fn(), + removeChatRun, + } as never, + isQueuedFollowupEnqueued: () => true, + persistUserTurnTranscript: vi.fn(), + session: { + agentId: "main", + backingSessionId: undefined, + cfg: {}, + clientRunId: "run-1", + now: 1, + rawSessionKey: "agent:main:main", + sessionKey: "agent:main:main", + }, + terminalizeRestartSafeAdmission: vi.fn(), + userTurnRecorder: { hasPersisted: () => false, isBlocked: () => false }, + }); + + await lifecycle.handleError(new Error("late failure")); + lifecycle.finalize(); + + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("dispatch failed after followup queue admission"), + ); + expect(dedupe.get("chat:run-1")).toMatchObject({ + ok: true, + payload: { runId: "run-1", status: "ok" }, + }); + expect(broadcast).toHaveBeenCalledWith( + "chat", + expect.objectContaining({ runId: "run-1", state: "final" }), + ); + expect(cleanupAdmittedRun).toHaveBeenCalledOnce(); + expect(removeChatRun).toHaveBeenCalledWith("run-1", "run-1", "agent:main:main"); + }); +}); diff --git a/src/gateway/server-methods/chat-send-dispatch-errors.ts b/src/gateway/server-methods/chat-send-dispatch-errors.ts new file mode 100644 index 000000000000..d69490407a56 --- /dev/null +++ b/src/gateway/server-methods/chat-send-dispatch-errors.ts @@ -0,0 +1,214 @@ +import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js"; +import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; +import { clearAgentRunContext } from "../../infra/agent-events.js"; +import { retainGatewayRootWorkAdmissionContinuation } from "../../process/gateway-work-admission.js"; +import type { UserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.js"; +import { persistGatewaySessionLifecycleEvent } from "../session-lifecycle-state.js"; +import { formatForLog } from "../ws-log.js"; +import { setGatewayDedupeEntry } from "./agent-job.js"; +import { broadcastChatError, broadcastChatFinal } from "./chat-broadcast.js"; +import type { AdmittedChatSend } from "./chat-send-admission.js"; +import type { PreparedChatSendSession } from "./chat-send-session.js"; +import { hasTrackedActiveSessionRun } from "./session-active-runs.js"; +import { emitSessionsChanged } from "./session-change-event.js"; +import type { GatewayRequestContext } from "./types.js"; + +type PendingDispatchLifecycleError = { + endedAt: number; + error: string; + sessionId: string; + startedAt: number; +}; + +/** Own dispatch rejection projection and post-cleanup lifecycle persistence. */ +export function createChatSendDispatchErrorLifecycle(params: { + admission: Pick< + AdmittedChatSend, + "activeRunAbort" | "cleanupAdmittedRun" | "lifecycleGeneration" | "restartSafeAdmission" + >; + context: GatewayRequestContext; + isQueuedFollowupEnqueued: () => boolean; + persistUserTurnTranscript: () => Promise; + session: Pick< + PreparedChatSendSession, + "agentId" | "backingSessionId" | "cfg" | "clientRunId" | "now" | "rawSessionKey" | "sessionKey" + >; + terminalizeRestartSafeAdmission: (state: { + retryable: boolean; + status: "failed" | "killed"; + }) => Promise; + userTurnRecorder: Pick; +}) { + const { + admission, + context, + isQueuedFollowupEnqueued, + persistUserTurnTranscript, + session, + terminalizeRestartSafeAdmission, + userTurnRecorder, + } = params; + const { activeRunAbort, cleanupAdmittedRun, lifecycleGeneration, restartSafeAdmission } = + admission; + const { agentId, backingSessionId, cfg, clientRunId, now, rawSessionKey, sessionKey } = session; + let pendingDispatchLifecycleError: PendingDispatchLifecycleError | undefined; + let persistDispatchErrorUserTurn: (() => Promise) | undefined; + + const handleError = async (err: unknown) => { + const errorMessage = String(err); + const queuedFollowupEnqueued = isQueuedFollowupEnqueued(); + let restartSafeDispatchFailureTerminalized = false; + if (restartSafeAdmission && !queuedFollowupEnqueued) { + restartSafeDispatchFailureTerminalized = await terminalizeRestartSafeAdmission({ + retryable: true, + status: "failed", + }).catch((terminalizeError: unknown) => { + context.logGateway.warn( + `failed to release restart-safe chat admission after dispatch error: ${formatForLog( + terminalizeError, + )}`, + ); + return false; + }); + if (restartSafeDispatchFailureTerminalized) { + emitSessionsChanged(context, { + sessionKey, + ...(agentId ? { agentId } : {}), + reason: "chat.dispatch-error", + }); + } + } + if (queuedFollowupEnqueued) { + context.logGateway.warn( + `webchat dispatch failed after followup queue admission: ${formatForLog(err)}`, + ); + if (!context.chatAbortedRuns.has(clientRunId)) { + setGatewayDedupeEntry({ + dedupe: context.dedupe, + key: `chat:${clientRunId}`, + entry: { + ts: Date.now(), + ok: true, + payload: { runId: clientRunId, status: "ok" as const }, + }, + }); + broadcastChatFinal({ + context, + runId: clientRunId, + sessionKey, + agentId, + }); + } + return; + } + persistDispatchErrorUserTurn = + userTurnRecorder.hasPersisted() || userTurnRecorder.isBlocked() + ? undefined + : async () => { + await persistUserTurnTranscript(); + }; + if ( + !restartSafeDispatchFailureTerminalized && + !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}`, + entry: { + ts: Date.now(), + ok: false, + payload: { + runId: clientRunId, + status: "error" as const, + summary: errorMessage, + }, + error, + }, + }); + broadcastChatError({ + context, + runId: clientRunId, + sessionKey, + agentId, + errorMessage, + }); + }; + + const finalize = () => { + const dispatchError = pendingDispatchLifecycleError; + // Reserve projection before cleanup retires the accepted dispatch root. + const releaseDispatchErrorRoot = dispatchError + ? retainGatewayRootWorkAdmissionContinuation() + : null; + cleanupAdmittedRun(); + clearAgentRunContext(clientRunId, lifecycleGeneration); + context.removeChatRun(clientRunId, clientRunId, sessionKey); + if (!dispatchError) { + return; + } + const persistDispatchLifecycleError = async () => { + const hasActiveRun = hasTrackedActiveSessionRun({ + context, + requestedKey: rawSessionKey, + canonicalKey: sessionKey, + ...(sessionKey === "global" && agentId ? { agentId } : {}), + defaultAgentId: resolveDefaultAgentId(cfg), + }); + if (hasActiveRun) { + return; + } + try { + await persistGatewaySessionLifecycleEvent({ + sessionKey, + ...(sessionKey === "global" && agentId ? { agentId } : {}), + event: { + runId: clientRunId, + sessionId: dispatchError.sessionId, + lifecycleGeneration, + ts: dispatchError.endedAt, + data: { + phase: "error", + startedAt: dispatchError.startedAt, + endedAt: dispatchError.endedAt, + error: dispatchError.error, + }, + }, + }); + emitSessionsChanged(context, { + sessionKey, + ...(agentId ? { agentId } : {}), + reason: "chat.dispatch-error", + }); + } catch (persistErr: unknown) { + context.logGateway.warn( + `webchat session lifecycle persist failed after error: ${formatForLog(persistErr)}`, + ); + } + }; + void (async () => { + await persistDispatchLifecycleError(); + await persistDispatchErrorUserTurn?.().catch((transcriptErr: unknown) => { + context.logGateway.warn( + `webchat user transcript update failed after error: ${formatForLog(transcriptErr)}`, + ); + }); + })() + .catch((continuationErr: unknown) => { + context.logGateway.warn( + `webchat session lifecycle continuation failed: ${formatForLog(continuationErr)}`, + ); + }) + .finally(() => releaseDispatchErrorRoot?.()); + }; + + return { finalize, handleError }; +} diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 0fbef5c6fb50..3e68cb4f918e 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -70,7 +70,6 @@ import { } from "../dashboard-session-title.js"; import type { ChatRunTiming } from "../server-chat-state.js"; import { getMaxChatHistoryMessagesBytes, MAX_PAYLOAD_BYTES } from "../server-constants.js"; -import { persistGatewaySessionLifecycleEvent } from "../session-lifecycle-state.js"; import { capArrayByJsonBytes, readSessionMessageByIdAsync, @@ -114,6 +113,7 @@ import { import { terminalizeRestartSafeChatAdmission } from "./chat-restart-recovery.js"; import { admitChatSend } from "./chat-send-admission.js"; import { prepareChatSendAttachments } from "./chat-send-attachments.js"; +import { createChatSendDispatchErrorLifecycle } from "./chat-send-dispatch-errors.js"; import { finalizeChatSendNonAgentReplies } from "./chat-send-nonagent-finalization.js"; import { respondChatSessionRoutingChanged, @@ -139,10 +139,7 @@ import { loadOptionalServerMethodModelCatalogSnapshot, startOptionalServerMethodModelCatalogSnapshotLoad, } from "./optional-model-catalog.js"; -import { - hasTrackedActiveSessionRun, - resolveVisibleActiveSessionRunState, -} from "./session-active-runs.js"; +import { resolveVisibleActiveSessionRunState } from "./session-active-runs.js"; import { emitSessionsChanged } from "./session-change-event.js"; import type { GatewayRequestContext, @@ -943,7 +940,6 @@ export const chatHandlers: GatewayRequestHandlers = { return; } const { - rawSessionKey, clientRunId, sessionLoadOptions, sessionLoadMs, @@ -1194,15 +1190,15 @@ export const chatHandlers: GatewayRequestHandlers = { userTurnRecorder, }); let queuedFollowupEnqueued = false; - let pendingDispatchLifecycleError: - | { - endedAt: number; - error: string; - sessionId: string; - startedAt: number; - } - | undefined; - let persistDispatchErrorUserTurn: (() => Promise) | undefined; + const dispatchErrorLifecycle = createChatSendDispatchErrorLifecycle({ + admission: admitted.value, + context, + isQueuedFollowupEnqueued: () => queuedFollowupEnqueued, + persistUserTurnTranscript: persistGatewayUserTurnTranscript, + session: preparedSession.value, + terminalizeRestartSafeAdmission, + userTurnRecorder, + }); const emitServerTiming = ( phase: ChatSendServerTimingPhase, extra?: Record, @@ -1518,160 +1514,8 @@ export const chatHandlers: GatewayRequestHandlers = { }); } }) - .catch(async (err: unknown) => { - const errorMessage = String(err); - let restartSafeDispatchFailureTerminalized = false; - if (restartSafeAdmission && !queuedFollowupEnqueued) { - restartSafeDispatchFailureTerminalized = await terminalizeRestartSafeAdmission({ - retryable: true, - status: "failed", - }).catch((terminalizeError: unknown) => { - context.logGateway.warn( - `failed to release restart-safe chat admission after dispatch error: ${formatForLog( - terminalizeError, - )}`, - ); - return false; - }); - if (restartSafeDispatchFailureTerminalized) { - emitSessionsChanged(context, { - sessionKey, - ...(agentId ? { agentId } : {}), - reason: "chat.dispatch-error", - }); - } - } - if (queuedFollowupEnqueued) { - context.logGateway.warn( - `webchat dispatch failed after followup queue admission: ${formatForLog(err)}`, - ); - if (!context.chatAbortedRuns.has(clientRunId)) { - setGatewayDedupeEntry({ - dedupe: context.dedupe, - key: `chat:${clientRunId}`, - entry: { - ts: Date.now(), - ok: true, - payload: { runId: clientRunId, status: "ok" as const }, - }, - }); - broadcastChatFinal({ - context, - runId: clientRunId, - sessionKey, - agentId, - }); - } - return; - } - persistDispatchErrorUserTurn = - userTurnRecorder.hasPersisted() || userTurnRecorder.isBlocked() - ? undefined - : async () => { - await persistGatewayUserTurnTranscript(); - }; - if ( - !restartSafeDispatchFailureTerminalized && - !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}`, - entry: { - ts: Date.now(), - ok: false, - payload: { - runId: clientRunId, - status: "error" as const, - summary: errorMessage, - }, - error, - }, - }); - broadcastChatError({ - context, - runId: clientRunId, - sessionKey, - agentId, - errorMessage, - }); - }) - .finally(() => { - const dispatchError = pendingDispatchLifecycleError; - // Reserve error projection before cleanup retires the dispatch root. Restart - // drain may already reject fresh roots, but this accepted request must finish. - const releaseDispatchErrorRoot = dispatchError - ? retainGatewayRootWorkAdmissionContinuation() - : null; - cleanupAdmittedRun(); - clearAgentRunContext(clientRunId, lifecycleGeneration); - context.removeChatRun(clientRunId, clientRunId, sessionKey); - if (!dispatchError) { - return; - } - const persistDispatchLifecycleError = async () => { - const hasActiveRun = hasTrackedActiveSessionRun({ - context, - requestedKey: rawSessionKey, - canonicalKey: sessionKey, - ...(sessionKey === "global" && agentId ? { agentId } : {}), - defaultAgentId: resolveDefaultAgentId(cfg), - }); - if (hasActiveRun) { - return; - } - try { - await persistGatewaySessionLifecycleEvent({ - sessionKey, - ...(sessionKey === "global" && agentId ? { agentId } : {}), - event: { - runId: clientRunId, - sessionId: dispatchError.sessionId, - lifecycleGeneration, - ts: dispatchError.endedAt, - data: { - phase: "error", - startedAt: dispatchError.startedAt, - endedAt: dispatchError.endedAt, - error: dispatchError.error, - }, - }, - }); - emitSessionsChanged(context, { - sessionKey, - ...(agentId ? { agentId } : {}), - reason: "chat.dispatch-error", - }); - } catch (persistErr: unknown) { - context.logGateway.warn( - `webchat session lifecycle persist failed after error: ${formatForLog(persistErr)}`, - ); - } - }; - void (async () => { - await persistDispatchLifecycleError(); - await persistDispatchErrorUserTurn?.().catch((transcriptErr: unknown) => { - context.logGateway.warn( - `webchat user transcript update failed after error: ${formatForLog(transcriptErr)}`, - ); - }); - })() - .catch((continuationErr: unknown) => { - context.logGateway.warn( - `webchat session lifecycle continuation failed: ${formatForLog(continuationErr)}`, - ); - }) - .finally(() => releaseDispatchErrorRoot?.()); - }); + .catch(dispatchErrorLifecycle.handleError) + .finally(dispatchErrorLifecycle.finalize); } catch (err) { if (restartSafeAdmission) { const terminalized = await terminalizeRestartSafeAdmission({