Agent State Machines and Durable Graphs

State, nodes, edges, reducers, checkpoints, interrupts and leased job recovery. LangGraph and a Go job queue.

The parts of a state graph#

State: a typed dict that flows through the graph. Each node gets the full state.
Node: a function state -> partial_state. One discrete step.
Edge: static edges go to one node. Conditional edges call a router function.
Reducer: (old, new) -> merged per field. Default is overwrite.
Thread: a thread_id string that scopes all checkpoints of one session.
Checkpoint: the full state after a node, keyed on (thread_id, checkpoint_id).
Send: a node returns Send(node_name, state) to run nodes in parallel.
Subgraph: a compiled graph used as one node in another graph.

class State(TypedDict):

messages: Annotated[list[AnyMessage], add_messages]append, not overwrite

def should_continue(state: State) -> str:

last = state["messages"][-1]

return "tools" if getattr(last, "tool_calls", None) else END

graph = StateGraph(State)

graph.add_node("agent", agent_node)calls the LLM

graph.add_node("tools", ToolNode(TOOLS))runs tool_calls

graph.set_entry_point("agent")

graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})

graph.add_edge("tools", "agent")static edge back

app = graph.compile(checkpointer=MemorySaver(),

interrupt_before=["tools"])pause before side effects

Task	Code
Set the thread	`{"configurable": {"thread_id": "<id>"}}`
Stream deltas	`app.stream(<state>, config, stream_mode="updates")`
Read paused state	`app.get_state(config)`
Approve and resume	`app.invoke(Command(resume=True), config)`
Deny with a message	`app.update_state(config, {"messages": [...]})`
List checkpoints	`list(app.get_state_history(config))`
Replay from a checkpoint	`app.stream(None, <old_config>)`
Postgres store	`PostgresSaver.from_conn_string(<url>)`

Factor	`updates`	`messages`	`values`
Yields	State deltas	LLM tokens	Full snapshots
Shape	`{node_name: state_delta}`	Tokens in model nodes	Whole state
Use for	The UI	Token streaming	Eval

Command	What it does
`go build -o /tmp/durable-jobs .`	Builds the reference worker
`enqueue --store "$STORE" --input samples/jobs.json`	Adds id/text records
`worker --store "$STORE" --max-jobs 100`	Claims and runs jobs
`status --store "$STORE"`	Shows jobs, effects and scope
`worker ... --crash-after effect`	Exits 86 after the receipt
`worker ... --now-ms 111 --lease-ms 10`	Recovers the expired claim
`--crash-after claim`	Crash before any output exists
`go run . demo`	Automated effect crash sequence

Symptom	Cause	Fix
Half the turn lost	No `add_messages` reducer	`Annotated[list, add_messages]`
No resume or time-travel	No checkpointer	`MemorySaver` in tests, a real store in prod
Harm before approval	Interrupt after the side effect	Interrupt before the side effect node
Tool state lost on resume	Checkpoints too small	Serialize the full state
Resume gives a new result	Non-deterministic node	Capture seeds, clock and API replies
Old worker completion fails	Version or lease changed	Expected. `Finish` fences stale workers
Enqueue `conflict`	Changed text under an old ID	Use a new identifier
Exit code 86	`--crash-after` set	Expected. Run the worker again

What does None as input do?: It replays from the given checkpoint.
Why completed v4 after one crash?: Claim, reclaim, claim and finish each add 1.
Is execution exactly once?: No. It is at least once.
Why no duplicate receipt?: The local immutable receipt makes the effect idempotent.
When does Reclaim fire?: On a running job with now at or past LeaseUntil.
Which three topologies?: Supervisor, swarm and hierarchical.