mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor(gateway): extract chat dispatch errors (#106743)
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
1816e83ca5
commit
ad1c0edede
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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<unknown>;
|
||||
session: Pick<
|
||||
PreparedChatSendSession,
|
||||
"agentId" | "backingSessionId" | "cfg" | "clientRunId" | "now" | "rawSessionKey" | "sessionKey"
|
||||
>;
|
||||
terminalizeRestartSafeAdmission: (state: {
|
||||
retryable: boolean;
|
||||
status: "failed" | "killed";
|
||||
}) => Promise<boolean>;
|
||||
userTurnRecorder: Pick<UserTurnTranscriptRecorder, "hasPersisted" | "isBlocked">;
|
||||
}) {
|
||||
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<void>) | 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 };
|
||||
}
|
||||
@@ -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<void>) | undefined;
|
||||
const dispatchErrorLifecycle = createChatSendDispatchErrorLifecycle({
|
||||
admission: admitted.value,
|
||||
context,
|
||||
isQueuedFollowupEnqueued: () => queuedFollowupEnqueued,
|
||||
persistUserTurnTranscript: persistGatewayUserTurnTranscript,
|
||||
session: preparedSession.value,
|
||||
terminalizeRestartSafeAdmission,
|
||||
userTurnRecorder,
|
||||
});
|
||||
const emitServerTiming = (
|
||||
phase: ChatSendServerTimingPhase,
|
||||
extra?: Record<string, string | number>,
|
||||
@@ -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({
|
||||
|
||||
Reference in New Issue
Block a user