diff --git a/CHANGELOG.md b/CHANGELOG.md index 732f585e..718399f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,8 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename - Cancelling a `task` or `wait_agents` worker reports wait status `interrupted`, not `failed`. +- Inference no longer fails over to a backup provider. A selected-provider + failure stays on that provider; switch with `/model`. ### Fixed @@ -46,6 +48,8 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename when a followup is already in flight. - `interrupt_agent` flips the wait mailbox so soft interrupt unblocks `wait_agents` while the background run is still in flight. +- Credential-refresh and auth send failures tell the user to log in again + instead of suggesting `/model`. ## [0.3.11] - 2026-08-31 diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index a61c07c5..2ae63a96 100644 --- a/docs/TELEMETRY.md +++ b/docs/TELEMETRY.md @@ -169,11 +169,11 @@ settles an unresolved turn once as an unsampled terminal failure. Attribution uses the latest `inference.usage` source, the first lifecycle payload carrying the runtime-resolved provider/model pair for an attempt. Each `inference.start` clears that authoritative source and records the newly attempted model. If a -fallback fails before usage exposes its source, telemetry retains that actual -model but uses the fixed `unknown` provider/source bucket rather than attributing -it to the previously selected provider. When the attempted model still matches -the selected source, that full source remains valid. Therefore a parent turn -emits at most one terminal `$ai_generation`, including retry and failover paths. +retry fails before usage exposes its source, telemetry retains that attempted +model but uses the fixed `unknown` provider/source bucket. When the attempted +model still matches the selected source, that full source remains valid. +Therefore a parent turn emits at most one terminal `$ai_generation`, including +retry paths. ## What's never collected diff --git a/src/config/inference-sources.ts b/src/config/inference-sources.ts index d82acce6..c0741715 100644 --- a/src/config/inference-sources.ts +++ b/src/config/inference-sources.ts @@ -13,7 +13,6 @@ import type { Settings } from "./settings.js"; import { resolveSessionEffort, type ReasoningEffort } from "../provider/reasoning-effort.js"; import { SOURCE_MAX_TOKENS } from "./index.js"; import { isOpenCodeGoProvider } from "../../packages/opencode-go/src/index.js"; -import { resolveDefaultModel } from "./providers.js"; export interface BuildSourceContext { sessionId: string; @@ -21,38 +20,13 @@ export interface BuildSourceContext { catalog: readonly ProviderCatalogEntry[]; } -// A resolved provider+model, with optional reasoningEffort — the unit both -// the primary source and its backups are built from. +// A resolved provider+model with optional reasoning effort. export interface ProviderRef { provider: string; model: string; reasoningEffort?: ReasoningEffort; } -function refKey(ref: ProviderRef): string { - return `${ref.provider}\0${ref.model}`; -} - -// Every other configured provider, one model each, so a primary source that -// fails to build (bad credentials, missing baseURL) still has somewhere to -// fall back to. Order follows settings.providers; providers already covered -// by `existing` are skipped. -function backupRefsFromSettings( - settings: Settings, - existing: readonly ProviderRef[], -): ProviderRef[] { - const seenProviders = new Set(existing.map((r) => r.provider)); - const tail: ProviderRef[] = []; - for (const [provider, p] of Object.entries(settings.providers)) { - if (seenProviders.has(provider)) continue; - const model = resolveDefaultModel(p); - if (model === undefined || model.length === 0) continue; - seenProviders.add(provider); - tail.push({ provider, model }); - } - return tail; -} - function catalogEntry( catalog: readonly ProviderCatalogEntry[], provider: string, @@ -170,38 +144,6 @@ export function buildInferenceSourceForRef( }; } -export function buildSourcesFromRefs( - refs: readonly ProviderRef[], - ctx: BuildSourceContext, - settings: Settings | undefined, -): InferenceSource[] { - const out: InferenceSource[] = []; - const seenIds = new Set(); - for (const ref of refs) { - let src: InferenceSource | null; - try { - src = buildInferenceSourceForRef(ref, ctx, settings); - } catch { - // A leftover sibling URL (e.g. Custom `/api/tags`) must not take down the - // whole bundle. Head failure is re-checked in `buildSourceBundle`. - continue; - } - if (src === null) continue; - if (seenIds.has(src.id)) continue; - seenIds.add(src.id); - out.push(src); - } - return out; -} - -export function prependActiveRef(refs: readonly ProviderRef[], active: ProviderRef): ProviderRef[] { - const without = refs.filter((r) => refKey(r) !== refKey(active)); - return [active, ...without]; -} - -// Builds the primary source for `head` plus one backup per other configured -// provider, so a mid-run failure (bad credentials, dropped connection) has -// somewhere else to go. `head` always wins as defaultSource when it builds. function buildSourceBundle(args: { settings: Settings | undefined; catalog: readonly ProviderCatalogEntry[]; @@ -215,30 +157,13 @@ function buildSourceBundle(args: { ...(args.reasoningEffort !== undefined ? { reasoningEffort: args.reasoningEffort } : {}), }; - const refs = - args.settings !== undefined - ? prependActiveRef(backupRefsFromSettings(args.settings, [args.head]), args.head) - : [args.head]; - - const sources = buildSourcesFromRefs(refs, ctx, args.settings); - const defaultId = args.head.provider; - const hasDefault = sources.some((s) => s.id === defaultId); - if (!hasDefault) { - let fallback: InferenceSource | null; - try { - fallback = buildInferenceSourceForRef(args.head, ctx, args.settings); - } catch (error) { - throw new Error(`No inference source for provider "${defaultId}"`, { cause: error }); - } - if (fallback === null) { - throw new Error(`No inference source for provider "${defaultId}"`); - } - return { sources: [fallback, ...sources], defaultSource: fallback.id }; + const source = buildInferenceSourceForRef(args.head, ctx, args.settings); + if (source === null) { + throw new Error( + `Unable to build inference source for selected provider "${args.head.provider}"`, + ); } - return { - sources, - defaultSource: defaultId, - }; + return { sources: [source], defaultSource: source.id }; } export function buildMainSessionSources(args: { diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 2a978a96..531bc9c7 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -13,12 +13,7 @@ import { noopAuditStore, permissiveAuthorize } from "@intx/agent/testing"; import { getLogger } from "@intx/log"; import { createOptimizedContextStore } from "../session/optimized-context-store.js"; import { type } from "arktype"; -import { - buildCodexSource, - buildOpenAISource, - buildXaiSource, - type Config, -} from "../config/index.js"; +import { type Config } from "../config/index.js"; import { loadLocalSettings, resolveLocalSettingsPath, @@ -67,6 +62,11 @@ import { createAgentToolset, type AgentToolset, type OperatorResult } from "../a import { createAgentWithLiveToolDispatch } from "../agent/live-tool-dispatch.js"; import { liveTelemetry } from "../telemetry/singleton.js"; import { createTurnObserver } from "../telemetry/ai-observability.js"; +import { + CREDENTIAL_FAILURE_USER_MESSAGE, + isResolvedProviderFailureError, + terminalProviderFailureMessage, +} from "../inference-error-message.js"; import { collectToolPlugins, resolveToolPlugins } from "../plugins/tool-plugins.js"; import { expandExistingPluginMembers, @@ -120,6 +120,35 @@ export function formatCaughtError(err: unknown): string { return err instanceof Error ? err.message : String(err); } +const SELECTED_PROVIDER_FAILURE = "SelectedProviderFailure"; + +export async function refreshSelectedProviderCredential(refresh: () => Promise): Promise { + try { + return await refresh(); + } catch (cause) { + const error = new Error(formatCaughtError(cause), { cause }); + error.name = SELECTED_PROVIDER_FAILURE; + throw error; + } +} + +export function execUserFailureMessage( + config: Config, + err: unknown, + providerFailureObserved: boolean, +): string { + if (err instanceof Error && err.name === SELECTED_PROVIDER_FAILURE) { + return CREDENTIAL_FAILURE_USER_MESSAGE; + } + if (providerFailureObserved || isResolvedProviderFailureError(err)) { + return terminalProviderFailureMessage( + config.providerName, + config.settings?.providers[config.providerName]?.name, + ); + } + return formatCaughtError(err); +} + /** * Headless analogue of TUI `runtime-shutdown`: abort live workers, then close * the primary agent and dispose the toolset. `cancelAll` is fire-and-forget — @@ -297,6 +326,7 @@ export async function runExec(config: Config): Promise { let finalized = false; let turnsUsed = 0; let runSink: RunSink | null = null; + let providerFailureObserved = false; const persist = async ( status: "running" | "done" | "failed" | "cancelled", @@ -586,56 +616,14 @@ export async function runExec(config: Config): Promise { const initialCodexProfile = codexProfileFromProviderName(config.providerName); const initialXaiProfile = xaiProfileFromProviderName(config.providerName); - const initialCodexAccountId = config.providers.find( - (p) => p.name === config.providerName, - )?.codexAccountId; - - const buildOpenAICompatibleInitialSource = (): InferenceSource => - buildOpenAISource({ - id: config.providerName, - baseURL: config.baseURL, - apiKey: config.apiKey, - model: config.model, - ...(config.reasoningEffort !== undefined - ? { reasoningEffort: config.reasoningEffort } - : {}), - }); - - const buildSessionSources = (): { sources: InferenceSource[]; defaultSource: string } => - buildSessionSourcesFromConfig(config, sessionId); - - const initialBundle = buildSessionSources(); + const initialBundle = buildSessionSourcesFromConfig(config, sessionId); const liveSources = initialBundle.sources; const liveDefaultSource = initialBundle.defaultSource; - - const buildInitialSourceFallback = (): InferenceSource => - initialCodexProfile !== undefined - ? buildCodexSource({ - id: config.providerName, - apiKey: config.apiKey, - model: config.model, - sessionId, - ...(initialCodexAccountId !== undefined ? { accountId: initialCodexAccountId } : {}), - ...(config.reasoningEffort !== undefined - ? { reasoningEffort: config.reasoningEffort } - : {}), - }) - : initialXaiProfile !== undefined - ? buildXaiSource({ - id: config.providerName, - apiKey: config.apiKey, - model: config.model, - sessionId, - ...(config.reasoningEffort !== undefined - ? { reasoningEffort: config.reasoningEffort } - : {}), - }) - : buildOpenAICompatibleInitialSource(); - - let liveSource: InferenceSource = - liveSources.find((s) => s.id === liveDefaultSource) ?? - liveSources[0] ?? - buildInitialSourceFallback(); + const selectedSource = liveSources[0]; + if (selectedSource === undefined) { + throw new Error("Selected inference source was not assembled"); + } + let liveSource: InferenceSource = selectedSource; // Refresh pinned Codex instructions before first inference, same as the // TUI path. Best-effort: a network failure falls back to the disk cache @@ -651,7 +639,9 @@ export async function runExec(config: Config): Promise { // Refresh OAuth tokens before first inference when starting on codex/xai. if (initialCodexProfile !== undefined) { - const { access } = await getValidCodexToken(initialCodexProfile); + const { access } = await refreshSelectedProviderCredential(() => + getValidCodexToken(initialCodexProfile), + ); liveSource = { ...liveSource, apiKey: access }; liveSubAgentProvider.current = { ...liveSubAgentProvider.current, @@ -659,7 +649,9 @@ export async function runExec(config: Config): Promise { }; } if (initialXaiProfile !== undefined) { - const { access } = await getValidXaiToken(initialXaiProfile); + const { access } = await refreshSelectedProviderCredential(() => + getValidXaiToken(initialXaiProfile), + ); liveSource = { ...liveSource, apiKey: access }; liveSubAgentProvider.current = { ...liveSubAgentProvider.current, @@ -762,6 +754,11 @@ export async function runExec(config: Config): Promise { // its partial output in partial.jsonl instead of vanishing. const cycleRecorder = createCycleTextRecorder(() => workdir); const sink = (event: ReactorEmittedEvent): void => { + if (event.type === "inference.start" || event.type === "inference.done") { + providerFailureObserved = false; + } else if (event.type === "inference.error") { + providerFailureObserved = true; + } liveSink.sink(event); cycleRecorder.handleEvent(event); if (event.type === "inference.text.delta") { @@ -881,17 +878,24 @@ export async function runExec(config: Config): Promise { }); if (!sendCompleted || runError !== undefined || summaryStatus === "failed") { - const message = + const diagnosticMessage = runError ?? (summaryStatus === "cancelled" ? "run cancelled before completion" : "run failed"); - stderr.write(`Error: ${message}\n`); + const userMessage = + summaryStatus === "failed" + ? terminalProviderFailureMessage( + config.providerName, + config.settings?.providers[config.providerName]?.name, + ) + : diagnosticMessage; + stderr.write(`Error: ${userMessage}\n`); const persistStatus = summaryStatus === "cancelled" ? "cancelled" : "failed"; - await persist(persistStatus, { error: message }); + await persist(persistStatus, { error: diagnosticMessage }); return { exitCode: 1, sessionId, text: textOut, - error: message, + error: userMessage, status: summaryStatus, durationMs: finishedAt - startedAt, turnsUsed: runSink.getTurnCount(), @@ -916,15 +920,16 @@ export async function runExec(config: Config): Promise { model: config.model, }; } catch (err) { - const message = err instanceof Error ? err.message : String(err); - logger.error("exec failed: {error}", { error: message }); - stderr.write(`Error: ${message}\n`); - await persist("failed", { error: message }); + const diagnosticMessage = formatCaughtError(err); + logger.error("exec failed: {error}", { error: diagnosticMessage }); + const userMessage = execUserFailureMessage(config, err, providerFailureObserved); + stderr.write(`Error: ${userMessage}\n`); + await persist("failed", { error: diagnosticMessage }); return { exitCode: 1, sessionId, text: textOut, - error: message, + error: userMessage, status: "failed", durationMs: Date.now() - startedAt, turnsUsed: runSink?.getTurnCount() ?? turnsUsed, diff --git a/src/inference-error-message.test.ts b/src/inference-error-message.test.ts index 4b9f2d77..17969610 100644 --- a/src/inference-error-message.test.ts +++ b/src/inference-error-message.test.ts @@ -1,6 +1,9 @@ import { describe, expect, test } from "bun:test"; -import { inferenceErrorMessage } from "./inference-error-message.js"; +import { + inferenceErrorMessage, + terminalProviderFailureMessage, +} from "./inference-error-message.js"; const CODEX_BODY = { detail: { @@ -89,3 +92,31 @@ describe("inferenceErrorMessage", () => { expect(line.toLowerCase()).toMatch(/log in again|sign in again/); }); }); + +describe("terminalProviderFailureMessage", () => { + test("uses the selected provider display label in the terminal guidance", () => { + expect(terminalProviderFailureMessage("openai", "OpenAI")).toBe( + 'OpenAI Provider failed. Try again or switch with "/model" and select another.', + ); + }); + + test("falls back to the selected provider id", () => { + expect(terminalProviderFailureMessage("custom-provider")).toBe( + 'custom-provider Provider failed. Try again or switch with "/model" and select another.', + ); + }); + + test("does not duplicate Provider in configured display labels", () => { + expect(terminalProviderFailureMessage("codex/work", "Codex Provider")).toBe( + 'Codex Provider failed. Try again or switch with "/model" and select another.', + ); + }); + + test("uses a safe label when the provider id contains only control sequences", () => { + const message = terminalProviderFailureMessage("\u001b[31m\u001b[0m"); + expect(message).toBe( + 'Unknown Provider failed. Try again or switch with "/model" and select another.', + ); + expect(message).not.toContain("\u001b"); + }); +}); diff --git a/src/inference-error-message.ts b/src/inference-error-message.ts index 3b75b9ea..772add3e 100644 --- a/src/inference-error-message.ts +++ b/src/inference-error-message.ts @@ -11,6 +11,7 @@ import { parseCodexUsageLimitError, } from "./auth/codex/usage-limit-error.js"; import { codexProfileFromProviderName, isCodexProviderName } from "./config/codex-providers.js"; +import { stripTerminalControlSequences } from "./util/control-char-strip.js"; import { gatewayOverloadUserMessage, isCodexShortRateLimitInferenceError, @@ -91,6 +92,44 @@ function codexUsageLimitLine(error: InferenceErrorLike): string | undefined { return undefined; } +export function terminalProviderFailureMessage(providerId: string, displayLabel?: string): string { + const preferred = displayLabel?.trim() || providerId; + const sanitized = stripTerminalControlSequences(preferred).replace(/\s+/g, " ").trim(); + const label = (sanitized.length > 0 ? sanitized : "Unknown").replace(/\s+Provider$/i, ""); + return `${label} Provider failed. Try again or switch with "/model" and select another.`; +} + +export type ResolvedProviderFailureError = Error & { + readonly name: "ResolvedProviderFailureError"; + readonly providerId: string; + readonly diagnosticMessage: string; +}; + +export function createResolvedProviderFailureError( + providerId: string, + diagnosticMessage: string, + displayLabel?: string, +): ResolvedProviderFailureError { + return Object.assign(new Error(terminalProviderFailureMessage(providerId, displayLabel)), { + name: "ResolvedProviderFailureError" as const, + providerId, + diagnosticMessage, + }); +} + +export function isResolvedProviderFailureError( + error: unknown, +): error is ResolvedProviderFailureError { + return ( + error instanceof Error && + error.name === "ResolvedProviderFailureError" && + "providerId" in error && + typeof error.providerId === "string" && + "diagnosticMessage" in error && + typeof error.diagnosticMessage === "string" + ); +} + /** One line describing the failure, falling back to the provider's own message. */ export function inferenceErrorMessage(error: InferenceErrorLike): string { if (isGatewayOverloadInferenceError(error)) return gatewayOverloadUserMessage(error); diff --git a/src/session/run-sink.test.ts b/src/session/run-sink.test.ts index 1e52e558..43628452 100644 --- a/src/session/run-sink.test.ts +++ b/src/session/run-sink.test.ts @@ -174,7 +174,7 @@ describe("createRunSink", () => { expect(failures).toEqual([{ turnIndex: 0, error: "429 rate limit" }]); }); - test("uses unknown attribution when a fallback model fails before usage", () => { + test("uses unknown attribution when a retry model fails before usage", () => { const { captured, runSink } = attributionHarness(); runSink.sink(event("inference.start", { model: "model-b" })); @@ -203,14 +203,14 @@ describe("createRunSink", () => { }); }); - test("uses authoritative usage attribution for a failed fallback", () => { + test("uses authoritative usage attribution for a failed retry attempt", () => { const { captured, runSink } = attributionHarness(); runSink.sink(event("inference.start", { model: "model-b" })); runSink.sink( event("inference.usage", { usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, - source: { sourceId: "fallback", provider: "provider-b", model: "model-b" }, + source: { sourceId: "retry", provider: "provider-b", model: "model-b" }, }), ); failMessageRun(runSink); @@ -223,7 +223,7 @@ describe("createRunSink", () => { }); }); - test("does not leak authoritative source attribution across retry attempts", () => { + test("retains selected provider attribution across a same-provider retry", () => { const { captured, runSink } = attributionHarness(); runSink.sink(event("inference.start", { model: "model-a" })); @@ -234,13 +234,13 @@ describe("createRunSink", () => { }), ); runSink.sink(event("inference.error", { error: { message: "retry" } })); - runSink.sink(event("inference.start", { model: "model-b" })); + runSink.sink(event("inference.start", { model: "model-a" })); failMessageRun(runSink); expect(captured).toHaveLength(1); expect(captured[0]?.properties).toMatchObject({ - $ai_provider: "unknown", - $ai_model: "model-b", + $ai_provider: "provider-a", + $ai_model: "model-a", $ai_is_error: true, }); }); diff --git a/src/session/run-sink.ts b/src/session/run-sink.ts index 068b5ba0..dbe04d68 100644 --- a/src/session/run-sink.ts +++ b/src/session/run-sink.ts @@ -26,8 +26,7 @@ export interface RunSinkArgs { // while the turn index is the collector's current in-flight turn count. onTurnStarted?: (info: { turnIndex: number; model: string }) => void; // inference.usage is the first attempt event carrying the runtime-resolved - // provider/model pair. It remains authoritative even when the selected source - // outside the reactor has not changed during fallback. + // provider/model pair. It remains authoritative across retry attempts. onTurnSourceObserved?: (info: { turnIndex: number; source: LastCycleSource }) => void; // Continues a resumed session's persisted run.json turn count instead of // restarting the collector at zero. diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index f9dec608..555691fa 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -87,6 +87,7 @@ import { } from "./authority.js"; import { formatSubAgentSpawnAuthFailureMessage } from "./inference-auth-failure.js"; +import { isResolvedProviderFailureError } from "../inference-error-message.js"; import { isSubAgentCancelError } from "./dispose.js"; import { createInterventionLog, type InterventionSink } from "./intervention-log.js"; @@ -97,6 +98,7 @@ interface FleetRecord { status: WaitJSONStatus; report?: string; error?: string; + providerFailure?: true; /** Set once a wait_agents caller has been handed this result. */ collected?: boolean; /** Set once the payload has been compacted away to bound memory. */ @@ -117,6 +119,7 @@ interface FleetOverlay { lastWaitStatus?: WaitJSONStatus; tombstoned?: boolean; hint?: string; + providerFailure?: true; } const RECOVERY_HINT = @@ -169,6 +172,12 @@ class FleetMailbox { return record !== undefined && record.status !== "running" && record.collected !== true; } + markProviderFailure(id: string): void { + const existing = this.records.get(id); + if (existing === undefined) return; + existing.providerFailure = true; + } + /** * Overlay wait-status override so wait unblocks while the session may still * be running (send_input interrupt:true followup, close_agent teardown). @@ -300,6 +309,7 @@ class FleetMailbox { ...(overlay.hint !== undefined ? { hint: overlay.hint } : {}), ...(payload?.report !== undefined ? { report: payload.report } : {}), ...(payload?.error !== undefined && status === "failed" ? { error: payload.error } : {}), + ...(overlay.providerFailure === true ? { providerFailure: true } : {}), }; } @@ -868,7 +878,13 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { if (!childCtl.signal.aborted) childCtl.abort(); }); + let providerFailureObserved = false; const onEvent = (event: ReactorEmittedEvent): void => { + if (event.type === "inference.start" || event.type === "inference.done") { + providerFailureObserved = false; + } else if (event.type === "inference.error") { + providerFailureObserved = true; + } deps.sessions.appendEvent(session.id, event); deps.onEvent?.(event); }; @@ -1078,9 +1094,17 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { deps.sessions.settleRun(session.id); return; } - // Auth failures keep the actionable Re-authenticate wording. + const diagnosticMessage = isResolvedProviderFailureError(err) + ? err.diagnosticMessage + : err instanceof Error + ? err.message + : String(err); + const isProviderFailure = isResolvedProviderFailureError(err); const authMessage = formatSubAgentSpawnAuthFailureMessage(description, err); - const failReason = authMessage ?? (err instanceof Error ? err.message : String(err)); + const failReason = authMessage ?? (isProviderFailure ? err.message : diagnosticMessage); + if (isProviderFailure || providerFailureObserved) { + deps.fleetRecords.markProviderFailure(session.id); + } deps.sessions.fail(session.id, failReason); }) .finally(() => { @@ -1225,6 +1249,7 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { ? { report: taken.report } : {}), ...(taken.error !== undefined ? { error: taken.error } : {}), + ...(taken.providerFailure === true ? { provider_failure: true } : {}), ...(taken.hint !== undefined ? { hint: taken.hint } : {}), }; }); diff --git a/src/subagent/run-event-settlement.ts b/src/subagent/run-event-settlement.ts new file mode 100644 index 00000000..ab77d98c --- /dev/null +++ b/src/subagent/run-event-settlement.ts @@ -0,0 +1,52 @@ +import type { ReactorEmittedEvent } from "@intx/inference"; + +export interface PendingRunSettlement { + readonly settled: Promise; + cancel: () => void; +} + +export interface RunEventSettlement { + beginSend: () => PendingRunSettlement; + handleEvent: (event: ReactorEmittedEvent) => void; + endStream: () => void; +} + +/** + * Coordinates Agent.send() with its streamed connector.reply. Agent.send resolves + * when the connector reply is produced, which can precede consumption of earlier + * inference.error events from the same run. + */ +export function createRunEventSettlement(): RunEventSettlement { + const pending: { resolve: () => void }[] = []; + let streamEnded = false; + + return { + beginSend: () => { + const entry: { resolve: () => void } = { + resolve: () => { + throw new Error("Run settlement resolver was not initialized"); + }, + }; + const settled = new Promise((resolve) => { + entry.resolve = resolve; + }); + if (streamEnded) entry.resolve(); + else pending.push(entry); + return { + settled, + cancel: () => { + const index = pending.indexOf(entry); + if (index >= 0) pending.splice(index, 1); + }, + }; + }, + handleEvent: (event) => { + if (event.type !== "connector.reply") return; + pending.shift()?.resolve(); + }, + endStream: () => { + streamEnded = true; + for (const entry of pending.splice(0)) entry.resolve(); + }, + }; +} diff --git a/src/subagent/run-resolved-provider-failure.test.ts b/src/subagent/run-resolved-provider-failure.test.ts new file mode 100644 index 00000000..2163e5d9 --- /dev/null +++ b/src/subagent/run-resolved-provider-failure.test.ts @@ -0,0 +1,174 @@ +import { describe, expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { AgentTool } from "@intx/agent"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js"; +import { + isResolvedProviderFailureError, + type ResolvedProviderFailureError, +} from "../inference-error-message.js"; +import { createPermissionGate } from "../permission/gate.js"; +import { createFleetMailbox, createSpawnAgentTool, createWaitAgentsTool } from "./agent-fleet.js"; +import { createSubAgentSessionStore } from "./session-store.js"; +import type { RunSubAgentParams, RunSubAgentResult } from "./types.js"; + +const RAW_DIAGNOSTIC = "POST https://provider.invalid returned secret response body"; +const SAFE_MESSAGE = + 'test-provider Provider failed. Try again or switch with "/model" and select another.'; +const provider = { + providerName: "test-provider", + baseURL: "http://localhost", + model: "test-model", +}; +const testPermissionGate = createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, +}); + +type Run = (params: RunSubAgentParams) => Promise; + +async function withResolvedProviderRun( + callback: (run: Run, cwd: string, observed: ReactorEmittedEvent[]) => Promise, +): Promise { + const cwd = await mkdtemp(join(tmpdir(), "resolved-provider-failure-")); + const observed: ReactorEmittedEvent[] = []; + try { + return await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => + ({ + send: async () => { + await new Promise((resolve) => queueMicrotask(resolve)); + return { reply: RAW_DIAGNOSTIC, turn: { role: "assistant", content: [] } }; + }, + stream: () => + (async function* (): AsyncGenerator { + yield { + type: "inference.start", + seq: 1, + data: { sourceId: "test", model: "test-model", input: [] }, + } as unknown as ReactorEmittedEvent; + yield { + type: "inference.error", + seq: 2, + data: { + error: { category: "fatal", message: RAW_DIAGNOSTIC }, + partial: { text: "" }, + }, + } as unknown as ReactorEmittedEvent; + yield { + type: "connector.reply", + seq: 3, + data: { content: RAW_DIAGNOSTIC }, + } as unknown as ReactorEmittedEvent; + })(), + deliver: () => {}, + close: async () => {}, + setSource: () => {}, + setSources: () => {}, + history: async () => [], + checkpoints: async () => [], + readAt: async () => [], + blobReader: {}, + }) as unknown as Awaited>, + }), + async () => { + const { runSubAgent } = await import("./run.js"); + const run: Run = (params) => + runSubAgent({ + ...params, + onEvent: (event) => { + observed.push(event); + params.onEvent?.(event); + }, + }); + return callback(run, cwd, observed); + }, + ); + } finally { + await rm(cwd, { recursive: true, force: true }); + } +} + +async function callTool(tool: AgentTool, name: string, args: Record) { + if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); + return tool.handler({ id: `${name}-call`, name, arguments: args }, new AbortController().signal); +} + +function runParams(cwd: string): RunSubAgentParams { + return { + cwd, + workdirBase: join(cwd, ".ctx"), + permissionGate: testPermissionGate, + provider, + description: "provider failure probe", + prompt: "trigger the provider", + }; +} + +describe("resolved sub-agent provider failures", () => { + test("runSubAgent rejects a raw director reply after inference.error", async () => { + const { caught, observed } = await withResolvedProviderRun(async (run, cwd, observed) => { + try { + await run(runParams(cwd)); + } catch (error) { + return { caught: error, observed }; + } + throw new Error("expected runSubAgent to reject"); + }); + + expect(isResolvedProviderFailureError(caught)).toBe(true); + expect((caught as ResolvedProviderFailureError).message).toBe(SAFE_MESSAGE); + expect((caught as ResolvedProviderFailureError).diagnosticMessage).toBe(RAW_DIAGNOSTIC); + expect(observed.some((event) => JSON.stringify(event).includes(RAW_DIAGNOSTIC))).toBe(true); + }); + + test("split spawn_agent and wait_agents return only the safe message", async () => { + await withResolvedProviderRun(async (run, cwd) => { + const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetMailbox(sessions); + const deps = { + ...runParams(cwd), + getWorkdirBase: () => join(cwd, ".ctx"), + sessions, + fleetRecords, + run, + }; + const spawned = await callTool(createSpawnAgentTool(deps), "spawn_agent", { + description: "provider failure", + prompt: "trigger it", + intent: "explore", + }); + const spawnPayload = JSON.parse(String(spawned.content)) as { agent_id?: unknown }; + if (typeof spawnPayload.agent_id !== "string") throw new Error("missing agent_id"); + const waited = await callTool( + createWaitAgentsTool({ sessions, fleetRecords }), + "wait_agents", + { targets: [spawnPayload.agent_id], timeout_ms: 5000 }, + ); + const waitPayload = JSON.parse(String(waited.content)) as { + results?: { + agent_id?: string; + status?: string; + error?: string; + provider_failure?: boolean; + }[]; + }; + + expect(waitPayload.results?.[0]).toEqual({ + agent_id: spawnPayload.agent_id, + status: "failed", + error: SAFE_MESSAGE, + provider_failure: true, + }); + expect(String(waited.content)).not.toContain(RAW_DIAGNOSTIC); + expect(sessions.get(spawnPayload.agent_id)?.error).toBe(SAFE_MESSAGE); + }); + }); +}); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index cfc2e260..95e0e96b 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -76,6 +76,8 @@ import { consumeStream } from "../session/stream-consumer.js"; import { createCycleTextRecorder } from "../session/stream-journal.js"; import { onTurnBoundary } from "../agent/reactor-events.js"; import { refreshInferenceSourceBundle } from "./refresh-inference-source.js"; +import { createResolvedProviderFailureError } from "../inference-error-message.js"; +import { createRunEventSettlement } from "./run-event-settlement.js"; import type { CapabilityFilter } from "../agent/profiles.js"; import type { Settings } from "../config/settings.js"; @@ -875,6 +877,7 @@ async function runSubAgentInner( // salvage Findings keep substantive mid-run text, not only the final cycle. const TURN_PROSE_CAP = 12_000; let accumulatedProse = ""; + let terminalProviderDiagnostic: string | undefined; // Thrash paths from tool.start so mid-tool cancel still lists files touched. let thrashState = EMPTY_THRASH_STATE; const withTelemetry = (result: RunSubAgentResult): RunSubAgentResult => ({ @@ -885,7 +888,9 @@ async function runSubAgentInner( // cancel/deadline has the cycle's tail as its payload, even though no // turn boundary has completed yet to carry it. const cycleRecorder = createCycleTextRecorder(() => workdir); + const runSettlement = createRunEventSettlement(); const streamSink = (event: ReactorEmittedEvent): void => { + runSettlement.handleEvent(event); const name = subAgentToolName(event); if (name !== null) { toolNamesUsed.push(name); @@ -907,9 +912,16 @@ async function runSubAgentInner( } if (event.type === "inference.start") { settlementState.latestModel = event.data.model; + terminalProviderDiagnostic = undefined; } if (event.type === "inference.done") { settlementState.latestModel = event.data.source.model; + terminalProviderDiagnostic = undefined; + } + if (event.type === "inference.error") { + const message = (event.data as { error?: { message?: unknown } }).error?.message; + terminalProviderDiagnostic = + typeof message === "string" && message.length > 0 ? message : "inference error"; } if (onTurnBoundary(event)) { telemetryRollup.turn_count += 1; @@ -955,7 +967,22 @@ async function runSubAgentInner( params.onEvent?.(event); }; - streamPromise = consumeStream(agent.stream(), streamSink); + streamPromise = consumeStream(agent.stream(), streamSink).finally(runSettlement.endStream); + + const sendAndSettle = async ( + message: string, + options: { signal: AbortSignal }, + ): ReturnType["send"]> => { + const pending = runSettlement.beginSend(); + try { + const result = await agent!.send(message, options); + await pending.settled; + return result; + } catch (error) { + pending.cancel(); + throw error; + } + }; // Aborting the send signal only rejects the promise; the child reactor keeps // running until close() (same hard-stop rule as the parent in runner.ts). @@ -1018,7 +1045,13 @@ async function runSubAgentInner( // agent object, reusing full context rather than starting fresh. const followup = async (message: string): Promise => { interruptController = new AbortController(); - const result = await agent!.send(message, { signal: sendAbortSignal() }); + const result = await sendAndSettle(message, { signal: sendAbortSignal() }); + if (terminalProviderDiagnostic !== undefined) { + throw createResolvedProviderFailureError( + params.provider.providerName, + terminalProviderDiagnostic, + ); + } return result.reply.trim().length > 0 ? result.reply.trim() : "Sub-agent finished without a textual result."; @@ -1074,7 +1107,13 @@ async function runSubAgentInner( params.catalog, ); agent.setSources(fresh.sources, fresh.defaultSource); - const result = await agent.send(fullPrompt, sendOpts); + const result = await sendAndSettle(fullPrompt, sendOpts); + if (terminalProviderDiagnostic !== undefined) { + throw createResolvedProviderFailureError( + params.provider.providerName, + terminalProviderDiagnostic, + ); + } // A successful non-empty reply must not be clobbered by a late cancel that // races the completion window — keep the completed report. Empty replies // still honor abort so we salvage (or rethrow) rather than fabricating diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 641184d7..0cd082fc 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -16,13 +16,7 @@ import { loadRecentTurns, } from "../session/optimized-context-store.js"; import { type } from "arktype"; -import { - buildCodexSource, - buildOpenAISource, - buildXaiSource, - refreshLiveProviderCatalog, - type Config, -} from "../config/index.js"; +import { refreshLiveProviderCatalog, type Config } from "../config/index.js"; import { globalSettingsPath, loadLocalSettings, @@ -222,6 +216,11 @@ import { shouldSettleUiAfterSendFailure, } from "./session-chrome.js"; import { ingestOperatorPrompt } from "./prompt-attachments.js"; +import { + CREDENTIAL_FAILURE_USER_MESSAGE, + isResolvedProviderFailureError, + terminalProviderFailureMessage, +} from "../inference-error-message.js"; import { listPathSuggestions } from "./components/at-mention/list.js"; import { imageAttachmentFromPath, type PendingImageAttachment } from "./image-attachments.js"; import { appendSentMessage, loadSentMessages } from "../session/sent-messages.js"; @@ -602,6 +601,36 @@ export function setUpCommandRegistry( setHiddenCommands(settings?.hiddenCommands ?? []); } +export function surfaceTerminalProviderFailure( + shell: Parameters[0], + providerId: string, + displayLabel?: string, +): void { + surfaceSystemNotice(shell, terminalProviderFailureMessage(providerId, displayLabel)); +} + +export interface InferenceAttemptIdentity { + providerId: string; + displayLabel?: string; +} + +export function tuiSendFailureMessage( + error: unknown, + failureKind: "auth" | "error", + providerFailureObserved: boolean, + attempt: InferenceAttemptIdentity, +): string { + if (failureKind === "auth") { + return CREDENTIAL_FAILURE_USER_MESSAGE; + } + if (!providerFailureObserved && !isResolvedProviderFailureError(error)) { + return error instanceof Error ? error.message : String(error); + } + const providerId = isResolvedProviderFailureError(error) ? error.providerId : attempt.providerId; + const displayLabel = providerId === attempt.providerId ? attempt.displayLabel : undefined; + return terminalProviderFailureMessage(providerId, displayLabel); +} + export async function runTUI(initialConfig: Config): Promise { let config = initialConfig; const inferenceDeps = await createInferenceDependencies(); @@ -1476,61 +1505,21 @@ export async function runTUI(initialConfig: Config): Promise { // connect after startup are not callable until the agent is rebuilt. buildAgent // re-runs tool resolution against the (now-populated) dynamic runner and resumes // conversation from the same git-backed store, so a reload is transparent. - // When the session starts on a Codex profile, seed the agent with a Responses - // source (account id pulled from the resolved catalog entry, session id from - // the run) rather than the OpenAI-compatible one. - const initialCodexAccountId = config.providers.find( - (p) => p.name === config.providerName, - )?.codexAccountId; - const buildOpenAICompatibleInitialSource = (): InferenceSource => - buildOpenAISource({ - id: config.providerName, - baseURL: config.baseURL, - apiKey: config.apiKey, - model: config.model, - ...(config.reasoningEffort !== undefined - ? { reasoningEffort: config.reasoningEffort } - : {}), - }); const buildSessionSources = (): { sources: InferenceSource[]; defaultSource: string } => buildSessionSourcesFromConfig(config, sessionId); const initialBundle = buildSessionSources(); let liveSources = initialBundle.sources; let liveDefaultSource = initialBundle.defaultSource; + const selectedSource = liveSources[0]; + if (selectedSource === undefined) { + throw new Error("Selected inference source was not assembled"); + } // The source the next inference will use, tracked live so the compaction // summarizer always summarizes with the current model (model switches and // Codex token refreshes update it below). - let liveSource: InferenceSource = - liveSources.find((s) => s.id === liveDefaultSource) ?? - liveSources[0] ?? - buildInitialSourceFallback(); - - function buildInitialSourceFallback(): InferenceSource { - return initialCodexProfile !== undefined - ? buildCodexSource({ - id: config.providerName, - apiKey: config.apiKey, - model: config.model, - sessionId, - ...(initialCodexAccountId !== undefined ? { accountId: initialCodexAccountId } : {}), - ...(config.reasoningEffort !== undefined - ? { reasoningEffort: config.reasoningEffort } - : {}), - }) - : initialXaiProfile !== undefined - ? buildXaiSource({ - id: config.providerName, - apiKey: config.apiKey, - model: config.model, - sessionId, - ...(config.reasoningEffort !== undefined - ? { reasoningEffort: config.reasoningEffort } - : {}), - }) - : buildOpenAICompatibleInitialSource(); - } + let liveSource: InferenceSource = selectedSource; // Compaction summarizer: produces a structured, workflow-aware handoff via a // one-shot call on the live model, falling back to the deterministic summary @@ -1677,8 +1666,14 @@ export async function runTUI(initialConfig: Config): Promise { // keeps the in-flight cycle's text so an errored or interrupted turn leaves // its partial output in partial.jsonl instead of vanishing. const cycleRecorder = createCycleTextRecorder(() => workdir); + let providerFailureObserved = false; flushPartialOnCrash = () => cycleRecorder.dispose("crashed").then(() => undefined); const streamSink = (event: Parameters[0]): void => { + if (event.type === "inference.start" || event.type === "inference.done") { + providerFailureObserved = false; + } else if (event.type === "inference.error") { + providerFailureObserved = true; + } runSink.sink(event); cycleRecorder.handleEvent(event); if (onTurnBoundary(event)) { @@ -2152,15 +2147,33 @@ export async function runTUI(initialConfig: Config): Promise { approvalPersistNotice.notify = systemNotice; /** Settle the shell after a rejected send so the run does not look live. */ - const handleSendFailure = (err: unknown): void => { + const handleSendFailure = (err: unknown, attempt: InferenceAttemptIdentity): void => { const failure = classifyAgentSendFailure(err, sendAborted, isCodexAuthError, isXaiAuthError); captureAuthFailure(getTelemetry(), failure); if (!shouldSettleUiAfterSendFailure(failure.kind)) return; + if (failure.kind === "abort") return; recordRunError(err); - systemNotice(err instanceof Error ? err.message : String(err)); + systemNotice(tuiSendFailureMessage(err, failure.kind, providerFailureObserved, attempt)); setShellRunState(host.shell, "idle"); }; + const currentAttemptIdentity = (): InferenceAttemptIdentity => { + const displayLabel = config.settings?.providers[config.providerName]?.name; + return { + providerId: config.providerName, + ...(displayLabel !== undefined ? { displayLabel } : {}), + }; + }; + + const sendWithAttemptIdentity = async (message: InboundMessage): Promise => { + const attempt = currentAttemptIdentity(); + try { + await agentProxy.send(message); + } catch (error) { + handleSendFailure(error, attempt); + } + }; + // The permissions surface addresses grants by their position in the last // listing, so revoke resolves against the same snapshot the operator saw. let listedGrants: readonly ScopedApproval[] = []; @@ -2176,7 +2189,7 @@ export async function runTUI(initialConfig: Config): Promise { // A command the operator typed and submitted at the prompt — same // provenance as a plain-text send, just composed by the command // handler instead of typed verbatim. - void agentProxy.send(userInboundMessage(result.text, [])).catch(handleSendFailure); + void sendWithAttemptIdentity(userInboundMessage(result.text, [])); return; case "workflow": systemNotice(workflowController.start(result.name)); @@ -2226,7 +2239,7 @@ export async function runTUI(initialConfig: Config): Promise { imageAttachmentFromPath, pending, ); - await agentProxy.send(userInboundMessage(ingested.text, ingested.attachments)); + await sendWithAttemptIdentity(userInboundMessage(ingested.text, ingested.attachments)); }; const dispatchCommand = (name: string, args: string): void => { @@ -2256,7 +2269,9 @@ export async function runTUI(initialConfig: Config): Promise { const send = createSubmitHandler({ dispatchCommand: (name, args) => dispatchCommand(name, args), sendPrompt: (text, attachments) => { - void sendUserPrompt(text, attachments ?? []).catch(handleSendFailure); + void sendUserPrompt(text, attachments ?? []).catch((error: unknown) => { + handleSendFailure(error, currentAttemptIdentity()); + }); }, onPromptSubmitted: () => { if (telemetryFirstRun && liveTelemetryIntent) { @@ -2329,7 +2344,7 @@ export async function runTUI(initialConfig: Config): Promise { ingestOperatorPrompt(text, config.cwd, imageAttachmentFromPath, pending), send: (text, pending) => { sendAborted = false; - void agentProxy.send(userInboundMessage(text, pending)).catch(handleSendFailure); + void sendWithAttemptIdentity(userInboundMessage(text, pending)); }, recordSent: (text) => { if (text.trim().length === 0) return; @@ -2340,7 +2355,7 @@ export async function runTUI(initialConfig: Config): Promise { }); }, captureGeneration: deliveryGeneration.capture, - onFailure: handleSendFailure, + onFailure: (error) => handleSendFailure(error, currentAttemptIdentity()), }), parentCycleLive: () => host.bridge.parentCycleLive, deliverSteer: createLiveSteerDeliver({ @@ -2351,7 +2366,7 @@ export async function runTUI(initialConfig: Config): Promise { agentProxy.deliver(userInboundMessage(text, pending)); }, captureGeneration: deliveryGeneration.capture, - onFailure: handleSendFailure, + onFailure: (error) => handleSendFailure(error, currentAttemptIdentity()), }), }), // Consent by proceeding requires the disclosure to be on screen before the @@ -2456,7 +2471,10 @@ export async function runTUI(initialConfig: Config): Promise { permissionGate.setProviderIdentity(providerName, modelName); }, rebuildInference: (next) => { - host.bridge.setInferenceProviderId(next.providerName); + host.bridge.setInferenceProviderId( + next.providerName, + config.settings?.providers[next.providerName]?.name, + ); const bundle = buildSessionSources(); agentProxy.setSources(bundle.sources, bundle.defaultSource); }, @@ -2766,7 +2784,11 @@ export async function runTUI(initialConfig: Config): Promise { // Harness inference.error events omit providerId; stamp the live catalog id // onto the stream map so transcript copy can identify known-xAI short 429s. - stampProvider.fn = (id) => host.bridge.setInferenceProviderId(id); + stampProvider.fn = (id) => + host.bridge.setInferenceProviderId( + id, + id === undefined ? undefined : config.settings?.providers[id]?.name, + ); stampProvider.fn(config.providerName); setMentionSuggestionSource(host.shell, (prefix) => listPathSuggestions(prefix, config.cwd)); @@ -2844,7 +2866,7 @@ export async function runTUI(initialConfig: Config): Promise { if (!resumeSkipInitialTask && config.task.trim().length > 0) { // The operator's initial task, typed as a CLI argument before launch — // same provenance as a prompt submit. - void agentProxy.send(userInboundMessage(config.task.trim(), [])).catch(handleSendFailure); + void sendWithAttemptIdentity(userInboundMessage(config.task.trim(), [])); } // Hydrate a resumed session's transcript after first paint. Reading history and diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index 1b9cb3c6..c86cdcca 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -723,11 +723,7 @@ describe("attachSessionBridge", () => { }); describe("failed sends", () => { - const errorRows = (shell: { - streamLog: readonly { role: string; meta?: string; text: string }[]; - }) => shell.streamLog.filter((r) => r.meta === "error").map((r) => r.text); - - test("a recognised auth expiry says what to press; anything else keeps its message", async () => { + test("a resolved terminal provider failure replaces the raw reply and resets", async () => { await withTestRenderer( async (h) => { const shell = createAppShell(h.renderer, { @@ -736,24 +732,24 @@ describe("failed sends", () => { run: "idle", }); const bridge = attachSessionBridge(shell, createRecordingPort()); + const rawDiagnostic = "upstream 401: secret response body"; + const normalReply = "The next request worked."; try { + bridge.setInferenceProviderId("codex/default", "Codex"); + bridge.handle({ type: "inference.start", data: { model: "gpt" } }); bridge.handle({ type: "inference.error", - data: { - error: { - message: 'Codex profile "default" is not authorized. Log in again.', - }, - }, - }); - bridge.handle({ - type: "inference.error", - data: { error: { message: "socket hang up" } }, + data: { error: { category: "credential_failure", message: rawDiagnostic } }, }); - - const rows = errorRows(shell); - expect(rows[0]).toContain("sign-in expired"); - expect(rows[0]).toContain("/model"); - expect(rows[1]).toBe("socket hang up"); + bridge.handle({ type: "connector.reply", data: { content: rawDiagnostic } }); + bridge.handle({ type: "inference.start", data: { model: "gpt" } }); + bridge.handle({ type: "connector.reply", data: { content: normalReply } }); + + const safeMessage = + 'Codex Provider failed. Try again or switch with "/model" and select another.'; + expect(shell.streamLog.filter((row) => row.text === safeMessage)).toHaveLength(1); + expect(shell.streamLog.filter((row) => row.text === normalReply)).toHaveLength(1); + expect(shell.streamLog.map((row) => row.text).join("\n")).not.toContain(rawDiagnostic); } finally { bridge.dispose(); shell.dispose(); @@ -814,7 +810,7 @@ describe("committed inference retry", () => { }); }); -describe("same-turn failover after inference.error", () => { +describe("same-turn retry after inference.error", () => { const errorRows = (shell: { streamLog: readonly { role: string; meta?: string; text: string }[]; }) => shell.streamLog.filter((r) => r.meta === "error").map((r) => r.text); @@ -934,7 +930,7 @@ describe("same-turn failover after inference.error", () => { expect(shell.streamLog.filter((r) => r.role === "user").map((r) => r.text)).toEqual([ "retry this", ]); - // Same-turn failover, not an operator stop — recovery must not borrow interrupt. + // Same-turn retry, not an operator stop — recovery must not borrow interrupt. expect(port.calls.some((c) => c.op === "interrupt")).toBe(false); expect(shell.streamLog.some((r) => r.meta === "stop")).toBe(false); } finally { @@ -1024,7 +1020,7 @@ describe("same-turn failover after inference.error", () => { const text = shell.streamLog.map((r) => r.text).join("\n"); expect(text).toContain("next prompt"); - expect(errorRows(shell)).toContain("Authentication failed — log in again."); + expect(errorRows(shell)).toEqual([]); } finally { bridge.dispose(); shell.dispose(); @@ -1034,7 +1030,7 @@ describe("same-turn failover after inference.error", () => { ); }); - test("a queued steer row survives failover rollback", async () => { + test("a queued steer row survives retry rollback", async () => { await withTestRenderer( async (h) => { const shell = createAppShell(h.renderer, { @@ -1096,7 +1092,7 @@ describe("same-turn failover after inference.error", () => { const text = shell.streamLog.map((r) => r.text).join("\n"); expect(text).toContain("restart from here"); expect(text).toContain("stop — restarting from your message"); - expect(errorRows(shell)).toContain("Authentication failed — log in again."); + expect(errorRows(shell)).toEqual([]); } finally { bridge.dispose(); shell.dispose(); @@ -1106,7 +1102,7 @@ describe("same-turn failover after inference.error", () => { ); }); - test("a terminal inference.error with no recovery still surfaces", async () => { + test("reactor.error still surfaces after a terminal inference.error", async () => { await withTestRenderer( async (h) => { const shell = createAppShell(h.renderer, { @@ -1129,7 +1125,7 @@ describe("same-turn failover after inference.error", () => { bridge.handle(event); } - expect(errorRows(shell)).toContain("Authentication failed — log in again."); + expect(errorRows(shell)).toEqual(["failed"]); } finally { bridge.dispose(); shell.dispose(); diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 14baed6a..824aa56a 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -214,7 +214,7 @@ export interface SessionBridge { * `inference.error` transcript lines can identify known-xAI short 429s * when the harness event omits `providerId`. */ - setInferenceProviderId: (id: string | undefined) => void; + setInferenceProviderId: (id: string | undefined, displayLabel?: string) => void; } const NOOP_PORT: SessionPort = { @@ -1407,12 +1407,18 @@ export function attachSessionBridge( bag.agentSessions = sessions; syncAgentProgress(shell, bag, sessions, now()); }, - setInferenceProviderId: (id) => { + setInferenceProviderId: (id, displayLabel) => { if (bag.disposed) return; if (id === undefined) { delete bag.mapCtx.providerId; + delete bag.mapCtx.providerLabel; } else { bag.mapCtx.providerId = id; + if (displayLabel === undefined) { + delete bag.mapCtx.providerLabel; + } else { + bag.mapCtx.providerLabel = displayLabel; + } } }, dispose: () => { diff --git a/src/tui/stream-event-map.test.ts b/src/tui/stream-event-map.test.ts index 54ab6877..0d6c34be 100644 --- a/src/tui/stream-event-map.test.ts +++ b/src/tui/stream-event-map.test.ts @@ -258,7 +258,7 @@ describe("inference.retry", () => { expect(actions(out)).toEqual(["mark", "rollback"]); }); - test("a same-turn failover start consumes the boundary handed off by inference.error", () => { + test("a same-turn retry start consumes the boundary handed off by inference.error", () => { const out = mapProductionSequence([ { type: "inference.start" }, { type: "inference.text.delta", data: { token: "partial" } }, @@ -273,7 +273,7 @@ describe("inference.retry", () => { expect(actions(out)).toEqual(["mark", "rollback", "mark"]); }); - test("a same-turn failover start after credential_failure also rolls back", () => { + test("a same-turn retry start after credential_failure also rolls back", () => { const out = mapProductionSequence([ { type: "inference.start" }, { @@ -334,106 +334,14 @@ describe("inference.retry", () => { }); describe("inference.error text", () => { - const message = (error: unknown) => { - const [event] = mapProductionEvent({ type: "inference.error", data: { error } }); - return event?.type === "error" ? event.message : undefined; - }; - - test("a classified failure gets its written line, not the provider body", () => { - expect(message({ category: "credential_failure", message: '{"error":{"code":401}}' })).toBe( - "Authentication failed — log in again.", - ); - expect(message({ category: "quota_exhausted", message: "429" })).toBe( - "Quota exhausted — usage limit reached.", - ); - }); - - test("a context overflow mislabeled as quota is read from the message", () => { - expect( - message({ category: "quota_exhausted", message: "input is too long for this model" }), - ).toContain("Context window full"); - }); - - test("Codex usage_limit_reached raw body surfaces reset ETA and profile switch", () => { - const line = message({ - category: "quota_exhausted", - message: "Too Many Requests", - statusCode: 429, - providerId: "codex/abk-labs", - raw: { - detail: { - error: { - code: "usage_limit_reached", - message: "You have reached your usage limit.", - plan_type: "workspace_member", - resets_in_seconds: 3435, - }, - }, - }, + test("provider diagnostics are not mapped to user-visible rows", () => { + const rawDiagnostic = "upstream 401: secret response body"; + const out = mapProductionEvent({ + type: "inference.error", + data: { error: { category: "credential_failure", message: rawDiagnostic } }, }); - expect(line).toContain('Codex profile "abk-labs"'); - expect(line).toMatch(/Resets in ~/); - expect(line).toContain("/model"); - }); - - test("ctx.providerId xAI + bare quota_exhausted 429 shows rate-limit copy", () => { - const ctx = createStreamMapContext({ providerId: "xai/thegreataxios" }); - const [event] = mapProductionEvent( - { - type: "inference.error", - data: { - error: { - category: "quota_exhausted", - message: "Too Many Requests", - statusCode: 429, - raw: { error: { message: "Too Many Requests" } }, - }, - }, - }, - ctx, - ); - expect(event?.type).toBe("error"); - if (event?.type !== "error") return; - expect(event.message.toLowerCase()).toMatch(/rate limit/); - expect(event.message).not.toContain("Quota exhausted"); - }); - - test("ctx.providerId Codex + ChatGPT usage-limit 429 shows rate-limit copy", () => { - const ctx = createStreamMapContext({ providerId: "codex/abk-labs" }); - const [event] = mapProductionEvent( - { - type: "inference.error", - data: { - error: { - category: "quota_exhausted", - message: "You have hit your ChatGPT usage limit", - statusCode: 429, - raw: "You have hit your ChatGPT usage limit", - }, - }, - }, - ctx, - ); - expect(event?.type).toBe("error"); - if (event?.type !== "error") return; - expect(event.message.toLowerCase()).toMatch(/rate limit/); - expect(event.message).not.toContain("Quota exhausted"); - expect(event.message.toLowerCase()).not.toContain("usage limit reached"); - }); - - test("bare quota_exhausted 429 without ctx/provider still shows Quota exhausted", () => { - expect( - message({ - category: "quota_exhausted", - message: "Too Many Requests", - statusCode: 429, - raw: { error: { message: "Too Many Requests" } }, - }), - ).toBe("Quota exhausted — usage limit reached."); - }); - test("an unclassified failure keeps the provider's own words", () => { - expect(message({ message: "socket hang up" })).toBe("socket hang up"); - expect(message({ category: "wat", message: "socket hang up" })).toBe("socket hang up"); + expect(out).toEqual([]); + expect(JSON.stringify(out)).not.toContain(rawDiagnostic); }); }); diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index 9ff63e9c..1773de08 100644 --- a/src/tui/stream-event-map.ts +++ b/src/tui/stream-event-map.ts @@ -9,7 +9,7 @@ import { splitPendingControlTail, stripTerminalControlSequences, } from "../util/control-char-strip.js"; -import { inferenceErrorMessage } from "../inference-error-message.js"; +import { terminalProviderFailureMessage } from "../inference-error-message.js"; import type { RunState } from "./session-queue.js"; /** Canonical inbound events the bridge understands (fixtures + mapped reactor). */ @@ -109,10 +109,9 @@ export interface StreamMapContext { attemptCallIds: Set; /** * A committed attempt can also end in `inference.error` with no - * `inference.done`. Recovery follows that error as either the reactor's - * committed-retry or a same-turn failover `inference.start`. The boundary - * must not stay armed across a terminal error, so the error hands it off - * here: the very next event either consumes it (retry or start) and + * `inference.done`. A same-provider retry follows that error with another + * `inference.start`. The boundary must not stay armed across a terminal + * error, so the error hands it off here: the very next event consumes it and * retracts the failed attempt, or expires it and keeps the error row. */ errorRollbackArmed: boolean; @@ -122,9 +121,15 @@ export interface StreamMapContext { * transcript formatting can reuse known-provider remappers. */ providerId?: string; + providerLabel?: string; + /** Raw diagnostics stay on the reactor event; only this marker reaches reply mapping. */ + pendingProviderFailure: boolean; } -export function createStreamMapContext(opts?: { providerId?: string }): StreamMapContext { +export function createStreamMapContext(opts?: { + providerId?: string; + providerLabel?: string; +}): StreamMapContext { return { callIdToName: new Map(), callIdToArgs: new Map(), @@ -134,7 +139,9 @@ export function createStreamMapContext(opts?: { providerId?: string }): StreamMa attemptArmed: false, attemptCallIds: new Set(), errorRollbackArmed: false, + pendingProviderFailure: false, ...(opts?.providerId !== undefined ? { providerId: opts.providerId } : {}), + ...(opts?.providerLabel !== undefined ? { providerLabel: opts.providerLabel } : {}), }; } @@ -294,8 +301,8 @@ export function mapProductionEvent( flushed.push(...flushDelta(ctx, "thinking")); } // The error handoff only survives to the very next event; consume it here so - // anything other than the retry or failover start it was meant for expires - // the boundary and keeps the error row. + // anything other than the retry start it was meant for expires the boundary + // and keeps the error row. const handoff = ctx?.errorRollbackArmed === true; if (ctx) ctx.errorRollbackArmed = false; const expired = handoff && !recoversErrorHandoff(event.type) ? [ATTEMPT_CLEAR] : []; @@ -337,6 +344,7 @@ function mapEvent( ctx.hadTextDelta = false; ctx.attemptArmed = true; ctx.attemptCallIds = new Set(ctx.callIdToName.keys()); + ctx.pendingProviderFailure = false; } return [ ...(recovered ? [ATTEMPT_ROLLBACK] : []), @@ -348,6 +356,7 @@ function mapEvent( case "inference.done": // Cycle settled: disarm so a pre-commit retry belonging to the *next* // cycle cannot retract this one's rows. + if (ctx) ctx.pendingProviderFailure = false; return disarmAttempt(ctx); case "inference.retry": { @@ -450,6 +459,16 @@ function mapEvent( case "connector.reply": { const content = typeof data.content === "string" ? data.content : ""; + if (ctx?.pendingProviderFailure === true) { + ctx.pendingProviderFailure = false; + ctx.hadTextDelta = false; + return [ + { + type: "assistant", + text: terminalProviderFailureMessage(ctx.providerId ?? "Unknown", ctx.providerLabel), + }, + ]; + } if (ctx?.hadTextDelta) { ctx.hadTextDelta = false; // Text already painted via assistant.delta; the reply would repeat it. @@ -460,7 +479,10 @@ function mapEvent( } case "reactor.done": - if (ctx) ctx.hadTextDelta = false; + if (ctx) { + ctx.hadTextDelta = false; + ctx.pendingProviderFailure = false; + } return [...disarmAttempt(ctx), { type: "run", state: "idle" }, { type: "tool.boundary" }]; case "reactor.error": { @@ -474,40 +496,16 @@ function mapEvent( } case "inference.error": { - const err = asRecord(data.error); - const rawMessage = - typeof err?.message === "string" - ? err.message - : typeof data.error === "string" - ? data.error - : "inference error"; - // A classified failure gets the line written for it; anything unclassified - // keeps the provider's own words rather than a generic stand-in. - const providerId = - typeof err?.providerId === "string" - ? err.providerId - : typeof ctx?.providerId === "string" - ? ctx.providerId - : undefined; - const message = - typeof err?.category === "string" - ? inferenceErrorMessage({ - category: err.category, - message: rawMessage, - ...(typeof err.statusCode === "number" ? { statusCode: err.statusCode } : {}), - ...(err.raw !== undefined ? { raw: err.raw } : {}), - ...(providerId !== undefined ? { providerId } : {}), - ...(typeof err.retryAfterMs === "number" ? { retryAfterMs: err.retryAfterMs } : {}), - }) - : rawMessage; // Hand the armed boundary to the next event rather than disarming: a - // committed retry or same-turn failover start must still retract the - // failed attempt, including the error row painted here. + // committed retry start must still retract the failed attempt. Terminal + // failures are surfaced once by the runner after agent.send rejects; + // provider diagnostics remain in the event stream for observability only. if (ctx?.attemptArmed === true) { ctx.attemptArmed = false; ctx.errorRollbackArmed = true; } - return [{ type: "error", message }]; + if (ctx) ctx.pendingProviderFailure = true; + return []; } default: diff --git a/tests/unit/exec/runner.test.ts b/tests/unit/exec/runner.test.ts index a2d9fe28..f6250c4b 100644 --- a/tests/unit/exec/runner.test.ts +++ b/tests/unit/exec/runner.test.ts @@ -2,7 +2,9 @@ import { describe, expect, test } from "bun:test"; import type { Config } from "../../../src/config/index.js"; import { disposeExecRuntime, + execUserFailureMessage, formatCaughtError, + refreshSelectedProviderCredential, resolveExecDirectorOverlay, runExec, } from "../../../src/exec/runner.js"; @@ -35,6 +37,33 @@ describe("formatCaughtError", () => { }); }); +describe("selected provider refresh failures", () => { + test("a non-provider failure remains distinct after inference has run", () => { + expect(execUserFailureMessage(bareConfig("hello"), new Error("disk full"), false)).toBe( + "disk full", + ); + }); + + test("pre-inference OAuth failure keeps diagnostics internal and returns safe copy", async () => { + const config = { + ...bareConfig("hello"), + providerName: "codex/work", + settings: { providers: { "codex/work": { name: "Codex" } } }, + } as unknown as Config; + const rawDiagnostic = '401 {"error":"refresh token rejected"}'; + + try { + await refreshSelectedProviderCredential(() => Promise.reject(new Error(rawDiagnostic))); + throw new Error("expected refresh to fail"); + } catch (err) { + expect(formatCaughtError(err)).toBe(rawDiagnostic); + const userMessage = execUserFailureMessage(config, err, false); + expect(userMessage).toBe("Authentication failed — log in again."); + expect(userMessage).not.toContain(rawDiagnostic); + } + }); +}); + describe("runExec", () => { test("empty prompt exits 2 with stderr message without bootstrapping", async () => { const stderrChunks: string[] = []; diff --git a/tests/unit/inference-sources.test.ts b/tests/unit/inference-sources.test.ts index 40a44afd..6276c6d7 100644 --- a/tests/unit/inference-sources.test.ts +++ b/tests/unit/inference-sources.test.ts @@ -119,7 +119,7 @@ test("buildInferenceSourceForRef forwards reasoning effort on xAI sources", () = expect(unset?.defaults?.providerOptions).not.toHaveProperty("reasoning_effort"); }); -test("buildMainSessionSources backs the active head with other configured providers", () => { +test("buildMainSessionSources includes only the selected provider and model", () => { const settings: Settings = { providers: { openai: { @@ -137,127 +137,12 @@ test("buildMainSessionSources backs the active head with other configured provid activeModel: "gpt-4o", sessionId: "sess", }); - expect(bundle.sources.length).toBeGreaterThanOrEqual(2); + expect(bundle.sources).toHaveLength(1); + expect(bundle.sources[0]).toMatchObject({ id: "openai", model: "gpt-4o" }); expect(bundle.defaultSource).toBe("openai"); }); -test("ollama backup with leftover extra path does not throw when OpenAI is active", () => { - const leftoverURL = "http://localhost:11434/api/tags"; - const settings: Settings = { - providers: { - openai: { - baseURL: "https://api.openai.com/v1", - apiKey: "k", - models: ["gpt-4o"], - }, - ollama: { baseURL: leftoverURL, keyless: true, models: ["llama3"] }, - }, - }; - const mixedCatalog: ProviderCatalogEntry[] = [ - { - name: "openai", - baseURL: "https://api.openai.com/v1", - apiKey: "k", - models: ["gpt-4o"], - defaultModel: "gpt-4o", - }, - { - name: "ollama", - baseURL: leftoverURL, - keyless: true, - models: ["llama3"], - defaultModel: "llama3", - }, - ]; - const bundle = buildMainSessionSources({ - settings, - catalog: mixedCatalog, - activeProvider: "openai", - activeModel: "gpt-4o", - sessionId: "sess", - }); - expect(bundle.defaultSource).toBe("openai"); - expect(bundle.sources.map((s) => s.id)).toEqual(["openai"]); -}); - -test("active ollama with leftover extra path still fails that provider", () => { - const leftoverURL = "http://localhost:11434/api/tags"; - const settings: Settings = { - providers: { - ollama: { baseURL: leftoverURL, keyless: true, models: ["llama3"] }, - openai: { - baseURL: "https://api.openai.com/v1", - apiKey: "k", - models: ["gpt-4o"], - }, - }, - }; - const mixedCatalog: ProviderCatalogEntry[] = [ - { - name: "ollama", - baseURL: leftoverURL, - keyless: true, - models: ["llama3"], - defaultModel: "llama3", - }, - { - name: "openai", - baseURL: "https://api.openai.com/v1", - apiKey: "k", - models: ["gpt-4o"], - defaultModel: "gpt-4o", - }, - ]; - expect(() => - buildMainSessionSources({ - settings, - catalog: mixedCatalog, - activeProvider: "ollama", - activeModel: "llama3", - sessionId: "sess", - }), - ).toThrow('No inference source for provider "ollama"'); -}); - -test("legacy ollama /v1 backup does not throw when OpenAI is active", () => { - const settings: Settings = { - providers: { - openai: { - baseURL: "https://api.openai.com/v1", - apiKey: "k", - models: ["gpt-4o"], - }, - ollama: { baseURL: "http://localhost:11434/v1", keyless: true, models: ["llama3"] }, - }, - }; - const mixedCatalog: ProviderCatalogEntry[] = [ - { - name: "openai", - baseURL: "https://api.openai.com/v1", - apiKey: "k", - models: ["gpt-4o"], - defaultModel: "gpt-4o", - }, - { - name: "ollama", - baseURL: "http://localhost:11434/v1", - keyless: true, - models: ["llama3"], - defaultModel: "llama3", - }, - ]; - const bundle = buildMainSessionSources({ - settings, - catalog: mixedCatalog, - activeProvider: "openai", - activeModel: "gpt-4o", - sessionId: "sess", - }); - expect(bundle.defaultSource).toBe("openai"); - expect(bundle.sources.find((s) => s.id === "ollama")?.baseURL).toBe("http://localhost:11434/v1"); -}); - -test("buildSubagentSources backs the head with other configured providers", () => { +test("buildSubagentSources includes only the selected provider and model", () => { const settings: Settings = { providers: { openai: { baseURL: "https://api.openai.com/v1", apiKey: "k", models: ["gpt-4o"] }, @@ -270,7 +155,8 @@ test("buildSubagentSources backs the head with other configured providers", () = head: { provider: "openai", model: "gpt-4o" }, sessionId: "sub", }); - expect(bundle.sources.length).toBeGreaterThanOrEqual(2); + expect(bundle.sources).toHaveLength(1); + expect(bundle.sources[0]).toMatchObject({ id: "openai", model: "gpt-4o" }); expect(bundle.defaultSource).toBe("openai"); }); diff --git a/tests/unit/tui/runner.test.ts b/tests/unit/tui/runner.test.ts index 8114531e..667558e8 100644 --- a/tests/unit/tui/runner.test.ts +++ b/tests/unit/tui/runner.test.ts @@ -8,6 +8,7 @@ import { getTUIRunSummaryStatus, loadLocalSettingsWriteBase, resumeTranscriptLoadErrorBlock, + tuiSendFailureMessage, } from "../../../src/tui/runner.js"; import { createSessionOperationQueue } from "../../../src/tui/session-operation-queue.js"; import { createRunSink } from "../../../src/session/run-sink.js"; @@ -39,6 +40,33 @@ test("resumeTranscriptLoadErrorBlock surfaces a user-visible error block", () => expect(resumeTranscriptLoadErrorBlock("disk full").message).toContain("disk full"); }); +test("TUI send failures keep non-provider errors distinct", () => { + expect( + tuiSendFailureMessage(new Error("disk full"), "error", false, { + providerId: "codex/work", + displayLabel: "Codex", + }), + ).toBe("disk full"); +}); + +test("TUI send failures retain the in-flight provider identity across model switches", () => { + expect( + tuiSendFailureMessage(new Error("raw provider body"), "error", true, { + providerId: "codex/work", + displayLabel: "Codex", + }), + ).toBe('Codex Provider failed. Try again or switch with "/model" and select another.'); +}); + +test("TUI auth failures tell the user to log in again instead of switching models", () => { + expect( + tuiSendFailureMessage(new Error("401 refresh token rejected"), "auth", false, { + providerId: "codex/work", + displayLabel: "Codex", + }), + ).toBe("Authentication failed — log in again."); +}); + test("loadLocalSettingsWriteBase distinguishes absent from unreadable", async () => { // Absent → empty base (safe to write a single key). expect(await loadLocalSettingsWriteBase("/nope", async () => null)).toEqual({});