Harden provider failure handling

Author: TheGreatAxios

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,
@@ -63,7 +58,10 @@ 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 { terminalProviderFailureMessage } from "../inference-error-message.js";
+import {
+  isResolvedProviderFailureError,
+  terminalProviderFailureMessage,
+} from "../inference-error-message.js";
 import { collectToolPlugins, resolveToolPlugins } from "../plugins/tool-plugins.js";
 import {
   expandExistingPluginMembers,
@@ -132,9 +130,13 @@ export async function refreshSelectedProviderCredential<T>(refresh: () => Promis
 export function execUserFailureMessage(
   config: Config,
   err: unknown,
-  inferenceStarted: boolean,
+  providerFailureObserved: boolean,
 ): string {
-  if (inferenceStarted || (err instanceof Error && err.name === SELECTED_PROVIDER_FAILURE)) {
+  if (
+    providerFailureObserved ||
+    isResolvedProviderFailureError(err) ||
+    (err instanceof Error && err.name === SELECTED_PROVIDER_FAILURE)
+  ) {
     return terminalProviderFailureMessage(
       config.providerName,
       config.settings?.providers[config.providerName]?.name,
@@ -279,7 +281,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
   let finalized = false;
   let turnsUsed = 0;
   let runSink: RunSink | null = null;
-  let inferenceStarted = false;
+  let providerFailureObserved = false;
 
   const persist = async (
     status: "running" | "done" | "failed" | "cancelled",
@@ -568,56 +570,14 @@ export async function runExec(config: Config): Promise<ExecResult> {
 
     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
@@ -748,6 +708,11 @@ export async function runExec(config: Config): Promise<ExecResult> {
     // 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") {
@@ -771,7 +736,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
     let runError: string | undefined;
     let sinkStatus: ReturnType<typeof liveSink.getStatus> = "cancelled";
     try {
-      inferenceStarted = true;
       // Final OAuth refresh immediately before send (token may have aged during MCP).
       if (initialCodexProfile !== undefined) {
         const { access } = await getValidCodexToken(initialCodexProfile);
@@ -912,7 +876,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
   } catch (err) {
     const diagnosticMessage = formatCaughtError(err);
     logger.error("exec failed: {error}", { error: diagnosticMessage });
-    const userMessage = execUserFailureMessage(config, err, inferenceStarted);
+    const userMessage = execUserFailureMessage(config, err, providerFailureObserved);
     stderr.write(`Error: ${userMessage}\n`);
     await persist("failed", { error: diagnosticMessage });
     return {

src/inference-error-message.test.ts

@@ -106,6 +106,12 @@ describe("terminalProviderFailureMessage", () => {
     );
   });
 
+  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(

src/inference-error-message.ts

@@ -95,12 +95,13 @@ function codexUsageLimitLine(error: InferenceErrorLike): string | 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";
+  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;
 };
 
@@ -111,6 +112,7 @@ export function createResolvedProviderFailureError(
 ): ResolvedProviderFailureError {
   return Object.assign(new Error(terminalProviderFailureMessage(providerId, displayLabel)), {
     name: "ResolvedProviderFailureError" as const,
+    providerId,
     diagnosticMessage,
   });
 }
@@ -121,6 +123,8 @@ export function isResolvedProviderFailureError(
   return (
     error instanceof Error &&
     error.name === "ResolvedProviderFailureError" &&
+    "providerId" in error &&
+    typeof error.providerId === "string" &&
     "diagnosticMessage" in error &&
     typeof error.diagnosticMessage === "string"
   );

src/session/run-sink.test.ts

@@ -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,
     });
   });

src/subagent/agent-fleet.ts

@@ -94,6 +94,7 @@ interface FleetRecord {
   status: "running" | "done" | "failed" | "interrupted";
   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. */
@@ -141,10 +142,14 @@ class FleetRecords {
     this.notify();
   }
 
-  reject(id: string, error: string): void {
+  reject(id: string, error: string, providerFailure = false): void {
     const existing = this.records.get(id);
     if (existing !== undefined && existing.status !== "running") return;
-    this.records.set(id, { status: "failed", error });
+    this.records.set(id, {
+      status: "failed",
+      error,
+      ...(providerFailure ? { providerFailure: true } : {}),
+    });
     this.enforceCap();
     this.notify();
   }
@@ -586,7 +591,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);
       };
@@ -786,13 +797,18 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
             : err instanceof Error
               ? err.message
               : String(err);
-          const parentMessage = isResolvedProviderFailureError(err)
-            ? err.message
-            : classifySubAgentInferenceAuthFailure(err) !== null
-              ? terminalProviderFailureMessage(provider.providerName)
-              : diagnosticMessage;
-          deps.fleetRecords.reject(session.id, parentMessage);
-          deps.sessions.fail(session.id, diagnosticMessage);
+          const providerId = isResolvedProviderFailureError(err)
+            ? err.providerId
+            : provider.providerName;
+          const isProviderFailure =
+            providerFailureObserved ||
+            isResolvedProviderFailureError(err) ||
+            classifySubAgentInferenceAuthFailure(err) !== null;
+          const parentMessage = isProviderFailure
+            ? terminalProviderFailureMessage(providerId, settings?.providers[providerId]?.name)
+            : diagnosticMessage;
+          deps.fleetRecords.reject(session.id, parentMessage, isProviderFailure);
+          deps.sessions.fail(session.id, parentMessage);
         })
         .finally(() => {
           finalizeEnd();
@@ -919,6 +935,7 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool {
             status: taken.status,
             ...(taken.report !== undefined ? { report: taken.report } : {}),
             ...(taken.error !== undefined ? { error: taken.error } : {}),
+            ...(taken.providerFailure === true ? { provider_failure: true } : {}),
             ...(taken.hint !== undefined ? { hint: taken.hint } : {}),
           };
         }

src/subagent/index.test.ts

@@ -1160,7 +1160,9 @@ describe("createTaskTool", () => {
     expect(String(result.content)).not.toContain("401 Unauthorized");
     const row = sessions.list().find((s) => s.description === "auth probe");
     expect(row?.status).toBe("failed");
-    expect(row?.error).toBe("401 Unauthorized");
+    expect(row?.error).toBe(
+      'test-provider Provider failed. Try again or switch with "/model" and select another.',
+    );
   });
 
   test("terminal child provider failure hides raw diagnostics from the parent result", async () => {

src/subagent/run-event-settlement.ts

@@ -0,0 +1,52 @@
+import type { ReactorEmittedEvent } from "@intx/inference";
+
+export interface PendingRunSettlement {
+  readonly settled: Promise<void>;
+  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<void>((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();
+    },
+  };
+}