From 72eb645067e066292bcfdb0dd753597950689a72 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 03:11:39 -0400 Subject: [PATCH] Q7: an agent runs one session at a time Claims skip agents that already have a running run, manual runs refuse a busy agent, a run whose task was reassigned fails before executing, and tasks with a run in progress cannot be deleted. Co-Authored-By: Claude Opus 5.5 --- src/main/app/task-queue-worker.ts | 4 +- src/main/services/organization-service.ts | 3 ++ src/main/services/task-execution-service.ts | 46 +++++++++++++++----- tests/integration/task-context-scope.test.ts | 45 +++++++++++++++++++ tests/unit/claim-priority.test.ts | 38 ++++++++++++++++ tests/unit/task-queue-worker.test.ts | 7 ++- 6 files changed, 129 insertions(+), 14 deletions(-) diff --git a/src/main/app/task-queue-worker.ts b/src/main/app/task-queue-worker.ts index 03181169..5c2b003d 100644 --- a/src/main/app/task-queue-worker.ts +++ b/src/main/app/task-queue-worker.ts @@ -46,7 +46,8 @@ export function isAutonomousPaused(connection: DatabaseConnection): boolean { * (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. + * 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. */ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProviderIds: string[] = [], lane: ClaimLane = "any"): QueuedRun | null { return connection.transaction(() => { @@ -59,6 +60,7 @@ export function claimNextQueuedRun(connection: DatabaseConnection, excludedProvi const row = connection.db .prepare(` select id, org_id, provider_id from task_runs where 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 diff --git a/src/main/services/organization-service.ts b/src/main/services/organization-service.ts index 94cb23f3..edb5ab3f 100644 --- a/src/main/services/organization-service.ts +++ b/src/main/services/organization-service.ts @@ -512,6 +512,9 @@ export class OrganizationService { } deleteTask(input: DeleteTaskInput): void { + const running = this.db.prepare("select 1 from task_runs where task_id = ? and org_id = ? and state = 'running' limit 1") + .get(input.taskId, input.orgId); + if (running) throw new Error("This task has a run in progress; wait for it to finish before deleting the task."); const result = this.db.prepare("delete from tasks where id = ? and org_id = ?").run(input.taskId, input.orgId); if (result.changes === 0) throw new Error("Task was not found."); this.audit.record({ diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 8ef3d5af..75fa6d75 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -107,17 +107,38 @@ export class TaskExecutionService { from task_runs where id = ? and org_id = ? `).get(runId, orgId) as RunRow | undefined; if (!run) throw new Error("Task run was not found."); - const task = this.requireTask(orgId, run.task_id); - const agent = this.requireAssignedAgent(orgId, task); - const provider = this.requireAgentProvider(orgId, agent); - const workspace = this.workspace(task.workspace_id); - const isolation: TaskRunIsolation = { - executionContextId: run.execution_context_id, - serviceIdentityId: run.service_identity_id, - workspaceMountJson: run.workspace_mount_json, - secretScopeJson: run.secret_scope_json, - }; - return this.executeAgentRun({ orgId, runId, task, agent, provider, workspace, isolation }); + let prepared: Parameters[0]; + try { + const task = this.requireTask(orgId, run.task_id); + if (task.assignee_agent_id !== run.agent_id) { + throw new Error(`Task was reassigned after this run was queued (run agent ${run.agent_id}, assignee ${task.assignee_agent_id ?? "none"}).`); + } + const agent = this.requireAssignedAgent(orgId, task); + const provider = this.requireAgentProvider(orgId, agent); + const workspace = this.workspace(task.workspace_id); + const isolation: TaskRunIsolation = { + executionContextId: run.execution_context_id, + serviceIdentityId: run.service_identity_id, + workspaceMountJson: run.workspace_mount_json, + secretScopeJson: run.secret_scope_json, + }; + prepared = { orgId, runId, task, agent, provider, workspace, isolation }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + console.error(`[task-run] run ${runId} failed before execution: ${message}`); + this.failRun(orgId, runId, null, message); + this.audit.record({ + orgId, + entityType: "task_run", + entityId: runId, + actorType: "system", + actorId: null, + action: "failed", + payload: { taskId: run.task_id, agentId: run.agent_id, providerId: run.provider_id, stage: "preparation", errorMessage: message }, + }); + throw error; + } + return this.executeAgentRun(prepared); } private async executeAgentRun(args: { @@ -297,6 +318,9 @@ export class TaskExecutionService { } private createRun(orgId: string, taskId: string, agentId: string, providerId: string, isolation: TaskRunIsolation): string { + const busy = this.db.prepare("select t.title from task_runs r join tasks t on t.id = r.task_id where r.agent_id = ? and r.state = 'running' limit 1") + .get(agentId) as { title: string } | undefined; + if (busy) throw new Error(`This agent is already working on "${busy.title}"; run again when that session ends.`); const runId = createId("run"); const now = new Date().toISOString(); this.db.prepare(` diff --git a/tests/integration/task-context-scope.test.ts b/tests/integration/task-context-scope.test.ts index e81006ae..70379683 100644 --- a/tests/integration/task-context-scope.test.ts +++ b/tests/integration/task-context-scope.test.ts @@ -7,6 +7,7 @@ import { OnboardingService } from "../../src/main/services/onboarding-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 { OrganizationService } from "../../src/main/services/organization-service"; import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; import { canonicalMentionHandle } from "../../src/shared/mention-handles"; import type { AgentTurnResult, ProviderToolTurnInput, ToolCall } from "../../src/main/services/provider-tool-turn"; @@ -64,9 +65,53 @@ describe("task prompt scope", () => { const task = connection.db.prepare("select id from tasks where org_id = ? and title = 'Build the booking API'").get(orgId) as { id: string }; await execution.runTask({ orgId, taskId: task.id }); + const danaId = (connection.db.prepare("select id from agents where org_id = ? and name = 'Dana Lee'").get(orgId) as { id: string }).id; + connection.db.prepare("update task_runs set state = 'running' where agent_id = ? and trigger = 'chat_mention'").run(danaId); + await expect(execution.runTask({ orgId, taskId: task.id })).rejects.toThrow("This agent is already working on"); + const busyTaskId = (connection.db.prepare("select task_id from task_runs where agent_id = ? and state = 'running'").get(danaId) as { task_id: string }).task_id; + expect(() => new OrganizationService(connection.db).deleteTask({ orgId, taskId: busyTaskId })).toThrow("has a run in progress"); + expect(connection.db.prepare("select count(*) as n from task_runs where task_id = ? and state = 'running'").get(busyTaskId)).toEqual({ n: 1 }); expect(danaPrompts).toHaveLength(1); expect(danaPrompts[0]).toContain("Build the booking API"); expect(danaPrompts[0]).not.toContain(CHANNEL_NOTE); connection.close(); }); + + test("a queued run whose task was reassigned fails instead of running as someone else", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-reassign-${randomUUID()}.db`)); + let calls = 0; + const turn = async (): Promise => { + calls += 1; + return { text: "done", toolCalls: [], rawResponseJson: "{}" }; + }; + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const organization = new OrganizationService(connection.db); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + let ceoId = ""; + connection.transaction(() => { + const summary = onboarding.bootstrap({ + companyName: "OrbitOps AI", companyDescription: "Ops.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Stand up a service business.", goalDescription: "Win customers.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }); + orgId = summary.orgId; + ceoId = summary.ceoAgentId; + }); + const providerId = (connection.db.prepare("select model_provider_id as id from agents where id = ?").get(ceoId) as { id: string }).id; + const otherId = organization.createAgent({ + orgId, name: "Ops Lead", role: "Ops", department: "Operations", + modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: ceoId, + }); + const kickoff = claimNextQueuedRun(connection)!; + const taskId = (connection.db.prepare("select task_id from task_runs where id = ?").get(kickoff.id) as { task_id: string }).task_id; + connection.db.prepare("update tasks set assignee_agent_id = ? where id = ?").run(otherId, taskId); + + await expect(execution.executeRun(orgId, kickoff.id)).rejects.toThrow("Task was reassigned after this run was queued"); + expect(connection.db.prepare("select state from task_runs where id = ?").get(kickoff.id)).toEqual({ state: "failed" }); + expect(calls).toBe(0); + connection.close(); + }); }); diff --git a/tests/unit/claim-priority.test.ts b/tests/unit/claim-priority.test.ts index d103a35f..45cbdc0c 100644 --- a/tests/unit/claim-priority.test.ts +++ b/tests/unit/claim-priority.test.ts @@ -98,3 +98,41 @@ describe("queue claim priority", () => { h.connection.close(); }); }); + +describe("one session per agent", () => { + test("a queued run for an agent that is already running waits; others proceed", () => { + const h = harness(); + const spaceWithCeo = h.chat.ensureDirectSpace(h.orgId, { type: "employee", id: h.employeeId }, { type: "agent", id: h.ceoId }); + const spaceWithDana = h.chat.ensureDirectSpace(h.orgId, { type: "employee", id: h.employeeId }, { type: "agent", id: h.danaId }); + const say = (spaceId: string, bodyMd: string) => h.chat.sendMessage({ + orgId: h.orgId, spaceId, threadId: null, senderType: "employee", senderId: h.employeeId, bodyMd, attachmentFileIds: [], + }); + say(spaceWithCeo, "First question?"); + say(spaceWithCeo, "Second, separate question?"); + say(spaceWithDana, "Dana, a question?"); + + const first = claimNextQueuedRun(h.connection)!; + expect(h.connection.db.prepare("select agent_id from task_runs where id = ?").get(first.id)).toEqual({ agent_id: h.ceoId }); + const second = claimNextQueuedRun(h.connection)!; + expect(h.connection.db.prepare("select agent_id from task_runs where id = ?").get(second.id)).toEqual({ agent_id: h.danaId }); + expect(claimNextQueuedRun(h.connection)).toBeNull(); + + h.connection.db.prepare("update task_runs set state = 'completed' where id = ?").run(first.id); + const third = claimNextQueuedRun(h.connection)!; + expect(h.connection.db.prepare("select agent_id from task_runs where id = ?").get(third.id)).toEqual({ agent_id: h.ceoId }); + h.connection.close(); + }); + + test("a running run whose agent was deleted blocks nobody", () => { + const h = harness(); + const spaceWithDana = h.chat.ensureDirectSpace(h.orgId, { type: "employee", id: h.employeeId }, { type: "agent", id: h.danaId }); + h.chat.sendMessage({ orgId: h.orgId, spaceId: spaceWithDana, threadId: null, senderType: "employee", senderId: h.employeeId, bodyMd: "Hi?", attachmentFileIds: [] }); + h.connection.db.prepare(` + insert into task_runs(id, org_id, task_id, agent_id, provider_id, execution_context_id, service_identity_id, workspace_mount_json, secret_scope_json, trigger, state, created_at) + select 'run_orphan', org_id, task_id, null, provider_id, 'ctx_orphan', service_identity_id, workspace_mount_json, '{}', 'manual', 'running', created_at + from task_runs where agent_id = ? + `).run(h.danaId); + expect(h.connection.db.prepare("select agent_id from task_runs where id = ?").get(claimNextQueuedRun(h.connection)!.id)).toEqual({ agent_id: h.danaId }); + h.connection.close(); + }); +}); diff --git a/tests/unit/task-queue-worker.test.ts b/tests/unit/task-queue-worker.test.ts index 52413d50..6b8a850d 100644 --- a/tests/unit/task-queue-worker.test.ts +++ b/tests/unit/task-queue-worker.test.ts @@ -40,14 +40,17 @@ describe("task queue worker claim", () => { const second = seedQueuedRun(connection, orgId, taskId, agentId, providerId); const claimedA = claimNextQueuedRun(connection); + // One session per agent: the second run waits while the first is running. + expect(claimNextQueuedRun(connection)).toBeNull(); + connection.db.prepare("update task_runs set state = 'completed' where id = ?").run(claimedA!.id); const claimedB = claimNextQueuedRun(connection); const claimedC = claimNextQueuedRun(connection); expect([claimedA?.id, claimedB?.id].sort()).toEqual([first, second].sort()); expect(claimedC).toBeNull(); - const states = (connection.db.prepare("select state from task_runs where org_id = ? order by created_at").all(orgId) as Array<{ state: string }>).map((r) => r.state); - expect(states).toEqual(["running", "running"]); + const states = (connection.db.prepare("select state from task_runs where org_id = ? order by created_at").all(orgId) as Array<{ state: string }>).map((r) => r.state).sort(); + expect(states).toEqual(["completed", "running"]); connection.close(); });