Implement retry semantics: requestIdToNodeKey reverse map, setRequestId method, full-history getEvents

src/analysis/index.ts

@@ -4,7 +4,8 @@ export {
   defaultEdgeType,
   resolveDefaultNodeAttrs,
 } from "./defaults.js";
-export { typeCompat, buildTypeEdges, type TypeCompatResult, type TypeMismatch } from "./type-compat.js";
+export { typeCompat, type TypeCompatResult, type TypeMismatch } from "./type-compat.js";
+export { buildTypeEdges } from "../graph/construction.js";
 export {
   validateSchema,
   validateGraph,

src/analysis/type-compat.ts

@@ -1,11 +1,4 @@
 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;
@@ -282,25 +275,4 @@ 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 } : {}),
-      });
-    }
-  }
 }

src/graph/construction.ts

@@ -21,10 +21,8 @@ import {
   OperationGraphSerialized,
   CallGraphSerialized,
 } from "../schema/index.js";
-import type { FlowGraphSerialized } from "../schema/index.js";
-import { buildTypeEdges, type TypeCompatResult } from "../analysis/type-compat.js";
-
-export { buildTypeEdges } from "../analysis/type-compat.js";
+import type { OperationNodeAttrs, FlowGraphSerialized } from "../schema/index.js";
+import { typeCompat, type TypeCompatResult } from "../analysis/type-compat.js";
 
 export interface FlowGraphOptions {
   type?: "directed";
@@ -457,4 +455,25 @@ 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 } : {}),
+      });
+    }
+  }
 }

src/index.ts

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

src/reactive/workflow.ts

@@ -88,6 +88,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,10 +107,16 @@ 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) ?? [];
@@ -213,15 +220,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 +275,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 +351,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;
-  }
 }

test/analysis/type-compat.test.ts

@@ -1,8 +1,6 @@
 import { describe, it, expect } from "vitest";
 import { Type, type TSchema } from "@alkdev/typebox";
-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";
+import { typeCompat, type TypeCompatResult, type TypeMismatch } from "../../src/analysis/type-compat.js";
 
 describe("typeCompat", () => {
   describe("exact match", () => {
@@ -413,111 +411,4 @@ describe("typeCompat", () => {
       expect(typeof mismatch.actual).toBe("string");
     });
   });
-});
-
-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);
-  });
 });

test/reactive/workflow.test.ts

@@ -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",
@@ -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",
@@ -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",
@@ -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,9 +841,9 @@ 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("b")).toBe("idle");