From b9d01e71270e15208d191e4ea4afdef31fbf51ac Mon Sep 17 00:00:00 2001 From: Vito Cappello Date: Thu, 27 Aug 2026 15:16:54 -0400 Subject: [PATCH] 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> --- .../run/attempt-recovery.test.ts | 177 ++++++++++++++++++ .../run/attempt-recovery.ts | 73 +++++++- .../run/context-recovery-state.ts | 1 + ...embedded-agent-subscribe.handlers.types.ts | 1 + .../embedded-agent-subscribe.run-state.ts | 1 + ...sion.subscribeembeddedagentsession.test.ts | 66 +++++++ src/agents/embedded-agent-subscribe.ts | 9 +- 7 files changed, 320 insertions(+), 8 deletions(-) diff --git a/src/agents/embedded-agent-runner/run/attempt-recovery.test.ts b/src/agents/embedded-agent-runner/run/attempt-recovery.test.ts index 6eb3bc74d714..2df70d5b8af4 100644 --- a/src/agents/embedded-agent-runner/run/attempt-recovery.test.ts +++ b/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[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), diff --git a/src/agents/embedded-agent-runner/run/attempt-recovery.ts b/src/agents/embedded-agent-runner/run/attempt-recovery.ts index 71cac7dacb65..e0269e895286 100644 --- a/src/agents/embedded-agent-runner/run/attempt-recovery.ts +++ b/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>; type NormalizedAttempt = Extract< Awaited>, @@ -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; } diff --git a/src/agents/embedded-agent-runner/run/context-recovery-state.ts b/src/agents/embedded-agent-runner/run/context-recovery-state.ts index c285b4cff206..132f4834b2d1 100644 --- a/src/agents/embedded-agent-runner/run/context-recovery-state.ts +++ b/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, }; } diff --git a/src/agents/embedded-agent-subscribe.handlers.types.ts b/src/agents/embedded-agent-subscribe.handlers.types.ts index b26fd188912d..452a434651d0 100644 --- a/src/agents/embedded-agent-subscribe.handlers.types.ts +++ b/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; diff --git a/src/agents/embedded-agent-subscribe.run-state.ts b/src/agents/embedded-agent-subscribe.run-state.ts index 64b67a704620..b054a5e69939 100644 --- a/src/agents/embedded-agent-subscribe.run-state.ts +++ b/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, diff --git a/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.subscribeembeddedagentsession.test.ts b/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.subscribeembeddedagentsession.test.ts index c63e84ac9688..9285ac12a211 100644 --- a/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.subscribeembeddedagentsession.test.ts +++ b/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 }) => { diff --git a/src/agents/embedded-agent-subscribe.ts b/src/agents/embedded-agent-subscribe.ts index 6d50de1b256b..9a5bf75ac7aa 100644 --- a/src/agents/embedded-agent-subscribe.ts +++ b/src/agents/embedded-agent-subscribe.ts @@ -70,7 +70,6 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess let lastAssistantUsage: ReturnType; 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; } };