From 71cada25f2f46f5e38282fcb03a298d2fe9d115b Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 21 May 2026 02:45:28 +0000 Subject: [PATCH] fix: classify provider runner transient failures --- .../provider-runner-triage-contract-test.ts | 69 ++++++++ scripts/src/check.ts | 5 + scripts/src/provider-triage.ts | 9 +- .../code-queue/src/code-agent/codex.ts | 20 ++- .../code-queue/src/code-agent/opencode.ts | 21 ++- .../microservices/code-queue/src/index.ts | 1 + .../code-queue/src/runner-error-classifier.ts | 163 ++++++++++++++++++ .../microservices/code-queue/src/types.ts | 2 + 8 files changed, 285 insertions(+), 5 deletions(-) create mode 100644 scripts/provider-runner-triage-contract-test.ts create mode 100644 src/components/microservices/code-queue/src/runner-error-classifier.ts diff --git a/scripts/provider-runner-triage-contract-test.ts b/scripts/provider-runner-triage-contract-test.ts new file mode 100644 index 00000000..0c44f973 --- /dev/null +++ b/scripts/provider-runner-triage-contract-test.ts @@ -0,0 +1,69 @@ +import { buildProviderTriageResult, type ProviderTriageSignal } from "./src/provider-triage"; +import { classifyRunnerError } from "../src/components/microservices/code-queue/src/runner-error-classifier"; + +type JsonRecord = Record; + +function assertCondition(condition: unknown, message: string, detail: unknown = {}): void { + if (!condition) throw new Error(`${message}: ${JSON.stringify(detail)}`); +} + +function signal( + id: string, + scope: ProviderTriageSignal["scope"], + status: ProviderTriageSignal["status"], + independentPath = true, +): ProviderTriageSignal { + return { + id, + scope, + status, + independentPath, + observedAt: "2026-05-21T00:00:00.000Z", + summary: `${id}:${scope}:${status}`, + }; +} + +function assertScope(message: string, expectedScope: string, expectedDisposition: string): JsonRecord { + const classification = classifyRunnerError(message, "D601") as unknown as JsonRecord; + assertCondition(classification.scope === expectedScope, `runner error should classify as ${expectedScope}`, classification); + assertCondition(classification.disposition === expectedDisposition, `runner error disposition should be ${expectedDisposition}`, classification); + assertCondition(classification.globalBlocker === false, "single runner error classification must not be global blocker", classification); + assertCondition(classification.retryable === true, "single runner error classification should remain retryable", classification); + assertCondition(String(classification.recommendedTriageCommand ?? "").includes("provider triage D601"), "classification should include triage command", classification); + return classification; +} + +export function runProviderRunnerTriageContract(): JsonRecord { + assertScope("provider is not online: D601", "runner-local", "runner-local-observation-gap"); + assertScope("provider-gateway http tunnel failed waiting for request", "provider-gateway", "provider-degraded"); + assertScope("artifact registry /v2 manifest HEAD failed on 127.0.0.1:5000", "registry", "service-degraded"); + assertScope("kubectl get pods failed: k3s api unavailable", "k3s", "service-degraded"); + assertScope("code-queue scheduler active run heartbeat is stale", "scheduler", "retryable-transient"); + assertScope("unexpected runner process exited with code 1", "unknown", "retryable-transient"); + + const singlePath = classifyRunnerError("provider is not online: D601", "D601"); + const result = buildProviderTriageResult("D601", [ + signal("observed-runner-error", singlePath.scope, "failed", false), + signal("backend-core-node", "provider-gateway", "ok"), + signal("host-ssh-probe", "ssh", "ok"), + signal("code-queue-health", "scheduler", "ok"), + ], "2026-05-21T00:00:00.000Z"); + + assertCondition(result.blockingDisposition === "runner-local-observation-gap", "single path provider offline must stay observation gap", result); + assertCondition(result.decision === "retryable-transient", "single path provider offline should be retryable transient", result); + assertCondition(result.retryable === true, "single path provider offline should be retryable", result); + assertCondition(result.contract.singlePathProviderOfflineIsGlobalBlocker === false, "triage contract should reject single-path global blocker", result); + + return { + ok: true, + checks: [ + "runner error classifier separates runner-local/provider-gateway/registry/k3s/scheduler/unknown", + "each single runner error classification has globalBlocker=false", + "provider triage keeps single provider is not online as retryable-transient, not global-blocker", + ], + }; +} + +if (import.meta.main) { + process.stdout.write(`${JSON.stringify(runProviderRunnerTriageContract(), null, 2)}\n`); +} diff --git a/scripts/src/check.ts b/scripts/src/check.ts index c054884a..b736178c 100644 --- a/scripts/src/check.ts +++ b/scripts/src/check.ts @@ -283,6 +283,9 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default fileItem("scripts/code-queue-trace-summary-contract-test.ts"), fileItem("scripts/code-queue-pr-preflight-contract-test.ts"), fileItem("scripts/code-queue-submit-routing-contract-test.ts"), + fileItem("scripts/provider-runner-triage-contract-test.ts"), + fileItem("scripts/src/provider-triage.ts"), + fileItem("src/components/microservices/code-queue/src/runner-error-classifier.ts"), fileItem("scripts/src/ci.ts"), fileItem("scripts/src/e2e.ts"), fileItem("scripts/deploy-artifact-matrix-contract-test.ts"), @@ -305,6 +308,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default items.push(commandItem("code-queue:trace-summary-contract", ["bun", "scripts/code-queue-trace-summary-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:pr-preflight-contract", ["bun", "scripts/code-queue-pr-preflight-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:submit-routing-contract", ["bun", "scripts/code-queue-submit-routing-contract-test.ts"], 30_000)); + items.push(commandItem("provider:runner-triage-contract", ["bun", "scripts/provider-runner-triage-contract-test.ts"], 30_000)); items.push(commandItem("deploy:artifact-matrix-contract", ["bun", "scripts/deploy-artifact-matrix-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:active-run-heartbeat-visible", ["bun", "scripts/code-queue-liveness-diagnostics-test.ts", "--only", "code-queue:active-run-heartbeat-visible"], 30_000)); items.push(commandItem("code-queue:trace-gap-not-stale", ["bun", "scripts/code-queue-liveness-diagnostics-test.ts", "--only", "code-queue:trace-gap-not-stale"], 30_000)); @@ -323,6 +327,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default items.push(skippedItem("code-queue:trace-summary-contract", "Code Queue trace summary contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:pr-preflight-contract", "Code Queue PR preflight contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:submit-routing-contract", "Code Queue submit routing contract is opt-in with script checks", "--scripts-typecheck or --full")); + items.push(skippedItem("provider:runner-triage-contract", "Provider runner triage contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("deploy:artifact-matrix-contract", "deploy artifact matrix contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:liveness-diagnostics-fixtures", "Code Queue liveness diagnostics fixtures are opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("baidu-netdisk:artifact-guard-contract", "Baidu Netdisk artifact guard contract is opt-in with script checks", "--scripts-typecheck or --full")); diff --git a/scripts/src/provider-triage.ts b/scripts/src/provider-triage.ts index d2b3d264..c1c12486 100644 --- a/scripts/src/provider-triage.ts +++ b/scripts/src/provider-triage.ts @@ -3,6 +3,7 @@ import { coreInternalFetch } from "./microservices"; import { debugDispatch, debugHealth } from "./debug"; import { runArtifactRegistryCommand } from "./artifact-registry"; import { runCodeQueueCommand } from "./code-queue"; +import { classifyRunnerError } from "../../src/components/microservices/code-queue/src/runner-error-classifier"; export type ProviderSignalScope = | "runner-local" @@ -270,6 +271,12 @@ function codeQueueTasksSignal(response: unknown): ProviderTriageSignal { } function classifyErrorMessage(message: string): ProviderSignalScope { + const runnerClassification = classifyRunnerError(message); + if (runnerClassification.scope === "provider-gateway") return "provider-gateway"; + if (runnerClassification.scope === "registry") return "registry"; + if (runnerClassification.scope === "k3s") return "k3s"; + if (runnerClassification.scope === "scheduler") return "scheduler"; + if (runnerClassification.scope === "runner-local") return "runner-local"; const normalized = message.toLowerCase(); if (/provider is not online|provider .*offline|provider .*not online/u.test(normalized)) return "runner-local"; if (/ssh|host\.ssh/u.test(normalized)) return "ssh"; @@ -282,7 +289,7 @@ function classifyErrorMessage(message: string): ProviderSignalScope { } function observedErrorSignal(message: string, scope: ProviderSignalScope): ProviderTriageSignal { - return signal("observed-error", scope, "failed", message, { message }, scope !== "runner-local"); + return signal("observed-error", scope, "failed", message, { message, runnerErrorClassification: classifyRunnerError(message) }, scope !== "runner-local"); } export function providerTriageRecommendedCrossChecks(providerId: string): string[] { diff --git a/src/components/microservices/code-queue/src/code-agent/codex.ts b/src/components/microservices/code-queue/src/code-agent/codex.ts index ecf1ae34..d10269c1 100644 --- a/src/components/microservices/code-queue/src/code-agent/codex.ts +++ b/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; @@ -393,6 +394,17 @@ function handleNotification(task: QueueTask, message: Record, t } } +function classifyCodexRunnerFailure(task: QueueTask, result: Pick): 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 { const queueId = ctx().queueIdOf(task); const events: CodexEventSummary[] = []; @@ -447,7 +459,7 @@ export async function runCodexTurn(task: QueueTask, prompt: string): Promise; @@ -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): 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 { const attemptedSessionId = task.codexThreadId; const first = await runOpenCodeTurnOnce(task, prompt); @@ -365,6 +377,7 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise 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 0 ? app.finalResponse : "", @@ -428,6 +443,8 @@ async function runOpenCodeTurnOnce(task: QueueTask, prompt: string): Promise 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) : ""; + 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; +} diff --git a/src/components/microservices/code-queue/src/types.ts b/src/components/microservices/code-queue/src/types.ts index ab6f041d..b7df7468 100644 --- a/src/components/microservices/code-queue/src/types.ts +++ b/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[]; }