diff --git a/eslint.config.mjs b/eslint.config.mjs index d353b11d..f8f168af 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -5,7 +5,9 @@ import reactRefresh from "eslint-plugin-react-refresh"; import tseslint from "typescript-eslint"; export default tseslint.config( - { ignores: ["out", "release", "node_modules", "design-source"] }, + // "landing" is a self-contained Next.js sub-project with its own tooling; + // "built-in skills" is bundled prompt content, not application code. + { ignores: ["out", "release", "node_modules", "design-source", "landing", "built-in skills"] }, js.configs.recommended, ...tseslint.configs.recommended, { diff --git a/src/main/services/provider-background-service.ts b/src/main/services/provider-background-service.ts index c98117b0..174171c4 100644 --- a/src/main/services/provider-background-service.ts +++ b/src/main/services/provider-background-service.ts @@ -9,7 +9,7 @@ import { type ProviderAuthType, type ProviderExecutionMode, } from "../../shared/domain"; -import { currentProviderUsageWindow, type ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; +import { currentProviderUsageWindow, PROVIDER_MONTHLY_LIMIT_MESSAGE, type ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; import { AuditService } from "./audit-service"; import { AuthBrokerService, type ProviderSecretVault } from "./auth-broker-service"; import { OperationsService } from "./operations-service"; @@ -238,7 +238,7 @@ export class ProviderBackgroundService { where org_id = ? and provider_id = ? and window_start = ? `).get(orgId, provider.id, window.start) as { request_count: number } | undefined; const requestCount = usage?.request_count ?? 0; - if (requestCount >= provider.monthly_request_limit) throw new Error("Provider monthly request limit has been reached."); + if (requestCount >= provider.monthly_request_limit) throw new Error(PROVIDER_MONTHLY_LIMIT_MESSAGE); } private normalizeWorkspaceId(orgId: string, workspaceId: string | null): string | null { diff --git a/src/main/services/provider-response-runner.ts b/src/main/services/provider-response-runner.ts index 22e81771..3b183464 100644 --- a/src/main/services/provider-response-runner.ts +++ b/src/main/services/provider-response-runner.ts @@ -2,7 +2,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { CancelProviderResponseInput, CreateProviderResponseInput, StreamProviderResponseInput } from "../../shared/commands"; import { FEATURE_FLAG_KEY, PROVIDER_AUTH_TYPE, PROVIDER_DATA_POLICY, PROVIDER_EXECUTION_MODE, PROVIDER_REQUEST_STATE, type ProviderExecutionMode } from "../../shared/domain"; import type { ProviderStreamRendererEvent } from "../../shared/provider-request-domain"; -import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain"; +import { currentProviderUsageWindow, PROVIDER_MONTHLY_LIMIT_MESSAGE } from "../../shared/provider-runtime-domain"; import type { AuditService } from "./audit-service"; import type { AuthBrokerService } from "./auth-broker-service"; import type { OperationsService } from "./operations-service"; @@ -331,7 +331,7 @@ export class ProviderResponseRunner { where org_id = ? and provider_id = ? and window_start = ? `).get(orgId, provider.id, window.start) as { request_count: number } | undefined; const requestCount = usage?.request_count ?? 0; - if (requestCount >= provider.monthly_request_limit) throw new Error("Provider monthly request limit has been reached."); + if (requestCount >= provider.monthly_request_limit) throw new Error(PROVIDER_MONTHLY_LIMIT_MESSAGE); } private normalizeWorkspaceId(orgId: string, workspaceId: string | null): string | null { diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 9e1aaa28..408f72f4 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -1,6 +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 { createId } from "../../shared/ids"; import { AuditService } from "./audit-service"; import { ProviderService } from "./provider-service"; @@ -53,6 +54,8 @@ export class TaskExecutionService { private readonly audit: AuditService; private readonly organization: OrganizationService; private readonly chat: ChatService; + /** Providers currently parked on their monthly quota, for transition logging. */ + private quotaBlockedProviders = new Map(); constructor( private readonly db: DatabaseSync, @@ -296,20 +299,31 @@ export class TaskExecutionService { * heartbeat-enabled agent with an enabled provider that has not run yet, so * 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. */ enqueueHeartbeatRuns(): string[] { + const window = currentProviderUsageWindow(); const rows = this.db.prepare(` select t.id as task_id, t.org_id as org_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 + left join provider_usage_windows u + on u.org_id = p.org_id and u.provider_id = p.id and u.window_start = ? where t.status in ('backlog', 'in_progress') and a.heartbeat_enabled = 1 and a.lifecycle = 'active' 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) < 3 - `).all() as Array<{ task_id: string; org_id: string }>; + 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 }>; + this.observeQuotaBlocks(window.start); const created: string[] = []; for (const row of rows) { try { @@ -321,6 +335,33 @@ export class TaskExecutionService { return created; } + /** + * Logs monthly-quota transitions (blocked <-> unblocked) exactly once per + * state change so a parked queue is diagnosable from the main-process log + * without spamming a line every 5s tick. + */ + private observeQuotaBlocks(windowStart: string): void { + const rows = this.db.prepare(` + select p.id as id, p.label as label + from providers p + join provider_usage_windows u + on u.org_id = p.org_id and u.provider_id = p.id and u.window_start = ? + where p.enabled = 1 and u.request_count >= p.monthly_request_limit + `).all(windowStart) as Array<{ id: string; label: string }>; + const current = new Map(rows.map((row) => [row.id, row.label])); + for (const [id, label] of current) { + if (!this.quotaBlockedProviders.has(id)) { + console.warn(`[task-queue] provider "${label}" hit its monthly request limit; its tasks are parked until the limit is raised or the window resets.`); + } + } + for (const [id, label] of this.quotaBlockedProviders) { + if (!current.has(id)) { + console.info(`[task-queue] provider "${label}" is under its monthly request limit again; parked tasks resume.`); + } + } + this.quotaBlockedProviders = current; + } + private enqueueRun(orgId: string, taskId: string): string { const task = this.requireTask(orgId, taskId); // Autonomous execution honours the same gate as a manual run: a high-risk or diff --git a/src/shared/provider-runtime-domain.ts b/src/shared/provider-runtime-domain.ts index 32b2b107..c915f506 100644 --- a/src/shared/provider-runtime-domain.ts +++ b/src/shared/provider-runtime-domain.ts @@ -29,6 +29,15 @@ export interface ProviderRateLimitObservationRecord extends ProviderObservedRate observedAt: string; } +/** + * Exact message recorded on runs blocked by a provider's monthly usage window. + * The heartbeat scheduler matches task_runs.error_message against this string + * so quota-blocked attempts never consume a task's retry budget — the text is + * a cross-module contract and also matches rows written by earlier builds + * that inlined it. + */ +export const PROVIDER_MONTHLY_LIMIT_MESSAGE = "Provider monthly request limit has been reached."; + 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/task-execution-service.test.ts b/tests/unit/task-execution-service.test.ts index 2a07444d..de0865d9 100644 --- a/tests/unit/task-execution-service.test.ts +++ b/tests/unit/task-execution-service.test.ts @@ -12,7 +12,8 @@ import type { AgentTurnResult, ProviderToolTurnInput } from "../../src/main/serv import { SkillService } from "../../src/main/services/skill-service"; import { SnapshotService } from "../../src/main/services/snapshot-service"; import { TaskExecutionService } from "../../src/main/services/task-execution-service"; -import { APPROVAL_STATE, RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../src/shared/domain"; +import { APPROVAL_STATE, PROVIDER_DATA_POLICY, RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../src/shared/domain"; +import { currentProviderUsageWindow, PROVIDER_MONTHLY_LIMIT_MESSAGE } from "../../src/shared/provider-runtime-domain"; function createConnection(): DatabaseConnection { return new DatabaseConnection(join(tmpdir(), `agentcompany-new-task-execution-${randomUUID()}.db`)); @@ -378,4 +379,165 @@ describe("task execution service", () => { expect(connection.db.prepare("select count(*) as count from task_runs").get()).toEqual({ count: 0 }); connection.close(); }); + + test("parks tasks while their provider is over its monthly quota and resumes once the limit is raised", () => { + 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 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" }); + providers.updatePolicy({ + orgId, + providerId, + dataPolicy: PROVIDER_DATA_POLICY.STANDARD, + backgroundEnabled: false, + monthlyRequestLimit: 5, + workspaceScope: "organization", + }); + 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, + }); + const window = currentProviderUsageWindow(); + connection.db.prepare(` + insert into provider_usage_windows(id, org_id, provider_id, window_start, window_end, request_count, input_char_count, output_char_count, created_at, updated_at) + values (?, ?, ?, ?, ?, 5, 0, 0, ?, ?) + `).run("puw_quota_test", orgId, providerId, window.start, window.end, window.start, window.start); + }); + + // At the limit: the task is parked, not enqueued into a doomed run. + expect(execution.enqueueHeartbeatRuns()).toEqual([]); + expect(connection.db.prepare("select count(*) as count from task_runs").get()).toEqual({ count: 0 }); + + // Raising the limit (what Settings saves) revives the queue on the next tick. + connection.transaction(() => providers.updatePolicy({ + orgId, + providerId, + dataPolicy: PROVIDER_DATA_POLICY.STANDARD, + backgroundEnabled: false, + monthlyRequestLimit: 50, + workspaceScope: "organization", + })); + 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("quota-blocked failures never consume the retry budget, while genuine failures still do", () => { + 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 filled quota window: three failed runs, all quota-blocked. + 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_quota_fail_${index}`, orgId, taskId, agentId, providerId, `ctx_quota_${index}`, serviceIdentityId, now, now, now, PROVIDER_MONTHLY_LIMIT_MESSAGE); + } + }); + + // Quota failures are excluded from the budget: the task revives without any data surgery. + const enqueued = connection.transaction(() => execution.enqueueHeartbeatRuns()); + expect(enqueued).toHaveLength(1); + + // Genuine failures still cap at three attempts. + connection.transaction(() => { + connection.db.prepare("delete from task_runs where state = 'queued'").run(); + const serviceIdentityId = (connection.db.prepare("select id from service_identities where org_id = ? and agent_id = ?") + .get(orgId, agentId) as { id: string }).id; + const now = new Date().toISOString(); + 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', ?, ?, ?, ?) + `); + for (const index of [1, 2, 3]) { + insertFailedRun.run(`run_real_fail_${index}`, orgId, taskId, agentId, providerId, `ctx_real_${index}`, serviceIdentityId, now, now, now, "provider rejected task"); + } + }); + expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toEqual([]); + connection.close(); + }); });