From 3bc341d7573bcbb6b6e1ec391d27376e8d19f5b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 08:02:43 -0400 Subject: [PATCH] Q45: the queue claims higher-priority work first Co-Authored-By: Claude Opus 5.5 --- src/main/app/task-queue-worker.ts | 28 +++++++++------------------- tests/unit/task-queue-worker.test.ts | 28 ++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 19 deletions(-) diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index 0b23e03d..02aabfa0 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -37,35 +37,25 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean { return row?.autorun_paused === 1; } -/** - * 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. - */ +/** Claims the next queued run atomically: human wakes first, then task priority, then age; skips cooling providers, busy agents and busy workspace writers. */ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = [], lane: ClaimLane = "any"): QueuedRun | null { return connection.transaction(() => { 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" - ? "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 .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) order by - case when json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end asc, - created_at asc + case when json_extract(task_runs.secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end 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; const candidate = candidates.find((run) => runningWorkspaceWriter(connection.db, run.task_id) === null); diff --git a/tests/unit/task-queue-worker.test.ts b/tests/unit/task-queue-worker.test.ts index e5f1667d..9d5087be 100644 --- a/tests/unit/task-queue-worker.test.ts +++ b/tests/unit/task-queue-worker.test.ts @@ -55,6 +55,34 @@ describe("task queue worker claim", () => { 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 = {}; + 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 = {}; + 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", () => { const connection = new DatabaseConnection(join(tmpdir(), `ac-queue-${randomUUID()}.db`)); const organization = new OrganizationService(connection.db);