Resume parked work when a provider's monthly limit is raised
The heartbeat scheduler enqueued runs even while a provider's monthly usage window was full, so every task burned its 3-attempt retry budget on instant quota failures within seconds and stayed dead forever - raising the limit (or restarting) never revived anything. - Park tasks (skip enqueue) while their provider is at its monthly_request_limit instead of creating runs doomed to fail - Exclude quota-blocked failures from the 3-attempt retry budget, retroactively reviving tasks abandoned by past filled windows - Share the gate message as PROVIDER_MONTHLY_LIMIT_MESSAGE and log quota-block transitions once per state change for diagnosability - Ignore landing/ (self-contained Next.js sub-project) and built-in skills/ (prompt content) in eslint so verify:ci lints only application code
This commit is contained in:
1 parent
8e01317a46
commit
349547a0de
6 files changed
+222
-8
No files matched your search
+3
-1
@@ -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,
|
||||
{
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<string, string>();
|
||||
|
||||
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
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user