From c6a046003582cef362f8b515985433078d9f385e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 12:03:37 -0400 Subject: [PATCH] Q69: provider failures classify by status and code; cooldowns follow Retry-After or back off Co-Authored-By: Claude Opus 5.5 --- .../migrations/074-run-failure-kind.ts | 12 ++++++ src/main/database/schema.ts | 2 + src/main/services/heartbeat-scheduler.ts | 8 ++-- src/main/services/provider-cooldowns.ts | 29 +++++++------ src/main/services/provider-failure.ts | 37 ++++++++++++++++ src/main/services/provider-http.ts | 11 +++-- .../provider-tool-turns/chat-format.ts | 9 +++- .../provider-tool-turns/gemini-format.ts | 4 +- .../provider-tool-turns/responses-format.ts | 20 +++++++-- src/main/services/task-execution-service.ts | 21 ++++++---- src/shared/provider-runtime-domain.ts | 2 +- tests/unit/domain-services.test.ts | 2 +- tests/unit/provider-cooldown-backoff.test.ts | 32 ++++++++++++++ tests/unit/provider-failure.test.ts | 42 +++++++++++++++++++ tests/unit/task-execution-service.test.ts | 8 ++-- 15 files changed, 198 insertions(+), 41 deletions(-) create mode 100644 src/main/database/migrations/074-run-failure-kind.ts create mode 100644 src/main/services/provider-failure.ts create mode 100644 tests/unit/provider-cooldown-backoff.test.ts create mode 100644 tests/unit/provider-failure.test.ts diff --git a/src/main/database/migrations/074-run-failure-kind.ts b/src/main/database/migrations/074-run-failure-kind.ts new file mode 100644 index 00000000..a8d6a03c --- /dev/null +++ b/src/main/database/migrations/074-run-failure-kind.ts @@ -0,0 +1,12 @@ +import { TRANSIENT_PROVIDER_FAILURE_SIGNATURES } from "../../../shared/provider-runtime-domain"; + +/** Runs record whether they failed for a passing provider reason; older rows are classified from their message once, here. */ +export const runFailureKindSchema = ` + alter table task_runs add column failure_kind text check (failure_kind in ('transient')); + + update task_runs set failure_kind = 'transient' + where state = 'failed' and (${TRANSIENT_PROVIDER_FAILURE_SIGNATURES.map((signature) => `lower(error_message) like '%${signature.replaceAll("'", "''")}%'`).join(" or ")}); + + insert into schema_migrations(version, name, applied_at) + values (74, 'run_failure_kind', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index cb642769..bee8f601 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -72,6 +72,7 @@ import { runActionsSchema } from "./migrations/070-run-actions"; import { backgroundPollFailuresSchema } from "./migrations/071-background-poll-failures"; import { spendSchema } from "./migrations/072-spend"; import { usageWindowTokensSchema } from "./migrations/073-usage-window-tokens"; +import { runFailureKindSchema } from "./migrations/074-run-failure-kind"; interface Migration { version: number; @@ -153,6 +154,7 @@ const MIGRATIONS: Migration[] = [ { version: 71, name: "background_poll_failures", sql: backgroundPollFailuresSchema }, { version: 72, name: "spend", sql: spendSchema }, { version: 73, name: "usage_window_tokens", sql: usageWindowTokensSchema }, + { version: 74, name: "run_failure_kind", sql: runFailureKindSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index 9f060b81..5e1b6ded 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -1,5 +1,5 @@ import type { DatabaseSync } from "node:sqlite"; -import { currentProviderUsageWindow, isTransientProviderFailure } from "../../shared/provider-runtime-domain"; +import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain"; import { createId } from "../../shared/ids"; import type { ProviderCooldowns } from "./provider-cooldowns"; import { assertExecutionApproval, createIsolationContext, loadWorkspace, requireAgentProvider, requireAssignedAgent, requireTask } from "./task-run-preparation"; @@ -54,7 +54,7 @@ export class HeartbeatScheduler { this.observeQuotaBlocks(window.start); const cooling = new Set(this.cooldowns.cooledProviderIds()); const sessionsOf = this.db.prepare(` - select r.error_message, + select r.failure_kind, exists (select 1 from run_actions a where a.run_id = r.id and a.result_excerpt not like 'ERROR%') as progressed from task_runs r join tasks t on t.id = r.task_id where r.task_id = ? and r.state not in ('superseded', 'stopped', 'queued', 'running') and r.rowid > t.run_budget_floor @@ -71,8 +71,8 @@ export class HeartbeatScheduler { continue; } this.budgetParkedAgents.delete(row.agent_id); - const sessions = (sessionsOf.all(row.task_id) as Array<{ error_message: string | null; progressed: number }>) - .filter((run) => !isTransientProviderFailure(run.error_message)); + const sessions = (sessionsOf.all(row.task_id) as Array<{ failure_kind: string | null; progressed: number }>) + .filter((run) => run.failure_kind !== "transient"); const stuck = sessions.filter((run) => run.progressed === 0).length; if (stuck >= TASK_RETRY_BUDGET) { this.markStalled(row.org_id, row.task_id, `${TASK_RETRY_BUDGET} sessions made no progress`); diff --git a/src/main/services/provider-cooldowns.ts b/src/main/services/provider-cooldowns.ts index e03a188e..f825a14d 100644 --- a/src/main/services/provider-cooldowns.ts +++ b/src/main/services/provider-cooldowns.ts @@ -1,13 +1,15 @@ -/** - * How long a provider's tasks stay parked after it reports a transient - * failure (usage limit, rate limit, overload). Long enough not to hammer a - * closed door every 5s tick, short enough to probe again the same hour. - */ -const PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS = 10 * 60_000; +/** Without a provider-given wait, parks double from this base with each failure in a row, up to the cap. */ +const BACKOFF_BASE_MS = 2 * 60_000; +const BACKOFF_CAP_MS = 30 * 60_000; +/** A provider-given wait is honored within these bounds. */ +const RETRY_AFTER_MIN_MS = 30_000; +const RETRY_AFTER_MAX_MS = 60 * 60_000; /** In-memory parks for providers that reported a transient failure. */ export class ProviderCooldowns { private readonly parks = new Map(); + /** Transient failures in a row per provider, reset by a served request. */ + private readonly strikes = new Map(); constructor(private readonly now: () => number) {} @@ -17,20 +19,22 @@ export class ProviderCooldowns { * The park is in-memory by design: after a restart the first probe run * either succeeds or re-arms the cooldown at the cost of one request. */ - park(provider: { id: string; label: string }, message: string): void { + park(provider: { id: string; label: string }, message: string, retryAfterMs: number | null): void { const active = this.parks.get(provider.id); const wasParked = active !== undefined && active.until > this.now(); + const strikes = (this.strikes.get(provider.id) ?? 0) + 1; + this.strikes.set(provider.id, strikes); + const waitMs = retryAfterMs !== null + ? Math.min(Math.max(retryAfterMs, RETRY_AFTER_MIN_MS), RETRY_AFTER_MAX_MS) + : Math.min(BACKOFF_BASE_MS * 2 ** (strikes - 1), BACKOFF_CAP_MS); this.parks.set(provider.id, { - until: this.now() + PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS, + until: this.now() + waitMs, label: provider.label, // A failed probe must not unlock endless probes: re-arming the park // keeps the spent-probe mark until the window expires naturally. interactiveProbedAt: wasParked ? active?.interactiveProbedAt : undefined, }); - if (!wasParked) { - const minutes = Math.round(PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS / 60_000); - console.warn(`[task-queue] provider "${provider.label}" reported a transient failure ("${message}"); its tasks are parked for ${minutes} minutes.`); - } + console.warn(`[task-queue] provider "${provider.label}" reported a transient failure ("${message}"); its tasks are parked for ${Math.round(waitMs / 1000)}s (failure ${strikes} in a row).`); } /** @@ -70,6 +74,7 @@ export class ProviderCooldowns { /** A served request is proof of health: the park ends for everyone at once. */ clear(provider: { id: string; label: string }): void { + this.strikes.delete(provider.id); if (this.parks.delete(provider.id)) { console.info(`[task-queue] provider "${provider.label}" served a request; cooldown lifted early.`); } diff --git a/src/main/services/provider-failure.ts b/src/main/services/provider-failure.ts new file mode 100644 index 00000000..4a8690f2 --- /dev/null +++ b/src/main/services/provider-failure.ts @@ -0,0 +1,37 @@ +import { isTransientProviderFailure } from "../../shared/provider-runtime-domain"; +import { ProviderHttpError } from "./provider-http"; + +/** Whether a failed call is the provider's passing trouble, and how long it asked us to wait. */ +export interface ProviderFailure { + transient: boolean; + retryAfterMs: number | null; +} + +/** Request timeout, too early, rate limited; every 5xx (server, gateway, Cloudflare 52x, Anthropic 529 overload) is transient too. */ +const TRANSIENT_STATUSES = new Set([408, 425, 429]); +/** Connection-level failures from Node's fetch (undici) and sockets. */ +const TRANSIENT_CODES = new Set([ + "ECONNRESET", "ETIMEDOUT", "ECONNREFUSED", "EAI_AGAIN", "EPIPE", + "UND_ERR_SOCKET", "UND_ERR_CONNECT_TIMEOUT", "UND_ERR_HEADERS_TIMEOUT", "UND_ERR_BODY_TIMEOUT", +]); + +/** Classifies by status, timeout and network error code first; the message text is the last resort. */ +export function classifyProviderFailure(error: unknown, now: number): ProviderFailure { + if (error instanceof ProviderHttpError) { + const state = error.rateLimitState; + const resetIn = state?.resetAt ? Date.parse(state.resetAt) - now : null; + const retryAfterMs = state?.retryAfterMs ?? (resetIn !== null && resetIn > 0 ? resetIn : null); + if (error.timedOut || (error.status !== null && (TRANSIENT_STATUSES.has(error.status) || (error.status >= 500 && error.status < 600)))) return { transient: true, retryAfterMs }; + if (error.status !== null) return { transient: isTransientProviderFailure(error.message), retryAfterMs }; + } + if (TRANSIENT_CODES.has(errorCode(error) ?? "")) return { transient: true, retryAfterMs: null }; + return { transient: isTransientProviderFailure(error instanceof Error ? error.message : String(error)), retryAfterMs: null }; +} + +function errorCode(error: unknown): string | null { + if (!(error instanceof Error)) return null; + const own = (error as NodeJS.ErrnoException).code; + if (typeof own === "string") return own; + const cause = (error as Error & { cause?: unknown }).cause; + return cause instanceof Error ? errorCode(cause) : null; +} diff --git a/src/main/services/provider-http.ts b/src/main/services/provider-http.ts index 86fb5044..5eac66ea 100644 --- a/src/main/services/provider-http.ts +++ b/src/main/services/provider-http.ts @@ -34,6 +34,9 @@ export class ProviderHttpError extends Error { constructor( message: string, readonly rateLimitState: ProviderObservedRateLimitState | null, + /** HTTP status of the failed response; null when no response arrived. */ + readonly status: number | null = null, + readonly timedOut = false, ) { super(message); this.name = "ProviderHttpError"; @@ -86,7 +89,7 @@ export async function postEventStream( const raw = await withDeadline(controller, timeouts.bodyTimeoutMs ?? BODY_TIMEOUT_MS, () => response.text()); const data = parseProviderJson(raw, false); const message = readProviderError(data) ?? (raw.trim() || `Provider request failed with HTTP ${response.status}.`); - throw new ProviderHttpError(message, rateLimitState); + throw new ProviderHttpError(message, rateLimitState, response.status); } if (!response.body) throw new ProviderHttpError("Provider returned an empty event stream.", rateLimitState); return { messages: readServerSentEvents(response.body, timeouts.idleTimeoutMs ?? STREAM_IDLE_TIMEOUT_MS), rateLimitState }; @@ -100,7 +103,7 @@ async function requestJson(endpoint: string, init: RequestInit, timeouts: Provid const data = parseProviderJson(raw, response.ok); if (!response.ok) { const message = readProviderError(data) ?? (raw.trim() || `Provider request failed with HTTP ${response.status}.`); - throw new ProviderHttpError(message, rateLimitState); + throw new ProviderHttpError(message, rateLimitState, response.status); } return { data, rateLimitState }; } @@ -126,7 +129,7 @@ async function withDeadline(controller: AbortController, timeoutMs: number, o return await operation(); } catch (error) { throwIfRunStopped(); - if (controller.signal.aborted) throw new ProviderHttpError("Provider request timed out.", null); + if (controller.signal.aborted) throw new ProviderHttpError("Provider request timed out.", null, null, true); throw error; } finally { clearTimeout(timer); @@ -175,7 +178,7 @@ async function readWithIdleDeadline( ): Promise> { let timer: NodeJS.Timeout | undefined; const idle = new Promise((_, reject) => { - timer = setTimeout(() => reject(new ProviderHttpError("Provider stream timed out.", null)), idleTimeoutMs); + timer = setTimeout(() => reject(new ProviderHttpError("Provider stream timed out.", null, null, true)), idleTimeoutMs); }); try { return await Promise.race([reader.read(), idle]); diff --git a/src/main/services/provider-tool-turns/chat-format.ts b/src/main/services/provider-tool-turns/chat-format.ts index 02e87bcf..b46dba4c 100644 --- a/src/main/services/provider-tool-turns/chat-format.ts +++ b/src/main/services/provider-tool-turns/chat-format.ts @@ -1,5 +1,5 @@ /** OpenAI Chat Completions encoding/decoding (xAI, Groq, Mistral, Cerebras, OpenRouter, Kimi family). */ -import { isRecord, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http"; +import { isRecord, ProviderHttpError, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http"; import { parseArguments, type AgentTurnMessage, type AgentTurnResult, type ToolCall, type ToolDefinition } from "../provider-tool-turn"; export function chatToolDefs(tools: ToolDefinition[]): unknown[] { @@ -63,7 +63,7 @@ export async function accumulateChatStream(messages: AsyncIterable(); for await (const message of messages) { raw.push(message.rawData); - if (message.data.error !== undefined) throw new Error(streamErrorText(message.data.error)); + if (message.data.error !== undefined) throw new ProviderHttpError(streamErrorText(message.data.error), null, streamErrorStatus(message.data.error)); const choices = Array.isArray(message.data.choices) ? message.data.choices : []; const first = choices.length > 0 && isRecord(choices[0]) ? choices[0] : null; const delta = first && isRecord(first.delta) ? first.delta : null; @@ -89,6 +89,11 @@ export async function accumulateChatStream(messages: AsyncIterable = { server_error: 500, rate_limit_exceeded: 429 }; + +/** The error of an `error` event, or of a `response.failed` event's response. */ +function streamError(data: JsonRecord): unknown { + return data.error ?? (isRecord(data.response) ? data.response.error : undefined); +} + function errorText(data: JsonRecord): string { - const error = data.error; + const error = streamError(data); if (typeof error === "string") return error; if (isRecord(error) && typeof error.message === "string") return error.message; if (typeof data.message === "string") return data.message; return "Provider stream failed."; } +function errorStatus(data: JsonRecord): number | null { + const error = streamError(data); + const code = isRecord(error) ? error.code : data.code; + return typeof code === "string" ? RESPONSES_ERROR_STATUS[code] ?? null : null; +} + /** * Accumulate a Responses SSE stream (Codex/Copilot) into the final result. * @@ -91,7 +105,7 @@ export async function accumulateResponsesStream(messages: AsyncIterable { + contextRecorder.commit(); + // A served request ends the provider's failure streak, so the next park starts small again. + this.cooldowns.clear(provider); + }, onToolCompleted: (call, output) => recordRunAction(this.db, { orgId, runId, taskId: task.id, call, output }), inputBudgetTokens: inputBudgetTokens(provider.context_window_tokens), pollNewMessages: combineMidRunPolls( @@ -298,8 +302,9 @@ export class TaskExecutionService { this.markStopped(orgId, runId, providerRequestId, stopReason, { taskId: task.id, agentId: agent.id, providerId: provider.id }); throw new Error(stopReason); } - this.failRun(orgId, runId, providerRequestId, message); - if (isTransientProviderFailure(message)) this.cooldowns.park(provider, message); + const failure = classifyProviderFailure(error, Date.now()); + this.failRun(orgId, runId, providerRequestId, message, failure.transient); + if (failure.transient) this.cooldowns.park(provider, message, failure.retryAfterMs); else if (task.origin === null) recordRunMemory(this.db, { orgId, runId, projectId: task.project_id, taskTitle: task.title, outcome: "failed", text: message }); this.audit.record({ orgId, @@ -462,12 +467,12 @@ export class TaskExecutionService { return runId; } - private failRun(orgId: string, runId: string, providerRequestId: string | null, errorMessage: string): void { + private failRun(orgId: string, runId: string, providerRequestId: string | null, errorMessage: string, transient: boolean): void { this.db.prepare(` update task_runs - set state = ?, provider_request_id = coalesce(?, provider_request_id), completed_at = ?, error_message = ? + set state = ?, provider_request_id = coalesce(?, provider_request_id), completed_at = ?, error_message = ?, failure_kind = ? where id = ? and org_id = ? - `).run(TASK_RUN_STATE.FAILED, providerRequestId, new Date().toISOString(), errorMessage, runId, orgId); + `).run(TASK_RUN_STATE.FAILED, providerRequestId, new Date().toISOString(), errorMessage, transient ? "transient" : null, runId, orgId); } private resolveSkillRuntimeSnapshot(orgId: string, agentId: string, runId: string, maxTokens: number, taskText: string, runModel: RunModel, houseStyle: boolean): SkillRuntimeSnapshotRow | null { diff --git a/src/shared/provider-runtime-domain.ts b/src/shared/provider-runtime-domain.ts index 87a8b897..1f2bb2c7 100644 --- a/src/shared/provider-runtime-domain.ts +++ b/src/shared/provider-runtime-domain.ts @@ -46,7 +46,7 @@ export const PROVIDER_MONTHLY_LIMIT_MESSAGE = "Provider monthly request limit ha * Matched case-insensitively against recorded task_runs.error_message rows, * which also heals historical rows written before this classification existed. */ -const TRANSIENT_PROVIDER_FAILURE_SIGNATURES = [ +export const TRANSIENT_PROVIDER_FAILURE_SIGNATURES = [ PROVIDER_MONTHLY_LIMIT_MESSAGE, "usage limit", // ChatGPT-account plans: "The usage limit has been reached" "rate limit", diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index a8effa19..fd1973ea 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -24,7 +24,7 @@ describe("domain services", () => { const connection = createConnection(); const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>; - expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73]); + expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74]); connection.close(); }); diff --git a/tests/unit/provider-cooldown-backoff.test.ts b/tests/unit/provider-cooldown-backoff.test.ts new file mode 100644 index 00000000..14fc3bbf --- /dev/null +++ b/tests/unit/provider-cooldown-backoff.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, test, vi } from "vitest"; +import { ProviderCooldowns } from "../../src/main/services/provider-cooldowns"; + +describe("provider cooldown sizing", () => { + test("a provider-given wait is honored; otherwise parks back off and reset once a request is served", () => { + vi.spyOn(console, "warn").mockImplementation(() => {}); + vi.spyOn(console, "info").mockImplementation(() => {}); + let now = 0; + const cooldowns = new ProviderCooldowns(() => now); + const provider = { id: "prv_1", label: "OpenAI" }; + const parkedFor = (): number => { + let elapsed = 0; + while (cooldowns.cooledProviderIds().includes(provider.id)) { + now += 1_000; + elapsed += 1_000; + } + return elapsed; + }; + cooldowns.park(provider, "429", 45_000); + expect(parkedFor()).toBe(45_000); + cooldowns.park(provider, "503", null); + expect(parkedFor()).toBe(4 * 60_000); + cooldowns.park(provider, "503", null); + expect(parkedFor()).toBe(8 * 60_000); + cooldowns.clear(provider); + cooldowns.park(provider, "503", null); + expect(parkedFor()).toBe(2 * 60_000); + cooldowns.park(provider, "429", 5 * 3_600_000); + expect(parkedFor()).toBe(60 * 60_000); + vi.restoreAllMocks(); + }); +}); diff --git a/tests/unit/provider-failure.test.ts b/tests/unit/provider-failure.test.ts new file mode 100644 index 00000000..5aa5f179 --- /dev/null +++ b/tests/unit/provider-failure.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, test } from "vitest"; +import { classifyProviderFailure } from "../../src/main/services/provider-failure"; +import { ProviderHttpError } from "../../src/main/services/provider-http"; +import { accumulateChatStream } from "../../src/main/services/provider-tool-turns/chat-format"; +import { accumulateResponsesStream } from "../../src/main/services/provider-tool-turns/responses-format"; + +const NOW = Date.parse("2026-10-08T12:00:00.000Z"); +const limits = (retryAfterMs: number | null, resetAt: string | null) => ({ requestLimit: null, remainingRequests: 0, resetAt, retryAfterMs, rawStateJson: "{}" }); + +describe("provider failure classification", () => { + test("status, timeouts and network codes decide, whatever the message says", () => { + expect(classifyProviderFailure(new ProviderHttpError("slow down", limits(45_000, null), 429), NOW)).toEqual({ transient: true, retryAfterMs: 45_000 }); + expect(classifyProviderFailure(new ProviderHttpError("", limits(null, "2026-10-08T12:05:00.000Z"), 503), NOW)).toEqual({ transient: true, retryAfterMs: 300_000 }); + expect(classifyProviderFailure(new ProviderHttpError("{\"type\":\"overloaded_error\"}", null, 529), NOW)).toEqual({ transient: true, retryAfterMs: null }); + expect(classifyProviderFailure(new ProviderHttpError("Provider request timed out.", null, null, true), NOW)).toEqual({ transient: true, retryAfterMs: null }); + expect(classifyProviderFailure(new ProviderHttpError("invalid api key", null, 401), NOW)).toEqual({ transient: false, retryAfterMs: null }); + expect(classifyProviderFailure(new ProviderHttpError("The usage limit has been reached", null, 400), NOW)).toEqual({ transient: true, retryAfterMs: null }); + const reset = Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }); + expect(classifyProviderFailure(new TypeError("fetch failed", { cause: reset }), NOW)).toEqual({ transient: true, retryAfterMs: null }); + expect(classifyProviderFailure(new ProviderHttpError("", null, 522), NOW)).toEqual({ transient: true, retryAfterMs: null }); + expect(classifyProviderFailure(new Error("Agent has no enabled skills."), NOW)).toEqual({ transient: false, retryAfterMs: null }); + }); + + test("an error event inside a chat stream keeps its status code", async () => { + async function* stream() { + yield { eventName: null, data: { error: { code: 500, message: "Internal server error" } }, rawData: "{}" }; + } + const failure = await accumulateChatStream(stream()).catch((error: unknown) => error); + expect(classifyProviderFailure(failure, NOW)).toEqual({ transient: true, retryAfterMs: null }); + }); + + test("a failed Responses stream keeps its server_error and rate-limit codes", async () => { + async function* failed(code: string) { + yield { eventName: null, data: { type: "response.failed", response: { error: { code, message: "The server had an error." } } }, rawData: "{}" }; + } + const serverError = await accumulateResponsesStream(failed("server_error")).catch((error: unknown) => error); + expect((serverError as Error).message).toBe("The server had an error."); + expect(classifyProviderFailure(serverError, NOW)).toEqual({ transient: true, retryAfterMs: null }); + expect(classifyProviderFailure(await accumulateResponsesStream(failed("rate_limit_exceeded")).catch((error: unknown) => error), NOW).transient).toBe(true); + expect(classifyProviderFailure(await accumulateResponsesStream(failed("invalid_prompt")).catch((error: unknown) => error), NOW).transient).toBe(false); + }); +}); diff --git a/tests/unit/task-execution-service.test.ts b/tests/unit/task-execution-service.test.ts index 09ff8671..fa7591f9 100644 --- a/tests/unit/task-execution-service.test.ts +++ b/tests/unit/task-execution-service.test.ts @@ -553,8 +553,8 @@ describe("task execution service", () => { insert into task_runs( id, org_id, task_id, agent_id, provider_id, provider_request_id, execution_context_id, service_identity_id, workspace_mount_json, secret_scope_json, - trigger, state, created_at, started_at, completed_at, error_message - ) values (?, ?, ?, ?, ?, null, ?, ?, '[]', '{}', 'heartbeat', 'failed', ?, ?, ?, ?) + trigger, state, created_at, started_at, completed_at, error_message, failure_kind + ) values (?, ?, ?, ?, ?, null, ?, ?, '[]', '{}', 'heartbeat', 'failed', ?, ?, ?, ?, 'transient') `); const now = new Date().toISOString(); for (const index of [1, 2, 3]) { @@ -638,8 +638,8 @@ describe("task execution service", () => { insert into task_runs( id, org_id, task_id, agent_id, provider_id, provider_request_id, execution_context_id, service_identity_id, workspace_mount_json, secret_scope_json, - trigger, state, created_at, started_at, completed_at, error_message - ) values (?, ?, ?, ?, ?, null, ?, ?, '[]', '{}', 'heartbeat', 'failed', ?, ?, ?, ?) + trigger, state, created_at, started_at, completed_at, error_message, failure_kind + ) values (?, ?, ?, ?, ?, null, ?, ?, '[]', '{}', 'heartbeat', 'failed', ?, ?, ?, ?, 'transient') `); const now = new Date().toISOString(); for (const index of [1, 2, 3]) {