diff --git a/src/main/database-base.ts b/src/main/database-base.ts index 9ea64bb4..2a2cd9ad 100644 --- a/src/main/database-base.ts +++ b/src/main/database-base.ts @@ -243,6 +243,7 @@ export class AppDatabaseBase { renewTaskLease(taskId: string, runId: string) { return this.tasks.renewTaskLease(taskId, runId); } listAgentTasks(agentId: string, companyId: string, statuses: string[]) { return this.tasks.listAgentTasks(agentId, companyId, statuses); } findReusableHeartbeatTask(agentId: string, companyId: string) { return this.tasks.findReusableHeartbeatTask(agentId, companyId); } + acquireHeartbeatTask(input: { agentId: string; companyId: string; runId: string; workspaceId: string | null; title: string; description: string; priority: string }) { return this.tasks.acquireHeartbeatTask(input); } cleanupDuplicateHeartbeatTasks(companyId: string) { return this.tasks.cleanupDuplicateHeartbeatTasks(companyId); } listReviewTasksForManager(managerId: string, companyId: string) { return this.tasks.listReviewTasksForManager(managerId, companyId); } countActiveTasksForCompany(companyId: string) { return this.tasks.countActiveTasksForCompany(companyId); } diff --git a/src/main/database/tasks-status.ts b/src/main/database/tasks-status.ts index f5b7593b..a2d7b19e 100644 --- a/src/main/database/tasks-status.ts +++ b/src/main/database/tasks-status.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import type { DatabaseSync } from "node:sqlite"; import type { TaskRecord } from "@shared/types"; import { @@ -124,6 +125,107 @@ export function buildTaskStatusMethods( return row ?? null; }, + acquireHeartbeatTask(input: { + agentId: string; + companyId: string; + runId: string; + workspaceId: string | null; + title: string; + description: string; + priority: string; + }): TaskRecord | null { + return transaction(() => { + const ts = nowIso(); + const db = getDb(); + + const claimable = db + .prepare( + `SELECT id, status FROM tasks + WHERE company_id = ? AND assignee_agent_id = ? AND title LIKE '[Heartbeat]%' + AND status IN ('todo', 'backlog', 'done', 'in_review') + AND checkout_run_id IS NULL + ORDER BY + CASE status WHEN 'todo' THEN 0 WHEN 'backlog' THEN 1 WHEN 'in_review' THEN 2 WHEN 'done' THEN 3 ELSE 4 END ASC, + updated_at DESC, rowid DESC + LIMIT 1`, + ) + .get(input.companyId, input.agentId) as { id: string; status: string } | undefined; + + let taskId: string | null = null; + if (claimable) { + const claimResult = db + .prepare( + `UPDATE tasks + SET checkout_run_id = ?, execution_locked_at = ?, status = 'in_progress', + started_at = COALESCE(started_at, ?), completed_at = NULL, cancelled_at = NULL, + updated_at = ? + WHERE id = ? AND checkout_run_id IS NULL + AND status IN ('todo', 'backlog', 'done', 'in_review')`, + ) + .run(input.runId, ts, ts, ts, claimable.id); + if (claimResult.changes > 0) { + taskId = claimable.id; + } else { + return null; + } + } else { + const inflight = db + .prepare( + `SELECT 1 FROM tasks + WHERE company_id = ? AND assignee_agent_id = ? AND title LIKE '[Heartbeat]%' + AND status NOT IN ('done', 'cancelled') + LIMIT 1`, + ) + .get(input.companyId, input.agentId); + if (inflight) return null; + + taskId = randomUUID(); + db.prepare( + `INSERT INTO tasks + (id, company_id, project_id, goal_id, parent_id, title, description, + assignee_agent_id, workspace_id, status, priority, + checkout_run_id, execution_locked_at, started_at, + created_at, updated_at) + VALUES (?, ?, NULL, NULL, NULL, ?, ?, ?, ?, 'in_progress', ?, ?, ?, ?, ?, ?)`, + ).run( + taskId, + input.companyId, + input.title, + input.description, + input.agentId, + input.workspaceId, + input.priority, + input.runId, + ts, + ts, + ts, + ts, + ); + } + + const row = db + .prepare( + `SELECT id, company_id as companyId, project_id as projectId, goal_id as goalId, + parent_id as parentId, title, description, + assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, + status, priority, active_run_id as activeRunId, + session_json as sessionJson, session_display_id as sessionDisplayId, + cost_usd as costUsd, requires_user_review as requiresUserReview, + task_type as taskType, + started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, + created_at as createdAt, updated_at as updatedAt + FROM tasks WHERE id = ?`, + ) + .get(taskId) as TaskRecord | undefined; + if (!row) return null; + row.requiresUserReview = Boolean(row.requiresUserReview); + row.startedAt = row.startedAt ?? null; + row.completedAt = row.completedAt ?? null; + row.cancelledAt = row.cancelledAt ?? null; + return row; + }); + }, + cleanupDuplicateHeartbeatTasks(companyId: string): number { const ts = nowIso(); const result = getDb() diff --git a/src/main/orchestration/heartbeat.ts b/src/main/orchestration/heartbeat.ts index d9057cb4..5357f050 100644 --- a/src/main/orchestration/heartbeat.ts +++ b/src/main/orchestration/heartbeat.ts @@ -112,9 +112,26 @@ export function createHeartbeat( const isTimer = trigger === "timer"; const candidateTasks = assignedTasks; - let task = candidateTasks.find((entry) => !entry.activeRunId) ?? null; + const candidateTask = candidateTasks.find((entry) => !entry.activeRunId) ?? null; + const runId = randomUUID(); + let task: ReturnType | null = null; - if (!task) { + if (candidateTask) { + const checkoutSuccess = db.checkoutTask(candidateTask.id, runId); + if (!checkoutSuccess) { + db.addActivity({ + companyId, + actor: "system", + action: "execution.checkout_rejected", + entityType: "task", + entityId: candidateTask.id, + detail: `Agent ${agent.name} lost checkout race on task "${candidateTask.title}"`, + }); + logger.info(`[heartbeat] Agent ${agent.name} lost checkout race on task ${candidateTask.title}`); + return null; + } + task = db.getTask(candidateTask.id); + } else { const hasActiveAssignedRun = candidateTasks.some( (entry) => entry.activeRunId && hasActiveRun({ taskId: entry.id, companyId }), ); @@ -133,52 +150,30 @@ export function createHeartbeat( return null; } - // Paperclip-style task reuse: coalesce into existing heartbeat task instead of creating duplicates - const reusableTask = db.findReusableHeartbeatTask(agentId, companyId); - if (reusableTask) { - db.advanceTaskStatus(reusableTask.id, "todo", `Heartbeat task recycled for ${trigger} cycle.`); - task = db.getTask(reusableTask.id); - logger.info(`[heartbeat] Reusing existing heartbeat task for ${agent.name}: ${task.title}`); - } else { - const taskTitle = isCeo - ? `[Heartbeat] ${agent.name} — Strategic planning & task delegation` - : isManager - ? `[Heartbeat] ${agent.name} — Review team & delegate work` - : `[Heartbeat] ${agent.name} — Check for work`; - const taskDescription = isCeo - ? "Review active company goals. Break each goal into projects and actionable tasks. Assign tasks to department leads and team members. Create tasks using the API. This is your primary responsibility as CEO — ensure every goal has a clear execution plan with assigned owners." - : isManager - ? "Review your department's goals and your team's workload. Create tasks for your direct reports if they have capacity. Review any work submitted for your approval. Ensure your team is productive and unblocked." - : `Automated heartbeat run for ${agent.name}. Review assigned tasks, check goals, and take action as needed.`; + const taskTitle = isCeo + ? `[Heartbeat] ${agent.name} — Strategic planning & task delegation` + : isManager + ? `[Heartbeat] ${agent.name} — Review team & delegate work` + : `[Heartbeat] ${agent.name} — Check for work`; + const taskDescription = isCeo + ? "Review active company goals. Break each goal into projects and actionable tasks. Assign tasks to department leads and team members. Create tasks using the API. This is your primary responsibility as CEO — ensure every goal has a clear execution plan with assigned owners." + : isManager + ? "Review your department's goals and your team's workload. Create tasks for your direct reports if they have capacity. Review any work submitted for your approval. Ensure your team is productive and unblocked." + : `Automated heartbeat run for ${agent.name}. Review assigned tasks, check goals, and take action as needed.`; - const taskId = db.saveTask({ - companyId, - title: taskTitle, - description: taskDescription, - assigneeAgentId: agentId, - workspaceId: agent.workspaceId ?? undefined, - status: "todo", - priority: isCeo ? "high" : "medium", - }); - task = db.getTask(taskId); - logger.info(`[heartbeat] Created heartbeat task for ${agent.name}: ${taskTitle}`); - } - } - - // Generate runId upfront for CAS checkout (EXEC-01, EXEC-02) - const runId = randomUUID(); - const checkoutSuccess = db.checkoutTask(task.id, runId); - if (!checkoutSuccess) { - db.addActivity({ + task = db.acquireHeartbeatTask({ + agentId, companyId, - actor: "system", - action: "execution.checkout_rejected", - entityType: "task", - entityId: task.id, - detail: `Agent ${agent.name} lost checkout race on task "${task.title}"`, + runId, + workspaceId: agent.workspaceId ?? null, + title: taskTitle, + description: taskDescription, + priority: isCeo ? "high" : "medium", }); - logger.info(`[heartbeat] Agent ${agent.name} lost checkout race on task ${task.title}`); - return null; + if (!task) { + return null; + } + logger.info(`[heartbeat] Acquired heartbeat task for ${agent.name}: ${task.title}`); } const workspaceResolution = resolveWorkspace(task, agent, companyId);