feat: implement node status signal management with computed preconditions and blockedByFailure
- Add computePreconditions and computeBlockedByFailure functions to node-status.ts - Add registerStartEffect and registerAbortEffect for automatic state transitions - Start effect: idle/waiting -> ready when preconditions met - Abort effect: idle/waiting -> aborted when blockedByFailure true - Refactor WorkflowReactiveRoot to use node-status.ts functions - Root nodes auto-transition from idle to ready (no predecessors = preconditions true) - Add AbortEffectOptions with abortDependents policy support - Add comprehensive unit tests for all precondition and failure isolation scenarios
This commit is contained in:
@@ -10,3 +10,12 @@ export {
|
|||||||
type EventLogProjection,
|
type EventLogProjection,
|
||||||
type AggregateStatus,
|
type AggregateStatus,
|
||||||
} from "./workflow.js";
|
} from "./workflow.js";
|
||||||
|
|
||||||
|
export {
|
||||||
|
computePreconditions,
|
||||||
|
computeBlockedByFailure,
|
||||||
|
registerStartEffect,
|
||||||
|
registerAbortEffect,
|
||||||
|
type NodeStatusContext,
|
||||||
|
type AbortEffectOptions,
|
||||||
|
} from "./node-status.js";
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
@@ -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 { Signal, ReadonlySignal } from "@preact/signals-core";
|
||||||
import type { DirectedGraph } from "graphology";
|
import type { DirectedGraph } from "graphology";
|
||||||
import type { NodeStatus } from "../schema/enums.js";
|
import type { NodeStatus } from "../schema/enums.js";
|
||||||
import type { CallResult } from "../schema/edge.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";
|
export type FailurePolicy = "continue-running" | "abort-dependents";
|
||||||
|
|
||||||
@@ -123,24 +130,17 @@ export class WorkflowReactiveRoot implements EventLogProjection {
|
|||||||
|
|
||||||
const status = signal<NodeStatus>("idle");
|
const status = signal<NodeStatus>("idle");
|
||||||
|
|
||||||
|
const ctx: NodeStatusContext = {
|
||||||
|
statusMap: this.statusMap,
|
||||||
|
predecessors,
|
||||||
|
};
|
||||||
|
|
||||||
const preconditionsComputed = computed(() => {
|
const preconditionsComputed = computed(() => {
|
||||||
return predecessors.every((pred: string) => {
|
return computePreconditions(node, ctx);
|
||||||
const predStatus = this.statusMap.get(pred);
|
|
||||||
if (!predStatus) return false;
|
|
||||||
return (
|
|
||||||
predStatus.value === "completed" || predStatus.value === "skipped"
|
|
||||||
);
|
|
||||||
});
|
|
||||||
});
|
});
|
||||||
|
|
||||||
const blockedByFailureComputed = computed(() => {
|
const blockedByFailureComputed = computed(() => {
|
||||||
return predecessors.some((pred: string) => {
|
return computeBlockedByFailure(node, ctx);
|
||||||
const predStatus = this.statusMap.get(pred);
|
|
||||||
if (!predStatus) return false;
|
|
||||||
return (
|
|
||||||
predStatus.value === "failed" || predStatus.value === "aborted"
|
|
||||||
);
|
|
||||||
});
|
|
||||||
});
|
});
|
||||||
|
|
||||||
const resultComputed = computed(() => {
|
const resultComputed = computed(() => {
|
||||||
@@ -195,23 +195,13 @@ export class WorkflowReactiveRoot implements EventLogProjection {
|
|||||||
|
|
||||||
for (const node of this.graph.nodes()) {
|
for (const node of this.graph.nodes()) {
|
||||||
const status = this.statusMap.get(node)!;
|
const status = this.statusMap.get(node)!;
|
||||||
|
const preconditions = this.preconditions.get(node)!;
|
||||||
const blocked = this.blockedByFailure.get(node)!;
|
const blocked = this.blockedByFailure.get(node)!;
|
||||||
|
|
||||||
const disposer = effect(() => {
|
registerStartEffect(status, preconditions, this.effectDisposers);
|
||||||
if (blocked.value) {
|
registerAbortEffect(status, blocked, this.effectDisposers, {
|
||||||
const current = status.value;
|
abortDependents: this._failurePolicy === "abort-dependents",
|
||||||
if (current === "idle" || current === "waiting" || current === "ready") {
|
|
||||||
if (this._failurePolicy === "abort-dependents") {
|
|
||||||
if (!TERMINAL_STATUSES.has(current)) {
|
|
||||||
status.value = "aborted";
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
status.value = "aborted";
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
});
|
||||||
this.effectDisposers.push(disposer);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -354,4 +344,4 @@ export class WorkflowReactiveRoot implements EventLogProjection {
|
|||||||
this.requestIdToNodeKey.clear();
|
this.requestIdToNodeKey.clear();
|
||||||
this.eventLog = [];
|
this.eventLog = [];
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
663
test/reactive/node-status.test.ts
Normal file
663
test/reactive/node-status.test.ts
Normal file
@@ -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);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -48,11 +48,11 @@ function makeForkJoinGraph(): DirectedGraph {
|
|||||||
|
|
||||||
describe("WorkflowReactiveRoot", () => {
|
describe("WorkflowReactiveRoot", () => {
|
||||||
describe("constructor and initializeSignals", () => {
|
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 graph = makeSimpleGraph();
|
||||||
const root = new WorkflowReactiveRoot(graph);
|
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("b")!.value).toBe("idle");
|
||||||
expect(root.statusMap.get("c")!.value).toBe("idle");
|
expect(root.statusMap.get("c")!.value).toBe("idle");
|
||||||
|
|
||||||
@@ -256,7 +256,7 @@ describe("WorkflowReactiveRoot", () => {
|
|||||||
timestamp: "2026-01-01T00:00:00Z",
|
timestamp: "2026-01-01T00:00:00Z",
|
||||||
});
|
});
|
||||||
|
|
||||||
expect(root.statusMap.get("a")!.value).toBe("idle");
|
expect(root.statusMap.get("a")!.value).toBe("ready");
|
||||||
|
|
||||||
root.dispose();
|
root.dispose();
|
||||||
});
|
});
|
||||||
@@ -291,11 +291,11 @@ describe("WorkflowReactiveRoot", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe("getStatus", () => {
|
describe("getStatus", () => {
|
||||||
it("returns idle for nodes with no events", () => {
|
it("returns ready for root nodes with no events", () => {
|
||||||
const graph = makeSimpleGraph();
|
const graph = makeSimpleGraph();
|
||||||
const root = new WorkflowReactiveRoot(graph);
|
const root = new WorkflowReactiveRoot(graph);
|
||||||
|
|
||||||
expect(root.getStatus("a")).toBe("idle");
|
expect(root.getStatus("a")).toBe("ready");
|
||||||
|
|
||||||
root.dispose();
|
root.dispose();
|
||||||
});
|
});
|
||||||
@@ -322,9 +322,9 @@ describe("WorkflowReactiveRoot", () => {
|
|||||||
const graph = makeSimpleGraph();
|
const graph = makeSimpleGraph();
|
||||||
const root = new WorkflowReactiveRoot(graph);
|
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();
|
root.dispose();
|
||||||
});
|
});
|
||||||
@@ -845,7 +845,7 @@ describe("WorkflowReactiveRoot", () => {
|
|||||||
root.setRequestId("b", "req-b");
|
root.setRequestId("b", "req-b");
|
||||||
root.setRequestId("c", "req-c");
|
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("b")).toBe("idle");
|
||||||
expect(root.getStatus("c")).toBe("idle");
|
expect(root.getStatus("c")).toBe("idle");
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user