From ead9475b71d80d872bf52aee258cf675c55066e7 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:26:17 -0400 Subject: [PATCH] refactor(scheduler): unify cadence phases --- src/main/orchestration/heartbeat.ts | 76 ++++- src/main/orchestration/index.ts | 1 - src/main/orchestration/scheduler-heartbeat.ts | 277 --------------- .../orchestration/scheduler-housekeeping.ts | 142 -------- .../scheduler-idle-automation.ts | 60 ---- .../orchestration/scheduler-lock-recovery.ts | 67 ---- src/main/orchestration/scheduler.ts | 315 +++++++++++++++++- tests/unit/scheduler-heartbeat.test.ts | 113 +++++-- 8 files changed, 463 insertions(+), 588 deletions(-) delete mode 100644 src/main/orchestration/scheduler-heartbeat.ts delete mode 100644 src/main/orchestration/scheduler-housekeeping.ts delete mode 100644 src/main/orchestration/scheduler-idle-automation.ts delete mode 100644 src/main/orchestration/scheduler-lock-recovery.ts diff --git a/src/main/orchestration/heartbeat.ts b/src/main/orchestration/heartbeat.ts index 092745f2..e68c6660 100644 --- a/src/main/orchestration/heartbeat.ts +++ b/src/main/orchestration/heartbeat.ts @@ -54,6 +54,77 @@ export function createHeartbeat( hasWorkerReady, isQuitting, } = deps; + const skipStreakByAgent = new Map(); + 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; } diff --git a/src/main/orchestration/index.ts b/src/main/orchestration/index.ts index 20055681..fac44569 100644 --- a/src/main/orchestration/index.ts +++ b/src/main/orchestration/index.ts @@ -22,7 +22,6 @@ export function createOrchestrationService(deps: OrchestrationDependencies) { const scheduler = createScheduler( deps, hasActiveRun, - heartbeat.startAgentHeartbeatRun, heartbeat.wakeAgentIfPossible, ); const taskRunner = createTaskRunner( diff --git a/src/main/orchestration/scheduler-heartbeat.ts b/src/main/orchestration/scheduler-heartbeat.ts deleted file mode 100644 index 99768ac1..00000000 --- a/src/main/orchestration/scheduler-heartbeat.ts +++ /dev/null @@ -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 | null = null; - const scheduleLastFiredMap = new Map(); - /** - * 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(); - const skipStreakByAgent = new Map(); - - /** - * 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 }; -} diff --git a/src/main/orchestration/scheduler-housekeeping.ts b/src/main/orchestration/scheduler-housekeeping.ts deleted file mode 100644 index 804b54b7..00000000 --- a/src/main/orchestration/scheduler-housekeeping.ts +++ /dev/null @@ -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 | 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 }; -} diff --git a/src/main/orchestration/scheduler-idle-automation.ts b/src/main/orchestration/scheduler-idle-automation.ts deleted file mode 100644 index 1ddd4f3b..00000000 --- a/src/main/orchestration/scheduler-idle-automation.ts +++ /dev/null @@ -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 | 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 }; -} diff --git a/src/main/orchestration/scheduler-lock-recovery.ts b/src/main/orchestration/scheduler-lock-recovery.ts deleted file mode 100644 index da3d3a5d..00000000 --- a/src/main/orchestration/scheduler-lock-recovery.ts +++ /dev/null @@ -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 | 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 }; -} diff --git a/src/main/orchestration/scheduler.ts b/src/main/orchestration/scheduler.ts index 825530ed..252e4bfa 100644 --- a/src/main/orchestration/scheduler.ts +++ b/src/main/orchestration/scheduler.ts @@ -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 = { + 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 | null = null; + let resumeListener: (() => void) | null = null; + let unlockListener: (() => void) | null = null; + const phaseLastRunAt = new Map(); + const scheduleLastFiredByCompany = new Map(); + const bootstrapAttemptByAgent = new Map(); + + 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 }; diff --git a/tests/unit/scheduler-heartbeat.test.ts b/tests/unit/scheduler-heartbeat.test.ts index 21e04db4..abddfabb 100644 --- a/tests/unit/scheduler-heartbeat.test.ts +++ b/tests/unit/scheduler-heartbeat.test.ts @@ -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 { 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 }); + }); });