Q69: provider failures classify by status and code; cooldowns follow Retry-After or back off
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
4325dfa874
commit
c6a0460035
15 files changed
+198
-41
No files matched your search
@@ -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'));
|
||||
`;
|
||||
@@ -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;
|
||||
|
||||
@@ -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`);
|
||||
|
||||
@@ -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<string, { until: number; label: string; interactiveProbedAt?: number }>();
|
||||
/** Transient failures in a row per provider, reset by a served request. */
|
||||
private readonly strikes = new Map<string, number>();
|
||||
|
||||
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.`);
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<T>(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<ReadableStreamReadResult<Uint8Array>> {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const idle = new Promise<never>((_, 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]);
|
||||
|
||||
@@ -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<ProviderEvent
|
||||
const fragments = new Map<number, { id: string; name: string; args: string }>();
|
||||
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<ProviderEvent
|
||||
return { text: textParts.join("").trim(), toolCalls, rawResponseJson: JSON.stringify(raw) };
|
||||
}
|
||||
|
||||
/** An in-stream error object's HTTP-style code (OpenRouter sends e.g. { code: 500 }), or null. */
|
||||
function streamErrorStatus(error: unknown): number | null {
|
||||
return isRecord(error) && typeof error.code === "number" ? error.code : null;
|
||||
}
|
||||
|
||||
function streamErrorText(error: unknown): string {
|
||||
if (typeof error === "string") return error;
|
||||
if (isRecord(error) && typeof error.message === "string") return error.message;
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
* Ported from Pi's google-shared `convertMessages` / `convertTools` and the
|
||||
* Antigravity request wrapper in google-gemini-cli.
|
||||
*/
|
||||
import { isRecord, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http";
|
||||
import { isRecord, ProviderHttpError, type JsonRecord, type ProviderEventStreamMessage } from "../provider-http";
|
||||
import { type AgentTurnMessage, type AgentTurnResult, type ToolCall, type ToolDefinition } from "../provider-tool-turn";
|
||||
|
||||
const ANTIGRAVITY_SYSTEM_INSTRUCTION =
|
||||
@@ -137,7 +137,7 @@ export async function accumulateGeminiStream(messages: AsyncIterable<ProviderEve
|
||||
raw.push(message.rawData);
|
||||
if (isRecord(message.data.error)) {
|
||||
const error = message.data.error;
|
||||
throw new Error(typeof error.message === "string" ? error.message : "Antigravity stream failed.");
|
||||
throw new ProviderHttpError(typeof error.message === "string" ? error.message : "Antigravity stream failed.", null, typeof error.code === "number" ? error.code : null);
|
||||
}
|
||||
const response = isRecord(message.data.response) ? message.data.response : null;
|
||||
if (!response) continue;
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
* (Codex) and github-copilot turns. Matches Pi's `convertResponsesMessages` /
|
||||
* `convertResponsesTools` and the Responses streaming event shapes.
|
||||
*/
|
||||
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 responsesToolDefs(tools: ToolDefinition[], strict: boolean | null = false): unknown[] {
|
||||
@@ -66,14 +66,28 @@ function eventType(message: ProviderEventStreamMessage): string | null {
|
||||
return typeof type === "string" ? type : message.eventName;
|
||||
}
|
||||
|
||||
/** Responses error codes that name passing provider trouble, as HTTP statuses for failure classification. */
|
||||
const RESPONSES_ERROR_STATUS: Record<string, number> = { 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<Provider
|
||||
raw.push(message.rawData);
|
||||
const type = eventType(message);
|
||||
if (!type) continue;
|
||||
if (type === "error" || type === "response.failed") throw new Error(errorText(message.data));
|
||||
if (type === "error" || type === "response.failed") throw new ProviderHttpError(errorText(message.data), null, errorStatus(message.data));
|
||||
if (type === "response.output_item.done" && isRecord(message.data.item)) {
|
||||
streamed.push(message.data.item);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { RunTaskInput } from "../../shared/commands";
|
||||
import { TASK_STATUS, WRITING_MODE } from "../../shared/domain";
|
||||
import { isTransientProviderFailure } from "../../shared/provider-runtime-domain";
|
||||
import { createId } from "../../shared/ids";
|
||||
import { AuditService } from "./audit-service";
|
||||
import { ProviderService } from "./provider-service";
|
||||
@@ -13,6 +12,7 @@ import { runAgentLoop, runTools } from "./agent-run-loop";
|
||||
import type { RunModel } from "./skill-runtime";
|
||||
import { combineMidRunPolls, createGoalChangePoller } from "./goal-change-poller";
|
||||
import { recordRunAction, runJournalLines } from "./run-journal";
|
||||
import { classifyProviderFailure } from "./provider-failure";
|
||||
import { runStopReason, SESSION_TIME_LIMIT_MESSAGE, SESSION_TIME_LIMIT_MS, withRunControl } from "./run-control";
|
||||
import type { AgentToolContext } from "./agent-tools";
|
||||
import {
|
||||
@@ -145,7 +145,7 @@ export class TaskExecutionService {
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
console.error(`[task-run] run ${runId} failed before execution: ${message}`);
|
||||
this.failRun(orgId, runId, null, message);
|
||||
this.failRun(orgId, runId, null, message, false);
|
||||
this.audit.record({
|
||||
orgId,
|
||||
entityType: "task_run",
|
||||
@@ -234,7 +234,11 @@ export class TaskExecutionService {
|
||||
}),
|
||||
toolContext,
|
||||
externalTools,
|
||||
onModelTurn: contextRecorder.commit,
|
||||
onModelTurn: () => {
|
||||
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 {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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]) {
|
||||
|
||||
Reference in new issue
Block a user