diff --git a/src/agents/embedded-agent-runner/message-visibility.ts b/src/agents/embedded-agent-runner/message-visibility.ts index 40dde7bedeae..4eb4d1c4869f 100644 --- a/src/agents/embedded-agent-runner/message-visibility.ts +++ b/src/agents/embedded-agent-runner/message-visibility.ts @@ -3,6 +3,7 @@ import { isSilentReplyText, SILENT_REPLY_TOKEN, } from "../../auto-reply/tokens.js"; +import { resolveAssistantMessagePhase } from "../../shared/chat-message-content.js"; type AgentPayloadLike = { text?: unknown; @@ -184,6 +185,33 @@ export function isMeaningfulTranscriptMessage(message: unknown): boolean { return Boolean(role && role !== "system"); } +/** Recognizes persisted progress without mistaking an ordinary assistant answer for completion. */ +export function isIntermediateAssistantTranscriptMessage(message: unknown): boolean { + if ( + !message || + typeof message !== "object" || + getTranscriptMessageRole(message) !== "assistant" + ) { + return false; + } + const record = message as Record; + if (record.stopReason !== undefined && record.stopReason !== "stop") { + return false; + } + const phase = resolveAssistantMessagePhase(message); + if (phase !== undefined) { + return phase === "commentary"; + } + const fallback = record.openclawStreamFallback; + if (!fallback || typeof fallback !== "object" || Array.isArray(fallback)) { + return false; + } + const { itemId, source } = fallback as { itemId?: unknown; source?: unknown }; + // Keyed segments are durable progress items; unkeyed/current fallbacks can + // become the final answer and must never bypass restart completion checks. + return source === "segment" && typeof itemId === "string" && itemId.trim().length > 0; +} + /** Returns whether a stopped assistant turn contains only reasoning and a silent marker. */ export function isTerminalSilentAssistantMessage(message: unknown): boolean { if ( diff --git a/src/agents/main-session-restart-recovery-resume-policy.test.ts b/src/agents/main-session-restart-recovery-resume-policy.test.ts new file mode 100644 index 000000000000..de0282b0853d --- /dev/null +++ b/src/agents/main-session-restart-recovery-resume-policy.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it, vi } from "vitest"; +import { resolveMainSessionResumePolicy } from "./main-session-restart-recovery-resume-policy.js"; + +vi.mock("./code-mode-control-tools.js", () => ({ + CODE_MODE_EXEC_TOOL_NAME: "exec", + CODE_MODE_WAIT_TOOL_NAME: "wait", +})); + +vi.mock("./tool-replay-safety.js", () => ({ + isAgentToolReplaySafe: ({ name }: { name?: string }) => name === "read", +})); + +vi.mock("./run-termination.js", () => ({ + AGENT_RUN_RESTART_ABORT_ERROR: "agent run aborted for restart", + AGENT_RUN_RESTART_ABORT_ERROR_CODE: "OPENCLAW_RESTART_ABORT", +})); + +function progressMessage(text: string, itemId: string): Record { + return { + role: "assistant", + content: [{ type: "text", text }], + stopReason: "stop", + openclawStreamFallback: { + replacementText: text, + source: "segment", + itemId, + }, + }; +} + +describe("resolveMainSessionResumePolicy progress tails", () => { + it("resumes when keyed progress messages arrive after the recovery mark", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + progressMessage("Checking the owner boundary.", "progress-1"), + progressMessage("Still tracing the restart lifecycle.", "progress-2"), + ]), + ).toEqual({ action: "resume", forceRestartSafeTools: false }); + }); + + it("resumes explicit commentary without making completed answers resumable", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + { + role: "assistant", + phase: "commentary", + content: [{ type: "text", text: "Checking the workspace." }], + stopReason: "stop", + }, + ]), + ).toEqual({ action: "resume", forceRestartSafeTools: false }); + + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + { role: "assistant", content: [{ type: "text", text: "The work is complete." }] }, + progressMessage("A later progress item.", "progress-late"), + ]), + ).toEqual({ action: "fail", reason: "transcript tail is not resumable" }); + }); + + it("recognizes the existing provider text-signature commentary contract", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + { + role: "assistant", + content: [ + { + type: "text", + text: "Checking the workspace.", + textSignature: JSON.stringify({ v: 1, id: "progress-signed", phase: "commentary" }), + }, + ], + stopReason: "stop", + }, + ]), + ).toEqual({ action: "resume", forceRestartSafeTools: false }); + }); + + it("keeps restart abort artifacts effective when progress arrives on either side", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + progressMessage("One last update before cancellation.", "progress-before-abort"), + { + role: "assistant", + content: [], + stopReason: "aborted", + errorMessage: "agent run aborted for restart", + }, + progressMessage("One delayed update after cancellation.", "progress-after-abort"), + ]), + ).toEqual({ action: "resume", forceRestartSafeTools: false }); + }); + + it("retains replay restrictions when progress follows a side-effecting tool call", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + { + role: "assistant", + stopReason: "toolUse", + content: [ + { type: "toolCall", id: "call-bash", name: "bash", arguments: { command: "true" } }, + ], + }, + progressMessage("Waiting for the command.", "progress-exec"), + ]), + ).toEqual({ action: "resume", forceRestartSafeTools: true }); + }); + + it("never treats unkeyed stream fallbacks as authoritative progress", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + { + role: "assistant", + content: [{ type: "text", text: "Possibly final output." }], + stopReason: "stop", + openclawStreamFallback: { replacementText: "Possibly final output.", source: "current" }, + }, + ]), + ).toEqual({ action: "fail", reason: "transcript tail is not resumable" }); + }); + + it("keeps explicit final-answer phase authoritative over keyed fallback metadata", () => { + expect( + resolveMainSessionResumePolicy([ + { role: "user", content: "finish the interrupted work" }, + { + ...progressMessage("The work is complete.", "final-item"), + phase: "final_answer", + }, + ]), + ).toEqual({ action: "fail", reason: "transcript tail is not resumable" }); + }); +}); diff --git a/src/agents/main-session-restart-recovery-resume-policy.ts b/src/agents/main-session-restart-recovery-resume-policy.ts index fa21d5f1f588..1cbc1063ab77 100644 --- a/src/agents/main-session-restart-recovery-resume-policy.ts +++ b/src/agents/main-session-restart-recovery-resume-policy.ts @@ -3,6 +3,7 @@ import type { InternalSessionEntry as SessionEntry } from "../config/sessions.js import { CODE_MODE_EXEC_TOOL_NAME, CODE_MODE_WAIT_TOOL_NAME } from "./code-mode-control-tools.js"; import { getTranscriptMessageRole as getMessageRole, + isIntermediateAssistantTranscriptMessage, isMeaningfulTranscriptMessage, readTerminalSourceReplyDeliveryMirror, } from "./embedded-agent-runner/message-visibility.js"; @@ -403,7 +404,16 @@ export function resolveMainSessionResumePolicy( } // `admitted` means no optional hook started. The dispatch boundary reloads // the current hook set before it permits this transcript to resume. - const meaningfulMessages = messages.toReversed().filter(isMeaningfulTranscriptMessage); + // Progress can commit after the recovery mark while the old run is winding + // down. It is not a terminal turn boundary; preserve it in the transcript + // while classifying the actual user/tool/assistant boundary beneath it. + const meaningfulMessages = messages + .toReversed() + .filter( + (message) => + isMeaningfulTranscriptMessage(message) && + !isIntermediateAssistantTranscriptMessage(message), + ); // A restart abort tail without tool calls is lifecycle noise whether or not // partial streamed text was persisted with it; the partial output stays in // the transcript for the continuation, and the message beneath decides diff --git a/src/agents/main-session-restart-recovery.test.ts b/src/agents/main-session-restart-recovery.test.ts index 7ff4efd9f911..32fd9fa0695e 100644 --- a/src/agents/main-session-restart-recovery.test.ts +++ b/src/agents/main-session-restart-recovery.test.ts @@ -863,6 +863,71 @@ describe("main-session-restart-recovery", () => { expect(store["agent:main:main"]?.abortedLastRun).toBe(false); }); + it("resumes when durable commentary is mirrored after the restart recovery mark", async () => { + const sessionsDir = await makeSessionsDir(); + const sessionKey = "agent:main:main"; + await writeStore(sessionsDir, { + [sessionKey]: runningSessionEntry("main-session"), + }); + await writeTranscript(sessionsDir, "main-session", [ + { role: "user", content: "finish the interrupted long-running turn" }, + ]); + + await expect( + markRestartAbortedMainSessions({ + stateDir: tmpDir, + sessionKeys: [sessionKey], + reason: "gateway restart drain", + }), + ).resolves.toEqual({ marked: 1, skipped: 0 }); + + await writeTranscript(sessionsDir, "main-session", [ + { + role: "assistant", + content: [{ type: "text", text: "Checking the remaining background task." }], + stopReason: "stop", + openclawStreamFallback: { + replacementText: "Checking the remaining background task.", + source: "segment", + itemId: "progress-after-recovery-mark", + }, + }, + { + role: "assistant", + content: [{ type: "text", text: "The restart handoff is in progress." }], + stopReason: "stop", + openclawStreamFallback: { + replacementText: "The restart handoff is in progress.", + source: "segment", + itemId: "progress-after-recovery-mark-2", + }, + }, + ]); + + await expectRecovery({ recovered: 1, failed: 0, skipped: 0 }); + expect(callGateway).toHaveBeenCalledOnce(); + expect(gatewayParams().sessionKey).toBe(sessionKey); + expect(readStore(path.join(sessionsDir, "sessions.json"))[sessionKey]).toMatchObject({ + status: "running", + abortedLastRun: false, + }); + + const transcript = await loadTestTranscript( + sessionKey, + path.join(sessionsDir, "sessions.json"), + ); + expect( + transcript + .map((event) => event.message) + .filter( + (message) => + message?.role === "assistant" && + (message as { openclawStreamFallback?: { source?: unknown } }).openclawStreamFallback + ?.source === "segment", + ), + ).toHaveLength(2); + }); + it.each([ { label: "same-process lifecycle rotation",