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; } };