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 <noreply@anthropic.com>
This commit is contained in:
1 parent
0fb4ca9ae6
commit
72eb645067
6 files changed
+129
-14
No files matched your search
@@ -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
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<TaskExecutionService["executeAgentRun"]>[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(`
|
||||
|
||||
@@ -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<AgentTurnResult> => {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user