diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index af5872fc..41dbe1e0 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -46,6 +46,7 @@ export class HeartbeatScheduler { and p.enabled = 1 and coalesce(u.request_count, 0) < p.monthly_request_limit and not exists (select 1 from task_runs r where r.task_id = t.id and r.state in ('queued', 'running')) + and not exists (select 1 from tasks c where c.parent_task_id = t.id and c.status <> 'done') `).all(window.start) as Array<{ task_id: string; org_id: string; provider_id: string }>; this.observeQuotaBlocks(window.start); const cooling = new Set(this.cooldowns.cooledProviderIds()); diff --git a/src/main/services/organization-service.ts b/src/main/services/organization-service.ts index 211120d3..ea181a5a 100644 --- a/src/main/services/organization-service.ts +++ b/src/main/services/organization-service.ts @@ -530,6 +530,7 @@ export class OrganizationService { } updateTaskStatus(orgId: string, taskId: string, status: string): void { + if (status === TASK_STATUS.DONE) this.assertNoOpenSubtasks(orgId, taskId); const result = this.db.prepare("update tasks set status = ?, updated_at = ? where id = ? and org_id = ?") .run(status, new Date().toISOString(), taskId, orgId); if (result.changes === 0) throw new Error("Task was not found."); @@ -543,6 +544,14 @@ export class OrganizationService { payload: { status }, }); } + + private assertNoOpenSubtasks(orgId: string, taskId: string): void { + const open = this.db.prepare("select title from tasks where org_id = ? and parent_task_id = ? and status <> 'done' order by created_at") + .all(orgId, taskId) as Array<{ title: string }>; + if (open.length === 0) return; + const parent = this.db.prepare("select title from tasks where id = ? and org_id = ?").get(taskId, orgId) as { title: string }; + throw new Error(`"${parent.title}" has open subtasks: ${open.map((task) => task.title).join(", ")}. It can close once they are done.`); + } } /** Avatar colors persist as #rrggbb; anything else is a caller bug. */ diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index a69746b1..f14aa0aa 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -121,6 +121,8 @@ export class TaskExecutionService { if (!sameWorkspaceAccess(run.workspace_mount_json, workspaceAccess(task, workspace))) { return this.supersedeRun(orgId, runId, run, "the task's workspace access changed after this run was queued."); } + const openSubtask = this.db.prepare("select title from tasks where parent_task_id = ? and status <> 'done' limit 1").get(task.id) as { title: string } | undefined; + if (openSubtask) return this.supersedeRun(orgId, runId, run, `the task has open subtasks ("${openSubtask.title}"); it runs again once they are done.`); const approvalProblem = executionApprovalProblem(this.db, orgId, task); if (approvalProblem) return this.supersedeRun(orgId, runId, run, approvalProblem); const isolation: TaskRunIsolation = { @@ -213,6 +215,8 @@ export class TaskExecutionService { handoffLines: renderTaskHandoff(this.db, orgId, task.handoff_json), knowledgeLines: relevantKnowledgeLines(this.db, { orgId, agentId: agent.id, projectId: task.project_id, goalId: task.goal_id, taskId: task.id }, `${task.title}\n${task.description}`), priorCompletedRuns: this.completedRunCount(task.id, runId), + subtasks: (this.db.prepare("select title, status = 'done' as done from tasks where parent_task_id = ? order by created_at").all(task.id) as Array<{ title: string; done: number }>) + .map((row) => ({ title: row.title, done: row.done === 1 })), availableToolNames: runTools(externalTools).map((tool) => tool.name), }), toolContext, diff --git a/src/main/services/task-prompts.ts b/src/main/services/task-prompts.ts index 76d1cd18..04a33c37 100644 --- a/src/main/services/task-prompts.ts +++ b/src/main/services/task-prompts.ts @@ -203,6 +203,8 @@ export function buildTaskPrompt(input: { priorCompletedRuns: number; /** Names of the tools the model can call this run. */ availableToolNames: string[]; + /** Subtasks this task delegated; autonomously it only runs again once they are all done. */ + subtasks: Array<{ title: string; done: boolean }>; }): string { const lines = [ "Execute this AgentCompany task.", @@ -266,6 +268,15 @@ export function buildTaskPrompt(input: { "This task carries filesystem access: read the existing code with list_files and read_file before changing anything, write changes with write_deliverable, and prove they work with run_command (install, build, test). Real output is code that builds and tests that pass — a document describing the work does not close an engineering task.", ); } + const openSubtasks = input.subtasks.filter((subtask) => !subtask.done).map((subtask) => subtask.title); + if (openSubtasks.length > 0) { + lines.push("", `Subtasks you delegated from this task are still open: ${openSubtasks.join(", ")}. Do not redo them; this task can close only after they are done.`); + } else if (input.subtasks.length > 0) { + lines.push( + "", + `The subtasks you delegated from this task are all done: ${input.subtasks.map((subtask) => subtask.title).join(", ")}. Check their results (read_messages, list_files, list_documents) against what this task needs, then close it, or create the follow-up work that is still missing.`, + ); + } if (input.priorCompletedRuns > 0) { lines.push( "", @@ -273,9 +284,10 @@ export function buildTaskPrompt(input: { ); } if (leadsPeople(input.agent.role_level, input.hasReports) && !input.hasReports) { + const park = `move it to "in_progress" with update_task_status(task_id="${input.task.id}", status="in_progress"): it waits while the tasks you created are open and comes back to you when they are done, so you can check the results and close it.`; const announce = input.isManager - ? `Post one company-wide kickoff to the #general channel. Then close this kickoff with update_task_status(task_id="${input.task.id}", status="done") and summarize.` - : `Tell your manager the plan in a direct message. Then close this task with update_task_status(task_id="${input.task.id}", status="done") if hiring and delegating was the work, or move it to "in_progress" and say what remains.`; + ? `Post one company-wide kickoff to the #general channel. Then ${park}` + : `Tell your manager the plan in a direct message. Then ${park}`; lines.push( "", `Your team is currently empty. First ${firstHiresFor(input.agent.role_level)}, create and assign their first tasks (create_task), and write the initial operating plan into the workspace (write_deliverable). Then use send_message with /handoff @Name and attachment_file_ids to pass the operating plan to the relevant report. Brief EACH report in a direct message (send_message with to="Their Name") about the work you assigned them — note the thread_id each returns so you can follow up in the same conversation. ${announce}`, diff --git a/src/renderer/src/pages/WorkPage.tsx b/src/renderer/src/pages/WorkPage.tsx index 5d4c992b..d50d9233 100644 --- a/src/renderer/src/pages/WorkPage.tsx +++ b/src/renderer/src/pages/WorkPage.tsx @@ -49,6 +49,7 @@ export function WorkPage({ snapshot, actions }: PageProps) { const [priorityFilter, setPriorityFilter] = useState("all"); const [assigneeFilter, setAssigneeFilter] = useState("all"); const [detailTaskId, setDetailTaskId] = useState(null); + const [actionError, setActionError] = useState(null); const allTasks = useMemo(() => snapshot.tasks.filter((task) => task.orgId === orgId), [snapshot.tasks, orgId]); const tasks = useMemo( @@ -75,10 +76,14 @@ export function WorkPage({ snapshot, actions }: PageProps) { const advance = (task: TaskRecord) => { const next = nextStatus(task.status); if (!next) return; - void actions.run(() => window.agentCompany.updateTaskStatus({ orgId, taskId: task.id, status: next })); + perform(() => actions.run(() => window.agentCompany.updateTaskStatus({ orgId, taskId: task.id, status: next }))); }; const runTask = (task: TaskRecord) => { - void actions.run(() => window.agentCompany.runTask({ orgId, taskId: task.id })); + perform(() => actions.run(() => window.agentCompany.runTask({ orgId, taskId: task.id }))); + }; + const perform = (operation: () => Promise) => { + setActionError(null); + operation().catch((error: unknown) => setActionError(error instanceof Error ? error.message : String(error))); }; const detailTask = detailTaskId ? allTasks.find((task) => task.id === detailTaskId) ?? null : null; @@ -102,6 +107,7 @@ export function WorkPage({ snapshot, actions }: PageProps) { {tasks.length === 1 ? t("{n} task", { n: tasks.length }) : t("{n} tasks", { n: tasks.length })} · {assigneeCount === 1 ? t("{n} assignee", { n: assigneeCount }) : t("{n} assignees", { n: assigneeCount })}

+ {actionError &&
{actionError}
}
diff --git a/tests/integration/autonomous-run.test.ts b/tests/integration/autonomous-run.test.ts index 6f43d004..31fba397 100644 --- a/tests/integration/autonomous-run.test.ts +++ b/tests/integration/autonomous-run.test.ts @@ -22,7 +22,7 @@ function call(name: string, args: Record): ToolCall { return { id: `call_${randomUUID()}`, name, arguments: args }; } -/** Scripts the CEO: staff two reports, delegate work, hand off the file in chat, close the kickoff, then stop. */ +/** Scripts the CEO: staff two reports, delegate work, hand off the file in chat, park the kickoff on its subtask, then stop. */ function scriptedCeo(): (input: ProviderToolTurnInput) => Promise { const script: Array<{ text: string; toolCalls: ToolCall[] }> = [ { text: "Staffing.", toolCalls: [ @@ -36,7 +36,7 @@ function scriptedCeo(): (input: ProviderToolTurnInput) => Promise Promise message.role === "user"); const taskId = firstUser ? /Task id: (\S+)/.exec(firstUser.text)?.[1] : undefined; index += 1; return { - text: "Closing the kickoff.", - toolCalls: taskId ? [call("update_task_status", { task_id: taskId, status: "done" })] : [], + text: "Parking the kickoff.", + toolCalls: taskId ? [call("update_task_status", { task_id: taskId, status: "in_progress" })] : [], rawResponseJson: JSON.stringify({ id: `resp_${index}` }), }; } @@ -162,13 +162,12 @@ describe("autonomous bootstrap run", () => { const run = state.taskRuns.find((r) => r.id === claimed!.id); expect(run?.state).toBe("completed"); - // The CEO closed the kickoff task using the task id the prompt carries. + // The CEO parked the kickoff using the task id the prompt carries; it waits on its subtask. const kickoffTask = connection.db.prepare("select status from tasks where id = (select task_id from task_runs where id = ?)") .get(claimed!.id) as { status: string }; - expect(kickoffTask).toEqual({ status: "done" }); + expect(kickoffTask).toEqual({ status: "in_progress" }); - // Stage 4 heartbeat: the CEO-delegated task (assigned to a hired, heartbeat-enabled - // agent, with no run yet) is enqueued exactly once for autonomous execution. + // Stage 4 heartbeat: only the delegated task is enqueued; the kickoff waits for it. const enqueued = taskExecution.enqueueHeartbeatRuns(); expect(enqueued).toHaveLength(1); const queued = connection.db.prepare("select task_id, state, trigger from task_runs where id = ?").get(enqueued[0]) as { task_id: string; state: string; trigger: string }; diff --git a/tests/integration/task-dependencies.test.ts b/tests/integration/task-dependencies.test.ts new file mode 100644 index 00000000..58458fec --- /dev/null +++ b/tests/integration/task-dependencies.test.ts @@ -0,0 +1,94 @@ +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 { OnboardingService } from "../../src/main/services/onboarding-service"; +import { OrganizationService } from "../../src/main/services/organization-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 { ChatService } from "../../src/main/services/chat-service"; +import { dispatchAgentTool } from "../../src/main/services/agent-tools"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import { RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../src/shared/domain"; +import type { AgentTurnResult, ProviderToolTurnInput } from "../../src/main/services/provider-tool-turn"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +function company() { + const connection = new DatabaseConnection(join(tmpdir(), `ac-deps-${randomUUID()}.db`)); + const prompts: string[] = []; + const turn = async (input: ProviderToolTurnInput): Promise => { + const first = input.messages[0]; + if (first?.role !== "user") throw new Error("The run must open with the task prompt."); + prompts.push(first.text); + 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 execution = new TaskExecutionService(connection.db, providers); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let ceoId = ""; + connection.transaction(() => { + const summary = onboarding.bootstrap({ + companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Ship the MVP.", goalDescription: "A working product.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }); + orgId = summary.orgId; + ceoId = summary.ceoAgentId; + }); + return { connection, execution, organization, orgId, ceoId, prompts }; +} + +describe("task dependencies", () => { + test("a parent cannot close while subtasks are open, and waits for them before running again", async () => { + const h = company(); + const { prompts } = h; + const kickoff = claimNextQueuedRun(h.connection)!; + const parentId = (h.connection.db.prepare("select task_id from task_runs where id = ?").get(kickoff.id) as { task_id: string }).task_id; + await h.execution.executeRun(h.orgId, kickoff.id); + const childId = h.organization.createTask({ + orgId: h.orgId, title: "Hire the CTO", description: "d", status: TASK_STATUS.BACKLOG, assigneeAgentId: h.ceoId, + risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId: null, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: false, + }); + h.connection.db.prepare("update tasks set parent_task_id = ? where id = ?").run(parentId, childId); + + const parentTitle = (h.connection.db.prepare("select title from tasks where id = ?").get(parentId) as { title: string }).title; + expect(() => h.organization.updateTaskStatus(h.orgId, parentId, TASK_STATUS.DONE)) + .toThrow(`"${parentTitle}" has open subtasks: Hire the CTO. It can close once they are done.`); + const ctx = { orgId: h.orgId, agent: { id: h.ceoId, modelProviderId: null, modelId: null }, taskProjectId: null, taskGoalId: null, taskId: null, sourceThreadId: null, workspace: null }; + const deps = { db: h.connection.db, organization: h.organization, chat: new ChatService(h.connection.db) }; + expect(dispatchAgentTool(deps, ctx, { id: "c1", name: "update_task_status", arguments: { task_id: parentId, status: "done" } })) + .toBe(`ERROR: "${parentTitle}" has open subtasks: Hire the CTO. It can close once they are done.`); + + // A run queued before the subtask existed is superseded instead of running ahead of it. + 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_early', org_id, task_id, agent_id, provider_id, 'ctx_early', service_identity_id, workspace_mount_json, secret_scope_json, 'heartbeat', 'running', created_at from task_runs where id = ?").run(kickoff.id); + await h.execution.executeRun(h.orgId, "run_early"); + expect(h.connection.db.prepare("select state, error_message from task_runs where id = 'run_early'").get()) + .toEqual({ state: "superseded", error_message: 'Superseded: the task has open subtasks ("Hire the CTO"); it runs again once they are done.' }); + + const waiting = h.connection.transaction(() => h.execution.enqueueHeartbeatRuns()); + const queuedTasks = () => (h.connection.db.prepare("select task_id from task_runs where state = 'queued'").all() as Array<{ task_id: string }>).map((row) => row.task_id); + expect(waiting).toHaveLength(1); + expect(queuedTasks()).toEqual([childId]); + + h.connection.db.prepare("update task_runs set state = 'completed' where state = 'queued'").run(); + h.organization.updateTaskStatus(h.orgId, childId, TASK_STATUS.DONE); + h.connection.transaction(() => h.execution.enqueueHeartbeatRuns()); + expect(queuedTasks()).toEqual([parentId]); + const resumed = claimNextQueuedRun(h.connection)!; + await h.execution.executeRun(h.orgId, resumed.id); + expect(prompts.at(-1)).toContain("The subtasks you delegated from this task are all done: Hire the CTO."); + expect(prompts[0]).toContain('move it to "in_progress"'); + expect(prompts[0]).not.toContain(`update_task_status(task_id="${parentId}", status="done") and summarize`); + h.organization.updateTaskStatus(h.orgId, parentId, TASK_STATUS.DONE); + h.connection.close(); + }); +});