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.
This commit is contained in:
1 parent
39aa292d09
commit
aa4cfc056d
3 files changed
+120
-4
No files matched your search
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user