fix: classify provider runner transient failures

src/components/microservices/code-queue/src/code-agent/codex.ts

@@ -6,6 +6,7 @@ import * as readline from "node:readline";
 import type { AppServerExit, CodexEventSummary, CodexRunResult, JsonValue, QueueTask, RuntimeConfig, SessionCommandOutput, TerminalStatus } from "../types";
 import type { ActiveRun, CodeAgentClient } from "./common";
 import { extractRecord, extractString, terminalStatus, textInput, withCodeAgentGitConfigEnv } from "./common";
+import { classifyRunnerError, runnerErrorClassificationJson } from "../runner-error-classifier";
 
 export interface CodexPortContext {
   config: Pick<RuntimeConfig, "approvalPolicy" | "codexHome" | "defaultWorkdir" | "sandbox" | "sourceCodexConfig" | "turnNoActivityTimeoutMs">;
@@ -393,6 +394,17 @@ function handleNotification(task: QueueTask, message: Record<string, unknown>, t
   }
 }
 
+function classifyCodexRunnerFailure(task: QueueTask, result: Pick<CodexRunResult, "terminalStatus" | "terminalError" | "transportClosedBeforeTerminal" | "appServerExit">): JsonValue | null {
+  if (result.terminalStatus === "completed" && !result.transportClosedBeforeTerminal) return null;
+  const message = [
+    result.terminalError ?? "",
+    result.transportClosedBeforeTerminal ? "transport closed before terminal event" : "",
+    result.appServerExit.stderrTail,
+  ].filter((part) => part.trim().length > 0).join("\n");
+  if (message.trim().length === 0) return null;
+  return runnerErrorClassificationJson(classifyRunnerError(message, task.providerId));
+}
+
 export async function runCodexTurn(task: QueueTask, prompt: string): Promise<CodexRunResult> {
   const queueId = ctx().queueIdOf(task);
   const events: CodexEventSummary[] = [];
@@ -447,7 +459,7 @@ export async function runCodexTurn(task: QueueTask, prompt: string): Promise<Cod
     const closedBeforeTerminal = race === "closed" && !terminalSeen;
     if (terminalSeen) app.stop();
     const exit = await app.closedPromise;
-    return {
+    const result: CodexRunResult = {
       threadId,
       turnId,
       finalResponse: task.finalResponse,
@@ -457,12 +469,14 @@ export async function runCodexTurn(task: QueueTask, prompt: string): Promise<Cod
       appServerExit: exit,
       events,
     };
+    result.runnerErrorClassification = classifyCodexRunnerFailure(task, result);
+    return result;
   } catch (error) {
     const message = error instanceof Error ? error.message : String(error);
     ctx().appendOutput(task, "error", `${message}\n`, "app-server");
     app.stop();
     const exit = await app.closedPromise;
-    return {
+    const result: CodexRunResult = {
       threadId: task.codexThreadId,
       turnId: task.activeTurnId,
       finalResponse: task.finalResponse,
@@ -472,6 +486,8 @@ export async function runCodexTurn(task: QueueTask, prompt: string): Promise<Cod
       appServerExit: exit,
       events,
     };
+    result.runnerErrorClassification = classifyCodexRunnerFailure(task, result);
+    return result;
   } finally {
     if (ctx().activeRuns.get(queueId)?.app === app) ctx().activeRuns.delete(queueId);
     app.stop();

src/components/microservices/code-queue/src/code-agent/opencode.ts

@@ -7,6 +7,7 @@ import * as readline from "node:readline";
 import type { AppServerExit, CodexEventSummary, CodexRunResult, JsonValue, QueueTask, RuntimeConfig, TerminalStatus } from "../types";
 import type { ActiveRun, CodeAgentClient } from "./common";
 import { codeAgentGitConfigEntries, extractRecord, minimaxM27Model, normalizeCodeModel, stripAnsi, withCodeAgentGitConfigEnv } from "./common";
+import { classifyRunnerError, runnerErrorClassificationJson } from "../runner-error-classifier";
 
 export interface OpenCodePortContext {
   config: Pick<RuntimeConfig, "defaultWorkdir" | "minimaxApiBase" | "minimaxApiKey" | "minimaxModel" | "turnNoActivityTimeoutMs">;
@@ -339,6 +340,17 @@ function openCodeSessionMissing(result: CodexRunResult): boolean {
   return /Session not found/iu.test(stripAnsi(`${result.terminalError ?? ""}\n${result.appServerExit.stderrTail}`));
 }
 
+function classifyOpenCodeRunnerFailure(task: QueueTask, result: Pick<CodexRunResult, "terminalStatus" | "terminalError" | "transportClosedBeforeTerminal" | "appServerExit">): JsonValue | null {
+  if (result.terminalStatus === "completed" && !result.transportClosedBeforeTerminal) return null;
+  const message = [
+    result.terminalError ?? "",
+    result.transportClosedBeforeTerminal ? "transport closed before terminal event" : "",
+    result.appServerExit.stderrTail,
+  ].filter((part) => part.trim().length > 0).join("\n");
+  if (message.trim().length === 0) return null;
+  return runnerErrorClassificationJson(classifyRunnerError(message, task.providerId));
+}
+
 export async function runOpenCodeTurn(task: QueueTask, prompt: string): Promise<CodexRunResult> {
   const attemptedSessionId = task.codexThreadId;
   const first = await runOpenCodeTurnOnce(task, prompt);
@@ -365,6 +377,7 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise<Cod
       terminalError: message,
       transportClosedBeforeTerminal: true,
       appServerExit: { code: 1, signal: null, stderrTail: message },
+      runnerErrorClassification: runnerErrorClassificationJson(classifyRunnerError(message, task.providerId)),
       events: [],
     };
   }
@@ -402,7 +415,7 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise<Cod
       `opencode completed status=${status} exit=${exit.code ?? "null"} signal=${exit.signal ?? "null"}${stderr.length > 0 ? ` stderr=${ctx().safePreview(stderr, 1200)}` : ""}\n`,
       "opencode/complete",
     );
-    return {
+    const result: CodexRunResult = {
       threadId: app.sessionId ?? task.codexThreadId,
       turnId: app.runId,
       finalResponse,
@@ -412,13 +425,15 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise<Cod
       appServerExit: exit,
       events: app.events,
     };
+    result.runnerErrorClassification = classifyOpenCodeRunnerFailure(task, result);
+    return result;
   } catch (error) {
     clearInterval(activityWatchdog);
     const message = error instanceof Error ? error.message : String(error);
     ctx().appendOutput(task, "error", `${message}\n`, "opencode");
     app.stop();
     const exit = await app.closedPromise;
-    return {
+    const result: CodexRunResult = {
       threadId: app.sessionId ?? task.codexThreadId,
       turnId: app.runId,
       finalResponse: app.finalResponse.trim().length > 0 ? app.finalResponse : "",
@@ -428,6 +443,8 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise<Cod
       appServerExit: exit,
       events: app.events,
     };
+    result.runnerErrorClassification = classifyOpenCodeRunnerFailure(task, result);
+    return result;
   } finally {
     clearInterval(activityWatchdog);
     if (ctx().activeRuns.get(queueId)?.app === app) ctx().activeRuns.delete(queueId);

src/components/microservices/code-queue/src/index.ts

@@ -3476,6 +3476,7 @@ function attemptFromResult(task: QueueTask, mode: RunMode, startedAt: string, fi
     stderrTail: safePreview(result.appServerExit.stderrTail, 3000),
     outputStartSeq,
     outputEndSeq,
+    runnerErrorClassification: result.runnerErrorClassification ?? null,
   };
   if (inputPrompt !== null) setAttemptInputPrompt(attempt, inputPrompt);
   return attempt;

src/components/microservices/code-queue/src/runner-error-classifier.ts

@@ -0,0 +1,163 @@
+import type { JsonValue } from "./types";
+
+export type RunnerFailureScope =
+  | "runner-local"
+  | "provider-gateway"
+  | "registry"
+  | "k3s"
+  | "scheduler"
+  | "unknown";
+
+export type RunnerFailureDisposition =
+  | "runner-local-observation-gap"
+  | "retryable-transient"
+  | "service-degraded"
+  | "provider-degraded"
+  | "global-blocker";
+
+export interface RunnerErrorClassification {
+  scope: RunnerFailureScope;
+  disposition: RunnerFailureDisposition;
+  retryable: boolean;
+  globalBlocker: boolean;
+  singlePathObservation: boolean;
+  reason: string;
+  evidence: string[];
+  recommendedTriageCommand: string;
+  contract: {
+    singlePathFailureIsGlobalBlocker: false;
+    globalBlockerRequiresMultiSignalTriage: true;
+  };
+}
+
+function unique(items: string[]): string[] {
+  return Array.from(new Set(items.filter((item) => item.length > 0)));
+}
+
+function collectEvidence(text: string, patterns: RegExp[], limit = 4): string[] {
+  const evidence: string[] = [];
+  for (const pattern of patterns) {
+    for (const match of text.matchAll(pattern)) {
+      const raw = String(match[0] ?? "").trim().replace(/\s+/gu, " ");
+      if (raw.length > 0) evidence.push(raw);
+      if (evidence.length >= limit) return unique(evidence);
+    }
+  }
+  return unique(evidence);
+}
+
+function providerTriageCommand(providerId: string | null | undefined, message: string, scope: RunnerFailureScope): string {
+  const safeProviderId = /^[A-Za-z0-9_.-]{1,64}$/u.test(String(providerId ?? "")) ? String(providerId) : "<providerId>";
+  const compactMessage = message.replace(/\s+/gu, " ").trim().slice(0, 500).replace(/'/gu, "'\\''");
+  return `bun scripts/cli.ts provider triage ${safeProviderId} --observed-scope ${scope} --observed-error '${compactMessage}'`;
+}
+
+export function classifyRunnerError(message: string, providerId?: string | null): RunnerErrorClassification {
+  const raw = String(message || "");
+  const normalized = raw.toLowerCase();
+  const providerOfflineEvidence = collectEvidence(normalized, [
+    /provider is not online/gu,
+    /provider is offline/gu,
+    /provider went offline/gu,
+    /provider .*not online/gu,
+    /provider .*offline/gu,
+  ]);
+  const providerGatewayEvidence = collectEvidence(normalized, [
+    /provider[- ]gateway/gu,
+    /provider gateway/gu,
+    /provider-gateway-online/gu,
+    /websocket|ws\/provider/gu,
+    /provider .*disconnected/gu,
+    /provider .*heartbeat/gu,
+    /tunnel failed|http tunnel|egress proxy/gu,
+  ]);
+  const registryEvidence = collectEvidence(normalized, [
+    /artifact[- ]registry/gu,
+    /registry \/v2/gu,
+    /\bregistry\b/gu,
+    /manifest unknown|docker push|docker pull/gu,
+    /127\.0\.0\.1:5000/gu,
+  ]);
+  const k3sEvidence = collectEvidence(normalized, [
+    /\bk3s\b/gu,
+    /\bk3sctl\b/gu,
+    /kubectl/gu,
+    /kubernetes/gu,
+    /containerd/gu,
+    /pod .*failed|deployment .*unavailable/gu,
+  ]);
+  const schedulerEvidence = collectEvidence(normalized, [
+    /code[- ]queue scheduler/gu,
+    /\bscheduler\b/gu,
+    /active run/gu,
+    /runtime-preflight/gu,
+    /database claim/gu,
+    /postgres|postgresql/gu,
+    /app-server closed|opencode exited|codex .*failed/gu,
+  ]);
+  const runnerLocalEvidence = collectEvidence(normalized, [
+    /spawn .*enoent/gu,
+    /command not found/gu,
+    /no such file or directory/gu,
+    /permission denied/gu,
+    /app-server is already closed/gu,
+    /no .*activity .*stopping/gu,
+    /etimedout|timed out|timeout/gu,
+  ]);
+
+  let scope: RunnerFailureScope = "unknown";
+  let disposition: RunnerFailureDisposition = "retryable-transient";
+  let reason = "The runner failure needs provider triage before it can be treated as a global blocker.";
+  let evidence: string[] = [];
+
+  if (providerOfflineEvidence.length > 0) {
+    scope = "runner-local";
+    disposition = "runner-local-observation-gap";
+    reason = "A single runner-local provider offline observation is not independent evidence of a global D601/provider outage.";
+    evidence = providerOfflineEvidence;
+  } else if (registryEvidence.length > 0) {
+    scope = "registry";
+    disposition = "service-degraded";
+    reason = "The failure is registry-scoped; cross-check provider gateway, SSH, k3s and scheduler before escalating globally.";
+    evidence = registryEvidence;
+  } else if (k3sEvidence.length > 0) {
+    scope = "k3s";
+    disposition = "service-degraded";
+    reason = "The failure is k3s-scoped unless provider gateway and SSH cross-checks also fail.";
+    evidence = k3sEvidence;
+  } else if (providerGatewayEvidence.length > 0) {
+    scope = "provider-gateway";
+    disposition = "provider-degraded";
+    reason = "The failure points at provider-gateway/proxy transport and requires independent SSH/scheduler checks.";
+    evidence = providerGatewayEvidence;
+  } else if (schedulerEvidence.length > 0) {
+    scope = "scheduler";
+    disposition = "retryable-transient";
+    reason = "The failure is scheduler/runner scoped and should be retried or triaged with runtime preflight.";
+    evidence = schedulerEvidence;
+  } else if (runnerLocalEvidence.length > 0) {
+    scope = "runner-local";
+    disposition = "runner-local-observation-gap";
+    reason = "The failure was observed locally by the runner and is not enough to declare provider-global outage.";
+    evidence = runnerLocalEvidence;
+  }
+
+  return {
+    scope,
+    disposition,
+    retryable: true,
+    globalBlocker: false,
+    singlePathObservation: true,
+    reason,
+    evidence,
+    recommendedTriageCommand: providerTriageCommand(providerId, raw, scope),
+    contract: {
+      singlePathFailureIsGlobalBlocker: false,
+      globalBlockerRequiresMultiSignalTriage: true,
+    },
+  };
+}
+
+export function runnerErrorClassificationJson(classification: RunnerErrorClassification): JsonValue {
+  return classification as unknown as JsonValue;
+}

src/components/microservices/code-queue/src/types.ts

@@ -270,6 +270,7 @@ export interface AttemptSummary {
   stderrTail: string;
   outputStartSeq?: number | null;
   outputEndSeq?: number | null;
+  runnerErrorClassification?: JsonValue | null;
   errorCount?: number;
 }
 
@@ -489,6 +490,7 @@ export interface CodexRunResult {
   terminalError: string | null;
   transportClosedBeforeTerminal: boolean;
   appServerExit: AppServerExit;
+  runnerErrorClassification?: JsonValue | null;
   events: CodexEventSummary[];
 }