diff --git a/src/main/database/agents-hierarchy.ts b/src/main/database/agents-hierarchy.ts index 0dd8781a..051d6c07 100644 --- a/src/main/database/agents-hierarchy.ts +++ b/src/main/database/agents-hierarchy.ts @@ -203,15 +203,17 @@ export function buildAgentHierarchyMethods( id: string; name: string; department: string | null; + heartbeatIntervalSec: number | null; lastHeartbeatAt: string | null; }> { return readAll<{ id: string; name: string; department: string | null; + heartbeatIntervalSec: number | null; lastHeartbeatAt: string | null; }>( - "select id, name, department, last_heartbeat_at as lastHeartbeatAt from agents where company_id = ? and status not in ('terminated', 'paused', 'pending_approval')", + "select id, name, department, heartbeat_interval_sec as heartbeatIntervalSec, last_heartbeat_at as lastHeartbeatAt from agents where company_id = ? and status not in ('terminated', 'paused', 'pending_approval')", companyId, ); }, diff --git a/src/main/orchestration/scheduler.ts b/src/main/orchestration/scheduler.ts index 252e4bfa..eb1d1fe1 100644 --- a/src/main/orchestration/scheduler.ts +++ b/src/main/orchestration/scheduler.ts @@ -36,6 +36,7 @@ export function createScheduler( let unlockListener: (() => void) | null = null; const phaseLastRunAt = new Map(); const scheduleLastFiredByCompany = new Map(); + const taskWakeAttemptByAgent = new Map(); const bootstrapAttemptByAgent = new Map(); function phaseIsDue(phase: SchedulerPhase, now: number): boolean { @@ -231,6 +232,25 @@ export function createScheduler( scheduleLastFiredByCompany.set(companyId, now); } + function runAssignedTaskWakeCadence(companyId: string, now: number) { + const activeAgents = db.core.agents.listActiveAgentsForCompany(companyId); + for (const activeAgent of activeAgents) { + if (hasActiveRun({ agentId: activeAgent.id, companyId })) continue; + if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, activeAgent.id)) continue; + + const lastHeartbeat = activeAgent.lastHeartbeatAt ? new Date(activeAgent.lastHeartbeatAt).getTime() : 0; + const lastAttempt = taskWakeAttemptByAgent.get(activeAgent.id) ?? 0; + const intervalMs = (activeAgent.heartbeatIntervalSec || 120) * 1000; + if (now - Math.max(lastHeartbeat, lastAttempt) < intervalMs) continue; + + const actionableTasks = db.core.tasks.listAgentTasks(activeAgent.id, companyId, ["todo"]); + if (!actionableTasks.some((task) => !task.activeRunId)) continue; + + taskWakeAttemptByAgent.set(activeAgent.id, now); + wakeAgentIfPossible(activeAgent.id, companyId, "timer"); + } + } + function runHeartbeatCadencePhase() { const activeCompanies = db.core.company.listActiveCompanyIds(); const now = Date.now(); @@ -238,6 +258,7 @@ export function createScheduler( for (const company of activeCompanies) { try { runScheduleAutomation(company.id, now); + runAssignedTaskWakeCadence(company.id, now); runCompanyBootstrap(company, now); } catch (error) { logAppError(logger, error, { diff --git a/tests/unit/scheduler-heartbeat.test.ts b/tests/unit/scheduler-heartbeat.test.ts index abddfabb..f4299d3b 100644 --- a/tests/unit/scheduler-heartbeat.test.ts +++ b/tests/unit/scheduler-heartbeat.test.ts @@ -168,6 +168,37 @@ describe("unified scheduler heartbeat cadence", () => { expect(wakeAgentIfPossible).not.toHaveBeenCalled(); }); + it("wakes an agent with assigned todo work on heartbeat cadence", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + db.core.tasks.saveTask({ + companyId, + title: "Continue operating work", + description: "A queued Phase 2 task needs periodic wakeups after the initial event.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const wakeAgentIfPossible = vi.fn(); + const scheduler = createScheduler( + createDeps(db), + () => false, + wakeAgentIfPossible, + ); + + try { + scheduler.startAllSchedulers(); + vi.advanceTimersByTime(60_000); + } finally { + scheduler.stopAllSchedulers(); + } + + expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "timer"); + }); + it("bootstraps the CEO through the wake pipeline", () => { const companyId = createCompany(db); createConnector(db);