@emre-yildiz-dev/effect-graph

An Effect-native superstep graph engine ("Pregel-lite") for building durable, resumable workflows and agent loops. Typed state with reducers, conditional routing, parallel supersteps, typed human-in-the-loop interrupts, checkpointing, and a typed event stream.

No AI, no database, no HTTP — it runs Effects in a loop. effect is its only peer dependency.

bun add @emre-yildiz-dev/effect-graph effect
# or: npm i @emre-yildiz-dev/effect-graph effect

Status: 0.x. The API is young and may change between minor versions. It is exercised by a full test suite (90 tests) and runs in production in one codebase, but it has not yet met a wide variety of use cases.

What it does

A graph is typed Schema state plus named nodes (plain Effect functions), static edges, and conditional routers. The executor runs the active node set one superstep at a time: run nodes concurrently, merge their writes through per-channel reducers in declaration order, checkpoint (per policy), route to the next active set. Runs emit a GraphEvent stream and can suspend on typed human-in-the-loop interrupts and resume from a checkpoint.

Key concepts

• GraphState.make / GraphState.channel: Schema struct state; each channel has a reducer (default last-write-wins) and an optional default
• Graph.make: nodes + edges + routers, topology validated at construction (GraphDefinitionError)
• execute: returns { events, outcome, cancel } — a per-run event stream, the run result, and native fiber cancellation
• GraphInterrupt.define: typed interrupt-as-value; suspends the run, resumes with a Schema-decoded value
• Checkpointer: Context.Tag port (save / loadLatest / deleteThread); InMemoryCheckpointer for tests
• GraphEvent: 11-tag Schema union with a per-run monotonic sequence

Usage

const State = GraphState.make({
  messages: GraphState.channel(Schema.Array(Schema.String), {
    reducer: (prev, next) => [...prev, ...next],
    default: () => [],
  }),
})

const graph = Graph.make(State, {
  stateVersion: "1",
  nodes: { reason: (state) => Effect.succeed({ update: { messages: ["hi"] } }) },
  edges: [[START, "reason"]],
})

const handle = yield* execute(graph, { context, input: {} })
const outcome = yield* handle.outcome

Visualizing a graph

toMermaid renders any Graph's topology — nodes, static edges, routers, and a legend of declared interrupts / checkpoint policy / state version:

import { toMermaid } from "@emre-yildiz-dev/effect-graph"

const diagram = toMermaid(graph)
// optional: a router's targets are dynamic, so they can be annotated for docs
const annotated = toMermaid(graph, { routeHints: { review: ["publish", "revise"] } })

Route hints are validated against the real topology — hinting an unknown target, or a node that isn't a router, throws rather than rendering a diagram that lies.

Example: showcase pipeline

A mini research pipeline that exercises the engine's distinctive capabilities in one graph. The diagram below is asserted against toMermaid output by the test suite, so it cannot drift from the code.

flowchart TD
  __start__([START])
  plan["plan"]
  searchWeb["searchWeb"]
  searchDocs["searchDocs"]
  draft["draft"]
  review["review"]
  revise["revise"]
  publish["publish"]
  __end__([END])
  __start__ --> plan
  plan --> searchWeb
  plan --> searchDocs
  searchWeb --> draft
  searchDocs --> draft
  draft --> review
  revise --> draft
  publish --> __end__
  review -.-> publish
  review -.-> revise
  classDef router fill:#f4d03f,stroke:#7d6608,stroke-width:2px;
  class review router

%% interrupts: review-approval
%% checkpoint: every-step
%% stateVersion: 1

What each part demonstrates:

| Part | Capability | |------|------------| | plan → searchWeb + searchDocs | Parallel fan-out — both run in one superstep | | findings append reducer | Deterministic merge — writes merge in node-declaration order even though searchWeb finishes last | | searchWeb + searchDocs → draft | Fan-in — deduped to a single activation | | review | Typed HITL interrupt — suspends before any publish side effect | | router on review | Conditional routing + loop — rejection cycles through revise → draft, bounded by recursionLimit | | checkpoint: "every-step" | Resume — the run continues across separate execute calls on the same threadId |

Source: test/examples/ShowcaseGraph.ts · Tests: test/examples/ShowcaseGraph.test.ts

Capabilities

| Supported | Not built (yet) | |---|---| | Typed Schema state + per-channel reducers | Subgraphs | | Static edges, conditional routers, goto | Dynamic map-reduce fan-out (Send-style) | | Parallel supersteps (sibling auto-interrupt on failure) | Time travel / fork-from-checkpoint | | Typed human-in-the-loop interrupts | Multiple concurrent interrupts per run | | Checkpointing, thread continuity, resume | Cross-thread memory store | | Native cancellation via Fiber.interrupt | Per-node retry policies / caching | | Typed event stream + OpenTelemetry spans | Prebuilt agent patterns | | Mermaid visualization | |

Design notes

• Errors are typed end-to-end. A node's failure type flows into execute's signature; infrastructure faults are defects. Every run publishes exactly one terminal event (run:completed / run:failed / run:interrupted / run:cancelled), including on defects.
• Interrupts are values, not exceptions. request() returns the schema-decoded resume value. The interrupted node re-executes from the top on resume, so side effects before the request must be idempotent.
• v1 interrupt restrictions: at most one pending interrupt per run, only in a single-node superstep, and only under checkpoint: "every-step". Violations are defects with explicit messages.
• Checkpointing is a port. Checkpointer is a Context.Tag; InMemoryCheckpointer ships for tests, and a durable implementation (Postgres, etc.) is yours to provide.

Development

bun install
bun run ci     # typecheck + lint + test + build

License

MIT