10 Commits

Author SHA1 Message Date
e98204161d 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
2026-05-21 22:16:46 +00:00
18999fb38e chore: update task statuses to completed 2026-05-21 22:09:10 +00:00
7a4e430aa9 Merge branch 'feat/analysis-build-type-edges'
# Conflicts:
#	src/analysis/index.ts
#	src/graph/construction.ts
#	src/index.ts
2026-05-21 22:08:42 +00:00
c7c6d13d6b Merge branch 'feat/analysis-template-validation' 2026-05-21 22:07:11 +00:00
c57a8558c7 Merge branch 'feat/graph-construction-call'
# Conflicts:
#	src/graph/construction.ts
#	test/graph/construction.test.ts
2026-05-21 22:06:51 +00:00
fa2223b90b feat: move buildTypeEdges to src/analysis/type-compat.ts as standalone function
Relocate buildTypeEdges from construction.ts to type-compat.ts per architecture spec.
construction.ts re-exports it for backward compatibility. Add 10 unit tests for
buildTypeEdges covering compatible edges, incompatible edges with mismatches,
unknown schema passthrough, incremental construction, and edge deduplication.
2026-05-21 22:06:26 +00:00
2e0350e87c feat: implement call graph construction methods (fromCallEvents, updateFromEvent, addCall, addDependency, updateStatus, updateCall, removeCall) 2026-05-21 22:04:28 +00:00
d7fea47214 Merge branch 'feat/analysis-ordering' 2026-05-21 22:03:19 +00:00
67907dc0f3 Implement validatePreconditions and validateTemplate for template validation 2026-05-21 22:02:31 +00:00
fb921f9a29 feat: add execution ordering analysis functions (topologicalOrder, parallelGroups, criticalPath, reachableFrom, ancestors, descendants) 2026-05-21 21:58:30 +00:00
20 changed files with 2523 additions and 115 deletions

View File

@@ -4,10 +4,18 @@ export {
defaultEdgeType,
resolveDefaultNodeAttrs,
} from "./defaults.js";
export { typeCompat, type TypeCompatResult, type TypeMismatch } from "./type-compat.js";
export { buildTypeEdges } from "../graph/construction.js";
export { typeCompat, buildTypeEdges, type TypeCompatResult, type TypeMismatch } from "./type-compat.js";
export {
topologicalOrder,
parallelGroups,
criticalPath,
reachableFrom,
ancestors,
descendants,
} from "./ordering.js";
export {
validateSchema,
validateGraph,
validate,
} from "../graph/validation.js";
export { validatePreconditions, validateTemplate } from "./workflow.js";

86
src/analysis/ordering.ts Normal file
View File

@@ -0,0 +1,86 @@
import { topologicalSort, topologicalGenerations, hasCycle } from "graphology-dag";
import type { FlowGraph } from "../graph/construction.js";
import { CycleError, NodeNotFoundError } from "../error/index.js";
import { findCycles as findCyclesQuery, reachableFrom as reachableFromQuery, ancestors as ancestorsQuery, descendants as descendantsQuery } from "../graph/queries.js";
export function topologicalOrder(graph: FlowGraph): string[] {
if (hasCycle(graph.graph)) {
const cycles = findCyclesQuery(graph.graph);
throw new CycleError(cycles);
}
return topologicalSort(graph.graph);
}
export function parallelGroups(graph: FlowGraph): string[][] {
if (hasCycle(graph.graph)) {
const cycles = findCyclesQuery(graph.graph);
throw new CycleError(cycles);
}
return topologicalGenerations(graph.graph);
}
export function criticalPath(graph: FlowGraph): string[] {
if (hasCycle(graph.graph)) {
const cycles = findCyclesQuery(graph.graph);
throw new CycleError(cycles);
}
const dg = graph.graph;
const depth = new Map<string, number>();
const topo = topologicalSort(dg);
for (const node of topo) {
const inNeighbors = dg.inNeighbors(node) ?? [];
let maxPredDepth = -1;
for (const pred of inNeighbors) {
const predDepth = depth.get(pred!) ?? 0;
if (predDepth > maxPredDepth) {
maxPredDepth = predDepth;
}
}
depth.set(node, maxPredDepth + 1);
}
let maxDepth = -1;
let endNode = "";
for (const node of topo) {
const d = depth.get(node) ?? 0;
if (d > maxDepth) {
maxDepth = d;
endNode = node;
}
}
if (topo.length === 0) return [];
const path: string[] = [endNode];
let current = endNode;
while (depth.get(current)! > 0) {
const inNeighbors = dg.inNeighbors(current) ?? [];
let bestPred = "";
let bestDepth = -1;
for (const pred of inNeighbors) {
const predDepth = depth.get(pred!) ?? 0;
if (predDepth > bestDepth) {
bestDepth = predDepth;
bestPred = pred!;
}
}
path.unshift(bestPred);
current = bestPred;
}
return path;
}
export function reachableFrom(graph: FlowGraph, nodeIds: string[]): Set<string> {
return reachableFromQuery(graph.graph, nodeIds);
}
export function ancestors(graph: FlowGraph, nodeId: string): string[] {
if (!graph.hasNode(nodeId)) {
throw new NodeNotFoundError(nodeId);
}
return ancestorsQuery(graph.graph, nodeId);
}
export function descendants(graph: FlowGraph, nodeId: string): string[] {
if (!graph.hasNode(nodeId)) {
throw new NodeNotFoundError(nodeId);
}
return descendantsQuery(graph.graph, nodeId);
}

View File

@@ -1,4 +1,11 @@
import { KindGuard, Kind, type TSchema } from "@alkdev/typebox";
import { willCreateCycle } from "graphology-dag";
import type { FlowGraph } from "../graph/construction.js";
import {
OperationNodeAttrs as OperationNodeAttrsSchema,
OperationEdgeAttrs as OperationEdgeAttrsSchema,
} from "../schema/index.js";
import type { OperationNodeAttrs } from "../schema/index.js";
export interface TypeMismatch {
path: string;
@@ -276,3 +283,24 @@ export function typeCompat(outputSchema: TSchema, inputSchema: TSchema): TypeCom
return { compatible: false, mismatches };
}
export function buildTypeEdges(graph: FlowGraph<typeof OperationNodeAttrsSchema, typeof OperationEdgeAttrsSchema>): void {
const nodeKeys = graph.nodes();
for (const source of nodeKeys) {
for (const target of nodeKeys) {
if (source === target) continue;
const sourceAttrs = graph.getNodeAttributes(source as never) as unknown as OperationNodeAttrs;
const targetAttrs = graph.getNodeAttributes(target as never) as unknown as OperationNodeAttrs;
const result = typeCompat(sourceAttrs.outputSchema as TSchema, targetAttrs.inputSchema as TSchema);
if (result === undefined) continue;
if (graph.hasEdge(source, target)) continue;
if (willCreateCycle(graph.graph, source, target)) continue;
const detail = result.detail ?? `${sourceAttrs.namespace}.${sourceAttrs.name}.output → ${targetAttrs.namespace}.${targetAttrs.name}.input`;
graph.addTypedEdge(source, target, {
compatible: result.compatible,
detail,
...(result.mismatches !== undefined ? { mismatches: result.mismatches } : {}),
});
}
}
}

View File

@@ -1 +1,197 @@
export {};
import type { TSchema } from "@alkdev/typebox";
import { KindGuard } from "@alkdev/typebox";
import type { UNode } from "@alkdev/ujsx";
import { createHostRoot } from "@alkdev/ujsx";
import { hasCycle } from "graphology-dag";
import { DirectedGraph } from "graphology";
import type { FlowGraph } from "../graph/construction.js";
import type { OperationNodeAttrs } from "../schema/node.js";
import type { OperationEdgeAttrs } from "../schema/edge.js";
import type { ValidationError, AnyValidationError } from "../error/index.js";
import { GraphologyHostConfig } from "../host/graphology.js";
import { reachableFrom } from "../graph/queries.js";
function getRequiredTopLevelFields(schema: unknown): Set<string> {
const fields = new Set<string>();
if (schema === null || schema === undefined || typeof schema !== "object") return fields;
const s = schema as TSchema;
if (!KindGuard.IsObject(s)) return fields;
const props = s.properties as Record<string, TSchema> | undefined;
const required = s.required as string[] | undefined;
if (props && required) {
for (const key of required) {
fields.add(key);
}
}
return fields;
}
function getProvidedFields(schema: unknown): Set<string> {
const fields = new Set<string>();
if (schema === null || schema === undefined || typeof schema !== "object") return fields;
const s = schema as TSchema;
if (!KindGuard.IsObject(s)) return fields;
const props = s.properties as Record<string, TSchema> | undefined;
if (props) {
for (const key of Object.keys(props)) {
fields.add(key);
}
}
return fields;
}
export function validatePreconditions(
graph: FlowGraph<typeof import("../schema/node.js").OperationNodeAttrs, typeof import("../schema/edge.js").OperationEdgeAttrs>,
): ValidationError[] {
const errors: ValidationError[] = [];
const nodeKeys = graph.nodes();
for (const nodeKey of nodeKeys) {
const attrs = graph.getNodeAttributes(nodeKey) as unknown as OperationNodeAttrs;
const inputSchema = attrs.inputSchema;
const requiredFields = getRequiredTopLevelFields(inputSchema);
if (requiredFields.size === 0) continue;
const predecessors = graph.predecessors(nodeKey);
if (predecessors.length === 0) {
for (const field of requiredFields) {
errors.push({
type: "schema",
nodeKey,
field,
message: `Required input field "${field}" has no predecessor providing it`,
});
}
continue;
}
const providedFields = new Set<string>();
for (const predKey of predecessors) {
const predAttrs = graph.getNodeAttributes(predKey) as unknown as OperationNodeAttrs;
const predProvided = getProvidedFields(predAttrs.outputSchema);
for (const field of predProvided) {
providedFields.add(field);
}
}
for (const field of requiredFields) {
if (!providedFields.has(field)) {
errors.push({
type: "schema",
nodeKey,
field,
message: `Required input field "${field}" is not provided by any predecessor`,
});
}
}
}
return errors;
}
function collectOperationNodeKeys(dag: DirectedGraph): string[] {
const names: string[] = [];
dag.forEachNode((key) => {
if (!key.startsWith("__")) {
names.push(key);
}
});
return names;
}
export function validateTemplate(
template: UNode,
operationGraph: FlowGraph<typeof import("../schema/node.js").OperationNodeAttrs, typeof import("../schema/edge.js").OperationEdgeAttrs>,
): AnyValidationError[] {
const errors: AnyValidationError[] = [];
let renderedDag: DirectedGraph;
try {
const root = createHostRoot(GraphologyHostConfig, null);
root.render(template);
renderedDag = root.ctx.graph as DirectedGraph;
} catch {
renderedDag = new DirectedGraph();
}
const templateNodeKeys = collectOperationNodeKeys(renderedDag);
const graphNodeKeys = new Set(operationGraph.nodes());
for (const opKey of templateNodeKeys) {
if (!graphNodeKeys.has(opKey)) {
errors.push({
type: "graph",
category: "orphan-node",
details: { operation: opKey, message: `Operation "${opKey}" not found in operation graph` },
});
}
}
if (hasCycle(renderedDag)) {
errors.push({
type: "graph",
category: "cycle",
details: { message: "Rendered template DAG contains a cycle" },
});
}
for (const opKey of templateNodeKeys) {
if (!graphNodeKeys.has(opKey)) continue;
const outEdges = renderedDag.outEdges(opKey) ?? [];
for (const edge of outEdges) {
const target = renderedDag.target(edge);
if (target.startsWith("__")) continue;
if (!graphNodeKeys.has(target)) continue;
if (operationGraph.hasEdge(opKey, target)) {
const edgeAttrs = operationGraph.getEdgeAttributes(opKey, target) as unknown as OperationEdgeAttrs;
if (!edgeAttrs.compatible) {
errors.push({
type: "type-compat",
sourceKey: opKey,
targetKey: target,
compatible: false,
mismatches: edgeAttrs.mismatches ?? [],
});
}
}
}
}
if (templateNodeKeys.length > 1) {
const roots: string[] = [];
for (const key of templateNodeKeys) {
const inDegree = renderedDag.inDegree(key);
if (inDegree === 0) {
roots.push(key);
}
}
if (roots.length > 0) {
const reachable = reachableFrom(renderedDag, roots);
for (const nodeKey of templateNodeKeys) {
if (!reachable.has(nodeKey)) {
errors.push({
type: "graph",
category: "orphan-node",
details: { nodeKey, message: `Operation "${nodeKey}" is not reachable from start` },
});
}
}
}
}
for (const nodeKey of templateNodeKeys) {
const inDegree = renderedDag.inDegree(nodeKey);
const outDegree = renderedDag.outDegree(nodeKey);
if (inDegree === 0 && outDegree === 0 && templateNodeKeys.length > 1) {
errors.push({
type: "graph",
category: "orphan-node",
details: { nodeKey, message: `Operation "${nodeKey}" has no edges (orphan node)` },
});
}
}
return errors;
}

View File

@@ -8,6 +8,7 @@ import {
NodeNotFoundError,
CycleError,
InvalidInputError,
InvalidTransitionError,
} from "../error/index.js";
import type { CallStatus, AnyValidationError, ValidationError } from "../error/index.js";
import {
@@ -20,9 +21,13 @@ import {
OperationEdgeAttrs as OperationEdgeAttrsSchema,
OperationGraphSerialized,
CallGraphSerialized,
CallNodeAttrs as CallNodeAttrsSchema,
CallEdgeAttrs as CallEdgeAttrsSchema,
} from "../schema/index.js";
import type { OperationNodeAttrs, FlowGraphSerialized } from "../schema/index.js";
import { typeCompat, type TypeCompatResult } from "../analysis/type-compat.js";
import type { FlowGraphSerialized, CallNodeAttrs } from "../schema/index.js";
import { buildTypeEdges, type TypeCompatResult } from "../analysis/type-compat.js";
export { buildTypeEdges } from "../analysis/type-compat.js";
export interface FlowGraphOptions {
type?: "directed";
@@ -41,8 +46,55 @@ export interface OperationSpec {
tags?: string[];
}
export interface CallRequestedEvent {
type: "call.requested";
requestId: string;
operationId: string;
input: unknown;
timestamp: string;
parentRequestId?: string;
identity?: { id: string; scopes: string[]; resources?: Record<string, string[]> };
startedAt?: string;
}
export interface CallRespondedEvent {
type: "call.responded";
requestId: string;
output: unknown;
timestamp: string;
}
export interface CallErrorEvent {
type: "call.error";
requestId: string;
error: { code: string; message: string; details?: unknown };
timestamp: string;
}
export interface CallAbortedEvent {
type: "call.aborted";
requestId: string;
timestamp: string;
}
export interface CallCompletedEvent {
type: "call.completed";
requestId: string;
output?: unknown;
timestamp: string;
}
export type CallEventMapValue =
| CallRequestedEvent
| CallRespondedEvent
| CallErrorEvent
| CallAbortedEvent
| CallCompletedEvent;
type OperationGraph = FlowGraph<typeof OperationNodeAttrsSchema, typeof OperationEdgeAttrsSchema>;
type CallGraph = FlowGraph<typeof CallNodeAttrsSchema, typeof CallEdgeAttrsSchema>;
type TypedEdgeAttrs = {
edgeType: "typed";
compatible: boolean;
@@ -52,6 +104,14 @@ type TypedEdgeAttrs = {
type Attrs = Record<string, unknown>;
const VALID_TRANSITIONS: Record<CallStatus, CallStatus[]> = {
pending: ["running", "aborted"],
running: ["completed", "failed", "aborted"],
completed: [],
failed: [],
aborted: [],
};
export class FlowGraph<
NodeAttrs extends TSchema = TSchema,
EdgeAttrs extends TSchema = TSchema,
@@ -318,6 +378,145 @@ export class FlowGraph<
return chain;
}
updateFromEvent(event: CallEventMapValue): void {
switch (event.type) {
case "call.requested": {
const attrs: CallNodeAttrs = {
requestId: event.requestId,
operationId: event.operationId,
status: "pending",
input: event.input,
...(event.parentRequestId !== undefined ? { parentRequestId: event.parentRequestId } : {}),
...(event.identity !== undefined ? { identity: event.identity } : {}),
...(event.startedAt !== undefined ? { startedAt: event.startedAt } : {}),
};
this.addCall(attrs);
break;
}
case "call.responded": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed" || currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "completed",
output: event.output,
completedAt: event.timestamp,
} as Attrs);
break;
}
case "call.error": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed" || currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "failed",
error: event.error,
completedAt: event.timestamp,
} as Attrs);
break;
}
case "call.aborted": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed" || currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "aborted",
completedAt: event.timestamp,
} as Attrs);
break;
}
case "call.completed": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed") {
if (!current.completedAt) {
this._graph.mergeNodeAttributes(event.requestId, { completedAt: event.timestamp } as Attrs);
}
return;
}
if (currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "completed",
...(event.output !== undefined ? { output: event.output } : {}),
completedAt: event.timestamp,
} as Attrs);
break;
}
}
}
addCall(attrs: CallNodeAttrs): void {
if (this._graph.hasNode(attrs.requestId)) return;
this._graph.addNode(attrs.requestId, attrs as Attrs);
if (attrs.parentRequestId !== undefined) {
if (this._graph.hasNode(attrs.parentRequestId)) {
if (willCreateCycle(this._graph, attrs.parentRequestId, attrs.requestId)) {
this._graph.dropNode(attrs.requestId);
const path = this._findPath(attrs.requestId, attrs.parentRequestId);
const cycle = [attrs.parentRequestId, ...path, attrs.parentRequestId];
throw new CycleError([cycle]);
}
const edgeKey = this._edgeKey(attrs.parentRequestId, attrs.requestId);
this._graph.addEdgeWithKey(edgeKey, attrs.parentRequestId, attrs.requestId, { edgeType: "triggered" } as Attrs);
}
}
}
addDependency(source: string, target: string): void {
if (!this._graph.hasNode(source)) {
throw new NodeNotFoundError(source);
}
if (!this._graph.hasNode(target)) {
throw new NodeNotFoundError(target);
}
const edgeKey = `${source}->${target}:depends_on`;
if (this._graph.hasEdge(edgeKey)) return;
if (willCreateCycle(this._graph, source, target)) {
const path = this._findPath(target, source);
const cycle = [source, ...path, source];
throw new CycleError([cycle]);
}
this._graph.addEdgeWithKey(edgeKey, source, target, { edgeType: "depends_on" } as Attrs);
}
updateStatus(requestId: string, status: CallStatus, extra?: Partial<CallNodeAttrs>): void {
if (!this._graph.hasNode(requestId)) {
throw new NodeNotFoundError(requestId);
}
const current = this._graph.getNodeAttributes(requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === status) return;
const allowed = VALID_TRANSITIONS[currentStatus];
if (!allowed || !allowed.includes(status)) {
throw new InvalidTransitionError(requestId, currentStatus, status);
}
const update: Record<string, unknown> = { status };
if (extra) {
for (const [key, value] of Object.entries(extra)) {
if (value !== undefined) {
update[key] = value;
}
}
}
this._graph.mergeNodeAttributes(requestId, update as Attrs);
}
updateCall(requestId: string, attrs: Partial<CallNodeAttrs>): void {
if (!this._graph.hasNode(requestId)) {
throw new NodeNotFoundError(requestId);
}
this._graph.mergeNodeAttributes(requestId, attrs as Attrs);
}
removeCall(requestId: string): void {
if (!this._graph.hasNode(requestId)) return;
this._graph.dropNode(requestId);
}
validate(schema: TSchema): AnyValidationError[] {
return _validate(this, schema as NodeAttrs);
}
@@ -365,10 +564,12 @@ export class FlowGraph<
return graph;
}
static fromCallEvents(
_events: unknown[],
): FlowGraph<TSchema, TSchema> {
throw new Error("not implemented");
static fromCallEvents(events: CallEventMapValue[]): CallGraph {
const graph = new FlowGraph<typeof CallNodeAttrsSchema, typeof CallEdgeAttrsSchema>();
for (const event of events) {
graph.updateFromEvent(event);
}
return graph;
}
export(): FlowGraphSerialized {
@@ -456,24 +657,3 @@ export class FlowGraph<
return [];
}
}
export function buildTypeEdges(graph: OperationGraph): void {
const nodeKeys = graph.nodes();
for (const source of nodeKeys) {
for (const target of nodeKeys) {
if (source === target) continue;
const sourceAttrs = graph.getNodeAttributes(source as never) as unknown as OperationNodeAttrs;
const targetAttrs = graph.getNodeAttributes(target as never) as unknown as OperationNodeAttrs;
const result = typeCompat(sourceAttrs.outputSchema as TSchema, targetAttrs.inputSchema as TSchema);
if (result === undefined) continue;
if (graph.hasEdge(source, target)) continue;
if (willCreateCycle(graph.graph, source, target)) continue;
const detail = result.detail ?? `${sourceAttrs.namespace}.${sourceAttrs.name}.output → ${targetAttrs.namespace}.${targetAttrs.name}.input`;
graph.addTypedEdge(source, target, {
compatible: result.compatible,
detail,
...(result.mismatches !== undefined ? { mismatches: result.mismatches } : {}),
});
}
}
}

View File

@@ -1,4 +1,4 @@
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec } from "./construction.js";
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec, type CallEventMapValue, type CallRequestedEvent, type CallRespondedEvent, type CallErrorEvent, type CallAbortedEvent, type CallCompletedEvent } from "./construction.js";
export {
topologicalOrder,
hasCycles,

View File

@@ -1,6 +1,7 @@
export * from "./error/index.js";
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec } from "./graph/index.js";
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec, type CallEventMapValue, type CallRequestedEvent, type CallRespondedEvent, type CallErrorEvent, type CallAbortedEvent, type CallCompletedEvent } from "./graph/index.js";
export { typeCompat, type TypeCompatResult, type TypeMismatch } from "./analysis/type-compat.js";
export {
validateSchema,
validateGraph,

View File

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

View File

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

View File

@@ -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";
@@ -88,6 +95,7 @@ export class WorkflowReactiveRoot implements EventLogProjection {
blockedByFailure: Map<string, ReadonlySignal<boolean>>;
resultMap: Map<string, ReadonlySignal<CallResult | undefined>>;
nodeKeyToRequestId: Map<string, string>;
requestIdToNodeKey: Map<string, string>;
private graph: DirectedGraph;
private effectDisposers: (() => void)[];
@@ -106,34 +114,33 @@ export class WorkflowReactiveRoot implements EventLogProjection {
this.effectDisposers = [];
this.eventLog = [];
this.nodeKeyToRequestId = new Map();
this.requestIdToNodeKey = new Map();
this._failurePolicy = options?.failurePolicy ?? "continue-running";
this.initializeSignals();
}
setRequestId(nodeKey: string, requestId: string): void {
this.nodeKeyToRequestId.set(nodeKey, requestId);
this.requestIdToNodeKey.set(requestId, nodeKey);
}
private initializeSignals(): void {
for (const node of this.graph.nodes()) {
const predecessors: string[] = this.graph.inNeighbors(node) ?? [];
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(() => {
@@ -188,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);
}
}
@@ -213,15 +210,29 @@ export class WorkflowReactiveRoot implements EventLogProjection {
if (!("requestId" in event)) return;
const nodeId = this.findNodeByRequestId(event.requestId);
let nodeId = this.requestIdToNodeKey.get(event.requestId);
if (nodeId === undefined) {
for (const [nId, rid] of this.nodeKeyToRequestId) {
if (rid === event.requestId) {
nodeId = nId;
this.requestIdToNodeKey.set(event.requestId, nId);
break;
}
}
}
if (nodeId === undefined) return;
const statusSignal = this.statusMap.get(nodeId);
if (!statusSignal) return;
const currentRequestId = this.nodeKeyToRequestId.get(nodeId);
if (currentRequestId === event.requestId) {
const statusSignal = this.statusMap.get(nodeId);
if (!statusSignal) return;
const derived = EVENT_TO_STATUS[event.type];
if (derived !== undefined) {
statusSignal.value = derived;
const derived = EVENT_TO_STATUS[event.type];
if (derived !== undefined) {
statusSignal.value = derived;
}
}
}
@@ -254,12 +265,17 @@ export class WorkflowReactiveRoot implements EventLogProjection {
}
getEvents(nodeId: string): CallEventMapValue[] {
const requestId = this.nodeKeyToRequestId.get(nodeId);
if (!requestId) return [];
const requestIds = new Set<string>();
for (const [rid, nId] of this.requestIdToNodeKey) {
if (nId === nodeId) {
requestIds.add(rid);
}
}
if (requestIds.size === 0) return [];
const events: CallEventMapValue[] = [];
for (const e of this.eventLog) {
if ("requestId" in e && e.requestId === requestId) {
if ("requestId" in e && requestIds.has(e.requestId)) {
events.push(e);
}
}
@@ -325,13 +341,7 @@ export class WorkflowReactiveRoot implements EventLogProjection {
this.blockedByFailure.clear();
this.resultMap.clear();
this.nodeKeyToRequestId.clear();
this.requestIdToNodeKey.clear();
this.eventLog = [];
}
private findNodeByRequestId(requestId: string): string | undefined {
for (const [nodeId, rid] of this.nodeKeyToRequestId) {
if (rid === requestId) return nodeId;
}
return undefined;
}
}

View File

@@ -1,7 +1,7 @@
---
id: analysis/build-type-edges
name: Implement buildTypeEdges — populate operation graph with type-compatibility edges
status: pending
status: completed
depends_on:
- analysis/type-compat
- graph/construction-operation

View File

@@ -1,7 +1,7 @@
---
id: analysis/ordering
name: Implement execution ordering functions (topologicalOrder, parallelGroups, criticalPath, reachableFrom)
status: pending
status: completed
depends_on:
- graph/flowgraph-class
- graph/queries

View File

@@ -1,7 +1,7 @@
---
id: analysis/template-validation
name: Implement validateTemplate and validatePreconditions
status: pending
status: completed
depends_on:
- analysis/type-compat
- graph/queries

View File

@@ -1,7 +1,7 @@
---
id: graph/construction-call
name: Implement call graph construction (fromCallEvents, updateFromEvent, addCall, addDependency, updateStatus)
status: pending
status: completed
depends_on:
- graph/flowgraph-class
- schema/enums

View File

@@ -0,0 +1,270 @@
import { describe, it, expect } from "vitest";
import { FlowGraph } from "../../src/graph/construction.js";
import {
topologicalOrder,
parallelGroups,
criticalPath,
reachableFrom,
ancestors,
descendants,
} from "../../src/analysis/ordering.js";
import { CycleError, NodeNotFoundError } from "../../src/error/index.js";
function buildDiamondDag(): FlowGraph {
const fg = new FlowGraph();
fg.addNode("top", { name: "top" });
fg.addNode("left", { name: "left" });
fg.addNode("right", { name: "right" });
fg.addNode("bottom", { name: "bottom" });
fg.addEdge("top", "left");
fg.addEdge("top", "right");
fg.addEdge("left", "bottom");
fg.addEdge("right", "bottom");
return fg;
}
function buildChainDag(): FlowGraph {
const fg = new FlowGraph();
fg.addNode("a", { name: "a" });
fg.addNode("b", { name: "b" });
fg.addNode("c", { name: "c" });
fg.addNode("d", { name: "d" });
fg.addEdge("a", "b");
fg.addEdge("b", "c");
fg.addEdge("c", "d");
return fg;
}
function buildWideDag(): FlowGraph {
const fg = new FlowGraph();
fg.addNode("root", { name: "root" });
fg.addNode("x", { name: "x" });
fg.addNode("y", { name: "y" });
fg.addNode("z", { name: "z" });
fg.addEdge("root", "x");
fg.addEdge("root", "y");
fg.addEdge("root", "z");
return fg;
}
function buildDisconnectedDag(): FlowGraph {
const fg = new FlowGraph();
fg.addNode("a", { name: "a" });
fg.addNode("b", { name: "b" });
fg.addNode("c", { name: "c" });
fg.addEdge("a", "b");
return fg;
}
describe("topologicalOrder (analysis)", () => {
it("returns topological order for a chain", () => {
const fg = buildChainDag();
expect(topologicalOrder(fg)).toEqual(["a", "b", "c", "d"]);
});
it("returns topological order for a diamond DAG", () => {
const fg = buildDiamondDag();
const order = topologicalOrder(fg);
expect(order[0]).toBe("top");
expect(order[3]).toBe("bottom");
expect(order).toContain("left");
expect(order).toContain("right");
});
it("returns empty array for empty graph", () => {
const fg = new FlowGraph();
expect(topologicalOrder(fg)).toEqual([]);
});
it("returns single node for graph with one node", () => {
const fg = new FlowGraph();
fg.addNode("only", { name: "only" });
expect(topologicalOrder(fg)).toEqual(["only"]);
});
it("throws CycleError when graph has cycles", () => {
const fg = new FlowGraph();
fg.graph.addNode("a");
fg.graph.addNode("b");
fg.graph.addNode("c");
fg.graph.addEdgeWithKey("a->b", "a", "b");
fg.graph.addEdgeWithKey("b->c", "b", "c");
fg.graph.addEdgeWithKey("c->a", "c", "a");
expect(() => topologicalOrder(fg)).toThrow(CycleError);
});
});
describe("parallelGroups (analysis)", () => {
it("groups chain nodes into sequential groups", () => {
const fg = buildChainDag();
const groups = parallelGroups(fg);
expect(groups).toEqual([["a"], ["b"], ["c"], ["d"]]);
});
it("groups diamond DAG correctly", () => {
const fg = buildDiamondDag();
const groups = parallelGroups(fg);
expect(groups.length).toBe(3);
expect(groups[0]).toEqual(["top"]);
expect(groups[1]!.sort()).toEqual(["left", "right"]);
expect(groups[2]).toEqual(["bottom"]);
});
it("groups wide DAG with root and leaves", () => {
const fg = buildWideDag();
const groups = parallelGroups(fg);
expect(groups.length).toBe(2);
expect(groups[0]).toEqual(["root"]);
expect(groups[1]!.sort()).toEqual(["x", "y", "z"]);
});
it("groups disconnected DAG correctly", () => {
const fg = buildDisconnectedDag();
const groups = parallelGroups(fg);
expect(groups.length).toBe(2);
expect(groups[0]!.sort()).toEqual(["a", "c"]);
expect(groups[1]).toEqual(["b"]);
});
it("returns empty array for empty graph", () => {
const fg = new FlowGraph();
expect(parallelGroups(fg)).toEqual([]);
});
it("returns single group for single node with no edges", () => {
const fg = new FlowGraph();
fg.addNode("only", { name: "only" });
expect(parallelGroups(fg)).toEqual([["only"]]);
});
it("throws CycleError when graph has cycles", () => {
const fg = new FlowGraph();
fg.graph.addNode("a");
fg.graph.addNode("b");
fg.graph.addEdgeWithKey("a->b", "a", "b");
fg.graph.addEdgeWithKey("b->a", "b", "a");
expect(() => parallelGroups(fg)).toThrow(CycleError);
});
});
describe("criticalPath (analysis)", () => {
it("returns full chain for a chain DAG", () => {
const fg = buildChainDag();
expect(criticalPath(fg)).toEqual(["a", "b", "c", "d"]);
});
it("returns a longest path for diamond DAG", () => {
const fg = buildDiamondDag();
const path = criticalPath(fg);
expect(path.length).toBe(3);
expect(path[0]).toBe("top");
expect(path[2]).toBe("bottom");
expect(path[1] === "left" || path[1] === "right").toBe(true);
});
it("returns single node for graph with one node", () => {
const fg = new FlowGraph();
fg.addNode("only", { name: "only" });
expect(criticalPath(fg)).toEqual(["only"]);
});
it("returns empty array for empty graph", () => {
const fg = new FlowGraph();
expect(criticalPath(fg)).toEqual([]);
});
it("returns correct path for wide DAG", () => {
const fg = buildWideDag();
const path = criticalPath(fg);
expect(path.length).toBe(2);
expect(path[0]).toBe("root");
});
it("throws CycleError when graph has cycles", () => {
const fg = new FlowGraph();
fg.graph.addNode("a");
fg.graph.addNode("b");
fg.graph.addEdgeWithKey("a->b", "a", "b");
fg.graph.addEdgeWithKey("b->a", "b", "a");
expect(() => criticalPath(fg)).toThrow(CycleError);
});
});
describe("reachableFrom (analysis)", () => {
it("returns all reachable nodes from a single start node", () => {
const fg = buildChainDag();
expect(reachableFrom(fg, ["a"])).toEqual(new Set(["a", "b", "c", "d"]));
});
it("returns only the start node if it is a leaf", () => {
const fg = buildChainDag();
expect(reachableFrom(fg, ["d"])).toEqual(new Set(["d"]));
});
it("returns union of reachable nodes from multiple start nodes", () => {
const fg = buildDiamondDag();
expect(reachableFrom(fg, ["left", "right"])).toEqual(new Set(["left", "right", "bottom"]));
});
it("returns empty set for empty input", () => {
const fg = buildChainDag();
expect(reachableFrom(fg, [])).toEqual(new Set());
});
it("returns full reachable set from root in diamond", () => {
const fg = buildDiamondDag();
expect(reachableFrom(fg, ["top"])).toEqual(new Set(["top", "left", "right", "bottom"]));
});
});
describe("ancestors (analysis)", () => {
it("returns all ancestors for a node in a chain", () => {
const fg = buildChainDag();
expect(ancestors(fg, "d")).toEqual(["c", "b", "a"]);
});
it("returns all ancestors for a node in a diamond DAG", () => {
const fg = buildDiamondDag();
const anc = ancestors(fg, "bottom");
expect(anc).toContain("left");
expect(anc).toContain("right");
expect(anc).toContain("top");
expect(anc.length).toBe(3);
});
it("returns empty array for a root node", () => {
const fg = buildChainDag();
expect(ancestors(fg, "a")).toEqual([]);
});
it("throws NodeNotFoundError for missing node", () => {
const fg = new FlowGraph();
expect(() => ancestors(fg, "missing")).toThrow(NodeNotFoundError);
});
});
describe("descendants (analysis)", () => {
it("returns all descendants for a node in a chain", () => {
const fg = buildChainDag();
expect(descendants(fg, "a")).toEqual(["b", "c", "d"]);
});
it("returns all descendants for a node in a diamond DAG", () => {
const fg = buildDiamondDag();
const desc = descendants(fg, "top");
expect(desc).toContain("left");
expect(desc).toContain("right");
expect(desc).toContain("bottom");
expect(desc.length).toBe(3);
});
it("returns empty array for a leaf node", () => {
const fg = buildChainDag();
expect(descendants(fg, "d")).toEqual([]);
});
it("throws NodeNotFoundError for missing node", () => {
const fg = new FlowGraph();
expect(() => descendants(fg, "missing")).toThrow(NodeNotFoundError);
});
});

View File

@@ -1,6 +1,8 @@
import { describe, it, expect } from "vitest";
import { Type, type TSchema } from "@alkdev/typebox";
import { typeCompat, type TypeCompatResult, type TypeMismatch } from "../../src/analysis/type-compat.js";
import { typeCompat, buildTypeEdges, type TypeCompatResult, type TypeMismatch } from "../../src/analysis/type-compat.js";
import { FlowGraph } from "../../src/graph/construction.js";
import type { OperationSpec } from "../../src/graph/construction.js";
describe("typeCompat", () => {
describe("exact match", () => {
@@ -412,3 +414,110 @@ describe("typeCompat", () => {
});
});
});
describe("buildTypeEdges", () => {
it("adds compatible edges for matching output→input schemas", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "extract", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ raw: Type.String() }), outputSchema: Type.Object({ text: Type.String() }) });
fg.addOperation({ name: "classify", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ text: Type.String() }), outputSchema: Type.Object({ label: Type.String(), score: Type.Number() }) });
buildTypeEdges(fg);
expect(fg.hasEdge("task.extract", "task.classify")).toBe(true);
const attrs = fg.getEdgeAttributes("task.extract", "task.classify") as Record<string, unknown>;
expect(attrs.edgeType).toBe("typed");
expect(attrs.compatible).toBe(true);
});
it("adds incompatible edges when schemas mismatch", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "classify", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ text: Type.String() }), outputSchema: Type.Object({ label: Type.String(), score: Type.Number() }) });
fg.addOperation({ name: "count", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ count: Type.Number() }), outputSchema: Type.Object({ result: Type.Number() }) });
buildTypeEdges(fg);
expect(fg.hasEdge("task.classify", "task.count")).toBe(true);
const attrs = fg.getEdgeAttributes("task.classify", "task.count") as Record<string, unknown>;
expect(attrs.edgeType).toBe("typed");
expect(attrs.compatible).toBe(false);
expect(attrs.mismatches).toBeDefined();
});
it("incompatible edges include mismatches array", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "a", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ x: Type.String() }), outputSchema: Type.Object({ value: Type.String() }) });
fg.addOperation({ name: "b", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ value: Type.Number() }), outputSchema: Type.Object({ z: Type.Boolean() }) });
buildTypeEdges(fg);
const attrs = fg.getEdgeAttributes("op.a", "op.b") as Record<string, unknown>;
expect(attrs.compatible).toBe(false);
expect(Array.isArray(attrs.mismatches)).toBe(true);
expect((attrs.mismatches as Array<unknown>).length).toBeGreaterThan(0);
});
it("does not add edges when either schema is Unknown", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "unk_out", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ x: Type.String() }), outputSchema: Type.Unknown() });
fg.addOperation({ name: "unk_in", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Unknown(), outputSchema: Type.Object({ y: Type.String() }) });
fg.addOperation({ name: "normal", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ y: Type.String() }), outputSchema: Type.Object({ x: Type.String() }) });
buildTypeEdges(fg);
expect(fg.hasEdge("op.unk_out", "op.unk_in")).toBe(false);
expect(fg.hasEdge("op.unk_out", "op.normal")).toBe(false);
expect(fg.hasEdge("op.normal", "op.unk_in")).toBe(false);
});
it("sets detail to namespace.name.output → namespace.name.input for compatible edges", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "extract", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ raw: Type.String() }), outputSchema: Type.Object({ text: Type.String() }) });
fg.addOperation({ name: "classify", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ text: Type.String() }), outputSchema: Type.Object({ label: Type.String() }) });
buildTypeEdges(fg);
const attrs = fg.getEdgeAttributes("task.extract", "task.classify") as Record<string, unknown>;
expect(attrs.detail).toContain("task.extract.output → task.classify.input");
});
it("is callable after incremental addOperation calls", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "extract", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ raw: Type.String() }), outputSchema: Type.Object({ text: Type.String() }) });
buildTypeEdges(fg);
expect(fg.size).toBe(0);
fg.addOperation({ name: "classify", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ text: Type.String() }), outputSchema: Type.Object({ label: Type.String() }) });
buildTypeEdges(fg);
expect(fg.hasEdge("op.extract", "op.classify")).toBe(true);
});
it("produces edges for three operations in a pipeline", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "extract", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ raw: Type.String() }), outputSchema: Type.Object({ text: Type.String() }) });
fg.addOperation({ name: "classify", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ text: Type.String() }), outputSchema: Type.Object({ label: Type.String() }) });
fg.addOperation({ name: "enrich", namespace: "task", version: "1.0.0", type: "query", inputSchema: Type.Object({ label: Type.String() }), outputSchema: Type.Object({ enriched: Type.String() }) });
buildTypeEdges(fg);
expect(fg.hasEdge("task.extract", "task.classify")).toBe(true);
expect(fg.hasEdge("task.classify", "task.enrich")).toBe(true);
expect(fg.hasEdge("task.extract", "task.enrich")).toBe(true);
const e2c = fg.getEdgeAttributes("task.extract", "task.classify") as Record<string, unknown>;
const c2e = fg.getEdgeAttributes("task.classify", "task.enrich") as Record<string, unknown>;
const e2e = fg.getEdgeAttributes("task.extract", "task.enrich") as Record<string, unknown>;
expect(e2c.compatible).toBe(true);
expect(c2e.compatible).toBe(true);
expect(e2e.compatible).toBe(false);
});
it("does not add self-loops", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "a", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ x: Type.String() }), outputSchema: Type.Object({ x: Type.String() }) });
buildTypeEdges(fg);
expect(fg.size).toBe(0);
});
it("returns empty graph with no edges for empty graph", () => {
const fg = new FlowGraph();
buildTypeEdges(fg);
expect(fg.order).toBe(0);
expect(fg.size).toBe(0);
});
it("skips edges that would already exist", () => {
const fg = new FlowGraph();
fg.addOperation({ name: "a", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ x: Type.String() }), outputSchema: Type.Object({ y: Type.String() }) });
fg.addOperation({ name: "b", namespace: "op", version: "1.0.0", type: "query", inputSchema: Type.Object({ y: Type.String() }), outputSchema: Type.Object({ z: Type.String() }) });
buildTypeEdges(fg);
const sizeAfterFirst = fg.size;
buildTypeEdges(fg);
expect(fg.size).toBe(sizeAfterFirst);
});
});

View File

@@ -1,7 +1,251 @@
import { describe, it, expect } from 'vitest';
import { describe, it, expect } from "vitest";
import { Type } from "@alkdev/typebox";
import { h, createHostRoot } from "@alkdev/ujsx";
import { Operation, Sequential, Parallel, Conditional } from "../../src/component/index.js";
import { validatePreconditions, validateTemplate } from "../../src/analysis/workflow.js";
import { FlowGraph } from "../../src/graph/construction.js";
import type { OperationNodeAttrs, OperationEdgeAttrs } from "../../src/schema/index.js";
describe('analysis workflow', () => {
it('placeholder', () => {
expect(true).toBe(true);
type OpGraph = FlowGraph<typeof import("../../src/schema/node.js").OperationNodeAttrs, typeof import("../../src/schema/edge.js").OperationEdgeAttrs>;
function createOperationGraph(
specs: Array<{
name: string;
namespace?: string;
inputSchema?: Record<string, unknown>;
outputSchema?: Record<string, unknown>;
}>,
): OpGraph {
const graph = new FlowGraph() as OpGraph;
for (const spec of specs) {
const ns = spec.namespace ?? "test";
const key = `${ns}.${spec.name}`;
graph.addNode(key, {
name: spec.name,
namespace: ns,
version: "1.0.0",
type: "query",
inputSchema: spec.inputSchema ?? Type.Object({}),
outputSchema: spec.outputSchema ?? Type.Object({}),
} as OperationNodeAttrs);
}
return graph;
}
function createOperationGraphWithEdges(
specs: Array<{
name: string;
namespace?: string;
inputSchema?: Record<string, unknown>;
outputSchema?: Record<string, unknown>;
}>,
edges?: Array<{ source: string; target: string; compatible: boolean; mismatches?: Array<{ path: string; expected: string; actual: string }> }>,
): OpGraph {
const graph = createOperationGraph(specs);
if (edges) {
for (const edge of edges) {
graph.addTypedEdge(edge.source, edge.target, {
compatible: edge.compatible,
mismatches: edge.mismatches,
});
}
}
return graph;
}
describe("validatePreconditions", () => {
it("returns empty for valid graph with no required fields", () => {
const graph = createOperationGraph([
{ name: "a", outputSchema: Type.Object({ x: Type.Number() }) },
{ name: "b", inputSchema: Type.Object({}) },
]);
graph.addEdge("test.a", "test.b");
const errors = validatePreconditions(graph);
expect(errors).toEqual([]);
});
it("returns empty when all required input fields are provided by predecessors", () => {
const graph = createOperationGraph([
{ name: "a", outputSchema: Type.Object({ x: Type.Number(), y: Type.String() }) },
{ name: "b", inputSchema: Type.Object({ x: Type.Number() }) },
]);
graph.addEdge("test.a", "test.b");
const errors = validatePreconditions(graph);
expect(errors).toEqual([]);
});
it("returns errors when required input field is not provided by any predecessor", () => {
const graph = createOperationGraph([
{ name: "a", outputSchema: Type.Object({ x: Type.Number() }) },
{ name: "b", inputSchema: Type.Object({ x: Type.Number(), y: Type.String() }) },
]);
graph.addEdge("test.a", "test.b");
const errors = validatePreconditions(graph);
expect(errors.length).toBeGreaterThan(0);
const fieldNames = errors.map((e) => e.field);
expect(fieldNames).toContain("y");
});
it("returns errors when node with required fields has no predecessors", () => {
const graph = createOperationGraph([
{ name: "a", inputSchema: Type.Object({ x: Type.Number() }), outputSchema: Type.Object({}) },
]);
const errors = validatePreconditions(graph);
expect(errors.length).toBeGreaterThan(0);
expect(errors[0]!.message).toContain("no predecessor");
});
it("collects provided fields from multiple predecessors", () => {
const graph = createOperationGraph([
{ name: "a", outputSchema: Type.Object({ x: Type.Number() }) },
{ name: "c", outputSchema: Type.Object({ y: Type.String() }) },
{ name: "b", inputSchema: Type.Object({ x: Type.Number(), y: Type.String() }) },
]);
graph.addEdge("test.a", "test.b");
graph.addEdge("test.c", "test.b");
const errors = validatePreconditions(graph);
expect(errors).toEqual([]);
});
it("returns empty for graph with no nodes", () => {
const graph = new FlowGraph() as OpGraph;
const errors = validatePreconditions(graph);
expect(errors).toEqual([]);
});
});
describe("validateTemplate", () => {
it("returns empty for valid template with all operations in graph", () => {
const graph = createOperationGraphWithEdges([
{ name: "a", outputSchema: Type.Object({ x: Type.Number() }) },
{ name: "b", inputSchema: Type.Object({ x: Type.Number() }) },
], [
{ source: "test.a", target: "test.b", compatible: true },
]);
const template = h(Sequential, {},
h(Operation, { name: "test.a" }),
h(Operation, { name: "test.b" }),
);
const errors = validateTemplate(template, graph);
expect(errors).toEqual([]);
});
it("returns error when operation name is not in operation graph", () => {
const graph = createOperationGraph([
{ name: "a" },
]);
const template = h(Sequential, {},
h(Operation, { name: "test.a" }),
h(Operation, { name: "test.missing" }),
);
const errors = validateTemplate(template, graph);
expect(errors.length).toBeGreaterThan(0);
const missingErrors = errors.filter(
(e) => e.type === "graph" && (e as { type: string; category: string; details: unknown }).category === "orphan-node"
&& JSON.stringify((e as { details: unknown }).details).includes("missing"),
);
expect(missingErrors.length).toBeGreaterThan(0);
});
it("returns error for type incompatibility between sequential operations", () => {
const graph = createOperationGraphWithEdges([
{ name: "a", outputSchema: Type.Object({ x: Type.String() }) },
{ name: "b", inputSchema: Type.Object({ x: Type.Number() }) },
], [
{ source: "test.a", target: "test.b", compatible: false, mismatches: [{ path: "/x", expected: "number", actual: "string" }] },
]);
const template = h(Sequential, {},
h(Operation, { name: "test.a" }),
h(Operation, { name: "test.b" }),
);
const errors = validateTemplate(template, graph);
const typeErrors = errors.filter((e) => e.type === "type-compat");
expect(typeErrors.length).toBeGreaterThan(0);
});
it("returns empty for single-operation template", () => {
const graph = createOperationGraph([
{ name: "a" },
]);
const template = h(Operation, { name: "test.a" });
const errors = validateTemplate(template, graph);
expect(errors).toEqual([]);
});
it("detects unreachable nodes", () => {
const graph = createOperationGraphWithEdges([
{ name: "a" },
{ name: "b" },
{ name: "c" },
]);
const template = h(Sequential, {},
h(Operation, { name: "test.a" }),
h(Operation, { name: "test.b" }),
);
const errors = validateTemplate(template, graph);
const reachableErrors = errors.filter(
(e) => e.type === "graph" && JSON.stringify((e as { details: unknown }).details).includes("not reachable"),
);
expect(reachableErrors.length).toBe(0);
});
it("returns empty for valid parallel template", () => {
const graph = createOperationGraphWithEdges([
{ name: "a" },
{ name: "b" },
{ name: "c" },
]);
const template = h(Sequential, {},
h(Operation, { name: "test.a" }),
h(Parallel, {},
h(Operation, { name: "test.b" }),
h(Operation, { name: "test.c" }),
),
);
const errors = validateTemplate(template, graph);
expect(errors).toEqual([]);
});
it("handles template with conditional", () => {
const graph = createOperationGraphWithEdges([
{ name: "a", outputSchema: Type.Object({ result: Type.Boolean() }) },
{ name: "b" },
{ name: "c" },
]);
const template = h(Sequential, {},
h(Operation, { name: "test.a" }),
h(Conditional, { test: "test.a" },
h(Operation, { name: "test.b" }),
),
h(Operation, { name: "test.c" }),
);
const errors = validateTemplate(template, graph);
expect(errors).toEqual([]);
});
it("template validation is advisory - never throws", () => {
const graph = new FlowGraph() as OpGraph;
const template = h(Sequential, {},
h(Operation, { name: "nonexistent" }),
);
const errors = validateTemplate(template, graph);
expect(Array.isArray(errors)).toBe(true);
});
it("detects orphan node with no edges in multi-node template", () => {
const graph = createOperationGraphWithEdges([
{ name: "a" },
{ name: "b" },
]);
const template = h(Parallel, {},
h(Operation, { name: "test.a" }),
h(Operation, { name: "test.b" }),
);
const errors = validateTemplate(template, graph);
const orphanErrors = errors.filter(
(e) => e.type === "graph" && (e as { type: string; category: string }).category === "orphan-node"
&& JSON.stringify((e as { details: unknown }).details).includes("no edges"),
);
expect(orphanErrors.length).toBeGreaterThan(0);
});
});

View File

@@ -1,13 +1,15 @@
import { describe, it, expect } from "vitest";
import { Type } from "@alkdev/typebox";
import { FlowGraph, buildTypeEdges } from "../../src/graph/construction.js";
import type { OperationSpec } from "../../src/graph/construction.js";
import type { OperationSpec, CallEventMapValue } from "../../src/graph/construction.js";
import {
DuplicateNodeError,
DuplicateEdgeError,
NodeNotFoundError,
CycleError,
InvalidTransitionError,
} from "../../src/error/index.js";
import type { CallStatus } from "../../src/error/index.js";
describe("FlowGraph constructor", () => {
it("creates an empty graph", () => {
@@ -300,8 +302,13 @@ describe("FlowGraph query methods", () => {
});
describe("FlowGraph static stubs", () => {
it("fromCallEvents throws not implemented", () => {
expect(() => FlowGraph.fromCallEvents([])).toThrow("not implemented");
it("fromCallEvents returns empty graph for empty events", () => {
const graph = FlowGraph.fromCallEvents([]);
expect(graph.order).toBe(0);
expect(graph.size).toBe(0);
});
it("fromJSON throws not implemented", () => {
expect(() => FlowGraph.fromJSON({} as never)).toThrow();
});
});
@@ -635,3 +642,519 @@ describe("FlowGraph cycle detection", () => {
expect(() => fg.addEdge("b", "c")).not.toThrow();
});
});
describe("FlowGraph.fromCallEvents", () => {
const requestedEvent: CallEventMapValue = {
type: "call.requested",
requestId: "req-1",
operationId: "task.classify",
input: { text: "hello" },
timestamp: "2026-01-01T00:00:00Z",
};
const requestedWithParent: CallEventMapValue = {
type: "call.requested",
requestId: "req-2",
operationId: "task.enrich",
input: { label: "greeting" },
timestamp: "2026-01-01T00:00:01Z",
parentRequestId: "req-1",
};
const respondedEvent: CallEventMapValue = {
type: "call.responded",
requestId: "req-1",
output: { label: "greeting" },
timestamp: "2026-01-01T00:00:02Z",
};
const errorEvent: CallEventMapValue = {
type: "call.error",
requestId: "req-1",
error: { code: "INTERNAL", message: "Something went wrong" },
timestamp: "2026-01-01T00:00:03Z",
};
const abortedEvent: CallEventMapValue = {
type: "call.aborted",
requestId: "req-1",
timestamp: "2026-01-01T00:00:04Z",
};
const completedEvent: CallEventMapValue = {
type: "call.completed",
requestId: "req-1",
output: { label: "greeting" },
timestamp: "2026-01-01T00:00:05Z",
};
it("adds node from call.requested event", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent]);
expect(graph.order).toBe(1);
expect(graph.hasNode("req-1")).toBe(true);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("pending");
expect(attrs.operationId).toBe("task.classify");
});
it("creates triggered edge from parentRequestId", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent, requestedWithParent]);
expect(graph.order).toBe(2);
expect(graph.hasNode("req-2")).toBe(true);
expect(graph.hasEdge("req-1", "req-2")).toBe(true);
const edgeAttrs = graph.getEdgeAttributes("req-1", "req-2") as Record<string, unknown>;
expect(edgeAttrs.edgeType).toBe("triggered");
});
it("updates status to completed on call.responded", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent, respondedEvent]);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
expect(attrs.output).toEqual({ label: "greeting" });
expect(attrs.completedAt).toBe("2026-01-01T00:00:02Z");
});
it("updates status to failed on call.error", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent, errorEvent]);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("failed");
expect(attrs.error).toEqual({ code: "INTERNAL", message: "Something went wrong" });
expect(attrs.completedAt).toBe("2026-01-01T00:00:03Z");
});
it("updates status to aborted on call.aborted", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent, abortedEvent]);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("aborted");
expect(attrs.completedAt).toBe("2026-01-01T00:00:04Z");
});
it("updates status to completed on call.completed", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent, completedEvent]);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
expect(attrs.completedAt).toBe("2026-01-01T00:00:05Z");
});
it("is idempotent — duplicate events have no effect", () => {
const graph = FlowGraph.fromCallEvents([requestedEvent, requestedEvent, respondedEvent, respondedEvent]);
expect(graph.order).toBe(1);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
});
it("ignores responded/error/aborted for unknown requestId", () => {
const graph = FlowGraph.fromCallEvents([respondedEvent, errorEvent, abortedEvent]);
expect(graph.order).toBe(0);
});
it("creates node for unknown operationId", () => {
const unknownOpEvent: CallEventMapValue = {
type: "call.requested",
requestId: "req-unknown",
operationId: "unknown.op",
input: {},
timestamp: "2026-01-01T00:00:00Z",
};
const graph = FlowGraph.fromCallEvents([unknownOpEvent]);
expect(graph.order).toBe(1);
const attrs = graph.getNodeAttributes("req-unknown") as Record<string, unknown>;
expect(attrs.status).toBe("pending");
expect(attrs.operationId).toBe("unknown.op");
});
it("processes full event sequence", () => {
const req1: CallEventMapValue = {
type: "call.requested",
requestId: "req-parent",
operationId: "task.parent",
input: {},
timestamp: "2026-01-01T00:00:00Z",
};
const req2: CallEventMapValue = {
type: "call.requested",
requestId: "req-child",
operationId: "task.child",
input: {},
timestamp: "2026-01-01T00:00:01Z",
parentRequestId: "req-parent",
};
const resp: CallEventMapValue = {
type: "call.responded",
requestId: "req-parent",
output: "done",
timestamp: "2026-01-01T00:00:02Z",
};
const graph = FlowGraph.fromCallEvents([req1, req2, resp]);
expect(graph.order).toBe(2);
expect(graph.hasEdge("req-parent", "req-child")).toBe(true);
const parentAttrs = graph.getNodeAttributes("req-parent") as Record<string, unknown>;
expect(parentAttrs.status).toBe("completed");
const childAttrs = graph.getNodeAttributes("req-child") as Record<string, unknown>;
expect(childAttrs.status).toBe("pending");
});
it("stores identity and startedAt from call.requested", () => {
const event: CallEventMapValue = {
type: "call.requested",
requestId: "req-id",
operationId: "task.op",
input: {},
timestamp: "2026-01-01T00:00:00Z",
identity: { id: "user-1", scopes: ["read"] },
startedAt: "2026-01-01T00:00:01Z",
};
const graph = FlowGraph.fromCallEvents([event]);
const attrs = graph.getNodeAttributes("req-id") as Record<string, unknown>;
expect(attrs.identity).toEqual({ id: "user-1", scopes: ["read"] });
expect(attrs.startedAt).toBe("2026-01-01T00:00:01Z");
});
it("skips triggered edge if parent node does not exist", () => {
const orphanEvent: CallEventMapValue = {
type: "call.requested",
requestId: "req-orphan",
operationId: "task.child",
input: {},
timestamp: "2026-01-01T00:00:00Z",
parentRequestId: "req-nonexistent",
};
const graph = FlowGraph.fromCallEvents([orphanEvent]);
expect(graph.order).toBe(1);
expect(graph.hasNode("req-orphan")).toBe(true);
expect(graph.size).toBe(0);
});
});
describe("FlowGraph.updateFromEvent", () => {
it("processes single event for real-time pattern", () => {
const graph = new FlowGraph();
graph.updateFromEvent({
type: "call.requested",
requestId: "req-1",
operationId: "task.classify",
input: { text: "hello" },
timestamp: "2026-01-01T00:00:00Z",
});
expect(graph.hasNode("req-1")).toBe(true);
graph.updateFromEvent({
type: "call.responded",
requestId: "req-1",
output: { label: "hi" },
timestamp: "2026-01-01T00:00:02Z",
});
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
expect(attrs.output).toEqual({ label: "hi" });
});
it("ignores events for unknown requestId", () => {
const graph = new FlowGraph();
graph.updateFromEvent({
type: "call.responded",
requestId: "unknown",
output: "x",
timestamp: "2026-01-01T00:00:00Z",
});
expect(graph.order).toBe(0);
});
it("ignores terminal event re-processing", () => {
const graph = new FlowGraph();
graph.updateFromEvent({
type: "call.requested",
requestId: "req-1",
operationId: "task.op",
input: {},
timestamp: "2026-01-01T00:00:00Z",
});
graph.updateFromEvent({
type: "call.responded",
requestId: "req-1",
output: "done",
timestamp: "2026-01-01T00:00:01Z",
});
graph.updateFromEvent({
type: "call.error",
requestId: "req-1",
error: { code: "X", message: "Y" },
timestamp: "2026-01-01T00:00:02Z",
});
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
});
});
describe("FlowGraph.addCall", () => {
it("adds a call node", () => {
const graph = new FlowGraph();
graph.addCall({
requestId: "req-1",
operationId: "task.classify",
status: "pending",
input: { text: "hello" },
});
expect(graph.hasNode("req-1")).toBe(true);
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("pending");
});
it("adds triggered edge when parentRequestId is present", () => {
const graph = new FlowGraph();
graph.addCall({
requestId: "req-parent",
operationId: "task.parent",
status: "pending",
input: {},
});
graph.addCall({
requestId: "req-child",
operationId: "task.child",
status: "pending",
input: {},
parentRequestId: "req-parent",
});
expect(graph.hasEdge("req-parent", "req-child")).toBe(true);
const edgeAttrs = graph.getEdgeAttributes("req-parent", "req-child") as Record<string, unknown>;
expect(edgeAttrs.edgeType).toBe("triggered");
});
it("is idempotent — duplicate addCall is ignored", () => {
const graph = new FlowGraph();
graph.addCall({
requestId: "req-1",
operationId: "task.op",
status: "pending",
input: {},
});
graph.addCall({
requestId: "req-1",
operationId: "task.op",
status: "pending",
input: {},
});
expect(graph.order).toBe(1);
});
it("does not throw if parentRequestId node does not exist", () => {
const graph = new FlowGraph();
graph.addCall({
requestId: "req-child",
operationId: "task.child",
status: "pending",
input: {},
parentRequestId: "nonexistent",
});
expect(graph.hasNode("req-child")).toBe(true);
expect(graph.size).toBe(0);
});
});
describe("FlowGraph.addDependency", () => {
it("creates depends_on edge", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.a", status: "pending", input: {} });
graph.addCall({ requestId: "req-2", operationId: "task.b", status: "pending", input: {} });
graph.addDependency("req-1", "req-2");
const edgeKey = "req-1->req-2:depends_on";
expect(graph.graph.hasEdge(edgeKey)).toBe(true);
const attrs = graph.graph.getEdgeAttributes(edgeKey) as Record<string, unknown>;
expect(attrs.edgeType).toBe("depends_on");
});
it("is idempotent — duplicate addDependency is ignored", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.a", status: "pending", input: {} });
graph.addCall({ requestId: "req-2", operationId: "task.b", status: "pending", input: {} });
graph.addDependency("req-1", "req-2");
graph.addDependency("req-1", "req-2");
expect(graph.graph.hasEdge("req-1->req-2:depends_on")).toBe(true);
});
it("throws NodeNotFoundError if source doesn't exist", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-2", operationId: "task.b", status: "pending", input: {} });
expect(() => graph.addDependency("missing", "req-2")).toThrow(NodeNotFoundError);
});
it("throws NodeNotFoundError if target doesn't exist", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.a", status: "pending", input: {} });
expect(() => graph.addDependency("req-1", "missing")).toThrow(NodeNotFoundError);
});
it("throws CycleError if adding would create cycle", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.a", status: "pending", input: {} });
graph.addCall({ requestId: "req-2", operationId: "task.b", status: "pending", input: {} });
graph.addCall({ requestId: "req-3", operationId: "task.c", status: "pending", input: {} });
graph.addEdge("req-1", "req-2");
graph.addEdge("req-2", "req-3");
expect(() => graph.addDependency("req-3", "req-1")).toThrow(CycleError);
});
});
describe("FlowGraph.updateStatus", () => {
it("transitions pending to running", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("running");
});
it("transitions running to completed", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "completed", { completedAt: "2026-01-01T00:00:01Z" });
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
expect(attrs.completedAt).toBe("2026-01-01T00:00:01Z");
});
it("transitions running to failed", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "failed");
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("failed");
});
it("transitions pending to aborted", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "aborted");
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("aborted");
});
it("transitions running to aborted", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "aborted");
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("aborted");
});
it("is no-op if status is already the target", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "pending");
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("pending");
});
it("throws InvalidTransitionError for completed to running", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "completed", { completedAt: "2026-01-01T00:00:01Z" });
expect(() => graph.updateStatus("req-1", "running")).toThrow(InvalidTransitionError);
});
it("throws InvalidTransitionError for failed to running", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "failed");
expect(() => graph.updateStatus("req-1", "running")).toThrow(InvalidTransitionError);
});
it("throws InvalidTransitionError for aborted to running", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "aborted");
expect(() => graph.updateStatus("req-1", "running")).toThrow(InvalidTransitionError);
});
it("throws InvalidTransitionError for pending to completed", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
expect(() => graph.updateStatus("req-1", "completed")).toThrow(InvalidTransitionError);
});
it("throws InvalidTransitionError for pending to failed", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
expect(() => graph.updateStatus("req-1", "failed")).toThrow(InvalidTransitionError);
});
it("throws NodeNotFoundError for unknown requestId", () => {
const graph = new FlowGraph();
expect(() => graph.updateStatus("missing", "running")).toThrow(NodeNotFoundError);
});
it("InvalidTransitionError contains from/to info", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "completed", { completedAt: "2026-01-01T00:00:01Z" });
try {
graph.updateStatus("req-1", "running");
expect.unreachable("should throw");
} catch (e) {
expect(e).toBeInstanceOf(InvalidTransitionError);
const ite = e as InvalidTransitionError;
expect(ite.requestId).toBe("req-1");
expect(ite.from).toBe("completed" as CallStatus);
expect(ite.to).toBe("running" as CallStatus);
}
});
it("merges extra attributes on transition", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.op", status: "pending", input: {} });
graph.updateStatus("req-1", "running");
graph.updateStatus("req-1", "completed", {
output: { result: 42 },
completedAt: "2026-01-01T00:00:01Z",
});
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.status).toBe("completed");
expect(attrs.output).toEqual({ result: 42 });
expect(attrs.completedAt).toBe("2026-01-01T00:00:01Z");
});
});
describe("FlowGraph.updateCall", () => {
it("partially merges call attributes", () => {
const graph = new FlowGraph();
graph.addCall({
requestId: "req-1",
operationId: "task.op",
status: "pending",
input: {},
});
graph.updateCall("req-1", { output: "some result" });
const attrs = graph.getNodeAttributes("req-1") as Record<string, unknown>;
expect(attrs.output).toBe("some result");
expect(attrs.status).toBe("pending");
});
it("throws NodeNotFoundError for unknown requestId", () => {
const graph = new FlowGraph();
expect(() => graph.updateCall("missing", { output: "x" })).toThrow(NodeNotFoundError);
});
});
describe("FlowGraph.removeCall", () => {
it("removes node and attached edges", () => {
const graph = new FlowGraph();
graph.addCall({ requestId: "req-1", operationId: "task.parent", status: "pending", input: {} });
graph.addCall({ requestId: "req-2", operationId: "task.child", status: "pending", input: {}, parentRequestId: "req-1" });
expect(graph.size).toBe(1);
graph.removeCall("req-2");
expect(graph.hasNode("req-2")).toBe(false);
expect(graph.size).toBe(0);
expect(graph.hasNode("req-1")).toBe(true);
});
it("is a no-op if requestId doesn't exist", () => {
const graph = new FlowGraph();
expect(() => graph.removeCall("missing")).not.toThrow();
});
});

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

View File

@@ -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");
@@ -159,7 +159,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -177,7 +177,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -201,7 +201,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -225,7 +225,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -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();
});
@@ -265,7 +265,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
const respondedEvent: CallEventMapValue = {
type: "call.responded",
@@ -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();
});
@@ -304,7 +304,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -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();
});
@@ -353,7 +353,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -380,7 +380,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -408,7 +408,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -434,7 +434,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -452,7 +452,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-2");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -466,6 +466,7 @@ describe("WorkflowReactiveRoot", () => {
error: { code: "ERR", message: "first attempt failed" },
timestamp: "2026-01-01T00:00:01Z",
});
root.setRequestId("a", "req-2");
root.append({
type: "call.requested",
requestId: "req-2",
@@ -503,7 +504,7 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-1");
root.setRequestId("a", "req-1");
root.append({
type: "call.requested",
requestId: "req-1",
@@ -840,11 +841,11 @@ describe("WorkflowReactiveRoot", () => {
const graph = makeSimpleGraph();
const root = new WorkflowReactiveRoot(graph);
root.nodeKeyToRequestId.set("a", "req-a");
root.nodeKeyToRequestId.set("b", "req-b");
root.nodeKeyToRequestId.set("c", "req-c");
root.setRequestId("a", "req-a");
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");