diff --git a/packages/core/src/__tests__/db-migrate.test.ts b/packages/core/src/__tests__/db-migrate.test.ts index 3b06532f27..b2f321eae4 100644 --- a/packages/core/src/__tests__/db-migrate.test.ts +++ b/packages/core/src/__tests__/db-migrate.test.ts @@ -716,7 +716,6 @@ describe("schema migration", () => { const row = db.prepare("SELECT deletedAt FROM tasks WHERE id = 'FN-legacy'").get() as { deletedAt: string | null }; expect(row.deletedAt).toBeNull(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -750,7 +749,6 @@ describe("schema migration", () => { { id: "WS-002", mode: "script", gateMode: "advisory" }, ]); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -801,7 +799,6 @@ describe("schema migration", () => { reviewerFallbackRetryCount: 0, }); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -831,7 +828,6 @@ describe("schema migration", () => { const columns = db.prepare("PRAGMA table_info(milestones)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("acceptanceCriteria"); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -872,7 +868,6 @@ describe("schema migration", () => { const missionColumns = db.prepare("PRAGMA table_info(missions)").all() as Array<{ name: string }>; expect(missionColumns.map((column) => column.name)).toContain("autoMerge"); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -908,7 +903,6 @@ describe("schema migration", () => { { id: "WS-003", mode: "prompt", enabled: 0, gateMode: "advisory" }, ]); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -946,7 +940,6 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(mission_goals)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idxMissionGoalsGoalId")).toBe(true); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1226,7 +1219,6 @@ describe("schema migration", () => { .get() as { migrated_fragment_id: string | null }; expect(stepRow.migrated_fragment_id).toBeNull(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1235,14 +1227,12 @@ describe("schema migration", () => { const db = new Database(fusionDir); db.init(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); // Re-open the same on-disk DB: already at 109, the 109 block must be a no-op. const reopened = new Database(fusionDir); reopened.init(); expect(reopened.getSchemaVersion()).toBe(115); - expect(reopened.getSchemaVersion()).toBe(115); const workflowColumns = reopened.prepare("PRAGMA table_info(workflows)").all() as Array<{ name: string }>; expect(workflowColumns.filter((c) => c.name === "kind")).toHaveLength(1); const stepColumns = reopened.prepare("PRAGMA table_info(workflow_steps)").all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index a4f89a3caa..a3fa334ad3 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -335,7 +335,6 @@ describe("Database", () => { it("seeds schema version", () => { expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); }); it("includes tokenUsageCacheWriteTokens on freshly initialized tasks table", () => { @@ -395,7 +394,6 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); }); it("does not overwrite existing config on re-init", () => { // Update the config @@ -1466,7 +1464,6 @@ describe("schema migrations", () => { // Verify version bumped to 29 (includes v1→v2 through v26→v29) expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1496,7 +1493,6 @@ describe("schema migrations", () => { // Re-init should not fail db.init(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); // Re-init should not fail db.init(); @@ -1535,7 +1531,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1577,7 +1572,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1650,7 +1644,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1891,7 +1884,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>; @@ -1966,7 +1958,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>; @@ -1991,7 +1982,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>; @@ -2097,7 +2087,6 @@ describe("schema migrations", () => { // Verify version bumped to 29 expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -2316,7 +2305,6 @@ describe("schema migrations", () => { localDb.init(); - expect(localDb.getSchemaVersion()).toBe(115); expect(localDb.getSchemaVersion()).toBe(115); const columns = localDb.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("tokenUsageCacheWriteTokens"); @@ -2628,7 +2616,6 @@ describe("createDatabase factory", () => { const db = createDatabase(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); expect(db.getLastModified()).toBeGreaterThan(0); @@ -2784,7 +2771,6 @@ describe("migration v77 task token budget columns", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const rows = migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const names = new Set(rows.map((row) => row.name)); expect(names.has("tokenBudgetSoftAlertedAt")).toBe(true); @@ -2816,7 +2802,6 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { try { fresh.init(); expect(fresh.getSchemaVersion()).toBe(115); - expect(fresh.getSchemaVersion()).toBe(115); const names = new Set( (fresh.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2845,7 +2830,6 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const names = new Set( (migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2872,7 +2856,6 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { try { fresh.init(); expect(fresh.getSchemaVersion()).toBe(115); - expect(fresh.getSchemaVersion()).toBe(115); const table = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2907,7 +2890,6 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const table = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2949,7 +2931,6 @@ describe("migration v67 drops orphan project auth tables", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const tables = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; @@ -2977,7 +2958,6 @@ describe("migration v67 drops orphan project auth tables", () => { try { fresh.init(); expect(fresh.getSchemaVersion()).toBe(115); - expect(fresh.getSchemaVersion()).toBe(115); const tables = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/run-audit.test.ts b/packages/core/src/__tests__/run-audit.test.ts index 087f71d8fc..6e15da28c5 100644 --- a/packages/core/src/__tests__/run-audit.test.ts +++ b/packages/core/src/__tests__/run-audit.test.ts @@ -583,7 +583,7 @@ describe("Run Audit", () => { expect(indexNames).toContain("idxRunAuditEventsTimestamp"); }); - it("schema version is bumped to 40", () => { + it("schema version is bumped to 115", () => { expect(db.getSchemaVersion()).toBe(115); }); }); diff --git a/packages/core/src/__tests__/store-workflow-runtime.test.ts b/packages/core/src/__tests__/store-workflow-runtime.test.ts index 0bee6ddf5f..0849cd6955 100644 --- a/packages/core/src/__tests__/store-workflow-runtime.test.ts +++ b/packages/core/src/__tests__/store-workflow-runtime.test.ts @@ -173,6 +173,33 @@ describe("TaskStore workflow work items", () => { }); }); + it("honors due-list state filters and validates lease duration", async () => { + const taskId = await createTaskId(); + const item = store.upsertWorkflowWorkItem({ + runId: "run-filter", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "runnable", + now: "2026-06-09T00:00:00.000Z", + }); + store.acquireWorkflowWorkItemLease(item.id, "worker-a", { + now: "2026-06-09T00:00:00.000Z", + leaseDurationMs: 60_000, + }); + + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:01:00.000Z", states: ["runnable"] })).toEqual([]); + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:01:00.000Z", states: ["running"] }).map((due) => due.id)).toEqual([ + item.id, + ]); + expect(() => + store.acquireWorkflowWorkItemLease(item.id, "worker-b", { + now: "2026-06-09T00:01:00.000Z", + leaseDurationMs: 0, + }), + ).toThrow("workflow work item leaseDurationMs must be > 0 (received 0)"); + }); + it("preserves lease and retry metadata on idempotent duplicate upserts", async () => { const taskId = await createTaskId(); const item = store.upsertWorkflowWorkItem({ diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index a7556d52b4..849398ab1b 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -9075,12 +9075,19 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} listDueWorkflowWorkItems(filter: WorkflowWorkItemDueFilter = {}): WorkflowWorkItem[] { const now = filter.now ?? new Date().toISOString(); + const includeExpiredRunning = !filter.states || filter.states.includes("running"); const states = filter.states?.length ? filter.states : ["runnable", "retrying"]; + const stateConditions = [`(state IN (${states.map(() => "?").join(", ")}) AND (leaseExpiresAt IS NULL OR leaseExpiresAt <= ?))`]; + const params: unknown[] = [...states, now]; + if (includeExpiredRunning) { + stateConditions.push("(state = 'running' AND leaseExpiresAt IS NOT NULL AND leaseExpiresAt <= ?)"); + params.push(now); + } const conditions = [ - `((state IN (${states.map(() => "?").join(", ")}) AND (leaseExpiresAt IS NULL OR leaseExpiresAt <= ?)) OR (state = 'running' AND leaseExpiresAt IS NOT NULL AND leaseExpiresAt <= ?))`, + `(${stateConditions.join(" OR ")})`, "(retryAfter IS NULL OR retryAfter <= ?)", ]; - const params: unknown[] = [...states, now, now, now]; + params.push(now); if (filter.kinds?.length) { conditions.push(`kind IN (${filter.kinds.map(() => "?").join(", ")})`); params.push(...filter.kinds); @@ -9104,6 +9111,10 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} leaseOwner: string, opts: { leaseDurationMs: number; now?: string }, ): WorkflowWorkItem | null { + if (opts.leaseDurationMs <= 0) { + throw new Error(`workflow work item leaseDurationMs must be > 0 (received ${opts.leaseDurationMs})`); + } + return this.db.transactionImmediate(() => { const now = opts.now ?? new Date().toISOString(); const leaseExpiresAt = new Date(new Date(now).getTime() + opts.leaseDurationMs).toISOString();