diff --git a/src/main/bootstrap-application.ts b/src/main/bootstrap-application.ts index cb4a2933..bc13352b 100644 --- a/src/main/bootstrap-application.ts +++ b/src/main/bootstrap-application.ts @@ -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; diff --git a/src/main/orchestration/index.ts b/src/main/orchestration/index.ts index 735344fa..84159dd3 100644 --- a/src/main/orchestration/index.ts +++ b/src/main/orchestration/index.ts @@ -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, }; } diff --git a/src/main/orchestration/scheduler-heartbeat.ts b/src/main/orchestration/scheduler-heartbeat.ts new file mode 100644 index 00000000..d386404f --- /dev/null +++ b/src/main/orchestration/scheduler-heartbeat.ts @@ -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 | null = null; + const scheduleLastFiredMap = new Map(); + + 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 }; +} diff --git a/src/main/orchestration/scheduler-housekeeping.ts b/src/main/orchestration/scheduler-housekeeping.ts new file mode 100644 index 00000000..3023c4d2 --- /dev/null +++ b/src/main/orchestration/scheduler-housekeeping.ts @@ -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 | 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 }; +} diff --git a/src/main/orchestration/scheduler-idle-automation.ts b/src/main/orchestration/scheduler-idle-automation.ts new file mode 100644 index 00000000..e58bc279 --- /dev/null +++ b/src/main/orchestration/scheduler-idle-automation.ts @@ -0,0 +1,50 @@ +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.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 }; +} diff --git a/src/main/orchestration/scheduler-lock-recovery.ts b/src/main/orchestration/scheduler-lock-recovery.ts new file mode 100644 index 00000000..5e977934 --- /dev/null +++ b/src/main/orchestration/scheduler-lock-recovery.ts @@ -0,0 +1,54 @@ +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.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 }; +} diff --git a/src/main/orchestration/scheduler.ts b/src/main/orchestration/scheduler.ts index 76ff044d..c9337a71 100644 --- a/src/main/orchestration/scheduler.ts +++ b/src/main/orchestration/scheduler.ts @@ -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 | null = null; - const scheduleLastFiredMap = new Map(); - - 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 }; } diff --git a/src/main/service-graph.ts b/src/main/service-graph.ts index 6887bb49..9637aaf3 100644 --- a/src/main/service-graph.ts +++ b/src/main/service-graph.ts @@ -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;