diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index 42f9dc55..1254a065 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -26,12 +26,17 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean { /** * Atomically claims the oldest queued task_run by transitioning it * `queued` -> `running` in one transaction; returns the claimed run or null. + * Runs whose provider sits in a transient-failure cooldown are left queued — + * they would fail on arrival, and waiting costs nothing. */ -export function claimNextQueuedRun(connection: DatabaseConnection): QueuedRun | null { +export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = []): QueuedRun | null { return connection.transaction(() => { + const exclusion = excludedProviderIds.length > 0 + ? `and provider_id not in (${excludedProviderIds.map(() => "?").join(", ")})` + : ""; const row = connection.db - .prepare("select id, org_id from task_runs where state = 'queued' order by created_at asc limit 1") - .get() as QueuedRun | undefined; + .prepare(`select id, org_id from task_runs where state = 'queued' ${exclusion} order by created_at asc limit 1`) + .get(...excludedProviderIds) as QueuedRun | undefined; if (!row) return null; const result = connection.db .prepare("update task_runs set state = 'running', started_at = coalesce(started_at, ?) where id = ? and state = 'queued'") @@ -70,7 +75,7 @@ export function startTaskQueueWorker(input: TaskQueueWorkerInput): () => void { } if (paused) return; const enqueued = input.connection.transaction(() => taskExecution.enqueueHeartbeatRuns()); - const claimed = claimNextQueuedRun(input.connection); + const claimed = claimNextQueuedRun(input.connection, taskExecution.cooledProviderIds()); if (!claimed) { if (enqueued.length > 0) notifyDataChanged(); return; diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 408f72f4..7a912b1d 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -1,7 +1,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { RunTaskInput } from "../../shared/commands"; import { APPROVAL_STATE, RISK_LEVEL } from "../../shared/domain"; -import { currentProviderUsageWindow, PROVIDER_MONTHLY_LIMIT_MESSAGE } from "../../shared/provider-runtime-domain"; +import { currentProviderUsageWindow, isTransientProviderFailure } from "../../shared/provider-runtime-domain"; import { createId } from "../../shared/ids"; import { AuditService } from "./audit-service"; import { ProviderService } from "./provider-service"; @@ -35,6 +35,15 @@ const TASK_RUN_STATE = { FAILED: "failed", } as const; +const TASK_RETRY_BUDGET = 3; + +/** + * 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; + interface TaskApprovalRow { id: string; state: string; @@ -56,6 +65,8 @@ export class TaskExecutionService { private readonly chat: ChatService; /** Providers currently parked on their monthly quota, for transition logging. */ private quotaBlockedProviders = new Map(); + /** Providers parked after a transient failure (usage/rate limit, overload): id -> park state. */ + private providerCooldowns = new Map(); constructor( private readonly db: DatabaseSync, @@ -63,6 +74,8 @@ export class TaskExecutionService { private readonly skills: SkillService = new SkillService(db, ""), /** Invoked whenever a run writes data mid-flight, so the UI can refresh while agents work. */ private readonly notifyDataChanged: () => void = () => undefined, + /** Injectable clock so tests can walk providers out of cooldown. */ + private readonly now: () => number = () => Date.now(), ) { this.audit = new AuditService(db); this.organization = new OrganizationService(db); @@ -191,6 +204,7 @@ export class TaskExecutionService { } catch (error) { const message = error instanceof Error ? error.message : String(error); this.failRun(orgId, runId, providerRequestId, message); + if (isTransientProviderFailure(message)) this.parkProvider(provider, message); this.audit.record({ orgId, entityType: "task_run", @@ -300,16 +314,17 @@ export class TaskExecutionService { * CEO-delegated work starts on its own. Each task is enqueued at most once * (guarded by existing runs), preventing runaway re-execution. * - * Two quota rules keep the loop self-healing when a provider's monthly - * window fills up: tasks on an exhausted provider are parked (not enqueued - * into certain failure), and past quota-blocked failures never count toward - * the 3-attempt retry budget. Raising the limit therefore resumes work on - * the next tick — no restart or manual unpausing. + * Three rules keep the loop self-healing when a provider stops serving: + * tasks on a provider over its monthly window are parked (not enqueued into + * certain failure), tasks on a provider inside a transient-failure cooldown + * wait it out, and transient failures (usage/rate limits, overload) never + * count toward the retry budget. Raising a limit — or the provider's own + * window resetting — therefore resumes work without a restart. */ enqueueHeartbeatRuns(): string[] { const window = currentProviderUsageWindow(); const rows = this.db.prepare(` - select t.id as task_id, t.org_id as org_id + select t.id as task_id, t.org_id as org_id, p.id as provider_id from tasks t join agents a on a.id = t.assignee_agent_id and a.org_id = t.org_id join providers p on p.id = a.model_provider_id and p.org_id = t.org_id @@ -321,11 +336,16 @@ export class TaskExecutionService { and p.enabled = 1 and coalesce(u.request_count, 0) < p.monthly_request_limit and not exists (select 1 from task_runs r where r.task_id = t.id and r.state in ('queued', 'running', 'completed')) - and (select count(*) from task_runs r where r.task_id = t.id and coalesce(r.error_message, '') <> ?) < 3 - `).all(window.start, PROVIDER_MONTHLY_LIMIT_MESSAGE) as Array<{ task_id: string; org_id: string }>; + `).all(window.start) as Array<{ task_id: string; org_id: string; provider_id: string }>; this.observeQuotaBlocks(window.start); + const cooling = new Set(this.cooledProviderIds()); + const attemptMessages = this.db.prepare("select error_message from task_runs where task_id = ?"); const created: string[] = []; for (const row of rows) { + if (cooling.has(row.provider_id)) continue; + const attempts = (attemptMessages.all(row.task_id) as Array<{ error_message: string | null }>) + .filter((run) => !isTransientProviderFailure(run.error_message)).length; + if (attempts >= TASK_RETRY_BUDGET) continue; try { created.push(this.enqueueRun(row.org_id, row.task_id)); } catch { @@ -335,6 +355,39 @@ export class TaskExecutionService { return created; } + /** + * Parks a provider after it reported a transient failure (usage limit, rate + * limit, overload), so the queue stops feeding it doomed runs every tick. + * 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. + */ + private parkProvider(provider: TaskExecutionProviderRow, message: string): void { + const active = this.providerCooldowns.get(provider.id); + const wasParked = active !== undefined && active.until > this.now(); + this.providerCooldowns.set(provider.id, { until: this.now() + PROVIDER_TRANSIENT_FAILURE_COOLDOWN_MS, label: provider.label }); + 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.`); + } + } + + /** + * Providers currently inside a transient-failure cooldown. Expired parks are + * pruned (with a resume log line) so the next tick probes the provider again. + */ + cooledProviderIds(): string[] { + const active: string[] = []; + for (const [id, park] of this.providerCooldowns) { + if (park.until > this.now()) { + active.push(id); + } else { + this.providerCooldowns.delete(id); + console.info(`[task-queue] provider "${park.label}" cooldown ended; its tasks resume.`); + } + } + return active; + } + /** * Logs monthly-quota transitions (blocked <-> unblocked) exactly once per * state change so a parked queue is diagnosable from the main-process log diff --git a/src/shared/provider-runtime-domain.ts b/src/shared/provider-runtime-domain.ts index c915f506..87a8b897 100644 --- a/src/shared/provider-runtime-domain.ts +++ b/src/shared/provider-runtime-domain.ts @@ -38,6 +38,28 @@ export interface ProviderRateLimitObservationRecord extends ProviderObservedRate */ export const PROVIDER_MONTHLY_LIMIT_MESSAGE = "Provider monthly request limit has been reached."; +/** + * Failure-message signatures of transient provider conditions — the account's + * usage window filling up, rate limiting, or momentary overload. These verdicts + * come from the provider's side of the wire and say nothing about the task, so + * schedulers must treat them as "try again later", never as attempts consumed. + * 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 = [ + PROVIDER_MONTHLY_LIMIT_MESSAGE, + "usage limit", // ChatGPT-account plans: "The usage limit has been reached" + "rate limit", + "too many requests", // HTTP 429 status text + "overloaded", // "Our servers are currently overloaded. Please try again later." +].map((signature) => signature.toLowerCase()); + +export function isTransientProviderFailure(message: string | null | undefined): boolean { + if (!message) return false; + const normalized = message.toLowerCase(); + return TRANSIENT_PROVIDER_FAILURE_SIGNATURES.some((signature) => normalized.includes(signature)); +} + export function currentProviderUsageWindow(now = new Date()): { start: string; end: string } { const start = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1)); const end = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + 1, 1)); diff --git a/tests/unit/provider-runtime-domain.test.ts b/tests/unit/provider-runtime-domain.test.ts new file mode 100644 index 00000000..fbc7ec31 --- /dev/null +++ b/tests/unit/provider-runtime-domain.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, test } from "vitest"; +import { isTransientProviderFailure, PROVIDER_MONTHLY_LIMIT_MESSAGE } from "../../src/shared/provider-runtime-domain"; + +describe("isTransientProviderFailure", () => { + test("classifies provider-side limit and overload verdicts as transient", () => { + expect(isTransientProviderFailure(PROVIDER_MONTHLY_LIMIT_MESSAGE)).toBe(true); + expect(isTransientProviderFailure("The usage limit has been reached")).toBe(true); + expect(isTransientProviderFailure("Our servers are currently overloaded. Please try again later.")).toBe(true); + expect(isTransientProviderFailure("Rate limit exceeded, retry after 20s")).toBe(true); + expect(isTransientProviderFailure("429 Too Many Requests")).toBe(true); + }); + + test("keeps genuine task failures counting toward the retry budget", () => { + expect(isTransientProviderFailure(null)).toBe(false); + expect(isTransientProviderFailure(undefined)).toBe(false); + expect(isTransientProviderFailure("")).toBe(false); + expect(isTransientProviderFailure("provider rejected task")).toBe(false); + expect(isTransientProviderFailure("Agent run did not converge within 12 iterations.")).toBe(false); + expect(isTransientProviderFailure("Provider returned an empty turn: no text and no tool calls.")).toBe(false); + }); +}); diff --git a/tests/unit/task-execution-service.test.ts b/tests/unit/task-execution-service.test.ts index 6cd1c86c..6b09f192 100644 --- a/tests/unit/task-execution-service.test.ts +++ b/tests/unit/task-execution-service.test.ts @@ -542,4 +542,135 @@ describe("task execution service", () => { expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toEqual([]); connection.close(); }); + + test("provider-side usage-limit failures never consume the retry budget", () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const providers = new ProviderService(connection.db, providerAdapters, testVault); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + let providerId = ""; + let agentId = ""; + let taskId = ""; + + connection.transaction(() => { + orgId = organization.createOrganization({ + organizationName: "Acme Operations", + employeeName: "Dana Lee", + mission: "Coordinate agents with auditable approvals.", + goal: "Ship a verified control plane.", + }); + providerId = openAiProviderId(connection, orgId); + providers.saveCredential({ orgId, providerId, secret: "sk-live-provider", model: "gpt-5.4" }); + agentId = organization.createAgent({ + orgId, + name: "Ada", + role: "Execution lead", + department: "Engineering", + modelProviderId: providerId, + modelId: "gpt-5.4", + budgetCents: 10000, + managerAgentId: null, + heartbeatEnabled: true, + }); + taskId = organization.createTask({ + orgId, + title: "Draft launch checklist", + description: "Prepare an auditable checklist for release readiness.", + status: TASK_STATUS.IN_PROGRESS, + assigneeAgentId: agentId, + risk: RISK_LEVEL.MEDIUM, + projectId: null, + goalId: null, + workspaceId: null, + priority: TASK_PRIORITY.HIGH, + requiresFilesystem: false, + }); + // Historical damage from a provider-account limit: three failed runs, all + // recorded with the raw provider verdict rather than the internal quota message. + const serviceIdentityId = (connection.db.prepare("select id from service_identities where org_id = ? and agent_id = ?") + .get(orgId, agentId) as { id: string }).id; + const insertFailedRun = connection.db.prepare(` + 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', ?, ?, ?, ?) + `); + const now = new Date().toISOString(); + for (const index of [1, 2, 3]) { + insertFailedRun.run(`run_usage_fail_${index}`, orgId, taskId, agentId, providerId, `ctx_usage_${index}`, serviceIdentityId, now, now, now, "The usage limit has been reached"); + } + }); + + const enqueued = connection.transaction(() => execution.enqueueHeartbeatRuns()); + expect(enqueued).toHaveLength(1); + expect(connection.db.prepare("select task_id, state from task_runs where id = ?").get(enqueued[0])) + .toMatchObject({ task_id: taskId, state: "queued" }); + connection.close(); + }); + + test("a transient provider failure parks the provider and the cooldown expiry resumes it", async () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const fakeToolTurn = async (): Promise => { + throw new Error("The usage limit has been reached"); + }; + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, fakeToolTurn); + let nowMs = Date.now(); + const execution = new TaskExecutionService(connection.db, providers, undefined, undefined, () => nowMs); + let orgId = ""; + let providerId = ""; + let taskId = ""; + + connection.transaction(() => { + orgId = organization.createOrganization({ + organizationName: "Acme Operations", + employeeName: "Dana Lee", + mission: "Coordinate agents with auditable approvals.", + goal: "Ship a verified control plane.", + }); + providerId = openAiProviderId(connection, orgId); + providers.saveCredential({ orgId, providerId, secret: "sk-live-provider", model: "gpt-5.4" }); + const agentId = organization.createAgent({ + orgId, + name: "Ada", + role: "Execution lead", + department: "Engineering", + modelProviderId: providerId, + modelId: "gpt-5.4", + budgetCents: 10000, + managerAgentId: null, + heartbeatEnabled: true, + }); + taskId = organization.createTask({ + orgId, + title: "Draft launch checklist", + description: "Prepare an auditable checklist for release readiness.", + status: TASK_STATUS.IN_PROGRESS, + assigneeAgentId: agentId, + risk: RISK_LEVEL.MEDIUM, + projectId: null, + goalId: null, + workspaceId: null, + priority: TASK_PRIORITY.HIGH, + requiresFilesystem: false, + }); + }); + + await expect(execution.runTask({ orgId, taskId })).rejects.toThrow("The usage limit has been reached"); + + // Parked: the worker skips the provider's queued runs and the heartbeat enqueues nothing new. + expect(execution.cooledProviderIds()).toEqual([providerId]); + expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toEqual([]); + + // Cooldown over: the provider resumes and the task re-enqueues — the failure spent no budget. + nowMs += 11 * 60_000; + expect(execution.cooledProviderIds()).toEqual([]); + const enqueued = connection.transaction(() => execution.enqueueHeartbeatRuns()); + expect(enqueued).toHaveLength(1); + expect(connection.db.prepare("select task_id, state from task_runs where id = ?").get(enqueued[0])) + .toMatchObject({ task_id: taskId, state: "queued" }); + connection.close(); + }); }); diff --git a/tests/unit/task-queue-worker.test.ts b/tests/unit/task-queue-worker.test.ts index 74636e4f..52413d50 100644 --- a/tests/unit/task-queue-worker.test.ts +++ b/tests/unit/task-queue-worker.test.ts @@ -50,4 +50,29 @@ describe("task queue worker claim", () => { expect(states).toEqual(["running", "running"]); connection.close(); }); + + test("leaves runs of a cooling-down provider queued instead of claiming them", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let agentId = ""; + let providerId = ""; + let taskId = ""; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "Acme", employeeName: "Dana", mission: "m", goal: "g" }); + providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + agentId = organization.createAgent({ orgId, name: "CEO", role: "CEO", department: "Executive", modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 25000, managerAgentId: null }); + taskId = organization.createTask({ + orgId, title: "Kickoff", description: "Start.", status: TASK_STATUS.BACKLOG, assigneeAgentId: agentId, + risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId: null, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: false, + }); + }); + const runId = seedQueuedRun(connection, orgId, taskId, agentId, providerId); + + expect(claimNextQueuedRun(connection, [providerId])).toBeNull(); + expect(connection.db.prepare("select state from task_runs where id = ?").get(runId)).toEqual({ state: "queued" }); + + expect(claimNextQueuedRun(connection, [])?.id).toBe(runId); + connection.close(); + }); });