StateGraph Runtime

The StateGraph runtime executes durable, checkpointed workflows. Nodes run in sequence or branch via conditional edges. State flows through the graph as a map[string]any. Checkpointing after every node enables resume after interrupts and time-travel debugging.

Where LLM Calls Happen

The StateGraph runtime (engine/graph) is intentionally LLM-agnostic — it imports no model provider and never calls an LLM itself. It is the durable orchestration engine: it runs your node functions, evaluates edges, and checkpoints state.

LLM calls happen inside node functions, which you write. A node receives the state, calls a model.Provider, and returns updated state:

import (
    "context"

    "github.com/spawn08/chronos/engine/graph"
    "github.com/spawn08/chronos/engine/model"
)

// provider is any model.Provider (OpenAI, Anthropic, Gemini, Ollama, …).
g.AddNode("classify", func(ctx context.Context, s graph.State) (graph.State, error) {
    question, _ := s["question"].(string)

    resp, err := provider.Chat(ctx, &model.ChatRequest{
        Messages: []model.Message{
            {Role: model.RoleSystem, Content: "Classify the question as 'technical' or 'general'. Reply with one word."},
            {Role: model.RoleUser, Content: question},
        },
    })
    if err != nil {
        return s, fmt.Errorf("classify: %w", err)
    }

    s["category"] = resp.Content // the LLM output flows on as state
    return s, nil
})

This layering mirrors the rest of the framework:

sdk/agent      → composes model + graph; agentic loops live here
engine/model   → Provider.Chat / StreamChat        (the LLM part)
engine/graph   → durable execution engine           (LLM-agnostic, on purpose)

Because the runtime is decoupled from the model, the same durability, resume, human-in-the-loop, and parallelism machinery works for any workflow — LLM-powered or not. "Agentic AI" emerges from what you put inside the nodes (model calls, tools, reasoning), not from the graph engine itself.

See examples/graph_with_llm for a full multi-node graph with real providers and tools.

Durability: How Checkpointing Survives Crashes

The runner checkpoints state after every node that completes, recording the next node to execute. If a later node fails — a crash, a transient provider/network error, or a process/replica restart — Resume loads the last checkpoint and continues from exactly that point. A completed, expensive LLM node is never re-executed.

draft (LLM) ──▶ review (LLM) ──▶ finalize
     ✓ checkpointed    ✗ crashes here

On Resume, draft is skipped (its output is in the checkpoint) and execution restarts at review:

// Attempt 1 — draft succeeds and is checkpointed; review crashes.
_, err := graph.NewRunner(compiled, store).Run(ctx, sessionID, graph.State{"topic": "durability"})
// err != nil, but draft's result is durably checkpointed.

// Later (even in a brand-new process/replica pointed at the same store):
// a Runner is single-use, so construct a fresh one — the state lives in the store.
result, err := graph.NewRunner(compiled, store).Resume(ctx, sessionID)
// draft does NOT run again; execution resumes at review → finalize.

Durability requires a persistent store (storage/adapters/sqlite for dev, storage/adapters/postgres for production). With Postgres, resume works across replicas, which is what makes the control plane horizontally scalable.

A complete, runnable demonstration (with a deterministic offline provider, so it needs no API key) lives in examples/durable_llm_graph — it proves the draft node's LLM call runs exactly once despite the crash.

Creating a Graph

g := graph.New("my-workflow")

Adding Nodes

Each node has an ID and a handler function:

g.AddNode("greet", func(ctx context.Context, s graph.State) (graph.State, error) {
    s["greeting"] = fmt.Sprintf("Hello, %s!", s["user"])
    return s, nil
})

g.AddNode("classify", func(ctx context.Context, s graph.State) (graph.State, error) {
    s["intent"] = "general_question"
    return s, nil
})

Interrupt Nodes

Interrupt nodes pause execution for human-in-the-loop approval. The runner checkpoints and returns before executing the node.

g.AddInterruptNode("approve", func(ctx context.Context, s graph.State) (graph.State, error) {
    // This node runs only after Resume; before that, execution pauses
    s["approved"] = true
    return s, nil
})

Entry and Finish Points

g.SetEntryPoint("greet")   // Start here
g.SetFinishPoint("respond") // End here

SetEntryPoint adds an edge from __start__ to the given node. SetFinishPoint adds an edge from the given node to __end__.

Edges

Static Edges

g.AddEdge("greet", "classify")
g.AddEdge("classify", "respond")

Conditional Edges

Route based on state. The condition function returns the target node ID.

g.AddConditionalEdge("classify", func(s graph.State) string {
    intent, _ := s["intent"].(string)
    switch intent {
    case "support":
        return "support_flow"
    case "sales":
        return "sales_flow"
    default:
        return "general_flow"
    }
})

Compiling

Validate the graph and produce an immutable CompiledGraph:

compiled, err := g.Compile()
if err != nil {
    return err  // e.g., missing entry point, invalid edge targets
}

Runner

The runner executes a compiled graph with checkpointing:

runner := graph.NewRunner(compiled, storage)

Run

Start a new execution with initial state:

result, err := runner.Run(ctx, sessionID, graph.State{"user": "Alice"})

Resume

Continue from the latest checkpoint (e.g., after an interrupt):

result, err := runner.Resume(ctx, sessionID)

ResumeFromCheckpoint

Resume from a specific checkpoint (time-travel debugging):

result, err := runner.ResumeFromCheckpoint(ctx, checkpointID)

Checkpointing

State is saved after every node. Each checkpoint stores:

Session ID, Run ID, Node ID
Full state
Sequence number

Storage must implement SaveCheckpoint and GetLatestCheckpoint (and GetCheckpoint for time-travel). Use SQLite or Postgres adapters.

StreamEvent

Subscribe to execution events for observability. Use the runner directly (not via agent.Run) to access the stream:

compiled, _ := g.Compile()
runner := graph.NewRunner(compiled, store)
stream := runner.Stream()
for evt := range stream {
    switch evt.Type {
    case "node_start":
        fmt.Printf("Starting node %s\n", evt.NodeID)
    case "node_end":
        fmt.Printf("Finished node %s\n", evt.NodeID)
    case "edge_transition":
        fmt.Printf("Transitioning to %s\n", evt.NodeID)
    case "interrupt":
        fmt.Printf("Paused at interrupt node %s\n", evt.NodeID)
    case "error":
        fmt.Printf("Error: %s\n", evt.Error)
    case "completed":
        fmt.Println("Graph completed")
    }
}

Type	Description
`node_start`	Node execution began
`node_end`	Node execution finished
`edge_transition`	Transitioning to next node
`interrupt`	Paused at interrupt node
`error`	Node failed
`completed`	Graph finished successfully

Integration with Agent

Attach a graph to an agent via the builder. The agent's Run and Resume methods use the graph:

g := graph.New("workflow").
    AddNode("greet", func(ctx context.Context, s graph.State) (graph.State, error) {
        s["greeting"] = fmt.Sprintf("Hello, %s!", s["user"])
        return s, nil
    }).
    AddNode("respond", func(ctx context.Context, s graph.State) (graph.State, error) {
        s["response"] = "How can I help?"
        return s, nil
    }).
    SetEntryPoint("greet").
    AddEdge("greet", "respond").
    SetFinishPoint("respond")

a, err := agent.New("run-agent", "Run Agent").
    WithStorage(store).
    WithGraph(g).
    Build()
if err != nil {
    log.Fatal(err)
}

result, err := a.Run(ctx, map[string]any{"user": "World"})
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Result: %v\n", result.State)

Time-Travel Debugging

Resume from any historical checkpoint to replay or debug:

checkpoints, _ := store.ListCheckpoints(ctx, sessionID)
// User selects checkpointID from UI or CLI
result, err := runner.ResumeFromCheckpoint(ctx, checkpointID)

Complete Example

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/spawn08/chronos/engine/graph"
    "github.com/spawn08/chronos/sdk/agent"
    "github.com/spawn08/chronos/storage/adapters/sqlite"
)

func main() {
    ctx := context.Background()

    store, err := sqlite.New("run.db")
    if err != nil {
        log.Fatal(err)
    }
    defer store.Close()
    if err := store.Migrate(ctx); err != nil {
        log.Fatal(err)
    }

    g := graph.New("workflow").
        AddNode("greet", func(_ context.Context, s graph.State) (graph.State, error) {
            s["greeting"] = fmt.Sprintf("Hello, %s!", s["user"])
            return s, nil
        }).
        AddNode("classify", func(_ context.Context, s graph.State) (graph.State, error) {
            s["intent"] = "general_question"
            return s, nil
        }).
        AddNode("respond", func(_ context.Context, s graph.State) (graph.State, error) {
            s["response"] = fmt.Sprintf("Intent: %s. How can I help?", s["intent"])
            return s, nil
        }).
        SetEntryPoint("greet").
        AddEdge("greet", "classify").
        AddEdge("classify", "respond").
        SetFinishPoint("respond")

    a, err := agent.New("run-agent", "Run Agent").
        WithStorage(store).
        WithGraph(g).
        Build()
    if err != nil {
        log.Fatal(err)
    }

    result, err := a.Run(ctx, map[string]any{"user": "World"})
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Result: %v\n", result.State)
}

Where LLM Calls Happen​

Durability: How Checkpointing Survives Crashes​

Creating a Graph​

Adding Nodes​

Interrupt Nodes​

Entry and Finish Points​

Edges​

Static Edges​

Conditional Edges​

Compiling​

Runner​

Run​

Resume​

ResumeFromCheckpoint​

Checkpointing​

StreamEvent​

Integration with Agent​

Time-Travel Debugging​

Complete Example​