From 8e53e0e49cf3ecec20c840c4cb03e692e96c50f1 Mon Sep 17 00:00:00 2001 From: David Brailovsky Date: Tue, 21 Jul 2026 10:07:20 +0000 Subject: [PATCH] compaction & branch summarization follow retry policy fixes #6647 compaction (auto & manual) and branch summarization retry on transient failures. use the same retry policy from settings. emit events for the tui to show indication of retries --- packages/ai/src/utils/retry.ts | 110 ++++++++++ packages/ai/test/retry.test.ts | 122 ++++++++++- .../coding-agent/src/core/agent-session.ts | 54 ++++- .../core/compaction/branch-summarization.ts | 17 +- .../src/core/compaction/compaction.ts | 41 +++- .../src/modes/interactive/interactive-mode.ts | 26 +++ ...tion-retries-transient-stream-drop.test.ts | 192 ++++++++++++++++++ 7 files changed, 546 insertions(+), 16 deletions(-) create mode 100644 packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts diff --git a/packages/ai/src/utils/retry.ts b/packages/ai/src/utils/retry.ts index 6332ff55..0a433439 100644 --- a/packages/ai/src/utils/retry.ts +++ b/packages/ai/src/utils/retry.ts @@ -85,6 +85,116 @@ const RETRYABLE_PROVIDER_ERROR_PATTERN = buildProviderErrorPattern([ "ResourceExhausted", ]); +/** + * Retry policy: bounded attempts with exponential backoff (`baseDelayMs * 2^(attempt-1)`). + * Matches `settings.retry` (`enabled`, `maxRetries`, `baseDelayMs`) in coding-agent; kept + * here so the classifier and the policy-driven retry loop live together and stay reusable + * by the SDK and other callers. + */ +export interface RetryPolicy { + enabled: boolean; + /** Max retry attempts (0 = no retries). The initial call never counts as a retry. */ + maxRetries: number; + /** Base delay in ms. Per-attempt delay is `baseDelayMs * 2^(attempt-1)` before jitter. */ + baseDelayMs: number; +} + +/** Optional callbacks emitted by {@link retryAssistantCall} around each retry. */ +export interface RetryCallbacks { + /** Emitted before the backoff sleep of each retry attempt (1-indexed). */ + onRetry?: (attempt: number, maxAttempts: number, delayMs: number, errorMessage: string) => void; + /** Emitted after the backoff sleep, immediately before the retried call starts. */ + onRetryAttemptStart?: () => void; + /** Emitted once when the loop ends: success if a later call returned a non-error message. */ + onRetryEnd?: (success: boolean, attempt: number, finalError?: string) => void; +} + +class RetrySleepAbortError extends Error { + constructor() { + super("Aborted"); + } +} + +function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new RetrySleepAbortError()); + return; + } + const timeout = setTimeout(resolve, ms); + signal?.addEventListener( + "abort", + () => { + clearTimeout(timeout); + reject(new RetrySleepAbortError()); + }, + { once: true }, + ); + }); +} + +/** + * Run a single assistant-producing call with bounded retry on transient errors. + * + * Behavior: + * - A non-error response (success or aborted) is returned immediately; aborts are + * never retried. Aborts during the backoff sleep are normalized to an aborted + * `AssistantMessage` too, so callers do not need to care when cancellation happened. + * - A non-retryable error (per {@link isRetryableAssistantError}, including quota/ + * billing exhaustion) is returned immediately so deterministic errors fail fast. + * - Otherwise retries up to `maxRetries` times with exponential backoff, emitting + * `onRetry` before each sleep, `onRetryAttemptStart` after each sleep before the + * retried call starts, and `onRetryEnd` once at the end (whether the loop ends in + * success, exhausted retries, or an aborted backoff). + * + * When `policy` is undefined or disabled, the first response is returned unchanged + * (equivalent to calling `produce()` directly). + */ +export async function retryAssistantCall( + produce: () => Promise, + policy: RetryPolicy | undefined, + signal: AbortSignal | undefined, + callbacks?: RetryCallbacks, +): Promise { + const maxAttempts = policy?.enabled ? policy.maxRetries : 0; + + let attempt = 0; + let lastRetry: { attempt: number; errorMessage: string } | undefined; + for (;;) { + const response = await produce(); + + // Success or abort: never retry an aborted message; non-error returns as-is. + if (response.stopReason !== "error") { + if (lastRetry) callbacks?.onRetryEnd?.(true, lastRetry.attempt); + return response; + } + + // Non-retryable, or budget exhausted: return the final error message. + if (attempt >= maxAttempts || !isRetryableAssistantError(response)) { + if (lastRetry) callbacks?.onRetryEnd?.(false, lastRetry.attempt, response.errorMessage); + return response; + } + + attempt++; + lastRetry = { attempt, errorMessage: response.errorMessage || "Unknown error" }; + const delayMs = policy!.baseDelayMs * 2 ** (attempt - 1); + callbacks?.onRetry?.(attempt, maxAttempts, delayMs, lastRetry.errorMessage); + + // Normalize aborts during retry backoff to the same AssistantMessage shape as + // provider stream aborts, so callers do not need to care when cancellation happened. + try { + await sleep(delayMs, signal); + } catch (error) { + callbacks?.onRetryEnd?.(false, attempt, lastRetry.errorMessage); + if (error instanceof RetrySleepAbortError) { + return { ...response, stopReason: "aborted", errorMessage: undefined }; + } + throw error; + } + callbacks?.onRetryAttemptStart?.(); + } +} + /** * Classifies whether a failed assistant message looks like a transient provider * or transport error, so callers can decide if the last assistant turn should be diff --git a/packages/ai/test/retry.test.ts b/packages/ai/test/retry.test.ts index abdc8716..71743f67 100644 --- a/packages/ai/test/retry.test.ts +++ b/packages/ai/test/retry.test.ts @@ -1,6 +1,6 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { fauxAssistantMessage } from "../src/providers/faux.ts"; -import { isRetryableAssistantError } from "../src/utils/retry.ts"; +import { isRetryableAssistantError, type RetryPolicy, retryAssistantCall } from "../src/utils/retry.ts"; const openAIExplicitRetryMessage = "An error occurred while processing your request. You can retry your request, or contact us through our help center at help.openai.com if the error persists. Please include the request ID req_******** in your message."; @@ -66,3 +66,121 @@ describe("provider retry classification", () => { expect(isRetryableAssistantError(fauxAssistantMessage("not an error"))).toBe(false); }); }); + +describe("retryAssistantCall", () => { + const disabled: RetryPolicy = { enabled: false, maxRetries: 3, baseDelayMs: 0 }; + const enabled: RetryPolicy = { enabled: true, maxRetries: 3, baseDelayMs: 0 }; + + it("returns a successful response immediately without retrying", async () => { + const produce = vi.fn(async () => fauxAssistantMessage("ok")); + const res = await retryAssistantCall(produce, enabled, undefined); + expect(res.content).toEqual([{ type: "text", text: "ok" }]); + expect(produce).toHaveBeenCalledTimes(1); + }); + + it("does not retry an aborted message", async () => { + const produce = vi.fn(async () => fauxAssistantMessage("", { stopReason: "aborted" })); + const onRetry = vi.fn(); + const res = await retryAssistantCall(produce, enabled, undefined, { onRetry }); + expect(res.stopReason).toBe("aborted"); + expect(produce).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + }); + + it("does not retry a non-retryable error (quota/billing)", async () => { + const produce = vi.fn(async () => + fauxAssistantMessage("", { stopReason: "error", errorMessage: "insufficient_quota" }), + ); + const onRetry = vi.fn(); + const onRetryEnd = vi.fn(); + const res = await retryAssistantCall(produce, enabled, undefined, { onRetry, onRetryEnd }); + expect(res.stopReason).toBe("error"); + expect(produce).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + expect(onRetryEnd).not.toHaveBeenCalled(); + }); + + it("retries a transient error up to maxRetries then returns the final error", async () => { + const produce = vi.fn(async () => fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" })); + const onRetry = vi.fn(); + const onRetryEnd = vi.fn(); + const res = await retryAssistantCall(produce, enabled, undefined, { onRetry, onRetryEnd }); + expect(res.stopReason).toBe("error"); + expect(produce).toHaveBeenCalledTimes(4); // 1 initial + 3 retries + expect(onRetry).toHaveBeenCalledTimes(3); + expect(onRetryEnd).toHaveBeenCalledWith(false, 3, "terminated"); + }); + + it("stops retrying once a call succeeds", async () => { + let n = 0; + const produce = vi.fn(async () => { + n++; + return n < 3 + ? fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" }) + : fauxAssistantMessage("recovered"); + }); + const onRetryEnd = vi.fn(); + const res = await retryAssistantCall(produce, enabled, undefined, { onRetryEnd }); + expect(res.content).toEqual([{ type: "text", text: "recovered" }]); + expect(produce).toHaveBeenCalledTimes(3); + expect(onRetryEnd).toHaveBeenCalledWith(true, 2); + }); + + it("does not retry when policy is disabled", async () => { + const produce = vi.fn(async () => fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" })); + const onRetry = vi.fn(); + const onRetryEnd = vi.fn(); + const res = await retryAssistantCall(produce, disabled, undefined, { onRetry, onRetryEnd }); + expect(res.stopReason).toBe("error"); + expect(produce).toHaveBeenCalledTimes(1); + expect(onRetry).not.toHaveBeenCalled(); + expect(onRetryEnd).not.toHaveBeenCalled(); + }); + + it("emits onRetryAttemptStart after backoff before each retried call", async () => { + const events: string[] = []; + let n = 0; + const produce = vi.fn(async () => { + events.push(`produce:${n}`); + n++; + return n < 3 + ? fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" }) + : fauxAssistantMessage("recovered"); + }); + const onRetry = vi.fn((attempt: number) => { + events.push(`retry:${attempt}`); + }); + const onRetryAttemptStart = vi.fn(() => { + events.push("attempt-start"); + }); + const res = await retryAssistantCall(produce, enabled, undefined, { onRetry, onRetryAttemptStart }); + expect(res.content).toEqual([{ type: "text", text: "recovered" }]); + expect(onRetry).toHaveBeenCalledTimes(2); + expect(onRetryAttemptStart).toHaveBeenCalledTimes(2); + expect(events).toEqual([ + "produce:0", + "retry:1", + "attempt-start", + "produce:1", + "retry:2", + "attempt-start", + "produce:2", + ]); + }); + + it("aborts backoff sleep via signal, returns an aborted message, and emits onRetryEnd(false)", async () => { + const controller = new AbortController(); + const produce = vi.fn(async () => fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" })); + const policy: RetryPolicy = { enabled: true, maxRetries: 5, baseDelayMs: 10_000 }; + const onRetryEnd = vi.fn(); + const p = retryAssistantCall(produce, policy, controller.signal, { onRetryEnd }); + // Let one error call resolve and the first backoff sleep start, then abort. + await vi.waitFor(() => expect(produce).toHaveBeenCalled()); + controller.abort(); + const res = await p; + expect(res.stopReason).toBe("aborted"); + expect(res.errorMessage).toBeUndefined(); + expect(produce).toHaveBeenCalledTimes(1); + expect(onRetryEnd).toHaveBeenCalledWith(false, 1, "terminated"); + }); +}); diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index b0fdbb07..19adf6dd 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -41,6 +41,7 @@ import { isContextOverflow, isRetryableAssistantError, modelsAreEqual, + type RetryCallbacks, resetApiProviders, streamSimple, } from "@earendil-works/pi-ai/compat"; @@ -161,7 +162,21 @@ export type AgentSessionEvent = errorMessage?: string; } | { type: "auto_retry_start"; attempt: number; maxAttempts: number; delayMs: number; errorMessage: string } - | { type: "auto_retry_end"; success: boolean; attempt: number; finalError?: string }; + | { type: "auto_retry_end"; success: boolean; attempt: number; finalError?: string } + | { + type: "summarization_retry_start"; + attempt: number; + maxAttempts: number; + delayMs: number; + errorMessage: string; + } + | { type: "summarization_retry_attempt_start"; source: "branchSummary" } + | { + type: "summarization_retry_attempt_start"; + source: "compaction"; + reason: "manual" | "threshold" | "overflow"; + } + | { type: "summarization_retry_end" }; /** Listener function for agent session events */ export type AgentSessionEventListener = (event: AgentSessionEvent) => void; @@ -1838,6 +1853,8 @@ export class AgentSession { this.thinkingLevel, this.agent.streamFunction, env, + this.settingsManager.getRetrySettings(), + this._summarizationRetryCallbacks({ source: "compaction", reason: "manual" }), ); summary = result.summary; firstKeptEntryId = result.firstKeptEntryId; @@ -2114,6 +2131,8 @@ export class AgentSession { this.thinkingLevel, this.agent.streamFunction, env, + this.settingsManager.getRetrySettings(), + this._summarizationRetryCallbacks({ source: "compaction", reason }), ); summary = compactResult.summary; firstKeptEntryId = compactResult.firstKeptEntryId; @@ -2620,6 +2639,37 @@ export class AgentSession { return isRetryableAssistantError(message); } + /** + * Retry policy + callbacks shared by compaction and branch-summary summarization calls. + * Uses the same `settings.retry` budget/backoff as agent-turn retries so a single transient + * stream drop no longer fails the whole operation. `source` carries the context + * the TUI needs to render the retry and recreate the underlying indicator. + */ + private _summarizationRetryCallbacks( + source: { source: "branchSummary" } | { source: "compaction"; reason: "manual" | "threshold" | "overflow" }, + ): RetryCallbacks { + return { + onRetry: (attempt, maxAttempts, delayMs, errorMessage) => { + this._emit({ + type: "summarization_retry_start", + attempt, + maxAttempts, + delayMs, + errorMessage, + }); + }, + onRetryAttemptStart: () => { + this._emit({ + type: "summarization_retry_attempt_start", + ...source, + }); + }, + onRetryEnd: () => { + this._emit({ type: "summarization_retry_end" }); + }, + }; + } + /** * Prepare a retryable error for continuation with exponential backoff. * @returns true if the caller should continue the agent, false otherwise @@ -2934,6 +2984,8 @@ export class AgentSession { replaceInstructions, reserveTokens: branchSummarySettings.reserveTokens, streamFn: this.agent.streamFunction, + retry: this.settingsManager.getRetrySettings(), + callbacks: this._summarizationRetryCallbacks({ source: "branchSummary" }), }); if (result.aborted) { return { cancelled: true, aborted: true }; diff --git a/packages/coding-agent/src/core/compaction/branch-summarization.ts b/packages/coding-agent/src/core/compaction/branch-summarization.ts index 3366f06a..dbb1217e 100644 --- a/packages/coding-agent/src/core/compaction/branch-summarization.ts +++ b/packages/coding-agent/src/core/compaction/branch-summarization.ts @@ -6,9 +6,9 @@ */ import type { AgentMessage, StreamFn } from "@earendil-works/pi-agent-core"; +import type { RetryCallbacks, RetryPolicy } from "@earendil-works/pi-ai"; import { contentText } from "@earendil-works/pi-ai"; import type { Model, SimpleStreamOptions, Usage } from "@earendil-works/pi-ai/compat"; -import { completeSimple } from "@earendil-works/pi-ai/compat"; import { convertToLlm, createBranchSummaryMessage, @@ -16,7 +16,7 @@ import { createCustomMessage, } from "../messages.ts"; import type { ReadonlySessionManager, SessionEntry } from "../session-manager.ts"; -import { estimateTokens } from "./compaction.ts"; +import { completeSummarization, estimateTokens } from "./compaction.ts"; import { computeFileLists, createFileOps, @@ -83,6 +83,10 @@ export interface GenerateBranchSummaryOptions { reserveTokens?: number; /** Optional session stream function. Used to preserve SDK request behavior without mutating agent state. */ streamFn?: StreamFn; + /** Retry policy for transient summarization errors. Reuses coding-agent's `settings.retry`. */ + retry?: RetryPolicy; + /** Optional callbacks for retry reporting (e.g. TUI retry indicators). */ + callbacks?: RetryCallbacks; } // ============================================================================ @@ -300,6 +304,8 @@ export async function generateBranchSummary( replaceInstructions, reserveTokens = 16384, streamFn, + retry, + callbacks, } = options; // Token budget = context window minus reserved space for prompt + response @@ -338,12 +344,11 @@ export async function generateBranchSummary( // Call LLM for summarization. Prefer the session stream function so SDK // request behavior (timeouts, retries, attribution headers) stays consistent - // without running through agent state/events. + // without running through agent state/events. Retried via completeSummarization + // so transient stream drops reuse the configured retry policy. const context = { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }; const requestOptions: SimpleStreamOptions = { apiKey, headers, env, signal, maxTokens: 2048 }; - const response = streamFn - ? await (await streamFn(model, context, requestOptions)).result() - : await completeSimple(model, context, requestOptions); + const response = await completeSummarization(model, context, requestOptions, streamFn, retry, callbacks); // Check if aborted or errored if (response.stopReason === "aborted") { diff --git a/packages/coding-agent/src/core/compaction/compaction.ts b/packages/coding-agent/src/core/compaction/compaction.ts index ffad75e8..0ae6deeb 100644 --- a/packages/coding-agent/src/core/compaction/compaction.ts +++ b/packages/coding-agent/src/core/compaction/compaction.ts @@ -6,7 +6,7 @@ */ import type { AgentMessage, StreamFn, ThinkingLevel } from "@earendil-works/pi-agent-core"; -import { contentText } from "@earendil-works/pi-ai"; +import { contentText, type RetryCallbacks, type RetryPolicy, retryAssistantCall } from "@earendil-works/pi-ai"; import type { AssistantMessage, Context, Model, SimpleStreamOptions, Usage } from "@earendil-works/pi-ai/compat"; import { completeSimple } from "@earendil-works/pi-ai/compat"; import { convertToLlm } from "../messages.ts"; @@ -552,17 +552,24 @@ function createSummarizationOptions( return options; } -async function completeSummarization( +/** + * Shared choke point for every compaction/branch-summary summarization call. Wraps the + * single LLM call in {@link retryAssistantCall} so transient stream drops (e.g. + * `terminated`, socket close) honor the configured retry policy instead of failing + * the whole compaction on the first attempt. Deterministic errors and aborts return + * immediately (see {@link retryAssistantCall}). + */ +export async function completeSummarization( model: Model, context: Context, options: SimpleStreamOptions, streamFn?: StreamFn, + retry?: RetryPolicy, + callbacks?: RetryCallbacks, ): Promise { - if (!streamFn) { - return completeSimple(model, context, options); - } - const stream = await streamFn(model, context, options); - return stream.result(); + const produce = async (): Promise => + streamFn ? (await streamFn(model, context, options)).result() : completeSimple(model, context, options); + return retryAssistantCall(produce, retry, options.signal, callbacks); } /** @@ -581,6 +588,8 @@ export async function generateSummary( thinkingLevel?: ThinkingLevel, streamFn?: StreamFn, env?: Record, + retry?: RetryPolicy, + callbacks?: RetryCallbacks, ): Promise { return ( await generateSummaryWithUsage( @@ -595,6 +604,8 @@ export async function generateSummary( thinkingLevel, streamFn, env, + retry, + callbacks, ) ).text; } @@ -612,6 +623,8 @@ export async function generateSummaryWithUsage( thinkingLevel?: ThinkingLevel, streamFn?: StreamFn, env?: Record, + retry?: RetryPolicy, + callbacks?: RetryCallbacks, ): Promise<{ text: string; usage: Usage }> { const maxTokens = Math.min( Math.floor(0.8 * reserveTokens), @@ -651,6 +664,8 @@ export async function generateSummaryWithUsage( { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, completionOptions, streamFn, + retry, + callbacks, ); if (response.stopReason === "error") { @@ -801,6 +816,8 @@ export async function compact( thinkingLevel?: ThinkingLevel, streamFn?: StreamFn, env?: Record, + retry?: RetryPolicy, + callbacks?: RetryCallbacks, ): Promise { const { firstKeptEntryId, @@ -833,6 +850,8 @@ export async function compact( thinkingLevel, streamFn, env, + retry, + callbacks, ); historyText = historyResult.text; historyUsage = historyResult.usage; @@ -847,6 +866,8 @@ export async function compact( signal, thinkingLevel, streamFn, + retry, + callbacks, ); // Merge into single summary summary = `${historyText}\n\n---\n\n**Turn Context (split turn):**\n\n${turnPrefixResult.text}`; @@ -865,6 +886,8 @@ export async function compact( thinkingLevel, streamFn, env, + retry, + callbacks, ); summary = result.text; summaryUsage = result.usage; @@ -900,6 +923,8 @@ async function generateTurnPrefixSummary( signal?: AbortSignal, thinkingLevel?: ThinkingLevel, streamFn?: StreamFn, + retry?: RetryPolicy, + callbacks?: RetryCallbacks, ): Promise<{ text: string; usage: Usage }> { const maxTokens = Math.min( Math.floor(0.5 * reserveTokens), @@ -921,6 +946,8 @@ async function generateTurnPrefixSummary( { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, createSummarizationOptions(model, maxTokens, apiKey, headers, env, signal, thinkingLevel), streamFn, + retry, + callbacks, ); if (response.stopReason === "error") { diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index 54415407..89e79c61 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -3110,6 +3110,32 @@ export class InteractiveMode { this.ui.requestRender(); break; } + + case "summarization_retry_start": { + this.showError(event.errorMessage); + this.showStatusIndicator( + new RetryStatusIndicator(this.ui, event.attempt, event.maxAttempts, event.delayMs), + ); + this.ui.requestRender(); + break; + } + + case "summarization_retry_attempt_start": { + this.clearStatusIndicator("retry"); + if (event.source === "branchSummary") { + this.showStatusIndicator(new BranchSummaryStatusIndicator(this.ui)); + } else { + this.showStatusIndicator(new CompactionStatusIndicator(this.ui, event.reason)); + } + this.ui.requestRender(); + break; + } + + case "summarization_retry_end": { + this.clearStatusIndicator("retry"); + this.ui.requestRender(); + break; + } } } 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 new file mode 100644 index 00000000..210b0a3b --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts @@ -0,0 +1,192 @@ +import type { StreamFn } from "@earendil-works/pi-agent-core"; +import { type AssistantMessage, createAssistantMessageEventStream, fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import { createHarness, type Harness } from "../harness.ts"; + +/** + * Regression for #6647: compaction runs a single non-retried summarization call, so a + * transient mid-stream socket death (`terminated`) failed the whole compaction. + * Verifies that summarization now reuses `settings.retry` (bounded retries with + * exponential backoff gated on isRetryableAssistantError), emits + * `summarization_retry_*` events, and that aborts / non-retryable errors are not retried. + */ +describe("#6647 compaction retries transient summarization failures", () => { + const harnesses: Harness[] = []; + + afterEach(() => { + while (harnesses.length > 0) { + harnesses.pop()?.cleanup(); + } + }); + + function createUsage(totalTokens: number) { + return { + input: totalTokens, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; + } + + function seedCompactableSession(harness: Harness): void { + harness.settingsManager.applyOverrides({ compaction: { keepRecentTokens: 1 } }); + const now = Date.now(); + harness.sessionManager.appendMessage({ + role: "user", + content: [{ type: "text", text: "message to compact" }], + timestamp: now - 1000, + }); + const model = harness.getModel(); + const assistant: AssistantMessage = { + ...fauxAssistantMessage("", { stopReason: "stop", timestamp: now - 500 }), + api: model.api, + provider: model.provider, + model: model.id, + usage: createUsage(100), + }; + assistant.content = [{ type: "text", text: "assistant response to compact" }]; + harness.sessionManager.appendMessage(assistant); + harness.session.agent.state.messages = harness.sessionManager.buildSessionContext().messages; + } + + /** streamFn that responds with the given sequence of assistant messages across calls. */ + function useScriptedStreamFn(harness: Harness, script: AssistantMessage[]): () => number { + let callCount = 0; + const streamFunction: StreamFn = (model) => { + const message = script[callCount] ?? script[script.length - 1]!; + callCount++; + const stream = createAssistantMessageEventStream(); + queueMicrotask(() => { + if (message.stopReason === "error" || message.stopReason === "aborted") { + stream.push({ + type: "error", + reason: message.stopReason, + error: { ...message, api: model.api, provider: model.provider, model: model.id }, + }); + } else { + stream.push({ + type: "done", + reason: message.stopReason, + message: { ...message, api: model.api, provider: model.provider, model: model.id }, + }); + } + }); + return stream; + }; + harness.session.agent.streamFunction = streamFunction; + return () => callCount; + } + + it("retries a transient `terminated` summarization error and compacts successfully", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + harness.settingsManager.applyOverrides({ retry: { enabled: true, maxRetries: 3, baseDelayMs: 0 } }); + + const model = harness.getModel(); + const error = (errorMessage: string): AssistantMessage => ({ + ...fauxAssistantMessage("", { stopReason: "error", errorMessage }), + usage: createUsage(10), + }); + const success: AssistantMessage = { + ...fauxAssistantMessage("recovered summary"), + usage: createUsage(10), + }; + const getCallCount = useScriptedStreamFn(harness, [error("terminated"), error("terminated"), success]); + + const result = await harness.session.compact(); + + expect(result.summary).toContain("recovered summary"); + expect(getCallCount()).toBe(3); // 1 initial + 2 retries + const starts = harness.eventsOfType("summarization_retry_start"); + const ends = harness.eventsOfType("summarization_retry_end"); + expect(starts).toHaveLength(2); + expect(ends).toHaveLength(1); + expect(starts[0]).toMatchObject({ attempt: 1, maxAttempts: 3, errorMessage: "terminated" }); + expect(starts[1]).toMatchObject({ attempt: 2, maxAttempts: 3 }); + expect(ends[0]).toMatchObject({ type: "summarization_retry_end" }); + // model.* referenced to keep imports honest + expect(model.id).toBeTruthy(); + }); + + it("does not retry a non-retryable error (insufficient_quota)", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + harness.settingsManager.applyOverrides({ retry: { enabled: true, maxRetries: 3, baseDelayMs: 0 } }); + + const error: AssistantMessage = { + ...fauxAssistantMessage("", { stopReason: "error", errorMessage: "insufficient_quota" }), + usage: createUsage(10), + }; + const getCallCount = useScriptedStreamFn(harness, [error]); + + await expect(harness.session.compact()).rejects.toThrow("insufficient_quota"); + expect(getCallCount()).toBe(1); + expect(harness.eventsOfType("summarization_retry_start")).toHaveLength(0); + }); + + it("does not retry when retry is disabled", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + harness.settingsManager.applyOverrides({ retry: { enabled: false, maxRetries: 3, baseDelayMs: 0 } }); + + const error: AssistantMessage = { + ...fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" }), + usage: createUsage(10), + }; + const getCallCount = useScriptedStreamFn(harness, [error]); + + await expect(harness.session.compact()).rejects.toThrow("terminated"); + expect(getCallCount()).toBe(1); + expect(harness.eventsOfType("summarization_retry_start")).toHaveLength(0); + }); + + it("stops retrying after maxRetries and reports failure", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + harness.settingsManager.applyOverrides({ retry: { enabled: true, maxRetries: 2, baseDelayMs: 0 } }); + + const error: AssistantMessage = { + ...fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" }), + usage: createUsage(10), + }; + const getCallCount = useScriptedStreamFn(harness, [error, error, error]); + + await expect(harness.session.compact()).rejects.toThrow("terminated"); + expect(getCallCount()).toBe(3); // 1 initial + 2 retries + const starts = harness.eventsOfType("summarization_retry_start"); + const ends = harness.eventsOfType("summarization_retry_end"); + expect(starts).toHaveLength(2); + expect(ends).toHaveLength(1); + expect(ends[0]).toMatchObject({ type: "summarization_retry_end" }); + }); + + it("aborts an in-flight retry backoff via abortCompaction", async () => { + const harness = await createHarness({ withConfiguredAuth: false }); + harnesses.push(harness); + seedCompactableSession(harness); + harness.settingsManager.applyOverrides({ retry: { enabled: true, maxRetries: 5, baseDelayMs: 30_000 } }); + + const error: AssistantMessage = { + ...fauxAssistantMessage("", { stopReason: "error", errorMessage: "terminated" }), + usage: createUsage(10), + }; + useScriptedStreamFn(harness, [error, error, error]); + + const compactPromise = harness.session.compact(); + // Let the first error resolve and the retry backoff sleep start. + await new Promise((resolve) => setTimeout(resolve, 0)); + harness.session.abortCompaction(); + + // The aborted retry backoff rejects with an AbortError (matching the real SDK + // abort path), which compaction classifies as aborted. + await expect(compactPromise).rejects.toThrow(); + const compactionEnd = harness.eventsOfType("compaction_end").at(-1); + expect(compactionEnd).toMatchObject({ aborted: true }); + }); +});