ri-agent-graph

Graph-based agent orchestration for Rust — a LangGraph-inspired execution engine with checkpointing, parallel fan-out/fan-in, interrupt/resume, and cryptographic execution receipts.

What it gives you

• Deterministic graph execution — define nodes as computational steps, edges as control flow, and execute with typed state
• 8 node types — llm, router, join, parallel, passthrough, state_transform, subgraph, human_approval
• Parallel fan-out/fan-in — automatic concurrent execution with configurable join policies (collect_array, merge_objects, first_non_null, all_success, quorum)
• Checkpointing & interrupt/resume — SQLite-backed persistence with atomic checkpoint transactions and crash recovery
• Cryptographic receipts — HMAC-SHA256 authenticated GraphExecutionReceiptV1 for every run
• Event streaming — lifecycle events (node start/complete/error), token streaming, and state snapshots
• stack-ids integration — TraceCtx, AttemptId, TrialId execution tracing at every layer
• Zero-cost abstractions — generic over user-defined state S, no heap allocation beyond what your nodes require

Installation

cargo add ri-agent-graph

Or in your Cargo.toml:

[dependencies]
ri-agent-graph = "0.2"

Feature flags

To disable checkpointing (embedded/no_std targets):

ri-agent-graph = { version = "0.2", default-features = false }

Quick start

use ri_agent_graph::prelude::*;

#[tokio::main]
async fn main() -> Result<()> {
    // Build a two-step graph
    let graph = AgentGraph::builder()
        .add_node("step1", node!(|state| async move {
            state.set("count", 1).await?;
            Ok(())
        }))
        .add_node("step2", node!(|state| async move {
            let count: i32 = state.get("count").await?;
            state.set("count", count + 1).await?;
            Ok(())
        }))
        .add_edge("step1", "step2")
        .build()?;

    // Execute
    let state = AgentState::new();
    let result = graph.execute("step1", state).await?;

    let final_count: i32 = result.get("count").await?;
    assert_eq!(final_count, 2);

    Ok(())
}

Node types

Router example

use ri_agent_graph::{AgentGraph, router, node, START, END};
use serde_json::json;

let graph = AgentGraph::builder()
    .add_node("classify", node!(|state| async move {
        // LLM classification returns "bug", "feature", or "question"
        state.set("category", "bug").await?;
        Ok(())
    }))
    .add_node("handle_bug", node!(|state| async move {
        state.set("response", "Bug triaged").await?;
        Ok(())
    }))
    .add_node("handle_feature", node!(|state| async move {
        state.set("response", "Feature scoped").await?;
        Ok(())
    }))
    .add_node("handle_question", node!(|state| async move {
        state.set("response", "Question answered").await?;
        Ok(())
    }))
    .add_edge(START, "classify")
    .add_router("classify", router!(|state| {
        let category: String = state.get("category").await?;
        Ok(match category.as_str() {
            "bug" => vec!["handle_bug"],
            "feature" => vec!["handle_feature"],
            _ => vec!["handle_question"],
        })
    }))
    .add_edge("handle_bug", END)
    .add_edge("handle_feature", END)
    .add_edge("handle_question", END)
    .build()?;

Parallel fan-out with join

let graph = AgentGraph::builder()
    .add_node("coordinator", node!(|state| async move {
        state.set("workstreams", json!(["A", "B", "C"])).await?;
        Ok(())
    }))
    .add_node("fanout", passthrough_node!())
    .add_node("worker_a", node!(|state| async move {
        state.set("result_a", "done").await?;
        Ok(())
    }))
    .add_node("worker_b", node!(|state| async move {
        state.set("result_b", "done").await?;
        Ok(())
    }))
    .add_node("worker_c", node!(|state| async move {
        state.set("result_c", "done").await?;
        Ok(())
    }))
    .add_node("merger", join_node!(JoinMode::CollectArray, ["result_a", "result_b", "result_c"], "findings"))
    .add_edge("coordinator", "fanout")
    .add_edge("fanout", "worker_a")
    .add_edge("fanout", "worker_b")
    .add_edge("fanout", "worker_c")
    .add_edge("worker_a", "merger")
    .add_edge("worker_b", "merger")
    .add_edge("worker_c", "merger")
    .add_edge("merger", END)
    .with_reducers(Reducers::new().append_to("findings"))
    .build()?;

Checkpointing & interrupt/resume

Enable checkpointing to persist execution state to SQLite:

use ri_agent_graph::checkpoint_store::SqliteCheckpointStore;

let store = SqliteCheckpointStore::open("executions.db").await?;
let mut executor = GraphExecutor::new(graph)
    .with_checkpoint_store(store);

// Execute with interrupt detection
match executor.execute_with_interrupt(initial_state).await {
    Ok(receipt) => println!("Completed: {:?}", receipt),
    Err(AgentGraphError::Interrupted { checkpoint_id, .. }) => {
        // Resume from checkpoint with injected input
        executor.resume_from(checkpoint_id, injected_input).await?;
    }
}

Execution receipts

Every completed run produces a GraphExecutionReceiptV1:

pub struct GraphExecutionReceiptV1 {
    pub run_id: String,
    pub graph_name: String,
    pub start_time: DateTime<Utc>,
    pub end_time: DateTime<Utc>,
    pub steps: Vec<StepExecutionReceiptV1>,
    pub final_state_digest: String,
    pub status: ExecutionOutcome,
}

Each step receipt carries:

• Node ID, attempt count, wall-clock duration
• Input/output state digests
• Error details (if any)
• TraceCtx / AttemptId / TrialId from stack-ids

Ecosystem

API overview

// Core types
AgentGraph<S>           // The graph: nodes + edges + reducers
AgentState              // Key-value state flowing through execution
GraphExecutor<S>        // The runtime engine
GraphExecutionReceiptV1 // Cryptographic execution receipt

// Node construction
node!(|state| async { ... })         // Inline closure node
passthrough_node!()                   // No-op passthrough
router!(|state| { ... })             // Conditional routing
join_node!(mode, inputs, output)     // Fan-in synchronization

// Sentinels
START  // Virtual entry node
END    // Virtual exit node

Claim boundaries

• This crate provides graph execution semantics — it does not include LLM provider clients, prompt templating, or response parsing. Those belong in llm-pipeline or your application layer.
• Receipts prove structural execution — they do not prove that an external LLM call occurred, what model responded, or what the provider's internal state was. Receipts carry cryptographic digests of the local execution trace only.
• Interrupt/resume is deterministic local resume — it supports linear chains of deterministic passthrough and local state_transform nodes with SQLite-bound state. It does not support resuming across LLM calls, network I/O, or external tool invocations.
• Parallelism is best-effort — concurrent branch execution uses Tokio's JoinSet. Unordered parallel writes to the same state key are rejected unless an explicit Reducer is declared.

Error handling

All fallible operations return Result<T, AgentGraphError>:

pub enum AgentGraphError {
    GraphBuild(String),
    NodeNotFound(String),
    EdgeNotFound(String),
    StateKeyNotFound(String),
    StateTypeMismatch { key: String, expected: String, actual: String },
    ParallelWriteConflict(String),
    CheckpointError(CheckpointStoreOperation),
    Interrupted { checkpoint_id: String, node_id: String },
    ExecutionTimeout { run_id: String, elapsed_ms: u64 },
    IntegrityKeyRequired,
    // ...
}

Verification

# Build
cargo build --release -p ri-agent-graph

# Test
cargo test -p ri-agent-graph

# Clippy (strict)
cargo clippy -p ri-agent-graph -- -D warnings

# Format check
cargo fmt --check

# Publish dry-run
cargo publish -p ri-agent-graph --dry-run

Roadmap

• Typed state extractors (derive macro for StateExtract)
• Graph visualization (Mermaid/DOT export)
• Streaming LLM token passthrough
• Distributed checkpoint backend (PostgreSQL, S3)
• Subgraph composition with state isolation
• WebAssembly target support (wasm-bindgen)

License

MIT — see LICENSE-MIT for details.

Built by RecursiveIntell — an applied R&D studio building local-first AI infrastructure.