ZacharyZhang-NY/AgentCompany
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

refactor(scheduler): unify cadence phases

Browse Source
This commit is contained in:
ZacharyZhang-NY committed 2026-04-24 03:26:17 -04:00
1 parent 8d98151027
commit ead9475b71
8 files changed
+463 -588
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/main/orchestration/heartbeat.ts
+75 -1
View File
@@ -54,6 +54,77 @@ export function createHeartbeat(
hasWorkerReady,
isQuitting,
} = deps;
const skipStreakByAgent = new Map<string, { reason: string; detail: string; count: number; since: string }>();
const SKIP_ONGOING_EVERY_N = 40;
function recordSkipActivity(
agentId: string,
companyId: string,
agentName: string,
reason: HeartbeatBlockReason,
detail: string,
trigger: string,
) {
const tickAt = new Date().toISOString();
const previous = skipStreakByAgent.get(agentId);
const sameReason = previous && previous.reason === reason && previous.detail === detail;
const nextStreak = sameReason
? { reason, detail, count: previous.count + 1, since: previous.since }
: { reason, detail, count: 1, since: tickAt };
skipStreakByAgent.set(agentId, nextStreak);
const action = !sameReason
? "heartbeat.skipped"
: nextStreak.count % SKIP_ONGOING_EVERY_N === 0
? "heartbeat.skip_ongoing"
: null;
if (!action) return;
try {
db.core.activity.addActivity({
companyId,
actor: "system",
action,
entityType: "agent",
entityId: agentId,
detail: JSON.stringify({
agentName,
reason,
detail,
streak: nextStreak.count,
since: nextStreak.since,
trigger,
}),
});
} catch (error) {
logger.warn(`[heartbeat] Failed to append ${action} activity for ${agentName}:`, error);
}
}
function clearSkipActivity(agentId: string, companyId: string, agentName: string) {
const previous = skipStreakByAgent.get(agentId);
if (!previous) return;
skipStreakByAgent.delete(agentId);
try {
db.core.activity.addActivity({
companyId,
actor: "system",
action: "heartbeat.skip_cleared",
entityType: "agent",
entityId: agentId,
detail: JSON.stringify({
agentName,
reason: previous.reason,
detail: previous.detail,
streak: previous.count,
since: previous.since,
resumedAt: new Date().toISOString(),
}),
});
} catch (error) {
logger.warn(`[heartbeat] Failed to append heartbeat.skip_cleared activity for ${agentName}:`, error);
}
}
function ensureAgentHeartbeatEnabled(agentId: string) {
const agent = db.core.entity.getAgent(agentId);
@@ -300,8 +371,9 @@ export function createHeartbeat(
// (e.g. a future automation action that forwards the `toAgentId`
// raw) from spamming the log with "Agent not found: __board__".
if (agentId === HUMAN_SENDER_ID) return null;
let agent: AgentRecord;
try {
const agent = db.core.entity.getAgent(agentId);
agent = db.core.entity.getAgent(agentId);
if (agent.companyId !== companyId) {
logger.warn(`[wake] Refusing cross-company wake for agent ${agentId}: ${agent.companyId} !== ${companyId}`);
return null;
@@ -325,6 +397,7 @@ export function createHeartbeat(
const result = startAgentHeartbeatRun(agentId, companyId, trigger);
if (result.status === "skipped") {
recordSkipActivity(agentId, companyId, agent.name, result.reason, result.detail, trigger);
db.automation.wakeup.enqueueWakeupRequest(
companyId,
agentId,
@@ -334,6 +407,7 @@ export function createHeartbeat(
);
return null;
}
clearSkipActivity(agentId, companyId, agent.name);
db.automation.wakeup.completePendingWakeupRequests(agentId, companyId, result.runId);
return result.runId;
}
src/main/orchestration/index.ts
-1
View File
@@ -22,7 +22,6 @@ export function createOrchestrationService(deps: OrchestrationDependencies) {
const scheduler = createScheduler(
deps,
hasActiveRun,
heartbeat.startAgentHeartbeatRun,
heartbeat.wakeAgentIfPossible,
);
const taskRunner = createTaskRunner(
src/main/orchestration/scheduler-heartbeat.ts
-277
View File
@@ -1,277 +0,0 @@
import { powerMonitor } from "electron";
import { logAppError } from "../errors/index.js";
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) => HeartbeatStartResult,
) {
const { db, logger, dispatchAutomation, 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>();
const skipStreakByAgent = new Map<string, { reason: string; detail: string; count: number; since: string }>();
/**
* Periodic heartbeat.skip_ongoing emission gate. The activity table
* would otherwise show a 5-day gap of blankness while a stall
* persists, because `writeSkipActivity` used to fire only on
* first-seen + reason-change. Every SKIP_ONGOING_EVERY_N skips of the
* same reason we emit one `heartbeat.skip_ongoing` row so the audit
* timeline carries the running count. At 15s ticks, N=40 ≈ once
* every ~10 minutes of sustained skipping — readable, not floody.
*/
const SKIP_ONGOING_EVERY_N = 40;
/**
* Structured skip write to the activity stream. Emits on first skip,
* on reason change, and every SKIP_ONGOING_EVERY_N consecutive skips
* of the same reason. That periodic row stops the "5 days of blank
* activity" pathology — after the initial `heartbeat.skipped` the
* 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 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 && previous.detail === detail;
const nextStreak = sameReason
? { reason, detail, count: previous.count + 1, since: previous.since }
: { reason, detail, count: 1, since: tickAt };
skipStreakByAgent.set(agentId, nextStreak);
const action = !sameReason
? "heartbeat.skipped"
: nextStreak.count % SKIP_ONGOING_EVERY_N === 0
? "heartbeat.skip_ongoing"
: null;
if (action) {
try {
db.core.activity.addActivity({
companyId,
actor: "system",
action,
entityType: "agent",
entityId: agentId,
detail: JSON.stringify({
agentName,
reason,
detail,
streak: nextStreak.count,
since: nextStreak.since,
trigger,
}),
});
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.activity_write_failed",
message: `Failed to append ${action} activity for ${agentName}`,
context: { agentId, companyId, reason },
}, "warn");
}
}
return { count: nextStreak.count, since: nextStreak.since };
}
function clearSkipStreak(agentId: string, companyId: string, agentName: string) {
const previous = skipStreakByAgent.get(agentId);
if (!previous) return;
skipStreakByAgent.delete(agentId);
try {
db.core.activity.addActivity({
companyId,
actor: "system",
action: "heartbeat.skip_cleared",
entityType: "agent",
entityId: agentId,
detail: JSON.stringify({
agentName,
reason: previous.reason,
detail: previous.detail,
streak: previous.count,
since: previous.since,
resumedAt: new Date().toISOString(),
}),
});
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.activity_write_failed",
message: `Failed to append heartbeat skip-cleared activity for ${agentName}`,
context: { agentId, companyId },
}, "warn");
}
}
function tryWake(agentId: string, companyId: string, agentName: string, trigger: string) {
const tickAt = new Date().toISOString();
attemptedAtByAgent.set(agentId, Date.now());
const result = startAgentHeartbeatRun(agentId, companyId, trigger);
if (result.status === "skipped") {
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 {
clearSkipStreak(agentId, companyId, agentName);
}
}
function runTick() {
try {
if (isQuitting()) return;
const activeCompanies = db.core.company.listActiveCompanyIds();
const now = Date.now();
for (const company of activeCompanies) {
try {
const eligibleAgents = db.core.agents.listHeartbeatEligibleAgents(company.id);
for (const agent of eligibleAgents) {
if (hasActiveRun({ agentId: agent.id, companyId: company.id })) continue;
// Unified co-founder mode pause: checkpoint engine checks mode + wait states + pending approvals
if (deps.checkpointEngine?.shouldPauseHeartbeat(company.id, agent.id)) {
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 - lastTouch < intervalMs) continue;
tryWake(agent.id, company.id, agent.name, "timer");
}
const goalCount = db.core.entity.countActiveGoalsForCompany(company.id);
// Bootstrap when the company has active goals but no work is
// actually advancing. The previous gate ("taskCount === 0")
// missed the case where every task is in_review or blocked —
// the company looks busy to the counter but produces zero
// progress, and the CEO would not be woken until the next
// regular timer interval. countAdvancingTasksForCompany
// counts only todo + in_progress so bootstrapping still
// fires once the team stalls.
const advancingCount = db.core.tasks.countAdvancingTasksForCompany(company.id);
if (goalCount > 0 && advancingCount === 0) {
const ceo = db.core.agents.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 - ceoLastTouch > bootstrapCooldown)) {
logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks — bootstrapping CEO ${ceo.name}`);
tryWake(ceo.id, company.id, ceo.name, "bootstrap");
}
}
const scheduleKey = `schedule-${company.id}`;
const lastScheduleFired = scheduleLastFiredMap.get(scheduleKey) ?? 0;
if (now - lastScheduleFired >= 60_000) {
dispatchAutomation("schedule", { companyId: company.id });
scheduleLastFiredMap.set(scheduleKey, now);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.company_tick_failed",
message: `Heartbeat tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.tick_failed",
message: "Heartbeat scheduler tick raised; next interval will retry",
});
}
}
/**
* Re-arm the interval after OS resume. On Windows (and to a lesser
* extent macOS App Nap), `setInterval` is throttled to ≥1 minute and
* sometimes stalls entirely across sleep/standby cycles — which is
* exactly why agents went 4+ days without a single successful
* heartbeat. We can't trust the handle to fire on schedule after
* resume, so we clear it, run one tick immediately to catch up, and
* restart the interval with a fresh timer. Idempotent; safe to call
* during normal operation.
*/
function rearm(trigger: string) {
if (isQuitting()) return;
logger.info(`[heartbeat] Re-arming scheduler (${trigger})`);
if (timer) clearInterval(timer);
timer = setInterval(runTick, 15_000);
runTick();
}
let resumeListener: (() => void) | null = null;
let unlockListener: (() => void) | null = null;
function start() {
if (timer) clearInterval(timer);
timer = setInterval(runTick, 15_000);
// Windows/macOS: when the OS resumes from sleep or the screen
// unlocks, setInterval timers set before sleep are unreliable
// (Windows modern-standby throttles background timers to minute
// granularity and can drop pending fires entirely). Listening to
// powerMonitor is the documented Electron path to get a
// deterministic kick that doesn't depend on a user clicking the
// window. Without this the CEO heartbeat silently stops firing
// whenever the laptop lid closes overnight.
if (!resumeListener) {
resumeListener = () => rearm("power_resume");
powerMonitor.on("resume", resumeListener);
}
if (!unlockListener) {
unlockListener = () => rearm("screen_unlock");
// `unlock-screen` fires on Windows/macOS when the user returns
// from a locked screen without a full sleep cycle — common path
// for "I left the app open, locked, came back, agents dead".
powerMonitor.on("unlock-screen", unlockListener);
}
}
function stop() {
if (timer) clearInterval(timer);
timer = null;
if (resumeListener) {
powerMonitor.removeListener("resume", resumeListener);
resumeListener = null;
}
if (unlockListener) {
powerMonitor.removeListener("unlock-screen", unlockListener);
unlockListener = null;
}
}
return { start, stop };
}
src/main/orchestration/scheduler-housekeeping.ts
-142
View File
@@ -1,142 +0,0 @@
import type { OrchestrationDependencies } from "./types";
import { sweepStaleReviews } from "../task-quality-gates";
import type { SweepStaleReviewsDeps } from "../task-quality-gates";
import { resolveApprovalDecision } from "../approval-policy";
import { dispatchAgentMessage } from "../messaging/dispatch";
import { logAppError } from "../errors/index.js";
export function createHousekeepingScheduler(
deps: OrchestrationDependencies,
wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => string | null | undefined,
) {
const {
db,
logger,
publishDomainChanged,
notify,
hasWorkerReady,
isQuitting,
budgetService,
} = deps;
let timer: ReturnType<typeof setInterval> | null = null;
function start() {
if (timer) clearInterval(timer);
timer = setInterval(() => {
try {
if (isQuitting() || !hasWorkerReady()) return;
const activeCompanies = db.core.company.listActiveCompanyIds();
for (const company of activeCompanies) {
try {
const currentMonth = new Date().toISOString().slice(0, 7);
const lastResetMonth = db.core.settings.getLastBudgetResetMonth(company.id) ?? "";
if (currentMonth !== lastResetMonth) {
db.core.agents.resetMonthlyBudgets(company.id);
db.core.settings.setLastBudgetResetMonth(company.id, currentMonth);
logger.info(`[housekeeping] Reset monthly budgets for company ${company.name} (${currentMonth})`);
}
// D-06: Auto-close incidents from previous budget windows
const closedIncidents = budgetService.autoCloseExpiredIncidents(company.id);
if (closedIncidents > 0) {
logger.info(`[housekeeping] Auto-closed ${closedIncidents} expired budget incident(s) for company ${company.name}`);
}
// TASK-04: Stale review escalation sweep
try {
const sweepDeps: SweepStaleReviewsDeps = {
db,
dispatchAgentMessage,
resolveApprovalDecision,
notify,
getCompanyExecutionMode: (cId: string) => db.core.company.getCompanyExecutionMode(cId),
handleTaskStatusChange: (taskId: string, cId: string, previousStatus: string, newStatus: string) => {
deps.dispatchAutomation("task_status_changed", { companyId: cId, taskId, previousStatus, newStatus });
publishDomainChanged();
},
handleApprovalResolved: (approvalId: string, cId: string, _decision: string) => {
deps.dispatchAutomation("approval_resolved", { companyId: cId });
publishDomainChanged();
},
emitEvent: deps.emitEvent,
publishDomainChanged,
wakeAgentIfPossible: (agentId: string | null | undefined, cId: string, trigger: string) =>
agentId != null ? wakeAgentIfPossible(agentId, cId, trigger) ?? null : null,
logger,
};
const escalated = sweepStaleReviews(company.id, sweepDeps);
if (escalated > 0) {
logger.info(`[housekeeping] Auto-escalated ${escalated} stale review(s) for company ${company.name}`);
publishDomainChanged();
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.stale_review_sweep_failed",
message: `Stale review sweep failed for company ${company.id}`,
context: { companyId: company.id },
});
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.company_tick_failed",
message: `Housekeeping tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
try {
const staleWakeups = db.automation.wakeup.purgeExpiredWakeupRequests();
if (staleWakeups > 0) {
logger.info(`[housekeeping] Purged ${staleWakeups} expired wakeup request(s)`);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.wakeup_purge_failed",
message: "Expired wakeup request purge failed; next tick will retry",
});
}
try {
// agent_message_projections accumulates one row per (inbox
// message, session) forever. Once a session has rotated away
// — no agent references it as its current session_id — its
// projection rows only consume disk. Keep a 24h safety margin
// so an active rotation window cannot strip the ledger out
// from under a live run.
const cutoff = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();
const purgedProjections = db.projection.inbox.purgeOrphanProjections(cutoff);
if (purgedProjections > 0) {
logger.info(`[housekeeping] Purged ${purgedProjections} orphan inbox projection ledger row(s)`);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.inbox_projection_purge_failed",
message: "Orphan inbox projection purge failed; next tick will retry",
});
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.tick_failed",
message: "Housekeeping scheduler tick raised; next interval will retry",
});
}
}, 300_000);
}
function stop() {
if (timer) clearInterval(timer);
timer = null;
}
return { start, stop };
}
src/main/orchestration/scheduler-idle-automation.ts
-60
View File
@@ -1,60 +0,0 @@
import { logAppError } from "../errors/index.js";
import type { OrchestrationDependencies } from "./types";
export function createIdleAutomationScheduler(deps: OrchestrationDependencies) {
const { db, logger, dispatchAutomation, hasWorkerReady, isQuitting } = deps;
let timer: ReturnType<typeof setInterval> | null = null;
function start() {
if (timer) clearInterval(timer);
timer = setInterval(() => {
try {
if (isQuitting() || !hasWorkerReady()) return;
const activeCompanies = db.core.company.listActiveCompanyIds();
const now = Date.now();
for (const company of activeCompanies) {
try {
const companyAgents = db.core.agents.listActiveAgentsForCompany(company.id);
for (const agent of companyAgents) {
if (!db.core.agents.hasAgentAssignedTasks(agent.id, company.id)) {
const lastActive = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
const idleThreshold = 20 * 60 * 1000;
if (now - lastActive > idleThreshold) {
dispatchAutomation("agent_idle", {
companyId: company.id,
agentId: agent.id,
department: agent.department ?? undefined,
});
}
}
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.idle_automation.company_tick_failed",
message: `Idle automation tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.idle_automation.tick_failed",
message: "Idle automation scheduler tick raised; next interval will retry",
});
}
}, 60_000);
}
function stop() {
if (timer) clearInterval(timer);
timer = null;
}
return { start, stop };
}
src/main/orchestration/scheduler-lock-recovery.ts
-67
View File
@@ -1,67 +0,0 @@
import { logAppError } from "../errors/index.js";
import type { OrchestrationDependencies } from "./types";
export function createLockRecoveryScheduler(deps: OrchestrationDependencies) {
const { db, logger, publishDomainChanged, hasWorkerReady, isQuitting } = deps;
let timer: ReturnType<typeof setInterval> | null = null;
function start() {
if (timer) clearInterval(timer);
timer = setInterval(() => {
try {
if (isQuitting() || !hasWorkerReady()) return;
// D-02: Stale lock recovery (EXEC-03)
try {
const recovered = db.recoverStaleLocks();
for (const task of recovered) {
db.core.activity.addActivity({
companyId: task.companyId,
actor: "system",
action: "execution.lock_recovered",
entityType: "task",
entityId: task.taskId,
detail: `Recovered stale lock on task "${task.title}" after 15min timeout`,
});
logger.info(`[lock-recovery] Recovered stale lock on task ${task.taskId}: "${task.title}"`);
}
} catch (err) {
logAppError(logger, err, {
kind: "recoverable",
code: "scheduler.lock_recovery.stale_lock_sweep_failed",
message: "Stale task lock recovery sweep failed; next tick will retry",
});
}
try {
const staleReleased = db.core.workspaces.releaseStaleWorkspaceLocks();
if (staleReleased > 0) {
logger.info(`[lock-recovery] Released ${staleReleased} stale workspace lock(s)`);
publishDomainChanged();
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.lock_recovery.workspace_lock_sweep_failed",
message: "Stale workspace lock cleanup failed; next tick will retry",
});
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.lock_recovery.tick_failed",
message: "Lock recovery scheduler tick raised; next interval will retry",
});
}
}, 60_000);
}
function stop() {
if (timer) clearInterval(timer);
timer = null;
}
return { start, stop };
}
src/main/orchestration/scheduler.ts
+297 -18
View File
@@ -1,33 +1,312 @@
import { powerMonitor } from "electron";
import { logAppError } from "../errors/index.js";
import { resolveApprovalDecision } from "../approval-policy";
import { dispatchAgentMessage } from "../messaging/dispatch";
import { sweepStaleReviews, type SweepStaleReviewsDeps } from "../task-quality-gates";
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";
import { createHousekeepingScheduler } from "./scheduler-housekeeping";
type SchedulerPhase = "lock_recovery" | "housekeeping" | "idle_automation" | "heartbeat_cadence";
const TICK_INTERVAL_MS = 10_000;
const PHASE_INTERVAL_MS: Record<SchedulerPhase, number> = {
lock_recovery: 60_000,
housekeeping: 300_000,
idle_automation: 60_000,
heartbeat_cadence: 60_000,
};
export function createScheduler(
deps: OrchestrationDependencies,
hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean,
startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => HeartbeatStartResult,
wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => string | null | undefined,
) {
const heartbeat = createHeartbeatScheduler(deps, hasActiveRun, startAgentHeartbeatRun);
const lockRecovery = createLockRecoveryScheduler(deps);
const idleAutomation = createIdleAutomationScheduler(deps);
const housekeeping = createHousekeepingScheduler(deps, wakeAgentIfPossible);
const {
db,
logger,
publishDomainChanged,
dispatchAutomation,
hasWorkerReady,
isQuitting,
notify,
budgetService,
} = deps;
let timer: ReturnType<typeof setInterval> | null = null;
let resumeListener: (() => void) | null = null;
let unlockListener: (() => void) | null = null;
const phaseLastRunAt = new Map<SchedulerPhase, number>();
const scheduleLastFiredByCompany = new Map<string, number>();
const bootstrapAttemptByAgent = new Map<string, number>();
function phaseIsDue(phase: SchedulerPhase, now: number): boolean {
const lastRunAt = phaseLastRunAt.get(phase) ?? 0;
if (now - lastRunAt < PHASE_INTERVAL_MS[phase]) return false;
phaseLastRunAt.set(phase, now);
return true;
}
function runLockRecoveryPhase() {
if (!hasWorkerReady()) return;
try {
const recovered = db.recoverStaleLocks();
for (const task of recovered) {
db.core.activity.addActivity({
companyId: task.companyId,
actor: "system",
action: "execution.lock_recovered",
entityType: "task",
entityId: task.taskId,
detail: `Recovered stale lock on task "${task.title}" after 15min timeout`,
});
logger.info(`[lock-recovery] Recovered stale lock on task ${task.taskId}: "${task.title}"`);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.lock_recovery.stale_lock_sweep_failed",
message: "Stale task lock recovery sweep failed; next tick will retry",
});
}
try {
const staleReleased = db.core.workspaces.releaseStaleWorkspaceLocks();
if (staleReleased > 0) {
logger.info(`[lock-recovery] Released ${staleReleased} stale workspace lock(s)`);
publishDomainChanged();
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.lock_recovery.workspace_lock_sweep_failed",
message: "Stale workspace lock cleanup failed; next tick will retry",
});
}
}
function runHousekeepingPhase() {
if (!hasWorkerReady()) return;
const activeCompanies = db.core.company.listActiveCompanyIds();
for (const company of activeCompanies) {
try {
const currentMonth = new Date().toISOString().slice(0, 7);
const lastResetMonth = db.core.settings.getLastBudgetResetMonth(company.id) ?? "";
if (currentMonth !== lastResetMonth) {
db.core.agents.resetMonthlyBudgets(company.id);
db.core.settings.setLastBudgetResetMonth(company.id, currentMonth);
logger.info(`[housekeeping] Reset monthly budgets for company ${company.name} (${currentMonth})`);
}
const closedIncidents = budgetService.autoCloseExpiredIncidents(company.id);
if (closedIncidents > 0) {
logger.info(`[housekeeping] Auto-closed ${closedIncidents} expired budget incident(s) for company ${company.name}`);
}
try {
const sweepDeps: SweepStaleReviewsDeps = {
db,
dispatchAgentMessage,
resolveApprovalDecision,
notify,
getCompanyExecutionMode: (companyId: string) => db.core.company.getCompanyExecutionMode(companyId),
handleTaskStatusChange: (taskId: string, companyId: string, previousStatus: string, newStatus: string) => {
dispatchAutomation("task_status_changed", { companyId, taskId, previousStatus, newStatus });
publishDomainChanged();
},
handleApprovalResolved: (approvalId: string, companyId: string, _decision: string) => {
dispatchAutomation("approval_resolved", { companyId });
publishDomainChanged();
},
emitEvent: deps.emitEvent,
publishDomainChanged,
wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) =>
agentId != null ? wakeAgentIfPossible(agentId, companyId, trigger) ?? null : null,
logger,
};
const escalated = sweepStaleReviews(company.id, sweepDeps);
if (escalated > 0) {
logger.info(`[housekeeping] Auto-escalated ${escalated} stale review(s) for company ${company.name}`);
publishDomainChanged();
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.stale_review_sweep_failed",
message: `Stale review sweep failed for company ${company.id}`,
context: { companyId: company.id },
});
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.company_tick_failed",
message: `Housekeeping tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
try {
const staleWakeups = db.automation.wakeup.purgeExpiredWakeupRequests();
if (staleWakeups > 0) {
logger.info(`[housekeeping] Purged ${staleWakeups} expired wakeup request(s)`);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.wakeup_purge_failed",
message: "Expired wakeup request purge failed; next tick will retry",
});
}
try {
const cutoff = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();
const purgedProjections = db.projection.inbox.purgeOrphanProjections(cutoff);
if (purgedProjections > 0) {
logger.info(`[housekeeping] Purged ${purgedProjections} orphan inbox projection ledger row(s)`);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.housekeeping.inbox_projection_purge_failed",
message: "Orphan inbox projection purge failed; next tick will retry",
});
}
}
function runIdleAutomationPhase() {
if (!hasWorkerReady()) return;
const activeCompanies = db.core.company.listActiveCompanyIds();
const now = Date.now();
for (const company of activeCompanies) {
try {
const companyAgents = db.core.agents.listActiveAgentsForCompany(company.id);
for (const agent of companyAgents) {
if (db.core.agents.hasAgentAssignedTasks(agent.id, company.id)) continue;
const lastActive = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
const idleThreshold = 20 * 60 * 1000;
if (now - lastActive > idleThreshold) {
dispatchAutomation("agent_idle", {
companyId: company.id,
agentId: agent.id,
department: agent.department ?? undefined,
});
}
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.idle_automation.company_tick_failed",
message: `Idle automation tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
}
function runCompanyBootstrap(company: { id: string; name: string }, now: number) {
const goalCount = db.core.entity.countActiveGoalsForCompany(company.id);
const advancingCount = db.core.tasks.countAdvancingTasksForCompany(company.id);
if (goalCount === 0 || advancingCount > 0) return;
const ceo = db.core.agents.getCeoForCompany(company.id);
if (!ceo) return;
if (hasActiveRun({ agentId: ceo.id, companyId: company.id })) return;
const lastHeartbeat = ceo.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
const lastAttempt = bootstrapAttemptByAgent.get(ceo.id) ?? 0;
const bootstrapCooldown = 120_000;
if (now - Math.max(lastHeartbeat, lastAttempt) <= bootstrapCooldown) return;
bootstrapAttemptByAgent.set(ceo.id, now);
logger.info(`[scheduler] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks; bootstrapping CEO ${ceo.name}`);
wakeAgentIfPossible(ceo.id, company.id, "bootstrap");
}
function runScheduleAutomation(companyId: string, now: number) {
const lastScheduleFired = scheduleLastFiredByCompany.get(companyId) ?? 0;
if (now - lastScheduleFired < 60_000) return;
dispatchAutomation("schedule", { companyId });
scheduleLastFiredByCompany.set(companyId, now);
}
function runHeartbeatCadencePhase() {
const activeCompanies = db.core.company.listActiveCompanyIds();
const now = Date.now();
for (const company of activeCompanies) {
try {
runScheduleAutomation(company.id, now);
runCompanyBootstrap(company, now);
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat_cadence.company_tick_failed",
message: `Heartbeat cadence failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
}
function runPhase(phase: SchedulerPhase, runner: () => void) {
try {
runner();
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: `scheduler.${phase}.tick_failed`,
message: `Scheduler phase ${phase} raised; next interval will retry`,
});
}
}
function runTick() {
if (isQuitting()) return;
const now = Date.now();
if (phaseIsDue("lock_recovery", now)) runPhase("lock_recovery", runLockRecoveryPhase);
if (phaseIsDue("housekeeping", now)) runPhase("housekeeping", runHousekeepingPhase);
if (phaseIsDue("idle_automation", now)) runPhase("idle_automation", runIdleAutomationPhase);
if (phaseIsDue("heartbeat_cadence", now)) runPhase("heartbeat_cadence", runHeartbeatCadencePhase);
}
function armTimer() {
if (timer) clearInterval(timer);
timer = setInterval(runTick, TICK_INTERVAL_MS);
}
function rearm(trigger: string) {
if (isQuitting()) return;
logger.info(`[scheduler] Re-arming unified scheduler (${trigger})`);
armTimer();
runTick();
}
function startAllSchedulers() {
heartbeat.start();
lockRecovery.start();
idleAutomation.start();
housekeeping.start();
armTimer();
if (!resumeListener) {
resumeListener = () => rearm("power_resume");
powerMonitor.on("resume", resumeListener);
}
if (!unlockListener) {
unlockListener = () => rearm("screen_unlock");
powerMonitor.on("unlock-screen", unlockListener);
}
}
function stopAllSchedulers() {
heartbeat.stop();
lockRecovery.stop();
idleAutomation.stop();
housekeeping.stop();
if (timer) clearInterval(timer);
timer = null;
if (resumeListener) {
powerMonitor.removeListener("resume", resumeListener);
resumeListener = null;
}
if (unlockListener) {
powerMonitor.removeListener("unlock-screen", unlockListener);
unlockListener = null;
}
}
return { startAllSchedulers, stopAllSchedulers };
tests/unit/scheduler-heartbeat.test.ts
+91 -22
View File
@@ -17,7 +17,8 @@ vi.mock("electron", () => ({
}));
import { AppDatabase } from "../../src/main/database";
import { createHeartbeatScheduler } from "../../src/main/orchestration/scheduler-heartbeat";
import { createHeartbeat } from "../../src/main/orchestration/heartbeat";
import { createScheduler } from "../../src/main/orchestration/scheduler";
import type { OrchestrationDependencies } from "../../src/main/orchestration/types";
function createLogger() {
@@ -61,8 +62,8 @@ function createAgent(db: AppDatabase, companyId: string, workspaceId: string): s
const agentId = db.core.agents.saveAgent({
companyId,
name: "Scheduler Agent",
role: "Engineer",
title: "Runtime Engineer",
role: "CEO",
title: "Chief Executive Officer",
status: "active",
reportsTo: null,
connectorId: "openai-codex",
@@ -75,7 +76,7 @@ function createAgent(db: AppDatabase, companyId: string, workspaceId: string): s
return agentId;
}
function createDeps(db: AppDatabase): OrchestrationDependencies {
function createDeps(db: AppDatabase, overrides: Partial<OrchestrationDependencies> = {}): OrchestrationDependencies {
return {
db,
logger: createLogger(),
@@ -85,18 +86,20 @@ function createDeps(db: AppDatabase): OrchestrationDependencies {
createRunToken: () => "test-token",
getApiPort: () => 3210,
dispatchAutomation: vi.fn(),
hasWorkerReady: () => false,
hasWorkerReady: () => true,
enqueueWorkerRun: vi.fn(),
isQuitting: () => false,
listRuntimeSkillEntries: () => [],
budgetService: {
autoCloseExpiredIncidents: vi.fn(() => 0),
evaluateBudgetGate: vi.fn(() => ({ allowed: true, blockedBy: [] })),
} as never,
checkpointEngine: { shouldPauseHeartbeat: () => false },
...overrides,
};
}
describe("heartbeat scheduler", () => {
describe("unified scheduler heartbeat cadence", () => {
let tempDir: string;
let db: AppDatabase;
@@ -114,30 +117,23 @@ describe("heartbeat scheduler", () => {
vi.useRealTimers();
});
it("records worker readiness skips from the heartbeat run decision", () => {
it("records worker readiness skips from the heartbeat wake 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),
const heartbeat = createHeartbeat(
createDeps(db, { hasWorkerReady: () => false }),
() => {
throw new Error("resolveWorkspace should stay behind canRunAgent.");
},
() => null,
vi.fn(),
() => false,
startAgentHeartbeatRun,
);
try {
scheduler.start();
vi.advanceTimersByTime(15_000);
} finally {
scheduler.stop();
}
expect(heartbeat.wakeAgentIfPossible(agentId, companyId, "timer")).toBeNull();
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);
@@ -149,4 +145,77 @@ describe("heartbeat scheduler", () => {
});
expect(db.core.agents.getRuntimeState(agentId)).not.toHaveProperty("heartbeatSkip");
});
it("keeps ordinary agent heartbeat cadence event-driven", () => {
const companyId = createCompany(db);
createConnector(db);
const workspaceId = createWorkspace(db, companyId, tempDir);
createAgent(db, companyId, workspaceId);
const wakeAgentIfPossible = vi.fn();
const scheduler = createScheduler(
createDeps(db),
() => false,
wakeAgentIfPossible,
);
try {
scheduler.startAllSchedulers();
vi.advanceTimersByTime(60_000);
} finally {
scheduler.stopAllSchedulers();
}
expect(wakeAgentIfPossible).not.toHaveBeenCalled();
});
it("bootstraps the CEO through the wake pipeline", () => {
const companyId = createCompany(db);
createConnector(db);
const workspaceId = createWorkspace(db, companyId, tempDir);
const agentId = createAgent(db, companyId, workspaceId);
db.core.entity.saveGoal({
companyId,
title: "Launch the company",
description: "Create the first operating plan.",
status: "active",
ownerAgentId: agentId,
});
const wakeAgentIfPossible = vi.fn();
const scheduler = createScheduler(
createDeps(db, { hasWorkerReady: () => false }),
() => false,
wakeAgentIfPossible,
);
try {
scheduler.startAllSchedulers();
vi.advanceTimersByTime(60_000);
} finally {
scheduler.stopAllSchedulers();
}
expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "bootstrap");
});
it("keeps schedule cadence running while the worker is unavailable", () => {
const companyId = createCompany(db);
const dispatchAutomation = vi.fn();
const scheduler = createScheduler(
createDeps(db, {
dispatchAutomation,
hasWorkerReady: () => false,
}),
() => false,
vi.fn(),
);
try {
scheduler.startAllSchedulers();
vi.advanceTimersByTime(60_000);
} finally {
scheduler.stopAllSchedulers();
}
expect(dispatchAutomation).toHaveBeenCalledWith("schedule", { companyId });
});
});