diff --git a/src/auto-reply/reply/agent-runner-execution.test.ts b/src/auto-reply/reply/agent-runner-execution.test.ts index b6e5c7c30599..09ccb93e3557 100644 --- a/src/auto-reply/reply/agent-runner-execution.test.ts +++ b/src/auto-reply/reply/agent-runner-execution.test.ts @@ -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(), }, }; } diff --git a/src/auto-reply/reply/agent-runner-memory.test.ts b/src/auto-reply/reply/agent-runner-memory.test.ts index 051a3ea7af8c..72ed1c495f80 100644 --- a/src/auto-reply/reply/agent-runner-memory.test.ts +++ b/src/auto-reply/reply/agent-runner-memory.test.ts @@ -46,6 +46,7 @@ function createReplyOperation(): TestReplyOperation { sessionId: "session", abortSignal: new AbortController().signal, resetTriggered: false, + terminalRecovery: false, phase: "queued", result: null, setPhase: vi.fn(), @@ -61,6 +62,7 @@ function createReplyOperation(): TestReplyOperation { fail: vi.fn(), abortByUser: vi.fn(), abortForRestart: vi.fn(), + markTerminalRecovery: vi.fn(), }; } diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index 0e2290113275..45cd0e160d96 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -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((release) => { + releaseFirstTurn = release; + }); + let signalFirstResolverEntered: () => void = () => {}; + const firstTurnEntered = new Promise((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((release) => { + releaseFirstTurn = release; + }); + let signalFirstResolverEntered: () => void = () => {}; + const firstTurnEntered = new Promise((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 | 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 | undefined; + let signalFreshRegistered: () => void = () => {}; + const freshRegistered = new Promise((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(); diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index de6cc647f787..bfe96e2764ff 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -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 { 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; diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts index 0521888f2e33..8d66a3784762 100644 --- a/src/auto-reply/reply/reply-run-registry.ts +++ b/src/auto-reply/reply/reply-run-registry.ts @@ -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, @@ -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; diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts index a0df25c52da9..c8b08a042b4c 100644 --- a/src/auto-reply/reply/session.test.ts +++ b/src/auto-reply/reply/session.test.ts @@ -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-"); diff --git a/src/auto-reply/reply/session.ts b/src/auto-reply/reply/session.ts index 0e1d63c253da..12b3656b071d 100644 --- a/src/auto-reply/reply/session.ts +++ b/src/auto-reply/reply/session.ts @@ -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, diff --git a/src/channels/turn/kernel.test.ts b/src/channels/turn/kernel.test.ts index 9be6df577f40..1d0ecfdf7840 100644 --- a/src/channels/turn/kernel.test.ts +++ b/src/channels/turn/kernel.test.ts @@ -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(); diff --git a/src/channels/turn/kernel.ts b/src/channels/turn/kernel.ts index 885e03e3b5a7..5dcfd1581de3 100644 --- a/src/channels/turn/kernel.ts +++ b/src/channels/turn/kernel.ts @@ -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( }) as TDispatchResult; } +function isSystemChannelTurn(ctx: FinalizedMsgContext): boolean { + return ( + ctx.Provider === "heartbeat" || ctx.Provider === "cron-event" || ctx.Provider === "exec-event" + ); +} + +function maybeWarnZeroCountVisibleDispatch( + params: Pick< + PreparedChannelTurn, + "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, diff --git a/src/channels/turn/types.ts b/src/channels/turn/types.ts index 2cdf2f33a730..16fa13c9a71f 100644 --- a/src/channels/turn/types.ts +++ b/src/channels/turn/types.ts @@ -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; diff --git a/src/gateway/server-methods/agent.test.ts b/src/gateway/server-methods/agent.test.ts index 11a95244a16a..d22a3de52125 100644 --- a/src/gateway/server-methods/agent.test.ts +++ b/src/gateway/server-methods/agent.test.ts @@ -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 } }) => ({ diff --git a/src/gateway/server-methods/agent.ts b/src/gateway/server-methods/agent.ts index a6d257fb8570..57a88d11283a 100644 --- a/src/gateway/server-methods/agent.ts +++ b/src/gateway/server-methods/agent.ts @@ -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; @@ -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 = { 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 } : {}), } : {}), };