feat(FN-1269): complete routine engine integration with RoutineRunner and RoutineScheduler
- Add RoutineRunner class for routine execution via heartbeat system - Add RoutineScheduler class for cron-based routine polling - Add triggerManual and triggerWebhook methods for API and webhook triggers - Wire RoutineScheduler into InProcessRuntime lifecycle - Add routine trigger and webhook API endpoints - Fix type mismatches between PROMPT and actual FN-1519 types
This commit is contained in:
gsxdsm
@@ -385,7 +385,12 @@ describe("RoutineStore", () => {
|
||||
store.on("routine:deleted", listener);
|
||||
|
||||
await store.deleteRoutine(routine.id);
|
||||
expect(listener).toHaveBeenCalledWith(routine);
|
||||
expect(listener).toHaveBeenCalled();
|
||||
const emittedRoutine = listener.mock.calls[0][0];
|
||||
// Verify the emitted routine has the expected fields
|
||||
expect(emittedRoutine.id).toBe(routine.id);
|
||||
expect(emittedRoutine.name).toBe("Delete test");
|
||||
expect(emittedRoutine.agentId).toBe("test-agent");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -1907,7 +1907,7 @@ export interface AgentHeartbeatEvent {
|
||||
}
|
||||
|
||||
/** What triggered a heartbeat run */
|
||||
export type HeartbeatInvocationSource = "on_demand" | "timer" | "assignment" | "automation";
|
||||
export type HeartbeatInvocationSource = "on_demand" | "timer" | "assignment" | "automation" | "routine";
|
||||
|
||||
/** Snapshot of the last blocked state for a task, used for dedup comparison. */
|
||||
export interface BlockedStateSnapshot {
|
||||
|
||||
@@ -8019,6 +8019,7 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
// ── Routine Routes ──────────────────────────────────────────────────
|
||||
|
||||
const routineStore = options?.routineStore;
|
||||
const routineRunner = options?.routineRunner;
|
||||
|
||||
// GET /routines — list all routines
|
||||
router.get("/routines", async (_req: Request, res: Response) => {
|
||||
@@ -8197,22 +8198,21 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
if (!routineStore) {
|
||||
throw new ApiError(503, "Routine store not available");
|
||||
}
|
||||
if (!routineRunner) {
|
||||
throw new ApiError(503, "Routine execution not available");
|
||||
}
|
||||
try {
|
||||
const id = Array.isArray(req.params.id) ? req.params.id[0] : req.params.id;
|
||||
const routine = await routineStore.getRoutine(id);
|
||||
|
||||
// Create a placeholder result for now (actual execution in future task)
|
||||
const startedAt = new Date().toISOString();
|
||||
const result: RoutineExecutionResult = {
|
||||
routineId: id,
|
||||
startedAt,
|
||||
triggerType: routine.trigger.type,
|
||||
success: true,
|
||||
output: "Manual run triggered",
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
// Validate routine is enabled
|
||||
if (!routine.enabled) {
|
||||
throw badRequest("Routine is disabled");
|
||||
}
|
||||
|
||||
const updated = await routineStore.recordRun(id, result);
|
||||
// Execute via RoutineRunner
|
||||
const result = await routineRunner.triggerManual(id);
|
||||
const updated = await routineStore.getRoutine(id);
|
||||
res.json({ routine: updated, result });
|
||||
} catch (err: any) {
|
||||
if (err instanceof ApiError) {
|
||||
@@ -8250,6 +8250,9 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
if (!routineStore) {
|
||||
throw new ApiError(503, "Routine store not available");
|
||||
}
|
||||
if (!routineRunner) {
|
||||
throw new ApiError(503, "Routine execution not available");
|
||||
}
|
||||
try {
|
||||
const id = Array.isArray(req.params.id) ? req.params.id[0] : req.params.id;
|
||||
const routine = await routineStore.getRoutine(id);
|
||||
@@ -8282,18 +8285,10 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
}
|
||||
}
|
||||
|
||||
// Create a placeholder result for now (actual execution in future task)
|
||||
const startedAt = new Date().toISOString();
|
||||
const result: RoutineExecutionResult = {
|
||||
routineId: id,
|
||||
startedAt,
|
||||
triggerType: "webhook",
|
||||
success: true,
|
||||
output: "Webhook trigger received",
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
const updated = await routineStore.recordRun(id, result);
|
||||
// Execute via RoutineRunner
|
||||
const payload = req.body;
|
||||
const result = await routineRunner.triggerWebhook(id, payload, signatureHeader);
|
||||
const updated = await routineStore.getRoutine(id);
|
||||
res.json({ routine: updated, result });
|
||||
} catch (err: any) {
|
||||
if (err instanceof ApiError) {
|
||||
|
||||
@@ -84,6 +84,11 @@ export interface ServerOptions {
|
||||
automationStore?: AutomationStore;
|
||||
/** Optional RoutineStore for recurring task automation */
|
||||
routineStore?: RoutineStore;
|
||||
/** Optional RoutineRunner for triggering routine execution via heartbeat */
|
||||
routineRunner?: {
|
||||
triggerManual(routineId: string): Promise<import("@fusion/core").RoutineExecutionResult>;
|
||||
triggerWebhook(routineId: string, payload: Record<string, unknown>, signature?: string): Promise<import("@fusion/core").RoutineExecutionResult>;
|
||||
};
|
||||
/** Optional AiSessionStore — if not provided, one is created from the default store's database */
|
||||
aiSessionStore?: AiSessionStore;
|
||||
/** Optional MissionAutopilot for autonomous mission progression */
|
||||
|
||||
@@ -26,7 +26,8 @@
|
||||
"@fusion/core": "workspace:*",
|
||||
"@mariozechner/pi-ai": "^0.62.0",
|
||||
"@mariozechner/pi-coding-agent": "^0.62.0",
|
||||
"@sinclair/typebox": "^0.34.48"
|
||||
"@sinclair/typebox": "^0.34.48",
|
||||
"cron-parser": "^5.5.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@modelcontextprotocol/sdk": "^1.0.0",
|
||||
|
||||
@@ -26,6 +26,8 @@ export { PrMonitor, type PrComment, type TrackedPr, type OnNewCommentsCallback }
|
||||
export { PrCommentHandler } from "./pr-comment-handler.js";
|
||||
export { NtfyNotifier, type NtfyNotifierOptions } from "./notifier.js";
|
||||
export { CronRunner, type CronRunnerOptions, type AiPromptExecutor, createAiPromptExecutor } from "./cron-runner.js";
|
||||
export { RoutineRunner, type RoutineRunnerOptions } from "./routine-runner.js";
|
||||
export { RoutineScheduler, type RoutineSchedulerOptions } from "./routine-scheduler.js";
|
||||
export { StuckTaskDetector, type StuckTaskDetectorOptions, type DisposableSession } from "./stuck-task-detector.js";
|
||||
export { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "./agent-heartbeat.js";
|
||||
export { TokenCapDetector, type TokenCapCheckResult } from "./token-cap-detector.js";
|
||||
|
||||
@@ -1,260 +1,512 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { Routine, RoutineStore } from "@fusion/core";
|
||||
import { RoutineRunner } from "./routine-runner.js";
|
||||
import { RoutineRunner, type RoutineRunnerOptions } from "./routine-runner.js";
|
||||
import type {
|
||||
RoutineStore,
|
||||
Routine,
|
||||
RoutineExecutionResult,
|
||||
AgentStore,
|
||||
TaskStore,
|
||||
Settings,
|
||||
} from "@fusion/core";
|
||||
import type { HeartbeatMonitor } from "./agent-heartbeat.js";
|
||||
|
||||
// Mock the logger
|
||||
vi.mock("./logger.js", () => ({
|
||||
createLogger: () => ({
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
// Default settings inline to avoid @fusion/core build dependency during tests
|
||||
const DEFAULT_SETTINGS: Settings = {
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
pollIntervalMs: 30000,
|
||||
autoResolveConflicts: true,
|
||||
requirePlanApproval: false,
|
||||
recycleWorktrees: false,
|
||||
worktreeNaming: "random",
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
ntfyEnabled: false,
|
||||
defaultProvider: "anthropic",
|
||||
defaultModelId: "claude-sonnet-4-5",
|
||||
planningProvider: "anthropic",
|
||||
planningModelId: "claude-sonnet-4-5",
|
||||
validatorProvider: "openai",
|
||||
validatorModelId: "gpt-4o",
|
||||
taskStuckTimeoutMs: undefined,
|
||||
groupOverlappingFiles: false,
|
||||
autoMerge: true,
|
||||
};
|
||||
|
||||
describe("RoutineRunner", () => {
|
||||
let mockHeartbeatMonitor: HeartbeatMonitor;
|
||||
let mockRoutineStore: RoutineStore;
|
||||
let runner: RoutineRunner;
|
||||
function createMockRoutine(overrides: Partial<Routine> = {}): Routine {
|
||||
return {
|
||||
id: "test-routine-id",
|
||||
agentId: "test-agent",
|
||||
name: "Test Routine",
|
||||
description: "A test routine",
|
||||
trigger: { type: "cron", cronExpression: "0 * * * *" },
|
||||
catchUpPolicy: "run_one",
|
||||
executionPolicy: "parallel",
|
||||
enabled: true,
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
cronExpression: "0 * * * *",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
const createMockRoutine = (overrides: Partial<Routine> = {}): Routine =>
|
||||
({
|
||||
id: "routine-1",
|
||||
agentId: "agent-1",
|
||||
name: "Test Routine",
|
||||
description: "A test routine",
|
||||
enabled: true,
|
||||
trigger: {
|
||||
type: "cron",
|
||||
cronExpression: "*/5 * * * *",
|
||||
},
|
||||
executionPolicy: "reject",
|
||||
catchUpPolicy: "skip",
|
||||
catchUpLimit: 5,
|
||||
lastRunAt: null,
|
||||
nextRunAt: new Date().toISOString(),
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
function createMockRoutineStore(routines: Routine[] = []): RoutineStore {
|
||||
const routineMap = new Map(routines.map((r) => [r.id, r]));
|
||||
|
||||
return {
|
||||
getRoutine: vi.fn().mockImplementation((id: string) => {
|
||||
const routine = routineMap.get(id);
|
||||
if (!routine) {
|
||||
throw Object.assign(new Error(`Routine '${id}' not found`), { code: "ENOENT" });
|
||||
}
|
||||
return routine;
|
||||
}),
|
||||
listRoutines: vi.fn().mockResolvedValue(routines),
|
||||
updateRoutine: vi.fn().mockImplementation((id: string, _updates: any) => {
|
||||
const routine = routineMap.get(id);
|
||||
if (!routine) {
|
||||
throw Object.assign(new Error(`Routine '${id}' not found`), { code: "ENOENT" });
|
||||
}
|
||||
return routine;
|
||||
}),
|
||||
getDueRoutines: vi.fn().mockResolvedValue([]),
|
||||
recordRun: vi.fn().mockImplementation((id: string, result: RoutineExecutionResult) => {
|
||||
return createMockRoutine({ id, lastRunResult: result });
|
||||
}),
|
||||
startRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
completeRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
cancelRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
init: vi.fn().mockResolvedValue(undefined),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as RoutineStore;
|
||||
}
|
||||
|
||||
function createMockAgentStore(): AgentStore {
|
||||
return {
|
||||
getAgent: vi.fn().mockImplementation(async (id: string) => ({
|
||||
id,
|
||||
name: "Test Agent",
|
||||
role: "executor" as const,
|
||||
state: "idle" as const,
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
...overrides,
|
||||
} as Routine);
|
||||
})),
|
||||
updateAgentState: vi.fn().mockResolvedValue(undefined),
|
||||
getBudgetStatus: vi.fn().mockResolvedValue({
|
||||
agentId: "",
|
||||
currentUsage: 0,
|
||||
budgetLimit: null,
|
||||
usagePercent: null,
|
||||
thresholdPercent: null,
|
||||
isOverBudget: false,
|
||||
isOverThreshold: false,
|
||||
lastResetAt: null,
|
||||
nextResetAt: null,
|
||||
}),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as AgentStore;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
mockHeartbeatMonitor = {
|
||||
executeHeartbeat: vi.fn().mockResolvedValue({
|
||||
id: "run-1",
|
||||
agentId: "agent-1",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "completed",
|
||||
}),
|
||||
start: vi.fn(),
|
||||
stop: vi.fn(),
|
||||
isAgentHealthy: vi.fn(),
|
||||
checkMissedHeartbeats: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
getAgentHeartbeatConfig: vi.fn(),
|
||||
} as unknown as HeartbeatMonitor;
|
||||
function createMockTaskStore(): TaskStore {
|
||||
return {
|
||||
getSettings: vi.fn().mockResolvedValue(DEFAULT_SETTINGS),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as TaskStore;
|
||||
}
|
||||
|
||||
mockRoutineStore = {
|
||||
getRoutine: vi.fn(),
|
||||
getRoutines: vi.fn(),
|
||||
createRoutine: vi.fn(),
|
||||
updateRoutine: vi.fn(),
|
||||
deleteRoutine: vi.fn(),
|
||||
getDueRoutines: vi.fn(),
|
||||
startRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
completeRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
cancelRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
recordRun: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as RoutineStore;
|
||||
function createMockHeartbeatMonitor(): HeartbeatMonitor {
|
||||
return {
|
||||
executeHeartbeat: vi.fn().mockResolvedValue({
|
||||
id: "run-123",
|
||||
agentId: "test-agent",
|
||||
status: "completed" as const,
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
}),
|
||||
start: vi.fn(),
|
||||
stop: vi.fn(),
|
||||
trackAgent: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as HeartbeatMonitor;
|
||||
}
|
||||
|
||||
runner = new RoutineRunner({
|
||||
heartbeatMonitor: mockHeartbeatMonitor,
|
||||
routineStore: mockRoutineStore,
|
||||
});
|
||||
function createRoutineRunner(options?: Partial<RoutineRunnerOptions>): RoutineRunner {
|
||||
return new RoutineRunner({
|
||||
routineStore: options?.routineStore ?? createMockRoutineStore(),
|
||||
heartbeatMonitor: options?.heartbeatMonitor ?? createMockHeartbeatMonitor(),
|
||||
rootDir: options?.rootDir ?? "/test/root",
|
||||
});
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.clearAllMocks();
|
||||
runner.clearInFlight("routine-1");
|
||||
runner.clearInFlight("routine-2");
|
||||
});
|
||||
describe("RoutineRunner", () => {
|
||||
describe("executeRoutine", () => {
|
||||
it("successfully executes a routine with trigger type 'cron'", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-1", name: "Test Routine" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
describe("execute", () => {
|
||||
it("should execute a routine successfully", async () => {
|
||||
const routine = createMockRoutine();
|
||||
const result = await runner.executeRoutine("routine-1", "cron");
|
||||
|
||||
const result = await runner.execute(routine);
|
||||
|
||||
expect(result.success).toBe(true);
|
||||
expect(result.routineId).toBe("routine-1");
|
||||
expect(mockRoutineStore.startRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-1",
|
||||
expect(result.success).toBe(true);
|
||||
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
triggeredAt: expect.any(String),
|
||||
invocationSource: "routine",
|
||||
})
|
||||
);
|
||||
expect(mockRoutineStore.completeRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-1",
|
||||
expect.objectContaining({
|
||||
success: true,
|
||||
})
|
||||
source: "routine",
|
||||
triggerDetail: "routine:routine-1:cron",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("should skip execution when already in-flight with reject policy", async () => {
|
||||
const routine = createMockRoutine({ executionPolicy: "reject" });
|
||||
it("throws descriptive error when routine not found", async () => {
|
||||
const routineStore = createMockRoutineStore([]);
|
||||
const runner = createRoutineRunner({ routineStore });
|
||||
|
||||
// First execution - starts but doesn't complete yet
|
||||
vi.mocked(mockHeartbeatMonitor.executeHeartbeat).mockImplementation(
|
||||
() => new Promise(() => {}) // Never resolves
|
||||
await expect(runner.executeRoutine("nonexistent", "cron")).rejects.toThrow(
|
||||
"Routine 'nonexistent' not found",
|
||||
);
|
||||
|
||||
// Start first execution (won't complete due to mock)
|
||||
const firstExecution = runner.execute(routine);
|
||||
// Give it a tick to start
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
|
||||
// Second execution should be skipped
|
||||
const secondResult = await runner.execute(routine);
|
||||
|
||||
expect(secondResult.success).toBe(true);
|
||||
expect(secondResult.error).toContain("reject");
|
||||
});
|
||||
|
||||
it("should handle execution failure", async () => {
|
||||
const routine = createMockRoutine();
|
||||
vi.mocked(mockHeartbeatMonitor.executeHeartbeat).mockResolvedValue({
|
||||
id: "run-1",
|
||||
agentId: "agent-1",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "failed",
|
||||
stderrExcerpt: "Agent session failed",
|
||||
} as any);
|
||||
it("throws descriptive error when routine is disabled", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-disabled", enabled: false });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const runner = createRoutineRunner({ routineStore });
|
||||
|
||||
const result = await runner.execute(routine);
|
||||
await expect(runner.executeRoutine("routine-disabled", "cron")).rejects.toThrow(
|
||||
"Routine 'routine-disabled' is disabled",
|
||||
);
|
||||
});
|
||||
|
||||
expect(result.success).toBe(false);
|
||||
expect(result.error).toBe("Agent session failed");
|
||||
expect(mockRoutineStore.completeRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-1",
|
||||
it("calls executeHeartbeat with source 'routine' and correct triggerDetail format", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-trigger", name: "Trigger Test" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
await runner.executeRoutine("routine-trigger", "webhook");
|
||||
|
||||
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
success: false,
|
||||
error: "Agent session failed",
|
||||
})
|
||||
source: "routine",
|
||||
triggerDetail: "routine:routine-trigger:webhook",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("should propagate catch-up context to heartbeat", async () => {
|
||||
const routine = createMockRoutine();
|
||||
it("includes routineId, routineName, triggerType in contextSnapshot", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-context", name: "Context Test" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
const catchUpTime = "2024-01-01T00:00:00.000Z";
|
||||
await runner.execute(routine, { catchUpFrom: catchUpTime });
|
||||
await runner.executeRoutine("routine-context", "api");
|
||||
|
||||
expect(mockHeartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(
|
||||
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
contextSnapshot: expect.objectContaining({
|
||||
routineId: "routine-1",
|
||||
catchUpFrom: catchUpTime,
|
||||
routineId: "routine-context",
|
||||
routineName: "Context Test",
|
||||
triggerType: "api",
|
||||
}),
|
||||
})
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("should allow concurrent execution with parallel policy", async () => {
|
||||
const routine = createMockRoutine({ executionPolicy: "parallel" });
|
||||
it("calls completeRoutineExecution after execution completes", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-record" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
// Both executions should succeed
|
||||
await runner.executeRoutine("routine-record", "cron");
|
||||
|
||||
// completeRoutineExecution is called once with the result
|
||||
expect(routineStore.completeRoutineExecution).toHaveBeenCalledTimes(1);
|
||||
expect(routineStore.completeRoutineExecution).toHaveBeenCalledWith(
|
||||
"routine-record",
|
||||
expect.objectContaining({
|
||||
success: true,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("marks execution as failed when executeHeartbeat rejects", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-fail" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
(heartbeatMonitor.executeHeartbeat as ReturnType<typeof vi.fn>).mockRejectedValue(
|
||||
new Error("Heartbeat failed"),
|
||||
);
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
const result = await runner.executeRoutine("routine-fail", "cron");
|
||||
|
||||
expect(result.success).toBe(false);
|
||||
expect(result.error).toBe("Heartbeat failed");
|
||||
});
|
||||
|
||||
it("cleans up inFlightExecutions map after successful completion", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-cleanup" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
expect(runner.isRoutineRunning("routine-cleanup")).toBe(false);
|
||||
|
||||
await runner.executeRoutine("routine-cleanup", "cron");
|
||||
|
||||
// After completion, should not be in-flight
|
||||
expect(runner.isRoutineRunning("routine-cleanup")).toBe(false);
|
||||
});
|
||||
|
||||
it("cleans up inFlightExecutions map even on error", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-error-cleanup", enabled: false });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const runner = createRoutineRunner({ routineStore });
|
||||
|
||||
try {
|
||||
await runner.executeRoutine("routine-error-cleanup", "cron");
|
||||
} catch {
|
||||
// Expected to throw
|
||||
}
|
||||
|
||||
// After an error, the routine should not be in the in-flight map
|
||||
expect(runner.isRoutineRunning("routine-error-cleanup")).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("concurrency policies", () => {
|
||||
it("parallel policy: runs even when another execution is in-flight", async () => {
|
||||
const routine = createMockRoutine({
|
||||
id: "routine-parallel",
|
||||
executionPolicy: "parallel",
|
||||
});
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
// Make heartbeat slow
|
||||
(heartbeatMonitor.executeHeartbeat as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
async () => {
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
return {
|
||||
id: "run-123",
|
||||
agentId: "test-agent",
|
||||
status: "completed" as const,
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
};
|
||||
},
|
||||
);
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
// Start two executions
|
||||
const [result1, result2] = await Promise.all([
|
||||
runner.execute(routine),
|
||||
runner.execute(routine),
|
||||
runner.executeRoutine("routine-parallel", "cron"),
|
||||
runner.executeRoutine("routine-parallel", "cron"),
|
||||
]);
|
||||
|
||||
// Both should succeed (parallel)
|
||||
expect(result1.success).toBe(true);
|
||||
expect(result2.success).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("determineCatchUp", () => {
|
||||
it("should not catch up when policy is skip", () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "skip" });
|
||||
const lastRunAt = new Date(Date.now() - 10 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(false);
|
||||
});
|
||||
|
||||
it("should catch up once when policy is run_one", () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "run_one" });
|
||||
const lastRunAt = new Date(Date.now() - 10 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(true);
|
||||
expect(result.catchUpFrom).toBe(lastRunAt);
|
||||
});
|
||||
|
||||
it("should handle bounded catch-up with run policy", () => {
|
||||
it("reject policy: returns failed result when another execution is in-flight", async () => {
|
||||
const routine = createMockRoutine({
|
||||
catchUpPolicy: "run",
|
||||
catchUpLimit: 3,
|
||||
id: "routine-reject",
|
||||
executionPolicy: "reject",
|
||||
});
|
||||
// Last executed 30 minutes ago
|
||||
const lastRunAt = new Date(Date.now() - 30 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(true);
|
||||
expect(result.catchUpFrom).toBeDefined();
|
||||
});
|
||||
|
||||
it("should not catch up when never executed", () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "run" });
|
||||
|
||||
const result = runner.determineCatchUp(routine, null, new Date());
|
||||
|
||||
expect(result.shouldCatchUp).toBe(false);
|
||||
});
|
||||
|
||||
it("should not catch up for non-cron triggers", () => {
|
||||
const routine = createMockRoutine({
|
||||
trigger: { type: "manual" },
|
||||
catchUpPolicy: "run",
|
||||
});
|
||||
const lastRunAt = new Date(Date.now() - 10 * 60 * 1000).toISOString();
|
||||
|
||||
const result = runner.determineCatchUp(routine, lastRunAt, new Date());
|
||||
|
||||
// Manual triggers return 5 * 60_000 default interval, 10min / 5min = 2 missed
|
||||
// With catchUpLimit 5, boundedCount = 2, which is > 1 so shouldCatchUp = true
|
||||
expect(result.shouldCatchUp).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("isExecuting", () => {
|
||||
it("should return false when routine is not executing", () => {
|
||||
expect(runner.isExecuting("routine-1")).toBe(false);
|
||||
});
|
||||
|
||||
it("should return true when routine is executing", async () => {
|
||||
const routine = createMockRoutine();
|
||||
|
||||
vi.mocked(mockHeartbeatMonitor.executeHeartbeat).mockImplementation(
|
||||
() => new Promise(() => {}) // Never resolves
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
// Make heartbeat slow
|
||||
(heartbeatMonitor.executeHeartbeat as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
async () => {
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
return {
|
||||
id: "run-123",
|
||||
agentId: "test-agent",
|
||||
status: "completed" as const,
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
};
|
||||
},
|
||||
);
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
const execution = runner.execute(routine);
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
// Start first execution
|
||||
const promise1 = runner.executeRoutine("routine-reject", "cron");
|
||||
|
||||
expect(runner.isExecuting("routine-1")).toBe(true);
|
||||
// Immediately try second execution - should be rejected
|
||||
const result2 = await runner.executeRoutine("routine-reject", "cron");
|
||||
|
||||
expect(result2.success).toBe(false);
|
||||
expect(result2.error).toBe("Routine rejected — already running");
|
||||
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledTimes(1); // Only first call
|
||||
|
||||
await promise1; // Clean up
|
||||
});
|
||||
|
||||
it("queue policy: waits for existing execution to complete", async () => {
|
||||
const routine = createMockRoutine({
|
||||
id: "routine-queue",
|
||||
executionPolicy: "queue",
|
||||
});
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
let callCount = 0;
|
||||
(heartbeatMonitor.executeHeartbeat as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
async () => {
|
||||
callCount++;
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
return {
|
||||
id: "run-123",
|
||||
agentId: "test-agent",
|
||||
status: "completed" as const,
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
};
|
||||
},
|
||||
);
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
// Start first execution
|
||||
const [result1, result2] = await Promise.all([
|
||||
runner.executeRoutine("routine-queue", "cron"),
|
||||
runner.executeRoutine("routine-queue", "cron"),
|
||||
]);
|
||||
|
||||
// Both should succeed (second waited for first)
|
||||
expect(result1.success).toBe(true);
|
||||
expect(result2.success).toBe(true);
|
||||
// Both heartbeats should have been called (sequential due to queue)
|
||||
expect(callCount).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("handleCatchUp", () => {
|
||||
it("skip policy: does NOT call executeRoutine, only logs", async () => {
|
||||
const routine = createMockRoutine({
|
||||
id: "routine-catchup-skip",
|
||||
catchUpPolicy: "skip",
|
||||
lastRunAt: new Date(Date.now() - 7200000).toISOString(), // 2 hours ago
|
||||
cronExpression: "0 * * * *",
|
||||
});
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
await runner.handleCatchUp(routine);
|
||||
|
||||
// No executions should have happened
|
||||
expect(heartbeatMonitor.executeHeartbeat).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("never-run routine (lastRunAt undefined): skips catch-up", async () => {
|
||||
const routine = createMockRoutine({
|
||||
id: "routine-never-run",
|
||||
lastRunAt: undefined,
|
||||
cronExpression: "0 * * * *",
|
||||
});
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
await runner.handleCatchUp(routine);
|
||||
|
||||
// No executions should have happened
|
||||
expect(heartbeatMonitor.executeHeartbeat).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("caps at MAX_CATCH_UP_INTERVALS (10) even when more intervals exist", async () => {
|
||||
const twoHoursAgo = new Date(Date.now() - 7200000);
|
||||
const routine = createMockRoutine({
|
||||
id: "routine-many-missed",
|
||||
catchUpPolicy: "run",
|
||||
lastRunAt: twoHoursAgo.toISOString(),
|
||||
cronExpression: "*/5 * * * *", // Every 5 minutes = 24 missed in 2 hours
|
||||
});
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
await runner.handleCatchUp(routine);
|
||||
|
||||
// Should be capped at 10
|
||||
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledTimes(10);
|
||||
});
|
||||
});
|
||||
|
||||
describe("helper methods", () => {
|
||||
it("getInFlightCount returns correct count", async () => {
|
||||
const routine1 = createMockRoutine({ id: "routine-count-1" });
|
||||
const routine2 = createMockRoutine({ id: "routine-count-2" });
|
||||
const routineStore = createMockRoutineStore([routine1, routine2]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
// Make heartbeat slow to allow checking in-flight count
|
||||
(heartbeatMonitor.executeHeartbeat as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
async () => {
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
return {
|
||||
id: "run-123",
|
||||
agentId: "test-agent",
|
||||
status: "completed" as const,
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
};
|
||||
},
|
||||
);
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
expect(runner.getInFlightCount()).toBe(0);
|
||||
|
||||
// Start first execution
|
||||
const promise1 = runner.executeRoutine("routine-count-1", "cron");
|
||||
// Allow microtask to complete to see the in-flight state
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
expect(runner.getInFlightCount()).toBe(1);
|
||||
|
||||
// Start second execution (will run in parallel since policy is "parallel")
|
||||
const promise2 = runner.executeRoutine("routine-count-2", "cron");
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
expect(runner.getInFlightCount()).toBe(2);
|
||||
|
||||
await Promise.all([promise1, promise2]);
|
||||
expect(runner.getInFlightCount()).toBe(0);
|
||||
});
|
||||
|
||||
it("isRoutineRunning returns true during execution, false after", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-running" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const heartbeatMonitor = createMockHeartbeatMonitor();
|
||||
// Make heartbeat slow to allow checking in-flight state
|
||||
(heartbeatMonitor.executeHeartbeat as ReturnType<typeof vi.fn>).mockImplementation(
|
||||
async () => {
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
return {
|
||||
id: "run-123",
|
||||
agentId: "test-agent",
|
||||
status: "completed" as const,
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
};
|
||||
},
|
||||
);
|
||||
const runner = createRoutineRunner({ routineStore, heartbeatMonitor });
|
||||
|
||||
expect(runner.isRoutineRunning("routine-running")).toBe(false);
|
||||
|
||||
const promise = runner.executeRoutine("routine-running", "cron");
|
||||
// Allow microtask to complete to see the in-flight state
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
expect(runner.isRoutineRunning("routine-running")).toBe(true);
|
||||
|
||||
await promise;
|
||||
expect(runner.isRoutineRunning("routine-running")).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,293 +1,334 @@
|
||||
/**
|
||||
* RoutineRunner — orchestrates routine execution via the heartbeat system.
|
||||
*
|
||||
* - Validates routine state before execution (enabled, has assigned agent)
|
||||
* - Enforces concurrency policies (parallel/skip/queue/replace)
|
||||
* - Handles catch-up for missed runs
|
||||
* - Triggers heartbeat execution for routines
|
||||
*/
|
||||
|
||||
import { CronExpressionParser } from "cron-parser";
|
||||
import type {
|
||||
Routine,
|
||||
RoutineStore,
|
||||
HeartbeatInvocationSource,
|
||||
Routine,
|
||||
RoutineExecutionResult,
|
||||
} from "@fusion/core";
|
||||
import type { HeartbeatMonitor } from "./agent-heartbeat.js";
|
||||
import { createLogger } from "./logger.js";
|
||||
|
||||
const logger = createLogger("routine-runner");
|
||||
const log = createLogger("routine-runner");
|
||||
|
||||
/**
|
||||
* Options for RoutineRunner.
|
||||
*/
|
||||
/** Options for RoutineRunner constructor */
|
||||
export interface RoutineRunnerOptions {
|
||||
/** The heartbeat monitor for executing routines */
|
||||
heartbeatMonitor: HeartbeatMonitor;
|
||||
/** The routine store for persisting execution state */
|
||||
/** RoutineStore for querying and updating routines */
|
||||
routineStore: RoutineStore;
|
||||
/** HeartbeatMonitor for triggering agent execution */
|
||||
heartbeatMonitor: HeartbeatMonitor;
|
||||
/** Project root directory */
|
||||
rootDir: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Tracks in-flight executions per routine ID.
|
||||
* Maximum number of catch-up executions to prevent runaway loops.
|
||||
*/
|
||||
const inFlightExecutions = new Map<string, boolean>();
|
||||
const MAX_CATCH_UP_INTERVALS = 10;
|
||||
|
||||
/**
|
||||
* Result of a routine execution.
|
||||
*/
|
||||
export interface RoutineExecutionResult {
|
||||
routineId: string;
|
||||
success: boolean;
|
||||
error?: string;
|
||||
executedAt: string;
|
||||
catchUpExecution?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* RoutineRunner orchestrates execution of a single routine through the heartbeat system.
|
||||
* RoutineRunner orchestrates routine execution via the heartbeat system.
|
||||
*
|
||||
* It handles:
|
||||
* - Concurrency policy enforcement (parallel/queue/reject)
|
||||
* - Catch-up policy handling for missed schedule windows
|
||||
* - Execution state persistence via RoutineStore
|
||||
* Key behaviors:
|
||||
* - Enforces concurrency policies before starting executions
|
||||
* - Handles catch-up for missed runs based on catch-up policy
|
||||
* - Triggers heartbeats with routine context in the trigger detail
|
||||
*/
|
||||
export class RoutineRunner {
|
||||
private heartbeatMonitor: HeartbeatMonitor;
|
||||
private routineStore: RoutineStore;
|
||||
private options: RoutineRunnerOptions;
|
||||
/** Tracks currently-running executions by routine ID */
|
||||
private inFlightExecutions: Map<string, Promise<RoutineExecutionResult>> = new Map();
|
||||
|
||||
constructor(options: RoutineRunnerOptions) {
|
||||
this.heartbeatMonitor = options.heartbeatMonitor;
|
||||
this.routineStore = options.routineStore;
|
||||
this.options = options;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a routine is currently being executed.
|
||||
*/
|
||||
isExecuting(routineId: string): boolean {
|
||||
return inFlightExecutions.get(routineId) === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a routine based on its configuration and policies.
|
||||
* Execute a routine by ID with a given trigger type.
|
||||
*
|
||||
* @param routine - The routine to execute
|
||||
* @param options.catchUpFrom - Optional timestamp to use for catch-up execution
|
||||
* @returns The result of the execution attempt
|
||||
* @param routineId - ID of the routine to execute
|
||||
* @param triggerType - What triggered this execution: "cron", "webhook", or "api"
|
||||
* @param context - Additional context passed to the heartbeat execution
|
||||
* @returns The execution result
|
||||
* @throws Error if routine not found or disabled
|
||||
*/
|
||||
async execute(
|
||||
routine: Routine,
|
||||
options: { catchUpFrom?: string } = {}
|
||||
async executeRoutine(
|
||||
routineId: string,
|
||||
triggerType: "cron" | "webhook" | "api",
|
||||
context?: Record<string, unknown>,
|
||||
): Promise<RoutineExecutionResult> {
|
||||
const { catchUpFrom } = options;
|
||||
const routineId = routine.id;
|
||||
// 1. Load routine
|
||||
let routine: Routine;
|
||||
try {
|
||||
routine = await this.options.routineStore.getRoutine(routineId);
|
||||
} catch {
|
||||
throw new Error(`Routine '${routineId}' not found`);
|
||||
}
|
||||
|
||||
// Check concurrency policy
|
||||
const policyResult = this.checkConcurrencyPolicy(routine);
|
||||
if (!policyResult.shouldExecute) {
|
||||
logger.log(
|
||||
`[${routineId}] Skipped by concurrency policy: ${policyResult.reason}`
|
||||
);
|
||||
// 2. Validate routine state
|
||||
if (!routine.enabled) {
|
||||
throw new Error(`Routine '${routineId}' is disabled`);
|
||||
}
|
||||
|
||||
if (!routine.agentId) {
|
||||
throw new Error(`Routine '${routineId}' has no assigned agent`);
|
||||
}
|
||||
|
||||
// 3. Enforce concurrency policy
|
||||
const concurrency = routine.executionPolicy ?? "queue";
|
||||
|
||||
if (concurrency === "reject" && this.inFlightExecutions.has(routineId)) {
|
||||
log.log(`Routine ${routineId} rejected — already running`);
|
||||
// Return a failed result without creating an execution record
|
||||
return {
|
||||
routineId,
|
||||
success: true,
|
||||
executedAt: new Date().toISOString(),
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
error: policyResult.reason,
|
||||
success: false,
|
||||
output: "Routine rejected — already running",
|
||||
error: "Routine rejected — already running",
|
||||
startedAt: new Date().toISOString(),
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
// Mark as in-flight
|
||||
inFlightExecutions.set(routineId, true);
|
||||
// If queue, wait for existing execution
|
||||
if (concurrency === "queue" && this.inFlightExecutions.has(routineId)) {
|
||||
log.log(`Routine ${routineId} queued — waiting for existing execution`);
|
||||
const existingResult = await this.inFlightExecutions.get(routineId);
|
||||
if (existingResult) {
|
||||
await existingResult;
|
||||
}
|
||||
}
|
||||
|
||||
// 4. Record execution start
|
||||
const startedAt = new Date().toISOString();
|
||||
|
||||
// Set in-flight BEFORE starting execution to prevent race conditions
|
||||
const executionPromise = this.runExecution(routine, triggerType, context, startedAt);
|
||||
this.inFlightExecutions.set(routineId, executionPromise);
|
||||
|
||||
try {
|
||||
// Persist execution start
|
||||
const startedAt = new Date().toISOString();
|
||||
await this.routineStore.startRoutineExecution(routineId, {
|
||||
await this.options.routineStore.startRoutineExecution(routineId, {
|
||||
triggeredAt: startedAt,
|
||||
catchUpFrom,
|
||||
invocationSource: "routine",
|
||||
});
|
||||
|
||||
logger.log(`[${routineId}] Starting routine execution`);
|
||||
const result = await executionPromise;
|
||||
return result;
|
||||
} finally {
|
||||
this.inFlightExecutions.delete(routineId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal execution logic for a routine.
|
||||
*/
|
||||
private async runExecution(
|
||||
routine: Routine,
|
||||
triggerType: string,
|
||||
context: Record<string, unknown> | undefined,
|
||||
startedAt: string,
|
||||
): Promise<RoutineExecutionResult> {
|
||||
const routineId = routine.id;
|
||||
|
||||
try {
|
||||
// Execute via heartbeat monitor
|
||||
const run = await this.heartbeatMonitor.executeHeartbeat({
|
||||
const run = await this.options.heartbeatMonitor.executeHeartbeat({
|
||||
agentId: routine.agentId,
|
||||
source: "routine" as HeartbeatInvocationSource,
|
||||
triggerDetail: `routine:${routineId}`,
|
||||
source: "routine",
|
||||
triggerDetail: `routine:${routine.id}:${triggerType}`,
|
||||
contextSnapshot: {
|
||||
routineId,
|
||||
catchUpFrom,
|
||||
executionPolicy: routine.executionPolicy,
|
||||
catchUpPolicy: routine.catchUpPolicy,
|
||||
routineId: routine.id,
|
||||
routineName: routine.name,
|
||||
triggerType,
|
||||
...context,
|
||||
},
|
||||
});
|
||||
|
||||
// Handle failed/terminated runs
|
||||
// Determine status from run
|
||||
let success = true;
|
||||
let output = "";
|
||||
let error: string | undefined;
|
||||
|
||||
if (run.status === "failed" || run.status === "terminated") {
|
||||
const error = run.stderrExcerpt || `Run ${run.status}`;
|
||||
logger.log(`[${routineId}] Execution ${run.status}: ${error}`);
|
||||
await this.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt: run.endedAt ?? new Date().toISOString(),
|
||||
success: false,
|
||||
error,
|
||||
});
|
||||
return {
|
||||
routineId,
|
||||
success: false,
|
||||
error,
|
||||
executedAt: startedAt,
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
};
|
||||
success = false;
|
||||
error = run.stderrExcerpt || `Run ${run.status}`;
|
||||
output = error;
|
||||
} else {
|
||||
output = run.resultJson ? JSON.stringify(run.resultJson) : "Routine completed successfully";
|
||||
}
|
||||
|
||||
// Persist execution completion
|
||||
const completedAt = run.endedAt ?? new Date().toISOString();
|
||||
await this.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt,
|
||||
success: true,
|
||||
// Complete the execution
|
||||
await this.options.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt: new Date().toISOString(),
|
||||
success,
|
||||
resultJson: run.resultJson,
|
||||
error,
|
||||
});
|
||||
|
||||
logger.log(`[${routineId}] Routine execution completed successfully`);
|
||||
|
||||
return {
|
||||
routineId,
|
||||
success: true,
|
||||
executedAt: completedAt,
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
success,
|
||||
output,
|
||||
startedAt,
|
||||
completedAt: new Date().toISOString(),
|
||||
error,
|
||||
};
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log(`[${routineId}] Routine execution failed: ${errorMessage}`);
|
||||
} catch (err) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Routine ${routineId} execution failed: ${errorMessage}`);
|
||||
|
||||
// Persist execution failure
|
||||
// Record failure
|
||||
try {
|
||||
await this.routineStore.completeRoutineExecution(routineId, {
|
||||
await this.options.routineStore.completeRoutineExecution(routineId, {
|
||||
completedAt: new Date().toISOString(),
|
||||
success: false,
|
||||
error: errorMessage,
|
||||
});
|
||||
} catch (persistError) {
|
||||
logger.error(`[${routineId}] Failed to persist error state: ${persistError}`);
|
||||
log.error(`[${routineId}] Failed to persist error state: ${persistError}`);
|
||||
}
|
||||
|
||||
return {
|
||||
routineId,
|
||||
success: false,
|
||||
output: errorMessage,
|
||||
startedAt,
|
||||
completedAt: new Date().toISOString(),
|
||||
error: errorMessage,
|
||||
executedAt: new Date().toISOString(),
|
||||
catchUpExecution: !!catchUpFrom,
|
||||
};
|
||||
} finally {
|
||||
// Clear in-flight flag
|
||||
inFlightExecutions.delete(routineId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a routine should be executed based on its concurrency policy.
|
||||
* Handle catch-up for missed routine executions based on the catch-up policy.
|
||||
*
|
||||
* @param routine - The routine to check for catch-up
|
||||
*/
|
||||
private checkConcurrencyPolicy(
|
||||
routine: Routine
|
||||
): { shouldExecute: boolean; reason?: string } {
|
||||
const routineId = routine.id;
|
||||
const isInFlight = this.isExecuting(routineId);
|
||||
async handleCatchUp(routine: Routine): Promise<void> {
|
||||
const catchUpPolicy = routine.catchUpPolicy ?? "skip";
|
||||
|
||||
if (isInFlight) {
|
||||
switch (routine.executionPolicy) {
|
||||
case "parallel":
|
||||
return { shouldExecute: true };
|
||||
|
||||
case "reject":
|
||||
return {
|
||||
shouldExecute: false,
|
||||
reason: `Routine ${routineId} is already being executed (policy: reject)`,
|
||||
};
|
||||
|
||||
case "queue":
|
||||
// Queue for later execution - return without executing
|
||||
return {
|
||||
shouldExecute: false,
|
||||
reason: `Routine ${routineId} is already being executed (policy: queue)`,
|
||||
};
|
||||
|
||||
default:
|
||||
return {
|
||||
shouldExecute: false,
|
||||
reason: `Unknown execution policy for ${routineId}`,
|
||||
};
|
||||
}
|
||||
if (catchUpPolicy === "skip") {
|
||||
return;
|
||||
}
|
||||
|
||||
return { shouldExecute: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if a catch-up execution should occur based on the routine's catch-up policy.
|
||||
*/
|
||||
determineCatchUp(
|
||||
routine: Routine,
|
||||
lastRunAt: string | null,
|
||||
currentTime: Date
|
||||
): { shouldCatchUp: boolean; catchUpFrom?: string } {
|
||||
if (!lastRunAt) {
|
||||
return { shouldCatchUp: false };
|
||||
// "run_one" or "run" policy - need to catch up
|
||||
if (!routine.lastRunAt) {
|
||||
// Never run before — nothing to catch up
|
||||
return;
|
||||
}
|
||||
|
||||
switch (routine.catchUpPolicy) {
|
||||
case "skip":
|
||||
return { shouldCatchUp: false };
|
||||
// Calculate missed intervals
|
||||
if (!routine.cronExpression) {
|
||||
return;
|
||||
}
|
||||
|
||||
case "run_one":
|
||||
return { shouldCatchUp: true, catchUpFrom: lastRunAt };
|
||||
try {
|
||||
const cronExpr = CronExpressionParser.parse(routine.cronExpression, {
|
||||
currentDate: new Date(routine.lastRunAt ?? Date.now()),
|
||||
});
|
||||
const lastRun = new Date(routine.lastRunAt ?? Date.now());
|
||||
const now = new Date();
|
||||
|
||||
case "run": {
|
||||
const catchUpLimit = routine.catchUpLimit ?? 5;
|
||||
const lastExecuted = new Date(lastRunAt);
|
||||
const diffMs = currentTime.getTime() - lastExecuted.getTime();
|
||||
const missedIntervals: Date[] = [];
|
||||
|
||||
const intervalMs = this.getRoutineIntervalMs(routine);
|
||||
if (intervalMs <= 0) {
|
||||
return { shouldCatchUp: false };
|
||||
// Get next interval after lastRun, then iterate
|
||||
let intervalDate = new Date(cronExpr.next().toISOString() ?? Date.now());
|
||||
|
||||
while (intervalDate.getTime() <= now.getTime() && missedIntervals.length < MAX_CATCH_UP_INTERVALS) {
|
||||
if (intervalDate.getTime() > lastRun.getTime()) {
|
||||
missedIntervals.push(new Date(intervalDate));
|
||||
}
|
||||
|
||||
const missedCount = Math.floor(diffMs / intervalMs);
|
||||
const boundedCount = Math.min(missedCount, catchUpLimit);
|
||||
|
||||
if (boundedCount <= 1) {
|
||||
return { shouldCatchUp: false };
|
||||
}
|
||||
|
||||
const catchUpFrom = new Date(
|
||||
lastExecuted.getTime() + intervalMs
|
||||
).toISOString();
|
||||
|
||||
logger.log(
|
||||
`[${routine.id}] Catch-up: ${boundedCount} missed executions (limit: ${catchUpLimit})`
|
||||
);
|
||||
|
||||
return { shouldCatchUp: true, catchUpFrom };
|
||||
const nextIso = cronExpr.next().toISOString();
|
||||
if (!nextIso) break;
|
||||
intervalDate = new Date(nextIso);
|
||||
}
|
||||
|
||||
default:
|
||||
return { shouldCatchUp: false };
|
||||
if (missedIntervals.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
log.log(`[${routine.id}] Running ${missedIntervals.length} catch-up executions`);
|
||||
|
||||
// Execute each missed interval
|
||||
for (const missedInterval of missedIntervals) {
|
||||
try {
|
||||
await this.executeRoutine(routine.id, "cron", {
|
||||
catchUp: true,
|
||||
missedInterval: missedInterval.toISOString(),
|
||||
});
|
||||
} catch (err) {
|
||||
log.error(`[${routine.id}] Catch-up execution failed: ${err}`);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
log.error(`[${routine.id}] Error calculating catch-up intervals: ${err}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the interval in milliseconds for a routine based on its cron schedule.
|
||||
* Trigger a routine manually (via API).
|
||||
*
|
||||
* @param routineId - The ID of the routine to trigger
|
||||
* @returns The execution result
|
||||
* @throws Error if routine not found or disabled
|
||||
*/
|
||||
private getRoutineIntervalMs(routine: Routine): number {
|
||||
if (routine.trigger.type === "cron") {
|
||||
const cron = routine.trigger.cronExpression;
|
||||
if (cron.includes("* * *")) return 60_000;
|
||||
if (cron.includes("*/5")) return 5 * 60_000;
|
||||
if (cron.includes("*/10")) return 10 * 60_000;
|
||||
if (cron.includes("*/15")) return 15 * 60_000;
|
||||
if (cron.includes("*/30")) return 30 * 60_000;
|
||||
if (cron.includes("0 * *")) return 60 * 60_000;
|
||||
if (cron.includes("0 0 *")) return 24 * 60 * 60_000;
|
||||
return 5 * 60_000;
|
||||
async triggerManual(routineId: string): Promise<RoutineExecutionResult> {
|
||||
const routine = await this.options.routineStore.getRoutine(routineId);
|
||||
|
||||
if (!routine.enabled) {
|
||||
throw new Error(`Routine '${routineId}' is disabled`);
|
||||
}
|
||||
return 5 * 60_000;
|
||||
|
||||
return this.executeRoutine(routineId, "api");
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the in-flight flag for a routine (for testing).
|
||||
* Trigger a routine via webhook.
|
||||
*
|
||||
* @param routineId - The ID of the routine to trigger
|
||||
* @param payload - The webhook payload
|
||||
* @param _signature - The webhook signature (verified by RoutineScheduler)
|
||||
* @returns The execution result
|
||||
* @throws Error if routine not found, not a webhook trigger, or disabled
|
||||
*/
|
||||
clearInFlight(routineId: string): void {
|
||||
inFlightExecutions.delete(routineId);
|
||||
async triggerWebhook(
|
||||
routineId: string,
|
||||
payload: Record<string, unknown>,
|
||||
_signature?: string
|
||||
): Promise<RoutineExecutionResult> {
|
||||
const routine = await this.options.routineStore.getRoutine(routineId);
|
||||
|
||||
if (routine.trigger.type !== "webhook") {
|
||||
throw new Error(
|
||||
`Routine '${routineId}' does not have webhook trigger type`
|
||||
);
|
||||
}
|
||||
|
||||
if (!routine.enabled) {
|
||||
throw new Error(`Routine '${routineId}' is disabled`);
|
||||
}
|
||||
|
||||
return this.executeRoutine(routineId, "webhook", { webhookPayload: payload });
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of currently-running executions.
|
||||
*/
|
||||
getInFlightCount(): number {
|
||||
return this.inFlightExecutions.size;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a routine is currently being executed.
|
||||
*/
|
||||
isRoutineRunning(routineId: string): boolean {
|
||||
return this.inFlightExecutions.has(routineId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,253 +1,458 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { Routine, RoutineStore, ProjectSettings } from "@fusion/core";
|
||||
import { RoutineScheduler } from "./routine-scheduler.js";
|
||||
import { RoutineScheduler, type RoutineSchedulerOptions } from "./routine-scheduler.js";
|
||||
import type { RoutineStore, Routine, TaskStore, RoutineExecutionResult, Settings } from "@fusion/core";
|
||||
import type { RoutineRunner } from "./routine-runner.js";
|
||||
|
||||
// Mock the logger
|
||||
vi.mock("./logger.js", () => ({
|
||||
createLogger: () => ({
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
// Default settings inline to avoid @fusion/core build dependency during tests
|
||||
const DEFAULT_SETTINGS: Settings = {
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
pollIntervalMs: 30000,
|
||||
autoResolveConflicts: true,
|
||||
requirePlanApproval: false,
|
||||
recycleWorktrees: false,
|
||||
worktreeNaming: "random",
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
ntfyEnabled: false,
|
||||
defaultProvider: "anthropic",
|
||||
defaultModelId: "claude-sonnet-4-5",
|
||||
planningProvider: "anthropic",
|
||||
planningModelId: "claude-sonnet-4-5",
|
||||
validatorProvider: "openai",
|
||||
validatorModelId: "gpt-4o",
|
||||
taskStuckTimeoutMs: undefined,
|
||||
groupOverlappingFiles: false,
|
||||
autoMerge: true,
|
||||
};
|
||||
|
||||
function createMockRoutine(overrides: Partial<Routine> = {}): Routine {
|
||||
return {
|
||||
id: "test-routine-id",
|
||||
agentId: "test-agent",
|
||||
name: "Test Routine",
|
||||
description: "A test routine",
|
||||
trigger: { type: "cron", cronExpression: "0 * * * *" },
|
||||
catchUpPolicy: "run_one",
|
||||
executionPolicy: "parallel",
|
||||
enabled: true,
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
cronExpression: "0 * * * *",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function createMockTaskStore(settingsOverrides: Partial<Settings> = {}): TaskStore {
|
||||
return {
|
||||
getSettings: vi.fn().mockResolvedValue({
|
||||
...DEFAULT_SETTINGS,
|
||||
...settingsOverrides,
|
||||
}),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as TaskStore;
|
||||
}
|
||||
|
||||
function createMockRoutineStore(routines: Routine[] = []): RoutineStore {
|
||||
const routineMap = new Map(routines.map((r) => [r.id, r]));
|
||||
|
||||
return {
|
||||
getRoutine: vi.fn().mockImplementation((id: string) => {
|
||||
const routine = routineMap.get(id);
|
||||
if (!routine) {
|
||||
throw Object.assign(new Error(`Routine '${id}' not found`), { code: "ENOENT" });
|
||||
}
|
||||
return routine;
|
||||
}),
|
||||
listRoutines: vi.fn().mockResolvedValue(routines),
|
||||
updateRoutine: vi.fn().mockImplementation((id: string, _updates: any) => {
|
||||
const routine = routineMap.get(id);
|
||||
if (!routine) {
|
||||
throw Object.assign(new Error(`Routine '${id}' not found`), { code: "ENOENT" });
|
||||
}
|
||||
return routine;
|
||||
}),
|
||||
getDueRoutines: vi.fn().mockResolvedValue(routines),
|
||||
recordRun: vi.fn().mockImplementation((id: string, result: RoutineExecutionResult) => {
|
||||
return createMockRoutine({ id, lastRunResult: result });
|
||||
}),
|
||||
startRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
completeRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
cancelRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
init: vi.fn().mockResolvedValue(undefined),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as RoutineStore;
|
||||
}
|
||||
|
||||
function createMockRoutineRunner(): RoutineRunner {
|
||||
return {
|
||||
executeRoutine: vi.fn().mockResolvedValue({
|
||||
routineId: "test-routine",
|
||||
success: true,
|
||||
output: "Success",
|
||||
startedAt: new Date().toISOString(),
|
||||
completedAt: new Date().toISOString(),
|
||||
} as RoutineExecutionResult),
|
||||
handleCatchUp: vi.fn().mockResolvedValue(undefined),
|
||||
getInFlightCount: vi.fn().mockReturnValue(0),
|
||||
isRoutineRunning: vi.fn().mockReturnValue(false),
|
||||
} as unknown as RoutineRunner;
|
||||
}
|
||||
|
||||
function createRoutineScheduler(
|
||||
taskStore?: TaskStore,
|
||||
routineStore?: RoutineStore,
|
||||
routineRunner?: RoutineRunner,
|
||||
options?: Partial<Pick<RoutineSchedulerOptions, "pollIntervalMs">>,
|
||||
): RoutineScheduler {
|
||||
return new RoutineScheduler({
|
||||
taskStore: taskStore ?? createMockTaskStore(),
|
||||
routineStore: routineStore ?? createMockRoutineStore(),
|
||||
routineRunner: routineRunner ?? createMockRoutineRunner(),
|
||||
pollIntervalMs: options?.pollIntervalMs,
|
||||
});
|
||||
}
|
||||
|
||||
describe("RoutineScheduler", () => {
|
||||
let mockRoutineStore: RoutineStore;
|
||||
let mockRoutineRunner: RoutineRunner;
|
||||
let mockGetSettings: ReturnType<typeof vi.fn>;
|
||||
let onStart: ReturnType<typeof vi.fn>;
|
||||
let onStop: ReturnType<typeof vi.fn>;
|
||||
let scheduler: RoutineScheduler;
|
||||
|
||||
const createMockRoutine = (overrides: Partial<Routine> = {}): Routine =>
|
||||
({
|
||||
id: "routine-1",
|
||||
agentId: "agent-1",
|
||||
name: "Test Routine",
|
||||
description: "A test routine",
|
||||
enabled: true,
|
||||
trigger: {
|
||||
type: "cron",
|
||||
cronExpression: "*/5 * * * *",
|
||||
},
|
||||
executionPolicy: "reject",
|
||||
catchUpPolicy: "skip",
|
||||
catchUpLimit: 5,
|
||||
lastRunAt: null,
|
||||
nextRunAt: new Date().toISOString(),
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
...overrides,
|
||||
} as Routine);
|
||||
|
||||
beforeEach(() => {
|
||||
mockRoutineStore = {
|
||||
getRoutine: vi.fn(),
|
||||
listRoutines: vi.fn(),
|
||||
createRoutine: vi.fn(),
|
||||
updateRoutine: vi.fn(),
|
||||
deleteRoutine: vi.fn(),
|
||||
getDueRoutines: vi.fn().mockResolvedValue([]),
|
||||
startRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
completeRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
cancelRoutineExecution: vi.fn().mockResolvedValue(undefined),
|
||||
recordRun: vi.fn(),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
} as unknown as RoutineStore;
|
||||
|
||||
mockRoutineRunner = {
|
||||
execute: vi.fn().mockResolvedValue({ success: true }),
|
||||
isExecuting: vi.fn().mockReturnValue(false),
|
||||
determineCatchUp: vi.fn().mockReturnValue({ shouldCatchUp: false }),
|
||||
clearInFlight: vi.fn(),
|
||||
} as unknown as RoutineRunner;
|
||||
|
||||
mockGetSettings = vi.fn().mockReturnValue({
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
});
|
||||
|
||||
onStart = vi.fn();
|
||||
onStop = vi.fn();
|
||||
|
||||
scheduler = new RoutineScheduler({
|
||||
routineStore: mockRoutineStore,
|
||||
routineRunner: mockRoutineRunner,
|
||||
pollIntervalMs: 1000, // Fast for testing
|
||||
getSettings: mockGetSettings as () => ProjectSettings,
|
||||
onStart,
|
||||
onStop,
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
scheduler.stop();
|
||||
if (scheduler) {
|
||||
scheduler.stop();
|
||||
}
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("lifecycle", () => {
|
||||
it("should start and stop correctly", () => {
|
||||
scheduler.start();
|
||||
expect(scheduler.getStatus()).toBe("running");
|
||||
expect(onStart).toHaveBeenCalled();
|
||||
describe("constructor", () => {
|
||||
it("clamps poll interval to minimum 10000ms", () => {
|
||||
scheduler = createRoutineScheduler(
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
{ pollIntervalMs: 100 }, // Below minimum
|
||||
);
|
||||
|
||||
scheduler.stop();
|
||||
expect(scheduler.getStatus()).toBe("stopped");
|
||||
expect(onStop).toHaveBeenCalled();
|
||||
// The scheduler should have clamped to 10000ms
|
||||
expect(scheduler["pollIntervalMs"]).toBe(10000);
|
||||
});
|
||||
|
||||
it("should not start twice", () => {
|
||||
scheduler.start();
|
||||
scheduler.start(); // Second start should be no-op
|
||||
expect(onStart).toHaveBeenCalledTimes(1);
|
||||
scheduler.stop();
|
||||
});
|
||||
|
||||
it("should not stop twice", () => {
|
||||
scheduler.start();
|
||||
scheduler.stop();
|
||||
scheduler.stop(); // Second stop should be no-op
|
||||
expect(onStop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("should subscribe to routine store events on start", () => {
|
||||
scheduler.start();
|
||||
expect(mockRoutineStore.on).toHaveBeenCalledWith(
|
||||
"routine:created",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.on).toHaveBeenCalledWith(
|
||||
"routine:updated",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.on).toHaveBeenCalledWith(
|
||||
"routine:deleted",
|
||||
expect.any(Function)
|
||||
);
|
||||
});
|
||||
|
||||
it("should unsubscribe from routine store events on stop", () => {
|
||||
scheduler.start();
|
||||
scheduler.stop();
|
||||
expect(mockRoutineStore.off).toHaveBeenCalledWith(
|
||||
"routine:created",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.off).toHaveBeenCalledWith(
|
||||
"routine:updated",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockRoutineStore.off).toHaveBeenCalledWith(
|
||||
"routine:deleted",
|
||||
expect.any(Function)
|
||||
);
|
||||
it("uses default poll interval of 60000ms when not specified", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
expect(scheduler["pollIntervalMs"]).toBe(60000);
|
||||
});
|
||||
});
|
||||
|
||||
describe("poll behavior", () => {
|
||||
it("should skip poll when globalPause is true", async () => {
|
||||
mockGetSettings.mockReturnValue({ globalPause: true });
|
||||
describe("start/stop", () => {
|
||||
it("sets running = true after start", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineStore.getDueRoutines).not.toHaveBeenCalled();
|
||||
expect(scheduler.isActive()).toBe(true);
|
||||
});
|
||||
|
||||
it("should skip poll when enginePaused is true", async () => {
|
||||
mockGetSettings.mockReturnValue({ enginePaused: true });
|
||||
it("clears interval and sets running = false after stop", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineStore.getDueRoutines).not.toHaveBeenCalled();
|
||||
scheduler.stop();
|
||||
expect(scheduler.isActive()).toBe(false);
|
||||
});
|
||||
|
||||
it("should process due routines", async () => {
|
||||
const routine = createMockRoutine();
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
it("runs first tick immediately on start", async () => {
|
||||
const routineStore = createMockRoutineStore([createMockRoutine({ id: "due-routine" })]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
scheduler.start();
|
||||
// Wait for the tick to complete
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
scheduler.stop();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineStore.getDueRoutines).toHaveBeenCalled();
|
||||
expect(mockRoutineRunner.execute).toHaveBeenCalledWith(routine);
|
||||
expect(routineRunner.handleCatchUp).toHaveBeenCalled();
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should skip disabled routines", async () => {
|
||||
const routine = createMockRoutine({ enabled: false });
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
it("is idempotent on start", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
scheduler.start();
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
|
||||
expect(mockRoutineRunner.execute).not.toHaveBeenCalled();
|
||||
scheduler.start(); // Should not throw
|
||||
expect(scheduler.isActive()).toBe(true);
|
||||
});
|
||||
|
||||
it("should skip routines that are already executing", async () => {
|
||||
const routine = createMockRoutine();
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
vi.mocked(mockRoutineRunner.isExecuting).mockReturnValue(true);
|
||||
scheduler.start();
|
||||
it("is safe to stop when not started", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
expect(() => scheduler.stop()).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
describe("tick", () => {
|
||||
it("skips when globalPause is true", async () => {
|
||||
const routineStore = createMockRoutineStore([createMockRoutine()]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(
|
||||
createMockTaskStore({ globalPause: true }),
|
||||
routineStore,
|
||||
routineRunner,
|
||||
);
|
||||
|
||||
expect(mockRoutineRunner.execute).not.toHaveBeenCalled();
|
||||
await scheduler.tick();
|
||||
|
||||
expect(routineStore.getDueRoutines).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should handle catch-up execution", async () => {
|
||||
const routine = createMockRoutine({ catchUpPolicy: "run_one" });
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([routine]);
|
||||
vi.mocked(mockRoutineRunner.determineCatchUp).mockReturnValue({
|
||||
shouldCatchUp: true,
|
||||
catchUpFrom: "2024-01-01T00:00:00.000Z",
|
||||
});
|
||||
scheduler.start();
|
||||
it("skips when enginePaused is true", async () => {
|
||||
const routineStore = createMockRoutineStore([createMockRoutine()]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(
|
||||
createMockTaskStore({ enginePaused: true }),
|
||||
routineStore,
|
||||
routineRunner,
|
||||
);
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
await scheduler.tick();
|
||||
|
||||
expect(mockRoutineRunner.execute).toHaveBeenCalledWith(routine, {
|
||||
catchUpFrom: "2024-01-01T00:00:00.000Z",
|
||||
});
|
||||
expect(routineStore.getDueRoutines).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should isolate per-routine failures", async () => {
|
||||
it("skips when no routines are due", async () => {
|
||||
const routineStore = createMockRoutineStore([]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
await scheduler.tick();
|
||||
|
||||
expect(routineRunner.handleCatchUp).not.toHaveBeenCalled();
|
||||
expect(routineRunner.executeRoutine).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("processes due routines in order", async () => {
|
||||
const routine1 = createMockRoutine({ id: "routine-1" });
|
||||
const routine2 = createMockRoutine({ id: "routine-2" });
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockResolvedValue([
|
||||
routine1,
|
||||
routine2,
|
||||
]);
|
||||
vi.mocked(mockRoutineRunner.execute)
|
||||
.mockRejectedValueOnce(new Error("Failed"))
|
||||
.mockResolvedValueOnce({ success: true, routineId: "routine-2", executedAt: new Date().toISOString() });
|
||||
scheduler.start();
|
||||
const routineStore = createMockRoutineStore([routine1, routine2]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 1500));
|
||||
await scheduler.tick();
|
||||
|
||||
// Both should have been attempted despite the first failure
|
||||
expect(mockRoutineRunner.execute).toHaveBeenCalledTimes(2);
|
||||
// Both routines should be processed
|
||||
expect(routineRunner.handleCatchUp).toHaveBeenCalledTimes(2);
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("calls handleCatchUp then executeRoutine for each routine", async () => {
|
||||
const routine1 = createMockRoutine({ id: "routine-order-1" });
|
||||
const routine2 = createMockRoutine({ id: "routine-order-2" });
|
||||
const routineStore = createMockRoutineStore([routine1, routine2]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
await scheduler.tick();
|
||||
|
||||
// Both routines should be processed
|
||||
expect(routineRunner.handleCatchUp).toHaveBeenCalledTimes(2);
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("handles errors in individual routine execution gracefully", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-error" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
vi.mocked(routineRunner.executeRoutine).mockRejectedValueOnce(
|
||||
new Error("Execution failed"),
|
||||
);
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
// Should not throw
|
||||
await expect(scheduler.tick()).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("re-entrance guard prevents overlapping ticks", async () => {
|
||||
const routine = createMockRoutine({ id: "routine-reentrant" });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
// Make execution slow
|
||||
vi.mocked(routineRunner.executeRoutine).mockImplementation(async () => {
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
return {
|
||||
routineId: "routine-reentrant",
|
||||
success: true,
|
||||
output: "",
|
||||
startedAt: new Date().toISOString(),
|
||||
completedAt: new Date().toISOString(),
|
||||
} as RoutineExecutionResult;
|
||||
});
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
// Start two ticks concurrently
|
||||
const [tick1, tick2] = [scheduler.tick(), scheduler.tick()];
|
||||
await Promise.all([tick1, tick2]);
|
||||
|
||||
// Should only have executed once (second tick was blocked by re-entrance guard)
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("re-entrance guard", () => {
|
||||
it("should skip concurrent polls", async () => {
|
||||
// Make getDueRoutines slow
|
||||
vi.mocked(mockRoutineStore.getDueRoutines).mockImplementation(
|
||||
() => new Promise((resolve) => setTimeout(() => resolve([]), 500))
|
||||
describe("triggerManual", () => {
|
||||
it("delegates to routineRunner.executeRoutine with 'api' trigger", async () => {
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), createMockRoutineStore(), routineRunner);
|
||||
|
||||
await scheduler.triggerManual("test-routine");
|
||||
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalledWith("test-routine", "api");
|
||||
});
|
||||
|
||||
it("passes through the result from routineRunner", async () => {
|
||||
const mockResult: RoutineExecutionResult = {
|
||||
routineId: "test-routine",
|
||||
success: true,
|
||||
output: "Success",
|
||||
startedAt: new Date().toISOString(),
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
vi.mocked(routineRunner.executeRoutine).mockResolvedValue(mockResult);
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), createMockRoutineStore(), routineRunner);
|
||||
|
||||
const result = await scheduler.triggerManual("test-routine");
|
||||
|
||||
expect(result).toEqual(mockResult);
|
||||
});
|
||||
});
|
||||
|
||||
describe("triggerWebhook", () => {
|
||||
it("delegates to routineRunner.executeRoutine with 'webhook' trigger", async () => {
|
||||
const routine = createMockRoutine({ id: "webhook-routine", trigger: { type: "webhook", webhookPath: "/test" } });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
await scheduler.triggerWebhook("webhook-routine", { data: "test" });
|
||||
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalledWith(
|
||||
"webhook-routine",
|
||||
"webhook",
|
||||
expect.objectContaining({ webhookPayload: { data: "test" } }),
|
||||
);
|
||||
});
|
||||
|
||||
it("throws for routine with non-webhook trigger type", async () => {
|
||||
const routine = createMockRoutine({ id: "cron-routine", trigger: { type: "cron", cronExpression: "0 * * * *" } });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore);
|
||||
|
||||
await expect(scheduler.triggerWebhook("cron-routine", {})).rejects.toThrow(
|
||||
"does not have webhook trigger type",
|
||||
);
|
||||
});
|
||||
|
||||
it("throws for nonexistent routine", async () => {
|
||||
const routineStore = createMockRoutineStore([]);
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore);
|
||||
|
||||
await expect(scheduler.triggerWebhook("nonexistent", {})).rejects.toThrow(
|
||||
"not found",
|
||||
);
|
||||
});
|
||||
|
||||
it("proceeds without signature check when no secret is configured", async () => {
|
||||
// Clear the env var if it exists
|
||||
const originalSecret = process.env.FUSION_ROUTINE_WEBHOOK_SECRET;
|
||||
delete process.env.FUSION_ROUTINE_WEBHOOK_SECRET;
|
||||
|
||||
try {
|
||||
const routine = createMockRoutine({ id: "webhook-no-secret", trigger: { type: "webhook", webhookPath: "/test" } });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
// Should not throw even without signature
|
||||
await scheduler.triggerWebhook("webhook-no-secret", { data: "test" });
|
||||
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalled();
|
||||
} finally {
|
||||
// Restore env var
|
||||
if (originalSecret !== undefined) {
|
||||
process.env.FUSION_ROUTINE_WEBHOOK_SECRET = originalSecret;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("throws for invalid HMAC signature when secret is configured", async () => {
|
||||
process.env.FUSION_ROUTINE_WEBHOOK_SECRET = "test-secret";
|
||||
|
||||
try {
|
||||
const routine = createMockRoutine({ id: "webhook-with-secret", trigger: { type: "webhook", webhookPath: "/test" } });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore);
|
||||
|
||||
await expect(
|
||||
scheduler.triggerWebhook("webhook-with-secret", { data: "test" }, "sha256=invalidsignature"),
|
||||
).rejects.toThrow("Invalid webhook signature");
|
||||
} finally {
|
||||
delete process.env.FUSION_ROUTINE_WEBHOOK_SECRET;
|
||||
}
|
||||
});
|
||||
|
||||
it("accepts valid HMAC signature", async () => {
|
||||
const secret = "test-secret-123";
|
||||
process.env.FUSION_ROUTINE_WEBHOOK_SECRET = secret;
|
||||
|
||||
try {
|
||||
const routine = createMockRoutine({ id: "webhook-valid-sig", trigger: { type: "webhook", webhookPath: "/test" } });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
const routineRunner = createMockRoutineRunner();
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore, routineRunner);
|
||||
|
||||
const payload = { data: "test" };
|
||||
const crypto = await import("node:crypto");
|
||||
const signature = "sha256=" + crypto.createHmac("sha256", secret).update(JSON.stringify(payload)).digest("hex");
|
||||
|
||||
await scheduler.triggerWebhook("webhook-valid-sig", payload, signature);
|
||||
|
||||
expect(routineRunner.executeRoutine).toHaveBeenCalled();
|
||||
} finally {
|
||||
delete process.env.FUSION_ROUTINE_WEBHOOK_SECRET;
|
||||
}
|
||||
});
|
||||
|
||||
it("throws when signature is missing but secret is configured", async () => {
|
||||
process.env.FUSION_ROUTINE_WEBHOOK_SECRET = "test-secret";
|
||||
|
||||
try {
|
||||
const routine = createMockRoutine({ id: "webhook-missing-sig", trigger: { type: "webhook", webhookPath: "/test" } });
|
||||
const routineStore = createMockRoutineStore([routine]);
|
||||
scheduler = createRoutineScheduler(createMockTaskStore(), routineStore);
|
||||
|
||||
await expect(
|
||||
scheduler.triggerWebhook("webhook-missing-sig", { data: "test" }),
|
||||
).rejects.toThrow("Missing webhook signature");
|
||||
} finally {
|
||||
delete process.env.FUSION_ROUTINE_WEBHOOK_SECRET;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("isActive", () => {
|
||||
it("returns false initially", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
expect(scheduler.isActive()).toBe(false);
|
||||
});
|
||||
|
||||
it("returns true after start", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
scheduler.start();
|
||||
expect(scheduler.isActive()).toBe(true);
|
||||
scheduler.stop();
|
||||
});
|
||||
|
||||
// Trigger immediate poll
|
||||
await scheduler.triggerPoll();
|
||||
|
||||
// The slow poll should be running, triggerPoll should complete quickly
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
expect(scheduler.getStatus()).toBe("running");
|
||||
it("returns false after stop", () => {
|
||||
scheduler = createRoutineScheduler();
|
||||
scheduler.start();
|
||||
scheduler.stop();
|
||||
expect(scheduler.isActive()).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,4 +1,16 @@
|
||||
import type { Routine, RoutineStore, ProjectSettings } from "@fusion/core";
|
||||
/**
|
||||
* RoutineScheduler — polls for due routines and triggers their execution via RoutineRunner.
|
||||
*
|
||||
* Handles:
|
||||
* - Polling interval with configurable interval
|
||||
* - Re-entrance guard (prevents overlapping polls)
|
||||
* - Pause awareness (globalPause / enginePaused)
|
||||
* - Catch-up execution before normal due execution
|
||||
* - Per-routine failure isolation
|
||||
*/
|
||||
|
||||
import { CronExpressionParser } from "cron-parser";
|
||||
import type { Routine, RoutineStore, TaskStore } from "@fusion/core";
|
||||
import { RoutineRunner } from "./routine-runner.js";
|
||||
import { createLogger } from "./logger.js";
|
||||
|
||||
@@ -8,151 +20,98 @@ const logger = createLogger("routine-scheduler");
|
||||
* Options for RoutineScheduler.
|
||||
*/
|
||||
export interface RoutineSchedulerOptions {
|
||||
/** The routine store */
|
||||
/** TaskStore for checking pause state */
|
||||
taskStore: TaskStore;
|
||||
/** RoutineStore for querying routines */
|
||||
routineStore: RoutineStore;
|
||||
/** The routine runner */
|
||||
/** RoutineRunner for executing routines */
|
||||
routineRunner: RoutineRunner;
|
||||
/** Polling interval in milliseconds */
|
||||
/** Polling interval in milliseconds. Default: 60000 (60s). Minimum: 10000 (10s). */
|
||||
pollIntervalMs?: number;
|
||||
/** Get current settings (for pause checks) */
|
||||
getSettings: () => ProjectSettings;
|
||||
/** Callback when scheduler is started */
|
||||
onStart?: () => void;
|
||||
/** Callback when scheduler is stopped */
|
||||
onStop?: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimum poll interval (30 seconds).
|
||||
*/
|
||||
const MIN_POLL_INTERVAL_MS = 30_000;
|
||||
|
||||
/**
|
||||
* Default poll interval (1 minute).
|
||||
*/
|
||||
const DEFAULT_POLL_INTERVAL_MS = 60_000;
|
||||
|
||||
/**
|
||||
* RoutineScheduler polls for due routines and triggers their execution.
|
||||
*
|
||||
* It handles:
|
||||
* - Polling interval with clamping
|
||||
* - Re-entrance guard (prevents overlapping polls)
|
||||
* - Pause awareness (globalPause / enginePaused)
|
||||
* - Catch-up execution before normal due execution
|
||||
* - Per-routine failure isolation
|
||||
*/
|
||||
export class RoutineScheduler {
|
||||
private taskStore: TaskStore;
|
||||
private routineStore: RoutineStore;
|
||||
private routineRunner: RoutineRunner;
|
||||
private pollIntervalMs: number;
|
||||
private getSettings: () => ProjectSettings;
|
||||
private onStart?: () => void;
|
||||
private onStop?: () => void;
|
||||
|
||||
private pollTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private isRunning = false;
|
||||
private isPolling = false;
|
||||
private running: boolean = false;
|
||||
private ticking: boolean = false;
|
||||
private pollInterval: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
constructor(options: RoutineSchedulerOptions) {
|
||||
this.taskStore = options.taskStore;
|
||||
this.routineStore = options.routineStore;
|
||||
this.routineRunner = options.routineRunner;
|
||||
this.getSettings = options.getSettings;
|
||||
this.onStart = options.onStart;
|
||||
this.onStop = options.onStop;
|
||||
|
||||
// Clamp poll interval to minimum
|
||||
this.pollIntervalMs = Math.max(
|
||||
options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS,
|
||||
MIN_POLL_INTERVAL_MS
|
||||
);
|
||||
this.pollIntervalMs = Math.max(10000, options.pollIntervalMs ?? 60000);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the scheduler.
|
||||
*/
|
||||
start(): void {
|
||||
if (this.isRunning) {
|
||||
if (this.running) {
|
||||
logger.log("RoutineScheduler already running");
|
||||
return;
|
||||
}
|
||||
|
||||
this.isRunning = true;
|
||||
logger.log(
|
||||
`RoutineScheduler started with ${this.pollIntervalMs}ms poll interval`
|
||||
);
|
||||
this.running = true;
|
||||
logger.log(`RoutineScheduler started with ${this.pollIntervalMs}ms poll interval`);
|
||||
|
||||
// Subscribe to routine store events
|
||||
this.routineStore.on("routine:created", this.handleRoutineCreated);
|
||||
this.routineStore.on("routine:updated", this.handleRoutineUpdated);
|
||||
this.routineStore.on("routine:deleted", this.handleRoutineDeleted);
|
||||
// Run first tick immediately
|
||||
void this.tick();
|
||||
|
||||
// Start polling
|
||||
this.pollTimer = setInterval(() => {
|
||||
void this.poll();
|
||||
// Start polling interval
|
||||
this.pollInterval = setInterval(() => {
|
||||
void this.tick();
|
||||
}, this.pollIntervalMs);
|
||||
|
||||
// Run initial poll
|
||||
void this.poll();
|
||||
|
||||
this.onStart?.();
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the scheduler.
|
||||
*/
|
||||
stop(): void {
|
||||
if (!this.isRunning) {
|
||||
if (!this.running) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.isRunning = false;
|
||||
this.running = false;
|
||||
logger.log("RoutineScheduler stopping");
|
||||
|
||||
// Clear timer
|
||||
if (this.pollTimer) {
|
||||
clearInterval(this.pollTimer);
|
||||
this.pollTimer = null;
|
||||
if (this.pollInterval) {
|
||||
clearInterval(this.pollInterval);
|
||||
this.pollInterval = null;
|
||||
}
|
||||
|
||||
// Unsubscribe from events
|
||||
this.routineStore.off("routine:created", this.handleRoutineCreated);
|
||||
this.routineStore.off("routine:updated", this.handleRoutineUpdated);
|
||||
this.routineStore.off("routine:deleted", this.handleRoutineDeleted);
|
||||
|
||||
this.onStop?.();
|
||||
logger.log("RoutineScheduler stopped");
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the scheduler is running.
|
||||
* Check if the scheduler is active (running).
|
||||
*/
|
||||
getStatus(): "running" | "stopped" {
|
||||
return this.isRunning ? "running" : "stopped";
|
||||
isActive(): boolean {
|
||||
return this.running;
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger an immediate poll (for testing).
|
||||
* Process a single tick — poll for due routines and execute them.
|
||||
*/
|
||||
async triggerPoll(): Promise<void> {
|
||||
await this.poll();
|
||||
}
|
||||
|
||||
/**
|
||||
* Poll for due routines and execute them.
|
||||
*/
|
||||
private async poll(): Promise<void> {
|
||||
async tick(): Promise<void> {
|
||||
// Re-entrance guard
|
||||
if (this.isPolling) {
|
||||
logger.log("Poll already in progress, skipping");
|
||||
if (this.ticking) {
|
||||
logger.log("Tick already in progress, skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
this.isPolling = true;
|
||||
this.ticking = true;
|
||||
|
||||
try {
|
||||
// Check pause state
|
||||
const settings = this.getSettings();
|
||||
const settings = await this.taskStore.getSettings();
|
||||
if (settings.globalPause || settings.enginePaused) {
|
||||
logger.log(
|
||||
`Paused: globalPause=${settings.globalPause}, enginePaused=${settings.enginePaused}`
|
||||
@@ -161,86 +120,117 @@ export class RoutineScheduler {
|
||||
}
|
||||
|
||||
// Get due routines
|
||||
const dueRoutines = await this.routineStore.getDueRoutines();
|
||||
const dueRoutines = await this.getDueRoutines();
|
||||
if (dueRoutines.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
logger.log(`Found ${dueRoutines.length} due routines`);
|
||||
|
||||
// Process each routine
|
||||
for (const routine of dueRoutines) {
|
||||
await this.processRoutine(routine);
|
||||
// Re-check pause state (may have changed mid-loop)
|
||||
const currentSettings = await this.taskStore.getSettings();
|
||||
if (currentSettings.globalPause || currentSettings.enginePaused) {
|
||||
logger.log("Paused mid-loop, stopping processing");
|
||||
break;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.processRoutine(routine);
|
||||
} catch (err) {
|
||||
logger.error(`[${routine.id}] Failed to process: ${err}`);
|
||||
// Continue to next routine
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Poll error: ${error}`);
|
||||
} finally {
|
||||
this.isPolling = false;
|
||||
this.ticking = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a single routine, handling catch-up and normal execution.
|
||||
* Process a single routine.
|
||||
*/
|
||||
private async processRoutine(routine: Routine): Promise<void> {
|
||||
const routineId = routine.id;
|
||||
|
||||
try {
|
||||
// Skip if routine is disabled
|
||||
if (!routine.enabled) {
|
||||
logger.log(`[${routineId}] Skipped: routine is disabled`);
|
||||
return;
|
||||
}
|
||||
// Skip if disabled
|
||||
if (!routine.enabled) {
|
||||
logger.log(`[${routineId}] Skipped: routine is disabled`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Skip if already executing
|
||||
if (this.routineRunner.isExecuting(routineId)) {
|
||||
logger.log(`[${routineId}] Skipped: already executing`);
|
||||
return;
|
||||
}
|
||||
// Handle catch-up
|
||||
await this.routineRunner.handleCatchUp(routine);
|
||||
|
||||
// Handle catch-up if needed
|
||||
const lastRunAt = routine.lastRunAt ?? null;
|
||||
const currentTime = new Date();
|
||||
const catchUp = this.routineRunner.determineCatchUp(
|
||||
routine,
|
||||
lastRunAt,
|
||||
currentTime
|
||||
);
|
||||
// Execute the routine
|
||||
await this.routineRunner.executeRoutine(routineId, "cron");
|
||||
|
||||
if (catchUp.shouldCatchUp && catchUp.catchUpFrom) {
|
||||
logger.log(
|
||||
`[${routineId}] Executing catch-up from ${catchUp.catchUpFrom}`
|
||||
);
|
||||
await this.routineRunner.execute(routine, {
|
||||
catchUpFrom: catchUp.catchUpFrom,
|
||||
});
|
||||
// Update next run time
|
||||
if (routine.cronExpression) {
|
||||
try {
|
||||
const nextRun = CronExpressionParser.parse(routine.cronExpression).next();
|
||||
// Note: We can't update nextRunAt directly as it's derived from trigger
|
||||
// The RoutineStore handles this internally
|
||||
} catch (err) {
|
||||
logger.error(`[${routineId}] Failed to calculate next run: ${err}`);
|
||||
}
|
||||
|
||||
// Normal execution
|
||||
if (!this.routineRunner.isExecuting(routineId)) {
|
||||
logger.log(`[${routineId}] Executing routine`);
|
||||
await this.routineRunner.execute(routine);
|
||||
}
|
||||
} catch (error) {
|
||||
// Per-routine failure isolation - don't let one failure affect others
|
||||
logger.error(`[${routineId}] Failed to process: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle routine created event.
|
||||
* Get routines that are due for execution.
|
||||
*/
|
||||
private handleRoutineCreated = (routine: Routine): void => {
|
||||
logger.log(`[${routine.id}] Routine created, will check at next poll`);
|
||||
};
|
||||
private async getDueRoutines(): Promise<Routine[]> {
|
||||
try {
|
||||
return await this.routineStore.getDueRoutines();
|
||||
} catch (err) {
|
||||
logger.error(`Failed to get due routines: ${err}`);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle routine updated event.
|
||||
* Trigger a routine manually via the API.
|
||||
*/
|
||||
private handleRoutineUpdated = (routine: Routine): void => {
|
||||
logger.log(`[${routine.id}] Routine updated`);
|
||||
};
|
||||
async triggerManual(routineId: string): Promise<import("@fusion/core").RoutineExecutionResult> {
|
||||
return this.routineRunner.executeRoutine(routineId, "api");
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle routine deleted event.
|
||||
* Trigger a routine via webhook.
|
||||
*/
|
||||
private handleRoutineDeleted = (routine: Routine): void => {
|
||||
logger.log(`[${routine.id}] Routine deleted`);
|
||||
};
|
||||
async triggerWebhook(
|
||||
routineId: string,
|
||||
payload: Record<string, unknown>,
|
||||
signature?: string
|
||||
): Promise<import("@fusion/core").RoutineExecutionResult> {
|
||||
// Load routine to validate webhook trigger type
|
||||
const routine = await this.routineStore.getRoutine(routineId);
|
||||
|
||||
if (routine.trigger.type !== "webhook") {
|
||||
throw new Error(`Routine '${routineId}' does not have webhook trigger type`);
|
||||
}
|
||||
|
||||
// Verify webhook signature if secret is configured
|
||||
const webhookSecret = process.env.FUSION_ROUTINE_WEBHOOK_SECRET;
|
||||
if (webhookSecret) {
|
||||
if (!signature) {
|
||||
throw new Error("Missing webhook signature");
|
||||
}
|
||||
const { createHmac, timingSafeEqual } = await import("node:crypto");
|
||||
const [algo, expectedSig] = signature.split("=");
|
||||
if (algo !== "sha256") {
|
||||
throw new Error("Invalid webhook signature algorithm");
|
||||
}
|
||||
const computed = createHmac("sha256", webhookSecret).update(JSON.stringify(payload)).digest("hex");
|
||||
const sigBuffer = Buffer.from(expectedSig, "hex");
|
||||
const computedBuffer = Buffer.from(computed, "hex");
|
||||
if (sigBuffer.length !== computedBuffer.length || !timingSafeEqual(sigBuffer, computedBuffer)) {
|
||||
throw new Error("Invalid webhook signature");
|
||||
}
|
||||
}
|
||||
|
||||
return this.routineRunner.executeRoutine(routineId, "webhook", { webhookPayload: payload });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,6 +14,8 @@ import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
|
||||
import { WorktreePool } from "../worktree-pool.js";
|
||||
import { AgentSemaphore } from "../concurrency.js";
|
||||
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
|
||||
import { RoutineRunner, type RoutineRunnerOptions } from "../routine-runner.js";
|
||||
import { RoutineScheduler } from "../routine-scheduler.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeConfig,
|
||||
@@ -82,6 +84,8 @@ export class InProcessRuntime
|
||||
private pluginRunner?: PluginRunner;
|
||||
private pluginStore?: PluginStore;
|
||||
private pluginLoader?: PluginLoader;
|
||||
private routineRunner?: RoutineRunner;
|
||||
private routineScheduler?: RoutineScheduler;
|
||||
|
||||
/**
|
||||
* @param config - Runtime configuration
|
||||
@@ -345,6 +349,39 @@ export class InProcessRuntime
|
||||
runtimeLog.warn(`AgentStore initialization failed (continuing without agent monitoring):`, agentErr);
|
||||
}
|
||||
|
||||
// Initialize RoutineScheduler (requires RoutineStore from FN-1519)
|
||||
try {
|
||||
const { RoutineStore: RoutineStoreClass } = await import("@fusion/core");
|
||||
// Verify RoutineStore actually has the expected methods (FN-1519 complete)
|
||||
if (typeof RoutineStoreClass.prototype.getDueRoutines === "function") {
|
||||
const routineStore = new RoutineStoreClass(this.taskStore.getFusionDir());
|
||||
await routineStore.init();
|
||||
|
||||
if (this.heartbeatMonitor) {
|
||||
const routineRunnerOptions: RoutineRunnerOptions = {
|
||||
routineStore,
|
||||
heartbeatMonitor: this.heartbeatMonitor,
|
||||
rootDir: this.config.workingDirectory,
|
||||
};
|
||||
this.routineRunner = new RoutineRunner(routineRunnerOptions);
|
||||
|
||||
this.routineScheduler = new RoutineScheduler({
|
||||
taskStore: this.taskStore,
|
||||
routineStore,
|
||||
routineRunner: this.routineRunner,
|
||||
pollIntervalMs: 60000,
|
||||
});
|
||||
this.routineScheduler.start();
|
||||
runtimeLog.log("RoutineScheduler initialized and started");
|
||||
}
|
||||
} else {
|
||||
runtimeLog.log("RoutineStore not available (FN-1519 types not complete) — skipping RoutineScheduler");
|
||||
}
|
||||
} catch (routineErr) {
|
||||
// Non-fatal — RoutineStore may not be exported if FN-1519 is not complete
|
||||
runtimeLog.warn("RoutineScheduler initialization skipped:", routineErr instanceof Error ? routineErr.message : routineErr);
|
||||
}
|
||||
|
||||
// 7. Initialize SelfHealingManager
|
||||
this.selfHealingManager = new SelfHealingManager(this.taskStore, {
|
||||
rootDir: this.config.workingDirectory,
|
||||
@@ -420,31 +457,37 @@ export class InProcessRuntime
|
||||
runtimeLog.log("SelfHealingManager stopped");
|
||||
}
|
||||
|
||||
// 2. Stop trigger scheduler
|
||||
// 2. Stop routine scheduler (stops new routine triggers; in-flight executions continue)
|
||||
if (this.routineScheduler) {
|
||||
this.routineScheduler.stop();
|
||||
runtimeLog.log("RoutineScheduler stopped");
|
||||
}
|
||||
|
||||
// 3. Stop trigger scheduler
|
||||
if (this.triggerScheduler) {
|
||||
this.triggerScheduler.stop();
|
||||
runtimeLog.log("TriggerScheduler stopped");
|
||||
}
|
||||
|
||||
// 3. Stop stuck task detector
|
||||
// 4. Stop stuck task detector
|
||||
if (this.stuckTaskDetector) {
|
||||
this.stuckTaskDetector.stop();
|
||||
runtimeLog.log("StuckTaskDetector stopped");
|
||||
}
|
||||
|
||||
// 4. Stop heartbeat monitor
|
||||
// 5. Stop heartbeat monitor
|
||||
if (this.heartbeatMonitor) {
|
||||
this.heartbeatMonitor.stop();
|
||||
runtimeLog.log("HeartbeatMonitor stopped");
|
||||
}
|
||||
|
||||
// 5. Stop scheduler (prevents new task scheduling)
|
||||
// 6. Stop scheduler (prevents new task scheduling)
|
||||
if (this.scheduler) {
|
||||
this.scheduler.stop();
|
||||
runtimeLog.log("Scheduler stopped");
|
||||
}
|
||||
|
||||
// 2. Wait for active tasks to complete (30 second timeout)
|
||||
// 7. Wait for active tasks to complete (30 second timeout)
|
||||
const shutdownTimeout = 30000;
|
||||
const startTime = Date.now();
|
||||
|
||||
@@ -467,13 +510,13 @@ export class InProcessRuntime
|
||||
);
|
||||
}
|
||||
|
||||
// 6. Shutdown plugin runner
|
||||
// 8. Shutdown plugin runner
|
||||
if (this.pluginRunner) {
|
||||
await this.pluginRunner.shutdown();
|
||||
runtimeLog.log("PluginRunner shutdown complete");
|
||||
}
|
||||
|
||||
// 7. Drain and cleanup worktree pool
|
||||
// 9. Drain and cleanup worktree pool
|
||||
if (this.worktreePool) {
|
||||
const worktrees = this.worktreePool.drain();
|
||||
if (worktrees.length > 0) {
|
||||
@@ -560,6 +603,22 @@ export class InProcessRuntime
|
||||
return this.triggerScheduler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the RoutineRunner instance (if initialized).
|
||||
* Returns undefined when RoutineStore is not available.
|
||||
*/
|
||||
getRoutineRunner(): RoutineRunner | undefined {
|
||||
return this.routineRunner;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the RoutineScheduler instance (if initialized).
|
||||
* Returns undefined when RoutineStore is not available.
|
||||
*/
|
||||
getRoutineScheduler(): RoutineScheduler | undefined {
|
||||
return this.routineScheduler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a heartbeat run for an agent.
|
||||
*
|
||||
|
||||
Reference in New Issue
Block a user