From 143a9d232a1d863d4ef42cc32e764eda456ceb4f Mon Sep 17 00:00:00 2001 From: ZouR Ma <37286405+ZouR-Ma@users.noreply.github.com> Date: Wed, 21 Oct 2026 17:32:26 +0800 Subject: [PATCH] fix(compaction): retry incomplete streams and persist safe failure diagnostics --- packages/coding-agent/docs/compaction.md | 10 ++ .../coding-agent/src/core/agent-session.ts | 40 +++++++- .../src/core/compaction/compaction.ts | 88 ++++++++++++++--- .../core/compaction/summary-diagnostics.ts | 84 ++++++++++++++++ packages/coding-agent/test/compaction.test.ts | 5 +- .../test/footer-data-provider.test.ts | 4 + .../test/subagent-invocation.test.ts | 4 + .../suite/agent-session-compaction.test.ts | 4 +- ...tion-retries-transient-stream-drop.test.ts | 95 ++++++++++++++++++- .../regressions/compaction-integrity.test.ts | 16 +++- packages/providers/README.md | 2 + .../providers/src/api/anthropic-messages.ts | 24 +++-- packages/providers/src/utils/error-kind.ts | 1 + .../test/anthropic-sse-parsing.test.ts | 51 ++++++++++ 14 files changed, 394 insertions(+), 34 deletions(-) create mode 100644 packages/coding-agent/src/core/compaction/summary-diagnostics.ts diff --git a/packages/coding-agent/docs/compaction.md b/packages/coding-agent/docs/compaction.md index 22b23f60..d3b23a36 100644 --- a/packages/coding-agent/docs/compaction.md +++ b/packages/coding-agent/docs/compaction.md @@ -34,6 +34,16 @@ The native pipeline has two summarization mechanisms: Both use the same structured summary format and track file operations cumulatively. Compaction and branch-summary requests use fresh routing session IDs and, where supported by the provider, disable prompt-cache writes because these one-off prompts are unlikely to be reused. +## Failure diagnostics + +Native compaction failures include a safe configuration snapshot in the error shown by TUI, RPC, and SDK. The snapshot reports the runtime model/provider/API, context window, model output limit, selected summary output budget, observed payload output budget when available, approximate input size, reasoning option, request attempts and elapsed time, stop reason and error kind. It also includes compaction reserve/retention and retry settings. + +`selectedMaxTokens` is the caller's budget; `observedPayloadMaxTokens` is the numeric output cap observed through the provider payload callback after context/adapter adjustments. Custom streams that do not invoke this callback report `unknown`. Input estimates use serialized characters/4, not the provider tokenizer. Configuration provenance is currently `unknown`; values describe effective runtime settings, not a guessed disk configuration. + +Non-cancelled failures append a versioned `step-compaction-failure` custom entry to the session JSONL (when persistence is enabled). This entry is not added to model context and contains no failed summary, prompt, headers, URL, or environment dump. Credential-shaped error text is redacted. Failure to persist diagnostics does not replace the original error. Cancelled requests do not create failure entries. + +A `length` response remains an incomplete summary and is never saved as a successful checkpoint. Check model output limits and input headroom rather than assuming increasing `reserveTokens` increases the output cap. Transient transport failures use the existing bounded retry policy; deterministic malformed responses and cancellations do not become network retries. + ## Compaction ### When It Triggers diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 2bd70504..bc861100 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -51,6 +51,7 @@ import { streamSimple, } from "@step-harness/providers/compat"; import { getAgentDir } from "../config.ts"; +import { redactSecretString } from "../step/secret-redaction.ts"; import { getThemeByName, theme } from "../theme/theme.ts"; import { stripFrontmatter } from "../utils/frontmatter.ts"; import { sleep } from "../utils/sleep.ts"; @@ -79,6 +80,7 @@ import { type LiveContextStatus, } from "./compaction/live-context/manager.ts"; import { StableContextNotices } from "./compaction/live-context/stable-notices.ts"; +import { formatRetryPolicy, formatSummaryDiagnostics, SummaryFailure } from "./compaction/summary-diagnostics.ts"; import { DEFAULT_THINKING_LEVEL, THINKING_LEVEL_OPTIONS } from "./defaults.ts"; import { exportSessionToHtml, type ToolHtmlRenderer } from "./export-html/index.ts"; import { createToolHtmlRenderer } from "./export-html/tool-renderer.ts"; @@ -1080,6 +1082,33 @@ export class AgentSession { }); } + private _recordCompactionFailure(error: unknown, reason: "manual" | "threshold" | "overflow"): string { + const settings = this.settingsManager.getCompactionSettings(); + const retry = this.settingsManager.getRetrySettings(); + const message = redactSecretString(error instanceof Error ? error.message : String(error)); + const diagnostics = error instanceof SummaryFailure ? error.diagnostics : undefined; + const model = this.model; + const configuration = `Compaction settings: reserveTokens=${settings.reserveTokens}, keepRecentTokens=${settings.keepRecentTokens}; ${formatRetryPolicy(retry)}; configurationSource=unknown.`; + const request = diagnostics + ? formatSummaryDiagnostics(diagnostics) + : `Selected model (request not observed): provider=${model?.provider ?? "unknown"}, model=${model?.id ?? "unknown"}, contextWindow=${model?.contextWindow ?? "unknown"}, model.maxTokens=${model?.maxTokens ?? "unknown"}.`; + let result = redactSecretString(`${message}\n${request}\n${configuration}`); + try { + this.sessionManager.appendCustomEntry("step-compaction-failure", { + version: 1, + reason, + errorMessage: message, + diagnostics, + settings: { reserveTokens: settings.reserveTokens, keepRecentTokens: settings.keepRecentTokens }, + retry: { enabled: retry.enabled, maxRetries: retry.maxRetries, baseDelayMs: retry.baseDelayMs }, + configurationSource: "unknown", + }); + } catch { + result += "\nCould not persist compaction diagnostics; original failure retained."; + } + return result; + } + private async _emitSessionCompactFailed(event: Omit): Promise { if (this._extensionRunner.hasHandlers("session_compact_failed")) { await this._extensionRunner.emit({ type: "session_compact_failed", ...event }); @@ -2709,7 +2738,9 @@ export class AgentSession { } catch (error) { const message = error instanceof Error ? error.message : String(error); const aborted = message === "Compaction cancelled" || (error instanceof Error && error.name === "AbortError"); - const errorMessage = aborted ? undefined : `Compaction failed: ${message}`; + const errorMessage = aborted + ? undefined + : `Compaction failed: ${this._recordCompactionFailure(error, "manual")}`; this._compactionAbortController = undefined; this._emit({ type: "compaction_end", @@ -2726,6 +2757,7 @@ export class AgentSession { willRetry: false, fromExtension, }); + if (!aborted && errorMessage) throw new Error(errorMessage, { cause: error }); throw error; } finally { this._compactionAbortController = undefined; @@ -3077,14 +3109,14 @@ export class AgentSession { // Continue once so queued messages are delivered. return this.agent.hasQueuedMessages(); } catch (error) { - const errorMessage = error instanceof Error ? error.message : "compaction failed"; if (started) { const aborted = this._autoCompactionAbortController?.signal.aborted ?? false; + const failureDetails = aborted ? undefined : this._recordCompactionFailure(error, reason); const formattedErrorMessage = aborted ? undefined : reason === "overflow" - ? `Context overflow recovery failed: ${errorMessage}` - : `Auto-compaction failed: ${errorMessage}`; + ? `Context overflow recovery failed: ${failureDetails}` + : `Auto-compaction failed: ${failureDetails}`; this._emit({ type: "compaction_end", reason, diff --git a/packages/coding-agent/src/core/compaction/compaction.ts b/packages/coding-agent/src/core/compaction/compaction.ts index 0b92ec9a..6b7fc0d0 100644 --- a/packages/coding-agent/src/core/compaction/compaction.ts +++ b/packages/coding-agent/src/core/compaction/compaction.ts @@ -31,6 +31,14 @@ import { type SkillInstructionContext, stripSkillInstructions, } from "./skill-instructions.ts"; +import { + createSummaryDiagnostics, + getSummaryDiagnostics, + observeSummaryPayload, + SummaryFailure, + saveSummaryDiagnostics, + summaryFailure, +} from "./summary-diagnostics.ts"; import { createSummarySourceReducer } from "./summary-overflow.ts"; import { computeFileLists, @@ -585,8 +593,8 @@ ${SUMMARY_DETAIL_RULES}`; /** * Returns an error message when a summarization response cannot safely be persisted. * A length stop contains partial text and must not become a session checkpoint. - * When the model reported its output cap, it is surfaced in the message so - * operators can see which limit was hit and tune `reserveTokens` accordingly. + * The displayed budget is selected by the caller, not read back from the HTTP payload. + * Effective request limits are recorded separately in the failure diagnostics. */ export function getSummarizationFailure( response: AssistantMessage, @@ -600,7 +608,9 @@ export function getSummarizationFailure( return `${label} failed: ${response.errorMessage || "Unknown error"}`; } if (response.stopReason === "length") { - const capNote = maxTokens ? ` (${maxTokens}-token output cap; raise reserveTokens if this recurs)` : ""; + const capNote = maxTokens + ? ` (${maxTokens}-token selected output cap; check model output limits and request context headroom)` + : ""; return `${label} failed: generation hit the token cap and the summary is incomplete${capNote}`; } return undefined; @@ -640,16 +650,40 @@ export async function completeSummarization( ): Promise { // Avoid cache writes for one-off summaries. Reuse caller-supplied routing when available; // callers without a session ID, including branch summaries, receive a fresh routing ID. + const diagnostics = createSummaryDiagnostics(model, context, options); + const startedAt = Date.now(); const requestOptions: SimpleStreamOptions = { ...options, cacheRetention: "none", sessionId: options.sessionId ?? uuidv7(), + onPayload: async (payload, requestModel) => { + const replacement = await options.onPayload?.(payload, requestModel); + diagnostics.observedMaxTokens = observeSummaryPayload(replacement ?? payload); + return replacement; + }, }; - const produce = async (): Promise => - streamFn + const produce = async (): Promise => { + diagnostics.attempts++; + diagnostics.observedMaxTokens = undefined; + return streamFn ? (await streamFn(model, context, requestOptions)).result() : completeSimple(model, context, requestOptions); - return retryAssistantCall(produce, retry, requestOptions.signal, callbacks); + }; + let response: AssistantMessage; + try { + response = await retryAssistantCall(produce, retry, requestOptions.signal, callbacks); + } catch (error) { + if (requestOptions.signal?.aborted) throw new DOMException("Summarization aborted", "AbortError"); + diagnostics.durationMs = Date.now() - startedAt; + throw new SummaryFailure(error instanceof Error ? error.message : String(error), diagnostics); + } + diagnostics.durationMs = Date.now() - startedAt; + diagnostics.stopReason = response.stopReason; + diagnostics.errorKind = response.errorKind; + diagnostics.rawStopReason = response.rawStopReason; + diagnostics.usage = response.usage; + saveSummaryDiagnostics(response, diagnostics); + return response; } /** @@ -735,6 +769,9 @@ async function completeSummaryWithOverflowRecovery( let reductions = 0; let reduceSource: ReturnType; let usage: Usage | undefined; + let lastResponse: AssistantMessage | undefined; + let attempts = 0; + const startedAt = Date.now(); const requestOptions = { ...options, sessionId: options.sessionId ?? uuidv7() }; // A local recovery/cancellation failure can end produce() while a transient retry is @@ -756,6 +793,10 @@ async function completeSummaryWithOverflowRecovery( for (;;) { if (options.signal?.aborted) throw new Error("Compaction cancelled"); const response = await completeSummarization(model, context, requestOptions, streamFn); + const diagnostics = getSummaryDiagnostics(response); + if (diagnostics) diagnostics.stage = label; + lastResponse = response; + attempts++; usage = usage ? combineUsage(usage, response.usage) : response.usage; if (options.signal?.aborted) throw new Error("Compaction cancelled"); // Do not infer overflow from successful usage or empty/length-limited output. @@ -784,11 +825,29 @@ async function completeSummaryWithOverflowRecovery( try { const response = await retryAssistantCall(produce, retry, options.signal, retryCallbacks); - return { ...response, usage: usage ?? response.usage }; + const result = { ...response, usage: usage ?? response.usage }; + const diagnostics = getSummaryDiagnostics(response); + if (diagnostics) { + saveSummaryDiagnostics(result, { ...diagnostics, attempts, durationMs: Date.now() - startedAt }); + } + return result; } catch (error) { if (activeRetry !== undefined) { await callbacks?.onRetryFinished?.(false, activeRetry, error instanceof Error ? error.message : String(error)); } + if (error instanceof SummaryFailure) { + error.diagnostics.stage = label; + error.diagnostics.attempts += attempts; + error.diagnostics.durationMs = Date.now() - startedAt; + throw error; + } + if (lastResponse && !options.signal?.aborted) { + const diagnostics = getSummaryDiagnostics(lastResponse); + if (diagnostics) { + saveSummaryDiagnostics(lastResponse, { ...diagnostics, attempts, durationMs: Date.now() - startedAt }); + throw summaryFailure(error instanceof Error ? error.message : String(error), lastResponse); + } + } throw error; } } @@ -854,16 +913,16 @@ export async function generateSummaryWithUsage( } const failure = getSummarizationFailure(response, "Summarization", maxTokens); if (failure) { - throw new Error(failure); + throw summaryFailure(failure, response); } if (response.content.some((block) => block.type === "toolCall")) { - throw new Error("Summarization attempted to call a tool"); + throw summaryFailure("Summarization attempted to call a tool", response); } const textContent = contentText(response.content); // Validate model text before split-turn scaffolding or file metadata can make it look nonempty. if (textContent.trim().length === 0) { - throw new Error("Summarization failed: empty summary (response contained no summary text)"); + throw summaryFailure("Summarization failed: empty summary (response contained no summary text)", response); } return { text: textContent, usage: response.usage }; @@ -1157,15 +1216,18 @@ async function generateTurnPrefixSummary( } const failure = getSummarizationFailure(response, "Turn prefix summarization", maxTokens); if (failure) { - throw new Error(failure); + throw summaryFailure(failure, response); } if (response.content.some((block) => block.type === "toolCall")) { - throw new Error("Turn prefix summarization attempted to call a tool"); + throw summaryFailure("Turn prefix summarization attempted to call a tool", response); } const textContent = contentText(response.content); // A valid history summary cannot substitute for a missing turn-prefix summary. if (textContent.trim().length === 0) { - throw new Error("Turn prefix summarization failed: empty summary (response contained no summary text)"); + throw summaryFailure( + "Turn prefix summarization failed: empty summary (response contained no summary text)", + response, + ); } return { diff --git a/packages/coding-agent/src/core/compaction/summary-diagnostics.ts b/packages/coding-agent/src/core/compaction/summary-diagnostics.ts new file mode 100644 index 00000000..b9d5f260 --- /dev/null +++ b/packages/coding-agent/src/core/compaction/summary-diagnostics.ts @@ -0,0 +1,84 @@ +import type { AssistantMessage, Context, Model, RetryPolicy, SimpleStreamOptions } from "@step-harness/providers"; +import { redactSecretString } from "../../step/secret-redaction.ts"; + +export interface SummaryRequestDiagnostics { + stage?: string; + provider: string; + model: string; + api: string; + contextWindow: number; + modelMaxTokens: number; + selectedMaxTokens?: number; + observedMaxTokens?: number; + inputTokensEstimated: number; + reasoning: string; + attempts: number; + durationMs: number; + stopReason?: string; + errorKind?: string; + rawStopReason?: string; + usage?: AssistantMessage["usage"]; +} + +const responses = new WeakMap(); + +/** Capture only numeric budget fields, never the prompt, headers, or credentials. */ +export function observeSummaryPayload(payload: unknown): number | undefined { + if (!payload || typeof payload !== "object") return undefined; + const record = payload as Record; + for (const key of ["max_output_tokens", "max_tokens", "max_completion_tokens"]) { + const value = record[key]; + if (typeof value === "number" && Number.isFinite(value) && value > 0) return value; + } + return undefined; +} + +export function createSummaryDiagnostics( + model: Model, + context: Context, + options: SimpleStreamOptions, +): SummaryRequestDiagnostics { + return { + provider: redactSecretString(model.provider), + model: redactSecretString(model.id), + api: model.api, + contextWindow: model.contextWindow, + modelMaxTokens: model.maxTokens, + selectedMaxTokens: options.maxTokens, + inputTokensEstimated: Math.ceil(JSON.stringify(context).length / 4), + reasoning: options.reasoning ?? "off", + attempts: 0, + durationMs: 0, + }; +} + +export function saveSummaryDiagnostics(response: AssistantMessage, diagnostics: SummaryRequestDiagnostics): void { + responses.set(response, diagnostics); +} + +export function getSummaryDiagnostics(response: AssistantMessage): SummaryRequestDiagnostics | undefined { + return responses.get(response); +} + +export class SummaryFailure extends Error { + readonly diagnostics: SummaryRequestDiagnostics; + + constructor(message: string, diagnostics: SummaryRequestDiagnostics) { + super(redactSecretString(message)); + this.name = "SummaryFailure"; + this.diagnostics = diagnostics; + } +} + +export function summaryFailure(message: string, response: AssistantMessage): Error { + const diagnostics = responses.get(response); + return diagnostics ? new SummaryFailure(message, diagnostics) : new Error(redactSecretString(message)); +} + +export function formatSummaryDiagnostics(d: SummaryRequestDiagnostics): string { + return `Summary request: stage=${d.stage ?? "unknown"}, provider=${d.provider}, model=${d.model}, api=${d.api}; contextWindow=${d.contextWindow}, model.maxTokens=${d.modelMaxTokens}, selectedMaxTokens=${d.selectedMaxTokens ?? "unknown"}, observedPayloadMaxTokens=${d.observedMaxTokens ?? "unknown"}, inputTokensEstimated=${d.inputTokensEstimated}, reasoning=${d.reasoning}; attempts=${d.attempts}, durationMs=${d.durationMs}, stopReason=${d.stopReason ?? "unknown"}, errorKind=${d.errorKind ?? "unknown"}.`; +} + +export function formatRetryPolicy(retry: RetryPolicy): string { + return `retry.enabled=${retry.enabled}, retry.maxRetries=${retry.maxRetries}, retry.baseDelayMs=${retry.baseDelayMs}`; +} diff --git a/packages/coding-agent/test/compaction.test.ts b/packages/coding-agent/test/compaction.test.ts index 0e78feee..f2df219a 100644 --- a/packages/coding-agent/test/compaction.test.ts +++ b/packages/coding-agent/test/compaction.test.ts @@ -371,8 +371,9 @@ describe("getSummarizationFailure length-stop diagnostics", () => { it("surfaces the token cap in the length-stop message when provided", () => { const message = getSummarizationFailure(lengthResponse, "Summarization", 19660); expect(message).toContain("hit the token cap"); - expect(message).toContain("19660-token output cap"); - expect(message).toContain("raise reserveTokens"); + expect(message).toContain("19660-token selected output cap"); + expect(message).toContain("check model output limits"); + expect(message).not.toContain("raise reserveTokens"); }); it("omits the cap note when maxTokens is not provided", () => { diff --git a/packages/coding-agent/test/footer-data-provider.test.ts b/packages/coding-agent/test/footer-data-provider.test.ts index 4ad9a23a..9b13895d 100644 --- a/packages/coding-agent/test/footer-data-provider.test.ts +++ b/packages/coding-agent/test/footer-data-provider.test.ts @@ -77,6 +77,10 @@ function createReftableWorktree(tempDir: string): WorktreeFixture { return { worktreeDir, reftableDir }; } +// These checks exercise real fs.watch delivery, not only the mocked git callback. +// A full-suite wait timed out but focused and later full reruns passed; event loss +// versus scheduling delay is unconfirmed. Capture watcher/debounce timing on recurrence +// rather than increasing this deadline without evidence. async function waitFor(condition: () => boolean, timeoutMs = 3000): Promise { const startedAt = Date.now(); while (!condition()) { diff --git a/packages/coding-agent/test/subagent-invocation.test.ts b/packages/coding-agent/test/subagent-invocation.test.ts index af442669..fcdfd54a 100644 --- a/packages/coding-agent/test/subagent-invocation.test.ts +++ b/packages/coding-agent/test/subagent-invocation.test.ts @@ -40,6 +40,10 @@ test("a source-launched subagent starts the real CLI from a different working di for (const name of ["PATH", "SystemRoot", "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT"]) { if (process.env[name] !== undefined) env[name] = process.env[name]; } + // macOS Unix socket paths are bounded: tsx adds an IPC suffix below TMPDIR. + // Run the isolated suite with TMPDIR=/tmp if the default temp root causes EINVAL. + // A subsequent 20s child timeout was intermittent; focused and full reruns passed. + // Keep the deadline unchanged until startup-stage evidence identifies the cause. const options = { cwd, env, encoding: "utf8" as const, timeout: 25_000 }; const launcher = [fileURLToPath(import.meta.resolve("tsx/cli")), "--tsconfig", tsconfig]; // Positive control: the source CLI itself works without built workspace packages. diff --git a/packages/coding-agent/test/suite/agent-session-compaction.test.ts b/packages/coding-agent/test/suite/agent-session-compaction.test.ts index b467b9b7..e0485d1e 100644 --- a/packages/coding-agent/test/suite/agent-session-compaction.test.ts +++ b/packages/coding-agent/test/suite/agent-session-compaction.test.ts @@ -371,7 +371,7 @@ describe("AgentSession compaction characterization", () => { reason: "threshold", aborted: false, willRetry: false, - errorMessage: "Auto-compaction failed: summary generator blew up", + errorMessage: expect.stringContaining("Auto-compaction failed: summary generator blew up"), }); expect(failedEvents).toEqual([ expect.objectContaining({ @@ -380,7 +380,7 @@ describe("AgentSession compaction characterization", () => { aborted: false, willRetry: false, fromExtension: false, - errorMessage: "Auto-compaction failed: summary generator blew up", + errorMessage: expect.stringContaining("Auto-compaction failed: summary generator blew up"), }), ]); }); diff --git a/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts b/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts index 69fb180c..9337cab5 100644 --- a/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts +++ b/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts @@ -4,7 +4,8 @@ import { createAssistantMessageEventStream, fauxAssistantMessage, } from "@step-harness/providers"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SessionManager } from "../../../src/core/session-manager.ts"; import { createHarness, type Harness } from "../harness.ts"; /** @@ -168,6 +169,93 @@ describe("#6647 compaction retries transient summarization failures", () => { expect(starts).toHaveLength(2); expect(ends).toHaveLength(1); expect(ends[0]).toMatchObject({ type: "summarization_retry_finished" }); + expect(harness.sessionManager.getEntries()).toContainEqual( + expect.objectContaining({ + type: "custom", + customType: "step-compaction-failure", + data: expect.objectContaining({ + diagnostics: expect.objectContaining({ attempts: 3, observedMaxTokens: undefined, stopReason: "error" }), + }), + }), + ); + }); + + it("persists effective request budgets without prompts or credentials on a length failure", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + const originalMessages = harness.sessionManager.buildSessionContext().messages; + harness.session.agent.streamFunction = (model, _context, options) => { + const stream = createAssistantMessageEventStream(); + queueMicrotask(async () => { + await options?.onPayload?.( + { max_output_tokens: 16, input: "private-prompt-fixture", apiKey: "private-key-fixture" }, + model, + ); + stream.push({ + type: "done", + reason: "length", + message: { + ...fauxAssistantMessage("partial summary must not be saved", { stopReason: "length" }), + usage: createUsage(10), + }, + }); + }); + return stream; + }; + const persisted = SessionManager.create(harness.tempDir, harness.tempDir); + for (const entry of harness.sessionManager.getEntries()) { + if ( + entry.type === "message" && + (entry.message.role === "user" || entry.message.role === "assistant" || entry.message.role === "toolResult") + ) + persisted.appendMessage(entry.message); + } + const append = vi + .spyOn(harness.sessionManager, "appendCustomEntry") + .mockImplementation((type, data) => persisted.appendCustomEntry(type, data)); + await expect(harness.session.compact()).rejects.toThrow("observedPayloadMaxTokens=16"); + append.mockRestore(); + const entry = SessionManager.open(persisted.getSessionFile()!) + .getEntries() + .find((entry) => entry.type === "custom" && entry.customType === "step-compaction-failure"); + expect(entry).toMatchObject({ + data: { + version: 1, + reason: "manual", + diagnostics: { + model: harness.getModel().id, + contextWindow: harness.getModel().contextWindow, + modelMaxTokens: harness.getModel().maxTokens, + selectedMaxTokens: harness.getModel().maxTokens, + observedMaxTokens: 16, + attempts: 1, + stopReason: "length", + }, + }, + }); + expect(JSON.stringify(entry)).not.toMatch(/private-prompt-fixture|private-key-fixture|partial summary/); + expect(harness.sessionManager.buildSessionContext().messages).toEqual(originalMessages); + expect(harness.sessionManager.getEntries().filter((entry) => entry.type === "compaction")).toHaveLength(0); + }); + + it("retains the original failure when diagnostic persistence fails", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + useScriptedStreamFn(harness, [ + fauxAssistantMessage("", { stopReason: "error", errorMessage: "insufficient_quota" }), + ]); + const append = vi.spyOn(harness.sessionManager, "appendCustomEntry").mockImplementation(() => { + throw new Error("disk unavailable"); + }); + try { + await expect(harness.session.compact()).rejects.toThrow( + /insufficient_quota[\s\S]*Could not persist compaction diagnostics/, + ); + } finally { + append.mockRestore(); + } }); it("aborts an in-flight retry backoff via abortCompaction", async () => { @@ -192,5 +280,10 @@ describe("#6647 compaction retries transient summarization failures", () => { await expect(compactPromise).rejects.toThrow(); const compactionEnd = harness.eventsOfType("compaction_end").at(-1); expect(compactionEnd).toMatchObject({ aborted: true }); + expect( + harness.sessionManager + .getEntries() + .some((entry) => entry.type === "custom" && entry.customType === "step-compaction-failure"), + ).toBe(false); }); }); diff --git a/packages/coding-agent/test/suite/regressions/compaction-integrity.test.ts b/packages/coding-agent/test/suite/regressions/compaction-integrity.test.ts index 2abd3cc5..a69fca5b 100644 --- a/packages/coding-agent/test/suite/regressions/compaction-integrity.test.ts +++ b/packages/coding-agent/test/suite/regressions/compaction-integrity.test.ts @@ -142,6 +142,16 @@ describe("compaction integrity", () => { }; } + function expectOnlyFailureDiagnosticAdded(harness: Harness, before: ReturnType): void { + const after = captureSession(harness); + const failure = after.entries.at(-1); + expect(failure).toMatchObject({ type: "custom", customType: "step-compaction-failure", parentId: before.leafId }); + expect(after.entries.slice(0, -1)).toEqual(before.entries); + expect(after.messages).toEqual(before.messages); + expect(after.context).toEqual(before.context); + expect(after.leafId).toBe(failure!.id); + } + // CP-01: the marker exists only in the previous checkpoint, outside the retained messages. it("preserves previous history through prepare, split-turn compaction, and context reload", async () => { const harness = await seedSession("prefix"); @@ -187,7 +197,7 @@ describe("compaction integrity", () => { await expect(harness.session.compact()).rejects.toThrow(`${scenario.label} failed: empty summary`); expect(appendCompaction).not.toHaveBeenCalled(); - expect(captureSession(harness)).toEqual(before); + expectOnlyFailureDiagnosticAdded(harness, before); expect(harness.session.messages).toBe(messages); expect(JSON.stringify(harness.session.messages)).toContain(HISTORY_MARKER); expect(harness.faux.state.callCount).toBe(scenario.validHistoryFirst ? 2 : 1); @@ -229,7 +239,7 @@ describe("compaction integrity", () => { await expect(session._runAutoCompaction("threshold", false)).resolves.toBe(false); expect(appendCompaction).not.toHaveBeenCalled(); - expect(captureSession(harness)).toEqual(before); + expectOnlyFailureDiagnosticAdded(harness, before); expect(harness.session.messages).toBe(messages); expect(harness.faux.state.callCount).toBe(1); expect(harness.eventsOfType("compaction_end").at(-1)).toMatchObject({ @@ -292,7 +302,7 @@ describe("compaction integrity", () => { await expect(harness.session.compact()).rejects.toThrow(error); - expect(captureSession(harness)).toEqual(before); + expectOnlyFailureDiagnosticAdded(harness, before); expect(harness.faux.state.callCount).toBe(1); expect(harness.eventsOfType("compaction_end").at(-1)).toMatchObject({ result: undefined, diff --git a/packages/providers/README.md b/packages/providers/README.md index 7c6cc0bc..f33d33a0 100644 --- a/packages/providers/README.md +++ b/packages/providers/README.md @@ -781,6 +781,8 @@ Every `AssistantMessage` includes a `stopReason` field that indicates how the ge ## Error Handling +Anthropic SSE transport EOF before `message_stop`, including an unparseable trailing event, is classified as `transient` using the structured `stream_incomplete` error code. Callers can reuse their bounded retry policy. A delimited invalid JSON event is not automatically classified as a dropped connection; cancellation remains `aborted`. Parse failures omit raw SSE payloads to avoid exposing response content in durable diagnostics. + Request failures never throw out of the stream functions: when a request ends with an error (including aborts and tool call validation errors), the streaming API emits an error event and the final message carries the details: ```typescript diff --git a/packages/providers/src/api/anthropic-messages.ts b/packages/providers/src/api/anthropic-messages.ts index 4fd36db0..63b79c9e 100644 --- a/packages/providers/src/api/anthropic-messages.ts +++ b/packages/providers/src/api/anthropic-messages.ts @@ -257,6 +257,8 @@ interface ServerSentEvent { event: string | null; data: string; raw: string[]; + /** False only for an event left in the decoder when the transport reaches EOF. */ + complete: boolean; } interface SseDecoderState { @@ -274,7 +276,7 @@ const ANTHROPIC_MESSAGE_EVENTS: ReadonlySet = new Set([ "content_block_stop", ]); -function flushSseEvent(state: SseDecoderState): ServerSentEvent | null { +function flushSseEvent(state: SseDecoderState, complete = true): ServerSentEvent | null { if (!state.event && state.data.length === 0) { return null; } @@ -283,6 +285,7 @@ function flushSseEvent(state: SseDecoderState): ServerSentEvent | null { event: state.event, data: state.data.join("\n"), raw: [...state.raw], + complete, }; state.event = null; state.data = []; @@ -395,7 +398,7 @@ async function* iterateSseMessages( } } - const trailingEvent = flushSseEvent(state); + const trailingEvent = flushSseEvent(state, false); if (trailingEvent) { yield trailingEvent; } @@ -432,16 +435,19 @@ async function* iterateAnthropicEvents( sawMessageEnd = true; } yield event; - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - throw new Error( - `Could not parse Anthropic SSE event ${sse.event}: ${message}; data=${sse.data}; raw=${sse.raw.join("\\n")}`, - ); + } catch { + if (!sse.complete) { + throw new ProviderStreamError(`Anthropic stream ended during SSE event ${sse.event}`, "stream_incomplete"); + } + throw new Error(`Could not parse Anthropic SSE event ${sse.event}: invalid JSON payload`); } } - if (sawMessageStart && !sawMessageEnd) { - throw new Error("Anthropic stream ended before message_stop"); + if (!sawMessageEnd) { + throw new ProviderStreamError( + sawMessageStart ? "Anthropic stream ended before message_stop" : "Anthropic stream ended before message_start", + "stream_incomplete", + ); } } diff --git a/packages/providers/src/utils/error-kind.ts b/packages/providers/src/utils/error-kind.ts index f11f57c7..8f1def3a 100644 --- a/packages/providers/src/utils/error-kind.ts +++ b/packages/providers/src/utils/error-kind.ts @@ -53,6 +53,7 @@ const TRANSIENT_CODES = new Set([ "service_unavailable", "internal_error", "timeout_error", + "stream_incomplete", ]); const AUTH_CODES = new Set(["authentication_error", "permission_error", "invalid_api_key"]); const NETWORK_CODES = new Set([ diff --git a/packages/providers/test/anthropic-sse-parsing.test.ts b/packages/providers/test/anthropic-sse-parsing.test.ts index b6dd036b..08ae28aa 100644 --- a/packages/providers/test/anthropic-sse-parsing.test.ts +++ b/packages/providers/test/anthropic-sse-parsing.test.ts @@ -79,6 +79,57 @@ function createFakeAnthropicClient(response: Response): Anthropic { } describe("Anthropic raw SSE parsing", () => { + it.each([ + "event: content_block_delta\n", + 'event: content_block_delta\ndata: {"type":"content_block_delta","secret":"private-fixture', + "", + ])("classifies an unfinished message stream as transient (%j)", async (tail) => { + const start = minimalAnthropicEvents[0]!; + const response = new Response(`event: ${start.event}\ndata: ${start.data}\n\n${tail}`); + const stream = streamAnthropic( + anthropicModel(), + { messages: [] }, + { + client: createFakeAnthropicClient(response), + }, + ); + const events = []; + for await (const event of stream) events.push(event); + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorKind).toBe("transient"); + expect(result.errorMessage).not.toContain("private-fixture"); + expect(events.filter((event) => event.type === "error")).toHaveLength(1); + }); + + it("does not classify a delimited invalid JSON event as a dropped connection", async () => { + const response = new Response("event: content_block_delta\ndata: invalid-private-fixture\n\n"); + const result = await streamAnthropic( + anthropicModel(), + { messages: [] }, + { + client: createFakeAnthropicClient(response), + }, + ).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorKind).not.toBe("transient"); + expect(result.errorMessage).not.toContain("private-fixture"); + }); + + it("does not retry-classify a cancelled stream", async () => { + const controller = new AbortController(); + controller.abort(); + const result = await streamAnthropic( + anthropicModel(), + { messages: [] }, + { + client: createFakeAnthropicClient(createSseResponse(minimalAnthropicEvents)), + signal: controller.signal, + }, + ).result(); + expect(result.stopReason).toBe("aborted"); + expect(result.errorKind).toBeUndefined(); + }); it("repairs malformed SSE JSON and malformed streamed tool JSON", async () => { const model = anthropicModel(); const context: Context = {