Serving & operations

Async graphs, sessions, the HTTP API, and reading a run afterwards.

How do I run a graph whose node has to await something?

Write the node async def and drive the graph with ainvoke(). Nothing else changes — the node still returns a dict, still declares its writes, still gets a deep-copied state. Sync nodes stay legal on the async path (LangGraph puts them on its own worker threads), so a graph may mix the two freely as long as you drive it with ainvoke()/astream(). max_seconds is the one thing that genuinely differs: on a sync node the ceiling arrives as an interrupt to the thread running it; on an async def node it arrives as task cancellation, because a signal aimed at "the thread" on an event loop would land in whichever coroutine happened to be running.

python

async def fetch(state: State) -> dict:
    await asyncio.sleep(0.01)          # a real client would await here
    return {"body": f"<html>{state.url}</html>"}


g = GraphARC(State, name="fetcher")
g.add_node("fetch", fetch, writes={"body"})
g.add_edge(START, "fetch")
g.add_edge("fetch", END)
graph = g.compile()

result = asyncio.run(graph.ainvoke({"url": "https://example.com"}))
print(result["body"])

sharp edge Per grapharc.runtime.graph._async_deadline, cancellation arrives at an await, so an async node doing blocking CPU work is not interrupted until it yields.

Why does invoke() refuse my graph before it runs anything?

Because it found an async def node and there is no event loop to await it on; the check happens at the entry point, before the first node. In the example cheap ran prints exactly once — from the ainvoke() call — proving the refused invoke() executed nothing. That is the whole point of a pre-flight check: plain LangGraph would have run every sync node first and then died on the first coroutine with a TypeError, leaving half a run and a confusing traceback. stream() refuses the same way and names astream() in its message.

console

AsyncNodeError: graph 'mixed' has async nodes ['costly']; use ainvoke(), because invoke() has no event loop to await them on
cheap ran
['cheap', 'costly']

How do I watch state land as the graph runs?

Use astream(). It takes the same stream_mode values LangGraph does; "updates" gives you one chunk per node with just what that node wrote. A "updates" chunk is a {node_name: delta} dict, not a (name, delta) pair. Passing a list of modes (stream_mode=["updates", "checkpoints"]) turns the chunks into (mode, payload) tuples instead — that second shape is what the session runtime uses, because checkpoints is the only chunk that means "the superstep is finished and on disk".

python

async def main() -> None:
    async for chunk in graph.astream(
        {"question": "how do budgets work?"}, stream_mode="updates"
    ):
        for node, delta in chunk.items():
            print(node, "->", delta)


asyncio.run(main())

sharp edge Unpacking an "updates" chunk as two values raises ValueError: not enough values to unpack.

How do I see every model call, live?

astream_events() is LangChain's event stream driven through GraphARC's budgeted path. It emits per-node and per-model-call events, so it is what you want behind a "thinking…" UI or a token meter — filter on on_chain_start for node starts and on_chat_model_end for usage_metadata. version defaults to "v2" and "v1" is accepted.

python

async for event in graph.astream_events({"question": "meaning of life?"}):
    if event["event"] == "on_chain_start":
        print("start", event["name"])
    elif event["event"] == "on_chat_model_end":
        usage = event["data"]["output"].usage_metadata
        print("model", event["name"], usage["total_tokens"], "tokens")

sharp edge "v3" is refused with a ValueError rather than passed through, because LangGraph returns a stream object for v3, not an async iterator — a different shape than this method's contract.

How do I stop someone bypassing the budget?

You do not have to. CompiledGraphARC.inner is the LangGraph object underneath and is reachable, but driving it directly fails closed at the first node with MissingRunContextError. Raw LangGraph entry points would silently bypass budgets and traces, so the node refuses to execute without a GraphARC run context.

console

MissingRunContextError: node 'bump' executed without a GraphARC run context; drive the graph via CompiledGraphARC.invoke()/stream()/ainvoke()/astream() — raw LangGraph entry points would silently bypass budgets and traces
ValueError: astream_events supports version 'v1' or 'v2', got 'v3'

How do I let a node pick the next node?

Return a langgraph.types.Command (re-exported from grapharc.runtime.graph). Its update goes through exactly the checks a returned dict does, so dynamic routing costs nothing in discipline. There is no add_edge(START, ...) for the destinations and no conditional edge at all — the goto is the transition — but the destination nodes still need their outgoing edges to END.

python

def triage(state: State) -> Command:
    if "urgent" in state.text:
        return Command(update={"verdict": "urgent"}, goto="escalate")
    return Command(update={"verdict": "routine"}, goto="archive")

sharp edge Command(update=...) must be a dict, and Command(graph=...) is refused — an update aimed at another graph's state cannot be checked against this graph's declared writes.

What happens when a goto names a node that does not exist?

It raises GraphRoutingError. This is the case worth knowing about because plain LangGraph does not raise: it logs wrote to unknown channel branch:to:<x>, ignoring it and carries on as if the node had never routed. A transition that silently does not happen is the same defect as one that happens unpermitted — the graph did something other than what the node asked for and nobody was told. The check covers node names, END, Send targets from an add_fanout_edge dispatcher, sequences of those, and Command(goto=None) (which LangGraph cannot iterate at all).

console

GraphRoutingError: node 'triage' routed to 'esclate', which is not a node of graph 'triage'; LangGraph drops an unknown destination and the run continues as if the routing had not happened. Valid destinations: 'escalate', 'triage', END

sharp edge The stated limit — the check covers destinations GraphARC hands to LangGraph; a Command passed as input to invoke()/stream() is unsupported (those take a dict, a state model, or None).

How do I pause a graph, let a human edit state, and carry on?

Compile with a checkpointer, stream with interrupt_before, then use get_state / update_state / resume-with-None. snapshot.next tells you which node is pending; update_state(..., as_node="write") attributes the human edit to the node that produced the value; invoke(None, thread_id=...) picks up from the checkpoint. get_state and friends read and write checkpoints rather than executing nodes, so they carry no budget and emit no trace events.

python

# Run until `send` is next, then stop.
for _ in graph.stream({"topic": "the outage"}, thread_id="t1", interrupt_before=["send"]):
    pass

snapshot = graph.get_state("t1")
print("next    :", snapshot.next)
graph.update_state("t1", {"draft": "Dear customer, we are sorry about the outage."}, as_node="write")
result = graph.invoke(None, thread_id="t1")

sharp edge interrupt_before is a stream keyword — not an invoke keyword and not a compile keyword; invoke() takes only input, thread_id, run_id, budget, and passing interrupt_before= raises TypeError.

sharp edge Without a checkpointer at compile time, get_state raises ValueError("No checkpointer set").

What does update_state actually check?

Field names and declared types always; the node's write allowlist only when you say which node the edit is attributed to. Unknown fields raise WritePermissionError, wrong types raise StateTypeError, and claiming a node whose declared writes do not cover the field raises WritePermissionError. With no as_node, the update is type-checked only — a human editing state mid-run is not a node, so there is no allowlist to apply; as_node="write" means "record this as if write had done it", and write's declared writes are then exactly the right contract.

python

graph.update_state("t1", {"sent": "forged"}, as_node="write")
# WritePermissionError: update_state(as_node='write') wrote undeclared fields ['sent']; declared writes: ['draft']

sharp edge Residual gap, stated rather than papered over — with as_node=None LangGraph attributes the update to whichever node last ran, and GraphARC does not reproduce that inference, so the update is type-checked but not allowlisted. If you want the allowlist, name the node.

How do I run something a human has to approve?

Use a Session: the graph's state schema derives from SessionState, gated nodes are named on the GraphSpec, and the runtime holds the graph before the gated node runs. grapharc.session.demo ships a four-node graph (ingest -> plan -> apply -> report) with apply gated — importing the module is what registers the graph. TurnResult.nodes is the honest answer to "what executed": apply is absent from the first turn because its body never ran. SessionManager(root) puts a session store and a checkpoint store in one directory, so "resume this session elsewhere" is "point another SessionManager at the same directory".

python

with SessionManager(root) as manager:
    session = manager.create(GRAPH_NAME)
    session.send("summarise the incident")

    first = session.run({})
    print("status  :", first.status.value)      # awaiting_approval
    print("holding :", [(h.node, h.action) for h in first.approvals])

    session.decide(approved=True, decided_by="ops")
    second = session.run()
    print("ran     :", second.nodes)            # ('apply', 'report')

sharp edge The gate belongs to the session, not to the graph — session.graph.invoke(...) runs gated nodes with nothing holding them; do not hand the compiled graph to anything that will not go through a session.

sharp edge run() is synchronous and occupies its caller until the session stops; an async turn is buildable, just not built.

How do I reject a step without unwinding it?

decide(approved=False). Nothing the gated node would have done has happened yet, so a rejection costs nothing to honour — the graph walks past the node rather than running it and undoing it. Calling run() while a hold is open raises ApprovalRequired before the session claims anything, so a refused run() leaves the queue, the holds and the checkpoint exactly as it found them. The rejected node appears in turn.skipped and never in the append-only log; what did run is the node's outgoing edges, filed against its own pending task, which is what advances the graph, after which report reads state.decision and routes on the verdict.

python

try:
    session.run()
except ApprovalRequired as exc:
    print("refused :", exc)

session.decide(approved=False, decided_by="ops", reason="wrong quarter")
turn = session.run()
print("ran     :", turn.nodes)      # ('report',)
print("skipped :", turn.skipped)    # ('apply',)

How do I resume a session after the process restarts?

Call SessionManager.resume(session_id) in the new process. Nothing about a session lives in memory: status, queue, holds and audit trail are rows in SQLite, graph state is in the checkpointer, and the graph itself is rebuilt by name from the registry. The chapter proves this by spawning two genuinely separate interpreters against one directory. log is the proof, not the status — the demo graph appends its own node name on every execution, so a resume that quietly re-ran ingest and plan would show them twice; it shows each once.

python

SECOND = """
import json, os, sys
from grapharc.session import SessionManager
from grapharc.session.demo import GRAPH_NAME

with SessionManager(sys.argv[1]) as manager:
    session = manager.resume("incident-42")          # rebuilt from the store
    session.decide(approved=True, decided_by="ops")
    turn = session.run()
    print(json.dumps({"pid": os.getpid(), "status": turn.status.value,
                      "ran": turn.nodes, "log": turn.state["log"]}))
"""

sharp edge The resuming process must be able to build the graph — register_graph is per-process and a session record only stores a name, so a resume in a process that never imported the registering module fails with UnknownGraphError, loudly, which is intended. That is why the demo import appears in both children: the import is the registration.

How do I stop a session that is already running?

session.interrupt(reason). It is a durable row, not a signal, so it works from another thread or another process; the process driving the session reads it after the next superstep. The turn comes back with status interrupted, interrupted_by set to your reason, the nodes that ran, and turn.pending naming what is still queued — and a later run() resumes from there. A stop queued while the session is asleep is honoured before the next turn's first node; an interrupt is never silently lost.

python

def stopper():
    INSIDE_STEP_ONE.wait(timeout=10)
    session.interrupt("operator asked for a stop")
    STOP_RECORDED.set()

threading.Thread(target=stopper).start()
turn = session.run({})
print("status       :", turn.status.value)   # interrupted
print("still pending:", turn.pending)        # ('two',)

sharp edge The boundary an interrupt is read at is the next one after it is written, and you do not control which one that is unless you synchronise (the snippet's STOP_RECORDED handshake is what keeps it illustrative rather than flaky). Cutting a node off mid-flight is what Budget(max_seconds=...) is for.

What does a hold mean when the gated node is fanned out?

It means one task, and you cannot tell which. A Send fan-out can put the same gated node on the boundary several times over; each task is held separately and each needs its own decision, so the count is exact and nothing runs unapproved — but the requests are indistinguishable, right down to the action text. Approving k of n runs k of them; which k is not yours to pick (an artefact of task ordering, not a guarantee). If the choice matters, gate a node that reads the decision from state, or fan out over distinct node names.

python

register_graph("mailshot", build, approval_nodes=("send",), replace=True)

holds = session.pending_approvals
session.decide(approved=True, request_id=holds[0].id)
session.decide(approved=True, request_id=holds[1].id)
session.decide(approved=False, request_id=holds[2].id, reason="bounced")

sharp edge request_id is required once more than one hold is open — decide() without it raises rather than guessing, because guessing is how an operator signs off send_email and releases delete_records.

sharp edge Only one verdict reaches the graph — SessionState.approval_decision is a single channel, so a boundary settling several holds shows the graph the first hold's verdict; every verdict is on the event log and in TurnResult.decisions.

sharp edge A Send payload that is a Pydantic model goes into the checkpoint and LangGraph prints Deserializing unregistered type <module>.Envelope from checkpoint. This will be blocked in a future version. on stderr — stderr noise today, a future error; the fix is LangGraph's allowed_msgpack_modules, not anything in GraphARC.

How do I put this behind HTTP?

grapharc.server.create_app(registry=...) returns a FastAPI app. A request may name a graph the operator registered and supply input and a budget; it may not describe a graph. Routes: POST /sessions (create and start, 201 + location), GET /sessions, GET /sessions/{id} (status, result, live meter reading), POST /sessions/{id}/events (message/approval/interrupt → 202), GET /sessions/{id}/stream (SSE: trace frames, then status, then done), GET /sessions/{id}/trace (JSONL as application/x-ndjson), GET /healthz (liveness, version, registered graph names). The stream takes a cursor query parameter and honours last-event-id, so a dropped SSE connection resumes without replaying frames you already saw.

python

registry = GraphRegistry({"qa": build_qa})
app = create_app(registry=registry)

with TestClient(app) as client:
    created = client.post(
        "/sessions", json={"graph": "qa", "input": {"question": "how do budgets work?"}}
    )
    session_id = created.json()["id"]
    for _ in range(200):
        view = client.get(f"/sessions/{session_id}").json()
        if view["status"] in ("succeeded", "failed", "interrupted"):
            break
        time.sleep(0.01)

sharp edge Read applied, not the status code — POST /events answers 202 for everything; whether it changed the run is event.applied, and this runtime only applies interrupt (message/approval are recorded with applied: false plus a detail).

sharp edge A stream frame and a trace line are the same record — BroadcastRecorder shapes each event once, including the trace format's 2000-character clip on long values; the unclipped answer is the session's result, never truncated.

sharp edge This default runtime does not resume across a process restart (sessions live in this process's memory; the trace files outlive them) and does not evict — every session and its event list is retained for the life of the process. For durable sessions use grapharc.session.

Which checkpointer does the server need?

An async one, if your nodes are async def. The server drives runs through astream, and SqliteSaver — this repo's own dependency, and what grapharc.session uses — has no async side. The runtime probes the saver once per run, on the run's own event loop, before anything executes; a sync-only saver quietly selects the sync driver instead. The one combination nothing can serve is async def nodes plus a sync-only saver, which yields CheckpointerNotAsyncError. Matrix: sync nodes with none/SqliteSaver run on the sync driver; sync nodes with InMemorySaver/AsyncSqliteSaver run async; async def with none/InMemorySaver/AsyncSqliteSaver runs async; async def with SqliteSaver errors.

python

registry = GraphRegistry(
    {
        "sync_saver": lambda trace: graph_with(SqliteSaver(conn), trace),
        "async_ok": lambda trace: graph_with(InMemorySaver(), trace),
    }
)

sharp edge AsyncSqliteSaver needs aiosqlite and must be constructed on a running loop; InMemorySaver is the zero-dependency answer when durability is not the point.

How do I actually serve it?

grapharc serve --registry module:attr, where the attribute is a grapharc.server.GraphRegistry (or a callable returning one). Without --registry the app starts with an empty registry — a legitimate way to check the process comes up and a useless way to run anything, and the CLI tells you which of the two you got. The banner (URL, graph names, "ctrl-c to stop") is printed before the server blocks, so a script watching stdout for the URL does not have to wait for the process to exit. To use a real model instead of the scripted one, swap the model in the builder with grapharc.gateway.get_model.

console

$ PYTHONPATH=. grapharc serve --registry mygraphs:REGISTRY --port 8124
serving grapharc.server on http://127.0.0.1:8124
graphs   : qa
ctrl-c to stop

$ curl -s localhost:8124/healthz
{"status":"ok","version":"0.1.0","graphs":["qa"]}

$ curl -s -X POST localhost:8124/sessions -H 'content-type: application/json' \
      -d '{"graph":"qa","input":{"question":"how do budgets work?"}}'
{"id":"38231fa41b8b4dad","graph":"qa","thread_id":"38231fa41b8b4dad","status":"queued", ...}

python

from grapharc.gateway import get_model

model = get_model("openrouter/anthropic/claude-haiku-4.5")
# or, on a Claude subscription with the CLI on PATH and no API key:
model = get_model("claude-cli/claude-sonnet-5")

sharp edge serve needs the server extra: uv sync --extra server.

sharp edge The real-model snippet was not executed — it needs a paid API key or a Claude subscription; grapharc models --check tells you which backends this machine could use without contacting any provider.

How do I reconstruct a run after it finished?

replay(trace, run_id). It is a reconstruction, not a re-execution: it reads the JSONL and rebuilds the node sequence, the deltas, the timings and the failures — nothing here calls a model, a tool, or a node. run.path, run.tokens, run.replay_state() and format_replay(run) are the readouts. Passing run_id= to invoke() is what makes a run findable later; without it you get a random hex id and have to fish it out of the trace file. replay_thread(trace, thread_id) gives you every run recorded against one thread, which is how you read a resumed session rather than its last attempt.

python

trace = TraceRecorder(path)
build(trace, ["budgets cap iterations"]).invoke({"question": "budgets?"}, run_id="before")

run = replay(trace, "before")
print("path   :", run.path)
print("tokens :", run.tokens)
print("state  :", run.replay_state())
print(format_replay(run))

sharp edge replay_state() is only as good as the deltas — the trace truncates strings past 2000 characters at write time, so a long value replays truncated.

sharp edge The trace does not record which state fields have reducers, so an appended field replays as last-write-wins unless you pass run.replay_state(reducers={"log": operator.add}).

How do I tell whether my change altered behaviour?

diff_trace(trace, run_a, run_b) aligns the two node sequences with difflib and reports where they diverged — which nodes wrote different deltas and which state fields differ — with format_diff for the readout and diff.identical for the verdict. identical deliberately ignores timing and tokens: two runs of a deterministic graph differ by milliseconds every time, and a diff that is never clean is a diff nobody reads. It compares path, deltas and termination reason; the token and duration numbers are still on the RunDiff if you want them.

python

diff = diff_trace(trace, "before", "after")
print("identical:", diff.identical)
print(format_diff(diff))

sharp edge Diffing runs of different graphs raises ReplayError rather than producing an alignment that is technically correct and means nothing.

Where did the tokens go?

attribute(trace, run_id, rates=...) splits a run's spend per node; a RateCard is USD per 1,000 tokens, blended across input and output, because a trace event records one tokens total and no split. summarize(trace, run_id) gives RunMetrics (nodes executed, tokens, per-node counts) and to_mermaid(trace, run_id) renders the executed path. RunCost.tokens and RunMetrics.tokens agree by construction — both count the end events of node executions, and the test suite asserts they match, because a cost report and an audit trail that disagree are worse than either alone. attribute_thread(trace, thread_id) does the same for a whole session across resumes, and by_node(trace) ranks every node in a file by cost.

python

cost = attribute(trace, "r1", rates=RateCard(default=3.0))  # USD per 1k tokens, blended
for node in cost.per_node:
    print(f"{node.node:<8} {node.tokens:>3} tok  ${node.estimated_cost_usd:.6f}")
print("total   ", cost.tokens, "tok  complete:", cost.complete)

metrics = summarize(trace, "r1")
print(to_mermaid(trace, "r1"))

sharp edge Nothing in GraphARC writes cost_usd onto a trace event today — the field exists and TraceRecorder.event accepts it, but the kernel does not pass it, so recorded_cost_usd is always None; everything shown is estimated_cost_usd (tokens × your rate card), in its own field so nobody mistakes an estimate for an invoice.

sharp edge complete is what tells you the total is a total — tokens with neither a recorded cost nor a matching rate land in unpriced_tokens, and a non-zero count means cost_usd is a lower bound; a RateCard with no default and no matching model prices nothing.

sharp edge A node that raised has no token count at node level (no end event); for an AgentNode the per-call sub-events still hold the spend as tokens_before_error, kept out of the total so the total keeps matching metrics.

sharp edge to_mermaid renders the executed path keyed by (node, step), so parallel fan-out instances are distinct boxes rather than one box with a self-loop the graph never had.

The CLI tour

Eleven commands, every one taking --json, which prints the same payload as one document on stdout — including failures, which become the document rather than a line on stderr. The set: demo <example> (stage0…stage6, capstone), run <graph.json> (through the admission gate; --check-only lints it), plan <goal> (governed loop: propose → admit → execute → replan), agent <task>, serve, models [spec] (--check probes this machine), replay <trace> <run>, diff <trace> <a> <b>, trace <trace>, metrics <trace> <run>, viz <trace> <run>. Exit codes are part of the interface: 0 did the job, 1 ran and the answer was negative (two runs differed, a run id had no events, no backend was usable), 2 could not run at all (missing file, missing component, unknown model spec). --allow / --deny / --ask on grapharc agent are repeatable tool-name globs (--deny wins), --executor local drops the sandbox, and --max-turns / --max-tokens / --max-seconds are the run's budget.

console

$ grapharc demo stage1 --trace trace.jsonl
$ grapharc trace trace.jsonl --json | jq -r '.events[0].run_id'
2a47f18064b7
$ grapharc metrics trace.jsonl 2a47f18064b7
run_id: 2a47f18064b7
graph: stage1_loop
nodes_executed: 8
tokens: 81
termination_reason: target_met
per_node: {'start': 1, 'plan': 2, 'act': 2, 'verify': 2, 'finish_target_met': 1}
$ grapharc diff trace.jsonl 2a47f18064b7 3c0d1b4b4b3e; echo "exit $?"
2a47f18064b7 == 3c0d1b4b4b3e: same path (8 nodes), same state
exit 0

console

$ grapharc models --check
claude-cli   usable    'claude' on PATH at /home/you/.local/bin/claude
                       credential: claude subscription login (no API key)
openrouter   unusable  no API key (set OPENROUTER_API_KEY, or add one to .env)
                       credential: <unset>
ollama       usable    local server at http://localhost:11434/v1
mock         usable    scripted test double; never reaches a provider

local probe only — no provider was contacted, so a configured key
is not a validated one.

sharp edge grapharc run does not print the run id — it prints the trace path, and you get the id out of the file (jq -r '.events[0].run_id' or a one-line python -c); everything downstream (metrics, viz, replay, diff) wants that id.

sharp edge models --check is a local probe: it reports that a credential is configured, never that it is valid, in credit, or entitled to a model; it exits 1 when no real provider is usable — mock alone does not count.

sharp edge grapharc agent always needs a model, so there is no scripted form and nothing in the chapter claims to have run it.

Source file: /home/shashank/Desktop/GraphARC/docs/cookbook/06-serving-and-ops.md (1709 lines); paired test: /home/shashank/Desktop/GraphARC/tests/test_cookbook_serving.py.