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

Address PR review feedback (#1363)

Browse Source 
Greptile + CodeRabbit findings across core/engine/dashboard. Stale findings
(written against earlier commits) verified and skipped; valid ones fixed.

Engine:
- await-input: do not clear pausedReason in the /input route (the node's
  marker must survive unpause); the node clears it after consuming input.
  Embed a colon-free epoch watermark in the marker so only post-pause steering
  comments count as the reply (ISO timestamps collided with the colon
  separator and the dashboard question parser).
- gate nodes without a registered runner now fail closed (throw) instead of
  silently passing.
- a thrown interpreter error in maybeExecuteWorkflowGraph now falls back to the
  legacy pipeline instead of stranding the task in-progress.
- approved-CLI path clears the stale awaiting-cli-approval status/marker.

Core:
- persist+cascade workflow selection: purge task_workflow_selection rows and
  compiled workflow_steps on physical task deletes; migration 105 cleans
  already-orphaned rows; catch-cleanup for materialized steps when the owner
  write fails; WF-id allocation now in a BEGIN IMMEDIATE transaction.
- compiler validates the canonical execute->review->merge seam order (rejects
  duplicate/misordered seams).
- disk-backed reopen round-trip + tightened updatedAt/list assertions.

Dashboard:
- WorkflowSelector clears stale default/options across project changes and on
  fetch failure; InlineCreateCard/NewTaskModal reset the workflow on all
  clear/discard paths and include it in dirty-state.
- WorkflowNodeEditor: config-key deletion now persists; removed an invalid
  eslint-disable that was itself a hard lint error.
- TaskCard: single status badge for awaiting-input (no duplicate).
- WorkflowResultsTab: reset paused-action UI between pauses; surface
  resume/approve failures inline.
- TaskDetailModal: treat awaiting-user-input/awaiting-cli-approval/paused as
  not-in-progress for the live-log subscription.
- workflow-flow-mapping: don't write synthetic node names back into IR.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-06-03 17:32:12 -07:00
parent 46f525bd28
commit eb67d08213
22 changed files with 587 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
packages/core/src/__tests__/store-persistence.test.ts
View File

@@ -428,5 +428,26 @@ describe("TaskStore", () => {
const listed = tasks.find((t) => t.id === paused.id);
expect(listed?.pausedReason).toBe("worktrunk_operation_failed");
});
it("survives a disk-backed store reload", async () => {
// The original bug was "pause state vanished on reload" — an in-memory
// cache could mask a missing persist column, so close and reopen the
// store from disk before asserting.
harness.store().close();
await harness.reopenDiskBackedStore();
const task = await harness.store().createTask({ description: "Pause across reload" });
await harness.store().updateTask(task.id, {
paused: true,
pausedReason: "workflow-input:ask: What environment should this deploy to?",
});
harness.store().close();
await harness.reopenDiskBackedStore();
const reloaded = await harness.store().getTask(task.id);
expect(reloaded.paused).toBe(true);
expect(reloaded.pausedReason).toBe("workflow-input:ask: What environment should this deploy to?");
});
});
});

42
packages/core/src/__tests__/workflow-compiler.test.ts
View File

@@ -145,6 +145,48 @@ describe("compileWorkflowToSteps (U2)", () => {
expect(err).toBeInstanceOf(WorkflowCompileError);
});
it("rejects seams that are out of the execute -> review -> merge order", () => {
const ir: WorkflowIr = {
version: "v1",
name: "misordered-seams",
nodes: [
{ id: "start", kind: "start" },
{ id: "merge", kind: "prompt", config: { seam: "merge" } },
{ id: "review", kind: "prompt", config: { seam: "review" } },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "merge", condition: "success" },
{ from: "merge", to: "review", condition: "success" },
{ from: "review", to: "end", condition: "success" },
],
};
const err = validateLinearity(ir);
expect(err).toBeInstanceOf(WorkflowCompileError);
expect(err?.message).toMatch(/execute -> review -> merge order/);
});
it("rejects a graph with a duplicated seam role", () => {
const ir: WorkflowIr = {
version: "v1",
name: "dup-merge",
nodes: [
{ id: "start", kind: "start" },
{ id: "merge1", kind: "prompt", config: { seam: "merge" } },
{ id: "merge2", kind: "prompt", config: { seam: "merge" } },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "merge1", condition: "success" },
{ from: "merge1", to: "merge2", condition: "success" },
{ from: "merge2", to: "end", condition: "success" },
],
};
const err = validateLinearity(ir);
expect(err).toBeInstanceOf(WorkflowCompileError);
expect(err?.message).toMatch(/appears more than once/);
});
it("returns an empty step set for a graph with only start/seams/end", () => {
const ir = graph([]);
expect(compileWorkflowToSteps(ir)).toEqual([]);

18
packages/core/src/__tests__/workflow-definition-store.test.ts
View File

@@ -1,6 +1,7 @@
import { describe, it, expect, beforeEach, afterEach } from "vitest";
import { WorkflowIrError } from "../workflow-ir.js";
import { isBuiltinWorkflowId } from "../builtin-workflows.js";
import type { WorkflowIr } from "../workflow-ir-types.js";
import { createTaskStoreTestHarness } from "./store-test-helpers.js";
@@ -43,11 +44,12 @@ describe("TaskStore workflow definitions (U1)", () => {
});
expect(created.id).toBe("WF-001");
const list = await store.listWorkflowDefinitions();
expect(list).toHaveLength(1);
expect(list[0].name).toBe("Quality Gate");
expect(list[0].ir.nodes).toHaveLength(3);
expect(list[0].layout.lint).toEqual({ x: 120, y: 0 });
// The list prepends read-only built-ins; assert on the user workflows only.
const userList = (await store.listWorkflowDefinitions()).filter((w) => !isBuiltinWorkflowId(w.id));
expect(userList).toHaveLength(1);
expect(userList[0].name).toBe("Quality Gate");
expect(userList[0].ir.nodes).toHaveLength(3);
expect(userList[0].layout.lint).toEqual({ x: 120, y: 0 });
});
it("rejects a workflow whose IR is missing start/end", async () => {
@@ -55,7 +57,7 @@ describe("TaskStore workflow definitions (U1)", () => {
await expect(
store.createWorkflowDefinition({ name: "Broken", ir: bad }),
).rejects.toBeInstanceOf(WorkflowIrError);
expect(await store.listWorkflowDefinitions()).toHaveLength(0);
expect((await store.listWorkflowDefinitions()).filter((w) => !isBuiltinWorkflowId(w.id))).toHaveLength(0);
});
it("requires a non-empty name", async () => {
@@ -88,7 +90,7 @@ describe("TaskStore workflow definitions (U1)", () => {
expect(updated.description).toBe("now with a prompt step");
expect(updated.ir.nodes.some((n) => n.id === "review")).toBe(true);
expect(updated.layout.start).toEqual({ x: 5, y: 5 });
expect(new Date(updated.updatedAt).getTime()).toBeGreaterThanOrEqual(
expect(new Date(updated.updatedAt).getTime()).toBeGreaterThan(
new Date(created.updatedAt).getTime(),
);
});
@@ -108,7 +110,7 @@ describe("TaskStore workflow definitions (U1)", () => {
const created = await store.createWorkflowDefinition({ name: "Temp", ir: makeIr() });
await store.deleteWorkflowDefinition(created.id);
expect(await store.getWorkflowDefinition(created.id)).toBeUndefined();
expect(await store.listWorkflowDefinitions()).toHaveLength(0);
expect((await store.listWorkflowDefinitions()).filter((w) => !isBuiltinWorkflowId(w.id))).toHaveLength(0);
});
it("throws when deleting a non-existent workflow", async () => {

21
packages/core/src/__tests__/workflow-selection-store.test.ts
View File

@@ -118,6 +118,27 @@ describe("TaskStore workflow selection (U3)", () => {
expect(detail.enabledWorkflowSteps ?? []).toHaveLength(0);
});
it("force-resurrecting over a tombstoned task purges its prior workflow selection", async () => {
const wf = await store.createWorkflowDefinition({ name: "QA", ir: linearIr() });
const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] });
await store.selectTaskWorkflow(task.id, wf.id);
const priorIds = store.getTaskWorkflowSelection(task.id)!.stepIds;
expect(priorIds).toHaveLength(2);
// Soft-delete then physically resurrect the same id; the physical purge of
// the old tasks row must drop the orphaned selection + its compiled steps.
await store.deleteTask(task.id);
await store.createTaskWithReservedId(
{ description: "resurrected", enabledWorkflowSteps: [], forceResurrect: true },
{ taskId: task.id, applyDefaultWorkflowSteps: false },
);
expect(store.getTaskWorkflowSelection(task.id)).toBeUndefined();
for (const id of priorIds) {
expect(await store.getWorkflowStep(id)).toBeUndefined();
}
});
it("throws when selecting an unknown workflow", async () => {
const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] });
await expect(store.selectTaskWorkflow(task.id, "WF-404")).rejects.toThrow(/not found/i);

25
packages/core/src/db.ts
View File

@@ -149,7 +149,7 @@ export function probeFts5(db: DatabaseSync): boolean {
// ── Schema Definition ────────────────────────────────────────────────
const SCHEMA_VERSION = 104;
const SCHEMA_VERSION = 105;
export { SCHEMA_VERSION };
@@ -4092,6 +4092,29 @@ export class Database {
});
}
// Migration 105: task_workflow_selection has no FK to tasks(id) (SQLite can't
// add one to an existing table without a rebuild), so physical task deletes
// before this version could leave orphaned selection rows and unreclaimable
// compiled workflow_steps. Drop any already-orphaned rows and their steps.
if (version < 105) {
this.applyMigration(105, () => {
// Delete the compiled steps referenced by orphaned selections first, then
// the orphaned selection rows themselves. json_each expands the stepIds
// JSON array; the WHERE guards against malformed (non-array) stepIds.
this.db.exec(`
DELETE FROM workflow_steps WHERE id IN (
SELECT je.value
FROM task_workflow_selection sel
JOIN json_each(sel.stepIds) je
WHERE json_valid(sel.stepIds)
AND sel.taskId NOT IN (SELECT id FROM tasks)
);
DELETE FROM task_workflow_selection
WHERE taskId NOT IN (SELECT id FROM tasks);
`);
});
}
}
/**

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

@@ -2529,6 +2529,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
const allowResurrection = existing.allowResurrection === true;
if (input.forceResurrect === true || allowResurrection) {
this.purgeTaskWorkflowSelectionRows(id);
this.db.prepare("DELETE FROM tasks WHERE id = ?").run(id);
this.db.bumpLastModified();
return;
@@ -3779,18 +3780,26 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
resolvedWorkflowSteps = undefined;
}
const task = await this.createTaskWithDistributedReservation(input, {
createTaskWithId: async (taskId) => {
await this.assertNoDependencyCycle(taskId, input.dependencies ?? [], "createTask");
return this._createTaskInternal(
input,
title,
resolvedWorkflowSteps,
taskId,
{ invokeTaskCreatedHook: shouldInvokeTaskCreatedHook && !hasPendingSummarization },
);
},
});
let task: Task;
try {
task = await this.createTaskWithDistributedReservation(input, {
createTaskWithId: async (taskId) => {
await this.assertNoDependencyCycle(taskId, input.dependencies ?? [], "createTask");
return this._createTaskInternal(
input,
title,
resolvedWorkflowSteps,
taskId,
{ invokeTaskCreatedHook: shouldInvokeTaskCreatedHook && !hasPendingSummarization },
);
},
});
} catch (err) {
// The task row was never created, so any default-workflow steps we
// materialized above would orphan with no task/selection pointing at them.
this.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds);
throw err;
}
// Record the inherited workflow selection now that the task row exists.
if (pendingWorkflowSelection) {
@@ -3942,12 +3951,20 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
resolvedWorkflowSteps = undefined;
}
const createdTask = await this._createTaskInternal(input, title, resolvedWorkflowSteps, id, {
createdAt: options.createdAt,
updatedAt: options.updatedAt,
promptOverride: options.prompt,
invokeTaskCreatedHook: options.invokeTaskCreatedHook,
});
let createdTask: Task;
try {
createdTask = await this._createTaskInternal(input, title, resolvedWorkflowSteps, id, {
createdAt: options.createdAt,
updatedAt: options.updatedAt,
promptOverride: options.prompt,
invokeTaskCreatedHook: options.invokeTaskCreatedHook,
});
} catch (err) {
// The task row was never created, so any default-workflow steps we
// materialized above would orphan with no task/selection pointing at them.
this.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds);
throw err;
}
// Record the inherited workflow selection now that the task row exists.
if (pendingWorkflowSelection) {
@@ -7827,6 +7844,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
private deleteTaskById(taskId: string): void {
this.clearLinkedAgentTaskIds(taskId);
this.purgeTaskWorkflowSelectionRows(taskId);
this.db.prepare('DELETE FROM tasks WHERE id = ?').run(taskId);
this.db.bumpLastModified();
}
@@ -8454,6 +8472,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
this.db.transaction(() => {
rewrittenLineageChildren = this.rewriteLineageChildrenForRemoval(id, lineageChildIds);
this.clearLinkedAgentTaskIds(id, task.updatedAt);
this.purgeTaskWorkflowSelectionRows(id);
this.db.prepare('DELETE FROM tasks WHERE id = ?').run(id);
this.db.bumpLastModified();
});
@@ -10472,6 +10491,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
const archivedAt = task.columnMovedAt ?? task.updatedAt ?? new Date().toISOString();
const entry = await this.taskToArchiveEntry(task, archivedAt);
this.archiveDb.upsert(entry);
this.purgeTaskWorkflowSelectionRows(task.id);
this.db.prepare("DELETE FROM tasks WHERE id = ?").run(task.id);
await rm(this.taskDir(task.id), { recursive: true, force: true });
if (this.isWatching) {
@@ -10507,6 +10527,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
this.archiveDb.upsert(entry);
// Remove task from tasks table
this.purgeTaskWorkflowSelectionRows(task.id);
this.db.prepare('DELETE FROM tasks WHERE id = ?').run(task.id);
this.db.bumpLastModified();
@@ -11018,16 +11039,21 @@ ${stepsSection}`;
/** Allocate the next workflow-definition id (WF-001, WF-002, …) using a
* monotonic counter persisted in __meta. Never reuses ids across deletes. */
private nextWorkflowDefinitionId(): string {
const row = this.db.prepare("SELECT value FROM __meta WHERE key = 'nextWorkflowDefinitionId'").get() as
| { value: string }
| undefined;
const next = row ? parseInt(row.value, 10) || 1 : 1;
this.db
.prepare(
"INSERT INTO __meta (key, value) VALUES ('nextWorkflowDefinitionId', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
)
.run(String(next + 1));
return `WF-${String(next).padStart(3, "0")}`;
// Serialize the read+increment in one write transaction so two TaskStore
// instances cannot both observe the same counter and allocate the same
// WF-id (which would collide on the workflows primary key).
return this.db.transactionImmediate(() => {
const row = this.db.prepare("SELECT value FROM __meta WHERE key = 'nextWorkflowDefinitionId'").get() as
| { value: string }
| undefined;
const next = row ? parseInt(row.value, 10) || 1 : 1;
this.db
.prepare(
"INSERT INTO __meta (key, value) VALUES ('nextWorkflowDefinitionId', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
)
.run(String(next + 1));
return `WF-${String(next).padStart(3, "0")}`;
});
}
private toWorkflowDefinition(row: {
@@ -11322,6 +11348,49 @@ ${stepsSection}`;
this.db.prepare("DELETE FROM task_workflow_selection WHERE taskId = ?").run(taskId);
}
/** Purge a task's workflow selection and its materialized WorkflowStep rows
* when the task row itself is being physically removed. `task_workflow_selection`
* has no FK to `tasks(id)` (SQLite can't add one to an existing table without a
* rebuild), so deletion must be mirrored here to avoid orphaned selection rows
* and unreclaimable compiled steps. Best-effort and synchronous: unlike
* clearTaskWorkflowSelection it does not touch enabledWorkflowSteps, since the
* owning task row no longer exists. */
private purgeTaskWorkflowSelectionRows(taskId: string): void {
const row = this.db
.prepare("SELECT stepIds FROM task_workflow_selection WHERE taskId = ?")
.get(taskId) as { stepIds: string } | undefined;
if (!row) return;
try {
const parsed = JSON.parse(row.stepIds) as unknown;
if (Array.isArray(parsed)) {
for (const stepId of parsed) {
if (typeof stepId === "string") {
this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId);
}
}
}
} catch {
// Corrupt stepIds list — still remove the selection row below.
}
this.db.prepare("DELETE FROM task_workflow_selection WHERE taskId = ?").run(taskId);
this.workflowStepsCache = null;
}
/** Delete a set of freshly materialized WorkflowStep rows that were never
* successfully attached to a task/selection (e.g. the owning task create
* failed). Best-effort; tolerates already-removed ids. */
private cleanupOrphanedMaterializedSteps(stepIds: string[] | undefined): void {
if (!stepIds || stepIds.length === 0) return;
for (const stepId of stepIds) {
try {
this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId);
} catch {
// Best-effort cleanup.
}
}
this.workflowStepsCache = null;
}
/** Persist pre-compiled workflow steps as fresh WorkflowStep rows and return
* their ids in execution order. Steps are tagged so they stay out of the
* step manager. Compile via compileWorkflowToSteps before calling. */
@@ -11372,8 +11441,23 @@ ${stepsSection}`;
// referencing already-deleted step ids.
const priorSelection = this.getTaskWorkflowSelection(taskId);
const ids = await this.materializeWorkflowSteps(workflowId, inputs);
await this.updateTask(taskId, { enabledWorkflowSteps: ids });
this.writeTaskWorkflowSelection(taskId, workflowId, ids);
try {
await this.updateTask(taskId, { enabledWorkflowSteps: ids });
this.writeTaskWorkflowSelection(taskId, workflowId, ids);
} catch (err) {
// The owner write (updateTask / selection upsert) failed, so the steps we
// just materialized would orphan with no selection row pointing at them.
// Delete them before propagating; the prior selection is left untouched.
for (const stepId of ids) {
try {
this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId);
} catch {
// Best-effort cleanup; surface the original error below.
}
}
this.workflowStepsCache = null;
throw err;
}
if (priorSelection) {
for (const stepId of priorSelection.stepIds) {

26
packages/core/src/workflow-compiler.ts
View File

@@ -102,10 +102,36 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null {
}
// Reachability: the single main path must reach end and cover every node.
// While walking, enforce the canonical seam pipeline: each of execute/review/
// merge may appear at most once and only in that order. The compiler treats
// seams as a fixed execute → review → merge boundary (merge flips pre- to
// post-merge), so out-of-order or duplicate seams would compile inconsistently
// with the runtime contract.
const expectedSeamOrder = ["execute", "review", "merge"] as const;
const seenSeams = new Set<string>();
let nextExpectedSeamIndex = 0;
const visited = new Set<string>();
let cursor: string | undefined = startNode.id;
while (cursor && !visited.has(cursor)) {
visited.add(cursor);
const node = nodesById.get(cursor);
const seam = node ? seamOf(node) : undefined;
if (seam) {
if (seenSeams.has(seam)) {
return new WorkflowCompileError(`seam '${seam}' appears more than once`);
}
while (
nextExpectedSeamIndex < expectedSeamOrder.length &&
expectedSeamOrder[nextExpectedSeamIndex] !== seam
) {
nextExpectedSeamIndex += 1;
}
if (expectedSeamOrder[nextExpectedSeamIndex] !== seam) {
return new WorkflowCompileError("seams must follow the execute -> review -> merge order");
}
seenSeams.add(seam);
nextExpectedSeamIndex += 1;
}
if (cursor === endNode.id) break;
cursor = mainEdge(outgoing.get(cursor) ?? [])?.to;
}