fix(agents): resume interrupted turns past progress commentary (#116725)

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<string, unknown>;
+  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 (

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<string, unknown> {
+  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" });
+  });
+});

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

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",