refactor(heartbeat): close the silent-skip loop; durable skip reason
Symptom: scheduler logs `[heartbeat] Waking agent X (interval: 120s)`
every 15 seconds for three ICs, but no `[heartbeat] Starting taskless
heartbeat run` ever follows. Agents are wedged, UI shows nothing, and
the operator has no way to see why.
The loop had three compounding flaws:
1. `startAgentHeartbeatRun` returned `string | null`. Every gate
rejection was a silent `return null` — most were `logger.debug`,
which prod log levels filter out. Neither the scheduler nor the
renderer could tell the difference between "ran fine" and
"blocked by budget / no tasks / pending approval." The function
name promised "run or not" and the enum of reasons lived in
comments, not in the type.
2. The scheduler logged `[heartbeat] Waking agent ${name}` BEFORE
calling start, so even pure no-ops spammed the log. The "wake"
wasn't actually a wake; it was an attempt.
3. `markHeartbeatAt` is only called when a run starts. When start
returns null, `lastHeartbeatAt` stays frozen. On the next 15s
tick, `now - lastHeartbeatAt >= intervalMs` is still true, and
the scheduler fires again. The loop was a fixed-point at "skip →
log → skip → log" running 8 times per minute until something
external changed.
Root fix (data structure + visibility):
- `startAgentHeartbeatRun` now returns a discriminated union:
`{ status: "started"; runId } | { status: "skipped"; reason }`,
with `HeartbeatSkipReason` closed over 16 named reasons. Every
exit path carries its reason. The type tells operators what the
function can actually do; the enum tells them what it did.
- Scheduler holds a per-agent `attemptedAtByAgent` map. The interval
check uses `max(lastHeartbeatAt, lastAttemptedAt)`, so a streak of
skips still respects the heartbeat interval. Failed attempts cool
down the same as successful runs.
- On `skipped`, the scheduler persists `{reason, at}` into
`agents.runtime_state_json` under `heartbeatSkip`. On `started`,
it clears the field. The UI can read per-agent skip state from
the snapshot without a new IPC or table; operators finally see
"agent waiting: no_actionable_tasks" instead of a blank screen.
- The `Waking agent` log is gone. Its replacement prints once per
actual skip, with the reason inline (`[heartbeat] X not ready:
no_actionable_tasks`), and the existing `Starting taskless
heartbeat run for X` keeps marking the happy path.
`wakeAgentIfPossible` keeps its external `string | null` contract;
only the internal call site adapts. No caller outside of
`heartbeat.ts` / `scheduler-heartbeat.ts` / `scheduler.ts` needed
changes.
File sizes: heartbeat.ts 290, scheduler-heartbeat.ts 133. All under
the 400 cap. Tests: 603/603 green. Typecheck clean.
This commit is contained in:
1 parent
d75bfd5851
commit
1261c2431f
3 files changed
+129
-48
No files matched your search
@@ -5,6 +5,38 @@ import type { OrchestrationDependencies } from "./types";
|
||||
import type { RunDispatchParams } from "./turn-context";
|
||||
import { pickModelForAgent } from "../model-dispatch";
|
||||
|
||||
/**
|
||||
* 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"
|
||||
| "max_concurrent_runs"
|
||||
| "cooldown"
|
||||
| "wake_on_demand_disabled"
|
||||
| "pending_approval"
|
||||
| "awaiting_user_review"
|
||||
| "pending_deliverable_review"
|
||||
| "checkout_race_lost"
|
||||
| "no_actionable_tasks";
|
||||
|
||||
export type HeartbeatStartResult =
|
||||
| { status: "started"; runId: string }
|
||||
| { status: "skipped"; reason: HeartbeatSkipReason };
|
||||
|
||||
const SKIPPED = (reason: HeartbeatSkipReason): HeartbeatStartResult => ({
|
||||
status: "skipped",
|
||||
reason,
|
||||
});
|
||||
|
||||
export function createHeartbeat(
|
||||
deps: OrchestrationDependencies,
|
||||
resolveWorkspace: (task: { workspaceId: string | null; projectId?: string | null } | null, agent: { workspaceId: string | null }, companyId: string) => { workspace: ReturnType<AppDatabase["getWorkspace"]>; lockMode: "exclusive" | "shared" | "none"; source: string },
|
||||
@@ -27,20 +59,17 @@ export function createHeartbeat(
|
||||
}
|
||||
}
|
||||
|
||||
function startAgentHeartbeatRun(agentId: string, companyId: string, trigger: string): string | null {
|
||||
if (isQuitting()) return null;
|
||||
function startAgentHeartbeatRun(agentId: string, companyId: string, trigger: string): HeartbeatStartResult {
|
||||
if (isQuitting()) return SKIPPED("quitting");
|
||||
const isMessageWake = trigger === "message";
|
||||
const isCommunicationWake = isMessageWake || trigger === "peer_request";
|
||||
if (!hasWorkerReady()) return null;
|
||||
if (!hasWorkerReady()) return SKIPPED("worker_not_ready");
|
||||
const companyStatus = db.getCompanyStatus(companyId);
|
||||
if (companyStatus !== "active") {
|
||||
logger.debug(`[heartbeat] Skipping ${agentId} because company ${companyId} is not active.`);
|
||||
return null;
|
||||
}
|
||||
if (companyStatus !== "active") return SKIPPED("company_inactive");
|
||||
const agent = db.getAgent(agentId);
|
||||
if (agent.companyId !== companyId) {
|
||||
logger.warn(`[heartbeat] Refusing cross-company run for agent ${agentId}: ${agent.companyId} !== ${companyId}`);
|
||||
return null;
|
||||
return SKIPPED("cross_company");
|
||||
}
|
||||
const assignment = pickModelForAgent(agent, {
|
||||
getConnector: (id) => db.getConnector(id),
|
||||
@@ -56,8 +85,19 @@ export function createHeartbeat(
|
||||
});
|
||||
const connector = db.getConnector(assignment.connectorId);
|
||||
const effectiveAgentForEligibility: AgentRecord = { ...agent, model: assignment.modelId };
|
||||
if (getRunEligibilityFailure(effectiveAgentForEligibility, connector, null)) {
|
||||
return null;
|
||||
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);
|
||||
}
|
||||
// Extended heartbeat policy checks (Paperclip parity).
|
||||
// Let getHeartbeatPolicy throw — a missing row signals data corruption
|
||||
@@ -67,30 +107,20 @@ export function createHeartbeat(
|
||||
|
||||
// maxConcurrentRuns check: count active runs for this agent
|
||||
const activeRunCount = db.countActiveRunsForAgent(agent.id, companyId);
|
||||
if (activeRunCount >= heartbeatPolicy.maxConcurrentRuns) {
|
||||
logger.debug(`[heartbeat] Skipping ${agent.name} — active runs (${activeRunCount}) >= maxConcurrentRuns (${heartbeatPolicy.maxConcurrentRuns}).`);
|
||||
return null;
|
||||
}
|
||||
if (activeRunCount >= heartbeatPolicy.maxConcurrentRuns) return SKIPPED("max_concurrent_runs");
|
||||
|
||||
// Cooldown check: ensure enough time has passed since last heartbeat
|
||||
if (!isMessageWake && agent.lastHeartbeatAt) {
|
||||
const elapsed = Date.now() - new Date(agent.lastHeartbeatAt).getTime();
|
||||
const cooldownMs = heartbeatPolicy.cooldownSec * 1000;
|
||||
if (elapsed < cooldownMs) {
|
||||
logger.debug(`[heartbeat] Skipping ${agent.name} — cooldown (${heartbeatPolicy.cooldownSec}s) not elapsed.`);
|
||||
return null;
|
||||
}
|
||||
if (elapsed < cooldownMs) return SKIPPED("cooldown");
|
||||
}
|
||||
|
||||
// wakeOnDemand gate: if disabled, only timer/bootstrap triggers allowed
|
||||
if (!heartbeatPolicy.wakeOnDemand && trigger !== "timer" && trigger !== "bootstrap") {
|
||||
logger.debug(`[heartbeat] Skipping ${agent.name} — wakeOnDemand is disabled and trigger is "${trigger}".`);
|
||||
return null;
|
||||
}
|
||||
if (db.isAgentPendingApproval(agentId)) {
|
||||
logger.debug(`[heartbeat] Skipping ${agent.name} because agent is pending hire approval.`);
|
||||
return null;
|
||||
return SKIPPED("wake_on_demand_disabled");
|
||||
}
|
||||
if (db.isAgentPendingApproval(agentId)) return SKIPPED("pending_approval");
|
||||
|
||||
const companyMode = db.getCompanyExecutionMode(companyId);
|
||||
const approvalSettings = db.getCompanyApprovalSettings(companyId);
|
||||
@@ -103,8 +133,7 @@ export function createHeartbeat(
|
||||
!isMessageWake &&
|
||||
(db.hasAgentPendingDeliverableReview(agentId, companyId) || db.hasAgentTaskAwaitingReview(agentId, companyId))
|
||||
) {
|
||||
logger.debug(`[heartbeat] Skipping ${agent.name} — awaiting user deliverable review.`);
|
||||
return null;
|
||||
return SKIPPED("awaiting_user_review");
|
||||
}
|
||||
|
||||
// Block CEO heartbeat when company has any pending deliverable reviews
|
||||
@@ -114,8 +143,7 @@ export function createHeartbeat(
|
||||
!agent.reportsTo &&
|
||||
db.hasPendingDeliverableReviews(companyId)
|
||||
) {
|
||||
logger.debug(`[heartbeat] Skipping CEO ${agent.name} — company has pending deliverable reviews.`);
|
||||
return null;
|
||||
return SKIPPED("pending_deliverable_review");
|
||||
}
|
||||
|
||||
const assignedTasks = isCommunicationWake
|
||||
@@ -139,17 +167,14 @@ export function createHeartbeat(
|
||||
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;
|
||||
return SKIPPED("checkout_race_lost");
|
||||
}
|
||||
task = db.getTask(candidateTask.id);
|
||||
} else {
|
||||
const hasActiveAssignedRun = candidateTasks.some(
|
||||
(entry) => entry.activeRunId && hasActiveRun({ taskId: entry.id, companyId }),
|
||||
);
|
||||
if (hasActiveAssignedRun) {
|
||||
return null;
|
||||
}
|
||||
if (hasActiveAssignedRun) return SKIPPED("max_concurrent_runs");
|
||||
|
||||
const isCeo = !agent.reportsTo;
|
||||
const isManager = db.getDirectReports(agent.id, companyId).length > 0;
|
||||
@@ -158,8 +183,7 @@ export function createHeartbeat(
|
||||
const companyNeedsBootstrap = goalCount > 0 && taskCount < goalCount;
|
||||
|
||||
if (isTimer && !isCeo && !isManager && !companyNeedsBootstrap) {
|
||||
logger.debug(`[heartbeat] Skipping ${agent.name} — no actionable tasks and trigger is timer.`);
|
||||
return null;
|
||||
return SKIPPED("no_actionable_tasks");
|
||||
}
|
||||
|
||||
// No task — this is a true heartbeat run. The run row carries
|
||||
@@ -221,7 +245,7 @@ export function createHeartbeat(
|
||||
detail: `${trigger} heartbeat for ${agent.name}`,
|
||||
});
|
||||
emitEvent({ type: "run-updated", runId: run.id, action: "created" });
|
||||
return run.id;
|
||||
return { status: "started", runId: run.id };
|
||||
}
|
||||
|
||||
function wakeAgentIfPossible(agentId: string | null | undefined, companyId: string, trigger: string, context?: Record<string, unknown>) {
|
||||
@@ -249,13 +273,13 @@ export function createHeartbeat(
|
||||
return null;
|
||||
}
|
||||
|
||||
const runId = startAgentHeartbeatRun(agentId, companyId, trigger);
|
||||
if (!runId) {
|
||||
db.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent (deferred): ${trigger}`, context);
|
||||
const result = startAgentHeartbeatRun(agentId, companyId, trigger);
|
||||
if (result.status === "skipped") {
|
||||
db.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent (deferred:${result.reason}): ${trigger}`, context);
|
||||
return null;
|
||||
}
|
||||
db.completePendingWakeupRequests(agentId, companyId, runId);
|
||||
return runId;
|
||||
db.completePendingWakeupRequests(agentId, companyId, result.runId);
|
||||
return result.runId;
|
||||
}
|
||||
|
||||
return {
|
||||
|
||||
@@ -1,14 +1,64 @@
|
||||
import type { HeartbeatStartResult } from "./heartbeat";
|
||||
import type { OrchestrationDependencies } from "./types";
|
||||
|
||||
export function createHeartbeatScheduler(
|
||||
deps: OrchestrationDependencies,
|
||||
hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean,
|
||||
startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => string | null,
|
||||
startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => HeartbeatStartResult,
|
||||
) {
|
||||
const { db, logger, dispatchAutomation, hasWorkerReady, isQuitting } = deps;
|
||||
|
||||
let timer: ReturnType<typeof setInterval> | null = null;
|
||||
const scheduleLastFiredMap = new Map<string, number>();
|
||||
/**
|
||||
* Per-agent last-attempt timestamp. Needed so that attempts which
|
||||
* return `skipped` still respect the heartbeat interval on the next
|
||||
* tick — otherwise `lastHeartbeatAt` (which only advances on success)
|
||||
* leaves the agent in a 15-second retry loop spamming logs and the
|
||||
* activity table. Keyed by agent id. In-memory is fine: on restart
|
||||
* the scheduler starts with a clean slate, and `lastHeartbeatAt`
|
||||
* plus the first skip-and-record cycle re-establish backoff.
|
||||
*/
|
||||
const attemptedAtByAgent = new Map<string, number>();
|
||||
|
||||
function persistSkip(agentId: string, companyId: string, reason: string) {
|
||||
try {
|
||||
const existing = db.getRuntimeState(agentId) as Record<string, unknown>;
|
||||
db.setRuntimeState(agentId, companyId, {
|
||||
...existing,
|
||||
heartbeatSkip: { reason, at: new Date().toISOString() },
|
||||
});
|
||||
} catch (error) {
|
||||
logger.warn(
|
||||
`[heartbeat] Failed to persist skip reason for agent ${agentId}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function clearSkip(agentId: string, companyId: string) {
|
||||
try {
|
||||
const existing = db.getRuntimeState(agentId) as Record<string, unknown>;
|
||||
if (!("heartbeatSkip" in existing)) return;
|
||||
const next = { ...existing };
|
||||
delete next.heartbeatSkip;
|
||||
db.setRuntimeState(agentId, companyId, next);
|
||||
} catch (error) {
|
||||
logger.warn(
|
||||
`[heartbeat] Failed to clear skip reason for agent ${agentId}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function tryWake(agentId: string, companyId: string, agentName: string, trigger: string) {
|
||||
attemptedAtByAgent.set(agentId, Date.now());
|
||||
const result = startAgentHeartbeatRun(agentId, companyId, trigger);
|
||||
if (result.status === "skipped") {
|
||||
persistSkip(agentId, companyId, result.reason);
|
||||
logger.info(`[heartbeat] ${agentName} not ready: ${result.reason}`);
|
||||
} else {
|
||||
clearSkip(agentId, companyId);
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
if (timer) clearInterval(timer);
|
||||
@@ -32,12 +82,16 @@ export function createHeartbeatScheduler(
|
||||
continue;
|
||||
}
|
||||
|
||||
// Use the max of last-success and last-attempt so a chain
|
||||
// of skipped attempts still cools down the cycle instead
|
||||
// of retrying on every 15-second tick.
|
||||
const lastHeartbeat = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
|
||||
const lastAttempt = attemptedAtByAgent.get(agent.id) ?? 0;
|
||||
const lastTouch = Math.max(lastHeartbeat, lastAttempt);
|
||||
const intervalMs = agent.heartbeatIntervalSec * 1000;
|
||||
if (now - lastHeartbeat < intervalMs) continue;
|
||||
if (now - lastTouch < intervalMs) continue;
|
||||
|
||||
logger.info(`[heartbeat] Waking agent ${agent.name} (interval: ${agent.heartbeatIntervalSec}s)`);
|
||||
startAgentHeartbeatRun(agent.id, company.id, "timer");
|
||||
tryWake(agent.id, company.id, agent.name, "timer");
|
||||
}
|
||||
|
||||
const goalCount = db.countActiveGoalsForCompany(company.id);
|
||||
@@ -45,10 +99,12 @@ export function createHeartbeatScheduler(
|
||||
if (goalCount > 0 && taskCount === 0) {
|
||||
const ceo = db.getCeoForCompany(company.id);
|
||||
const ceoLastWake = ceo?.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
|
||||
const ceoLastAttempt = ceo ? attemptedAtByAgent.get(ceo.id) ?? 0 : 0;
|
||||
const ceoLastTouch = Math.max(ceoLastWake, ceoLastAttempt);
|
||||
const bootstrapCooldown = 120_000;
|
||||
if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastWake > bootstrapCooldown)) {
|
||||
if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastTouch > bootstrapCooldown)) {
|
||||
logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goals but no tasks — bootstrapping CEO ${ceo.name}`);
|
||||
startAgentHeartbeatRun(ceo.id, company.id, "bootstrap");
|
||||
tryWake(ceo.id, company.id, ceo.name, "bootstrap");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { OrchestrationDependencies } from "./types";
|
||||
import type { HeartbeatStartResult } from "./heartbeat";
|
||||
import { createHeartbeatScheduler } from "./scheduler-heartbeat";
|
||||
import { createLockRecoveryScheduler } from "./scheduler-lock-recovery";
|
||||
import { createIdleAutomationScheduler } from "./scheduler-idle-automation";
|
||||
@@ -7,7 +8,7 @@ import { createHousekeepingScheduler } from "./scheduler-housekeeping";
|
||||
export function createScheduler(
|
||||
deps: OrchestrationDependencies,
|
||||
hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean,
|
||||
startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => string | null,
|
||||
startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => HeartbeatStartResult,
|
||||
wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => string | null | undefined,
|
||||
) {
|
||||
const heartbeat = createHeartbeatScheduler(deps, hasActiveRun, startAgentHeartbeatRun);
|
||||
|
||||
Reference in new issue
Block a user