feat(FN-1715): add scope-aware automation and routine scheduling
- Add scope field to automations and routines for granular control - Update automation-store and routine-store with scope-aware query methods - Extend database schema with scope column for automations and routines - Update cron-runner and routine-scheduler to respect scope boundaries - Add project context injection to in-process runtime for scoped execution - Include changeset for minor version bump
This commit is contained in:
@@ -51,7 +51,7 @@ describe("TaskStore task documents", () => {
|
||||
|
||||
expect(tableNames.has("task_documents")).toBe(true);
|
||||
expect(tableNames.has("task_document_revisions")).toBe(true);
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
const index = db
|
||||
.prepare(
|
||||
|
||||
@@ -507,7 +507,7 @@ describe("AutomationStore", () => {
|
||||
// Let's just test that getDueSchedules works with disabled/enabled correctly.
|
||||
|
||||
// For the actual due test, verify the schedule is NOT due (nextRunAt is in the future)
|
||||
const due = await store.getDueSchedules();
|
||||
const due = await store.getDueSchedules("project");
|
||||
// The schedule's nextRunAt is in the future after recordRun, so it shouldn't be due
|
||||
// Instead, let's verify it returns enabled schedules only
|
||||
expect(Array.isArray(due)).toBe(true);
|
||||
@@ -523,7 +523,7 @@ describe("AutomationStore", () => {
|
||||
enabled: false,
|
||||
});
|
||||
|
||||
const due = await store.getDueSchedules();
|
||||
const due = await store.getDueSchedules("project");
|
||||
expect(due.some((d) => d.id === schedule.id)).toBe(false);
|
||||
});
|
||||
|
||||
@@ -535,7 +535,7 @@ describe("AutomationStore", () => {
|
||||
});
|
||||
|
||||
// nextRunAt is in the future by default
|
||||
const due = await store.getDueSchedules();
|
||||
const due = await store.getDueSchedules("project");
|
||||
expect(due.some((d) => d.id === schedule.id)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -71,6 +71,7 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
|
||||
lastRunResult: fromJson<AutomationRunResult>(row.lastRunResult),
|
||||
runCount: row.runCount || 0,
|
||||
runHistory: fromJson<AutomationRunResult[]>(row.runHistory) || [],
|
||||
scope: (row.scope as "global" | "project") || "project",
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
};
|
||||
@@ -81,8 +82,8 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
|
||||
INSERT OR REPLACE INTO automations (
|
||||
id, name, description, scheduleType, cronExpression, command,
|
||||
enabled, timeoutMs, steps, nextRunAt, lastRunAt, lastRunResult,
|
||||
runCount, runHistory, createdAt, updatedAt
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
runCount, runHistory, scope, createdAt, updatedAt
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`).run(
|
||||
schedule.id,
|
||||
schedule.name,
|
||||
@@ -98,6 +99,7 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
|
||||
schedule.lastRunResult ? JSON.stringify(schedule.lastRunResult) : null,
|
||||
schedule.runCount || 0,
|
||||
JSON.stringify(schedule.runHistory || []),
|
||||
schedule.scope ?? "project",
|
||||
schedule.createdAt,
|
||||
schedule.updatedAt,
|
||||
);
|
||||
@@ -237,6 +239,7 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
|
||||
timeoutMs: input.timeoutMs,
|
||||
steps: hasSteps ? input.steps : undefined,
|
||||
nextRunAt: enabled ? this.computeNextRun(cronExpression) : undefined,
|
||||
scope: input.scope ?? "project",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
};
|
||||
@@ -406,8 +409,21 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
|
||||
|
||||
/**
|
||||
* Get all schedules that are due to run (nextRunAt <= now and enabled).
|
||||
* Filters by scope: "global" or "project".
|
||||
*/
|
||||
async getDueSchedules(): Promise<ScheduledTask[]> {
|
||||
async getDueSchedules(scope: "global" | "project"): Promise<ScheduledTask[]> {
|
||||
const now = new Date().toISOString();
|
||||
const rows = this.db.prepare(
|
||||
'SELECT * FROM automations WHERE enabled = 1 AND nextRunAt IS NOT NULL AND nextRunAt <= ? AND scope = ?'
|
||||
).all(now, scope) as any[];
|
||||
return rows.map((row) => this.rowToSchedule(row));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all schedules that are due to run (nextRunAt <= now and enabled) for both scopes.
|
||||
* Returns schedules from both "global" and "project" scopes.
|
||||
*/
|
||||
async getDueSchedulesAllScopes(): Promise<ScheduledTask[]> {
|
||||
const now = new Date().toISOString();
|
||||
const rows = this.db.prepare(
|
||||
'SELECT * FROM automations WHERE enabled = 1 AND nextRunAt IS NOT NULL AND nextRunAt <= ?'
|
||||
|
||||
@@ -111,6 +111,8 @@ export interface ScheduledTask {
|
||||
timeoutMs?: number;
|
||||
/** History of recent run results (most recent first, capped at 50). */
|
||||
runHistory: AutomationRunResult[];
|
||||
/** Scope of this schedule: "global" (shared) or "project" (isolated). */
|
||||
scope?: "global" | "project";
|
||||
/** ISO-8601 timestamp of when this schedule was created. */
|
||||
createdAt: string;
|
||||
/** ISO-8601 timestamp of when this schedule was last updated. */
|
||||
@@ -130,6 +132,8 @@ export interface ScheduledTaskCreateInput {
|
||||
timeoutMs?: number;
|
||||
/** Multi-step workflow. When provided, `command` is ignored in favor of sequential step execution. */
|
||||
steps?: AutomationStep[];
|
||||
/** Scope of this schedule: "global" (shared) or "project" (isolated). Default: "project". */
|
||||
scope?: "global" | "project";
|
||||
}
|
||||
|
||||
/** Input for updating an existing scheduled task. */
|
||||
@@ -143,6 +147,8 @@ export interface ScheduledTaskUpdateInput {
|
||||
timeoutMs?: number;
|
||||
/** Multi-step workflow. When provided, `command` is ignored in favor of sequential step execution. */
|
||||
steps?: AutomationStep[];
|
||||
/** Scope of this schedule: "global" (shared) or "project" (isolated). */
|
||||
scope?: "global" | "project";
|
||||
}
|
||||
|
||||
/** Maximum number of run history entries to retain per schedule. */
|
||||
|
||||
@@ -119,7 +119,7 @@ describe("Database", () => {
|
||||
});
|
||||
|
||||
it("seeds schema version", () => {
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
});
|
||||
|
||||
it("seeds lastModified", () => {
|
||||
@@ -142,7 +142,7 @@ describe("Database", () => {
|
||||
|
||||
it("is idempotent - calling init() twice does not fail", () => {
|
||||
expect(() => db.init()).not.toThrow();
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
});
|
||||
|
||||
it("does not overwrite existing config on re-init", () => {
|
||||
@@ -749,7 +749,7 @@ describe("schema migrations", () => {
|
||||
db.init();
|
||||
|
||||
// Verify version bumped to 29 (includes v1→v2 through v26→v29)
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
// Verify new columns exist and existing data is intact
|
||||
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
|
||||
@@ -774,11 +774,11 @@ describe("schema migrations", () => {
|
||||
const db = new Database(kbDir);
|
||||
db.init();
|
||||
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
// Re-init should not fail
|
||||
db.init();
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
db.close();
|
||||
});
|
||||
@@ -794,7 +794,7 @@ describe("schema migrations", () => {
|
||||
|
||||
db.init();
|
||||
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>;
|
||||
expect(tables).toEqual([{ name: "agentRatings" }]);
|
||||
@@ -818,7 +818,7 @@ describe("schema migrations", () => {
|
||||
|
||||
db.init();
|
||||
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>;
|
||||
expect(tables).toEqual([{ name: "mission_events" }]);
|
||||
@@ -922,7 +922,7 @@ describe("schema migrations", () => {
|
||||
db.init();
|
||||
|
||||
// Verify version bumped to 29
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
|
||||
// Verify new columns exist and existing data is intact
|
||||
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
|
||||
@@ -1288,7 +1288,7 @@ describe("createDatabase factory", () => {
|
||||
const db = createDatabase(kbDir);
|
||||
db.init();
|
||||
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
expect(db.getLastModified()).toBeGreaterThan(0);
|
||||
|
||||
db.close();
|
||||
|
||||
@@ -59,7 +59,7 @@ export function fromJson<T>(json: string | null | undefined): T | undefined {
|
||||
|
||||
// ── Schema Definition ────────────────────────────────────────────────
|
||||
|
||||
const SCHEMA_VERSION = 33;
|
||||
const SCHEMA_VERSION = 34;
|
||||
|
||||
function normalizeTaskComments(
|
||||
steeringComments: SteeringComment[] | undefined,
|
||||
@@ -1398,6 +1398,20 @@ export class Database {
|
||||
`);
|
||||
});
|
||||
}
|
||||
|
||||
// Scope columns for automations and routines (FN-1714)
|
||||
// Enables dual-lane execution: global scope (shared) and project scope (isolated)
|
||||
if (version < 34) {
|
||||
this.applyMigration(34, () => {
|
||||
// Add scope column to automations table
|
||||
this.addColumnIfMissing("automations", "scope", "TEXT DEFAULT 'project'");
|
||||
this.db.exec(`CREATE INDEX IF NOT EXISTS idxAutomationsScope ON automations(scope)`);
|
||||
|
||||
// Add scope column to routines table
|
||||
this.addColumnIfMissing("routines", "scope", "TEXT DEFAULT 'project'");
|
||||
this.db.exec(`CREATE INDEX IF NOT EXISTS idxRoutinesScope ON routines(scope)`);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -776,7 +776,7 @@ describe("Migration: pre-33 DB upgrade", () => {
|
||||
// Step 1: Create a fresh database at v33 (runs all migrations up to 33)
|
||||
const db1 = createDatabase(legacyDir);
|
||||
db1.init();
|
||||
expect(db1.getSchemaVersion()).toBe(33);
|
||||
expect(db1.getSchemaVersion()).toBe(34);
|
||||
db1.close();
|
||||
|
||||
// Step 2: Manually downgrade to version 32 and drop insight tables
|
||||
@@ -811,7 +811,7 @@ describe("Migration: pre-33 DB upgrade", () => {
|
||||
expect(tableNamesBefore).not.toContain("project_insight_runs");
|
||||
// Now run init — this triggers the v32→v33 migration
|
||||
db3.init();
|
||||
expect(db3.getSchemaVersion()).toBe(33);
|
||||
expect(db3.getSchemaVersion()).toBe(34);
|
||||
|
||||
// Step 4: Verify insight tables exist after migration
|
||||
const tablesAfter = db3.prepare(
|
||||
@@ -842,12 +842,12 @@ describe("Migration: pre-33 DB upgrade", () => {
|
||||
try {
|
||||
const db1 = createDatabase(testDir);
|
||||
db1.init();
|
||||
expect(db1.getSchemaVersion()).toBe(33);
|
||||
expect(db1.getSchemaVersion()).toBe(34);
|
||||
db1.close();
|
||||
|
||||
const db2 = createDatabase(testDir);
|
||||
expect(() => db2.init()).not.toThrow();
|
||||
expect(db2.getSchemaVersion()).toBe(33);
|
||||
expect(db2.getSchemaVersion()).toBe(34);
|
||||
db2.close();
|
||||
} finally {
|
||||
rmSync(testDir, { recursive: true, force: true });
|
||||
|
||||
@@ -2544,7 +2544,7 @@ describe("MissionStore", () => {
|
||||
|
||||
describe("Loop State & Validator Run Schema (v31)", () => {
|
||||
it("schema version is 32 after migration", () => {
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
});
|
||||
|
||||
it("mission_features table has loop state columns", () => {
|
||||
|
||||
@@ -739,7 +739,7 @@ describe("RoadmapStore", () => {
|
||||
|
||||
describe("schema version", () => {
|
||||
it("schema version is 32 after init", () => {
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -520,7 +520,7 @@ describe("RoutineStore", () => {
|
||||
|
||||
describe("getDueRoutines", () => {
|
||||
it("returns empty array when no routines", async () => {
|
||||
const due = await store.getDueRoutines();
|
||||
const due = await store.getDueRoutines("project");
|
||||
expect(due).toEqual([]);
|
||||
});
|
||||
|
||||
@@ -532,7 +532,7 @@ describe("RoutineStore", () => {
|
||||
enabled: false,
|
||||
});
|
||||
|
||||
const due = await store.getDueRoutines();
|
||||
const due = await store.getDueRoutines("project");
|
||||
expect(due.some((d) => d.id === routine.id)).toBe(false);
|
||||
});
|
||||
|
||||
@@ -544,7 +544,7 @@ describe("RoutineStore", () => {
|
||||
});
|
||||
|
||||
// nextRunAt is in the future by default
|
||||
const due = await store.getDueRoutines();
|
||||
const due = await store.getDueRoutines("project");
|
||||
expect(due.some((d) => d.id === routine.id)).toBe(false);
|
||||
});
|
||||
|
||||
@@ -564,7 +564,7 @@ describe("RoutineStore", () => {
|
||||
).run(pastDate, routine.id);
|
||||
|
||||
// Now getDueRoutines should include it
|
||||
const due = await store.getDueRoutines();
|
||||
const due = await store.getDueRoutines("project");
|
||||
expect(due.some((d) => d.id === routine.id)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -119,6 +119,7 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
runHistory: fromJson<RoutineExecutionResult[]>(row.runHistory) || [],
|
||||
catchUpLimit: row.catchUpLimit ?? 5,
|
||||
cronExpression: isCronTrigger(trigger) ? trigger.cronExpression : undefined,
|
||||
scope: (row.scope as "global" | "project") || "project",
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
};
|
||||
@@ -149,8 +150,8 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
id, agentId, name, description, triggerType, triggerConfig,
|
||||
catchUpPolicy, executionPolicy, catchUpLimit, enabled,
|
||||
lastRunAt, lastRunResult, nextRunAt,
|
||||
runCount, runHistory, createdAt, updatedAt
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
runCount, runHistory, scope, createdAt, updatedAt
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`).run(
|
||||
routine.id,
|
||||
routine.agentId,
|
||||
@@ -167,6 +168,7 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
routine.nextRunAt ?? null,
|
||||
routine.runCount || 0,
|
||||
JSON.stringify(routine.runHistory || []),
|
||||
routine.scope ?? "project",
|
||||
routine.createdAt,
|
||||
routine.updatedAt,
|
||||
);
|
||||
@@ -257,6 +259,7 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
enabled,
|
||||
runCount: 0,
|
||||
runHistory: [],
|
||||
scope: input.scope ?? "project",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
};
|
||||
@@ -450,8 +453,21 @@ export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
|
||||
|
||||
/**
|
||||
* Get all routines that are due to run (nextRunAt <= now and enabled).
|
||||
* Filters by scope: "global" or "project".
|
||||
*/
|
||||
async getDueRoutines(): Promise<Routine[]> {
|
||||
async getDueRoutines(scope: "global" | "project"): Promise<Routine[]> {
|
||||
const now = new Date().toISOString();
|
||||
const rows = this.db.prepare(
|
||||
"SELECT * FROM routines WHERE enabled = 1 AND nextRunAt IS NOT NULL AND nextRunAt <= ? AND scope = ?"
|
||||
).all(now, scope) as any[];
|
||||
return rows.map((row) => this.rowToRoutine(row));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all routines that are due to run (nextRunAt <= now and enabled) for both scopes.
|
||||
* Returns routines from both "global" and "project" scopes.
|
||||
*/
|
||||
async getDueRoutinesAllScopes(): Promise<Routine[]> {
|
||||
const now = new Date().toISOString();
|
||||
const rows = this.db.prepare(
|
||||
"SELECT * FROM routines WHERE enabled = 1 AND nextRunAt IS NOT NULL AND nextRunAt <= ?"
|
||||
|
||||
@@ -135,6 +135,8 @@ export interface Routine {
|
||||
catchUpLimit?: number;
|
||||
/** Optional cron expression stored directly for due-routine queries (derived from trigger). */
|
||||
cronExpression?: string;
|
||||
/** Scope of this routine: "global" (shared) or "project" (isolated). */
|
||||
scope?: "global" | "project";
|
||||
/** ISO-8601 timestamp of when this routine was created. */
|
||||
createdAt: string;
|
||||
/** ISO-8601 timestamp of when this routine was last updated. */
|
||||
@@ -159,6 +161,8 @@ export interface RoutineCreateInput {
|
||||
executionPolicy?: RoutineExecutionPolicy;
|
||||
/** Whether enabled. Default: true. */
|
||||
enabled?: boolean;
|
||||
/** Scope of this routine: "global" (shared) or "project" (isolated). Default: "project". */
|
||||
scope?: "global" | "project";
|
||||
}
|
||||
|
||||
/** Input for updating an existing routine. */
|
||||
@@ -175,6 +179,8 @@ export interface RoutineUpdateInput {
|
||||
executionPolicy?: RoutineExecutionPolicy;
|
||||
/** Whether enabled. */
|
||||
enabled?: boolean;
|
||||
/** Scope of this routine: "global" (shared) or "project" (isolated). */
|
||||
scope?: "global" | "project";
|
||||
}
|
||||
|
||||
// ── Constants ─────────────────────────────────────────────────────────
|
||||
|
||||
@@ -465,7 +465,7 @@ describe("Run Audit", () => {
|
||||
});
|
||||
|
||||
it("schema version is bumped to 28", () => {
|
||||
expect(db.getSchemaVersion()).toBe(33);
|
||||
expect(db.getSchemaVersion()).toBe(34);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -54,15 +54,37 @@ export interface CronRunnerOptions {
|
||||
schedule: ScheduledTask,
|
||||
result: AutomationRunResult,
|
||||
) => void | Promise<void>;
|
||||
/**
|
||||
* Scope to poll for due schedules.
|
||||
* - "project": Only poll schedules scoped to this project (default)
|
||||
* - "global": Only poll global/shared schedules
|
||||
* - "all": Poll both project and global scopes
|
||||
*
|
||||
* When polling "all", the runner executes schedules from both scopes but
|
||||
* deduplicates by schedule ID to prevent double-execution.
|
||||
*/
|
||||
scope?: "global" | "project" | "all";
|
||||
}
|
||||
|
||||
/**
|
||||
* CronRunner polls the AutomationStore for due schedules and executes them.
|
||||
*
|
||||
* UTILITY PATH: This component runs on the utility lane and does NOT receive the
|
||||
* task-lane semaphore. Automation execution is independent of task execution concurrency.
|
||||
*
|
||||
* - Respects `globalPause` and `enginePaused` settings — skips execution when either is true.
|
||||
* - Prevents concurrent runs of the same schedule.
|
||||
* - Enforces per-schedule timeouts and output size limits.
|
||||
* - Uses a re-entrance guard like Scheduler to prevent overlapping ticks.
|
||||
* - Supports scope-aware polling: "global", "project", or "all" (both scopes).
|
||||
*
|
||||
* SCOPED POLLING SEMANTICS:
|
||||
* - scope="project" (default): Only polls schedules scoped to this project
|
||||
* - scope="global": Only polls global/shared schedules (e.g., memory insight extraction)
|
||||
* - scope="all": Polls both scopes with deterministic de-duplication by schedule ID
|
||||
*
|
||||
* Each ProjectEngine instance runs with scope="project", ensuring project isolation.
|
||||
* A separate global engine (FN-1714) runs with scope="global" for shared schedules.
|
||||
*/
|
||||
export class CronRunner {
|
||||
private running = false;
|
||||
@@ -73,6 +95,8 @@ export class CronRunner {
|
||||
private onScheduleRunProcessed?: CronRunnerOptions["onScheduleRunProcessed"];
|
||||
/** Schedule IDs currently being executed — prevents concurrent runs of the same schedule. */
|
||||
private inFlight = new Set<string>();
|
||||
/** Scope to poll: "global", "project", or "all" (both scopes). */
|
||||
private scope: "global" | "project" | "all";
|
||||
|
||||
constructor(
|
||||
private store: TaskStore,
|
||||
@@ -85,13 +109,14 @@ export class CronRunner {
|
||||
);
|
||||
this.aiPromptExecutor = options.aiPromptExecutor;
|
||||
this.onScheduleRunProcessed = options.onScheduleRunProcessed;
|
||||
this.scope = options.scope ?? "project";
|
||||
}
|
||||
|
||||
/** Start the polling loop. */
|
||||
start(): void {
|
||||
if (this.running) return;
|
||||
this.running = true;
|
||||
log.log(`Started (poll every ${this.pollIntervalMs / 1000}s)`);
|
||||
log.log(`Started (poll every ${this.pollIntervalMs / 1000}s, scope: ${this.scope})`);
|
||||
|
||||
// Run first tick immediately
|
||||
void this.tick();
|
||||
@@ -127,9 +152,20 @@ export class CronRunner {
|
||||
return;
|
||||
}
|
||||
|
||||
const dueSchedules = await this.automationStore.getDueSchedules();
|
||||
// Get due schedules based on configured scope
|
||||
let dueSchedules: ScheduledTask[];
|
||||
if (this.scope === "all") {
|
||||
// Poll both scopes and deduplicate by ID
|
||||
dueSchedules = await this.automationStore.getDueSchedulesAllScopes();
|
||||
} else {
|
||||
dueSchedules = await this.automationStore.getDueSchedules(this.scope);
|
||||
}
|
||||
|
||||
if (dueSchedules.length === 0) return;
|
||||
|
||||
// Track executed schedule IDs to prevent double-execution when polling all scopes
|
||||
const executedIds = new Set<string>();
|
||||
|
||||
for (const schedule of dueSchedules) {
|
||||
// Skip if already in-flight (prevents concurrent runs of same schedule)
|
||||
if (this.inFlight.has(schedule.id)) {
|
||||
@@ -137,6 +173,22 @@ export class CronRunner {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Skip if already executed this tick (de-duplication across scopes)
|
||||
if (executedIds.has(schedule.id)) {
|
||||
log.log(`Skipping ${schedule.name} (${schedule.id}) — already executed from another scope this tick`);
|
||||
continue;
|
||||
}
|
||||
executedIds.add(schedule.id);
|
||||
|
||||
// Log which scope this schedule is from
|
||||
const scheduleScope = schedule.scope ?? "project";
|
||||
if (scheduleScope !== this.scope && this.scope !== "all") {
|
||||
log.log(`Skipping ${schedule.name} (${schedule.id}) — belongs to ${scheduleScope} scope, not polling`);
|
||||
continue;
|
||||
}
|
||||
|
||||
log.log(`Executing ${schedule.name} (${schedule.id}) [scope: ${scheduleScope}]`);
|
||||
|
||||
// Re-check pause on each schedule (may have changed mid-loop)
|
||||
const currentSettings = await this.store.getSettings();
|
||||
if (currentSettings.globalPause || currentSettings.enginePaused) {
|
||||
|
||||
@@ -165,6 +165,7 @@ export class ProjectEngine {
|
||||
this.cronRunner = new CronRunner(store, this.automationStore, {
|
||||
aiPromptExecutor,
|
||||
onScheduleRunProcessed: this.buildInsightRunHandler(cwd),
|
||||
scope: "project", // Project-scoped execution — global schedules run separately
|
||||
});
|
||||
|
||||
// Sync insight extraction automation on startup
|
||||
|
||||
@@ -1,12 +1,23 @@
|
||||
/**
|
||||
* RoutineScheduler — polls for due routines and triggers their execution via RoutineRunner.
|
||||
*
|
||||
* UTILITY PATH: This component runs on the utility lane and does NOT receive the
|
||||
* task-lane semaphore. Routine execution is independent of task execution concurrency.
|
||||
*
|
||||
* 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
|
||||
* - Scope-aware polling: "global", "project", or "all" (both scopes)
|
||||
*
|
||||
* SCOPED POLLING SEMANTICS:
|
||||
* - scope="project" (default): Only polls routines scoped to this project
|
||||
* - scope="global": Only polls global/shared routines
|
||||
* - scope="all": Polls both scopes with deterministic de-duplication by routine ID
|
||||
*
|
||||
* Each ProjectEngine instance runs with scope="project", ensuring project isolation.
|
||||
*/
|
||||
|
||||
import { CronExpressionParser } from "cron-parser";
|
||||
@@ -28,6 +39,13 @@ export interface RoutineSchedulerOptions {
|
||||
routineRunner: RoutineRunner;
|
||||
/** Polling interval in milliseconds. Default: 60000 (60s). Minimum: 10000 (10s). */
|
||||
pollIntervalMs?: number;
|
||||
/**
|
||||
* Scope to poll for due routines.
|
||||
* - "project": Only poll routines scoped to this project (default)
|
||||
* - "global": Only poll global/shared routines
|
||||
* - "all": Poll both project and global scopes
|
||||
*/
|
||||
scope?: "global" | "project" | "all";
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -38,6 +56,8 @@ export class RoutineScheduler {
|
||||
private routineStore: RoutineStore;
|
||||
private routineRunner: RoutineRunner;
|
||||
private pollIntervalMs: number;
|
||||
/** Scope to poll: "global", "project", or "all". */
|
||||
private scope: "global" | "project" | "all";
|
||||
|
||||
private running: boolean = false;
|
||||
private ticking: boolean = false;
|
||||
@@ -48,6 +68,7 @@ export class RoutineScheduler {
|
||||
this.routineStore = options.routineStore;
|
||||
this.routineRunner = options.routineRunner;
|
||||
this.pollIntervalMs = Math.max(10000, options.pollIntervalMs ?? 60000);
|
||||
this.scope = options.scope ?? "project";
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -60,7 +81,7 @@ export class RoutineScheduler {
|
||||
}
|
||||
|
||||
this.running = true;
|
||||
logger.log(`RoutineScheduler started with ${this.pollIntervalMs}ms poll interval`);
|
||||
logger.log(`RoutineScheduler started with ${this.pollIntervalMs}ms poll interval (scope: ${this.scope})`);
|
||||
|
||||
// Run first tick immediately
|
||||
void this.tick();
|
||||
@@ -119,16 +140,42 @@ export class RoutineScheduler {
|
||||
return;
|
||||
}
|
||||
|
||||
// Get due routines
|
||||
const dueRoutines = await this.getDueRoutines();
|
||||
// Get due routines based on configured scope
|
||||
let dueRoutines: Routine[];
|
||||
if (this.scope === "all") {
|
||||
// Poll both scopes and deduplicate by ID
|
||||
dueRoutines = await this.routineStore.getDueRoutinesAllScopes();
|
||||
} else {
|
||||
dueRoutines = await this.routineStore.getDueRoutines(this.scope);
|
||||
}
|
||||
|
||||
if (dueRoutines.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
logger.log(`Found ${dueRoutines.length} due routines`);
|
||||
logger.log(`Found ${dueRoutines.length} due routines (scope: ${this.scope})`);
|
||||
|
||||
// Track executed routine IDs to prevent double-execution when polling all scopes
|
||||
const executedIds = new Set<string>();
|
||||
|
||||
// Process each routine
|
||||
for (const routine of dueRoutines) {
|
||||
// Skip if already executed this tick (de-duplication across scopes)
|
||||
if (executedIds.has(routine.id)) {
|
||||
logger.log(`[${routine.id}] Skipped: already executed from another scope this tick`);
|
||||
continue;
|
||||
}
|
||||
executedIds.add(routine.id);
|
||||
|
||||
// Log which scope this routine is from
|
||||
const routineScope = routine.scope ?? "project";
|
||||
if (routineScope !== this.scope && this.scope !== "all") {
|
||||
logger.log(`[${routine.id}] Skipped: belongs to ${routineScope} scope, not polling`);
|
||||
continue;
|
||||
}
|
||||
|
||||
logger.log(`[${routine.id}] Processing [scope: ${routineScope}]`);
|
||||
|
||||
// Re-check pause state (may have changed mid-loop)
|
||||
const currentSettings = await this.taskStore.getSettings();
|
||||
if (currentSettings.globalPause || currentSettings.enginePaused) {
|
||||
@@ -178,18 +225,6 @@ export class RoutineScheduler {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get routines that are due for execution.
|
||||
*/
|
||||
private async getDueRoutines(): Promise<Routine[]> {
|
||||
try {
|
||||
return await this.routineStore.getDueRoutines();
|
||||
} catch (err) {
|
||||
logger.error(`Failed to get due routines: ${err}`);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger a routine manually via the API.
|
||||
*/
|
||||
|
||||
@@ -463,6 +463,7 @@ export class InProcessRuntime
|
||||
routineStore,
|
||||
routineRunner: this.routineRunner,
|
||||
pollIntervalMs: 60000,
|
||||
scope: "project", // Project-scoped execution — global routines run separately
|
||||
});
|
||||
this.routineScheduler.start();
|
||||
runtimeLog.log("RoutineScheduler initialized and started");
|
||||
|
||||
Reference in New Issue
Block a user