refactor(scheduler): split 15s tick into 4 independent intervals

Split the single 15s setInterval in scheduler.ts (6 serial jobs) into 4 independent schedulers so a slow job no longer stalls the others: - scheduler-heartbeat.ts (15s): wake eligible agents, CEO bootstrap, schedule automation trigger - scheduler-lock-recovery.ts (60s): stale task-lock recovery and stale workspace-lock release - scheduler-idle-automation.ts (60s): idle-agent automation dispatch - scheduler-housekeeping.ts (300s): monthly budget reset, expired budget incident auto-close, stale review sweep, expired wakeup purge scheduler.ts is now a thin composer exposing startAllSchedulers / stopAllSchedulers. Bootstrap and service-graph updated accordingly. Business logic, error isolation, and atomic actions preserved.
2026-04-16 21:47:09 -04:00
1 parent 2676400e0c
commit f8052c3741
@@ -321,7 +321,7 @@ export async function bootstrapApplication({
    log.info(`[bootstrap] Connector health checks completed (${healthCheckResults.filter((r) => r.status === "fulfilled").length}/${connectors.length} succeeded)`);
  }

-  serviceGraph.getOrchestrationService()?.startHeartbeatScheduler();
+  serviceGraph.getOrchestrationService()?.startAllSchedulers();

  // Check for crash recovery -- emit recovery notification before resuming queued runs
  const interruptedCount = db.lastInterruptedCount;
@@ -40,7 +40,7 @@ export function createOrchestrationService(deps: OrchestrationDependencies) {
    queueTaskRun: taskRunner.queueTaskRun,
    resumeQueuedRuns: taskRunner.resumeQueuedRuns,
    startAgentHeartbeatRun: heartbeat.startAgentHeartbeatRun,
-    startHeartbeatScheduler: scheduler.startHeartbeatScheduler,
-    stopHeartbeatScheduler: scheduler.stopHeartbeatScheduler,
+    startAllSchedulers: scheduler.startAllSchedulers,
+    stopAllSchedulers: scheduler.stopAllSchedulers,
  };
 }
@@ -0,0 +1,77 @@
+import type { OrchestrationDependencies } from "./types";
+
+export function createHeartbeatScheduler(
+  deps: OrchestrationDependencies,
+  hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean,
+  startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => string | null,
+) {
+  const { db, logger, dispatchAutomation, hasWorkerReady, isQuitting } = deps;
+
+  let timer: ReturnType<typeof setInterval> | null = null;
+  const scheduleLastFiredMap = new Map<string, number>();
+
+  function start() {
+    if (timer) clearInterval(timer);
+
+    timer = setInterval(() => {
+      try {
+        if (isQuitting() || !hasWorkerReady()) return;
+
+        const activeCompanies = db.listActiveCompanyIds();
+        const now = Date.now();
+
+        for (const company of activeCompanies) {
+          try {
+            const eligibleAgents = db.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;
+              }
+
+              const lastHeartbeat = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
+              const intervalMs = agent.heartbeatIntervalSec * 1000;
+              if (now - lastHeartbeat < intervalMs) continue;
+
+              logger.info(`[heartbeat] Waking agent ${agent.name} (interval: ${agent.heartbeatIntervalSec}s)`);
+              startAgentHeartbeatRun(agent.id, company.id, "timer");
+            }
+
+            const goalCount = db.countActiveGoalsForCompany(company.id);
+            const taskCount = db.countActiveTasksForCompany(company.id);
+            if (goalCount > 0 && taskCount === 0) {
+              const ceo = db.getCeoForCompany(company.id);
+              const ceoLastWake = ceo?.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
+              const bootstrapCooldown = 120_000;
+              if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastWake > bootstrapCooldown)) {
+                logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goals but no tasks — bootstrapping CEO ${ceo.name}`);
+                startAgentHeartbeatRun(ceo.id, company.id, "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) {
+            logger.error(`[heartbeat] Error processing company ${company.id}: ${error instanceof Error ? error.message : String(error)}`);
+          }
+        }
+      } catch (error) {
+        logger.error(`[heartbeat] Scheduler error: ${error instanceof Error ? error.message : String(error)}`);
+      }
+    }, 15_000);
+  }
+
+  function stop() {
+    if (timer) clearInterval(timer);
+    timer = null;
+  }
+
+  return { start, stop };
+}
@@ -0,0 +1,103 @@
+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 "../message-dispatch";
+
+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.listActiveCompanyIds();
+
+        for (const company of activeCompanies) {
+          try {
+            const currentMonth = new Date().toISOString().slice(0, 7);
+            const lastResetMonth = db.getLastBudgetResetMonth(company.id) ?? "";
+            if (currentMonth !== lastResetMonth) {
+              db.resetMonthlyBudgets(company.id);
+              db.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.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) {
+              logger.error(`[housekeeping] Stale review sweep failed: ${error instanceof Error ? error.message : String(error)}`);
+            }
+          } catch (error) {
+            logger.error(`[housekeeping] Error processing company ${company.id}: ${error instanceof Error ? error.message : String(error)}`);
+          }
+        }
+
+        try {
+          const staleWakeups = db.purgeExpiredWakeupRequests();
+          if (staleWakeups > 0) {
+            logger.info(`[housekeeping] Purged ${staleWakeups} expired wakeup request(s)`);
+          }
+        } catch (error) {
+          logger.error(`[housekeeping] Wakeup purge failed: ${error instanceof Error ? error.message : String(error)}`);
+        }
+      } catch (error) {
+        logger.error(`[housekeeping] Scheduler error: ${error instanceof Error ? error.message : String(error)}`);
+      }
+    }, 300_000);
+  }
+
+  function stop() {
+    if (timer) clearInterval(timer);
+    timer = null;
+  }
+
+  return { start, stop };
+}
@@ -0,0 +1,50 @@
+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.listActiveCompanyIds();
+        const now = Date.now();
+
+        for (const company of activeCompanies) {
+          try {
+            const companyAgents = db.listActiveAgentsForCompany(company.id);
+            for (const agent of companyAgents) {
+              if (!db.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) {
+            logger.error(`[idle-automation] Error processing company ${company.id}: ${error instanceof Error ? error.message : String(error)}`);
+          }
+        }
+      } catch (error) {
+        logger.error(`[idle-automation] Scheduler error: ${error instanceof Error ? error.message : String(error)}`);
+      }
+    }, 60_000);
+  }
+
+  function stop() {
+    if (timer) clearInterval(timer);
+    timer = null;
+  }
+
+  return { start, stop };
+}
@@ -0,0 +1,54 @@
+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.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) {
+          logger.error(`[lock-recovery] Stale lock recovery failed: ${err instanceof Error ? err.message : String(err)}`);
+        }
+
+        try {
+          const staleReleased = db.releaseStaleWorkspaceLocks();
+          if (staleReleased > 0) {
+            logger.info(`[lock-recovery] Released ${staleReleased} stale workspace lock(s)`);
+            publishDomainChanged();
+          }
+        } catch (error) {
+          logger.error(`[lock-recovery] Stale workspace lock cleanup failed: ${error instanceof Error ? error.message : String(error)}`);
+        }
+      } catch (error) {
+        logger.error(`[lock-recovery] Scheduler error: ${error instanceof Error ? error.message : String(error)}`);
+      }
+    }, 60_000);
+  }
+
+  function stop() {
+    if (timer) clearInterval(timer);
+    timer = null;
+  }
+
+  return { start, stop };
+}
@@ -1,8 +1,8 @@
 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 "../message-dispatch";
+import { createHeartbeatScheduler } from "./scheduler-heartbeat";
+import { createLockRecoveryScheduler } from "./scheduler-lock-recovery";
+import { createIdleAutomationScheduler } from "./scheduler-idle-automation";
+import { createHousekeepingScheduler } from "./scheduler-housekeeping";

 export function createScheduler(
  deps: OrchestrationDependencies,
@@ -10,174 +10,24 @@ export function createScheduler(
  startAgentHeartbeatRun: (agentId: string, companyId: string, trigger: string) => string | null,
  wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => string | null | undefined,
 ) {
-  const {
-    db,
-    logger,
-    publishDomainChanged,
-    notify,
-    dispatchAutomation,
-    hasWorkerReady,
-    isQuitting,
-    budgetService,
-  } = deps;
+  const heartbeat = createHeartbeatScheduler(deps, hasActiveRun, startAgentHeartbeatRun);
+  const lockRecovery = createLockRecoveryScheduler(deps);
+  const idleAutomation = createIdleAutomationScheduler(deps);
+  const housekeeping = createHousekeepingScheduler(deps, wakeAgentIfPossible);

-  let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
-  const scheduleLastFiredMap = new Map<string, number>();
-
-  function startHeartbeatScheduler() {
-    if (heartbeatTimer) clearInterval(heartbeatTimer);
-
-    heartbeatTimer = setInterval(() => {
-      try {
-        if (isQuitting() || !hasWorkerReady()) return;
-
-        // D-02: Stale lock recovery piggybacked on heartbeat cycle (EXEC-03)
-        try {
-          const recovered = db.recoverStaleLocks();
-          for (const task of recovered) {
-            db.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(`[heartbeat] Recovered stale lock on task ${task.taskId}: "${task.title}"`);
-          }
-        } catch (err) {
-          logger.error(`[heartbeat] Stale lock recovery failed: ${err instanceof Error ? err.message : String(err)}`);
-        }
-
-        const activeCompanies = db.listActiveCompanyIds();
-        const now = Date.now();
-
-        for (const company of activeCompanies) {
-          try {
-            const eligibleAgents = db.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;
-              }
-
-              const lastHeartbeat = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
-              const intervalMs = agent.heartbeatIntervalSec * 1000;
-              if (now - lastHeartbeat < intervalMs) continue;
-
-              logger.info(`[heartbeat] Waking agent ${agent.name} (interval: ${agent.heartbeatIntervalSec}s)`);
-              startAgentHeartbeatRun(agent.id, company.id, "timer");
-            }
-
-            const goalCount = db.countActiveGoalsForCompany(company.id);
-            const taskCount = db.countActiveTasksForCompany(company.id);
-            if (goalCount > 0 && taskCount === 0) {
-              const ceo = db.getCeoForCompany(company.id);
-              const ceoLastWake = ceo?.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
-              const bootstrapCooldown = 120_000;
-              if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastWake > bootstrapCooldown)) {
-                logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goals but no tasks — bootstrapping CEO ${ceo.name}`);
-                startAgentHeartbeatRun(ceo.id, company.id, "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);
-            }
-
-            const companyAgents = db.listActiveAgentsForCompany(company.id);
-            for (const agent of companyAgents) {
-              if (!db.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,
-                  });
-                }
-              }
-            }
-
-            const currentMonth = new Date().toISOString().slice(0, 7);
-            const lastResetMonth = db.getLastBudgetResetMonth(company.id) ?? "";
-            if (currentMonth !== lastResetMonth) {
-              db.resetMonthlyBudgets(company.id);
-              db.setLastBudgetResetMonth(company.id, currentMonth);
-              logger.info(`[heartbeat] 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(`[heartbeat] 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.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(`[heartbeat] Auto-escalated ${escalated} stale review(s) for company ${company.name}`);
-                publishDomainChanged();
-              }
-            } catch (error) {
-              logger.error(`[heartbeat] Stale review sweep failed: ${error instanceof Error ? error.message : String(error)}`);
-            }
-          } catch (error) {
-            logger.error(`[heartbeat] Error processing company ${company.id}: ${error instanceof Error ? error.message : String(error)}`);
-          }
-        }
-
-        try {
-          const staleReleased = db.releaseStaleWorkspaceLocks();
-          if (staleReleased > 0) {
-            logger.info(`[heartbeat] Released ${staleReleased} stale workspace lock(s)`);
-            publishDomainChanged();
-          }
-          const staleWakeups = db.purgeExpiredWakeupRequests();
-          if (staleWakeups > 0) {
-            logger.info(`[heartbeat] Purged ${staleWakeups} expired wakeup request(s)`);
-          }
-        } catch (error) {
-          logger.error(`[heartbeat] Stale lock cleanup failed: ${error instanceof Error ? error.message : String(error)}`);
-        }
-      } catch (error) {
-        logger.error(`[heartbeat] Scheduler error: ${error instanceof Error ? error.message : String(error)}`);
-      }
-    }, 15_000);
+  function startAllSchedulers() {
+    heartbeat.start();
+    lockRecovery.start();
+    idleAutomation.start();
+    housekeeping.start();
  }

-  function stopHeartbeatScheduler() {
-    if (heartbeatTimer) clearInterval(heartbeatTimer);
-    heartbeatTimer = null;
+  function stopAllSchedulers() {
+    heartbeat.stop();
+    lockRecovery.stop();
+    idleAutomation.stop();
+    housekeeping.stop();
  }

-  return { startHeartbeatScheduler, stopHeartbeatScheduler };
+  return { startAllSchedulers, stopAllSchedulers };
 }
@@ -321,7 +321,7 @@ export function createServiceGraph({
      shouldQuit: () => isQuittingRef.current,
      onBeforeQuit: () => {
        isQuittingRef.current = true;
-        orchestrationService?.stopHeartbeatScheduler();
+        orchestrationService?.stopAllSchedulers();
        automationService?.destroy();
        apiServerRef.current?.stop();
        apiServerRef.current = null;