root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(review): address workflow work item feedback

Browse Source
This commit is contained in:
gsxdsm
2026-06-09 13:37:46 -07:00
parent c11f1700b7
commit cdd86fc3f6
5 changed files with 41 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
packages/core/src/__tests__/db-migrate.test.ts
View File

@@ -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 }>;

20
packages/core/src/__tests__/db.test.ts
View File

@@ -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 }>;

2
packages/core/src/__tests__/run-audit.test.ts
View File

@@ -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);
});
});

27
packages/core/src/__tests__/store-workflow-runtime.test.ts
View File

@@ -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({

15
packages/core/src/store.ts
View File

@@ -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();