From 61f2162236624bcea78b1ee3b99e4909dd802f19 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Fri, 29 May 2026 21:12:09 -0400 Subject: [PATCH] Wake CEO-hired agents via heartbeat so delegated tasks run autonomously The queue worker now also enqueues one run per actionable assigned task of a heartbeat-enabled agent with an enabled provider and no run yet, so after the CEO hires and delegates, its reports start their tasks on their own. Each task is enqueued at most once. The team-building prompt is now scoped to managers (no manager) with an empty team, so reports execute their task instead. --- .../src/main/app/task-queue-worker.ts | 8 ++- .../main/services/task-execution-service.ts | 69 ++++++++++++++++++- .../tests/integration/autonomous-run.test.ts | 12 ++++ 3 files changed, 85 insertions(+), 4 deletions(-) diff --git a/AgentCompany-New/src/main/app/task-queue-worker.ts b/AgentCompany-New/src/main/app/task-queue-worker.ts index 185d6964..2821cf18 100644 --- a/AgentCompany-New/src/main/app/task-queue-worker.ts +++ b/AgentCompany-New/src/main/app/task-queue-worker.ts @@ -48,8 +48,14 @@ export function startTaskQueueWorker(input: TaskQueueWorkerInput): () => void { if (running) return; running = true; try { + const enqueued = input.connection.transaction(() => taskExecution.enqueueHeartbeatRuns()); const claimed = claimNextQueuedRun(input.connection); - if (!claimed) return; + if (!claimed) { + if (enqueued.length > 0 && !input.mainWindow.isDestroyed()) { + input.mainWindow.webContents.send(IPC_CHANNELS.rendererEvent, { type: "snapshot-changed" }); + } + return; + } try { await taskExecution.executeRun(claimed.org_id, claimed.id); } catch { diff --git a/AgentCompany-New/src/main/services/task-execution-service.ts b/AgentCompany-New/src/main/services/task-execution-service.ts index 97df9d35..36d3bdef 100644 --- a/AgentCompany-New/src/main/services/task-execution-service.ts +++ b/AgentCompany-New/src/main/services/task-execution-service.ts @@ -10,6 +10,7 @@ import { runAgentLoop } from "./agent-run-loop"; import type { AgentToolContext } from "./agent-tools"; const TASK_RUN_STATE = { + QUEUED: "queued", RUNNING: "running", COMPLETED: "completed", FAILED: "failed", @@ -39,6 +40,7 @@ interface TaskExecutionAgentRow { role_level: string; model_provider_id: string | null; model_id: string | null; + manager_agent_id: string | null; } interface TaskExecutionProviderRow { @@ -180,6 +182,7 @@ export class TaskExecutionService { workspace, isolation, skillRuntimeSnapshot, + isManager: !agent.manager_agent_id, hasReports: this.agentHasReports(orgId, agent.id), }), toolContext, @@ -241,7 +244,7 @@ export class TaskExecutionService { private requireAssignedAgent(orgId: string, task: TaskExecutionTaskRow): TaskExecutionAgentRow { if (!task.assignee_agent_id) throw new Error("Task requires an assigned agent."); const agent = this.db.prepare(` - select id, name, role, department, title, role_level, model_provider_id, model_id + select id, name, role, department, title, role_level, model_provider_id, model_id, manager_agent_id from agents where id = ? and org_id = ? `).get(task.assignee_agent_id, orgId) as TaskExecutionAgentRow | undefined; @@ -313,6 +316,65 @@ export class TaskExecutionService { return runId; } + /** + * Heartbeat: enqueues one run per actionable assigned task of a + * heartbeat-enabled agent with an enabled provider that has not run yet, so + * CEO-delegated work starts on its own. Each task is enqueued at most once + * (guarded by existing runs), preventing runaway re-execution. + */ + enqueueHeartbeatRuns(): string[] { + const rows = this.db.prepare(` + select t.id as task_id, t.org_id as org_id + from tasks t + join agents a on a.id = t.assignee_agent_id and a.org_id = t.org_id + join providers p on p.id = a.model_provider_id and p.org_id = t.org_id + where t.status in ('backlog', 'in_progress') + and a.heartbeat_enabled = 1 + and a.lifecycle = 'active' + and p.enabled = 1 + and not exists (select 1 from task_runs r where r.task_id = t.id) + `).all() as Array<{ task_id: string; org_id: string }>; + const created: string[] = []; + for (const row of rows) { + try { + created.push(this.enqueueRun(row.org_id, row.task_id)); + } catch { + // Skip tasks that cannot be prepared yet (e.g. provider not fully configured). + } + } + return created; + } + + private enqueueRun(orgId: string, taskId: string): string { + const task = this.requireTask(orgId, taskId); + const agent = this.requireAssignedAgent(orgId, task); + const provider = this.requireAgentProvider(orgId, agent); + const workspace = this.workspace(task.workspace_id); + const isolation = this.createIsolationContext(orgId, task, agent, provider, workspace); + const runId = createId("run"); + const now = new Date().toISOString(); + this.db.prepare(` + insert into task_runs( + id, org_id, task_id, agent_id, provider_id, provider_request_id, + execution_context_id, service_identity_id, workspace_mount_json, secret_scope_json, + trigger, state, created_at, started_at, completed_at, error_message + ) values (?, ?, ?, ?, ?, null, ?, ?, ?, ?, 'heartbeat', ?, ?, null, null, null) + `).run( + runId, + orgId, + task.id, + agent.id, + provider.id, + isolation.executionContextId, + isolation.serviceIdentityId, + isolation.workspaceMountJson, + isolation.secretScopeJson, + TASK_RUN_STATE.QUEUED, + now, + ); + return runId; + } + private agentHasReports(orgId: string, agentId: string): boolean { const row = this.db.prepare("select 1 from agents where org_id = ? and manager_agent_id = ? and lifecycle = 'active' limit 1") .get(orgId, agentId); @@ -422,6 +484,7 @@ function buildTaskPrompt(input: { workspace: WorkspaceRow | null; isolation: TaskRunIsolation; skillRuntimeSnapshot: SkillRuntimeSnapshotRow | null; + isManager: boolean; hasReports: boolean; }): string { const lines = [ @@ -461,13 +524,13 @@ function buildTaskPrompt(input: { `Skill revisions: ${input.skillRuntimeSnapshot.source_revision_ids}`, ); } - if (!input.hasReports) { + if (input.isManager && !input.hasReports) { lines.push( "", "Your team is currently empty. Hire 3-5 direct reports for the roles this goal needs (hire_agent), create and assign their first tasks (create_task), and write the initial operating plan into the workspace (write_deliverable). Then summarize.", ); } else { - lines.push("", "Deliver completed work: make decisions, delegate or update tasks, and write real files with write_deliverable where appropriate."); + lines.push("", "Complete this task: do the work, write your output as real files with write_deliverable, update the task status when done, and summarize."); } return lines.join("\n"); } diff --git a/AgentCompany-New/tests/integration/autonomous-run.test.ts b/AgentCompany-New/tests/integration/autonomous-run.test.ts index 15768e11..ec1c06a5 100644 --- a/AgentCompany-New/tests/integration/autonomous-run.test.ts +++ b/AgentCompany-New/tests/integration/autonomous-run.test.ts @@ -96,6 +96,18 @@ describe("autonomous bootstrap run", () => { const run = state.taskRuns.find((r) => r.id === claimed!.id); expect(run?.state).toBe("completed"); + + // 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. + 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 }; + expect(queued.task_id).toBe(delegated!.id); + expect(queued.state).toBe("queued"); + expect(queued.trigger).toBe("heartbeat"); + // Idempotent: the delegated task now has a run, so it is not enqueued again. + expect(taskExecution.enqueueHeartbeatRuns()).toEqual([]); + connection.close(); }); });