fix(compaction): recover a session from a provider-killed summary stream

packages/coding-agent/CHANGELOG.md

@@ -13,6 +13,9 @@
 - A provider that accepts a request and never starts streaming no longer ends the turn with the watchdog's own message (`Provider stream start timed out after 180000ms ...`). The stall is still retried on the same model with the configured stream-start bound, and still hands the turn to the next model in a configured `retry.fallbackChains` entry whose answer becomes the turn result. What changed is what you read: the transcript (and `senpi -p`) describes the stall in plain language, and when nothing can take the turn over the final line names the stalled model, the attempts spent and the next step - `/fallback`, resending, or raising `retry.provider.streamStartTimeoutMs` (`0` disables). The wording on the assistant message is unchanged, so retry classification and fallback routing behave exactly as before ([#1740](https://github.com/code-yeongyu/senpi/issues/1740)).
 
 - A provider that keeps streaming at a uselessly low rate is now detected instead of looking healthy forever. Every previous guard on a live stream watched for silence (the stream-start bound stops applying at the first event; the idle bound is re-armed by every event), so a turn crawling at ~2 tok/s never failed, never retried and never walked a fallback chain. After the first stream event senpi now ignores `retry.provider.throughputGraceMs` (default 5000) of streaming and then measures streamed text and thinking units over a trailing `retry.provider.throughputWindowMs` (default 20000); a full window carrying at least 16 units whose sustained rate is below `retry.provider.minThroughputTokensPerSecond` (default 8, `0` disables) aborts the request with `Provider stream throughput degraded: <n> tok/s over <n>s (floor <n> tok/s)`. That failure is retryable but spends no same-model attempts - replaying the payload cannot make the upstream faster - so it goes straight to the configured fallback chain, and with no candidate the turn ends on that error with the usual "No fallback chain configured - set one with /fallback." guidance instead of continuing to crawl. Time the provider spends running local tools is excluded from the measurement, and the interactive working line now shows the live rate (`Working (1m 12s - 2.1 tok/s - esc to interrupt)`) so a degraded turn is visible while it runs ([#1739](https://github.com/code-yeongyu/senpi/issues/1739)).
+
+- A session no longer dead-ends when the compaction summarizer is killed by a provider or credential failure. Such a stream used to end required compaction with the raw internal `senpi:no-turn-retry:` marker, which also disabled auto-retry and model fallback; because compaction never applied, the context stayed above the threshold and every following prompt failed identically. Those failures now apply the deterministic compaction checkpoint (retained-suffix safety checks unchanged) and the turn continues, with one plain warning saying a provider summary could not be completed, that a checkpoint was applied and older detail was dropped, and that it is safe to continue. Provider refusals, missing credentials and user aborts still surface loudly instead of reducing context. One compaction is also bounded at 15 minutes total across every attempt and retry regardless of input size (an explicit `compaction.summarizationMaxDurationMs` above that still wins), and the summary stream's final settlement now happens inside the watchdog, so a provider whose stream ends without a terminal event can no longer park compaction with no timer armed ([#1741](https://github.com/code-yeongyu/senpi/issues/1741)).
+
 ### Removed
 
 ## [2026.9.16] - 2026-09-16

packages/coding-agent/src/core/compaction/changes.md

@@ -1,3 +1,24 @@
+## 2026-09-16 - Bound one compaction and settle its stream inside the watchdog (#1741)
+
+### What changed
+
+- `packages/coding-agent/src/core/compaction/stream-watchdog.ts`: `consumeStreamWithIdleTimeout` gains an optional `settle()` callback and returns its value, awaiting the stream's final `result()` under the SAME idle and wall-clock timers as iteration (overloads keep the settle-less call sites at `Promise<void>`). Adds the compaction-wide bound `SUMMARIZATION_TOTAL_BUDGET_MS` (900,000 ms), `summarizationTotalBudgetMs(attemptOverrideMs?)`, `SummarizationTotalBudgetError`, and `createSummarizationDeadline(totalBudgetMs, now?)` whose `attemptBudgetMs()` clamps one attempt to the compaction's remaining budget and throws once nothing is left.
+- `packages/coding-agent/src/core/compaction/compaction.ts`: `completeSummarization` returns the value settled inside `consumeStreamWithIdleTimeout` instead of awaiting `responseStream.result()` after the watchdog's `finally` cleared its timers.
+
+### Why
+
+- Issue #1741: final `result()` settlement sat outside the watchdog, so a provider whose iterator ends without a terminal `done`/`error` event parked compaction forever with no timer armed at all.
+- The per-attempt budget is size-scaled (2 ms per estimated input token, 30-minute ceiling) and every retry re-arms it, so a large session's total wait grew with the very thing that made it slow. One compaction now shares a single deadline that never scales with the input; only an explicit `compaction.summarizationMaxDurationMs` override raises it.
+
+### Why an extension could not handle it
+
+- The watchdog is core compaction mechanics shared by the core route and the builtin extension route; an extension cannot arm a timer around a stream core owns, nor bound an operation whose attempts core and the extension split between them.
+
+### Expected merge conflict zones
+
+- MEDIUM: `stream-watchdog.ts` `consumeStreamWithIdleTimeout` signature and its loop exits.
+- LOW: `compaction.ts` `completeSummarization` stream settlement.
+
 ## 2026-09-07 - Effective admission reserve (#7921 case 2)
 
 ### What changed

packages/coding-agent/src/core/compaction/compaction.ts

@@ -764,13 +764,15 @@ export async function completeSummarization(
 			const responseStream = Promise.resolve(
 				streamFn ? streamFn(model, context, requestOptions) : streamSimple(model, context, requestOptions),
 			);
-			await consumeStreamWithIdleTimeout(responseStream, {
+			// Settlement rides inside the watchdog: a provider whose iterator ends
+			// without a terminal event used to park here with every timer cleared.
+			return await consumeStreamWithIdleTimeout(responseStream, {
 				idleTimeoutMs: DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS,
 				maxDurationMs,
 				abort: () => requestController.abort(),
 				signal: callerSignal,
+				settle: async () => await (await responseStream).result(),
 			});
-			return await (await responseStream).result();
 		} finally {
 			if (callerSignal) callerSignal.removeEventListener("abort", onCallerAbort);
 		}

packages/coding-agent/src/core/compaction/stream-watchdog.ts

@@ -64,6 +64,79 @@ export const SUMMARIZATION_MAX_DURATION_PER_TOKEN_MS = 2;
  */
 export const SUMMARIZATION_MAX_DURATION_CAP_MS = 1_800_000;
 
+/**
+ * Total wall clock ONE compaction may hold the session across every attempt,
+ * retry and overflow shrink.
+ *
+ * The per-attempt budget is deliberately proportional to the input, so a large
+ * session legally licenses a 690s attempt (345k tokens) or the full 30-minute
+ * ceiling (900k tokens), and each retry re-arms that budget from scratch: the
+ * user's wait grew with the very thing that made it slow, without bound (#1741).
+ * This cap is the session-health bound the per-attempt budget cannot be: it
+ * never scales with the input, and every attempt of one compaction shares it.
+ */
+export const SUMMARIZATION_TOTAL_BUDGET_MS = 900_000;
+
+/**
+ * Total budget for one compaction. Size never raises it; only an explicit
+ * `compaction.summarizationMaxDurationMs` override does, because an operator who
+ * deliberately allows a longer single attempt must not have that attempt cut
+ * short by the total. Clamped to {@link SUMMARIZATION_MAX_DURATION_CAP_MS}.
+ */
+export function summarizationTotalBudgetMs(attemptOverrideMs?: number): number {
+	const override =
+		attemptOverrideMs !== undefined && Number.isFinite(attemptOverrideMs) && attemptOverrideMs > 0
+			? Math.min(SUMMARIZATION_MAX_DURATION_CAP_MS, attemptOverrideMs)
+			: 0;
+	return Math.max(SUMMARIZATION_TOTAL_BUDGET_MS, override);
+}
+
+/**
+ * One compaction outlived {@link SUMMARIZATION_TOTAL_BUDGET_MS}. Distinct from
+ * {@link StreamDurationBudgetError}, which bounds a single attempt: this one says
+ * no further attempt may start, so recovery must come from the deterministic
+ * fallback rather than another provider request.
+ */
+export class SummarizationTotalBudgetError extends Error {
+	readonly totalBudgetMs: number;
+	constructor(totalBudgetMs: number) {
+		super(
+			`Compaction exceeded its ${totalBudgetMs}ms total wall-clock budget across every summarization attempt and retry`,
+		);
+		this.name = "SummarizationTotalBudgetError";
+		this.totalBudgetMs = totalBudgetMs;
+	}
+}
+
+export interface SummarizationDeadline {
+	readonly totalBudgetMs: number;
+	/** Time left before the whole compaction is out of budget; never negative. */
+	remainingMs(): number;
+	/**
+	 * Clamp one attempt's wall-clock budget to what the compaction has left, so a
+	 * retry started near the deadline cannot re-arm a full attempt budget. Throws
+	 * {@link SummarizationTotalBudgetError} once nothing is left.
+	 */
+	attemptBudgetMs(requestedMs: number): number;
+}
+
+export function createSummarizationDeadline(
+	totalBudgetMs: number,
+	now: () => number = Date.now,
+): SummarizationDeadline {
+	const startedMs = now();
+	const remainingMs = (): number => Math.max(0, totalBudgetMs - (now() - startedMs));
+	return {
+		totalBudgetMs,
+		remainingMs,
+		attemptBudgetMs: (requestedMs: number): number => {
+			const remaining = remainingMs();
+			if (remaining <= 0) throw new SummarizationTotalBudgetError(totalBudgetMs);
+			return Math.min(requestedMs, remaining);
+		},
+	};
+}
+
 /**
  * Total time one summarization attempt may hold the session, sized to its input.
  *
@@ -83,7 +156,7 @@ export function summarizationMaxDurationMs(estimatedInputTokens: number, overrid
 	return Math.min(SUMMARIZATION_MAX_DURATION_CAP_MS, Math.max(DEFAULT_SUMMARIZATION_MAX_DURATION_MS, scaled));
 }
 
-export interface ConsumeStreamWithIdleTimeoutOptions<T> {
+export interface ConsumeStreamWithIdleTimeoutOptions<T, R = void> {
 	/** Silence budget per read; the timer resets on every event. */
 	readonly idleTimeoutMs: number;
 	/** Total wall-clock budget for the whole stream; omit to leave it unbounded. */
@@ -93,6 +166,14 @@ export interface ConsumeStreamWithIdleTimeoutOptions<T> {
 	readonly onEvent?: (event: T) => void;
 	/** Caller cancellation; an abort here ends the wait without an idle error. */
 	readonly signal?: AbortSignal;
+	/**
+	 * Final settlement of the stream (its `result()`), awaited under the SAME
+	 * timers as iteration. A provider whose iterator ends without pushing a
+	 * terminal `done`/`error` event leaves `result()` pending forever; settling it
+	 * after the watchdog's timers were cleared parked compaction with no timer
+	 * armed at all (#1741).
+	 */
+	readonly settle?: () => Promise<R>;
 }
 
 const IDLE_TRIP = "idle-trip" as const;
@@ -104,19 +185,23 @@ const CALLER_ABORTED = "caller-aborted" as const;
  * event arrives within `idleTimeoutMs`. Caller aborts propagate as the
  * stream's own abort outcome, never masked as an idle timeout.
  */
-export async function consumeStreamWithIdleTimeout<T>(
+export function consumeStreamWithIdleTimeout<T>(
 	stream: AsyncIterable<T> | PromiseLike<AsyncIterable<T>>,
-	options: ConsumeStreamWithIdleTimeoutOptions<T>,
-): Promise<void> {
-	const { idleTimeoutMs, maxDurationMs, abort, onEvent, signal } = options;
+	options: ConsumeStreamWithIdleTimeoutOptions<T, void> & { settle?: undefined },
+): Promise<void>;
+export function consumeStreamWithIdleTimeout<T, R>(
+	stream: AsyncIterable<T> | PromiseLike<AsyncIterable<T>>,
+	options: ConsumeStreamWithIdleTimeoutOptions<T, R> & { settle: () => Promise<R> },
+): Promise<R>;
+export async function consumeStreamWithIdleTimeout<T, R>(
+	stream: AsyncIterable<T> | PromiseLike<AsyncIterable<T>>,
+	options: ConsumeStreamWithIdleTimeoutOptions<T, R>,
+): Promise<R | undefined> {
+	const { idleTimeoutMs, maxDurationMs, abort, onEvent, signal, settle } = options;
 	let iterator: AsyncIterator<T> | undefined;
 	let removeAbortListener: (() => void) | undefined;
 	let callerAbortPromise: Promise<typeof CALLER_ABORTED> | undefined;
-	if (signal?.aborted) {
-		return;
-	}
-	// One absolute deadline for the whole stream, not a per-read budget. Created
-	// only after the already-aborted early return so no timer is ever leaked.
+	// One absolute deadline for the whole stream, not a per-read budget.
 	let budgetPromise: Promise<typeof BUDGET_TRIP> | undefined;
 	let budgetTimer: ReturnType<typeof setTimeout> | undefined;
 	let budgetMs = 0;
@@ -127,14 +212,46 @@ export async function consumeStreamWithIdleTimeout<T>(
 		budgetTimer.unref?.();
 		budgetPromise = promise;
 	}
-	if (signal !== undefined) {
+	if (signal !== undefined && !signal.aborted) {
 		const { promise, resolve } = Promise.withResolvers<typeof CALLER_ABORTED>();
 		const onAbort = () => resolve(CALLER_ABORTED);
 		signal.addEventListener("abort", onAbort, { once: true });
 		removeAbortListener = () => signal.removeEventListener("abort", onAbort);
 		callerAbortPromise = promise;
 	}
+	// Settle the stream under the timers this call already armed. Every exit that
+	// is not a thrown watchdog error goes through here, so `result()` can never be
+	// awaited with no deadline in force.
+	const settleUnderWatchdogs = async (): Promise<R | undefined> => {
+		if (!settle) return undefined;
+		const { promise: idlePromise, resolve: resolveIdle } = Promise.withResolvers<typeof IDLE_TRIP>();
+		const timer = setTimeout(() => resolveIdle(IDLE_TRIP), idleTimeoutMs);
+		timer.unref?.();
+		const contenders: Array<Promise<{ settled: R } | typeof IDLE_TRIP | typeof BUDGET_TRIP>> = [
+			settle().then((value) => ({ settled: value })),
+			idlePromise,
+		];
+		if (budgetPromise) contenders.push(budgetPromise);
+		let outcome: { settled: R } | typeof IDLE_TRIP | typeof BUDGET_TRIP;
+		try {
+			outcome = await Promise.race(contenders);
+		} finally {
+			clearTimeout(timer);
+		}
+		if (outcome === IDLE_TRIP) {
+			abort();
+			throw new StreamIdleTimeoutError(idleTimeoutMs);
+		}
+		if (outcome === BUDGET_TRIP) {
+			abort();
+			throw new StreamDurationBudgetError(budgetMs);
+		}
+		return outcome.settled;
+	};
 	try {
+		// A caller that already cancelled still settles its stream - the terminal
+		// aborted message is what callers return - but under the same watchdogs.
+		if (signal?.aborted) return await settleUnderWatchdogs();
 		let resolvedStream: AsyncIterable<T>;
 		if (Symbol.asyncIterator in stream) {
 			resolvedStream = stream;
@@ -149,7 +266,7 @@ export async function consumeStreamWithIdleTimeout<T>(
 				abort();
 				throw new StreamDurationBudgetError(budgetMs);
 			}
-			if (resolution === CALLER_ABORTED) return;
+			if (resolution === CALLER_ABORTED) return await settleUnderWatchdogs();
 			resolvedStream = resolution;
 		}
 		iterator = resolvedStream[Symbol.asyncIterator]();
@@ -181,9 +298,9 @@ export async function consumeStreamWithIdleTimeout<T>(
 			}
 			if (result === CALLER_ABORTED) {
 				void iterator?.return?.();
-				return;
+				return await settleUnderWatchdogs();
 			}
-			if (result.done) return;
+			if (result.done) return await settleUnderWatchdogs();
 			onEvent?.(result.value);
 		}
 	} finally {

packages/coding-agent/src/core/extensions/builtin/compaction/changes.md

@@ -1,3 +1,28 @@
+## Authorize the deterministic fallback for a provider-killed summary stream (2026-09-16)
+
+### What changed
+
+- `packages/coding-agent/src/core/extensions/builtin/compaction/deterministic-fallback.ts`: `RequiredCompactionFallbackFailure` gains `summarization-provider-failure`, and `classifyRequiredCompactionFallbackFailure` now recognizes a summary stream terminated by a provider or credential fault - any `CredentialFailoverError`, any error whose message carries the `senpi:no-turn-retry:` marker, and a non-transient, non-refused `SummaryRequestError` with no structured failure kind - plus `SummarizationTotalBudgetError` as `summarization-timeout`. User aborts, policy refusals, missing credentials and ordinary bugs stay unauthorized. Adds `stripTurnRetrySuppressionPrefix()` and `formatRequiredCompactionFallbackNotice()`.
+- `packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts`: `SummaryRequestError` carries an explicit `refused` flag (set from `refusal`/`sensitive` stop details); `isRetryableSummaryAttempt` mirrors the new class so a marker-bearing or credential-failover error is never re-billed while genuinely transient failures still retry; `runExtensionCompaction` opens one `createSummarizationDeadline` for the whole compaction, re-clamps every attempt budget to what is left, and refuses a retry once the deadline passed.
+- `packages/coding-agent/src/core/extensions/builtin/compaction/speculative-summary.ts`: `generateSummaryMessage` returns the message settled inside `consumeStreamWithIdleTimeout` instead of awaiting `responseStream.result()` after the watchdog cleared its timers.
+- `packages/coding-agent/src/core/extensions/builtin/compaction/transient-failure.ts`: a compaction-wide total-budget trip degrades like the other watchdog trips.
+- `packages/coding-agent/src/core/extensions/builtin/compaction/index.ts`: `recoverRequiredCompaction` takes the context and the causing error, notifies the user once through `ctx.ui.notify` when the deterministic checkpoint is applied, and every compaction message built from an error message is stripped of the `senpi:no-turn-retry:` prefix.
+
+### Why
+
+- Issue #1741: a summarization stream killed by a credential-rotation or provider error classified as `undefined`, so the deterministic fallback that exists precisely for "summarization did not complete" never ran. The blocking route rethrew, the marker disabled both session retry and model fallback, the context stayed above the threshold, and the next prompt repeated the identical failure forever; the circuit breaker never debited because the failure was non-transient.
+- The marker is a session-internal replay-suppression signal. It must stay on the error object the session predicates read, and never appear in what the user is shown.
+
+### Why an extension could not handle it
+
+- The classification is this builtin's private authorization contract for destructive context reduction, and the retry predicate lives inside its own summarization loop; no external extension can observe a summary attempt's provenance or participate in required-compaction admission.
+
+### Expected merge conflict zones
+
+- LOW: `deterministic-fallback.ts` failure-kind union and classifier.
+- LOW: `speculative.ts` `SummaryRequestError` shape, `isRetryableSummaryAttempt`, and the summarization while-loop.
+- LOW: `speculative-summary.ts` stream settlement, `transient-failure.ts` predicate, `index.ts` `recoverRequiredCompaction` call sites.
+
 # changes.md — builtin compaction policy
 
 ## Deterministic resume slice for an over-window restored context (2026-09-10)