mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-28 05:16:23 -06:00
fix: recover terminal session status on visible inbound turns (#89045)
* fix: recover terminal session status on visible inbound turns (#86827) When a group chat session enters a terminal status (failed/timeout/killed), subsequent visible inbound messages now automatically recover the session by clearing stale lifecycle fields while preserving the session ID and transcript continuity. Changes: - session.ts: detect terminal status on visible turns and clear status/ startedAt/endedAt/runtimeMs/abortedLastRun without rotating sessionId - dispatch-from-config.ts: force-clear stale active reply operations for terminal sessions and retry admission once - agent.ts: mirror terminal recovery in the agent API dispatch path - kernel.ts: add zero-count visible dispatch warning diagnostic - types.ts: add 'warning' to ChannelTurnLogEvent event union * fix: guard terminal recovery from concurrent force-clear race When two visible messages arrive against the same terminal session snapshot, the second turn could force-fail the first turn's freshly admitted recovery operation, dropping the very message the recovery path exists to save. Add a terminalRecovery flag on ReplyOperation that is set after a recovery turn clears the proven stale leftover and admits its own operation. The force-clear branch now skips operations marked as in-flight terminal recoveries, letting concurrent turns fall through to normal busy/wait handling instead. Add a two-turn regression test that gates the first recovery turn open, races a second visible turn against the same terminal snapshot, and asserts neither turn's operation is incorrectly killed. Also fix missing FinalizedMsgContext import in kernel.ts. * fix: avoid return value in Promise executor to satisfy lint * fix: gate terminal force-clear to visible reply turns A heartbeat/control turn can pass the early active-run short-circuit, reach the terminal force-clear branch, and abort an in-flight visible recovery operation that a concurrent visible turn just admitted (before that op is marked terminalRecovery). Gate the force-clear to visible reply turns so non-visible turns fall through to normal busy/skip handling instead of killing the recovery they are meant to protect. Adds a focused regression test exercising a heartbeat turn that reaches the force-clear branch against a terminal session snapshot with an active recovery operation present; it must leave that operation intact and skip. * fix: verify session identity before terminal force-clear * fix: mark clean no-stale terminal recovery to survive concurrent visible turn The same-session terminal-recovery race could still drop a reply when two visible turns raced the same failed snapshot. The terminalRecovery marker was set only on the post-force-clear re-admission path, never on the clean no-stale admission path. So when the first turn admitted cleanly (no stale op to clear), its recovery op stayed terminalRecovery=false and a second concurrent visible turn force-cleared it, dropping the first reply (#86827). Consolidate marking to the single owned-admission choke point that both the clean no-stale admission and the re-admission-after-force-clear flow through. Genuine stale leftovers from the original failed run never pass through this admission, so they stay unmarked and remain force-clearable. Add a no-stale regression twin to the existing stale-race test. * fix: suppress zero-count visible-dispatch warning for observed-delivery turns maybeWarnZeroCountVisibleDispatch re-implemented a partial visibility check that omitted observedReplyDelivery, so visible turns delivered via the observed-delivery path (queuedFinal=false, zero counts, observedReplyDelivery=true) falsely tripped the silent-drop sentinel and emitted a bogus zero-count-visible-dispatch warning/event. Use the canonical hasVisibleChannelTurnDispatch helper for the warning suppression so all non-count delivery paths (observedReplyDelivery, fallback, summary, queuedFinal) are honored. Add regression tests covering the observed-delivery case (no warning) and a genuinely empty visible dispatch (still warns). * test: read persisted session store in failed-group recovery test --------- Co-authored-by: 忻役 <xinyi@mininglamp.com>
This commit is contained in:
Jerry-Xin
@@ -448,6 +448,7 @@ function createMockReplyOperation(): {
|
||||
sessionId: "session",
|
||||
abortSignal: new AbortController().signal,
|
||||
resetTriggered: false,
|
||||
terminalRecovery: false,
|
||||
phase: "running",
|
||||
result: null,
|
||||
setPhase: vi.fn(),
|
||||
@@ -461,6 +462,7 @@ function createMockReplyOperation(): {
|
||||
fail: failMock,
|
||||
abortByUser: vi.fn(),
|
||||
abortForRestart: vi.fn(),
|
||||
markTerminalRecovery: vi.fn(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -46,6 +46,7 @@ function createReplyOperation(): TestReplyOperation {
|
||||
sessionId: "session",
|
||||
abortSignal: new AbortController().signal,
|
||||
resetTriggered: false,
|
||||
terminalRecovery: false,
|
||||
phase: "queued",
|
||||
result: null,
|
||||
setPhase: vi.fn<ReplyOperation["setPhase"]>(),
|
||||
@@ -61,6 +62,7 @@ function createReplyOperation(): TestReplyOperation {
|
||||
fail: vi.fn(),
|
||||
abortByUser: vi.fn(),
|
||||
abortForRestart: vi.fn(),
|
||||
markTerminalRecovery: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -1713,6 +1713,377 @@ describe("dispatchReplyFromConfig", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("clears stale active reply operations for terminal sessions and retries admission", async () => {
|
||||
setNoAbort();
|
||||
const sessionKey = "agent:main:telegram:group:-1003774691294";
|
||||
const sessionId = "failed-session";
|
||||
const activeOperation = createReplyOperation({
|
||||
sessionKey,
|
||||
sessionId,
|
||||
resetTriggered: false,
|
||||
});
|
||||
activeOperation.setPhase("running");
|
||||
sessionStoreMocks.currentEntry = {
|
||||
sessionId,
|
||||
updatedAt: Date.now(),
|
||||
status: "failed",
|
||||
};
|
||||
const dispatcher = createDispatcher();
|
||||
const replyResolver = vi.fn(async () => ({ text: "fresh reply" }) satisfies ReplyPayload);
|
||||
|
||||
const result = await dispatchReplyFromConfig({
|
||||
ctx: buildTestCtx({
|
||||
Provider: "telegram",
|
||||
Surface: "telegram",
|
||||
OriginatingChannel: "telegram",
|
||||
ChatType: "group",
|
||||
SessionKey: sessionKey,
|
||||
MessageSid: "visible-after-failure",
|
||||
To: "telegram:-1003774691294",
|
||||
BodyForAgent: "@openclaw recover",
|
||||
}),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher,
|
||||
replyResolver,
|
||||
});
|
||||
|
||||
expect(activeOperation.result).toMatchObject({ kind: "failed", code: "run_failed" });
|
||||
expect(replyResolver).toHaveBeenCalledTimes(1);
|
||||
expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
|
||||
expect(result).toMatchObject({
|
||||
queuedFinal: true,
|
||||
counts: { tool: 0, block: 0, final: 0 },
|
||||
});
|
||||
expect(replyRunRegistry.isActive(sessionKey)).toBe(false);
|
||||
});
|
||||
|
||||
it("does not kill a sibling recovery turn when a second visible turn races the same terminal snapshot", async () => {
|
||||
setNoAbort();
|
||||
const sessionKey = "agent:main:telegram:group:-1003774691295";
|
||||
const sessionId = "failed-session-race";
|
||||
// Leftover stuck run from the failed lifecycle; both racing turns read the
|
||||
// same terminal store snapshot below.
|
||||
const staleOperation = createReplyOperation({
|
||||
sessionKey,
|
||||
sessionId,
|
||||
resetTriggered: false,
|
||||
});
|
||||
staleOperation.setPhase("running");
|
||||
sessionStoreMocks.currentEntry = {
|
||||
sessionId,
|
||||
updatedAt: Date.now(),
|
||||
status: "failed",
|
||||
};
|
||||
|
||||
let releaseFirstTurn: () => void = () => {};
|
||||
const firstResolverGate = new Promise<void>((release) => {
|
||||
releaseFirstTurn = release;
|
||||
});
|
||||
let signalFirstResolverEntered: () => void = () => {};
|
||||
const firstTurnEntered = new Promise<void>((resolve) => {
|
||||
signalFirstResolverEntered = resolve;
|
||||
});
|
||||
const firstReplyResolver = vi.fn(async () => {
|
||||
signalFirstResolverEntered();
|
||||
await firstResolverGate;
|
||||
return { text: "first recovery reply" } satisfies ReplyPayload;
|
||||
});
|
||||
const secondReplyResolver = vi.fn(
|
||||
async () => ({ text: "second reply" }) satisfies ReplyPayload,
|
||||
);
|
||||
const firstDispatcher = createDispatcher();
|
||||
const secondDispatcher = createDispatcher();
|
||||
|
||||
const buildRaceCtx = (messageSid: string) =>
|
||||
buildTestCtx({
|
||||
Provider: "telegram",
|
||||
Surface: "telegram",
|
||||
OriginatingChannel: "telegram",
|
||||
ChatType: "group",
|
||||
SessionKey: sessionKey,
|
||||
MessageSid: messageSid,
|
||||
To: "telegram:-1003774691295",
|
||||
BodyForAgent: "@openclaw recover",
|
||||
});
|
||||
|
||||
const firstTurn = dispatchReplyFromConfig({
|
||||
ctx: buildRaceCtx("visible-race-first"),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher: firstDispatcher,
|
||||
replyResolver: firstReplyResolver,
|
||||
});
|
||||
|
||||
// First turn cleared the leftover run and now owns the in-flight recovery
|
||||
// operation; capture it before the second turn races in.
|
||||
await firstTurnEntered;
|
||||
expect(staleOperation.result).toMatchObject({ kind: "failed", code: "run_failed" });
|
||||
const recoveryOperation = replyRunRegistry.get(sessionKey);
|
||||
expect(recoveryOperation).toBeDefined();
|
||||
expect(recoveryOperation).not.toBe(staleOperation);
|
||||
|
||||
const secondTurn = dispatchReplyFromConfig({
|
||||
ctx: buildRaceCtx("visible-race-second"),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher: secondDispatcher,
|
||||
replyResolver: secondReplyResolver,
|
||||
});
|
||||
|
||||
// Give the second turn time to run its admission/recovery path. With the
|
||||
// bug it would force-fail the first turn's fresh recovery operation here.
|
||||
await new Promise((resolve) => {
|
||||
setTimeout(resolve, 100);
|
||||
});
|
||||
expect(recoveryOperation?.result).toBeNull();
|
||||
expect(secondReplyResolver).not.toHaveBeenCalled();
|
||||
|
||||
releaseFirstTurn();
|
||||
const firstResult = await firstTurn;
|
||||
const secondResult = await secondTurn;
|
||||
|
||||
// The first recovery completed normally; the second turn was never allowed
|
||||
// to kill it and got its own admission once the first finished.
|
||||
expect(recoveryOperation?.result).toMatchObject({ kind: "completed" });
|
||||
expect(firstReplyResolver).toHaveBeenCalledTimes(1);
|
||||
expect(secondReplyResolver).toHaveBeenCalledTimes(1);
|
||||
expect(firstResult).toMatchObject({ queuedFinal: true });
|
||||
expect(secondResult).toMatchObject({ queuedFinal: true });
|
||||
expect(firstDispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
|
||||
expect(secondDispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
|
||||
expect(replyRunRegistry.isActive(sessionKey)).toBe(false);
|
||||
});
|
||||
|
||||
it("marks a clean no-stale terminal recovery so a racing visible turn cannot force-clear it", async () => {
|
||||
setNoAbort();
|
||||
const sessionKey = "agent:main:telegram:group:-1003774691297";
|
||||
const sessionId = "failed-session-no-stale-race";
|
||||
// No leftover op is pre-registered: the first visible turn reaches the clean
|
||||
// admission path (nothing to force-clear). Both racing turns read the same
|
||||
// terminal store snapshot below.
|
||||
sessionStoreMocks.currentEntry = {
|
||||
sessionId,
|
||||
updatedAt: Date.now(),
|
||||
status: "failed",
|
||||
};
|
||||
|
||||
let releaseFirstTurn: () => void = () => {};
|
||||
const firstResolverGate = new Promise<void>((release) => {
|
||||
releaseFirstTurn = release;
|
||||
});
|
||||
let signalFirstResolverEntered: () => void = () => {};
|
||||
const firstTurnEntered = new Promise<void>((resolve) => {
|
||||
signalFirstResolverEntered = resolve;
|
||||
});
|
||||
const firstReplyResolver = vi.fn(async () => {
|
||||
signalFirstResolverEntered();
|
||||
await firstResolverGate;
|
||||
return { text: "first recovery reply" } satisfies ReplyPayload;
|
||||
});
|
||||
const secondReplyResolver = vi.fn(
|
||||
async () => ({ text: "second reply" }) satisfies ReplyPayload,
|
||||
);
|
||||
const firstDispatcher = createDispatcher();
|
||||
const secondDispatcher = createDispatcher();
|
||||
|
||||
const buildRaceCtx = (messageSid: string) =>
|
||||
buildTestCtx({
|
||||
Provider: "telegram",
|
||||
Surface: "telegram",
|
||||
OriginatingChannel: "telegram",
|
||||
ChatType: "group",
|
||||
SessionKey: sessionKey,
|
||||
MessageSid: messageSid,
|
||||
To: "telegram:-1003774691295",
|
||||
BodyForAgent: "@openclaw recover",
|
||||
});
|
||||
|
||||
const firstTurn = dispatchReplyFromConfig({
|
||||
ctx: buildRaceCtx("visible-no-stale-first"),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher: firstDispatcher,
|
||||
replyResolver: firstReplyResolver,
|
||||
});
|
||||
|
||||
// First turn admitted cleanly and now owns the in-flight recovery operation;
|
||||
// capture it before the second turn races in.
|
||||
await firstTurnEntered;
|
||||
const recoveryOperation = replyRunRegistry.get(sessionKey);
|
||||
expect(recoveryOperation).toBeDefined();
|
||||
// The marker must be set on the clean no-stale admission path too; without it
|
||||
// the racing second visible turn would force-clear this op (#86827).
|
||||
expect(recoveryOperation?.terminalRecovery).toBe(true);
|
||||
|
||||
const secondTurn = dispatchReplyFromConfig({
|
||||
ctx: buildRaceCtx("visible-no-stale-second"),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher: secondDispatcher,
|
||||
replyResolver: secondReplyResolver,
|
||||
});
|
||||
|
||||
// Give the second turn time to run its admission/recovery path. With the
|
||||
// bug it would force-fail the first turn's fresh recovery operation here.
|
||||
await new Promise((resolve) => {
|
||||
setTimeout(resolve, 100);
|
||||
});
|
||||
expect(recoveryOperation?.result).toBeNull();
|
||||
expect(secondReplyResolver).not.toHaveBeenCalled();
|
||||
|
||||
releaseFirstTurn();
|
||||
const firstResult = await firstTurn;
|
||||
const secondResult = await secondTurn;
|
||||
|
||||
// The first recovery completed normally; the second turn was never allowed
|
||||
// to kill it and got its own admission once the first finished.
|
||||
expect(recoveryOperation?.result).toMatchObject({ kind: "completed" });
|
||||
expect(firstReplyResolver).toHaveBeenCalledTimes(1);
|
||||
expect(secondReplyResolver).toHaveBeenCalledTimes(1);
|
||||
expect(firstResult).toMatchObject({ queuedFinal: true });
|
||||
expect(secondResult).toMatchObject({ queuedFinal: true });
|
||||
expect(firstDispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
|
||||
expect(secondDispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
|
||||
expect(replyRunRegistry.isActive(sessionKey)).toBe(false);
|
||||
});
|
||||
|
||||
it("does not force-clear an active recovery operation for a heartbeat turn on a terminal session", async () => {
|
||||
setNoAbort();
|
||||
const sessionKey = "agent:main:telegram:group:-1003774691296";
|
||||
const sessionId = "failed-session-heartbeat";
|
||||
sessionStoreMocks.currentEntry = {
|
||||
sessionId,
|
||||
updatedAt: Date.now(),
|
||||
status: "failed",
|
||||
};
|
||||
const dispatcher = createDispatcher();
|
||||
const replyResolver = vi.fn(
|
||||
async () => ({ text: "heartbeat should not run" }) satisfies ReplyPayload,
|
||||
);
|
||||
|
||||
// A concurrent visible turn already cleared the failed leftover and admitted
|
||||
// a fresh recovery operation. Register it inside the fast-abort seam, which
|
||||
// runs after the early heartbeat short-circuit but before admission, so the
|
||||
// heartbeat reaches the terminal force-clear branch with this op active. The
|
||||
// op is intentionally NOT marked `terminalRecovery`, so only the visible-turn
|
||||
// guard can stop the heartbeat from force-failing it.
|
||||
let recoveryOperation: ReturnType<typeof createReplyOperation> | undefined;
|
||||
|
||||
const result = await dispatchReplyFromConfig({
|
||||
ctx: buildTestCtx({
|
||||
Provider: "telegram",
|
||||
Surface: "telegram",
|
||||
OriginatingChannel: "telegram",
|
||||
ChatType: "group",
|
||||
SessionKey: sessionKey,
|
||||
MessageSid: "heartbeat-after-failure",
|
||||
To: "telegram:-1003774691296",
|
||||
BodyForAgent: "[OpenClaw heartbeat poll]",
|
||||
}),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher,
|
||||
replyOptions: { isHeartbeat: true },
|
||||
fastAbortResolver: async () => {
|
||||
recoveryOperation = createReplyOperation({
|
||||
sessionKey,
|
||||
sessionId,
|
||||
resetTriggered: false,
|
||||
});
|
||||
recoveryOperation.setPhase("running");
|
||||
return { handled: false, aborted: false };
|
||||
},
|
||||
formatAbortReplyTextResolver: () => "aborted",
|
||||
replyResolver,
|
||||
});
|
||||
|
||||
// The heartbeat left the active visible recovery operation untouched and
|
||||
// skipped itself instead of force-clearing the in-flight visible turn.
|
||||
expect(recoveryOperation).toBeDefined();
|
||||
expect(recoveryOperation?.result).toBeNull();
|
||||
expect(replyRunRegistry.get(sessionKey)).toBe(recoveryOperation);
|
||||
expect(replyResolver).not.toHaveBeenCalled();
|
||||
expect(dispatcher.sendFinalReply).not.toHaveBeenCalled();
|
||||
expect(result).toMatchObject({
|
||||
queuedFinal: false,
|
||||
counts: { tool: 0, block: 0, final: 0 },
|
||||
});
|
||||
recoveryOperation?.complete();
|
||||
});
|
||||
|
||||
it("does not force-clear an active operation whose session id differs from the terminal snapshot", async () => {
|
||||
setNoAbort();
|
||||
const sessionKey = "agent:main:telegram:group:-1003774691297";
|
||||
// Terminal store snapshot still reports the failed lifecycle's session id.
|
||||
sessionStoreMocks.currentEntry = {
|
||||
sessionId: "failed-session-rotated",
|
||||
updatedAt: Date.now(),
|
||||
status: "failed",
|
||||
};
|
||||
const dispatcher = createDispatcher();
|
||||
const replyResolver = vi.fn(
|
||||
async () => ({ text: "visible recovery reply" }) satisfies ReplyPayload,
|
||||
);
|
||||
|
||||
// A concurrent reset/rotation admitted a fresh op under the same session key
|
||||
// but with a NEW session id, after this turn already captured the stale
|
||||
// terminal snapshot. Register it inside the fast-abort seam, which runs after
|
||||
// the early short-circuit but before admission, so the visible turn reaches
|
||||
// the terminal force-clear branch with this op active. The op is NOT marked
|
||||
// `terminalRecovery`, so only the session-id guard can stop the force-clear.
|
||||
let freshOperation: ReturnType<typeof createReplyOperation> | undefined;
|
||||
let signalFreshRegistered: () => void = () => {};
|
||||
const freshRegistered = new Promise<void>((resolve) => {
|
||||
signalFreshRegistered = resolve;
|
||||
});
|
||||
|
||||
const turn = dispatchReplyFromConfig({
|
||||
ctx: buildTestCtx({
|
||||
Provider: "telegram",
|
||||
Surface: "telegram",
|
||||
OriginatingChannel: "telegram",
|
||||
ChatType: "group",
|
||||
SessionKey: sessionKey,
|
||||
MessageSid: "visible-after-rotation",
|
||||
To: "telegram:-1003774691297",
|
||||
BodyForAgent: "@openclaw recover",
|
||||
}),
|
||||
cfg: automaticGroupReplyConfig,
|
||||
dispatcher,
|
||||
fastAbortResolver: async () => {
|
||||
freshOperation = createReplyOperation({
|
||||
sessionKey,
|
||||
sessionId: "fresh-rotated-session",
|
||||
resetTriggered: false,
|
||||
});
|
||||
freshOperation.setPhase("running");
|
||||
signalFreshRegistered();
|
||||
return { handled: false, aborted: false };
|
||||
},
|
||||
formatAbortReplyTextResolver: () => "aborted",
|
||||
replyResolver,
|
||||
});
|
||||
|
||||
// Let the visible turn run its admission/force-clear path. With the bug it
|
||||
// would force-fail the rotated op here, mistaking a valid in-flight reply for
|
||||
// the stale terminal leftover and recreating the message loss (#86827).
|
||||
await freshRegistered;
|
||||
await new Promise((resolve) => {
|
||||
setTimeout(resolve, 100);
|
||||
});
|
||||
|
||||
// The session-id guard kept the rotated op untouched: it still owns the
|
||||
// session key and was never force-failed, so the visible turn simply parks
|
||||
// behind it instead of dropping it.
|
||||
expect(freshOperation).toBeDefined();
|
||||
expect(freshOperation?.result).toBeNull();
|
||||
expect(replyRunRegistry.get(sessionKey)).toBe(freshOperation);
|
||||
expect(replyResolver).not.toHaveBeenCalled();
|
||||
|
||||
// Releasing the rotated op lets the parked visible turn admit and deliver,
|
||||
// proving the message survived the rotation rather than being silently lost.
|
||||
freshOperation?.complete();
|
||||
const result = await turn;
|
||||
expect(replyResolver).toHaveBeenCalledTimes(1);
|
||||
expect(result).toMatchObject({ queuedFinal: true });
|
||||
expect(replyRunRegistry.isActive(sessionKey)).toBe(false);
|
||||
});
|
||||
|
||||
it("routes when OriginatingChannel differs from Provider", async () => {
|
||||
setNoAbort();
|
||||
mocks.routeReply.mockClear();
|
||||
|
||||
@@ -206,6 +206,10 @@ function isDispatchReplyOperationAbortedError(
|
||||
return error instanceof DispatchReplyOperationAbortedError;
|
||||
}
|
||||
|
||||
function isRecoverableTerminalSessionStatus(status: SessionEntry["status"] | undefined): boolean {
|
||||
return status === "failed" || status === "timeout" || status === "killed";
|
||||
}
|
||||
|
||||
function composeAbortSignals(...signals: Array<AbortSignal | undefined>): AbortSignal | undefined {
|
||||
const activeSignals: AbortSignal[] = [];
|
||||
for (const signal of signals) {
|
||||
@@ -1421,6 +1425,48 @@ export async function dispatchReplyFromConfig(
|
||||
});
|
||||
}
|
||||
}
|
||||
if (
|
||||
admission.status === "skipped" &&
|
||||
admission.reason === "active-run" &&
|
||||
// Only visible reply turns may force-clear a stale terminal operation.
|
||||
// A heartbeat/control turn can also see the terminal snapshot, but it must
|
||||
// not abort an in-flight visible recovery a concurrent visible turn just
|
||||
// admitted (before that op is marked `terminalRecovery`); let it fall
|
||||
// through to normal busy/skip handling instead.
|
||||
replyTurnKind === "visible" &&
|
||||
isRecoverableTerminalSessionStatus(sessionStoreEntry.entry?.status) &&
|
||||
// Only clear the leftover op that belongs to the SAME terminal session.
|
||||
// A concurrent reset/rotation can admit a fresh op (new sessionId) under
|
||||
// this session key while we still hold the stale terminal snapshot;
|
||||
// force-clearing by the active op's id would drop that valid in-flight
|
||||
// reply and recreate the message loss this fix exists to prevent (#86827).
|
||||
admission.activeOperation?.sessionId === sessionStoreEntry.entry?.sessionId &&
|
||||
// Only clear the proven stale leftover from the failed lifecycle. A
|
||||
// freshly-admitted visible recovery op is marked `terminalRecovery` at the
|
||||
// admission choke point below; force-failing that op would drop the very
|
||||
// recovery turn this path exists to protect (concurrent visible turns can
|
||||
// read the same terminal snapshot before it clears).
|
||||
!admission.activeOperation?.terminalRecovery
|
||||
) {
|
||||
const cleared = forceClearReplyRunBySessionId(
|
||||
admission.activeOperation?.sessionId ?? operationSessionId,
|
||||
new Error("clearing stale terminal reply operation"),
|
||||
);
|
||||
if (cleared) {
|
||||
logVerbose(
|
||||
`dispatch-from-config: cleared stale active reply operation for terminal session ${dispatchOperationSessionKey}`,
|
||||
);
|
||||
admission = await admitReplyTurn({
|
||||
sessionKey: dispatchOperationSessionKey,
|
||||
sessionId: operationSessionId,
|
||||
kind: replyTurnKind,
|
||||
resetTriggered: false,
|
||||
routeThreadId,
|
||||
upstreamAbortSignal: params.replyOptions?.abortSignal,
|
||||
waitForActive: !allowActivePreDispatch && !allowSlackRoutedThreadBypass,
|
||||
});
|
||||
}
|
||||
}
|
||||
if (admission.status === "skipped") {
|
||||
if (allowActivePreDispatch && admission.reason === "active-run") {
|
||||
preDispatchAbortOperation = admission.activeOperation;
|
||||
@@ -1445,6 +1491,20 @@ export async function dispatchReplyFromConfig(
|
||||
);
|
||||
return { status: "busy" };
|
||||
}
|
||||
// Mark every freshly-admitted visible recovery of a terminal session at this
|
||||
// single choke point (both the clean no-stale admission and the
|
||||
// re-admission after a sibling force-clear flow through here). The marker
|
||||
// protects this op from being force-cleared by a concurrent sibling visible
|
||||
// turn that reads the same terminal snapshot (#86827). Genuine stale
|
||||
// leftovers from the original failed run never pass through this admission,
|
||||
// so they stay unmarked and remain force-clearable.
|
||||
if (
|
||||
replyTurnKind === "visible" &&
|
||||
isRecoverableTerminalSessionStatus(sessionStoreEntry.entry?.status) &&
|
||||
operationSessionId === sessionStoreEntry.entry?.sessionId
|
||||
) {
|
||||
admission.operation.markTerminalRecovery();
|
||||
}
|
||||
dispatchReplyOperation = admission.operation;
|
||||
dispatchReplyOperation.retainFailureUntilComplete();
|
||||
dispatchAbortOperation = admission.operation;
|
||||
|
||||
@@ -56,9 +56,18 @@ export type ReplyOperation = {
|
||||
readonly routeThreadId?: string | number;
|
||||
readonly abortSignal: AbortSignal;
|
||||
readonly resetTriggered: boolean;
|
||||
/**
|
||||
* True when this operation was admitted to recover a terminal session (a
|
||||
* leftover failed/timeout/killed run). Concurrent visible turns reading the
|
||||
* same terminal store snapshot must NOT force-clear such an operation: it is a
|
||||
* sibling recovery already in flight, not the proven stale leftover.
|
||||
*/
|
||||
readonly terminalRecovery: boolean;
|
||||
readonly phase: ReplyOperationPhase;
|
||||
readonly result: ReplyOperationResult | null;
|
||||
setPhase(next: "queued" | "preflight_compacting" | "memory_flushing" | "running"): void;
|
||||
/** Mark this operation as an in-flight terminal-session recovery. */
|
||||
markTerminalRecovery(): void;
|
||||
updateSessionId(nextSessionId: string): void;
|
||||
attachBackend(handle: ReplyBackendHandle): void;
|
||||
detachBackend(handle: ReplyBackendHandle): void;
|
||||
@@ -391,6 +400,7 @@ export function createReplyOperation(params: {
|
||||
let result: ReplyOperationResult | null = null;
|
||||
let stateCleared = false;
|
||||
let retainFailureUntilComplete = false;
|
||||
let terminalRecovery = false;
|
||||
|
||||
const clearState = (
|
||||
afterClearBarrier?: PromiseLike<unknown>,
|
||||
@@ -472,6 +482,9 @@ export function createReplyOperation(params: {
|
||||
get resetTriggered() {
|
||||
return params.resetTriggered;
|
||||
},
|
||||
get terminalRecovery() {
|
||||
return terminalRecovery;
|
||||
},
|
||||
get phase() {
|
||||
return phase;
|
||||
},
|
||||
@@ -484,6 +497,9 @@ export function createReplyOperation(params: {
|
||||
}
|
||||
phase = next;
|
||||
},
|
||||
markTerminalRecovery() {
|
||||
terminalRecovery = true;
|
||||
},
|
||||
updateSessionId(nextSessionId) {
|
||||
if (result) {
|
||||
return;
|
||||
|
||||
@@ -11,6 +11,7 @@ import * as bootstrapCache from "../../agents/bootstrap-cache.js";
|
||||
import type { OpenClawConfig } from "../../config/config.js";
|
||||
import type { SessionEntry } from "../../config/sessions.js";
|
||||
import { runExclusiveSessionStoreWrite } from "../../config/sessions/store-writer.js";
|
||||
import { readSessionStoreForTest } from "../../config/sessions/test-helpers.js";
|
||||
import { formatZonedTimestamp } from "../../infra/format-time/format-datetime.ts";
|
||||
import {
|
||||
testing as sessionBindingTesting,
|
||||
@@ -2157,13 +2158,14 @@ describe("initSessionState reset policy", () => {
|
||||
expectNewSession: true,
|
||||
},
|
||||
{
|
||||
name: "failed main terminal rows reuse when the transcript exists",
|
||||
name: "failed main terminal rows recover on visible turns when the transcript exists",
|
||||
sessionKey: "agent:main:main",
|
||||
status: "failed" as const,
|
||||
updatedAtOffsetMs: -10_000,
|
||||
endedAtOffsetMs: -11_000,
|
||||
transcriptMtimeOffsetMs: 0,
|
||||
expectNewSession: false,
|
||||
expectRecovered: true,
|
||||
},
|
||||
{
|
||||
name: "main terminal rows reuse when updatedAt already reflects the transcript",
|
||||
@@ -2228,6 +2230,14 @@ describe("initSessionState reset policy", () => {
|
||||
expect(entry?.startedAt).toBeUndefined();
|
||||
expect(entry?.endedAt).toBeUndefined();
|
||||
expect(entry?.runtimeMs).toBeUndefined();
|
||||
} else if (scenario.expectRecovered) {
|
||||
// Visible turns recover recoverable terminal rows in place: the session id
|
||||
// is reused, but the terminal lifecycle fields are cleared (#86827).
|
||||
expect(result.sessionId).toBe(existingSessionId);
|
||||
expect(entry?.status).toBeUndefined();
|
||||
expect(entry?.startedAt).toBeUndefined();
|
||||
expect(entry?.endedAt).toBeUndefined();
|
||||
expect(entry?.runtimeMs).toBeUndefined();
|
||||
} else {
|
||||
expect(result.sessionId).toBe(existingSessionId);
|
||||
expect(entry?.status).toBe(scenario.status ?? "done");
|
||||
@@ -2235,6 +2245,54 @@ describe("initSessionState reset policy", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("recovers failed group sessions without rotating the transcript", async () => {
|
||||
vi.setSystemTime(new Date(2026, 0, 18, 5, 30, 0));
|
||||
const root = await makeCaseDir("openclaw-reset-failed-entry-");
|
||||
const storePath = path.join(root, "sessions.json");
|
||||
const sessionKey = "agent:main:telegram:group:-1001";
|
||||
const existingSessionId = "failed-entry-old";
|
||||
await writeSessionStoreFast(storePath, {
|
||||
[sessionKey]: {
|
||||
sessionId: existingSessionId,
|
||||
updatedAt: Date.now(),
|
||||
startedAt: Date.now() - 10_000,
|
||||
endedAt: Date.now() - 1_000,
|
||||
runtimeMs: 9_000,
|
||||
status: "failed",
|
||||
abortedLastRun: true,
|
||||
chatType: "group",
|
||||
},
|
||||
});
|
||||
|
||||
const cfg = { session: { store: storePath } } as OpenClawConfig;
|
||||
const result = await initSessionState({
|
||||
ctx: {
|
||||
Body: "@openclaw hello",
|
||||
RawBody: "@openclaw hello",
|
||||
CommandBody: "@openclaw hello",
|
||||
SessionKey: sessionKey,
|
||||
ChatType: "group",
|
||||
Provider: "telegram",
|
||||
BotUsername: "openclaw",
|
||||
},
|
||||
cfg,
|
||||
commandAuthorized: true,
|
||||
});
|
||||
|
||||
expect(result.isNewSession).toBe(false);
|
||||
expect(result.sessionId).toBe(existingSessionId);
|
||||
expect(result.abortedLastRun).toBe(false);
|
||||
expect(result.sessionEntry.abortedLastRun).toBeUndefined();
|
||||
|
||||
const persisted = readSessionStoreForTest(storePath);
|
||||
expect(persisted[sessionKey]?.sessionId).toBe(existingSessionId);
|
||||
expect(persisted[sessionKey]?.status).toBeUndefined();
|
||||
expect(persisted[sessionKey]?.startedAt).toBeUndefined();
|
||||
expect(persisted[sessionKey]?.endedAt).toBeUndefined();
|
||||
expect(persisted[sessionKey]?.runtimeMs).toBeUndefined();
|
||||
expect(persisted[sessionKey]?.abortedLastRun).toBeUndefined();
|
||||
});
|
||||
|
||||
it("keeps the existing stale session for /reset soft", async () => {
|
||||
vi.setSystemTime(new Date(2026, 0, 18, 5, 30, 0));
|
||||
const root = await makeCaseDir("openclaw-reset-soft-stale-");
|
||||
|
||||
@@ -160,6 +160,21 @@ function hasProviderOwnedSession(entry: SessionEntry | undefined): boolean {
|
||||
return Boolean(provider && getCliSessionBinding(entry, provider));
|
||||
}
|
||||
|
||||
function isRecoverableTerminalSessionStatus(status: SessionEntry["status"] | undefined): boolean {
|
||||
return status === "failed" || status === "timeout" || status === "killed";
|
||||
}
|
||||
|
||||
function recoverTerminalSessionEntryForVisibleTurn(entry: SessionEntry): SessionEntry {
|
||||
return {
|
||||
...entry,
|
||||
status: undefined,
|
||||
startedAt: undefined,
|
||||
endedAt: undefined,
|
||||
runtimeMs: undefined,
|
||||
abortedLastRun: undefined,
|
||||
};
|
||||
}
|
||||
|
||||
export type SessionInitResult = {
|
||||
sessionCtx: TemplateContext;
|
||||
sessionEntry: SessionEntry;
|
||||
@@ -538,9 +553,16 @@ async function initSessionStateAttemptLocked(
|
||||
mainKey,
|
||||
storePath,
|
||||
}));
|
||||
const recoverTerminalVisibleEntry =
|
||||
canReuseExistingEntry &&
|
||||
!isSystemEvent &&
|
||||
!resetTriggered &&
|
||||
(entryFreshness?.fresh ?? false) &&
|
||||
isRecoverableTerminalSessionStatus(entry?.status);
|
||||
const freshEntry =
|
||||
(isSystemEvent && canReuseExistingEntry) ||
|
||||
(((reconnectResumeRequested && canReuseExistingEntry) ||
|
||||
recoverTerminalVisibleEntry ||
|
||||
(entryFreshness?.fresh ?? false) ||
|
||||
(softResetAllowed && canReuseExistingEntry)) &&
|
||||
!terminalMainTranscriptNewerThanRegistry);
|
||||
@@ -577,23 +599,29 @@ async function initSessionStateAttemptLocked(
|
||||
clearSessionResetRuntimeState([sessionKey, previousSessionEntry.sessionId]);
|
||||
}
|
||||
|
||||
if (!isNewSession && effectiveFreshEntry && canReuseExistingEntry) {
|
||||
sessionId = entry.sessionId;
|
||||
systemSent = entry.systemSent ?? false;
|
||||
abortedLastRun = entry.abortedLastRun ?? false;
|
||||
persistedThinking = entry.thinkingLevel;
|
||||
persistedVerbose = entry.verboseLevel;
|
||||
persistedTrace = entry.traceLevel;
|
||||
persistedReasoning = entry.reasoningLevel;
|
||||
persistedTtsAuto = entry.ttsAuto;
|
||||
persistedResponseUsage = entry.responseUsage;
|
||||
persistedModelOverride = entry.modelOverride;
|
||||
persistedProviderOverride = entry.providerOverride;
|
||||
persistedModelOverrideSource = entry.modelOverrideSource;
|
||||
persistedAuthProfileOverride = entry.authProfileOverride;
|
||||
persistedAuthProfileOverrideSource = entry.authProfileOverrideSource;
|
||||
persistedAuthProfileOverrideCompactionCount = entry.authProfileOverrideCompactionCount;
|
||||
persistedLabel = entry.label;
|
||||
const recoveredTerminalEntry =
|
||||
entry && recoverTerminalVisibleEntry
|
||||
? recoverTerminalSessionEntryForVisibleTurn(entry)
|
||||
: undefined;
|
||||
const reusableEntry = recoveredTerminalEntry ?? entry;
|
||||
|
||||
if (!isNewSession && effectiveFreshEntry && canReuseExistingEntry && reusableEntry) {
|
||||
sessionId = reusableEntry.sessionId;
|
||||
systemSent = reusableEntry.systemSent ?? false;
|
||||
abortedLastRun = reusableEntry.abortedLastRun ?? false;
|
||||
persistedThinking = reusableEntry.thinkingLevel;
|
||||
persistedVerbose = reusableEntry.verboseLevel;
|
||||
persistedTrace = reusableEntry.traceLevel;
|
||||
persistedReasoning = reusableEntry.reasoningLevel;
|
||||
persistedTtsAuto = reusableEntry.ttsAuto;
|
||||
persistedResponseUsage = reusableEntry.responseUsage;
|
||||
persistedModelOverride = reusableEntry.modelOverride;
|
||||
persistedProviderOverride = reusableEntry.providerOverride;
|
||||
persistedModelOverrideSource = reusableEntry.modelOverrideSource;
|
||||
persistedAuthProfileOverride = reusableEntry.authProfileOverride;
|
||||
persistedAuthProfileOverrideSource = reusableEntry.authProfileOverrideSource;
|
||||
persistedAuthProfileOverrideCompactionCount = reusableEntry.authProfileOverrideCompactionCount;
|
||||
persistedLabel = reusableEntry.label;
|
||||
} else {
|
||||
sessionId = crypto.randomUUID();
|
||||
isNewSession = true;
|
||||
@@ -648,7 +676,7 @@ async function initSessionStateAttemptLocked(
|
||||
}
|
||||
}
|
||||
|
||||
const baseEntry = !isNewSession && effectiveFreshEntry ? entry : undefined;
|
||||
const baseEntry = !isNewSession && effectiveFreshEntry ? reusableEntry : undefined;
|
||||
const usageFamilyKey = previousSessionEntry
|
||||
? (previousSessionEntry.usageFamilyKey ?? sessionKey)
|
||||
: baseEntry?.usageFamilyKey;
|
||||
@@ -738,7 +766,7 @@ async function initSessionStateAttemptLocked(
|
||||
: (baseEntry?.sessionStartedAt ?? lifecycleTimestamps.sessionStartedAt),
|
||||
lastInteractionAt: isSystemEvent ? baseEntry?.lastInteractionAt : now,
|
||||
systemSent,
|
||||
abortedLastRun,
|
||||
abortedLastRun: recoveredTerminalEntry ? undefined : abortedLastRun,
|
||||
// Persist previously stored thinking/verbose levels when present.
|
||||
thinkingLevel: persistedThinking ?? baseEntry?.thinkingLevel,
|
||||
verboseLevel: persistedVerbose ?? baseEntry?.verboseLevel,
|
||||
|
||||
@@ -804,6 +804,108 @@ describe("channel turn kernel", () => {
|
||||
expect(logRecord?.trace?.traceId).toBe(traceId);
|
||||
});
|
||||
|
||||
it("logs a warning when a visible prepared dispatch queues no payloads", async () => {
|
||||
const events: string[] = [];
|
||||
const log = vi.fn();
|
||||
const recordInboundSession = createRecordInboundSession(events);
|
||||
const runDispatch = vi.fn(async () => ({
|
||||
queuedFinal: false,
|
||||
counts: { tool: 0, block: 0, final: 0 },
|
||||
}));
|
||||
|
||||
const result = await runPreparedChannelTurn({
|
||||
channel: "test",
|
||||
routeSessionKey: "agent:main:test:peer",
|
||||
storePath: "/tmp/sessions.json",
|
||||
ctxPayload: createCtx(),
|
||||
recordInboundSession,
|
||||
runDispatch,
|
||||
log,
|
||||
messageId: "msg-zero",
|
||||
record: {
|
||||
onRecordError: vi.fn(),
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.dispatchResult?.queuedFinal).toBe(false);
|
||||
expect(log.mock.calls).toContainEqual([
|
||||
expect.objectContaining({
|
||||
stage: "dispatch",
|
||||
event: "warning",
|
||||
messageId: "msg-zero",
|
||||
reason: "zero-count-visible-dispatch",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not warn for observed-path deliveries with zero queued counts", async () => {
|
||||
const events: string[] = [];
|
||||
const log = vi.fn();
|
||||
const recordInboundSession = createRecordInboundSession(events);
|
||||
// Observed-delivery path: queuedFinal false and all counts zero, but the reply was
|
||||
// delivered via observedReplyDelivery and must not trip the silent-drop sentinel.
|
||||
const runDispatch = vi.fn(async () => ({
|
||||
queuedFinal: false,
|
||||
counts: { tool: 0, block: 0, final: 0 },
|
||||
observedReplyDelivery: true,
|
||||
}));
|
||||
|
||||
const result = await runPreparedChannelTurn({
|
||||
channel: "test",
|
||||
routeSessionKey: "agent:main:test:peer",
|
||||
storePath: "/tmp/sessions.json",
|
||||
ctxPayload: createCtx(),
|
||||
recordInboundSession,
|
||||
runDispatch,
|
||||
log,
|
||||
messageId: "msg-observed",
|
||||
record: {
|
||||
onRecordError: vi.fn(),
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.dispatchResult?.observedReplyDelivery).toBe(true);
|
||||
expect(log.mock.calls).not.toContainEqual([
|
||||
expect.objectContaining({ reason: "zero-count-visible-dispatch" }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("still warns when a visible turn has zero counts and no observed delivery", async () => {
|
||||
const events: string[] = [];
|
||||
const log = vi.fn();
|
||||
const recordInboundSession = createRecordInboundSession(events);
|
||||
// Guard against over-suppression: a genuinely empty visible dispatch must still warn.
|
||||
const runDispatch = vi.fn(async () => ({
|
||||
queuedFinal: false,
|
||||
counts: { tool: 0, block: 0, final: 0 },
|
||||
observedReplyDelivery: false,
|
||||
}));
|
||||
|
||||
const result = await runPreparedChannelTurn({
|
||||
channel: "test",
|
||||
routeSessionKey: "agent:main:test:peer",
|
||||
storePath: "/tmp/sessions.json",
|
||||
ctxPayload: createCtx(),
|
||||
recordInboundSession,
|
||||
runDispatch,
|
||||
log,
|
||||
messageId: "msg-empty",
|
||||
record: {
|
||||
onRecordError: vi.fn(),
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.dispatchResult?.observedReplyDelivery).toBe(false);
|
||||
expect(log.mock.calls).toContainEqual([
|
||||
expect.objectContaining({
|
||||
stage: "dispatch",
|
||||
event: "warning",
|
||||
messageId: "msg-empty",
|
||||
reason: "zero-count-visible-dispatch",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("drops direct prepared turns with bot-loop protection before record and dispatch", async () => {
|
||||
const events: string[] = [];
|
||||
const log = vi.fn();
|
||||
|
||||
@@ -4,15 +4,21 @@ import {
|
||||
clearHistoryEntriesIfEnabled,
|
||||
recordPendingHistoryEntryWithMedia,
|
||||
} from "../../auto-reply/reply/history.js";
|
||||
import type { FinalizedMsgContext } from "../../auto-reply/templating.js";
|
||||
import {
|
||||
createDiagnosticTraceContextFromActiveScope,
|
||||
runWithDiagnosticTraceContext,
|
||||
} from "../../infra/diagnostic-trace-context.js";
|
||||
import { createSubsystemLogger } from "../../logging/subsystem.js";
|
||||
import { toHistoryMediaEntries } from "../inbound-event/media.js";
|
||||
import { createChannelReplyPipeline } from "../message/reply-pipeline.js";
|
||||
import type { CreateChannelReplyPipelineParams } from "../message/reply-pipeline.js";
|
||||
import { recordChannelBotPairLoopAndCheckSuppression } from "./bot-loop-protection.js";
|
||||
import { EMPTY_CHANNEL_TURN_DISPATCH_COUNTS } from "./dispatch-result.js";
|
||||
import {
|
||||
EMPTY_CHANNEL_TURN_DISPATCH_COUNTS,
|
||||
hasVisibleChannelTurnDispatch,
|
||||
type ChannelTurnDispatchResultLike,
|
||||
} from "./dispatch-result.js";
|
||||
import {
|
||||
deliverInboundReplyWithMessageSendContext,
|
||||
isDurableInboundReplyDeliveryHandled,
|
||||
@@ -100,6 +106,7 @@ const DEFAULT_EVENT_CLASS: ChannelEventClass = {
|
||||
kind: "message",
|
||||
canStartAgentTurn: true,
|
||||
};
|
||||
const log = createSubsystemLogger("channels/turn/kernel");
|
||||
|
||||
/**
|
||||
* @deprecated Compatibility assembly for legacy buffered reply dispatchers.
|
||||
@@ -268,6 +275,50 @@ function resolveObserveOnlyDispatchResult<TDispatchResult>(
|
||||
}) as TDispatchResult;
|
||||
}
|
||||
|
||||
function isSystemChannelTurn(ctx: FinalizedMsgContext): boolean {
|
||||
return (
|
||||
ctx.Provider === "heartbeat" || ctx.Provider === "cron-event" || ctx.Provider === "exec-event"
|
||||
);
|
||||
}
|
||||
|
||||
function maybeWarnZeroCountVisibleDispatch<TDispatchResult>(
|
||||
params: Pick<
|
||||
PreparedChannelTurn<TDispatchResult>,
|
||||
"admission" | "channel" | "ctxPayload" | "messageId" | "routeSessionKey"
|
||||
> & {
|
||||
dispatchResult: TDispatchResult;
|
||||
log?: (event: ChannelTurnLogEvent) => void;
|
||||
},
|
||||
): void {
|
||||
if (params.admission?.kind === "observeOnly" || isSystemChannelTurn(params.ctxPayload)) {
|
||||
return;
|
||||
}
|
||||
const dispatchResult = params.dispatchResult as ChannelTurnDispatchResultLike;
|
||||
// Suppress the silent-drop warning using the canonical visible-delivery signal, which
|
||||
// includes observedReplyDelivery and other non-count delivery paths. A partial count-only
|
||||
// check would falsely flag observed-path deliveries (queuedFinal=false, zero counts) as drops.
|
||||
if (hasVisibleChannelTurnDispatch(dispatchResult)) {
|
||||
return;
|
||||
}
|
||||
log.warn(
|
||||
`visible channel turn dispatched with no queued reply payloads: channel=${params.channel} ` +
|
||||
`messageId=${params.messageId ?? "unknown"} sessionKey=${
|
||||
params.ctxPayload.SessionKey ?? params.routeSessionKey
|
||||
}`,
|
||||
);
|
||||
emit({
|
||||
...params,
|
||||
event: {
|
||||
stage: "dispatch",
|
||||
event: "warning",
|
||||
messageId: params.messageId,
|
||||
sessionKey: params.ctxPayload.SessionKey ?? params.routeSessionKey,
|
||||
admission: params.admission?.kind ?? "dispatch",
|
||||
reason: "zero-count-visible-dispatch",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function isExplicitlyNonVisibleChannelDelivery(result: unknown): boolean {
|
||||
return (
|
||||
typeof result === "object" &&
|
||||
@@ -549,6 +600,11 @@ async function runPreparedChannelTurnCoreInTrace<
|
||||
options.suppressObserveOnlyDispatch && admission.kind === "observeOnly"
|
||||
? resolveObserveOnlyDispatchResult(params)
|
||||
: await params.runDispatch();
|
||||
maybeWarnZeroCountVisibleDispatch({
|
||||
...params,
|
||||
admission,
|
||||
dispatchResult,
|
||||
});
|
||||
} catch (err) {
|
||||
emit({
|
||||
...params,
|
||||
|
||||
@@ -416,7 +416,7 @@ export type ChannelTurnStage =
|
||||
/** Structured channel turn log event. */
|
||||
export type ChannelTurnLogEvent = {
|
||||
stage: ChannelTurnStage;
|
||||
event: "start" | "done" | "drop" | "handled" | "error";
|
||||
event: "start" | "done" | "drop" | "handled" | "error" | "warning";
|
||||
channel: string;
|
||||
accountId?: string;
|
||||
messageId?: string;
|
||||
|
||||
@@ -1085,7 +1085,7 @@ describe("gateway agent handler", () => {
|
||||
expect(capturedEntry?.sessionFile).toBeUndefined();
|
||||
});
|
||||
|
||||
it("keeps a failed session reusable when its default transcript exists", async () => {
|
||||
it("recovers a failed session when its default transcript exists", async () => {
|
||||
const now = Date.parse("2026-05-18T09:49:00.000Z");
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
dateOnlyFakeClockActive = true;
|
||||
@@ -1098,6 +1098,10 @@ describe("gateway agent handler", () => {
|
||||
const failedEntryWithDefaultTranscript = {
|
||||
sessionId: "failed-present-default-session-id",
|
||||
status: "failed",
|
||||
startedAt: now - 1_000,
|
||||
endedAt: now,
|
||||
runtimeMs: 1_000,
|
||||
abortedLastRun: true,
|
||||
updatedAt: now,
|
||||
sessionStartedAt: now,
|
||||
lastInteractionAt: now,
|
||||
@@ -1116,12 +1120,16 @@ describe("gateway agent handler", () => {
|
||||
const call = await waitForAgentCommandCall<{ sessionId?: string }>();
|
||||
expect(call.sessionId).toBe("failed-present-default-session-id");
|
||||
expect(capturedEntry?.sessionId).toBe("failed-present-default-session-id");
|
||||
expect(capturedEntry?.status).toBe("failed");
|
||||
expect(capturedEntry?.status).toBeUndefined();
|
||||
expect(capturedEntry?.startedAt).toBeUndefined();
|
||||
expect(capturedEntry?.endedAt).toBeUndefined();
|
||||
expect(capturedEntry?.runtimeMs).toBeUndefined();
|
||||
expect(capturedEntry?.abortedLastRun).toBeUndefined();
|
||||
expect(capturedEntry?.sessionFile).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps a failed session reusable when its relative transcript resolves and exists", async () => {
|
||||
it("recovers a failed session when its relative transcript resolves and exists", async () => {
|
||||
const now = Date.parse("2026-05-18T09:50:00.000Z");
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
dateOnlyFakeClockActive = true;
|
||||
@@ -1135,6 +1143,10 @@ describe("gateway agent handler", () => {
|
||||
sessionId: "failed-present-session-id",
|
||||
sessionFile: "relative-present.jsonl",
|
||||
status: "failed",
|
||||
startedAt: now - 1_000,
|
||||
endedAt: now,
|
||||
runtimeMs: 1_000,
|
||||
abortedLastRun: true,
|
||||
updatedAt: now,
|
||||
sessionStartedAt: now,
|
||||
lastInteractionAt: now,
|
||||
@@ -1153,7 +1165,11 @@ describe("gateway agent handler", () => {
|
||||
const call = await waitForAgentCommandCall<{ sessionId?: string }>();
|
||||
expect(call.sessionId).toBe("failed-present-session-id");
|
||||
expect(capturedEntry?.sessionId).toBe("failed-present-session-id");
|
||||
expect(capturedEntry?.status).toBe("failed");
|
||||
expect(capturedEntry?.status).toBeUndefined();
|
||||
expect(capturedEntry?.startedAt).toBeUndefined();
|
||||
expect(capturedEntry?.endedAt).toBeUndefined();
|
||||
expect(capturedEntry?.runtimeMs).toBeUndefined();
|
||||
expect(capturedEntry?.abortedLastRun).toBeUndefined();
|
||||
expect(capturedEntry?.sessionFile).toBe("relative-present.jsonl");
|
||||
});
|
||||
});
|
||||
@@ -1727,6 +1743,43 @@ describe("gateway agent handler", () => {
|
||||
});
|
||||
expect(mocks.agentCommand).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("recovers terminal failed agent API sessions without rotating the session id", async () => {
|
||||
const sessionId = "failed-agent-session";
|
||||
await withTempDir({ prefix: "openclaw-gateway-terminal-recovery-" }, async (root) => {
|
||||
const sessionsDir = `${root}/sessions`;
|
||||
await fs.mkdir(sessionsDir, { recursive: true });
|
||||
await fs.writeFile(`${sessionsDir}/${sessionId}.jsonl`, "", "utf8");
|
||||
mocks.loadSessionEntry.mockReturnValue({
|
||||
cfg: {},
|
||||
storePath: `${sessionsDir}/sessions.json`,
|
||||
entry: {
|
||||
sessionId,
|
||||
status: "failed",
|
||||
startedAt: 100,
|
||||
endedAt: 200,
|
||||
runtimeMs: 100,
|
||||
abortedLastRun: true,
|
||||
updatedAt: Date.now(),
|
||||
},
|
||||
canonicalKey: "agent:main:main",
|
||||
});
|
||||
|
||||
const capturedEntry = await runMainAgentAndCaptureEntry("recover-terminal-agent-session");
|
||||
const call = await waitForAgentCommandCall();
|
||||
|
||||
expect(call.sessionId).toBe(sessionId);
|
||||
expectRecordFields(capturedEntry, {
|
||||
sessionId,
|
||||
status: undefined,
|
||||
startedAt: undefined,
|
||||
endedAt: undefined,
|
||||
runtimeMs: undefined,
|
||||
abortedLastRun: undefined,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
it("does not restore a stale session id over a fresh store rotation (#5369)", async () => {
|
||||
mocks.resolveSessionLifecycleTimestamps.mockImplementation(
|
||||
({ entry }: { entry?: { sessionId?: string; sessionStartedAt?: number } }) => ({
|
||||
|
||||
@@ -174,6 +174,10 @@ import type {
|
||||
|
||||
const RESET_COMMAND_RE = /^\/(new|reset)(?:\s+([\s\S]*))?$/i;
|
||||
|
||||
function isRecoverableTerminalSessionStatus(status: SessionEntry["status"] | undefined): boolean {
|
||||
return status === "failed" || status === "timeout" || status === "killed";
|
||||
}
|
||||
|
||||
type AgentSendSessionLifecycleTransition = {
|
||||
cfg: OpenClawConfig;
|
||||
sessionKey: string;
|
||||
@@ -1767,6 +1771,10 @@ export const agentHandlers: GatewayRequestHandlers = {
|
||||
policy: resetPolicy,
|
||||
})
|
||||
: undefined;
|
||||
const visibleRequest =
|
||||
request.bootstrapContextRunKind !== "cron" &&
|
||||
request.bootstrapContextRunKind !== "heartbeat" &&
|
||||
!request.internalEvents?.length;
|
||||
const resolveFailedSessionTranscriptMissingForEntry = (
|
||||
candidateEntry: SessionEntry | undefined,
|
||||
) => {
|
||||
@@ -1834,7 +1842,7 @@ export const agentHandlers: GatewayRequestHandlers = {
|
||||
(!canReuseSession && !usableRequestedSessionId) ||
|
||||
Boolean(usableRequestedSessionId && entry?.sessionId !== usableRequestedSessionId);
|
||||
let rotatedSessionId = Boolean(entry?.sessionId && entry.sessionId !== sessionId);
|
||||
const touchInteraction = !isSystemGatewayRun && !request.internalEvents?.length;
|
||||
const touchInteraction = visibleRequest;
|
||||
const sessionAgent = canonicalSessionAgentId;
|
||||
type AgentSessionPatchBuild = {
|
||||
patch: Partial<SessionEntry>;
|
||||
@@ -1996,6 +2004,14 @@ export const agentHandlers: GatewayRequestHandlers = {
|
||||
? freshEntry?.sessionId
|
||||
: freshSessionId;
|
||||
const shouldClearRotatedState = freshRotatedSessionId && !freshSessionRotatedSinceLoad;
|
||||
const freshRecoverTerminalSession =
|
||||
freshCanReuseSession &&
|
||||
visibleRequest &&
|
||||
isRecoverableTerminalSessionStatus(freshEntry?.status);
|
||||
const shouldClearTerminalState =
|
||||
freshRecoverTerminalSession &&
|
||||
!freshSessionRotatedSinceLoad &&
|
||||
patchSessionId === freshEntry?.sessionId;
|
||||
const patch: Partial<SessionEntry> = {
|
||||
sessionId: patchSessionId,
|
||||
updatedAt: now,
|
||||
@@ -2024,14 +2040,14 @@ export const agentHandlers: GatewayRequestHandlers = {
|
||||
groupChannel: nextGroup.groupChannel,
|
||||
space: nextGroup.groupSpace,
|
||||
...(pluginOwnerId ? { pluginOwnerId } : {}),
|
||||
...(shouldClearRotatedState
|
||||
...(shouldClearRotatedState || shouldClearTerminalState
|
||||
? {
|
||||
status: undefined,
|
||||
startedAt: undefined,
|
||||
endedAt: undefined,
|
||||
runtimeMs: undefined,
|
||||
abortedLastRun: undefined,
|
||||
sessionFile: undefined,
|
||||
...(shouldClearRotatedState ? { sessionFile: undefined } : {}),
|
||||
}
|
||||
: {}),
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user