fix(agents): continue settled tool batches after a transport drop (#130721)

* fix(agents): continue settled tool batches after a transport drop

A transient provider transport failure (WebSocket drop, socket reset) that
lands after every tool call in the batch has settled used to kill the turn:
any executed side-effecting tool marks the attempt replay-unsafe, and every
recovery path is gated on replay safety, so the runner surfaced the generic
"Agent couldn't generate a response" warning even though the tool results
were already persisted and nothing needed to be re-run. Under code mode every
tool call is exec, so any turn with a tool call died on a socket hiccup.

Continue such attempts from the persisted transcript with the existing
mid-turn continuation prompt, bounded to two continuations per run and only
for silent errored turns with transient evidence; prompt resubmission, profile
rotation, and model fallback stay closed for side-effecting attempts.

* fix(agents): narrow settled transport recovery

Require provider transport diagnostics before continuing a replay-unsafe settled tool batch, leaving rate-limit failures with the existing failover owner. Preserve the prior last-call usage only when a compaction retry fails before recording a replacement, without overwriting a newer successful call after later processing errors.

* fix(agents): restore retry usage only on failure

Keep the previous exact usage hidden while a compaction retry is active. Restore it only when the retry terminates with an error before recording a new call; a successful replacement call remains authoritative across later processing failures.

* fix(agents): require settled tools for transport resume

Keep the parked Code Mode exception scoped to overflow recovery. Provider transport recovery now requires every tool lifecycle item to be settled, with regression coverage for a suspended nested Code Mode run.

* fix(agents): fence suspended code mode recovery

Reject transcript continuation whenever the settled batch still carries producer-recorded suspended Code Mode work, even if the outer lifecycle count has reached zero. Cover that exact state in the recovery regression.

* fix(agents): resume parked code mode after transport drop

Use the canonical mid-turn settled evidence for transport continuation so producer-recorded parked Code Mode runs can resume from the current transcript. This lets the model call wait for the existing cell without replaying the original prompt or exec.

Cover parked active-count states and preserve the terminal, abort, timeout, visible-output, diagnostic, rate-limit, retry-budget, and no-fallback gates.

Worked on by:
- @VACInc

Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com>

---------

Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com>
Co-authored-by: roboclaw-bot <309084314+roboclaw-bot@users.noreply.github.com>

src/agents/embedded-agent-runner/run/attempt-recovery.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it, vi } from "vitest";
+import type { AssistantMessage } from "../../../llm/types.js";
 import {
   buildEmbeddedRunnerAssistant,
   createMockUsage,
@@ -10,7 +11,183 @@ import { recoverEmbeddedRunAttempt } from "./attempt-recovery.js";
 import { createEmbeddedRunContextRecoveryState } from "./context-recovery-state.js";
 import { resolveEmbeddedRunAttemptTerminalState } from "./terminal-outcome.js";
 
+type TransportDropScenario = {
+  errorMessage?: string;
+  content?: AssistantMessage["content"];
+  diagnostics?: AssistantMessage["diagnostics"];
+  activeCount?: number;
+  codeModeSuspended?: boolean;
+  transportDropContinuations?: number;
+  terminal?: Parameters<typeof makeEmbeddedRunnerAttempt>[0]["terminal"];
+  yieldDetected?: boolean;
+};
+
+// Live shape: a code-mode exec batch settled, then the ChatGPT Responses stream
+// died while the model was still reasoning, so the errored turn is thinking-only.
+async function recoverAfterTransportDrop(scenario: TransportDropScenario = {}) {
+  const toolCalls = ["call_1", "call_2"];
+  const toolAssistant = buildEmbeddedRunnerAssistant({
+    stopReason: "toolUse",
+    content: toolCalls.map((id) => ({ type: "toolCall", id, name: "exec", arguments: {} })),
+  });
+  const erroredAssistant = buildEmbeddedRunnerAssistant({
+    stopReason: "error",
+    errorMessage: scenario.errorMessage ?? "WebSocket error",
+    diagnostics:
+      scenario.diagnostics ??
+      ([
+        {
+          type: "provider_transport_failure",
+          error: { message: "WebSocket error" },
+          details: { phase: "after_message_stream_start" },
+        },
+      ] as never),
+    content: scenario.content ?? [{ type: "thinking", thinking: "checking the results" }],
+    usage: createMockUsage(0, 0),
+  });
+  const messagesSnapshot = [
+    { role: "user", content: "why is it unauthorized?" },
+    toolAssistant,
+    ...toolCalls.map((id) => ({ role: "toolResult", toolCallId: id, toolName: "exec" })),
+    erroredAssistant,
+  ] as never;
+  const attempt = makeEmbeddedRunnerAttempt({
+    messagesSnapshot,
+    toolMetas: toolCalls.map((toolCallId) => ({
+      toolCallId,
+      toolName: "exec",
+      replaySafe: false,
+      ...(scenario.codeModeSuspended ? { codeModeSuspended: true } : {}),
+    })) as never,
+    lastAssistant: erroredAssistant,
+    currentAttemptAssistant: erroredAssistant,
+    itemLifecycle: {
+      startedCount: toolCalls.length,
+      completedCount: toolCalls.length,
+      activeCount: scenario.activeCount ?? 0,
+    },
+    ...(scenario.terminal ? { terminal: scenario.terminal } : {}),
+    ...(scenario.yieldDetected ? { yieldDetected: true } : {}),
+  });
+  const terminalState = resolveEmbeddedRunAttemptTerminalState({
+    attempt,
+    assistant: erroredAssistant,
+  });
+  const continueFromCurrentTranscript = vi.fn();
+  const contextRecoveryState = createEmbeddedRunContextRecoveryState();
+  contextRecoveryState.transportDropContinuations = scenario.transportDropContinuations ?? 0;
+  const failoverRetryController = {
+    resolveAuthProfileFailureReason: vi.fn(),
+    advanceAuthProfile: vi.fn(),
+    advanceRateLimitAuthProfile: vi.fn(),
+    maybeMarkAuthProfileFailure: vi.fn(),
+    maybeBackoffBeforeOverloadFailover: vi.fn(),
+  };
+  const recovery = await recoverEmbeddedRunAttempt({
+    runInput: {
+      runParams: {
+        config: {},
+        agentId: "main",
+        sessionId: "session:transport-drop",
+        runId: "run:transport-drop",
+      },
+      resolvedSessionKey: "agent:main:transport-drop",
+      startedAtMs: Date.now(),
+      laneController: { throwIfAborted: vi.fn() },
+    },
+    preparedRuntime: {
+      provider: "openai",
+      modelId: "gpt-5.6-luna",
+      model: { id: "gpt-5.6-luna" },
+      genericCompactionRecoveryAllowed: false,
+      snapshot: () => ({
+        thinkLevel: "off",
+        agentHarness: { id: "openclaw" },
+        outerContextTokenMeta: {},
+        pluginHarnessOwnsTransport: false,
+      }),
+    },
+    normalizedAttempt: {
+      attempt,
+      sessionIdUsed: attempt.sessionIdUsed,
+      attemptAssistant: erroredAssistant,
+      currentAttemptAssistant: erroredAssistant,
+      currentAttemptCompletedAssistant: undefined,
+      terminalState,
+      setTerminalLifecycleMeta: vi.fn(),
+      attemptCompactionCount: 0,
+      activeErrorContext: { provider: "openai", model: "gpt-5.6-luna" },
+      resolveReplayInvalidForAttempt: () => true,
+      canRestartForLiveSwitch: false,
+    },
+    runtimePlan: { auth: {} },
+    sessionPromptState: { sessionFile: "/tmp/session.jsonl", continueFromCurrentTranscript },
+    failoverRetryController,
+    compactionRuntime: {},
+    contextRecoveryState,
+    usageAccumulator: createUsageAccumulator(),
+    lastRunPromptUsage: undefined,
+    runtimeAuthRetry: false,
+    codexAppServerRecoveryRetryAvailable: false,
+    codexAppServerRecoveryRetries: 0,
+    lastRetryFailoverReason: null,
+    traceAttempts: [],
+    sessionAgentId: "main",
+  } as never);
+  return { recovery, continueFromCurrentTranscript, contextRecoveryState, failoverRetryController };
+}
+
 describe("recoverEmbeddedRunAttempt", () => {
+  it("continues from the transcript after a transient transport drop on a settled exec batch", async () => {
+    const {
+      recovery,
+      continueFromCurrentTranscript,
+      contextRecoveryState,
+      failoverRetryController,
+    } = await recoverAfterTransportDrop();
+
+    expect(recovery).toMatchObject({ action: "retry" });
+    expect(contextRecoveryState.transportDropContinuations).toBe(1);
+    expect(continueFromCurrentTranscript).toHaveBeenCalledTimes(1);
+    expect(failoverRetryController.advanceAuthProfile).not.toHaveBeenCalled();
+    expect(failoverRetryController.maybeMarkAuthProfileFailure).not.toHaveBeenCalled();
+  });
+
+  it.each([0, 1])(
+    "continues a parked Code Mode run from its persisted waiting result with activeCount=%i",
+    async (activeCount) => {
+      const { recovery, continueFromCurrentTranscript } = await recoverAfterTransportDrop({
+        codeModeSuspended: true,
+        activeCount,
+      });
+
+      expect(recovery).toMatchObject({ action: "retry" });
+      expect(continueFromCurrentTranscript).toHaveBeenCalledTimes(1);
+    },
+  );
+
+  it.each<[string, TransportDropScenario]>([
+    ["the exec batch is still running", { activeCount: 1 }],
+    ["the run was externally aborted", { terminal: { kind: "aborted", source: "external" } }],
+    ["the run timed out", { terminal: { kind: "timeout", phase: "prompt", source: "runtime" } }],
+    ["the attempt already has terminal state", { yieldDetected: true }],
+    ["the assistant error is not transient", { errorMessage: "invalid request: bad schema" }],
+    [
+      "the failure is retryable but not a transport drop",
+      { errorMessage: "429 rate limit exceeded; retry after 2 seconds", diagnostics: [] },
+    ],
+    [
+      "the errored turn already carried visible text",
+      { content: [{ type: "text", text: "Partial" }] },
+    ],
+    ["the continuation budget is spent", { transportDropContinuations: 2 }],
+  ])("keeps the replay gate closed when %s", async (_label, scenario) => {
+    const { recovery, continueFromCurrentTranscript } = await recoverAfterTransportDrop(scenario);
+
+    expect(recovery).toEqual({ action: "proceed", shouldSurfaceCodexCompletionTimeout: false });
+    expect(continueFromCurrentTranscript).not.toHaveBeenCalled();
+  });
+
   it("surfaces before_agent_run blocks with current carried usage", async () => {
     const historicalAssistant = buildEmbeddedRunnerAssistant({
       usage: createMockUsage(128_814, 3_000),

src/agents/embedded-agent-runner/run/attempt-recovery.ts

@@ -1,9 +1,12 @@
 import { formatErrorMessage, toErrorObject } from "../../../infra/errors.js";
+import type { AssistantMessage } from "../../../llm/types.js";
+import { isRetryableAssistantError } from "../../../llm/utils/retry.js";
 import { projectAgentRunAttemptTerminal } from "../../agent-run-terminal-outcome.js";
 import { DEFAULT_MODEL, DEFAULT_PROVIDER } from "../../defaults.js";
 import type { FailoverReason } from "../../embedded-agent-helpers.js";
 import { LiveSessionModelSwitchError } from "../../live-model-switch-error.js";
 import { shouldSwitchToLiveModel, clearLiveModelSwitchPending } from "../../live-model-switch.js";
+import { hasOnlyAssistantReasoningContent } from "../../replay-turn-classification.js";
 import type { normalizeUsage } from "../../usage.js";
 import { log } from "../logger.js";
 import { getEmbeddedSessionPromptState } from "../session-prompt-state.js";
@@ -11,7 +14,11 @@ import type { EmbeddedAgentRunResult, TraceAttempt } from "../types.js";
 import type { createUsageAccumulator } from "../usage-accumulator.js";
 import type { prepareAndDispatchEmbeddedRunAttempt } from "./attempt-dispatch-preparation.js";
 import type { normalizeEmbeddedRunAttempt } from "./attempt-normalization.js";
-import { hasAsyncActivity, isCurrentAttemptReplaySafe } from "./attempt-terminal-evidence.js";
+import {
+  hasAsyncActivity,
+  hasAttemptTerminalState,
+  isCurrentAttemptReplaySafe,
+} from "./attempt-terminal-evidence.js";
 import { buildEmbeddedRunBlockedResult } from "./blocked-run-result.js";
 import { resolveCodexAppServerRecoveryRetry } from "./codex-app-server-recovery.js";
 import { resolveCompactionLiveModelSelection } from "./compaction-live-model-selection.js";
@@ -28,6 +35,22 @@ import type { createEmbeddedRunSessionPromptState } from "./session-prompt-state
 import { isEmbeddedRunTerminalInterrupted } from "./terminal-outcome.js";
 import { recoverEmbeddedRunTimeout } from "./timeout-context-recovery.js";
 
+const MAX_TRANSPORT_DROP_CONTINUATIONS = 2;
+
+/** Errored assistant turn with transient transport evidence and no visible output. */
+function isSilentTransportDropAssistant(assistant: AssistantMessage | undefined): boolean {
+  if (
+    !assistant ||
+    assistant.stopReason !== "error" ||
+    !isRetryableAssistantError(assistant) ||
+    !assistant.diagnostics?.some((diagnostic) => diagnostic.type === "provider_transport_failure")
+  ) {
+    return false;
+  }
+  const content = Array.isArray(assistant.content) ? assistant.content : [];
+  return content.length === 0 || hasOnlyAssistantReasoningContent(assistant);
+}
+
 type PreparedRuntime = Awaited<ReturnType<typeof prepareEmbeddedRunRuntime>>;
 type NormalizedAttempt = Extract<
   Awaited<ReturnType<typeof normalizeEmbeddedRunAttempt>>,
@@ -126,6 +149,26 @@ export async function recoverEmbeddedRunAttempt(input: {
     attempt.preflightRecovery?.source === "mid-turn" &&
     midTurnBatchSettled &&
     !hasAsyncActivity(attempt.toolMetas);
+  // A transient transport failure that lands after the whole tool batch settled
+  // is a resume, not a replay: the continuation prompt re-enters after the
+  // persisted tool results and nothing from the failed attempt is resubmitted.
+  // Only a silent errored assistant qualifies; partial visible text would be
+  // duplicated or replaced. Everything #122516 closed for side-effecting
+  // attempts (prompt resubmission, profile rotation, model fallback) stays
+  // closed below this branch.
+  const settledTransportDropAssistant =
+    !currentAttemptReplaySafe &&
+    !promptError &&
+    !aborted &&
+    !timedOut &&
+    !terminalInterrupted &&
+    !hasAttemptTerminalState(attempt) &&
+    midTurnBatchSettled &&
+    // A parked Code Mode result is persisted same-session state. Continuing is
+    // how the model reaches wait; it does not resubmit the prompt or exec call.
+    isSilentTransportDropAssistant(currentAttemptAssistant)
+      ? currentAttemptAssistant
+      : undefined;
   const { signalOwnedInterruption } = terminalState;
   const assistantOverflowCandidate =
     currentAttemptCompletedAssistant !== undefined
@@ -184,7 +227,11 @@ export async function recoverEmbeddedRunAttempt(input: {
       }),
     };
   }
-  if (!currentAttemptReplaySafe && !canContinueSettledMidTurnOverflow) {
+  if (
+    !currentAttemptReplaySafe &&
+    !canContinueSettledMidTurnOverflow &&
+    !settledTransportDropAssistant
+  ) {
     return replayUnsafeOutcome;
   }
 
@@ -313,8 +360,26 @@ export async function recoverEmbeddedRunAttempt(input: {
       }),
     };
   }
-  // Settled-tool continuation authorizes only current-transcript overflow recovery.
-  // Every path below can replay or replace the original attempt and remains fail-closed.
+  const recoveryState = input.contextRecoveryState;
+  if (
+    settledTransportDropAssistant &&
+    recoveryState.transportDropContinuations < MAX_TRANSPORT_DROP_CONTINUATIONS
+  ) {
+    runInput.laneController.throwIfAborted();
+    recoveryState.transportDropContinuations += 1;
+    sessionPromptState.continueFromCurrentTranscript();
+    log.warn(
+      `provider transport dropped after a settled tool batch; continuing from the transcript ` +
+        `attempt=${recoveryState.transportDropContinuations}/${MAX_TRANSPORT_DROP_CONTINUATIONS} ` +
+        `provider=${preparedRuntime.provider} model=${preparedRuntime.modelId} ` +
+        `error=${settledTransportDropAssistant.errorMessage?.trim() ?? "unknown"} ` +
+        `runId=${params.runId} sessionId=${params.sessionId}`,
+    );
+    return retry();
+  }
+  // Settled-tool continuation authorizes only current-transcript overflow and
+  // transport-drop recovery. Every path below can replay or replace the original
+  // attempt and remains fail-closed.
   if (!currentAttemptReplaySafe) {
     return replayUnsafeOutcome;
   }

src/agents/embedded-agent-runner/run/context-recovery-state.ts

@@ -8,6 +8,7 @@ export function createEmbeddedRunContextRecoveryState() {
     overflowCompactionAttempts: 0,
     timeoutCompactionAttempts: 0,
     toolResultTruncationAttempted: false,
+    transportDropContinuations: 0,
   };
 }
 

src/agents/embedded-agent-subscribe.handlers.types.ts

@@ -178,6 +178,7 @@ export type EmbeddedAgentSubscribeState = {
   lastReasoningSent?: string;
   pendingAssistantUsage?: NormalizedUsage;
   assistantUsageCommitted: boolean;
+  retryUsage?: NormalizedUsage;
 
   compactionInFlight: boolean;
   lastCompactionTokensAfter?: number;

src/agents/embedded-agent-subscribe.run-state.ts

@@ -63,6 +63,7 @@ export function createEmbeddedAgentSubscribeState(
     lastReasoningSent: undefined,
     pendingAssistantUsage: undefined,
     assistantUsageCommitted: false,
+    retryUsage: undefined,
     compactionInFlight: false,
     lastCompactionTokensAfter: undefined,
     pendingCompactionRetry: 0,

src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.subscribeembeddedagentsession.test.ts

@@ -425,6 +425,72 @@ describe("subscribeEmbeddedAgentSession", () => {
     });
   });
 
+  it("keeps a successful retry call when later post-call processing fails", () => {
+    const { emit, subscription } = createSubscribedSessionHarness({ runId: "run" });
+
+    emit({ type: "message_start", message: { role: "assistant" } });
+    emit({
+      type: "message_end",
+      message: {
+        role: "assistant",
+        usage: { input: 100, output: 20, totalTokens: 120 },
+      },
+    });
+    emit(retryingCompactionEnd());
+    emit({ type: "message_start", message: { role: "assistant" } });
+    emit({
+      type: "message_end",
+      message: {
+        role: "assistant",
+        usage: { input: 240, output: 30, totalTokens: 270 },
+      },
+    });
+    emit({ type: "message_start", message: { role: "assistant" } });
+    emit({
+      type: "message_end",
+      message: {
+        role: "assistant",
+        stopReason: "error",
+        usage: makeZeroUsageSnapshot(),
+      },
+    });
+
+    expect(subscription.getLastAssistantUsage()).toEqual({
+      input: 240,
+      output: 30,
+      total: 270,
+    });
+  });
+
+  it("restores the previous call when a retry fails before recording usage", () => {
+    const { emit, subscription } = createSubscribedSessionHarness({ runId: "run" });
+
+    emit({ type: "message_start", message: { role: "assistant" } });
+    emit({
+      type: "message_end",
+      message: {
+        role: "assistant",
+        usage: { input: 100, output: 20, totalTokens: 120 },
+      },
+    });
+    emit(retryingCompactionEnd());
+    emit({ type: "message_start", message: { role: "assistant" } });
+    emit({
+      type: "message_end",
+      message: {
+        role: "assistant",
+        stopReason: "error",
+        usage: makeZeroUsageSnapshot(),
+      },
+    });
+
+    expect(subscription.getLastAssistantUsage()).toEqual({
+      input: 100,
+      output: 20,
+      total: 120,
+    });
+  });
+
   it.each(THINKING_TAG_CASES)(
     "streams <%s> reasoning via onReasoningStream without leaking into final text",
     async ({ open, close }) => {

src/agents/embedded-agent-subscribe.ts

@@ -70,7 +70,6 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
   let lastAssistantUsage: ReturnType<typeof normalizeUsage>;
   let compactionCount = 0;
   let currentAttemptAssistant: AssistantMessage | undefined;
-
   const assistantTexts = state.assistantTexts;
   const toolMetas = state.toolMetas;
   const toolMetaById = state.toolMetaById;
@@ -283,7 +282,7 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
       total: usageTotals.total || derivedTotal || undefined,
     };
   };
-  const getLastAssistantUsage = () => (lastAssistantUsage ? { ...lastAssistantUsage } : undefined);
+  const getLastAssistantUsage = () => normalizeUsage(lastAssistantUsage);
   const incrementCompactionCount = () => {
     compactionCount += 1;
   };
@@ -451,9 +450,9 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
     state.deterministicApprovalPromptSent = false;
     state.lastDeliveredBlockReplyText = undefined;
     state.toolExecutionSinceLastBlockReply = false;
-    // A retry is a new model attempt. A silent retry must not inherit the
-    // completed assistant or pre-compaction context snapshot.
+    // Keep prior usage until the retry records its own call or terminal error.
     currentAttemptAssistant = undefined;
+    state.retryUsage = lastAssistantUsage ?? state.retryUsage;
     lastAssistantUsage = undefined;
     state.replayState = mergeEmbeddedRunReplayState(state.replayState, params.initialReplayState);
     state.livenessState = "working";
@@ -470,6 +469,8 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
       // Context-engine projection may later replace or mutate transcript
       // objects. Final delivery needs the model event owned by this run.
       currentAttemptAssistant = structuredClone(msg) as AssistantMessage;
+      lastAssistantUsage ??= msg.stopReason === "error" ? state.retryUsage : undefined;
+      state.retryUsage = undefined;
     }
   };