The flow graph model

A flow is { nodes, edges } and nothing else. Everything the builder draws, the engine walks, the history diffs and the export carries is derived from that one document, which is stored as a single JSON column. Get this shape right first — the canvas, the inspector and the engine are all downstream of it, and a graph that cannot be parsed is a flow that silently stops running.

The three types

interface FlowNode {
  id: string;
  type: NodeType;                    // the discriminant
  position: { x: number; y: number };
  config: /* narrowed by `type` */;
}

interface FlowEdge {
  id: string;
  source: string;
  target: string;
  /** Outlet label for branching sources; null = the single default outlet. */
  branch: string | null;
}

interface FlowGraph { nodes: FlowNode[]; edges: FlowEdge[] }

Three decisions are load-bearing:

type is a discriminant, not a string. Nodes are a Zod discriminated union, so the engine's switch narrows config per arm and a never check at the bottom makes an unhandled node type a compile error.
The branch lives on the edge, not on the node. A node declares which outlets it has; an edge says which one it leaves by. Adding a branch is therefore an edit to one node's config, and the edges follow.
position is part of the model. Layout is authored, saved, versioned and diffed. Never recompute it behind the author's back — see flow-builder-canvas.

Execution is single-path

Exactly one node is active at a time, and a branching node chooses one outgoing edge by its label. The frontier is a single node id, which is what makes durable resume a one-line concept instead of a distributed join.

export function nextFrom(bySource: Map<string, FlowEdge[]>, nodeId: string,
                         branch: string | null): string | null {
  const out = bySource.get(nodeId);
  if (!out?.length) return null;
  if (branch === null) return out[0]!.target;
  const exact = out.find((e) => e.branch === branch);
  if (exact) return exact.target;
  // An outlet with nothing on it falls back to the default path rather than
  // ending the run — a half-built branch must not silently drop subjects.
  return out.find((e) => e.branch == null || e.branch === "default")?.target ?? null;
}

Bucket edges by source once (edgesBySource) and pass the map down. A filter() inside a walk is O(nodes × edges), and the walk visits every node.

Configs must tolerate a half-built flow

The builder saves drafts. A schema that only accepts finished configs makes a freshly dropped node unsaveable, which makes the builder unusable.

/** A connection/destination reference: a uuid, or the empty string. */
export const ConfigRefSchema = z.string().uuid().or(z.literal("")).default("");

Apply the same rule to every field a person fills in second: an unchosen predicate path, an empty output_key, a mapping with no to. Each one is skipped by the engine, so '' is a valid work-in-progress state rather than an error. Give every field a .default() so a bare {} parses.

Normalize legacy shapes on read — never migrate

A graph lives in three places at once: the flow's own row, every pinned run snapshot, and every history revision. You cannot migrate all three, and a graph that fails to parse drops out of flow selection — the flow just stops running, with no error anywhere.

export const TriggerConfigSchema = z.preprocess((raw) => {
  if (typeof raw !== "object" || raw === null) return raw;
  const cfg = raw as Record<string, unknown>;
  if ("event_types" in cfg || !("event_type" in cfg)) return cfg;
  const { event_type, ...rest } = cfg;               // v1 → v2, on read
  return { ...rest, event_types: event_type ? [event_type] : [] };
}, TriggerConfigShape);

Widening a field is free. Renaming one costs a preprocess forever — which is cheaper than the alternative.

Purity decides what gets journaled

A durable engine memoises what it cannot re-run. Declare that per node type as an opt-in, and derive the dangerous set from it:

export const PURE_NODE_TYPES =
  new Set(BLOCK_CATALOG.filter((b) => b.pure).map((b) => b.type));
export const JOURNALED_NODE_TYPES =
  new Set(BLOCK_CATALOG.filter((b) => !b.pure).map((b) => b.type));

Omitting pure is the safe default: anything unmarked is journaled. A node wrongly marked pure is executed twice on replay — re-delivering, re-charging a model call, re-emitting an event. See flow-node-catalog for the paired-evaluator rule that keeps the claim honest.

Caps exist for a reason — write the reason down

/** A suspended run's journal must outlive the run, and partition retention
 *  is finite: a wait longer than the shortest retention window lets a
 *  partition drop destroy the memo of a still-waiting run, which then
 *  replays from the trigger and re-executes every node. */
export const MAX_WAIT_MS = 30 * 24 * 60 * 60 * 1000;

Also cap nodes (500) and edges (2000) so a pathological document cannot be saved. A cap with no comment gets raised by the next person; a cap with the failure mode attached does not.

Validation: errors stop a run, warnings are usually a mistake

export interface GraphIssue { level: "error" | "warning"; nodeId?: string; message: string }

Level	Condition
error	duplicate node id
error	no trigger node, or more than one
error	edge whose `source`/`target` is not in the graph
warning	edge on an outlet its source no longer has
warning	node not reachable from the trigger

Surface both in the builder's banner strip, and never block a save on a warning — an unreachable step is a normal intermediate state while building.

One canonical comparison

canonicalGraph sorts nodes and edges by id, sorts object keys, drops undefined, and rounds positions to whole pixels. Then:

export function graphsDiffer(a: FlowGraph, b: FlowGraph): boolean {
  return canonicalGraph(a) !== canonicalGraph(b);
}

Import this on both sides. The server's write gate and the UI's "Unsaved" badge must be the same predicate: if the badge's check is finer than the gate, you press Save, the server no-ops, and the badge can never clear.

Nested groups: inline once, at snapshot time

A group node (flow.call) holds a whole FlowGraph inline, so the types are recursive and that one arm of the union is hand-written. expandFlowCalls splices every group into the parent — namespacing inner ids as outer::inner, rewiring the group's outgoing edge into the sub-graph's entry, and wiring every terminal inner node back to the group's continuation. Run it once, when the run's graph snapshot is built, so the walk, the journal and resume never see sub-flows as a special case. Bound the recursion (depth 5) and fall back to the un-expanded graph if expansion produces an invalid one.

Parsing at the boundary

/** Parse a stored graph, returning null on failure. Callers MUST log the null. */
export function parseGraph(raw: unknown): FlowGraph | null {
  const res = FlowGraphSchema.safeParse(raw ?? EMPTY_GRAPH);
  return res.success ? res.data : null;
}

A null here is the failure mode described above, so it is never swallowed: the API records it and the builder shows a banner — "this flow's graph could not be read, so it will not match any event. Fix it here and save to recover." An unparseable graph must be editable, or the only fix is a database write.

flow-graph-model

What it does