Q45: the queue claims higher-priority work first
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8b521ef5d9
commit
3bc341d757
2 files changed
+37
-19
No files matched your search
@@ -37,35 +37,25 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean {
|
|||||||
return row?.autorun_paused === 1;
|
return row?.autorun_paused === 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/** Claims the next queued run atomically: human wakes first, then task priority, then age; skips cooling providers, busy agents and busy workspace writers. */
|
||||||
* Atomically claims the next 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.
|
|
||||||
*
|
|
||||||
* Claim order is interactive-first: a wake triggered by a human message
|
|
||||||
* (the founder DMing the CEO, a board member asking in a channel) means a
|
|
||||||
* person is waiting at the keyboard, so it outranks any amount of queued
|
|
||||||
* autonomous work; within each lane, first come first served. Non-chat runs
|
|
||||||
* carry no sourceSenderType and fall into the autonomous lane. An agent
|
|
||||||
* runs one session at a time: its queued runs wait while one is running.
|
|
||||||
* A workspace takes one writing run at a time: write-granted runs wait.
|
|
||||||
*/
|
|
||||||
export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = [], lane: ClaimLane = "any"): QueuedRun | null {
|
export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = [], lane: ClaimLane = "any"): QueuedRun | null {
|
||||||
return connection.transaction(() => {
|
return connection.transaction(() => {
|
||||||
const exclusion = excludedProviderIds.length > 0
|
const exclusion = excludedProviderIds.length > 0
|
||||||
? `and provider_id not in (${excludedProviderIds.map(() => "?").join(", ")})`
|
? `and task_runs.provider_id not in (${excludedProviderIds.map(() => "?").join(", ")})`
|
||||||
: "";
|
: "";
|
||||||
const laneFilter = lane === "interactive"
|
const laneFilter = lane === "interactive"
|
||||||
? "and json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board')"
|
? "and json_extract(task_runs.secret_scope_json, '$.sourceSenderType') in ('employee', 'board')"
|
||||||
: "";
|
: "";
|
||||||
const candidates = connection.db
|
const candidates = connection.db
|
||||||
.prepare(`
|
.prepare(`
|
||||||
select id, org_id, provider_id, task_id from task_runs where state = 'queued' ${exclusion} ${laneFilter}
|
select task_runs.id, task_runs.org_id, task_runs.provider_id, task_runs.task_id
|
||||||
|
from task_runs join tasks on tasks.id = task_runs.task_id
|
||||||
|
where task_runs.state = 'queued' ${exclusion} ${laneFilter}
|
||||||
and not exists (select 1 from task_runs busy where busy.state = 'running' and busy.agent_id = task_runs.agent_id)
|
and not exists (select 1 from task_runs busy where busy.state = 'running' and busy.agent_id = task_runs.agent_id)
|
||||||
order by
|
order by
|
||||||
case when json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end asc,
|
case when json_extract(task_runs.secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end asc,
|
||||||
created_at asc
|
case tasks.priority when 'high' then 0 when 'normal' then 1 else 2 end asc,
|
||||||
|
task_runs.created_at asc
|
||||||
`)
|
`)
|
||||||
.all(...excludedProviderIds) as unknown as Array<QueuedRun & { task_id: string }>;
|
.all(...excludedProviderIds) as unknown as Array<QueuedRun & { task_id: string }>;
|
||||||
const candidate = candidates.find((run) => runningWorkspaceWriter(connection.db, run.task_id) === null);
|
const candidate = candidates.find((run) => runningWorkspaceWriter(connection.db, run.task_id) === null);
|
||||||
|
|||||||
@@ -55,6 +55,34 @@ describe("task queue worker claim", () => {
|
|||||||
connection.close();
|
connection.close();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("within a lane, higher-priority tasks are claimed first, then oldest first", () => {
|
||||||
|
const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`));
|
||||||
|
const organization = new OrganizationService(connection.db);
|
||||||
|
let orgId = "";
|
||||||
|
let providerId = "";
|
||||||
|
const tasks: Record<string, { taskId: string; agentId: string }> = {};
|
||||||
|
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;
|
||||||
|
for (const [name, priority] of [["low", TASK_PRIORITY.LOW], ["normal", TASK_PRIORITY.NORMAL], ["high", TASK_PRIORITY.HIGH]] as const) {
|
||||||
|
const agentId = organization.createAgent({ orgId, name: `Agent ${name}`, role: "Staff", department: "Operations", modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: null });
|
||||||
|
const taskId = organization.createTask({
|
||||||
|
orgId, title: name, description: "d", status: TASK_STATUS.BACKLOG, assigneeAgentId: agentId,
|
||||||
|
risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId: null, priority, requiresFilesystem: false,
|
||||||
|
});
|
||||||
|
tasks[name] = { taskId, agentId };
|
||||||
|
}
|
||||||
|
});
|
||||||
|
const runs: Record<string, string> = {};
|
||||||
|
for (const [index, name] of ["low", "normal", "high"].entries()) {
|
||||||
|
runs[name] = seedQueuedRun(connection, orgId, tasks[name].taskId, tasks[name].agentId, providerId);
|
||||||
|
connection.db.prepare("update task_runs set created_at = ? where id = ?").run(`2026-01-0${index + 1}T00:00:00.000Z`, runs[name]);
|
||||||
|
}
|
||||||
|
expect([claimNextQueuedRun(connection)?.id, claimNextQueuedRun(connection)?.id, claimNextQueuedRun(connection)?.id])
|
||||||
|
.toEqual([runs.high, runs.normal, runs.low]);
|
||||||
|
connection.close();
|
||||||
|
});
|
||||||
|
|
||||||
test("leaves runs of a cooling-down provider queued instead of claiming them", () => {
|
test("leaves runs of a cooling-down provider queued instead of claiming them", () => {
|
||||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`));
|
const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`));
|
||||||
const organization = new OrganizationService(connection.db);
|
const organization = new OrganizationService(connection.db);
|
||||||
|
|||||||
Reference in new issue
Block a user