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,
@@ -67,7 +62,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,
@@ -136,9 +134,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,
@@ -324,7 +326,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",
@@ -614,56 +616,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
@@ -794,6 +754,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") {
@@ -817,7 +782,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);
@@ -958,7 +922,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

@@ -98,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. */
@@ -118,6 +119,7 @@ interface FleetOverlay {
   lastWaitStatus?: WaitJSONStatus;
   tombstoned?: boolean;
   hint?: string;
+  providerFailure?: true;
 }
 
 const RECOVERY_HINT =
@@ -170,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).
@@ -301,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 } : {}),
     };
   }
 
@@ -869,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);
       };
@@ -1084,10 +1099,14 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
             : err instanceof Error
               ? err.message
               : String(err);
+          const isProviderFailure = isResolvedProviderFailureError(err);
           const authMessage = formatSubAgentSpawnAuthFailureMessage(description, err);
           const failReason =
             authMessage ??
-            (isResolvedProviderFailureError(err) ? err.message : diagnosticMessage);
+            (isProviderFailure ? err.message : diagnosticMessage);
+          if (isProviderFailure || providerFailureObserved) {
+            deps.fleetRecords.markProviderFailure(session.id);
+          }
           deps.sessions.fail(session.id, failReason);
         })
         .finally(() => {
@@ -1232,6 +1251,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 } : {}),
         };
       });

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