From aa4cfc056d261473e27e04f69cc61db776c32f56 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Tue, 14 Jul 2026 20:46:51 -0400 Subject: [PATCH] Human-triggered wakes claim before queued autonomous work Wake runs stamp sourceSenderType into their secret scope, and the claim query orders interactive-first: a wake born from an employee or board message means a person is waiting at the keyboard, so it outranks any autonomous backlog; within each lane FIFO holds. Non-chat runs carry no stamp and stay in the autonomous lane. With wake expiry this closes the live incident: the founder's CEO DM now claims on the next tick instead of 13th in line. Red-green tested. --- src/main/app/task-queue-worker.ts | 16 ++- src/main/services/chat-agent-wake-service.ts | 8 +- tests/unit/claim-priority.test.ts | 100 +++++++++++++++++++ 3 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 tests/unit/claim-priority.test.ts diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index f8c47b61..de3c5fcf 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -27,10 +27,16 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean { } /** - * Atomically claims the oldest queued task_run by transitioning it + * 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. */ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = []): QueuedRun | null { return connection.transaction(() => { @@ -38,7 +44,13 @@ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProvi ? `and provider_id not in (${excludedProviderIds.map(() => "?").join(", ")})` : ""; const row = connection.db - .prepare(`select id, org_id from task_runs where state = 'queued' ${exclusion} order by created_at asc limit 1`) + .prepare(` + select id, org_id from task_runs where state = 'queued' ${exclusion} + order by + case when json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end asc, + created_at asc + limit 1 + `) .get(...excludedProviderIds) as QueuedRun | undefined; if (!row) return null; const result = connection.db diff --git a/src/main/services/chat-agent-wake-service.ts b/src/main/services/chat-agent-wake-service.ts index 602b2f8c..ff5cad2f 100644 --- a/src/main/services/chat-agent-wake-service.ts +++ b/src/main/services/chat-agent-wake-service.ts @@ -95,7 +95,7 @@ export class ChatAgentWakeService { requiresFilesystem: false, origin: "chat_wake", }); - const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, now }); + const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, senderType: input.senderType, now }); this.recordPresence(agent.id, { command: "mention", args: "" }, now); return { targetId: agent.id, taskId, runId }; } @@ -118,7 +118,7 @@ export class ChatAgentWakeService { requiresFilesystem: false, origin: "chat_wake", }); - const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, now }); + const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, senderType: input.senderType, now }); this.recordPresence(agent.id, slash, now); return { targetId: agent.id, taskId, runId }; } @@ -178,6 +178,7 @@ export class ChatAgentWakeService { workspace: WorkspaceContext | null; messageId: string; threadId: string; + senderType: SendMessageInput["senderType"]; now: string; }): string { const providerId = input.agent.model_provider_id; @@ -199,6 +200,9 @@ export class ChatAgentWakeService { credentialAccess: "provider_runtime_only", sourceMessageId: input.messageId, sourceThreadId: input.threadId, + // Who spoke decides scheduling: the queue claims human-triggered wakes + // before autonomous work (see claimNextQueuedRun). + sourceSenderType: input.senderType, }); this.db.prepare(` insert into task_runs( diff --git a/tests/unit/claim-priority.test.ts b/tests/unit/claim-priority.test.ts new file mode 100644 index 00000000..523631f5 --- /dev/null +++ b/tests/unit/claim-priority.test.ts @@ -0,0 +1,100 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OrganizationService } from "../../src/main/services/organization-service"; +import { ChatService } from "../../src/main/services/chat-service"; +import { dispatchAgentTool } from "../../src/main/services/agent-tools"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; + +interface Harness { + connection: DatabaseConnection; + organization: OrganizationService; + chat: ChatService; + orgId: string; + employeeId: string; + ceoId: string; + danaId: string; +} + +function harness(): Harness { + const connection = new DatabaseConnection(join(tmpdir(), `ac-claim-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let ceoId = ""; + let danaId = ""; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "OrbitOps", employeeName: "Yang", mission: "m", goal: "g" }); + const providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + connection.db.prepare("update providers set enabled = 1 where id = ? and org_id = ?").run(providerId, orgId); + ceoId = organization.createAgent({ + orgId, name: "CEO", role: "CEO", department: "Executive", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 25000, managerAgentId: null, roleLevel: "c_suite", + }); + danaId = organization.createAgent({ + orgId, name: "Dana", role: "Engineer", department: "Engineering", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: ceoId, + }); + }); + const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + return { connection, organization, chat: new ChatService(connection.db), orgId, employeeId, ceoId, danaId }; +} + +describe("queue claim priority", () => { + test("a human-triggered wake claims before older autonomous work", () => { + const h = harness(); + const deps = { db: h.connection.db, organization: h.organization, chat: h.chat }; + + // T0: agent-triggered wake (CEO DMs Dana) — autonomous conversation. + dispatchAgentTool(deps, { + orgId: h.orgId, + agent: { id: h.ceoId, modelProviderId: null, modelId: null }, + taskProjectId: null, taskGoalId: null, workspace: null, + }, { id: `call_${randomUUID()}`, name: "send_message", arguments: { to: "Dana", body: "Status?" } }); + const danaRun = h.connection.db.prepare( + "select id from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention'", + ).get(h.orgId, h.danaId) as { id: string }; + // Make it decisively older than what follows. + h.connection.db.prepare("update task_runs set created_at = ? where id = ?") + .run(new Date(Date.now() - 60_000).toISOString(), danaRun.id); + + // T1: the founder DMs the CEO — a human is waiting at the keyboard. + const spaceId = h.chat.ensureDirectSpace(h.orgId, { type: "employee", id: h.employeeId }, { type: "agent", id: h.ceoId }); + h.chat.sendMessage({ + orgId: h.orgId, spaceId, threadId: null, senderType: "employee", senderId: h.employeeId, + bodyMd: "How is the project going?", attachmentFileIds: [], + }); + const ceoRun = h.connection.db.prepare( + "select id from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention'", + ).get(h.orgId, h.ceoId) as { id: string }; + + // The founder's newer wake outranks the older autonomous one. + expect(claimNextQueuedRun(h.connection)?.id).toBe(ceoRun.id); + expect(claimNextQueuedRun(h.connection)?.id).toBe(danaRun.id); + expect(claimNextQueuedRun(h.connection)).toBeNull(); + h.connection.close(); + }); + + test("among human-triggered wakes, order stays first come first served", () => { + const h = harness(); + const wakeByFounder = (agentId: string): string => { + const spaceId = h.chat.ensureDirectSpace(h.orgId, { type: "employee", id: h.employeeId }, { type: "agent", id: agentId }); + h.chat.sendMessage({ + orgId: h.orgId, spaceId, threadId: null, senderType: "employee", senderId: h.employeeId, + bodyMd: "ping", attachmentFileIds: [], + }); + return (h.connection.db.prepare( + "select id from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention'", + ).get(h.orgId, agentId) as { id: string }).id; + }; + const first = wakeByFounder(h.danaId); + h.connection.db.prepare("update task_runs set created_at = ? where id = ?") + .run(new Date(Date.now() - 30_000).toISOString(), first); + const second = wakeByFounder(h.ceoId); + + expect(claimNextQueuedRun(h.connection)?.id).toBe(first); + expect(claimNextQueuedRun(h.connection)?.id).toBe(second); + h.connection.close(); + }); +});