refactor(heartbeat): collapse skip decisions
This commit is contained in:
1 parent
1383805d66
commit
8d98151027
7 files changed
+247
-197
No files matched your search
@@ -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<HeartbeatRunDecision, { allowed: false }> => ({
|
||||
allowed: false,
|
||||
reason,
|
||||
detail,
|
||||
});
|
||||
|
||||
const SKIPPED = (decision: Extract<HeartbeatRunDecision, { allowed: false }>): 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,
|
||||
|
||||
@@ -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<typeof setInterval> | null = null;
|
||||
const scheduleLastFiredMap = new Map<string, number>();
|
||||
@@ -22,15 +22,7 @@ export function createHeartbeatScheduler(
|
||||
* plus the first skip-and-record cycle re-establish backoff.
|
||||
*/
|
||||
const attemptedAtByAgent = new Map<string, number>();
|
||||
/**
|
||||
* 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<string, { reason: string; count: number; since: string }>();
|
||||
const skipStreakByAgent = new Map<string, { reason: string; detail: string; count: number; since: string }>();
|
||||
|
||||
/**
|
||||
* 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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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();
|
||||
|
||||
@@ -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<string, unknown>` 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<string, unknown>;
|
||||
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);
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user