fix(gateway): persist webchat dispatch failures

This commit is contained in:
jesse-merhi
2026-06-22 22:15:42 +10:00
parent a822c9abaa
commit ea33aac648
2 changed files with 136 additions and 4 deletions
+69 -4
View File
@@ -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 });
@@ -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<string, unknown> =>
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);