Merge branch 'feat/reactive-node-status'

src/reactive/index.ts

@@ -10,3 +10,12 @@ export {
   type EventLogProjection,
   type AggregateStatus,
 } from "./workflow.js";
+
+export {
+  computePreconditions,
+  computeBlockedByFailure,
+  registerStartEffect,
+  registerAbortEffect,
+  type NodeStatusContext,
+  type AbortEffectOptions,
+} from "./node-status.js";

src/reactive/node-status.ts

@@ -1 +1,81 @@
-export {};
+import { effect } from "@preact/signals-core";
+import type { Signal, ReadonlySignal } from "@preact/signals-core";
+import type { NodeStatus } from "../schema/enums.js";
+
+const TERMINAL_STATUSES: Set<NodeStatus> = new Set([
+  "completed",
+  "failed",
+  "aborted",
+  "skipped",
+]);
+
+export interface NodeStatusContext {
+  statusMap: Map<string, Signal<NodeStatus>>;
+  predecessors: string[];
+}
+
+export function computePreconditions(
+  _nodeKey: string,
+  ctx: NodeStatusContext,
+): boolean {
+  if (ctx.predecessors.length === 0) return true;
+  return ctx.predecessors.every((pred: string) => {
+    const predStatus = ctx.statusMap.get(pred);
+    if (!predStatus) return false;
+    return predStatus.value === "completed" || predStatus.value === "skipped";
+  });
+}
+
+export function computeBlockedByFailure(
+  _nodeKey: string,
+  ctx: NodeStatusContext,
+): boolean {
+  return ctx.predecessors.some((pred: string) => {
+    const predStatus = ctx.statusMap.get(pred);
+    if (!predStatus) return false;
+    return predStatus.value === "failed" || predStatus.value === "aborted";
+  });
+}
+
+export function registerStartEffect(
+  status: Signal<NodeStatus>,
+  preconditions: ReadonlySignal<boolean>,
+  effectDisposers: (() => void)[],
+): void {
+  const disposer = effect(() => {
+    if (preconditions.value) {
+      const current = status.value;
+      if (current === "idle" || current === "waiting") {
+        status.value = "ready";
+      }
+    }
+  });
+  effectDisposers.push(disposer);
+}
+
+export interface AbortEffectOptions {
+  abortDependents?: boolean;
+}
+
+export function registerAbortEffect(
+  status: Signal<NodeStatus>,
+  blockedByFailure: ReadonlySignal<boolean>,
+  effectDisposers: (() => void)[],
+  options?: AbortEffectOptions,
+): void {
+  const disposer = effect(() => {
+    if (blockedByFailure.value) {
+      const current = status.value;
+      if (options?.abortDependents) {
+        if (!TERMINAL_STATUSES.has(current)) {
+          status.value = "aborted";
+        }
+      } else {
+        if (current === "idle" || current === "waiting") {
+          status.value = "aborted";
+        }
+      }
+    }
+  });
+  effectDisposers.push(disposer);
+}

src/reactive/workflow.ts

@@ -1,8 +1,15 @@
-import { signal, computed, effect } from "@preact/signals-core";
+import { signal, computed } from "@preact/signals-core";
 import type { Signal, ReadonlySignal } from "@preact/signals-core";
 import type { DirectedGraph } from "graphology";
 import type { NodeStatus } from "../schema/enums.js";
 import type { CallResult } from "../schema/edge.js";
+import {
+  computePreconditions,
+  computeBlockedByFailure,
+  registerStartEffect,
+  registerAbortEffect,
+} from "./node-status.js";
+import type { NodeStatusContext } from "./node-status.js";
 
 export type FailurePolicy = "continue-running" | "abort-dependents";
 
@@ -123,24 +130,17 @@ export class WorkflowReactiveRoot implements EventLogProjection {
 
       const status = signal<NodeStatus>("idle");
 
+      const ctx: NodeStatusContext = {
+        statusMap: this.statusMap,
+        predecessors,
+      };
+
       const preconditionsComputed = computed(() => {
-        return predecessors.every((pred: string) => {
-          const predStatus = this.statusMap.get(pred);
-          if (!predStatus) return false;
-          return (
-            predStatus.value === "completed" || predStatus.value === "skipped"
-          );
-        });
+        return computePreconditions(node, ctx);
       });
 
       const blockedByFailureComputed = computed(() => {
-        return predecessors.some((pred: string) => {
-          const predStatus = this.statusMap.get(pred);
-          if (!predStatus) return false;
-          return (
-            predStatus.value === "failed" || predStatus.value === "aborted"
-          );
-        });
+        return computeBlockedByFailure(node, ctx);
       });
 
       const resultComputed = computed(() => {
@@ -195,23 +195,13 @@ export class WorkflowReactiveRoot implements EventLogProjection {
 
     for (const node of this.graph.nodes()) {
       const status = this.statusMap.get(node)!;
+      const preconditions = this.preconditions.get(node)!;
       const blocked = this.blockedByFailure.get(node)!;
 
-      const disposer = effect(() => {
-        if (blocked.value) {
-          const current = status.value;
-          if (current === "idle" || current === "waiting" || current === "ready") {
-            if (this._failurePolicy === "abort-dependents") {
-              if (!TERMINAL_STATUSES.has(current)) {
-                status.value = "aborted";
-              }
-            } else {
-              status.value = "aborted";
-            }
-          }
-        }
+      registerStartEffect(status, preconditions, this.effectDisposers);
+      registerAbortEffect(status, blocked, this.effectDisposers, {
+        abortDependents: this._failurePolicy === "abort-dependents",
       });
-      this.effectDisposers.push(disposer);
     }
   }
 
@@ -354,4 +344,4 @@ export class WorkflowReactiveRoot implements EventLogProjection {
     this.requestIdToNodeKey.clear();
     this.eventLog = [];
   }
-}
+}

test/reactive/node-status.test.ts

@@ -0,0 +1,663 @@
+import { describe, it, expect } from "vitest";
+import { signal, computed } from "@preact/signals-core";
+import type { Signal, ReadonlySignal } from "@preact/signals-core";
+import { DirectedGraph } from "graphology";
+import {
+  computePreconditions,
+  computeBlockedByFailure,
+  registerStartEffect,
+  registerAbortEffect,
+} from "../../src/reactive/node-status.js";
+import type { NodeStatusContext } from "../../src/reactive/node-status.js";
+import { WorkflowReactiveRoot } from "../../src/reactive/workflow.js";
+import type { NodeStatus } from "../../src/schema/enums.js";
+
+function makeContext(
+  statusMap: Map<string, Signal<NodeStatus>>,
+  predecessors: string[],
+): NodeStatusContext {
+  return { statusMap, predecessors };
+}
+
+function makeStatusMap(
+  entries: [string, NodeStatus][],
+): Map<string, Signal<NodeStatus>> {
+  const map = new Map<string, Signal<NodeStatus>>();
+  for (const [key, value] of entries) {
+    map.set(key, signal<NodeStatus>(value));
+  }
+  return map;
+}
+
+describe("computePreconditions", () => {
+  it("returns true for root node with no predecessors", () => {
+    const statusMap = makeStatusMap([]);
+    const ctx = makeContext(statusMap, []);
+    expect(computePreconditions("a", ctx)).toBe(true);
+  });
+
+  it("returns false when predecessor is idle", () => {
+    const statusMap = makeStatusMap([["a", "idle"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(false);
+  });
+
+  it("returns true when all predecessors are completed", () => {
+    const statusMap = makeStatusMap([["a", "completed"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(true);
+  });
+
+  it("returns true when all predecessors are skipped", () => {
+    const statusMap = makeStatusMap([["a", "skipped"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(true);
+  });
+
+  it("returns true when predecessors are mix of completed and skipped", () => {
+    const statusMap = makeStatusMap([
+      ["a", "completed"],
+      ["b", "skipped"],
+    ]);
+    const ctx = makeContext(statusMap, ["a", "b"]);
+    expect(computePreconditions("c", ctx)).toBe(true);
+  });
+
+  it("returns false when predecessor is failed", () => {
+    const statusMap = makeStatusMap([["a", "failed"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(false);
+  });
+
+  it("returns false when predecessor is aborted", () => {
+    const statusMap = makeStatusMap([["a", "aborted"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(false);
+  });
+
+  it("returns false when one of multiple predecessors is failed", () => {
+    const statusMap = makeStatusMap([
+      ["a", "completed"],
+      ["b", "failed"],
+    ]);
+    const ctx = makeContext(statusMap, ["a", "b"]);
+    expect(computePreconditions("c", ctx)).toBe(false);
+  });
+
+  it("returns false when predecessor is running", () => {
+    const statusMap = makeStatusMap([["a", "running"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(false);
+  });
+
+  it("returns false when predecessor is waiting", () => {
+    const statusMap = makeStatusMap([["a", "waiting"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computePreconditions("b", ctx)).toBe(false);
+  });
+
+  it("returns false for missing predecessor in statusMap", () => {
+    const statusMap = makeStatusMap([]);
+    const ctx = makeContext(statusMap, ["unknown"]);
+    expect(computePreconditions("b", ctx)).toBe(false);
+  });
+});
+
+describe("computeBlockedByFailure", () => {
+  it("returns false for root node with no predecessors", () => {
+    const statusMap = makeStatusMap([]);
+    const ctx = makeContext(statusMap, []);
+    expect(computeBlockedByFailure("a", ctx)).toBe(false);
+  });
+
+  it("returns true when predecessor is failed", () => {
+    const statusMap = makeStatusMap([["a", "failed"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computeBlockedByFailure("b", ctx)).toBe(true);
+  });
+
+  it("returns true when predecessor is aborted", () => {
+    const statusMap = makeStatusMap([["a", "aborted"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computeBlockedByFailure("b", ctx)).toBe(true);
+  });
+
+  it("returns false when predecessor is completed", () => {
+    const statusMap = makeStatusMap([["a", "completed"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computeBlockedByFailure("b", ctx)).toBe(false);
+  });
+
+  it("returns false when predecessor is skipped", () => {
+    const statusMap = makeStatusMap([["a", "skipped"]]);
+    const ctx = makeContext(statusMap, ["a"]);
+    expect(computeBlockedByFailure("b", ctx)).toBe(false);
+  });
+
+  it("returns true when any of multiple predecessors is failed", () => {
+    const statusMap = makeStatusMap([
+      ["a", "completed"],
+      ["b", "failed"],
+    ]);
+    const ctx = makeContext(statusMap, ["a", "b"]);
+    expect(computeBlockedByFailure("c", ctx)).toBe(true);
+  });
+
+  it("returns false when all predecessors are completed", () => {
+    const statusMap = makeStatusMap([
+      ["a", "completed"],
+      ["b", "completed"],
+    ]);
+    const ctx = makeContext(statusMap, ["a", "b"]);
+    expect(computeBlockedByFailure("c", ctx)).toBe(false);
+  });
+
+  it("returns false for missing predecessor in statusMap", () => {
+    const statusMap = makeStatusMap([]);
+    const ctx = makeContext(statusMap, ["unknown"]);
+    expect(computeBlockedByFailure("b", ctx)).toBe(false);
+  });
+});
+
+describe("registerStartEffect", () => {
+  it("transitions idle node to ready when preconditions are true", () => {
+    const status = signal<NodeStatus>("idle");
+    const preconditions = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(status.value).toBe("ready");
+
+    for (const d of disposers) d();
+  });
+
+  it("transitions waiting node to ready when preconditions are true", () => {
+    const status = signal<NodeStatus>("waiting");
+    const preconditions = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(status.value).toBe("ready");
+
+    for (const d of disposers) d();
+  });
+
+  it("does not transition running node when preconditions become true", () => {
+    const status = signal<NodeStatus>("running");
+    const preconditions = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(status.value).toBe("running");
+
+    for (const d of disposers) d();
+  });
+
+  it("does not transition completed node when preconditions become true", () => {
+    const status = signal<NodeStatus>("completed");
+    const preconditions = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(status.value).toBe("completed");
+
+    for (const d of disposers) d();
+  });
+
+  it("does not transition idle node when preconditions are false", () => {
+    const status = signal<NodeStatus>("idle");
+    const preconditions = computed(() => false);
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(status.value).toBe("idle");
+
+    for (const d of disposers) d();
+  });
+
+  it("reactively transitions to ready when preconditions change from false to true", () => {
+    const trigger = signal<NodeStatus>("idle");
+    const preconditions = computed(() => trigger.value === "completed");
+    const status = signal<NodeStatus>("idle");
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(status.value).toBe("idle");
+
+    trigger.value = "completed";
+
+    expect(status.value).toBe("ready");
+
+    for (const d of disposers) d();
+  });
+});
+
+describe("registerAbortEffect", () => {
+  it("transitions idle node to aborted when blockedByFailure is true", () => {
+    const status = signal<NodeStatus>("idle");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers);
+
+    expect(status.value).toBe("aborted");
+
+    for (const d of disposers) d();
+  });
+
+  it("transitions waiting node to aborted when blockedByFailure is true", () => {
+    const status = signal<NodeStatus>("waiting");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers);
+
+    expect(status.value).toBe("aborted");
+
+    for (const d of disposers) d();
+  });
+
+  it("does not transition ready node with default policy", () => {
+    const status = signal<NodeStatus>("ready");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers);
+
+    expect(status.value).toBe("ready");
+
+    for (const d of disposers) d();
+  });
+
+  it("transitions ready node to aborted with abortDependents option", () => {
+    const status = signal<NodeStatus>("ready");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers, {
+      abortDependents: true,
+    });
+
+    expect(status.value).toBe("aborted");
+
+    for (const d of disposers) d();
+  });
+
+  it("transitions running node to aborted with abortDependents option", () => {
+    const status = signal<NodeStatus>("running");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers, {
+      abortDependents: true,
+    });
+
+    expect(status.value).toBe("aborted");
+
+    for (const d of disposers) d();
+  });
+
+  it("does not transition completed node even with abortDependents", () => {
+    const status = signal<NodeStatus>("completed");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers, {
+      abortDependents: true,
+    });
+
+    expect(status.value).toBe("completed");
+
+    for (const d of disposers) d();
+  });
+
+  it("does not transition failed node even with abortDependents", () => {
+    const status = signal<NodeStatus>("failed");
+    const blockedByFailure = computed(() => true);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers, {
+      abortDependents: true,
+    });
+
+    expect(status.value).toBe("failed");
+
+    for (const d of disposers) d();
+  });
+
+  it("reactively transitions to aborted when blockedByFailure changes", () => {
+    const trigger = signal<NodeStatus>("idle");
+    const blockedByFailure = computed(
+      () => trigger.value === "failed" || trigger.value === "aborted",
+    );
+    const status = signal<NodeStatus>("idle");
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers);
+
+    expect(status.value).toBe("idle");
+
+    trigger.value = "failed";
+
+    expect(status.value).toBe("aborted");
+
+    for (const d of disposers) d();
+  });
+});
+
+describe("sequential preconditions via WorkflowReactiveRoot", () => {
+  function makeSeqGraph(): DirectedGraph {
+    const graph = new DirectedGraph();
+    graph.addNode("a", { name: "a" });
+    graph.addNode("b", { name: "b" });
+    graph.addNode("c", { name: "c" });
+    graph.addEdgeWithKey("a->b", "a", "b", { edgeType: "sequential" });
+    graph.addEdgeWithKey("b->c", "b", "c", { edgeType: "sequential" });
+    return graph;
+  }
+
+  it("root node transitions to ready (no predecessors)", () => {
+    const graph = makeSeqGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    expect(root.statusMap.get("a")!.value).toBe("ready");
+
+    root.dispose();
+  });
+
+  it("downstream nodes stay idle until predecessor completes", () => {
+    const graph = makeSeqGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    expect(root.statusMap.get("b")!.value).toBe("idle");
+
+    root.dispose();
+  });
+
+  it("downstream node transitions to ready after predecessor completes", () => {
+    const graph = makeSeqGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("a")!.value = "completed";
+
+    expect(root.statusMap.get("b")!.value).toBe("ready");
+
+    root.dispose();
+  });
+
+  it("full sequential chain transitions", () => {
+    const graph = makeSeqGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("a")!.value = "completed";
+    expect(root.statusMap.get("b")!.value).toBe("ready");
+
+    root.statusMap.get("b")!.value = "completed";
+    expect(root.statusMap.get("c")!.value).toBe("ready");
+
+    root.dispose();
+  });
+});
+
+describe("parallel preconditions via WorkflowReactiveRoot", () => {
+  function makeParallelGraph(): DirectedGraph {
+    const graph = new DirectedGraph();
+    graph.addNode("top", { name: "top" });
+    graph.addNode("left", { name: "left" });
+    graph.addNode("right", { name: "right" });
+    graph.addEdgeWithKey("top->left", "top", "left", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("top->right", "top", "right", {
+      edgeType: "sequential",
+    });
+    return graph;
+  }
+
+  it("parallel siblings both become ready when parent completes", () => {
+    const graph = makeParallelGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+
+    expect(root.statusMap.get("left")!.value).toBe("ready");
+    expect(root.statusMap.get("right")!.value).toBe("ready");
+
+    root.dispose();
+  });
+
+  it("parallel siblings are independent - failure on one does not abort the other", () => {
+    const graph = makeParallelGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+    root.statusMap.get("left")!.value = "ready";
+    root.statusMap.get("right")!.value = "ready";
+
+    root.statusMap.get("left")!.value = "failed";
+
+    expect(root.statusMap.get("left")!.value).toBe("failed");
+    expect(root.statusMap.get("right")!.value).toBe("ready");
+
+    root.dispose();
+  });
+});
+
+describe("join (fork-join) preconditions via WorkflowReactiveRoot", () => {
+  function makeDiamondGraph(): DirectedGraph {
+    const graph = new DirectedGraph();
+    graph.addNode("top", { name: "top" });
+    graph.addNode("left", { name: "left" });
+    graph.addNode("right", { name: "right" });
+    graph.addNode("bottom", { name: "bottom" });
+    graph.addEdgeWithKey("top->left", "top", "left", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("top->right", "top", "right", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("left->bottom", "left", "bottom", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("right->bottom", "right", "bottom", {
+      edgeType: "sequential",
+    });
+    return graph;
+  }
+
+  it("join node stays idle until all predecessors complete", () => {
+    const graph = makeDiamondGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+    root.statusMap.get("left")!.value = "completed";
+
+    expect(root.statusMap.get("bottom")!.value).toBe("idle");
+
+    root.dispose();
+  });
+
+  it("join node becomes ready when all predecessors are completed", () => {
+    const graph = makeDiamondGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+    root.statusMap.get("left")!.value = "completed";
+    root.statusMap.get("right")!.value = "completed";
+
+    expect(root.statusMap.get("bottom")!.value).toBe("ready");
+
+    root.dispose();
+  });
+
+  it("join node blocked when one predecessor fails", () => {
+    const graph = makeDiamondGraph();
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+    root.statusMap.get("left")!.value = "completed";
+    root.statusMap.get("right")!.value = "failed";
+
+    expect(root.preconditions.get("bottom")!.value).toBe(false);
+    expect(root.statusMap.get("bottom")!.value).toBe("aborted");
+
+    root.dispose();
+  });
+});
+
+describe("blockedByFailure cascade via WorkflowReactiveRoot", () => {
+  it("failure cascades through multiple downstream nodes", () => {
+    const graph = new DirectedGraph();
+    graph.addNode("a", { name: "a" });
+    graph.addNode("b", { name: "b" });
+    graph.addNode("c", { name: "c" });
+    graph.addEdgeWithKey("a->b", "a", "b", { edgeType: "sequential" });
+    graph.addEdgeWithKey("b->c", "b", "c", { edgeType: "sequential" });
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("a")!.value = "failed";
+
+    expect(root.statusMap.get("b")!.value).toBe("aborted");
+    expect(root.statusMap.get("c")!.value).toBe("aborted");
+
+    root.dispose();
+  });
+
+  it("aborted predecessor causes cascade just like failed", () => {
+    const graph = new DirectedGraph();
+    graph.addNode("a", { name: "a" });
+    graph.addNode("b", { name: "b" });
+    graph.addEdgeWithKey("a->b", "a", "b", { edgeType: "sequential" });
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("a")!.value = "aborted";
+
+    expect(root.statusMap.get("b")!.value).toBe("aborted");
+
+    root.dispose();
+  });
+});
+
+describe("skipped satisfies preconditions via WorkflowReactiveRoot", () => {
+  it("skipped predecessor satisfies preconditions for downstream node", () => {
+    const graph = new DirectedGraph();
+    graph.addNode("a", { name: "a" });
+    graph.addNode("b", { name: "b" });
+    graph.addEdgeWithKey("a->b", "a", "b", { edgeType: "sequential" });
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("a")!.value = "skipped";
+
+    expect(root.preconditions.get("b")!.value).toBe(true);
+    expect(root.statusMap.get("b")!.value).toBe("ready");
+
+    root.dispose();
+  });
+});
+
+describe("failure isolation in parallel branches via WorkflowReactiveRoot", () => {
+  it("failure in one branch does not abort sibling branch", () => {
+    const graph = new DirectedGraph();
+    graph.addNode("top", { name: "top" });
+    graph.addNode("left", { name: "left" });
+    graph.addNode("right", { name: "right" });
+    graph.addNode("bottom", { name: "bottom" });
+    graph.addEdgeWithKey("top->left", "top", "left", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("top->right", "top", "right", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("left->bottom", "left", "bottom", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("right->bottom", "right", "bottom", {
+      edgeType: "sequential",
+    });
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+    root.statusMap.get("left")!.value = "running";
+    root.statusMap.get("right")!.value = "failed";
+
+    expect(root.statusMap.get("left")!.value).toBe("running");
+    expect(root.statusMap.get("bottom")!.value).toBe("aborted");
+
+    root.dispose();
+  });
+
+  it("completed branch is not affected by failure in other branch", () => {
+    const graph = new DirectedGraph();
+    graph.addNode("top", { name: "top" });
+    graph.addNode("left", { name: "left" });
+    graph.addNode("right", { name: "right" });
+    graph.addNode("bottom", { name: "bottom" });
+    graph.addEdgeWithKey("top->left", "top", "left", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("top->right", "top", "right", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("left->bottom", "left", "bottom", {
+      edgeType: "sequential",
+    });
+    graph.addEdgeWithKey("right->bottom", "right", "bottom", {
+      edgeType: "sequential",
+    });
+    const root = new WorkflowReactiveRoot(graph);
+
+    root.statusMap.get("top")!.value = "completed";
+    root.statusMap.get("left")!.value = "completed";
+    root.statusMap.get("right")!.value = "failed";
+
+    expect(root.statusMap.get("left")!.value).toBe("completed");
+
+    root.dispose();
+  });
+});
+
+describe("effect disposal", () => {
+  it("start effect is tracked by effectDisposers and cleaned up", () => {
+    const status = signal<NodeStatus>("idle");
+    const preconditions = computed(() => false);
+    const disposers: (() => void)[] = [];
+
+    registerStartEffect(status, preconditions, disposers);
+
+    expect(disposers.length).toBe(1);
+
+    disposers[0]!();
+  });
+
+  it("abort effect is tracked by effectDisposer and cleaned up", () => {
+    const status = signal<NodeStatus>("idle");
+    const blockedByFailure = computed(() => false);
+    const disposers: (() => void)[] = [];
+
+    registerAbortEffect(status, blockedByFailure, disposers);
+
+    expect(disposers.length).toBe(1);
+
+    disposers[0]!();
+  });
+
+  it("WorkflowReactiveRoot dispose clears all effects", () => {
+    const graph = new DirectedGraph();
+    graph.addNode("a", { name: "a" });
+    graph.addNode("b", { name: "b" });
+    graph.addEdgeWithKey("a->b", "a", "b", { edgeType: "sequential" });
+    const root = new WorkflowReactiveRoot(graph);
+
+    expect(root.statusMap.get("a")!.value).toBe("ready");
+
+    root.dispose();
+
+    expect(root.statusMap.size).toBe(0);
+  });
+});

test/reactive/workflow.test.ts

@@ -48,11 +48,11 @@ function makeForkJoinGraph(): DirectedGraph {
 
 describe("WorkflowReactiveRoot", () => {
   describe("constructor and initializeSignals", () => {
-    it("initializes all nodes with idle status", () => {
+    it("root node starts as ready, downstream nodes start as idle", () => {
       const graph = makeSimpleGraph();
       const root = new WorkflowReactiveRoot(graph);
 
-      expect(root.statusMap.get("a")!.value).toBe("idle");
+      expect(root.statusMap.get("a")!.value).toBe("ready");
       expect(root.statusMap.get("b")!.value).toBe("idle");
       expect(root.statusMap.get("c")!.value).toBe("idle");
 
@@ -256,7 +256,7 @@ describe("WorkflowReactiveRoot", () => {
         timestamp: "2026-01-01T00:00:00Z",
       });
 
-      expect(root.statusMap.get("a")!.value).toBe("idle");
+      expect(root.statusMap.get("a")!.value).toBe("ready");
 
       root.dispose();
     });
@@ -291,11 +291,11 @@ describe("WorkflowReactiveRoot", () => {
   });
 
   describe("getStatus", () => {
-    it("returns idle for nodes with no events", () => {
+    it("returns ready for root nodes with no events", () => {
       const graph = makeSimpleGraph();
       const root = new WorkflowReactiveRoot(graph);
 
-      expect(root.getStatus("a")).toBe("idle");
+      expect(root.getStatus("a")).toBe("ready");
 
       root.dispose();
     });
@@ -322,9 +322,9 @@ describe("WorkflowReactiveRoot", () => {
       const graph = makeSimpleGraph();
       const root = new WorkflowReactiveRoot(graph);
 
-      root.statusMap.get("a")!.value = "waiting";
+      root.statusMap.get("b")!.value = "waiting";
 
-      expect(root.getStatus("a")).toBe("waiting");
+      expect(root.getStatus("b")).toBe("waiting");
 
       root.dispose();
     });
@@ -1058,7 +1058,7 @@ describe("WorkflowReactiveRoot", () => {
       root.setRequestId("b", "req-b");
       root.setRequestId("c", "req-c");
 
-      expect(root.getStatus("a")).toBe("idle");
+      expect(root.getStatus("a")).toBe("ready");
       expect(root.getStatus("b")).toBe("idle");
       expect(root.getStatus("c")).toBe("idle");