From 8d98151027f2da03db3fdc788aa281d7f0885157 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, 24 Apr 2026 03:13:57 -0400 Subject: [PATCH] refactor(heartbeat): collapse skip decisions --- src/main/orchestration/heartbeat.ts | 119 +++++++------- src/main/orchestration/scheduler-heartbeat.ts | 81 ++-------- src/main/orchestration/turn-context.ts | 39 ----- src/main/prompt-builder/index.ts | 18 --- src/main/prompt-builder/messages/system-en.ts | 17 -- .../prompt-manager-classification.test.ts | 18 +++ tests/unit/scheduler-heartbeat.test.ts | 152 ++++++++++++++++++ 7 files changed, 247 insertions(+), 197 deletions(-) create mode 100644 tests/unit/scheduler-heartbeat.test.ts diff --git a/src/main/orchestration/heartbeat.ts b/src/main/orchestration/heartbeat.ts index cc771832..092745f2 100644 --- a/src/main/orchestration/heartbeat.ts +++ b/src/main/orchestration/heartbeat.ts @@ -8,38 +8,36 @@ import { pickModelForAgent } from "../model-dispatch"; import { HUMAN_SENDER_ID } from "../messaging/dispatch"; import { decideRunAttempt } from "../run-attempt-policy"; -/** - * Every reason a heartbeat attempt can be rejected before a run is - * dispatched. The enum is closed so callers (scheduler, UI) can reason - * about persistence and operator display exhaustively. - */ -export type HeartbeatSkipReason = - | "quitting" - | "worker_not_ready" - | "company_inactive" - | "cross_company" - | "agent_not_runnable" - | "budget_blocked" - | "connector_not_ready" - | "connector_model_not_allowed" - | "model_pick_failed" - | "max_concurrent_runs" - | "cooldown" - | "wake_on_demand_disabled" - | "pending_approval" - | "awaiting_user_review" - | "pending_deliverable_review" - | "checkout_race_lost" - | "no_actionable_tasks" - | "workspace_unresolved"; +export type HeartbeatBlockReason = + | "system_unavailable" + | "agent_unavailable" + | "policy_blocked" + | "human_blocked" + | "no_work" + | "workspace_unavailable" + | "dispatch_race"; + +export type HeartbeatRunDecision = + | { allowed: true; agent: AgentRecord; connector: ConnectorRecord; modelId: string | null; isCommunicationWake: boolean } + | { allowed: false; reason: HeartbeatBlockReason; detail: string }; export type HeartbeatStartResult = | { status: "started"; runId: string } - | { status: "skipped"; reason: HeartbeatSkipReason }; + | { status: "skipped"; reason: HeartbeatBlockReason; detail: string }; -const SKIPPED = (reason: HeartbeatSkipReason): HeartbeatStartResult => ({ - status: "skipped", +const BLOCKED = ( + reason: HeartbeatBlockReason, + detail: string, +): Extract => ({ + allowed: false, reason, + detail, +}); + +const SKIPPED = (decision: Extract): HeartbeatStartResult => ({ + status: "skipped", + reason: decision.reason, + detail: decision.detail, }); export function createHeartbeat( @@ -64,17 +62,17 @@ export function createHeartbeat( } } - function startAgentHeartbeatRun(agentId: string, companyId: string, trigger: string): HeartbeatStartResult { - if (isQuitting()) return SKIPPED("quitting"); + function canRunAgent(agentId: string, companyId: string, trigger: string): HeartbeatRunDecision { + if (isQuitting()) return BLOCKED("system_unavailable", "quitting"); const isMessageWake = trigger === "message"; const isCommunicationWake = isMessageWake || trigger === "peer_request"; - if (!hasWorkerReady()) return SKIPPED("worker_not_ready"); + if (!hasWorkerReady()) return BLOCKED("system_unavailable", "worker_not_ready"); const companyStatus = db.core.company.getCompanyStatus(companyId); - if (companyStatus !== "active") return SKIPPED("company_inactive"); + if (companyStatus !== "active") return BLOCKED("system_unavailable", `company_${companyStatus}`); const agent = db.core.entity.getAgent(agentId); if (agent.companyId !== companyId) { logger.warn(`[heartbeat] Refusing cross-company run for agent ${agentId}: ${agent.companyId} !== ${companyId}`); - return SKIPPED("cross_company"); + return BLOCKED("agent_unavailable", "cross_company"); } const pick = pickModelForAgent(agent, { getConnector: (id) => db.core.connectors.getConnector(id), @@ -92,24 +90,14 @@ export function createHeartbeat( entityId: agent.id, detail: pick.message, }); - return SKIPPED("model_pick_failed"); + return BLOCKED("policy_blocked", "model_pick_failed"); } const assignment = pick.assignment; const connector = db.core.connectors.getConnector(assignment.connectorId); const effectiveAgentForEligibility: AgentRecord = { ...agent, model: assignment.modelId }; const eligibilityFailure = getRunEligibilityFailure(effectiveAgentForEligibility, connector, null); if (eligibilityFailure) { - // getRunEligibilityFailure returns one of three string codes; map to - // the corresponding skip reason so the caller can surface it. - const mapped: HeartbeatSkipReason = - eligibilityFailure === "AGENT_NOT_RUNNABLE" - ? "agent_not_runnable" - : eligibilityFailure === "BUDGET_POLICY_BLOCKED" - ? "budget_blocked" - : eligibilityFailure === "CONNECTOR_NOT_READY" - ? "connector_not_ready" - : "connector_model_not_allowed"; - return SKIPPED(mapped); + return BLOCKED("policy_blocked", eligibilityFailure); } // Extended heartbeat policy checks (Paperclip parity). // Let getHeartbeatPolicy throw — a missing row signals data corruption @@ -129,14 +117,14 @@ export function createHeartbeat( bypassCooldown: isMessageWake, }); if (runAttemptDecision.action === "skip") { - return SKIPPED(runAttemptDecision.reason); + return BLOCKED("policy_blocked", runAttemptDecision.reason); } // wakeOnDemand gate: if disabled, only timer/bootstrap triggers allowed if (!heartbeatPolicy.wakeOnDemand && trigger !== "timer" && trigger !== "bootstrap") { - return SKIPPED("wake_on_demand_disabled"); + return BLOCKED("policy_blocked", "wake_on_demand_disabled"); } - if (db.core.agents.isAgentPendingApproval(agentId)) return SKIPPED("pending_approval"); + if (db.core.agents.isAgentPendingApproval(agentId)) return BLOCKED("human_blocked", "pending_approval"); const companyMode = db.core.company.getCompanyExecutionMode(companyId); const approvalSettings = db.core.company.getCompanyApprovalSettings(companyId); @@ -149,7 +137,7 @@ export function createHeartbeat( !isMessageWake && (db.core.tasks.hasAgentPendingDeliverableReview(agentId, companyId) || db.core.tasks.hasAgentTaskAwaitingReview(agentId, companyId)) ) { - return SKIPPED("awaiting_user_review"); + return BLOCKED("human_blocked", "awaiting_user_review"); } // Block CEO heartbeat when company has any pending deliverable reviews @@ -159,9 +147,23 @@ export function createHeartbeat( !agent.reportsTo && db.core.tasks.hasPendingDeliverableReviews(companyId) ) { - return SKIPPED("pending_deliverable_review"); + return BLOCKED("human_blocked", "pending_deliverable_review"); } + return { + allowed: true, + agent, + connector, + modelId: assignment.modelId, + isCommunicationWake, + }; + } + + function startAgentHeartbeatRun(agentId: string, companyId: string, trigger: string): HeartbeatStartResult { + const decision = canRunAgent(agentId, companyId, trigger); + if (!decision.allowed) return SKIPPED(decision); + const { agent, connector, modelId, isCommunicationWake } = decision; + const assignedTasks = isCommunicationWake ? [] : db.core.tasks.listAgentTasks(agentId, companyId, ["todo", "in_progress", "blocked"]); @@ -177,7 +179,7 @@ export function createHeartbeat( const hasActiveAssignedRun = candidateTasks.some( (entry) => entry.activeRunId && hasActiveRun({ taskId: entry.id, companyId }), ); - if (hasActiveAssignedRun) return SKIPPED("max_concurrent_runs"); + if (hasActiveAssignedRun) return SKIPPED(BLOCKED("policy_blocked", "max_concurrent_runs")); const isCeo = !agent.reportsTo; const isManager = db.core.agents.getDirectReports(agent.id, companyId).length > 0; @@ -186,7 +188,7 @@ export function createHeartbeat( const companyNeedsBootstrap = goalCount > 0 && taskCount < goalCount; if (isTimer && !isCeo && !isManager && !companyNeedsBootstrap) { - return SKIPPED("no_actionable_tasks"); + return SKIPPED(BLOCKED("no_work", "no_actionable_tasks")); } // No task — this is a true heartbeat run. The run row carries @@ -209,7 +211,7 @@ export function createHeartbeat( detail: error.message, }); logger.warn(`[heartbeat] Workspace unresolved for ${agent.name}: ${error.message}`); - return SKIPPED("workspace_unresolved"); + return SKIPPED(BLOCKED("workspace_unavailable", "workspace_unresolved")); } throw error; } @@ -245,7 +247,7 @@ export function createHeartbeat( entityId: task.id, detail: `Agent ${agent.name} lost checkout race on task "${task.title}"`, }); - return SKIPPED("checkout_race_lost"); + return SKIPPED(BLOCKED("dispatch_race", "checkout_race_lost")); } const companyBasic = db.core.company.getCompanyBasic(companyId); @@ -262,7 +264,7 @@ export function createHeartbeat( companyDescription: companyBasic?.description ?? "", snapshot: db.projection.snapshot.listSnapshot(), trigger, - resolvedModel: assignment.modelId, + resolvedModel: modelId, }); db.core.agents.markHeartbeatAt(agentId); @@ -323,7 +325,13 @@ export function createHeartbeat( const result = startAgentHeartbeatRun(agentId, companyId, trigger); if (result.status === "skipped") { - db.automation.wakeup.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent (deferred:${result.reason}): ${trigger}`, context); + db.automation.wakeup.enqueueWakeupRequest( + companyId, + agentId, + trigger, + `Wake for agent (deferred:${result.reason}/${result.detail}): ${trigger}`, + context, + ); return null; } db.automation.wakeup.completePendingWakeupRequests(agentId, companyId, result.runId); @@ -331,6 +339,7 @@ export function createHeartbeat( } return { + canRunAgent, ensureAgentHeartbeatEnabled, startAgentHeartbeatRun, wakeAgentIfPossible, diff --git a/src/main/orchestration/scheduler-heartbeat.ts b/src/main/orchestration/scheduler-heartbeat.ts index c91bdd5d..99768ac1 100644 --- a/src/main/orchestration/scheduler-heartbeat.ts +++ b/src/main/orchestration/scheduler-heartbeat.ts @@ -8,7 +8,7 @@ export function createHeartbeatScheduler( hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean, startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => HeartbeatStartResult, ) { - const { db, logger, dispatchAutomation, hasWorkerReady, isQuitting } = deps; + const { db, logger, dispatchAutomation, isQuitting } = deps; let timer: ReturnType | null = null; const scheduleLastFiredMap = new Map(); @@ -22,15 +22,7 @@ export function createHeartbeatScheduler( * plus the first skip-and-record cycle re-establish backoff. */ const attemptedAtByAgent = new Map(); - /** - * Tracks consecutive skip streak per agent so the per-turn prompt - * builder (which reads `runtime_state_json` at dispatch time) can - * surface "skipped: worker_not_ready (last N ticks since T)" and stop - * being blind to a chain of zero-delivery heartbeats. Keyed by agent - * id → { reason, count, since }. Reset when status flips to `started` - * or the reason changes. - */ - const skipStreakByAgent = new Map(); + const skipStreakByAgent = new Map(); /** * Periodic heartbeat.skip_ongoing emission gate. The activity table @@ -43,51 +35,6 @@ export function createHeartbeatScheduler( */ const SKIP_ONGOING_EVERY_N = 40; - function persistSkip( - agentId: string, - companyId: string, - reason: string, - streak: { count: number; since: string }, - lastTickAt: string, - ) { - try { - const existing = db.core.agents.getRuntimeState(agentId) as Record; - db.core.agents.setRuntimeState(agentId, companyId, { - ...existing, - heartbeatSkip: { - reason, - streak: streak.count, - since: streak.since, - lastTickAt, - }, - }); - } catch (error) { - logAppError(logger, error, { - kind: "recoverable", - code: "scheduler.heartbeat.persist_skip_failed", - message: `Failed to persist heartbeat skip reason for agent ${agentId}`, - context: { agentId, companyId }, - }, "warn"); - } - } - - function clearSkip(agentId: string, companyId: string) { - try { - const existing = db.core.agents.getRuntimeState(agentId) as Record; - if (!("heartbeatSkip" in existing)) return; - const next = { ...existing }; - delete next.heartbeatSkip; - db.core.agents.setRuntimeState(agentId, companyId, next); - } catch (error) { - logAppError(logger, error, { - kind: "recoverable", - code: "scheduler.heartbeat.clear_skip_failed", - message: `Failed to clear heartbeat skip reason for agent ${agentId}`, - context: { agentId, companyId }, - }, "warn"); - } - } - /** * Structured skip write to the activity stream. Emits on first skip, * on reason change, and every SKIP_ONGOING_EVERY_N consecutive skips @@ -96,25 +43,23 @@ export function createHeartbeatScheduler( * next visible row used to be the flip/resume, which meant long * sleep/standby stalls were invisible in the audit feed. * - * `detail` is emitted as JSON `{reason, streak, since, trigger, - * agentName}` so downstream readers parse structured fields instead - * of regexing English. The prompt-side consumer reads - * `runtime_state_json.heartbeatSkip` (written by `persistSkip`), so - * activity here is pure audit. + * `detail` is emitted as JSON so downstream readers parse structured + * fields instead of regexing English. */ function writeSkipActivity( agentId: string, companyId: string, agentName: string, reason: string, + detail: string, trigger: string, tickAt: string, ): { count: number; since: string } { const previous = skipStreakByAgent.get(agentId); - const sameReason = previous && previous.reason === reason; + const sameReason = previous && previous.reason === reason && previous.detail === detail; const nextStreak = sameReason - ? { reason, count: previous.count + 1, since: previous.since } - : { reason, count: 1, since: tickAt }; + ? { reason, detail, count: previous.count + 1, since: previous.since } + : { reason, detail, count: 1, since: tickAt }; skipStreakByAgent.set(agentId, nextStreak); const action = !sameReason ? "heartbeat.skipped" @@ -132,6 +77,7 @@ export function createHeartbeatScheduler( detail: JSON.stringify({ agentName, reason, + detail, streak: nextStreak.count, since: nextStreak.since, trigger, @@ -163,6 +109,7 @@ export function createHeartbeatScheduler( detail: JSON.stringify({ agentName, reason: previous.reason, + detail: previous.detail, streak: previous.count, since: previous.since, resumedAt: new Date().toISOString(), @@ -183,18 +130,16 @@ export function createHeartbeatScheduler( attemptedAtByAgent.set(agentId, Date.now()); const result = startAgentHeartbeatRun(agentId, companyId, trigger); if (result.status === "skipped") { - const streak = writeSkipActivity(agentId, companyId, agentName, result.reason, trigger, tickAt); - persistSkip(agentId, companyId, result.reason, streak, tickAt); - logger.info(`[heartbeat] ${agentName} not ready: ${result.reason} (streak=${streak.count})`); + const streak = writeSkipActivity(agentId, companyId, agentName, result.reason, result.detail, trigger, tickAt); + logger.info(`[heartbeat] ${agentName} not ready: ${result.reason}/${result.detail} (streak=${streak.count})`); } else { - clearSkip(agentId, companyId); clearSkipStreak(agentId, companyId, agentName); } } function runTick() { try { - if (isQuitting() || !hasWorkerReady()) return; + if (isQuitting()) return; const activeCompanies = db.core.company.listActiveCompanyIds(); const now = Date.now(); diff --git a/src/main/orchestration/turn-context.ts b/src/main/orchestration/turn-context.ts index 7ad58ef5..88ea6464 100644 --- a/src/main/orchestration/turn-context.ts +++ b/src/main/orchestration/turn-context.ts @@ -291,43 +291,6 @@ function buildConnectorCatalog( .filter((c) => c.enabledModels.length > 0); } -/** - * Lift the scheduler-written `heartbeatSkip` blob out of - * `runtime_state_json` into a typed PromptContext slice. The heartbeat - * scheduler writes the reader-free fields {reason, streak, since, - * lastTickAt}; we validate the shape here because `getRuntimeState` - * returns `Record` and bad JSON shouldn't poison the - * prompt. Missing / malformed entries return null — the prompt section - * is omitted rather than rendered with placeholder values. - */ -function readHeartbeatSkip( - db: AppDatabase, - agentId: string, -): { - reason: string; - streak: number; - since: string; - lastTickAt: string; -} | null { - const state = db.core.agents.getRuntimeState(agentId); - const raw = state["heartbeatSkip"]; - if (!raw || typeof raw !== "object") return null; - const record = raw as Record; - const reason = record.reason; - const streak = record.streak; - const since = record.since; - const lastTickAt = record.lastTickAt; - if ( - typeof reason !== "string" || - typeof streak !== "number" || - typeof since !== "string" || - typeof lastTickAt !== "string" - ) { - return null; - } - return { reason, streak, since, lastTickAt }; -} - /** * Assemble the full run payload from resolved entities and dependencies. * This is the core composition step that was previously inlined in dispatchRunToWorker. @@ -345,7 +308,6 @@ export function buildRunPayload(params: RunDispatchParams, deps: TurnContextDeps const initialHandoff = resolveSessionHandoff(agent.id, deps); const { taskSessionId, sessionHandoff } = resolveSession(db, agent, companyId, connector.id, initialHandoff, deps); const skills = resolveSkills(companyId, agent.id, connector.id, agent.name, deps); - const heartbeatSkip = readHeartbeatSkip(db, agent.id); const systemPrompt = apiUrl ? buildTaskPrompt({ agent, @@ -391,7 +353,6 @@ export function buildRunPayload(params: RunDispatchParams, deps: TurnContextDeps .filter((workspace) => workspace.companyId === companyId) .map((workspace) => ({ name: workspace.name, localPath: workspace.localPath })), companyDocuments: snapshot.documents.filter((d) => d.companyId === companyId), - heartbeatSkip, }) : null; const thinkingEffort = extractThinkingEffort(agent.metadataJson, deps.logger, agent.id); diff --git a/src/main/prompt-builder/index.ts b/src/main/prompt-builder/index.ts index 5bd6b015..a1f67ae3 100644 --- a/src/main/prompt-builder/index.ts +++ b/src/main/prompt-builder/index.ts @@ -45,20 +45,6 @@ export interface PromptContext { companySkillCatalog?: Array<{ name: string; slug: string; description: string | null }>; /** Company workspaces visible to the current run. */ companyWorkspaces?: Array<{ name: string; localPath: string }>; - /** - * Pending-skip context from the heartbeat scheduler. Present only - * when the scheduler most recently refused to dispatch this agent - * (e.g. worker not ready, cooldown). Surfacing it into the system - * prompt is the reader half of the scheduler-side write — without - * it, a multi-day skip streak would be invisible to the agent even - * though `runtime_state_json.heartbeatSkip` tracks it precisely. - */ - heartbeatSkip?: { - reason: string; - streak: number; - since: string; - lastTickAt: string; - } | null; } export function buildTaskPrompt(ctx: PromptContext): string { @@ -83,10 +69,6 @@ export function buildTaskPrompt(ctx: PromptContext): string { parts.push(`\n${m.wakeReasonHeading}\n${explanation}\n`); } - if (ctx.heartbeatSkip) { - parts.push(`\n${m.heartbeatSkipSection(ctx.heartbeatSkip)}`); - } - parts.push(`\n${m.timeAwarenessSection}`); // Taskless runs used to emit a "respond with a brief status summary" diff --git a/src/main/prompt-builder/messages/system-en.ts b/src/main/prompt-builder/messages/system-en.ts index 6ab73832..fac366d3 100644 --- a/src/main/prompt-builder/messages/system-en.ts +++ b/src/main/prompt-builder/messages/system-en.ts @@ -271,23 +271,6 @@ export function importantRulesSection(companyId: string, agentId: string, runId: export const wakeReasonHeading = "# Wake Reason"; -export function heartbeatSkipSection(args: { - reason: string; - streak: number; - since: string; - lastTickAt: string; -}): string { - return `# Heartbeat Skip Status -The scheduler most recently declined to dispatch this agent. -- Reason: ${args.reason} -- Consecutive skips: ${args.streak} -- First skip in this streak: ${args.since} -- Last tick: ${args.lastTickAt} - -If this streak is long and unexplained, it likely means a subsystem (worker pool, connector, quota, checkpoint hold) has been blocking wakes. Check the subsystem before assuming silence means nothing is wrong. -`; -} - export const timeAwarenessSection = `# Time Awareness Consider what phase of operations the company is in and prioritize accordingly: - Early in a sprint cycle → planning, task breakdown, delegation diff --git a/tests/unit/prompt-manager-classification.test.ts b/tests/unit/prompt-manager-classification.test.ts index a2c4da08..8199b2cd 100644 --- a/tests/unit/prompt-manager-classification.test.ts +++ b/tests/unit/prompt-manager-classification.test.ts @@ -238,6 +238,24 @@ describe("buildSystemPrompt leadership classification", () => { expect(changedFullPrompt).toContain("1 blocked"); }); + it("keeps heartbeat skip state out of the task prompt", () => { + const ctx = buildCtx() as PromptContext & { + heartbeatSkip: { reason: string; streak: number; since: string; lastTickAt: string }; + }; + ctx.heartbeatSkip = { + reason: "worker_not_ready", + streak: 7, + since: "2026-04-23T00:00:00.000Z", + lastTickAt: "2026-04-23T00:10:00.000Z", + }; + + const prompt = buildTaskPrompt(ctx); + + expect(prompt).not.toContain("Heartbeat Skip Status"); + expect(prompt).not.toContain("worker_not_ready"); + expect(prompt).not.toContain("Consecutive skips"); + }); + it("keeps volatile task context after the cacheable system prefix", () => { const ctx = buildCtx({ wakeReason: "timer" }); const systemPrompt = buildSystemPrompt(ctx); diff --git a/tests/unit/scheduler-heartbeat.test.ts b/tests/unit/scheduler-heartbeat.test.ts new file mode 100644 index 00000000..21e04db4 --- /dev/null +++ b/tests/unit/scheduler-heartbeat.test.ts @@ -0,0 +1,152 @@ +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { mkdtemp, rm } from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("electron", () => ({ + powerMonitor: { + on: vi.fn(), + removeListener: vi.fn(), + }, + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +import { AppDatabase } from "../../src/main/database"; +import { createHeartbeatScheduler } from "../../src/main/orchestration/scheduler-heartbeat"; +import type { OrchestrationDependencies } from "../../src/main/orchestration/types"; + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createCompany(db: AppDatabase): string { + return db.core.company.saveCompany({ + name: "Heartbeat Scheduler Co", + description: "Scheduler gate contract test", + status: "active", + }); +} + +function createConnector(db: AppDatabase): void { + db.core.connectors.saveConnector({ + id: "openai-codex", + model: "gpt-5.4", + envBindingText: "", + notes: "Test connector", + }); +} + +function createWorkspace(db: AppDatabase, companyId: string, localPath: string): string { + return db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); +} + +function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string { + const agentId = db.core.agents.saveAgent({ + companyId, + name: "Scheduler Agent", + role: "Engineer", + title: "Runtime Engineer", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Checks scheduler heartbeat gates", + budgetMonthlyUsd: 100, + }); + db.core.agents.setHeartbeat(agentId, true, 1); + return agentId; +} + +function createDeps(db: AppDatabase): OrchestrationDependencies { + return { + db, + logger: createLogger(), + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + notify: vi.fn(), + createRunToken: () => "test-token", + getApiPort: () => 3210, + dispatchAutomation: vi.fn(), + hasWorkerReady: () => false, + enqueueWorkerRun: vi.fn(), + isQuitting: () => false, + listRuntimeSkillEntries: () => [], + budgetService: { + evaluateBudgetGate: vi.fn(() => ({ allowed: true, blockedBy: [] })), + } as never, + checkpointEngine: { shouldPauseHeartbeat: () => false }, + }; +} + +describe("heartbeat scheduler", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-04-23T00:00:00.000Z")); + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-scheduler-heartbeat-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + vi.useRealTimers(); + }); + + it("records worker readiness skips from the heartbeat run decision", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const startAgentHeartbeatRun = vi.fn(() => ({ + status: "skipped" as const, + reason: "system_unavailable" as const, + detail: "worker_not_ready", + })); + const scheduler = createHeartbeatScheduler( + createDeps(db), + () => false, + startAgentHeartbeatRun, + ); + + try { + scheduler.start(); + vi.advanceTimersByTime(15_000); + } finally { + scheduler.stop(); + } + + expect(startAgentHeartbeatRun).toHaveBeenCalledWith(agentId, companyId, "timer"); + const activity = db.projection.snapshot.listSnapshot({ companyId }).activity; + const skip = activity.find((entry) => entry.action === "heartbeat.skipped"); + expect(skip?.entityId).toBe(agentId); + expect(JSON.parse(skip?.detail ?? "{}")).toMatchObject({ + agentName: "Scheduler Agent", + reason: "system_unavailable", + detail: "worker_not_ready", + trigger: "timer", + }); + expect(db.core.agents.getRuntimeState(agentId)).not.toHaveProperty("heartbeatSkip"); + }); +});