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.
This commit is contained in:
ZacharyZhang-NY committed 2026-05-29 21:12:09 -04:00
1 parent 0575712e02
commit 61f2162236
3 files changed
+85 -4

No files matched your search

@@ -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 {
@@ -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");
}
@@ -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();
});
});