diff --git a/.agents/skills/senpi-qa/scripts/claude-sdk-oauth-stream-stall-retry-probe.mjs b/.agents/skills/senpi-qa/scripts/claude-sdk-oauth-stream-stall-retry-probe.mjs new file mode 100644 index 0000000000..b3e9487204 --- /dev/null +++ b/.agents/skills/senpi-qa/scripts/claude-sdk-oauth-stream-stall-retry-probe.mjs @@ -0,0 +1,201 @@ +#!/usr/bin/env node +// Issue #723 real-surface probe: a stream-start-stalled headless turn must retry +// by forking at the pre-turn assistant boundary (delta re-send, prefix-cache +// read), never by re-attaching and re-sending the full conversation. + +import { spawn } from "node:child_process"; +import { mkdirSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { bootHermeticStack, SOURCE_ROOT } from "./lib/claude-sdk-oauth-fullstack-harness.mjs"; +import { classifyPayload } from "./lib/claude-sdk-oauth-fullstack-support.mjs"; +import { safeDetail } from "./lib/output-safety.mjs"; + +const TURN_TIMEOUT_MS = 120_000; +const tsxImport = fileURLToPath(import.meta.resolve("tsx")); +const evidenceArg = process.argv.indexOf("--evidence"); +const evidenceSlug = evidenceArg === -1 ? undefined : process.argv[evidenceArg + 1]; + +function killTree(child) { + if (child.exitCode !== null) return; + try { + if (process.platform === "win32") child.kill("SIGKILL"); + else process.kill(-child.pid, "SIGKILL"); + } catch { + child.kill("SIGKILL"); + } +} + +function runCli(stack, args, onLine) { + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, ["--import", tsxImport, join(SOURCE_ROOT, "cli.ts"), ...args], { + cwd: stack.box.cwd, + detached: process.platform !== "win32", + env: { ...process.env }, + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + stdout += chunk; + if (onLine) for (const line of chunk.split("\n")) onLine(line); + }); + child.stderr.on("data", (chunk) => { + stderr += chunk; + }); + const timeout = setTimeout(() => { + killTree(child); + reject(new Error(`CLI turn exceeded ${TURN_TIMEOUT_MS}ms`)); + }, TURN_TIMEOUT_MS); + child.once("error", (error) => { + clearTimeout(timeout); + reject(error); + }); + child.once("close", (code) => { + clearTimeout(timeout); + resolve({ code: code ?? -1, stdout, stderr }); + }); + }); +} + +function events(stdout, type) { + const out = []; + for (const line of stdout.split("\n")) { + if (!line.trim()) continue; + let parsed; + try { + parsed = JSON.parse(line); + } catch { + continue; + } + const visit = (value) => { + if (Array.isArray(value)) return value.forEach(visit); + if (!value || typeof value !== "object") return; + if (value.type === type) out.push(value); + for (const nested of Object.values(value)) visit(nested); + }; + visit(parsed); + } + return out; +} + +function seedFastTimeoutSettings(stack) { + writeFileSync( + join(stack.box.agentDir, "settings.json"), + JSON.stringify({ + retry: { enabled: true, baseDelayMs: 0, provider: { streamStartTimeoutMs: 3000, streamRetryTimeoutMs: 0 } }, + }), + ); +} + +const common = (sessionDir) => [ + "-p", + "--provider", "claude-sdk-oauth", + "--model", "claude-haiku-4-5", + "--thinking", "off", + "--mode", "json", + "--session-dir", sessionDir, + "--no-tools", + "--no-context-files", + "--offline", + "--no-model-fallback", + "--no-recommended-models", + "--system-prompt", "Reply with the requested marker only.", +]; + +let stack; +let summary; +const cleanup = []; +try { + stack = await bootHermeticStack({ sandboxLabel: "issue-723-stream-stall-retry" }); + seedFastTimeoutSettings(stack); + + // Turn 1: normal — establishes lineage + persisted binding. + const first = await runCli(stack, [...common(stack.box.sessionDir), "Reply exactly STALL-1."]); + if (first.code !== 0) throw new Error(`turn 1 failed (${first.code}): ${safeDetail(first.stderr.slice(-400))}`); + const payloadsBeforeStall = stack.creations.reduce((n, c) => n + c.payloads.length, 0); + + // Turn 2: stall the FIRST provider request, watch for the retry's payload. + const releaseStall = stack.stallNextResponse(); + const retryPayloads = []; + const second = await runCli( + stack, + [...common(stack.box.sessionDir), "-c", "Reply exactly STALL-2."], + (line) => { + if (line.includes("auto_retry_start")) { + // The stalled first attempt timed out and the retry is about to fire: + // let the loopback answer normally now. + releaseStall(); + } + }, + ); + // The stall only held ONE response; the retry streams a fresh body, so no + // further release is needed even if the retry-start line never surfaced. + releaseStall(); + + const secondTurnCreations = stack.creations.slice(1); // everything after the turn-1 resident query + const classified = secondTurnCreations.flatMap((c) => + c.payloads.map((message) => ({ path: c.path, lineage: c.lineage, forked: c.forked, resumeAt: c.resumeAt, ...classifyPayload(message) })), + ); + const continuity = events(second.stdout, "claude_sdk_oauth_session_continuity").map((e) => e.details ?? e); + const retries = events(second.stdout, "auto_retry_start").length; + // The decisive signals (issue #723): after the stall, every continuity + // decision resumes lineage (fork|delta|reattach) carrying ONLY the turn's own + // message (deltaMessages === 1); a flatten/bootstrap or a larger delta means + // the retry re-billed the conversation. + const noColdSeedAfterStall = continuity.every((o) => o.kind !== "flatten" && o.kind !== "bootstrap"); + const deltaOnlyAfterStall = continuity.length > 0 && continuity.every((o) => o.deltaMessages === 1); + const resumedWithFork = continuity.some((o) => o.kind === "fork" || o.kind === "reattach"); + const noFlattenPayload = classified.every((p) => p.kind !== "flatten"); + + const passed = + second.code === 0 && + retries >= 1 && + resumedWithFork && + noColdSeedAfterStall && + deltaOnlyAfterStall && + noFlattenPayload; + + summary = { + passed, + turn2: { code: second.code, retries, continuity }, + classified, + providerRequests: stack.providerRequests.length, + resumedWithFork, + noColdSeedAfterStall, + deltaOnlyAfterStall, + stderr: safeDetail(second.stderr.split("\n").filter(Boolean).slice(-6).join("\n")), + }; +} catch (error) { + summary = { passed: false, error: safeDetail(error instanceof Error ? error.stack : String(error)) }; +} finally { + if (stack) { + await stack.shutdown().catch(() => {}); + stack.authGuard.assertUnchanged(); + stack.box.cleanup(); + cleanup.push("loopback server closed", "sandbox removed", "real auth unchanged"); + } +} + +if (evidenceSlug) { + const directory = join( + process.cwd(), + "local-ignore", + "qa-evidence", + `${new Date().toISOString().slice(0, 10).replaceAll("-", "")}-${evidenceSlug}`, + ); + mkdirSync(directory, { recursive: true }); + writeFileSync(join(directory, "probe-resume.json"), `${JSON.stringify(summary, null, 2)}\n`); + writeFileSync( + join(directory, "probe-resume.log"), + `command: node .agents/skills/senpi-qa/scripts/claude-sdk-oauth-stream-stall-retry-probe.mjs --evidence ${evidenceSlug}\n` + + `cleanup: ${cleanup.join("; ")}\n\n${JSON.stringify(summary, null, 2)}\n`, + ); + process.stdout.write(`EVIDENCE ${join(directory, "probe-resume.log")}\n`); +} +const finalPassed = summary?.passed === true; +process.stdout.write(`${JSON.stringify(summary)}\n`); +process.stdout.write(finalPassed ? "VERDICT: PASS claude-sdk-oauth stream-stall retry continuity\n" : "VERDICT: FAIL claude-sdk-oauth stream-stall retry continuity\n"); +process.exitCode = finalPassed ? 0 : 1; diff --git a/.agents/skills/senpi-qa/scripts/lib/claude-sdk-oauth-fullstack-harness.mjs b/.agents/skills/senpi-qa/scripts/lib/claude-sdk-oauth-fullstack-harness.mjs index 0fb9b7f6fa..07eac9c0e0 100644 --- a/.agents/skills/senpi-qa/scripts/lib/claude-sdk-oauth-fullstack-harness.mjs +++ b/.agents/skills/senpi-qa/scripts/lib/claude-sdk-oauth-fullstack-harness.mjs @@ -9,6 +9,7 @@ */ import { createServer } from "node:http"; +import { mkdirSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { pathToFileURL } from "node:url"; import { guardRealAuth, makeSandbox, repoRoot, track } from "./common.mjs"; @@ -69,6 +70,13 @@ async function startLoopbackServer(onRequest, holdRelease) { response.end(sse); return; } + if (hold.stall === true) { + // Stream-start stall: headers flushed above, first SSE event withheld + // until release — the client stream-start watchdog must fire. + response.flushHeaders(); + void hold.release.then(() => response.end(sse)); + return; + } // Stream the opening events so the turn is genuinely in flight, tell the // phase it may act now, and finish only when it releases the hold. const { head, tail } = splitSseBody(sse); @@ -113,6 +121,23 @@ export async function bootHermeticStack({ sandboxLabel = "claude-sdk-fullstack-p ); seedProbeAgentDir(box.agentDir); + // Post-#969 the ambient lane is opt-in AND the SDK subprocess validates its own + // credential store, so seed the sandbox CLAUDE_CONFIG_DIR with a dummy OAuth blob + // in the exact shape `claude auth status` accepts. + const claudeConfigDir = join(box.dir, "claude-config"); + mkdirSync(claudeConfigDir, { recursive: true, mode: 0o700 }); + writeFileSync( + join(claudeConfigDir, ".credentials.json"), + JSON.stringify({ + claudeAiOauth: { + accessToken: "fullstack-probe-dummy-access", + refreshToken: "fullstack-probe-dummy-refresh", + expiresAt: 4102444800000, + scopes: ["user:inference", "user:profile", "user:sessions:claude_code"], + }, + }), + { mode: 0o600 }, + ); // The ambient auth lane hands the probe's own environment to the Claude Code // subprocess, so inherited credentials and proxies are scrubbed BEFORE the // hermetic pins are applied, and the result is asserted below. @@ -126,10 +151,11 @@ export async function bootHermeticStack({ sandboxLabel = "claude-sdk-fullstack-p PI_TELEMETRY: "0", ANTHROPIC_BASE_URL: baseUrl, ANTHROPIC_API_KEY: "fullstack-probe-dummy-key", + SENPI_CLAUDE_SDK_OAUTH_TOKEN_INJECTION: "ambient", + SENPI_CLAUDE_SDK_OAUTH_ENABLED: "1", CLAUDE_CONFIG_DIR: join(box.dir, "claude-config"), CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: "1", CLAUDE_CODE_DISABLE_TELEMETRY: "1", - SENPI_CLAUDE_SDK_OAUTH_TOKEN_INJECTION: "ambient", NO_PROXY: "127.0.0.1,localhost", no_proxy: "127.0.0.1,localhost", }); @@ -222,6 +248,21 @@ export async function bootHermeticStack({ sandboxLabel = "claude-sdk-fullstack-p }; return () => releaseHold?.(); }, + /** + * Stalls the NEXT loopback response after headers: the first SSE event is + * never written until release(), so the client stream-start watchdog fires. + * issue #723 timeout-retry probe. Release with the returned function. + */ + stallNextResponse() { + let releaseHold; + pendingHold = { + stall: true, + release: new Promise((resolve) => { + releaseHold = resolve; + }), + }; + return () => releaseHold?.(); + }, createAgentSession: () => createAgentSession({ cwd: box.cwd, agentDir: box.agentDir, noTools: "all", autoTitleSessions: false }), /** diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 0eed8571ff..e897862b5b 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -6,6 +6,14 @@ ### Fixed +- claude-sdk-oauth stream-start-timeout retries now fork the SDK conversation at the last assistant + boundary before the stalled turn instead of re-attaching and re-sending it, so each retry re-bills + only the turn's own message on a prefix cache read instead of re-writing the whole conversation + (fixes #723 retry-storm re-billing: cache writes grew ~8K per attempt, $25/6min, $1084/3days on + worker dispatch). A stalled first turn with no boundary to fork at re-seeds byte-identically, which + the provider serves from prefix cache after the first write. The retry watchdog cap semantics + (`streamRetryTimeoutMs` caps the retry continuation, reconciled to the granted stream-start guard) + are now documented on the setting itself. - The Cursor exec bridge fails closed when a session bridge has no captured owning run, and rechecks run ownership after awaited preflight work so a run that ends during an approval prompt cannot start a tool side effect afterward. diff --git a/packages/coding-agent/src/core/changes.md b/packages/coding-agent/src/core/changes.md index d5978b8fcd..8f71e1ed0c 100644 --- a/packages/coding-agent/src/core/changes.md +++ b/packages/coding-agent/src/core/changes.md @@ -1,5 +1,30 @@ # changes +## 2026-08-20 - streamRetryTimeoutMs docstring aligned with the reconciled watchdog (issue #723 lane) + +### What changed + +- `core/retry-fallback/settings.ts`: the `streamRetryTimeoutMs` interface comment now states the actual + post-2026-08-18 semantics — it caps the retry-CONTINUATION watchdog, reconciled to + `max(cap, streamStartTimeoutMs)` — instead of the stale "first-request liveness cap after a provider + timeout" wording. Comment-only; no behavior change. + +### Why + +- Issue #723 diagnosis (M3) read that comment and concluded the setting clamps the stream-start guard + itself. It does not: since the 2026-08-18 reconciliation the retry request keeps its full granted + guard and only the continuation watchdog takes this cap. A wrong comment on the exact knob a + retry-storm investigation reaches first sends the next diagnosis down the same dead end. + +### Why an extension could not handle it + +- The setting is a core `ProviderRetrySettings` field consumed by `core/provider-timeout-retry.ts`; the + doc contract lives with the interface. + +### Expected merge conflict zones + +- `core/retry-fallback/settings.ts` `ProviderRetrySettings` field list only (comment line). + ## 2026-08-20 - Append-only goal continuations and exponentially floored 429 waits ### What changed diff --git a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/changes.md b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/changes.md index 97e05a0146..a645edb90d 100644 --- a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/changes.md +++ b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/changes.md @@ -1,5 +1,57 @@ # claude-sdk-oauth +## 2026-08-20 - Same-turn timeout retries fork at the pre-turn boundary (issue #723) + +### What changed + +- `session-reattach.ts`: `ContinuityBinding` gained an optional in-memory-only `unansweredTurnDigest`. It + rides the existing clone/remember paths but is NEVER persisted: `storedBindingFromEntry` + (`session-binding.ts`) builds the sidecar record from an explicit field list, and the strict + `schemaVersion: 1` schema (`session-binding-store.ts`) rejects any record carrying it. + Tests: `test/claude-sdk-oauth-binding-store.test.ts` (round-trip rejects it). +- `session-turn-attempt.ts`: an attempt that pushed its user payload but ended aborted, failed, or + discarded now remembers a retry checkpoint binding anchored at the PRE-TURN boundary + (`bindingFromEntry(entry, hashes.slice(0, entry.sentCount))` + the attempted turn's full sent-stream + digest). Covers the `turn.aborted` resolution, the queue-failure (completion rejected) path, and + `discard()` before `closeSession`. +- `session-continuity.ts`: `decideFromBinding` gains a branch ahead of the existing prefix logic — when + the binding carries a checkpoint, the FULL current sent stream hashes to it, and the prefix at + `binding.sentCount` matches, it returns `fork` at `binding.lastAssistantUuid` (`reason: + "timeout_retry"`), or the cold-seed `flatten` with the same reason when no boundary exists (first + turn). A digest mismatch falls through to the pre-existing branches unchanged. +- `session-observability.ts`: `ContinuityReason` union and the sanitizer allowlist admit + `timeout_retry`. No new event types; one observation per main turn is preserved. +- Tests: `test/suite/regressions/723-claude-sdk-oauth-timeout-abort-retry-continuity.test.ts`, + `test/claude-sdk-oauth-continuity-decision.test.ts`, `test/claude-sdk-oauth-continuity-retry-checkpoint.test.ts`. + +### Why + +A stream-start-timeout abort closes the SDK session with the turn's user message already appended and + un-answered. The retry then re-attached to that lineage and appended the SAME message again — one + duplicate per attempt, ~8K tokens of cache re-billing per attempt, and for a first turn (no assistant + boundary, binding absent) a full re-flatten of the whole conversation at full price on every attempt + (issue #723: $25 per 6 minutes, $1084 over 3 days on worker dispatch). Forking at the pre-turn + boundary rewinds past the orphaned message, so the retry's request byte-layout matches the failed + attempt's and the provider serves it from prefix cache; a first turn re-seeds byte-identically + (flatten is a deterministic function of context), which is likewise cache-read after the first write. + +### Why an extension could not handle it + +- The retry checkpoint must be recorded where the attempt's outcome is known (`session-turn-attempt.ts`) + and consumed by the resident-lane continuity decision table (`session-continuity.ts`) — both are + internal to this builtin's resident session machinery; no extension hook observes attempt outcomes or + continuity bindings. + +### Expected merge conflict zones + +- `session-continuity.ts` in `decideFromBinding` (head of the function) — upstream continuity reworks + touch the same function. +- `session-turn-attempt.ts` attempt-outcome block and `discard()` — same file upstream reworked in the + 2026-08-01 continuity pass. +- `session-reattach.ts` `ContinuityBinding` field list. +- `session-observability.ts` `ContinuityReason` union tail and `SANITIZED_REASONS` set (mechanically + duplicated literals; both must gain the member). + ## 2026-08-19 - Kill-switched lane leaves implicit fallback expansion ### What changed diff --git a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-continuity.ts b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-continuity.ts index cfa6517fdc..40b214b041 100644 --- a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-continuity.ts +++ b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-continuity.ts @@ -25,6 +25,8 @@ export type ContinuityBindingSnapshot = { modelId: string; systemPromptHash: string; toolsetHash: string; + /** Sent-stream digest of a turn that was pushed but never answered (retry checkpoint). */ + unansweredTurnDigest?: string; }; export type ContinuityDecisionInput = { @@ -97,10 +99,45 @@ function identityDrift( return null; } +/** + * Same-turn retry after a stream-start timeout: the abandoned attempt already + * appended its user message to the lineage, so re-attaching would append it a + * SECOND time and re-bill the whole conversation. Forking at the pre-turn + * assistant boundary rewinds past the un-answered message, so the retry's + * request byte-layout matches the failed attempt's (prefix cache read). + * Requires the FULL current turn to hash-match the checkpoint, so a different + * turn falls through to the ordinary branches below. + */ +function retryCheckpointDecision( + input: ContinuityDecisionInput, + binding: ContinuityBindingSnapshot, +): ContinuityDecision | undefined { + if (binding.unansweredTurnDigest === undefined) return undefined; + if (sentHashPrefixDigest(input.currentHashes, input.currentHashes.length) !== binding.unansweredTurnDigest) { + return undefined; + } + if (input.currentHashes.length < binding.sentCount) return undefined; + const prefixMatches = + binding.sentPrefixHash !== undefined + ? sentHashPrefixDigest(input.currentHashes, binding.sentCount) === binding.sentPrefixHash + : commonPrefixLength(binding.sentHashes, input.currentHashes) === binding.sentCount; + if (!prefixMatches) return undefined; + if (!binding.lastAssistantUuid) return { kind: "flatten", reason: "timeout_retry" }; + return { + kind: "fork", + sdkSessionId: binding.sdkSessionId, + atUuid: binding.lastAssistantUuid, + from: binding.sentCount, + reason: "timeout_retry", + }; +} + function decideFromBinding(input: ContinuityDecisionInput, binding: ContinuityBindingSnapshot): ContinuityDecision { if (!input.transcriptAvailable) return { kind: "flatten", reason: "transcript_missing" }; const drift = identityDrift(input, binding); if (drift) return { kind: "flatten", reason: drift }; + const retry = retryCheckpointDecision(input, binding); + if (retry) return retry; if (binding.sentPrefixHash !== undefined) { const prefixMatches = input.currentHashes.length >= binding.sentCount && diff --git a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-observability.ts b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-observability.ts index 9ae02ffd41..7f1f1b7188 100644 --- a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-observability.ts +++ b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-observability.ts @@ -47,6 +47,7 @@ export type ContinuityReason = | "abort_timeout" | "extensions_removed" | "session_shutdown" + | "timeout_retry" | "other"; export type ContinuityObservation = { @@ -96,6 +97,7 @@ const SANITIZED_REASONS = new Set([ "abort_timeout", "extensions_removed", "session_shutdown", + "timeout_retry", "other", ]); diff --git a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-reattach.ts b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-reattach.ts index 679ed5dac1..c32ea157d3 100644 --- a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-reattach.ts +++ b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-reattach.ts @@ -17,6 +17,14 @@ export type ContinuityBinding = { toolsetHash: string; /** Assistant boundaries kept as entries so a later fork still has a resume point. */ assistantUuidByIndex?: readonly (readonly [number, string])[]; + /** + * Digest of the FULL sent stream an attempt pushed but never got answered + * (stream-start timeout abort/failure). Purely in-memory: it lets the SAME + * turn's retry fork at the pre-turn boundary instead of re-appending its user + * message to a lineage that already carries it. Never persisted — the sidecar + * schema is fixed at schemaVersion 1 and restart retries are out of scope. + */ + unansweredTurnDigest?: string; }; export type ReattachInput = { diff --git a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-turn-attempt.ts b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-turn-attempt.ts index c22f9e6da4..9ca8e3a95e 100644 --- a/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-turn-attempt.ts +++ b/packages/coding-agent/src/core/extensions/builtin/claude-sdk-oauth/session-turn-attempt.ts @@ -8,7 +8,7 @@ import { sessionRegistry, } from "./session-registry.ts"; import { submitSessionTurn } from "./session-registry-pump.ts"; -import { recordSyncedStream } from "./session-sync.ts"; +import { recordSyncedStream, sentHashPrefixDigest } from "./session-sync.ts"; type StagedContinuityDecision = { emit(): void }; @@ -22,6 +22,22 @@ function recordAssistantUuid(entry: ClaudeSdkOauthSessionEntry, sentCount: numbe } } +/** + * An attempt that pushed its user payload and then aborted or failed leaves that + * message on the lineage un-answered: `recordSyncedStream` never ran, so the + * entry still points at the PRE-TURN boundary. Remembering the binding at that + * boundary, tagged with the attempted turn's full sent-stream digest, lets the + * SAME turn's retry fork past the orphaned message instead of appending it + * twice (issue #723 retry storm). In-memory only — nothing here is persisted. + */ +function rememberRetryCheckpoint(entry: ClaudeSdkOauthSessionEntry, hashes: readonly string[]): void { + if (entry.sentCount < 0 || entry.sentCount > hashes.length) return; + rememberBinding({ + ...bindingFromEntry(entry, hashes.slice(0, entry.sentCount)), + unansweredTurnDigest: sentHashPrefixDigest(hashes, hashes.length), + }); +} + export function createSessionTurnAttempt( entry: ClaudeSdkOauthSessionEntry, message: SDKUserMessage["message"], @@ -51,12 +67,21 @@ export function createSessionTurnAttempt( if (!turn.aborted && successfulTurn(turn.messages)) { recordSyncedStream(entry, hashes); rememberBinding(bindingFromEntry(entry, hashes)); + } else { + rememberRetryCheckpoint(entry, hashes); } + } catch (error) { + // The queue failed (completion rejected: pump failure, query end, + // attribution error). The payload was still pushed, so the retry needs + // the same checkpoint the aborted path records. + rememberRetryCheckpoint(entry, hashes); + throw error; } finally { staged.emit(); } })(), discard: (): void => { + rememberRetryCheckpoint(entry, hashes); if (isCurrentGeneration(entry.senpiSessionId, generation)) { closeSession(entry.senpiSessionId, "attempt_discarded"); } diff --git a/packages/coding-agent/src/core/retry-fallback/settings.ts b/packages/coding-agent/src/core/retry-fallback/settings.ts index bc2612911a..90ae52b931 100644 --- a/packages/coding-agent/src/core/retry-fallback/settings.ts +++ b/packages/coding-agent/src/core/retry-fallback/settings.ts @@ -3,7 +3,7 @@ import type { FallbackChains } from "./chains.ts"; export interface ProviderRetrySettings { timeoutMs?: number; streamStartTimeoutMs?: number; - streamRetryTimeoutMs?: number; // first-request liveness cap after a provider timeout; default: 30000, 0 disables + streamRetryTimeoutMs?: number; // retry-continuation watchdog cap after a provider timeout; reconciled to max(cap, streamStartTimeoutMs) so a granted stream-start budget is never cut short; default: 30000, 0 disables maxRetries?: number; maxRetryDelayMs?: number; } diff --git a/packages/coding-agent/test/claude-sdk-oauth-binding-store.test.ts b/packages/coding-agent/test/claude-sdk-oauth-binding-store.test.ts index bc81f985a4..ec30a94fa9 100644 --- a/packages/coding-agent/test/claude-sdk-oauth-binding-store.test.ts +++ b/packages/coding-agent/test/claude-sdk-oauth-binding-store.test.ts @@ -51,6 +51,24 @@ describe("claude-sdk-oauth session binding store", () => { expect(stored?.sessionId).toBe("sess-abc-123"); }); + // Issue #723 added an in-memory-only `unansweredTurnDigest` to the process + // binding. The sidecar schema is strict and fixed at schemaVersion 1: a record + // carrying it must be REJECTED at write, and one smuggled onto disk must not + // survive a read, so restart retries can never resume off a stale checkpoint. + it("never persists the in-memory retry checkpoint field", async () => { + const { sessionFile } = makeSessionFixture(); + const withDigest = { ...record(sessionFile), unansweredTurnDigest: "5".repeat(64) } as StoredBinding; + + await expect(writeStoredBinding(sessionFile, withDigest)).rejects.toThrow(); + + writeFileSync(bindingSidecarPath(sessionFile), `${JSON.stringify(withDigest)}\n`, "utf8"); + await expect(readStoredBinding(sessionFile)).resolves.toBeUndefined(); + + await writeStoredBinding(sessionFile, record(sessionFile)); + const stored = await readStoredBinding(sessionFile); + expect(stored && "unansweredTurnDigest" in stored).toBe(false); + }); + it("rejects a malformed sidecar file", async () => { const { sessionFile } = makeSessionFixture(); const sidecar = bindingSidecarPath(sessionFile); diff --git a/packages/coding-agent/test/claude-sdk-oauth-continuity-decision.test.ts b/packages/coding-agent/test/claude-sdk-oauth-continuity-decision.test.ts index 4447054dd6..f18f9d550b 100644 --- a/packages/coding-agent/test/claude-sdk-oauth-continuity-decision.test.ts +++ b/packages/coding-agent/test/claude-sdk-oauth-continuity-decision.test.ts @@ -7,6 +7,20 @@ import { sentHashPrefixDigest } from "../src/core/extensions/builtin/claude-sdk- const FINGERPRINT = { systemPromptHash: "prompt-v1", toolsetHash: "tools-v1" }; +function restored(overrides: Partial> = {}) { + return { + sdkSessionId: "sdk-1", + sentCount: 2, + sentHashes: ["h1", "h2"], + lastAssistantUuid: "uuid-a2", + accountName: "primary", + modelId: "claude-opus-4-5", + systemPromptHash: FINGERPRINT.systemPromptHash, + toolsetHash: FINGERPRINT.toolsetHash, + ...overrides, + } satisfies NonNullable; +} + function resident(overrides: Partial = {}) { return { sdkSessionId: "sdk-1", @@ -198,6 +212,40 @@ describe("claude-sdk-oauth native continuity decisions", () => { expect(decision).toEqual({ kind: "flatten", reason: "sent_stream_diverged" }); }); + it("forks at the pre-turn boundary when the same turn is retried after a timeout abort", () => { + const decision = decideNativeContinuity( + input({ + entry: undefined, + binding: restored({ unansweredTurnDigest: sentHashPrefixDigest(["h1", "h2", "h3"]) }), + }), + ); + + expect(decision).toEqual({ + kind: "fork", + sdkSessionId: "sdk-1", + atUuid: "uuid-a2", + from: 2, + reason: "timeout_retry", + }); + }); + + it("cold-seeds a retried first turn that has no assistant boundary to fork at", () => { + const decision = decideNativeContinuity( + input({ + entry: undefined, + currentHashes: ["h1"], + binding: restored({ + sentCount: 0, + sentHashes: [], + lastAssistantUuid: null, + unansweredTurnDigest: sentHashPrefixDigest(["h1"]), + }), + }), + ); + + expect(decision).toEqual({ kind: "flatten", reason: "timeout_retry" }); + }); + it("flattens when a hash divergence has no assistant boundary to fork at", () => { const decision = decideNativeContinuity( input({ diff --git a/packages/coding-agent/test/claude-sdk-oauth-continuity-retry-checkpoint.test.ts b/packages/coding-agent/test/claude-sdk-oauth-continuity-retry-checkpoint.test.ts new file mode 100644 index 0000000000..1fd59c156d --- /dev/null +++ b/packages/coding-agent/test/claude-sdk-oauth-continuity-retry-checkpoint.test.ts @@ -0,0 +1,128 @@ +import { describe, expect, it } from "vitest"; +import { + type ContinuityDecisionInput, + decideNativeContinuity, +} from "../src/core/extensions/builtin/claude-sdk-oauth/session-continuity.ts"; +import { sentHashPrefixDigest } from "../src/core/extensions/builtin/claude-sdk-oauth/session-sync.ts"; + +/** + * Shadowing guards for the issue #723 retry checkpoint branch: it must fire ONLY + * for a re-send of the exact turn that was left un-answered, and must never + * outrank the fail-closed checks that precede it. + */ + +const FINGERPRINT = { systemPromptHash: "prompt-v1", toolsetHash: "tools-v1" }; + +function restored(overrides: Partial> = {}) { + return { + sdkSessionId: "sdk-1", + sentCount: 2, + sentHashes: ["h1", "h2"], + lastAssistantUuid: "uuid-a2", + accountName: "primary", + modelId: "claude-opus-4-5", + systemPromptHash: FINGERPRINT.systemPromptHash, + toolsetHash: FINGERPRINT.toolsetHash, + ...overrides, + } satisfies NonNullable; +} + +function input(overrides: Partial = {}): ContinuityDecisionInput { + return { + entry: undefined, + binding: restored({ unansweredTurnDigest: sentHashPrefixDigest(["h1", "h2", "h3"]) }), + currentHashes: ["h1", "h2", "h3"], + accountName: "primary", + modelId: "claude-opus-4-5", + fingerprint: FINGERPRINT, + transcriptAvailable: true, + ...overrides, + }; +} + +describe("claude-sdk-oauth retry checkpoint continuity", () => { + it("ignores a stale checkpoint once the conversation moved past that turn", () => { + const decision = decideNativeContinuity(input({ currentHashes: ["h1", "h2", "h3", "h4"] })); + + expect(decision).toMatchObject({ kind: "reattach", reason: "registry_miss", from: 2 }); + }); + + it("ignores a checkpoint whose pre-turn prefix no longer matches", () => { + const decision = decideNativeContinuity( + input({ + binding: restored({ unansweredTurnDigest: sentHashPrefixDigest(["h1", "h2-rewritten", "h3"]) }), + currentHashes: ["h1", "h2-rewritten", "h3"], + }), + ); + + // Falls through to the pre-existing divergence branch, which owns this shape. + expect(decision).toMatchObject({ kind: "fork", reason: "history_rolled_back", atUuid: "uuid-a2" }); + }); + + it("ignores a checkpoint whose restored prefix digest no longer matches", () => { + const decision = decideNativeContinuity( + input({ + binding: restored({ + sentHashes: [], + sentPrefixHash: sentHashPrefixDigest(["h1", "h2-elsewhere"]), + unansweredTurnDigest: sentHashPrefixDigest(["h1", "h2", "h3"]), + }), + }), + ); + + expect(decision).toEqual({ kind: "flatten", reason: "sent_stream_diverged" }); + }); + + it("honours a checkpoint carried on a restored prefix-digest binding", () => { + const decision = decideNativeContinuity( + input({ + binding: restored({ + sentHashes: [], + sentPrefixHash: sentHashPrefixDigest(["h1", "h2"]), + unansweredTurnDigest: sentHashPrefixDigest(["h1", "h2", "h3"]), + }), + }), + ); + + expect(decision).toMatchObject({ kind: "fork", atUuid: "uuid-a2", from: 2, reason: "timeout_retry" }); + }); + + it("does not let a checkpoint outrank an identity drift", () => { + expect(decideNativeContinuity(input({ modelId: "claude-sonnet-5" }))).toEqual({ + kind: "flatten", + reason: "model_changed", + }); + expect(decideNativeContinuity(input({ accountName: "secondary" }))).toEqual({ + kind: "flatten", + reason: "account_changed", + }); + }); + + it("does not let a checkpoint outrank a missing transcript", () => { + expect(decideNativeContinuity(input({ transcriptAvailable: false }))).toEqual({ + kind: "flatten", + reason: "transcript_missing", + }); + }); + + it("never overrides a live resident entry, which owns its own decision", () => { + const decision = decideNativeContinuity( + input({ + entry: { + sdkSessionId: "sdk-1", + accountName: "primary", + modelId: "claude-opus-4-5", + systemPromptHash: FINGERPRINT.systemPromptHash, + toolsetHash: FINGERPRINT.toolsetHash, + sentCount: 2, + sentHashes: ["h1", "h2"], + lastAssistantUuid: "uuid-a2", + assistantUuidByIndex: new Map([[2, "uuid-a2"]]), + pendingForkReason: null, + }, + }), + ); + + expect(decision).toEqual({ kind: "delta", from: 2 }); + }); +}); diff --git a/packages/coding-agent/test/suite/regressions/723-claude-sdk-oauth-timeout-abort-retry-continuity.test.ts b/packages/coding-agent/test/suite/regressions/723-claude-sdk-oauth-timeout-abort-retry-continuity.test.ts new file mode 100644 index 0000000000..5476935f94 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/723-claude-sdk-oauth-timeout-abort-retry-continuity.test.ts @@ -0,0 +1,287 @@ +import type { Api, AssistantMessage, Context, Model } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import { + type Options, + overrideSdkBoundary, + resetSdkBoundary, + type SDKMessage, + type SDKUserMessage, + type SdkQuery, + type SdkQueryHandle, +} from "../../../src/core/extensions/builtin/claude-sdk-oauth/sdk-boundary.ts"; +import { forgetBinding } from "../../../src/core/extensions/builtin/claude-sdk-oauth/session-reattach.ts"; +import { + closeSession, + overrideSessionRegistryBoundary, + resetSessionRegistryBoundary, +} from "../../../src/core/extensions/builtin/claude-sdk-oauth/session-registry.ts"; +import { streamClaudeSdkOauth } from "../../../src/core/extensions/builtin/claude-sdk-oauth/stream.ts"; + +/** + * Issue #723 mechanism M2: a stream-start-timeout abort must not make the + * SAME turn's retry pay for the conversation again. Mid-conversation the retry + * forks at the pre-turn assistant boundary, so the turn's user message is sent + * exactly once per lineage. A first turn has no boundary to fork at, so it + * re-seeds - but exactly once per attempt and byte-identically, which the + * provider serves from prefix cache instead of re-billing. + */ + +const SESSION_ID = "issue-723-retry-continuity"; +const FLATTEN_MARKER = ""; +const FLATTEN_PREAMBLE = "The above is the conversation history so far"; + +const model: Model = { + id: "claude-test", + name: "Claude test", + api: "claude-sdk-oauth", + provider: "claude-sdk-oauth", + baseUrl: "claude-sdk-oauth", + reasoning: true, + input: ["text", "image"], + cost: { input: 3, output: 15, cacheRead: 0.3, cacheWrite: 3 }, + contextWindow: 200_000, + maxTokens: 8_192, +}; + +function sdkMessage(value: unknown): SDKMessage { + return value as SDKMessage; +} + +/** Resident scripted query: answers each submission unless the turn is scripted to stall. */ +class ResidentQuery implements SdkQueryHandle, AsyncIterator { + readonly submitted: SDKUserMessage[] = []; + readonly options: Options; + closes = 0; + private readonly stalls: () => boolean; + private readonly onSubmit: (message: SDKUserMessage) => void; + private readonly queued: SDKMessage[] = []; + private readonly readers: Array<(value: IteratorResult) => void> = []; + + constructor( + prompt: AsyncIterable, + options: Options, + stalls: () => boolean, + onSubmit: (message: SDKUserMessage) => void, + ) { + this.options = options; + this.stalls = stalls; + this.onSubmit = onSubmit; + void this.consume(prompt); + } + + [Symbol.asyncIterator](): AsyncIterator { + return this; + } + + next(): Promise> { + const value = this.queued.shift(); + if (value) return Promise.resolve({ value, done: false }); + return new Promise((resolve) => this.readers.push(resolve)); + } + + /** Clean interrupt receipt: nothing left queued, so the abort must not taint the lineage. */ + async interrupt(): Promise { + return { still_queued: [] }; + } + + close(): void { + this.closes++; + for (const reader of this.readers.splice(0)) reader({ value: undefined, done: true }); + } + + private emit(message: SDKMessage): void { + const reader = this.readers.shift(); + if (reader) reader({ value: message, done: false }); + else this.queued.push(message); + } + + private async consume(prompt: AsyncIterable): Promise { + for await (const message of prompt) { + this.submitted.push(message); + this.onSubmit(message); + if (this.stalls()) continue; // stream-start timeout: accepted, never answered + const uuid = message.uuid ?? `submitted-${this.submitted.length}`; + const session = message.session_id; + this.emit(sdkMessage({ ...message, uuid, session_id: session, isReplay: true })); + this.emit( + sdkMessage({ + type: "assistant", + message: { id: `m-${uuid}`, type: "message", role: "assistant", content: [] }, + parent_tool_use_id: null, + uuid: `assistant-${uuid}`, + session_id: session, + }), + ); + this.emit( + sdkMessage({ + type: "result", + subtype: "success", + result: `answer-${this.submitted.length}`, + user_message_uuid: uuid, + uuid: `result-${uuid}`, + session_id: session, + }), + ); + } + } +} + +type Submission = { text: string; lineage: string; flattened: boolean }; + +/** Lineage identity as the SDK sees it: a fork mints a new branch, resume/seed do not. */ +function lineageOf(options: Options): string { + if (options.forkSession) return `fork:${String(options.resumeSessionAt)}`; + return String(options.resume ?? options.sessionId ?? "unknown"); +} + +function textFrom(message: SDKUserMessage): string { + const content = message.message.content; + if (typeof content === "string") return content; + return content.map((block) => (block.type === "text" ? block.text : "[image]")).join(""); +} + +function residentBoundary(stalledSubmissions: ReadonlySet) { + const queries: ResidentQuery[] = []; + const submissions: Submission[] = []; + const waiters: Array<{ count: number; resolve: () => void }> = []; + const query: SdkQuery = ({ prompt, options = {} }) => { + if (typeof prompt === "string") throw new Error("Expected streaming input"); + const resident: ResidentQuery = new ResidentQuery( + prompt, + options, + () => stalledSubmissions.has(submissions.length - 1), + (message) => { + const text = textFrom(message); + submissions.push({ + text, + lineage: lineageOf(resident.options), + flattened: text.includes(FLATTEN_MARKER) || text.includes(FLATTEN_PREAMBLE), + }); + for (const waiter of waiters.splice(0)) { + if (submissions.length >= waiter.count) waiter.resolve(); + else waiters.push(waiter); + } + }, + ); + queries.push(resident); + return resident; + }; + overrideSdkBoundary({ query }); + overrideSessionRegistryBoundary({ queryFactory: query }); + return { + queries, + submissions, + waitForSubmissions(count: number): Promise { + if (submissions.length >= count) return Promise.resolve(); + return new Promise((resolve) => waiters.push({ count, resolve })); + }, + }; +} + +function assistant(text: string, timestamp: number): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "claude-sdk-oauth", + provider: "claude-sdk-oauth", + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp, + }; +} + +function continuityKinds(diagnostics: AssistantMessage["diagnostics"]): string[] { + return (diagnostics ?? []) + .filter((diagnostic) => diagnostic.type === "claude_sdk_oauth_session_continuity") + .map((diagnostic) => String((diagnostic.details as { kind?: unknown } | undefined)?.kind)); +} + +function continuityReasons(diagnostics: AssistantMessage["diagnostics"]): string[] { + return (diagnostics ?? []) + .filter((diagnostic) => diagnostic.type === "claude_sdk_oauth_session_continuity") + .map((diagnostic) => String((diagnostic.details as { reason?: unknown } | undefined)?.reason)); +} + +async function runTurn(context: Context, signal?: AbortSignal): Promise { + return await streamClaudeSdkOauth(model, context, { + sessionId: SESSION_ID, + streamKind: "main", + ...(signal ? { signal } : {}), + }).result(); +} + +afterEach(() => { + closeSession(SESSION_ID, "test_cleanup"); + forgetBinding(SESSION_ID); + resetSessionRegistryBoundary(); + resetSdkBoundary(); +}); + +describe("issue #723 claude-sdk-oauth stream-start-timeout retry continuity", () => { + it("resumes the aborted turn's lineage instead of re-sending it", async () => { + const { submissions, waitForSubmissions } = residentBoundary(new Set([1])); + const user1 = { role: "user" as const, content: "first", timestamp: 1 }; + const user2 = { role: "user" as const, content: "second", timestamp: 3 }; + await runTurn({ messages: [user1] }); + + const turn2: Context = { messages: [user1, assistant("answer-1", 2), user2] }; + const abort = new AbortController(); + const stalled = runTurn(turn2, abort.signal); + await waitForSubmissions(2); + abort.abort(); + expect((await stalled).stopReason).toBe("aborted"); + + const retry = await runTurn(turn2); + const turnSends = submissions.slice(1); + + // Sub-defect (a): the retry re-sends the delta only, never the flattened conversation. + expect(retry.stopReason).toBe("stop"); + expect(turnSends.map((send) => send.text)).toEqual(turnSends.map(() => "second")); + expect(turnSends.filter((send) => send.flattened)).toEqual([]); + // Sub-defect (b): the retry stays on the established lineage. + expect(continuityKinds(retry.diagnostics)).toEqual([expect.stringMatching(/^(?:delta|reattach|fork)$/)]); + // Sub-defect (c): the aborted attempt already appended the delta to its + // lineage, so re-appending it there bills the turn twice and duplicates the + // user message. The retry must rewind (fork) or otherwise not re-append. + const perLineage = new Map(); + for (const send of turnSends) perLineage.set(send.lineage, (perLineage.get(send.lineage) ?? 0) + 1); + expect({ maxSendsPerLineage: Math.max(...perLineage.values()) }).toEqual({ maxSendsPerLineage: 1 }); + }, 10_000); + + it("re-seeds a stalled first turn byte-identically instead of storming", async () => { + const { submissions, waitForSubmissions } = residentBoundary(new Set([0])); + const turn1: Context = { messages: [{ role: "user", content: "first", timestamp: 1 }] }; + + const abort = new AbortController(); + const stalled = runTurn(turn1, abort.signal); + await waitForSubmissions(1); + abort.abort(); + expect((await stalled).stopReason).toBe("aborted"); + + const retry = await runTurn(turn1); + const coldSeeds = submissions.filter((send) => send.flattened); + + // Sub-defect (d): a first turn has no assistant boundary to fork at, so the + // retry must re-seed - but exactly ONCE per attempt (never a storm), and + // byte-identically, so the provider serves the repeat from prefix cache + // instead of re-billing the write. + expect(retry.stopReason).toBe("stop"); + expect({ coldSeedSends: coldSeeds.length, attempts: submissions.length }).toEqual({ + coldSeedSends: 2, + attempts: 2, + }); + expect(new Set(coldSeeds.map((send) => send.text)).size).toBe(1); + // Sub-defect (e): the retry is attributed to the same-turn timeout retry, not + // to a fresh unexplained bootstrap. + expect(continuityKinds(retry.diagnostics)).toEqual(["flatten"]); + expect(continuityReasons(retry.diagnostics)).toEqual(["timeout_retry"]); + }, 10_000); +}); diff --git a/packages/coding-agent/test/suite/regressions/723-provider-timeout-retry-request-identity.test.ts b/packages/coding-agent/test/suite/regressions/723-provider-timeout-retry-request-identity.test.ts new file mode 100644 index 0000000000..1a7fb9d641 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/723-provider-timeout-retry-request-identity.test.ts @@ -0,0 +1,119 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import { createHarness, getMessageText, type Harness } from "../harness.ts"; + +const STREAM_START_STALL_ERROR = "Provider stream start timed out after 90000ms"; + +function summarizeMessages(messages: unknown[]) { + return messages.map((message) => { + if (!message || typeof message !== "object") { + return { shape: typeof message, value: message }; + } + const record = message as { + role?: string; + stopReason?: string; + errorMessage?: string; + timestamp?: number; + content?: unknown; + }; + return { + role: record.role, + text: getMessageText(message), + stopReason: record.stopReason, + errorMessage: record.errorMessage, + timestamp: record.timestamp, + keys: Object.keys(record).sort(), + content: record.content, + }; + }); +} + +function describeUnknown(value: unknown): unknown { + if (value === undefined) return { type: "undefined" }; + if (value === null) return { type: "null" }; + if (typeof value === "function") return { type: "function", name: value.name }; + if (typeof value !== "object") return { type: typeof value, value }; + if (value instanceof AbortSignal) { + return { type: "AbortSignal", aborted: value.aborted }; + } + if (Array.isArray(value)) return { type: "array", length: value.length }; + return { type: "object", keys: Object.keys(value).sort() }; +} + +function describeOptions(options: unknown) { + if (!options || typeof options !== "object") return options; + const record = options as Record; + return Object.fromEntries( + Object.keys(record) + .sort() + .map((key) => [key, describeUnknown(record[key])]), + ); +} + +describe("issue #723 provider timeout retry request identity", () => { + const harnesses: Harness[] = []; + + afterEach(() => { + while (harnesses.length > 0) { + harnesses.pop()?.cleanup(); + } + }); + + it("retries a stream-start stall with the same provider request messages", async () => { + const harness = await createHarness({ + settings: { retry: { enabled: true, maxRetries: 2, baseDelayMs: 0 } }, + }); + harnesses.push(harness); + harness.setResponses([ + fauxAssistantMessage("seed-one"), + fauxAssistantMessage("seed-two"), + fauxAssistantMessage("", { + stopReason: "error", + errorMessage: STREAM_START_STALL_ERROR, + }), + fauxAssistantMessage("recovered after stall"), + ]); + + await harness.session.prompt("seed one"); + await harness.session.prompt("seed two"); + await harness.session.prompt("trigger stall"); + + const calls = harness.faux.getCallLog(); + expect(harness.faux.state.callCount).toBe(4); + expect(calls).toHaveLength(4); + expect( + harness.eventsOfType("auto_retry_start").map((event) => ({ + attempt: event.attempt, + maxAttempts: event.maxAttempts, + delayMs: event.delayMs, + errorMessage: event.errorMessage, + })), + ).toEqual([ + { + attempt: 1, + maxAttempts: 2, + delayMs: 0, + errorMessage: STREAM_START_STALL_ERROR, + }, + ]); + expect(harness.eventsOfType("auto_retry_end").map((event) => event.success)).toEqual([true]); + + const failed = calls[2]; + const retried = calls[3]; + const failedMessages = failed.context.messages; + const retriedMessages = retried.context.messages; + + // Discovery dump: print failed vs retried arrays (and next-layer options) before identity. + console.log("=== #723 failed vs retried request messages ==="); + console.log("failed.length", failedMessages.length, "retried.length", retriedMessages.length); + console.log("failed.summaries", JSON.stringify(summarizeMessages(failedMessages), null, 2)); + console.log("retried.summaries", JSON.stringify(summarizeMessages(retriedMessages), null, 2)); + console.log("failed.raw", JSON.stringify(failedMessages, null, 2)); + console.log("retried.raw", JSON.stringify(retriedMessages, null, 2)); + console.log("failed.options", JSON.stringify(describeOptions(failed.options), null, 2)); + console.log("retried.options", JSON.stringify(describeOptions(retried.options), null, 2)); + console.log("call.timestamps", failed.timestamp, retried.timestamp); + + expect(retriedMessages).toEqual(failedMessages); + }); +});