diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index de3c5fcf..7053d4f5 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -12,13 +12,23 @@ export interface TaskQueueWorkerInput { connection: DatabaseConnection; mainWindow: BrowserWindow; intervalMs?: number; + /** Injectable execution service so tests can script the provider. */ + taskExecution?: TaskExecutionService; } export interface QueuedRun { id: string; org_id: string; + provider_id: string | null; } +/** + * Which queued runs a claim may take: "interactive" claims only wakes born + * from a human message (the founder is waiting at the keyboard); "any" claims + * everything, interactive-first. + */ +export type ClaimLane = "interactive" | "any"; + /** True while the operator has pulled the global autonomous handbrake. */ export function isAutonomousPaused(connection: DatabaseConnection): boolean { const row = connection.db.prepare("select autorun_paused from app_state where id = 1") @@ -38,14 +48,17 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean { * 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 { +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(", ")})` : ""; + const laneFilter = lane === "interactive" + ? "and json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board')" + : ""; const row = connection.db .prepare(` - select id, org_id from task_runs where state = 'queued' ${exclusion} + select id, org_id, provider_id from task_runs where state = 'queued' ${exclusion} ${laneFilter} order by case when json_extract(secret_scope_json, '$.sourceSenderType') in ('employee', 'board') then 0 else 1 end asc, created_at asc @@ -61,10 +74,13 @@ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProvi } /** - * Autonomous executor: every tick it claims the oldest `queued` task_run (the - * bootstrap CEO run, plus any future heartbeat-created runs) and executes its - * real agent loop. The claim is a single transaction (`queued` -> `running`, - * verified by row changes) so a run is never executed twice. + * Autonomous executor with two execution slots. The autonomous slot grinds + * the queue one run at a time (bootstrap kickoff, heartbeat work, agent + * chatter). The interactive slot is reserved capacity for the human: a wake + * born from an employee or board message executes immediately, concurrently + * with whatever long autonomous run is in flight — the founder never queues + * behind batch work. Each claim is a single transaction (`queued` -> + * `running`, verified by row changes) so a run is never executed twice. */ export function startTaskQueueWorker(input: TaskQueueWorkerInput): () => void { const notifyDataChanged = () => { @@ -72,57 +88,58 @@ export function startTaskQueueWorker(input: TaskQueueWorkerInput): () => void { input.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); } }; - const taskExecution = new TaskExecutionService(input.connection.db, undefined, undefined, notifyDataChanged); + const taskExecution = input.taskExecution ?? new TaskExecutionService(input.connection.db, undefined, undefined, notifyDataChanged); const organization = new OrganizationService(input.connection.db); const intervalMs = input.intervalMs ?? DEFAULT_INTERVAL_MS; - let running = false; + const inFlight = { interactive: false, autonomous: false }; let wasPaused = false; - const tick = async (): Promise => { - if (running) return; - running = true; - try { - // The operator's global pause parks the whole engine: nothing new is - // enqueued or claimed until they resume. Queued work simply waits. - const paused = isAutonomousPaused(input.connection); - if (paused !== wasPaused) { - console.info(paused ? "[task-queue] paused by the operator; no new runs start." : "[task-queue] resumed by the operator."); - wasPaused = paused; - } - if (paused) return; - // Stale conversation wakes die before anything is claimed: replying to - // a thread hours later is noise, and a backlog of them would starve - // fresh messages (the founder's included) for the whole grind. - const expired = input.connection.transaction(() => expireStaleChatWakes(input.connection.db, new Date().toISOString())); - if (expired > 0) { - console.info(`[task-queue] expired ${expired} stale chat wake run(s); newer messages re-wake their agents.`); + const dispatch = (slot: "interactive" | "autonomous"): void => { + if (inFlight[slot]) return; + const claimed = slot === "interactive" + ? claimNextQueuedRun(input.connection, taskExecution.cooledProviderIds(), "interactive") + : claimNextQueuedRun(input.connection, taskExecution.cooledProviderIds(), "any"); + if (!claimed) return; + inFlight[slot] = true; + notifyDataChanged(); + void taskExecution.executeRun(claimed.org_id, claimed.id) + .catch(() => { + // executeRun records the failure on the run row; the UI reflects it. + }) + .finally(() => { + inFlight[slot] = false; notifyDataChanged(); - } - // Goal checkpoints first, so a review task created this tick is - // eligible for the same tick's heartbeat enqueue below. - const reviews = input.connection.transaction(() => spawnGoalReviews(input.connection.db, organization)); - if (reviews.length > 0) notifyDataChanged(); - const enqueued = input.connection.transaction(() => taskExecution.enqueueHeartbeatRuns()); - const claimed = claimNextQueuedRun(input.connection, taskExecution.cooledProviderIds()); - if (!claimed) { - if (enqueued.length > 0) notifyDataChanged(); - return; - } - notifyDataChanged(); - try { - await taskExecution.executeRun(claimed.org_id, claimed.id); - } catch { - // executeRun records the failure on the run row; the UI reflects it below. - } - notifyDataChanged(); - } finally { - running = false; - } + }); }; - const interval = setInterval(() => { - void tick(); - }, intervalMs); - void tick(); + const tick = (): void => { + // The operator's global pause parks the whole engine: nothing new is + // enqueued or claimed until they resume. Queued work simply waits. + const paused = isAutonomousPaused(input.connection); + if (paused !== wasPaused) { + console.info(paused ? "[task-queue] paused by the operator; no new runs start." : "[task-queue] resumed by the operator."); + wasPaused = paused; + } + if (paused) return; + // Stale conversation wakes die before anything is claimed: replying to + // a thread hours later is noise, and a backlog of them would starve + // fresh messages (the founder's included) for the whole grind. + const expired = input.connection.transaction(() => expireStaleChatWakes(input.connection.db, new Date().toISOString())); + if (expired > 0) { + console.info(`[task-queue] expired ${expired} stale chat wake run(s); newer messages re-wake their agents.`); + notifyDataChanged(); + } + // Goal checkpoints first, so a review task created this tick is + // eligible for the same tick's heartbeat enqueue below. + const reviews = input.connection.transaction(() => spawnGoalReviews(input.connection.db, organization)); + if (reviews.length > 0) notifyDataChanged(); + const enqueued = input.connection.transaction(() => taskExecution.enqueueHeartbeatRuns()); + if (enqueued.length > 0) notifyDataChanged(); + dispatch("interactive"); + dispatch("autonomous"); + }; + + const interval = setInterval(tick, intervalMs); + tick(); return () => clearInterval(interval); } diff --git a/tests/integration/interactive-lane.test.ts b/tests/integration/interactive-lane.test.ts new file mode 100644 index 00000000..93ad6d82 --- /dev/null +++ b/tests/integration/interactive-lane.test.ts @@ -0,0 +1,129 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { BrowserWindow } from "electron"; +import { describe, expect, test, vi } 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 { ProviderService } from "../../src/main/services/provider-service"; +import { providerAdapters } from "../../src/main/services/provider-adapters"; +import { TaskExecutionService } from "../../src/main/services/task-execution-service"; +import { dispatchAgentTool } from "../../src/main/services/agent-tools"; +import { startTaskQueueWorker } from "../../src/main/app/task-queue-worker"; +import type { AgentTurnResult, ProviderToolTurnInput, ToolCall } from "../../src/main/services/provider-tool-turn"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +function call(name: string, args: Record): ToolCall { + return { id: `call_${randomUUID()}`, name, arguments: args }; +} + +function turn(text: string, toolCalls: ToolCall[] = []): AgentTurnResult { + return { text, toolCalls, rawResponseJson: "{}" }; +} + +interface Deferred { + promise: Promise; + resolve: (value: AgentTurnResult) => void; +} + +function deferred(): Deferred { + let resolve!: (value: AgentTurnResult) => void; + const promise = new Promise((r) => { resolve = r; }); + return { promise, resolve }; +} + +describe("interactive execution lane", () => { + test("the founder's wake executes concurrently instead of queueing behind a long autonomous run", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-lane-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let ceoId = ""; + let danaId = ""; + let providerId = ""; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "OrbitOps", employeeName: "Yang", 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; + 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; + const chat = new ChatService(connection.db); + + // Dana's autonomous run blocks on this gate — the "30-minute engineering + // run" in miniature. The CEO's interactive run completes normally. + const danaGate = deferred(); + let ceoReplied = false; + const script = async (input: ProviderToolTurnInput): Promise => { + const prompt = input.messages.find((message) => message.role === "user")?.text ?? ""; + if (prompt.includes("Grind through the migration backlog")) return danaGate.promise; + const threadId = /thread_id="([^"]+)"/.exec(prompt)?.[1]; + if (prompt.includes("Add headcount") && threadId && !ceoReplied) { + ceoReplied = true; + return turn("Answering.", [call("send_message", { to: "Yang", body: "On it — hiring plan coming today.", thread_id: threadId })]); + } + return turn("done"); + }; + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, script); + providers.saveCredential({ orgId, providerId, secret: "sk-test-key", model: "gpt-5.4" }); + const taskExecution = new TaskExecutionService(connection.db, providers); + + // Older autonomous work: the CEO DMs Dana, waking her. + dispatchAgentTool({ db: connection.db, organization, chat }, { + orgId, agent: { id: ceoId, modelProviderId: null, modelId: null }, + taskProjectId: null, taskGoalId: null, workspace: null, + }, call("send_message", { to: "Dana", body: "Grind through the migration backlog." })); + + // Newer interactive work: the founder DMs the CEO. + const spaceId = chat.ensureDirectSpace(orgId, { type: "employee", id: employeeId }, { type: "agent", id: ceoId }); + chat.sendMessage({ + orgId, spaceId, threadId: null, senderType: "employee", senderId: employeeId, + bodyMd: "Add headcount wherever you need — your call.", attachmentFileIds: [], + }); + + const stopWorker = startTaskQueueWorker({ + connection, + mainWindow: { isDestroyed: () => true } as unknown as BrowserWindow, + intervalMs: 25, + taskExecution, + }); + try { + // The CEO's reply lands while Dana's run is still in flight: the + // interactive slot did not wait for the autonomous slot. + await vi.waitFor(() => { + const reply = connection.db.prepare( + "select id from chat_messages where org_id = ? and sender_type = 'agent' and sender_id = ? and space_id = ?", + ).get(orgId, ceoId, spaceId); + expect(reply).toBeTruthy(); + }, { timeout: 3000, interval: 50 }); + + const danaRun = connection.db.prepare( + "select state from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention'", + ).get(orgId, danaId) as { state: string }; + expect(danaRun.state).toBe("running"); + + // Release the long run; the autonomous slot completes it. + danaGate.resolve(turn("Backlog ground through.")); + await vi.waitFor(() => { + const state = (connection.db.prepare( + "select state from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention'", + ).get(orgId, danaId) as { state: string }).state; + expect(state).toBe("completed"); + }, { timeout: 3000, interval: 50 }); + } finally { + stopWorker(); + connection.close(); + } + }); +});