diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 36aaaffd4..7ba2b33f7 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -334,7 +334,7 @@ prerequisites; Ollama installation remains outside this flow. ### Profiles (`src/config/profiles.ts`) -Profiles supply per-project or named-profile overrides for `model` and `systemPromptExtensions` (the only allowed keys; any other key is rejected on load). +Profiles supply per-project or named-profile overrides for `model`, `systemPromptExtensions`, `inactivityTimeoutMs`, `totalTimeoutMs`, and `summarizerTimeoutMs` (any other key is rejected on load). `summarizerTimeoutMs` caps the compaction summary call per inference round-trip and defaults to 90 s — well under the director's `totalTimeoutMs`, because compaction runs inline on the reactor and a stalled summary call freezes the session. - Project profile: `.corbits/profile.json` in the repo root — committed, credential-free. - Named profiles: `~/.corbits/profiles/.json` — user-level overrides, inherited via the `profile` key or the `--profile` flag. A missing named file fails closed. A missing project `profile.json` overlay is optional. diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index c8d424bdf..ce949be44 100644 --- a/docs/TELEMETRY.md +++ b/docs/TELEMETRY.md @@ -11,27 +11,34 @@ env kill switches (see Intentional feedback below). Each event carries a small set of properties: -| Event | When | Properties | -| ------------------- | ----------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `cli_start` | Once per used session (see First-run disclosure) | (none beyond common properties) | -| `session_end` | When a TUI session finishes | `status`, `turn_count`, `duration_ms`, `session_mode`, `exit_reason` | -| `$ai_generation` | Once per completed turn (may be sampled); always on turn failure | `$ai_trace_id`, `$ai_provider`, `$ai_model`, `$ai_input_tokens`, `$ai_output_tokens`, `$ai_latency`, `$ai_is_error`, `$ai_error`, `$ai_cache_read_input_tokens`, `$ai_cache_creation_input_tokens`, `$ai_reasoning_tokens`, `tool_call_count`, `tool_error_count`, `subagent_call_count` | -| `$ai_span` | Opt-in only — once per top-level tool call when `CORBITS_TELEMETRY_AI_SPANS` is set | `$ai_trace_id`, `$ai_span_id`, `$ai_parent_id`, `$ai_span_name`, `$ai_is_error` | -| `slash_command` | A slash command is dispatched (shared product-event path) | `command_name` | -| `skill_used` | `use_skill` loads a skill that resolved | (none beyond common properties) | -| `plugin_loaded` | First successful load of a plugin identity in this process | `origin` | -| `subagent_start` | A `spawn_agent` dispatch begins | `agent_name` | -| `subagent_end` | A `spawn_agent` dispatch finishes | `agent_name`, `status`, `duration_ms`, `model`, `turn_count`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_write_tokens`, `reasoning_tokens`, `tool_call_count`, `tool_error_count`, `stop_reason`, `parent_trace_id` | -| `permission_prompt` | An approval prompt is answered (or abandoned) | `decision`, `permission_kind` | -| `compaction` | The compactor actually folds turns away | `mode`, `duration_ms`, `turns_before`, `turns_after` | -| `crash` | A fatal error reaches the process-level handler | `kind`, `error_class` | -| `auth_failure` | A provider rejects the stored credentials | `auth_provider` | -| `survey sent` | User submits intentional feedback via `/feedback` | `$survey_id`, `$survey_response`, `$survey_questions`, `turn_trace_id` | +| Event | When | Properties | +| -------------------- | ----------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `cli_start` | Once per used session (see First-run disclosure) | (none beyond common properties) | +| `session_end` | When a TUI session finishes | `status`, `turn_count`, `duration_ms`, `session_mode`, `exit_reason` | +| `$ai_generation` | Once per completed turn (may be sampled); always on turn failure | `$ai_trace_id`, `$ai_provider`, `$ai_model`, `$ai_input_tokens`, `$ai_output_tokens`, `$ai_latency`, `$ai_is_error`, `$ai_error`, `$ai_cache_read_input_tokens`, `$ai_cache_creation_input_tokens`, `$ai_reasoning_tokens`, `tool_call_count`, `tool_error_count`, `subagent_call_count` | +| `$ai_span` | Opt-in only — once per top-level tool call when `CORBITS_TELEMETRY_AI_SPANS` is set | `$ai_trace_id`, `$ai_span_id`, `$ai_parent_id`, `$ai_span_name`, `$ai_is_error` | +| `slash_command` | A slash command is dispatched (shared product-event path) | `command_name` | +| `skill_used` | `use_skill` loads a skill that resolved | (none beyond common properties) | +| `plugin_loaded` | First successful load of a plugin identity in this process | `origin` | +| `subagent_start` | A `spawn_agent` dispatch begins | `agent_name` | +| `subagent_end` | A `spawn_agent` dispatch finishes | `agent_name`, `status`, `duration_ms`, `model`, `turn_count`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_write_tokens`, `reasoning_tokens`, `tool_call_count`, `tool_error_count`, `stop_reason`, `parent_trace_id` | +| `permission_prompt` | An approval prompt is answered (or abandoned) | `decision`, `permission_kind` | +| `compaction` | The compactor actually folds turns away | `mode`, `duration_ms`, `turns_before`, `turns_after` | +| `summarizer_failure` | The compaction summary call fails after its retry budget is spent | `provider`, `model`, `error_kind`, `duration_ms` | +| `crash` | A fatal error reaches the process-level handler | `kind`, `error_class` | +| `auth_failure` | A provider rejects the stored credentials | `auth_provider` | +| `survey sent` | User submits intentional feedback via `/feedback` | `$survey_id`, `$survey_response`, `$survey_questions`, `turn_trace_id` | `compaction` is deliberately silent on the runs where the compactor decides there is nothing to compact — an event that also fires on no-ops makes its own duration and turn-count averages meaningless. +`summarizer_failure` fires once per failed summary call, not per attempt. +`error_kind` is a first-party enum (`auth`, `provider`, `timeout`, `aborted`, +`empty`, `failed`) — the provider's error text is never sent. `provider` and +`model` are the canonical runtime ids, the same trust class as +`$ai_provider`/`$ai_model`. + Common properties attached to every event: a random installation UUID (`distinct_id`), `session_id`, `$app_version` (PostHog's standard Version property, the running package version), `service_version` (same value, kept diff --git a/src/config/index.ts b/src/config/index.ts index 4e87be859..2552bdd36 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -486,6 +486,9 @@ export interface Config { inactivityTimeoutMs?: number; // Per-call total wall-clock cap in ms (default 600_000 in the harness). totalTimeoutMs?: number; + // Per-call wall-clock cap for the compaction summary call in ms + // (default 90_000 in the summarizer). + summarizerTimeoutMs?: number; reasoningEffort?: ReasoningEffort; mcpServers?: ResolvedMCPServerConfig[]; /** Local project MCP lists replace global lists and require project trust. */ @@ -1005,6 +1008,9 @@ export async function loadConfig( ...(profile.totalTimeoutMs !== undefined ? { totalTimeoutMs: profile.totalTimeoutMs } : {}), + ...(profile.summarizerTimeoutMs !== undefined + ? { summarizerTimeoutMs: profile.summarizerTimeoutMs } + : {}), ...(local?.reasoningEffort !== undefined ? { reasoningEffort: local.reasoningEffort } : {}), diff --git a/src/config/profiles.ts b/src/config/profiles.ts index c915de1da..33433b2da 100644 --- a/src/config/profiles.ts +++ b/src/config/profiles.ts @@ -19,6 +19,10 @@ const ProfileSchema = type({ // Default in the inference harness is 600_000 (10 min). Backstop for // streams that keep emitting forever without terminating. "totalTimeoutMs?": "number >= 1", + // Per-call cap for the compaction summary call in milliseconds. Default + // 90_000 — well under totalTimeoutMs because compaction runs inline on the + // reactor and a stuck summary call freezes the session. + "summarizerTimeoutMs?": "number >= 1", "+": "reject", }); @@ -100,6 +104,8 @@ export async function resolveProfile( merged.inactivityTimeoutMs = projectProfile.inactivityTimeoutMs; if (projectProfile.totalTimeoutMs !== undefined) merged.totalTimeoutMs = projectProfile.totalTimeoutMs; + if (projectProfile.summarizerTimeoutMs !== undefined) + merged.summarizerTimeoutMs = projectProfile.summarizerTimeoutMs; } const resolvedName = profileName ?? projectProfile?.profile; diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 13d71a570..bb31ef370 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -113,6 +113,7 @@ import { createModelSummarizer } from "../session/summarizer.js"; import { ID_PREFIX, LOG_NAMESPACE_ROOT } from "../branding.js"; import type { ReactorEmittedEvent } from "@intx/inference"; import { setAgentSourceUnlessClosed } from "../tui/agent-source-sync.js"; +import { ensureFreshInferenceSource } from "../subagent/refresh-inference-source.js"; import { getToolApprovalBudget } from "../tui/tool-execution-watchdog.js"; import { WorkflowHost } from "../workflows/host.js"; @@ -678,6 +679,20 @@ export async function runExec(config: Config): Promise { getSource: () => liveSource, deps: inferenceDeps, getArchive: () => evidenceArchiveHolder.current, + timeoutMs: config.summarizerTimeoutMs, + telemetry: liveTelemetry, + // A 401 here usually means the shared OAuth file rotated under another + // process; re-read it so the retry runs on the fresh token. + refreshAuth: async () => { + const fresh = await ensureFreshInferenceSource( + liveSource, + config.providers, + ); + if (fresh.apiKey === liveSource.apiKey) return; + liveSource = fresh; + if (currentAgent !== null) + setAgentSourceUnlessClosed(currentAgent, fresh); + }, }); const { activated: activatedToolNames, computeAdvertised } = diff --git a/src/session/summarizer.ts b/src/session/summarizer.ts index 4526f37ba..948869f1a 100644 --- a/src/session/summarizer.ts +++ b/src/session/summarizer.ts @@ -8,11 +8,18 @@ // compact cycle: the caller retains the prior context instead of substituting // a statistics-only stub. +import { type } from "arktype"; import { runInference, type Dependencies } from "@intx/inference"; import { createDefaultDependencies } from "@intx/inference/providers"; import { getLogger } from "@intx/log"; -import type { ConversationTurn, InferenceSource } from "@intx/types/runtime"; +import { + InferenceError, + type ConversationTurn, + type InferenceSource, + type RetryPolicy, +} from "@intx/types/runtime"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js"; import { buildArchiveSummaryExcerpt, type SummaryExcerptArchive, @@ -153,6 +160,18 @@ export function buildSummaryPrompt( return `${workflowPreamble(ctx)}Session excerpt:\n\n${body}`; } +// Per-call wall-clock cap for the summary call. Compaction runs inline on the +// reactor, so a summarizer that inherits the director's 600 s budget freezes +// the session for the full window; the summary prompt is small and a slow +// answer is almost always a stuck call, not a thinking model. +export const DEFAULT_SUMMARIZER_TIMEOUT_MS = 90_000; + +// The harness's default policy retries retryable and timeout categories up to +// three times inside one call. The summarizer owns its retry budget instead — +// one retry per failure class below — so a stalled call cannot multiply into +// minutes of frozen reactor. +const NO_HARNESS_RETRY: RetryPolicy = () => ({ kind: "abort" }); + // Low-level completion: one inference round-trip returning assistant text. // Injectable so tests can drive the summarizer without a live model. export type CompletionFn = ( @@ -161,7 +180,7 @@ export type CompletionFn = ( signal: AbortSignal, ) => Promise; -function defaultComplete(deps: Dependencies): CompletionFn { +function defaultComplete(deps: Dependencies, timeoutMs: number): CompletionFn { return async (turns, source, signal) => { let seq = 0; let out = ""; @@ -171,19 +190,98 @@ function defaultComplete(deps: Dependencies): CompletionFn { signal, nextSeq: () => seq++, deps, + inferenceOptions: { + totalTimeoutMs: timeoutMs, + retryPolicy: NO_HARNESS_RETRY, + }, })) { if (event.type === "inference.done") { for (const block of event.data.turn.content) { if (block.type === "text") out += block.text; } } else if (event.type === "inference.error") { - throw new Error(event.data.error.message); + throw new Error(event.data.error.message, { + cause: event.data.error, + }); } } return out.trim(); }; } +// The class a failed summary call falls into. `auth` and `provider` each earn +// one retry; `timeout` never does — the point of the smaller cap is to stop a +// stalled call from freezing the reactor, and retrying would double the stall. +export type SummarizerFailureClass = + | "auth" + | "provider" + | "timeout" + | "aborted" + | "empty" + | "failed"; + +const EMPTY_SUMMARY_MESSAGE = "compaction summary returned empty text"; + +// xAI's Responses proxy reports mid-stream generation failures as a +// response.failed envelope, which the adapter classifies protocol_mismatch — +// a category the harness never retries, though the fault is transient. +const PROVIDER_INTERNAL_ERROR = /internal error during token generation/i; + +// defaultComplete attaches the harness's classified InferenceError as `cause`; +// errors without one (injected fakes, thrown parser detail) classify by +// bounded message markers. +function inferenceErrorCause(error: unknown): InferenceError | undefined { + if (!(error instanceof Error) || error.cause === undefined) return undefined; + const parsed = InferenceError(error.cause); + return parsed instanceof type.errors ? undefined : parsed; +} + +function classifySummarizerFailure(error: unknown): SummarizerFailureClass { + const cause = inferenceErrorCause(error); + if (cause !== undefined) { + if (cause.category === "aborted") return "aborted"; + if (cause.category === "timeout") return "timeout"; + if ( + cause.category === "credential_failure" || + cause.statusCode === 401 || + cause.statusCode === 403 + ) + return "auth"; + if ( + (cause.statusCode !== undefined && + cause.statusCode >= 500 && + cause.statusCode < 600) || + PROVIDER_INTERNAL_ERROR.test(cause.message) + ) + return "provider"; + return "failed"; + } + const message = error instanceof Error ? error.message : String(error); + if (message === EMPTY_SUMMARY_MESSAGE) return "empty"; + if (error instanceof Error && error.name === "AbortError") return "aborted"; + if (/\b(?:timeout|timed out)\b/i.test(message)) return "timeout"; + if (/\b(?:401|403)\b|\bunauthorized\b/i.test(message)) return "auth"; + if (/\b5\d\d\b/.test(message) || PROVIDER_INTERNAL_ERROR.test(message)) + return "provider"; + return "failed"; +} + +// One-line operator notice for a final failure. The reason named is the +// provider's own first line when short enough to be useful, else the class. +function failureNotice( + failureClass: SummarizerFailureClass, + error: Error, +): string { + const firstLine = error.message.split("\n", 1)[0]?.trim() ?? ""; + const reason = + firstLine.length > 0 + ? firstLine.length > 140 + ? `${firstLine.slice(0, 140)}...` + : firstLine + : failureClass; + return `Compaction summary failed — keeping prior context (${reason})`; +} + export interface ModelSummarizerOptions { /** Returns the source to summarize with — read live so model switches apply. */ getSource: () => InferenceSource; @@ -194,6 +292,22 @@ export interface ModelSummarizerOptions { deps?: Dependencies; /** Cap on the returned summary length. */ maxChars?: number; + /** + * Per-call wall-clock cap for the summary call, profile-configurable via + * `summarizerTimeoutMs`. Deliberately far below the director's + * `totalTimeoutMs` — compaction blocks the reactor, so a stuck summary call + * must give up in seconds, not minutes. + */ + timeoutMs?: number | undefined; + /** + * Re-read the provider credential (OAuth token store) before the single + * `auth` retry. Several processes share one auth file, so a 401 may only + * mean this process holds a token another already rotated. + */ + refreshAuth?: (() => Promise) | undefined; + /** Fires once per failed `summarize` call, after the retry budget is spent. */ + onFailure?: ((text: string) => void) | undefined; + telemetry?: Telemetry | undefined; /** Primary sessions pass the evidence archive so the prompt is not a clipped stub. */ getArchive?: () => SummaryExcerptArchive | undefined; } @@ -208,42 +322,87 @@ export function createModelSummarizer( options: ModelSummarizerOptions, ): (turns: ConversationTurn[], ctx?: SummaryContext) => Promise { const deps = options.deps ?? createDefaultDependencies(); - const complete = options.complete ?? defaultComplete(deps); + const timeoutMs = options.timeoutMs ?? DEFAULT_SUMMARIZER_TIMEOUT_MS; + const complete = options.complete ?? defaultComplete(deps, timeoutMs); const maxChars = options.maxChars ?? 4000; + const telemetry = options.telemetry ?? NOOP_TELEMETRY; return async (turns, ctx) => { - try { - const archive = options.getArchive?.(); - const excerpt = - archive !== undefined - ? await buildArchiveSummaryExcerpt(archive) - : undefined; - const promptTurns: ConversationTurn[] = [ - { - role: "system", - content: [{ type: "text", text: SYSTEM_INSTRUCTION }], - timestamp: turns[0]?.timestamp ?? 0, - }, - { - role: "user", - content: [ - { type: "text", text: buildSummaryPrompt(turns, ctx, excerpt) }, - ], - timestamp: 0, - }, - ]; - const signal = options.getSignal?.() ?? new AbortController().signal; - const text = await complete(promptTurns, options.getSource(), signal); - if (text.length === 0) { - logger.warn("compaction summary call returned empty text"); - throw new Error("compaction summary returned empty text"); + const startedAt = Date.now(); + const archive = options.getArchive?.(); + const excerpt = + archive !== undefined + ? await buildArchiveSummaryExcerpt(archive) + : undefined; + const promptTurns: ConversationTurn[] = [ + { + role: "system", + content: [{ type: "text", text: SYSTEM_INSTRUCTION }], + timestamp: turns[0]?.timestamp ?? 0, + }, + { + role: "user", + content: [ + { type: "text", text: buildSummaryPrompt(turns, ctx, excerpt) }, + ], + timestamp: 0, + }, + ]; + + // One retry per failure class. Auth retries refresh the credential first + // when a hook is wired; provider retries replay the call as-is. + const retried = new Set(); + for (;;) { + try { + const signal = options.getSignal?.() ?? new AbortController().signal; + const text = await complete(promptTurns, options.getSource(), signal); + if (text.length === 0) { + logger.warn("compaction summary call returned empty text"); + throw new Error(EMPTY_SUMMARY_MESSAGE); + } + return text.length > maxChars ? text.slice(0, maxChars) : text; + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + const failureClass = classifySummarizerFailure(err); + const retryable = + (failureClass === "auth" && options.refreshAuth !== undefined) || + failureClass === "provider"; + if (retryable && !retried.has(failureClass)) { + retried.add(failureClass); + logger.warn( + "compaction summary call failed ({class}); retrying once: {error}", + { class: failureClass, error: err.message }, + ); + if (failureClass === "auth") { + try { + await options.refreshAuth?.(); + } catch (refreshError) { + logger.warn( + "credential re-read after summary auth failure failed: {error}", + { + error: + refreshError instanceof Error + ? refreshError.message + : String(refreshError), + }, + ); + } + } + continue; + } + logger.warn("compaction summary call failed: {error}", { + error: err.message, + }); + const source = options.getSource(); + telemetry.capture("summarizer_failure", { + provider: source.provider, + model: source.model, + error_kind: failureClass, + duration_ms: Date.now() - startedAt, + }); + options.onFailure?.(failureNotice(failureClass, err)); + throw err; } - return text.length > maxChars ? text.slice(0, maxChars) : text; - } catch (error) { - logger.warn("compaction summary call failed: {error}", { - error: error instanceof Error ? error.message : String(error), - }); - throw error instanceof Error ? error : new Error(String(error)); } }; } diff --git a/src/telemetry/index.ts b/src/telemetry/index.ts index af7818d86..7c7fa3a23 100644 --- a/src/telemetry/index.ts +++ b/src/telemetry/index.ts @@ -60,6 +60,7 @@ export type TelemetryEvent = | "subagent_end" | "permission_prompt" | "compaction" + | "summarizer_failure" | "crash" | "auth_failure" // PostHog Surveys event name (space included). Intentional operator feedback @@ -190,6 +191,10 @@ const EVENT_PROPERTY_ALLOWLIST: Record = { permission_prompt: ["decision", "permission_kind"], compaction: ["mode", "duration_ms", "turns_before", "turns_after"], + // provider/model are the canonical runtime ids (same trust class as + // $ai_provider/$ai_model); error_kind is the summarizer's first-party + // failure enum, never the provider's error text. + summarizer_failure: ["provider", "model", "error_kind", "duration_ms"], crash: ["kind", "error_class"], // Which provider rejected the credentials, not why — the rejection detail is // provider-authored text and error_class means a JS constructor name. diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 74f16d64f..631c94931 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -74,6 +74,8 @@ import { createModelSummarizer, type SummaryContext, } from "../../session/summarizer.js"; +import { ensureFreshInferenceSource } from "../../subagent/refresh-inference-source.js"; +import { setAgentSourceUnlessClosed } from "../agent-source-sync.js"; import { createSessionCostAccumulator } from "../../cost/session-cost.js"; import { createSessionOperationQueue } from "../session-operation-queue.js"; import { createDeliveryGeneration } from "../queued-delivery.js"; @@ -510,6 +512,22 @@ export async function assembleTUISession( getSource: () => state.liveSource, deps: start.inferenceDeps, getArchive: () => evidenceArchiveHolder.current, + timeoutMs: config.summarizerTimeoutMs, + telemetry: liveTelemetry, + // A 401 here usually means the shared OAuth file rotated under another + // process; re-read it so the retry runs on the fresh token, and keep the + // live source in step so the summarizer picks it up. + refreshAuth: async () => { + const fresh = await ensureFreshInferenceSource( + state.liveSource, + state.config.providers, + ); + if (fresh.apiKey === state.liveSource.apiKey) return; + state.liveSource = fresh; + if (state.currentAgent !== undefined) + setAgentSourceUnlessClosed(state.currentAgent, fresh); + }, + onFailure: (text) => state.systemNotice?.(text), }); const summaryContext = (): SummaryContext | undefined => { const status = workflowHost.status(); diff --git a/tests/unit/summarizer.test.ts b/tests/unit/summarizer.test.ts index 1a0bc9db1..8f63acf11 100644 --- a/tests/unit/summarizer.test.ts +++ b/tests/unit/summarizer.test.ts @@ -1,10 +1,13 @@ import { test, expect } from "bun:test"; +import { setupHarness } from "@intx/inference-testing"; import type { ConversationTurn, InferenceSource } from "@intx/types/runtime"; import { buildSummaryPrompt, condenseTurns, createModelSummarizer, + DEFAULT_SUMMARIZER_TIMEOUT_MS, } from "../../src/session/summarizer.js"; +import type { Telemetry, TelemetryEvent } from "../../src/telemetry/index.js"; const source: InferenceSource = { id: "test", @@ -144,3 +147,222 @@ test("buildSummaryPrompt uses a supplied excerpt instead of condensing turns", ( expect(prompt).toContain("ARCHIVE_EXCERPT_BODY"); expect(prompt).not.toContain("Turns dropped"); }); + +function stubTelemetry(): { + telemetry: Telemetry; + events: { + event: TelemetryEvent; + properties?: Record | undefined; + }[]; +} { + const events: { + event: TelemetryEvent; + properties?: Record | undefined; + }[] = []; + const telemetry: Telemetry = { + enabled: true, + installationId: "test", + capture: (event, properties) => { + events.push({ event, properties }); + }, + captureIntentional: () => false, + flush: async () => undefined, + discard: () => undefined, + }; + return { telemetry, events }; +} + +// Structured InferenceError riding `cause`, exactly how defaultComplete +// rethrows the harness's classified error. +function inferenceFailure(fields: { + category: string; + message: string; + statusCode?: number; +}): Error { + return new Error(fields.message, { + cause: { + category: fields.category, + message: fields.message, + ...(fields.statusCode !== undefined + ? { statusCode: fields.statusCode } + : {}), + }, + }); +} + +test("summarizer timeout is honoured independently of the director total timeout", async () => { + const harness = setupHarness({ enableInferenceTimers: true }); + try { + // The stream parks forever; only the summarizer's own timer can end the call. + harness.scenario.stall(); + const summarize = createModelSummarizer({ + getSource: () => ({ + id: "anthropic", + provider: "anthropic", + model: "claude-test", + baseURL: "https://api.anthropic.com", + apiKey: "k", + }), + deps: harness.deps, + timeoutMs: 30_000, + }); + const pending = summarize(turns()); + await harness.run(); + await expect(pending).rejects.toThrow("total timeout (30000 ms"); + } finally { + harness.dispose(); + } +}); + +test("default summarizer timeout stays well under the director's 600s", () => { + expect(DEFAULT_SUMMARIZER_TIMEOUT_MS).toBeLessThanOrEqual(120_000); + expect(DEFAULT_SUMMARIZER_TIMEOUT_MS).toBeGreaterThanOrEqual(60_000); +}); + +test("a 401 retries once after a credential re-read", async () => { + let calls = 0; + let refreshes = 0; + const summarize = createModelSummarizer({ + getSource: () => source, + refreshAuth: async () => { + refreshes++; + }, + complete: async () => { + calls++; + if (calls === 1) + throw inferenceFailure({ + category: "credential_failure", + message: "Unauthorized", + statusCode: 401, + }); + return "## What Happened\n- recovered"; + }, + }); + const result = await summarize(turns()); + expect(result).toContain("recovered"); + expect(calls).toBe(2); + expect(refreshes).toBe(1); +}); + +test("a second 401 fails: retry budget is spent once", async () => { + let calls = 0; + let refreshes = 0; + const notices: string[] = []; + const summarize = createModelSummarizer({ + getSource: () => source, + refreshAuth: async () => { + refreshes++; + }, + onFailure: (text) => notices.push(text), + complete: async () => { + calls++; + throw new Error("HTTP 401 Unauthorized"); + }, + }); + await expect(summarize(turns())).rejects.toThrow("401"); + expect(calls).toBe(2); + expect(refreshes).toBe(1); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain("401"); +}); + +test("a 401 without a refresh hook is not retried", async () => { + let calls = 0; + const summarize = createModelSummarizer({ + getSource: () => source, + complete: async () => { + calls++; + throw new Error("HTTP 401 Unauthorized"); + }, + }); + await expect(summarize(turns())).rejects.toThrow("401"); + expect(calls).toBe(1); +}); + +test("a provider 5xx retries exactly once", async () => { + let calls = 0; + let refreshes = 0; + const summarize = createModelSummarizer({ + getSource: () => source, + refreshAuth: async () => { + refreshes++; + }, + complete: async () => { + calls++; + if (calls <= 2) + throw inferenceFailure({ + category: "retryable", + message: "HTTP 503 Service Unavailable", + statusCode: 503, + }); + return "done"; + }, + }); + await expect(summarize(turns())).rejects.toThrow("503"); + expect(calls).toBe(2); + expect(refreshes).toBe(0); +}); + +test("a provider internal-generation error retries once", async () => { + let calls = 0; + const summarize = createModelSummarizer({ + getSource: () => source, + complete: async () => { + calls++; + if (calls === 1) + throw inferenceFailure({ + category: "protocol_mismatch", + message: "grok-responses: Internal error during token generation", + }); + return "## What Happened\n- recovered"; + }, + }); + const result = await summarize(turns()); + expect(result).toContain("recovered"); + expect(calls).toBe(2); +}); + +test("a timeout is never retried", async () => { + let calls = 0; + const summarize = createModelSummarizer({ + getSource: () => source, + complete: async () => { + calls++; + throw inferenceFailure({ + category: "timeout", + message: "inference call exceeded total timeout (90000 ms wall-clock)", + }); + }, + }); + await expect(summarize(turns())).rejects.toThrow("total timeout"); + expect(calls).toBe(1); +}); + +test("final failure throws, notices once, and reports telemetry", async () => { + const { telemetry, events } = stubTelemetry(); + const notices: string[] = []; + let calls = 0; + const summarize = createModelSummarizer({ + getSource: () => source, + telemetry, + onFailure: (text) => notices.push(text), + complete: async () => { + calls++; + throw inferenceFailure({ + category: "retryable", + message: "HTTP 500 Internal Server Error", + statusCode: 500, + }); + }, + }); + await expect(summarize(turns())).rejects.toThrow("500"); + expect(calls).toBe(2); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain("500"); + expect(events).toHaveLength(1); + expect(events[0]?.event).toBe("summarizer_failure"); + expect(events[0]?.properties?.error_kind).toBe("provider"); + expect(events[0]?.properties?.provider).toBe("openai"); + expect(events[0]?.properties?.model).toBe("test-model"); + expect(typeof events[0]?.properties?.duration_ms).toBe("number"); +});