Contract Tests for a LangGraph Agent: Test the Property, Not the Setting

Here is a preview of the test suite you will build, run against the LangGraph research graph you start with:

text

.FF.FF...FF                                                              [100%]=========================== short test summary info ===========================FAILED tests/test_crash.py::test_crash_budget[sync] - AssertionError: ran 37FAILED tests/test_crash.py::test_crash_budget[async] - AssertionError: ran 37FAILED tests/test_merge.py::test_split[compliance] - AssertionError: approvedFAILED tests/test_merge.py::test_split[reversed] - AssertionError: approvedFAILED tests/test_reentry.py::test_thread_budget - AssertionError: ran 78FAILED tests/test_width.py::test_dispatch_capped - AssertionError: ran 10006 failed, 5 passed in 42.72s

The two crash counts can differ on your machine, for reasons Step 9 explains. Here is the same suite against the graph you finish with:

text

...........                                                              [100%]11 passed in 30.77s

By the end you will have that suite and a graph that passes it. The suite checks four properties of a LangGraph agent: how much work one fan-out can create, whether a merge depends on node names, whether a spend budget survives resumes and new turns, and whether that budget survives a process crash. It reaches the graph through one adapter file, so pointing it at your own graph means rewriting that file and nothing else.

This is an intermediate tutorial. You should already know Python, basic pytest, and LangGraph's StateGraph, Send, reducers and checkpointers. If those are new, start with LangGraph or a While Loop? You Already Have a Runtime. It takes about 45 minutes, and nothing in it calls a model or needs an API key.

Verified against Python 3.12.2, langgraph==1.2.11, langgraph-checkpoint==4.2.0, langgraph-checkpoint-sqlite==3.1.1, pytest==9.1.1 on 2026-09-14.

Prerequisites

Python 3.12 (3.13 also works; langgraph 1.2.11 lists 3.10 to 3.13)
langgraph==1.2.11
langgraph-checkpoint==4.2.0 (installed by langgraph anyway, pinned because the tests use its InMemorySaver)
langgraph-checkpoint-sqlite==3.1.1 (a separate package, needed for the crash test in Step 9)
pytest==9.1.1

You need no accounts and no API keys. A stub stands in for every paid call.

On macOS or Linux:

bash

mkdir contract-tests && cd contract-testspython3.12 -m venv .venvsource .venv/bin/activatepip install langgraph==1.2.11 langgraph-checkpoint==4.2.0pip install langgraph-checkpoint-sqlite==3.1.1 pytest==9.1.1mkdir agent teststouch agent/__init__.py

On Windows (PowerShell):

powershell

mkdir contract-tests; cd contract-testspy -3.12 -m venv .venv.venv\Scripts\activatepip install langgraph==1.2.11 langgraph-checkpoint==4.2.0pip install langgraph-checkpoint-sqlite==3.1.1 pytest==9.1.1mkdir agent, testsNew-Item agent/__init__.py

From here on, run every command from the contract-tests directory with the virtual environment active. Create pytest.ini there. It lets tests import agent from the project root and keeps pytest's output to one line per failure:

ini

[pytest]pythonpath = .testpaths = testsaddopts = -q --tb=no -rf

The multi-line python -c "..." commands in this tutorial work unchanged in PowerShell. Verify the install before you write any code:

bash

python -c "from importlib.metadata import versionfor p in ('langgraph', 'langgraph-checkpoint', 'langgraph-checkpoint-sqlite', 'pytest'):    print(p, version(p))"

text

langgraph 1.2.11langgraph-checkpoint 4.2.0langgraph-checkpoint-sqlite 3.1.1pytest 9.1.1

The run time pytest prints at the end of each run will differ from the ones on this page. Everything else in the output should match.

How a pytest property test differs from a setting check

A super-step is one tick of the graph: every node that is ready runs, and then all of their writes are applied together. recursion_limit counts these ticks, not nodes or tasks.

Several LangGraph settings have names that sound like the guarantee you want: recursion_limit for a bound on work, a checkpointer for safe resume, a reducer for a defined merge. In Graph Engineering Draws the Graph. Your Runtime Redraws It. I measured each of these on langgraph 1.2.11 and found that each one enforces something nearby: super-steps per invocation rather than tasks, a step budget that comes back at full on resume, and a merge whose result follows node names.

A setting check asks "is recursion_limit set?" A property test asks "when I request 1,000 research tasks, how many run?" The second test does not care what any setting is called, so it catches the gap between the name and the behaviour.

This tutorial calls these tests contract tests. Martin Fowler uses the term for tests that check whether an external dependency still behaves the way your code assumes, and you re-run them whenever that dependency changes. Here the dependency is the LangGraph runtime, and the change is a LangGraph upgrade. The suite is also a worked version of the checklist at the end of LangGraph Evals Test the Answer, Not the Thread: a real checkpointer, a reused thread_id, and budgets checked when the same thread is entered a second time.

Four properties, and the setting that looks like it provides each one:

Property you want	Setting that sounds like it	What the test measures
One dispatch creates at most 8 research tasks	`recursion_limit`	request 1,000 tasks, count the calls that ran
A split vote is never approved, whatever the nodes are called	an `operator.add` reducer	rename the reviewer nodes, check the decision
One thread spends at most 15 research calls, ever	`recursion_limit` plus a checkpointer	resume twice and start a new turn, count the calls
That budget holds when the process dies mid-call	`durability="sync"` (when checkpoints are written)	kill the process after a call, resume, count

The suite has three layers. The graph under test lives in agent/, and the tests live in tests/ and never import it. Between them sits tests/adapter.py, which builds the graph, counts its paid calls and runs it. That file is the only place the contract touches your code.

Step 1: Build the LangGraph research graph under test

Goal

Write the research graph the suite will test, with no protections at all.

Why this step

A test you have only seen pass might be a test that cannot fail, so start with a graph that fails every property. This one is a smaller version of the graph in the graph engineering article. A planner picks sources, Send fans out one research task per source, two reviewers vote in parallel, and a gate either ends the run or loops back to the planner.

Two things are stubbed so the tests run without a model. requested_width stands in for the number of sources a planner model would choose. search is any object with a fetch(target) method. It plays the part of a paid search API, and the tests will replace it with a counter.

Code

Create agent/graph.py:

python

# agent/graph.pyimport operatorfrom typing import Annotated, TypedDictfrom langgraph.graph import END, START, StateGraphfrom langgraph.types import Sendclass State(TypedDict):    requested_width: int                          # the planner model's choice, stubbed    targets: list[str]    findings: Annotated[list[str], operator.add]    verdicts: Annotated[list[str], operator.add]    approved: booldef build(checkpointer, search, reviewers=("facts_review", "policy_review"),          votes=("approve", "reject")):    def plan(state):        return {"targets": [f"source-{i}" for i in range(state["requested_width"])]}    def fan_out(state):        return [Send("research", {"target": t}) for t in state["targets"]]    def research(payload):        return {"findings": [search.fetch(payload["target"])]}    def reviewer(vote):        return lambda state: {"verdicts": [vote]}    def gate(state):        return {"approved": state["verdicts"][-1] == "approve"}    def decide(state):        return END if state["approved"] else "plan"    g = StateGraph(State)    g.add_node("plan", plan)    g.add_node("research", research)    g.add_node("gate", gate)    g.add_edge(START, "plan")    g.add_conditional_edges("plan", fan_out, ["research"])    for name, vote in zip(reviewers, votes):        g.add_node(name, reviewer(vote))        g.add_edge("research", name)    g.add_edge(list(reviewers), "gate")    g.add_conditional_edges("gate", decide, ["plan", END])    return g.compile(checkpointer=checkpointer)

build takes the reviewer node names and their votes as arguments. The merge test in Step 5 has to rename nodes, and you cannot rename nodes in a compiled graph, so the names go in before compile. g.add_edge(list(reviewers), "gate") is a join: gate runs only after both reviewers have finished, and the two reviewers run in the same super-step.

Run it

Draw the compiled graph:

bash

python -c "from langgraph.checkpoint.memory import InMemorySaverfrom agent.graph import buildprint(build(InMemorySaver(), None).get_graph().draw_mermaid(), end='')"

Expected output

text

---config:  flowchart:    curve: linear---graph TD;	__start__([<p>__start__</p>]):::first	plan(plan)	research(research)	gate(gate)	facts_review(facts_review)	policy_review(policy_review)	__end__([<p>__end__</p>]):::last	__start__ --> plan;	facts_review --> gate;	gate -.-> __end__;	gate -.-> plan;	plan -.-> research;	policy_review --> gate;	research --> facts_review;	research --> policy_review;	classDef default fill:#f2f0ff,line-height:1.2	classDef first fill-opacity:0	classDef last fill:#bfb6fc

What just happened

The graph compiles, and the topology is the one you meant: plan, research, two reviewers, a gate that loops. Now look at plan -.-> research. That single dotted edge is the whole fan-out, and it is drawn the same way whether the planner requests 3 tasks or 1,000. Nothing in this drawing tells you how much work one run does. The next nine steps measure it instead.

Step 2: Write a pytest adapter so the tests never import the graph

Goal

Give the test suite one file that knows how to build, run and meter the graph.

Why this step

If every test imports agent.graph and calls build directly, the suite is welded to this graph. Put that knowledge in tests/adapter.py instead, and the test files depend only on the adapter's functions. To test a different graph later, you rewrite the adapter and keep the tests.

The adapter also holds the contract's numbers. MAX_FANOUT and BUDGET are what your team promised, written where the tests can read them. Do not import them from the graph. A test that reads its limit from the code under test passes whenever the code agrees with itself.

Code

Create tests/adapter.py:

python

# tests/adapter.py - the only file in tests/ that knows which graph it is testingimport threadingfrom pathlib import Pathfrom langgraph.errors import GraphRecursionErrorfrom agent.graph import build# The contract: numbers your team promised, not settings the graph happens to have.MAX_FANOUT = 8      # most research tasks one dispatch may createBUDGET = 15         # most research tasks one thread may ever runREVIEWERS = ("facts_review", "policy_review")SPLIT = ("approve", "reject")UNANIMOUS = ("approve", "approve")class CountingSearch:    """Stands in for a paid search API. Each call appends one line to a file."""    def __init__(self, path):        self.path = Path(path)        self.lock = threading.Lock()    def fetch(self, target):        with self.lock, self.path.open("a") as f:            f.write(target + "\n")        return f"notes on {target}"    def count(self):        return len(self.path.read_text().splitlines()) if self.path.exists() else 0def calls(workdir):    return CountingSearch(Path(workdir) / "calls.txt").count()def make_graph(workdir, checkpointer, reviewers=REVIEWERS, votes=SPLIT, search=None):    search = search or CountingSearch(Path(workdir) / "calls.txt")    return build(checkpointer, search, reviewers=reviewers, votes=votes)def settle(app, inputs, thread_id, recursion_limit=25, durability=None):    """Run one invocation and return the saved state, however the run stopped."""    config = {"recursion_limit": recursion_limit,              "configurable": {"thread_id": thread_id}}    try:        app.invoke(inputs, config, durability=durability)    except GraphRecursionError:        pass    return app.get_state(config)def approved(snapshot):    return snapshot.next == () and snapshot.values.get("approved") is True

Create tests/test_adapter.py, a smoke test that proves the adapter works before you trust it:

python

# tests/test_adapter.pyfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import UNANIMOUS, approved, calls, make_graph, settledef test_adapter_runs_one_approved_round(tmp_path):    app = make_graph(tmp_path, InMemorySaver(), votes=UNANIMOUS)    snapshot = settle(app, {"requested_width": 3}, "smoke")    assert approved(snapshot)    assert calls(tmp_path) == 3

Run it

bash

pytest

Expected output

text

.                                                                        [100%]1 passed in 0.55s

What just happened

The rest of the tutorial leans on four decisions in the adapter:

CountingSearch meters paid calls in a file rather than in memory, because a list in memory vanishes when its process dies and Step 9 kills processes on purpose. calls(workdir) reads the count back. The lock is there because LangGraph runs the tasks of one Send fan-out in parallel threads, and their appends to the file must not interleave.
settle swallows GraphRecursionError and returns the saved state. A property test cares how much work ran and how the run ended, not whether the runtime complained on the way. It also passes recursion_limit=25 every time, so all tests run under the same limit. LangGraph's own default is far higher, and its documentation (1000) and its 1.2.11 source (10007) disagree on the number, so a test relying on it would take much longer to stop.
approved requires a finished run (next == () means nothing is left to execute) and a recorded approval. A run that the recursion limit stopped is not an approval.
make_graph accepts a search override, and settle accepts a durability argument. Nothing uses either yet. durability controls when checkpoints are written, and None keeps LangGraph's default, "async". Step 9 explains the modes and passes in a search that crashes its own process.

tests/ has no __init__.py, so pytest puts the tests directory itself on the import path. That is what lets the tests write from adapter import ....

Step 3: Test fan-out width by requesting 1,000 tasks in one dispatch

Goal

Write a test that fails when one fan-out can create more than MAX_FANOUT research tasks.

Why this step

A code reviewer who sees recursion_limit=25 in a config can easily read it as a bound on work. LangGraph counts super-steps, the ticks defined above, and every task in a Send fan-out runs in the same super-step. A fan-out of any width costs one step against the limit. The only way to learn how many tasks one dispatch creates is to ask for a lot of them and count.

The test uses a unanimous vote, so the graph approves after one round and every call it counts came from a single fan-out.

Code

Create tests/test_width.py:

python

# tests/test_width.pyfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import MAX_FANOUT, UNANIMOUS, calls, make_graph, settledef test_dispatch_capped(tmp_path):    app = make_graph(tmp_path, InMemorySaver(), votes=UNANIMOUS)    settle(app, {"requested_width": 1000}, "width")    ran = calls(tmp_path)    assert ran <= MAX_FANOUT, f"ran {ran}"

Run it

bash

pytest tests/test_width.py

Expected output

text

F                                                                        [100%]=========================== short test summary info ===========================FAILED tests/test_width.py::test_dispatch_capped - AssertionError: ran 10001 failed in 3.40s

What just happened

All 1,000 requested searches ran under a recursion_limit of 25. You now have a red test that proves the gap exists, and you have not changed the graph yet. Every property after this one follows the same order: write the test, watch it fail for the reason you expect, then fix.

Step 4: Cap the fan-out where Send creates the tasks

Goal

Make test_dispatch_capped pass by capping width at the one place tasks are dispatched.

Why this step

Put the cap where the work is created. In this graph that is fan_out, the only function that builds Send objects. A cap placed anywhere else can be bypassed by any code path that dispatches without going through it.

Code

In agent/graph.py, add a constant below the imports:

python

from langgraph.types import SendMAX_FANOUT = 8      # most research tasks one dispatch may createclass State(TypedDict):

Then replace fan_out inside build with:

python

    def fan_out(state):        return [Send("research", {"target": t}) for t in state["targets"][:MAX_FANOUT]]

Run it

bash

pytest

Expected output

text

..                                                                       [100%]2 passed in 0.63s

What just happened

The planner still records 1,000 targets in state, but only the first 8 become tasks, so the smoke test and the width test both pass. This graph has one dispatch site, and one slice covers it. If another node also routes by returning Command(goto=[Send(...)]), which lets a node choose its own next step, that is a second dispatch site. It needs its own cap, and its own width test to exercise it.

Step 5: Test that a reducer merge does not depend on node names

Goal

Write a test that fails when renaming a reviewer node changes whether a split vote is approved.

Why this step

Both reviewers write verdicts in the same super-step, and operator.add concatenates their writes. LangGraph applies the writes from one super-step in an order sorted by task path. A task path identifies one task inside a super-step. For a node that runs because an edge points at it, like these reviewers, the path is essentially the node name, so the writes are sorted by node name. gate reads verdicts[-1], the last write. Which vote counts therefore depends on which reviewer's name sorts later.

Do not build on that order. LangGraph's documentation warns that "updates from a parallel superstep may not be ordered consistently". What you want is a decision that stays the same whatever the nodes are called, so the test runs a split vote and a unanimous vote under three sets of names and requires the right answer every time.

Code

Create tests/test_merge.py:

python

# tests/test_merge.pyimport pytestfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import SPLIT, UNANIMOUS, approved, make_graph, settleNAMES = [    ("facts_review", "policy_review"),    ("facts_review", "compliance_review"),    ("zz_facts", "aa_policy"),]IDS = ["policy", "compliance", "reversed"]@pytest.mark.parametrize("reviewers", NAMES, ids=IDS)def test_split(tmp_path, reviewers):    app = make_graph(tmp_path, InMemorySaver(), reviewers=reviewers, votes=SPLIT)    result = approved(settle(app, {"requested_width": 1}, "split"))    assert not result, "approved"@pytest.mark.parametrize("reviewers", NAMES, ids=IDS)def test_unanimous(tmp_path, reviewers):    app = make_graph(tmp_path, InMemorySaver(), reviewers=reviewers, votes=UNANIMOUS)    result = approved(settle(app, {"requested_width": 1}, "unanimous"))    assert result, "not approved"

In a split vote the first reviewer always votes approve and the second always votes reject. NAMES is a list and not a generator, because pytest 9.1 deprecates generators as parametrize values.

Run it

bash

pytest tests/test_merge.py

Expected output

text

.FF...                                                                   [100%]=========================== short test summary info ===========================FAILED tests/test_merge.py::test_split[compliance] - AssertionError: approvedFAILED tests/test_merge.py::test_split[reversed] - AssertionError: approved2 failed, 4 passed in 1.09s

What just happened

The same split vote got two different decisions. With policy_review the rejection sorts last, so the gate rejects and the graph loops until the recursion limit stops it. The run is never approved, and that case passes. With compliance_review the approval sorts last and the run ends approved, and renaming to zz_facts and aa_policy puts the approval last as well. All the unanimous cases pass, which tells you the failures come from merge order and not from a broken graph.

I chose these names knowing how LangGraph sorts. For your own graph you do not need to know. Vary every input that should not change the answer, node names among them, and require the same answer each time.

Step 6: Give each reviewer its own key and make the gate fail closed

Goal

Make the merge test pass by removing the dependence on write order.

Why this step

Sorting the writes some other way would still depend on order, so take order out of the question. Each reviewer writes its own key. operator.or_ merges the dictionaries, and because the keys never collide, the result is the same dictionary in whichever order the writes are merged.

The gate then checks each required role by name, and a missing or renamed role counts as a refusal. It has to fail closed: a check over "whatever verdicts are present" would approve an empty dictionary.

Code

Replace agent/graph.py with:

python

# agent/graph.pyimport operatorfrom typing import Annotated, TypedDictfrom langgraph.graph import END, START, StateGraphfrom langgraph.types import SendMAX_FANOUT = 8      # most research tasks one dispatch may createROLES = ("facts", "policy")class State(TypedDict):    requested_width: int                          # the planner model's choice, stubbed    targets: list[str]    findings: Annotated[list[str], operator.add]    verdicts: Annotated[dict[str, str], operator.or_]   # one key per reviewer role    approved: booldef build(checkpointer, search, reviewers=("facts_review", "policy_review"),          votes=("approve", "reject")):    def plan(state):        return {"targets": [f"source-{i}" for i in range(state["requested_width"])]}    def fan_out(state):        return [Send("research", {"target": t}) for t in state["targets"][:MAX_FANOUT]]    def research(payload):        return {"findings": [search.fetch(payload["target"])]}    def reviewer(role, vote):        return lambda state: {"verdicts": {role: vote}}    def gate(state):        verdicts = state.get("verdicts", {})        return {"approved": all(verdicts.get(r) == "approve" for r in ROLES)}    def decide(state):        return END if state["approved"] else "plan"    g = StateGraph(State)    g.add_node("plan", plan)    g.add_node("research", research)    g.add_node("gate", gate)    g.add_edge(START, "plan")    g.add_conditional_edges("plan", fan_out, ["research"])    for name, role, vote in zip(reviewers, ROLES, votes):        g.add_node(name, reviewer(role, vote))        g.add_edge("research", name)    g.add_edge(list(reviewers), "gate")    g.add_conditional_edges("gate", decide, ["plan", END])    return g.compile(checkpointer=checkpointer)

The changes from Step 4 are ROLES, the verdicts type and reducer, reviewer taking a role, the gate check, and the zip over three sequences.

Run it

bash

pytest

Expected output

text

........                                                                 [100%]8 passed in 0.75s

What just happened

Node names now decide nothing. A reviewer's role comes from its position in reviewers, and its vote lands under that role's key. Under any names, a split vote loops until the recursion limit without ever being approved, and a unanimous vote ends approved.

Some graphs genuinely need the writes in a particular order, not just a result that ignores order. For those, LangGraph's documentation suggests an explicit ordering field that you sort at the join.

Step 7: Test that a spend budget survives resume and a new turn

Goal

Write a test that fails when one thread can run more than BUDGET research calls across several invocations.

Why this step

A split vote loops forever, so something has to stop it. With a width of 3, recursion_limit=25 stops the first invocation after a bounded number of searches, and it is tempting to treat that as the thread's spend limit. It is not one. LangGraph computes the step allowance fresh, from the step it resumes at, every time the graph is entered. Each invocation on the thread gets a new allowance.

One round of this graph is four super-steps: plan, research, the two reviewers together, and gate. At a width of 3, a limit of 25 allows about six rounds, or roughly 18 searches, before it stops the invocation.

Threads get re-entered in ordinary ways: a resume with invoke(None, config) after the limit, or a new message on the same thread_id. The test does both and counts every call the thread ever made.

Code

Create tests/test_reentry.py:

python

# tests/test_reentry.pyfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import BUDGET, SPLIT, calls, make_graph, settledef test_thread_budget(tmp_path):    app = make_graph(tmp_path, InMemorySaver(), votes=SPLIT)    settle(app, {"requested_width": 3}, "t1")   # first run stops at the limit    settle(app, None, "t1")                     # resume after the limit    settle(app, None, "t1")                     # resume again    settle(app, {"requested_width": 3}, "t1")   # a new turn on the same thread    ran = calls(tmp_path)    assert ran <= BUDGET, f"ran {ran}"

Run it

bash

pytest tests/test_reentry.py

Expected output

text

F                                                                        [100%]=========================== short test summary info ===========================FAILED tests/test_reentry.py::test_thread_budget - AssertionError: ran 781 failed in 0.94s

What just happened

Four invocations made 78 searches against a budget of 15. When I printed the count after each one, they made 18, 21, 21 and 18, and the thread's saved step number went 25, 52, 79, 106. That number is snapshot.metadata["step"] from get_state. Each entry was allowed about 25 steps, counted from wherever the thread had stopped, not from zero. A fresh run spends two of them applying its input, and a resume has no input to apply, so each resume ran 27 steps and fit a seventh round. That is why the resumes made 21 searches instead of 18. The recursion limit did exactly what it was designed to do on each invocation, but no invocation knew what the others had spent. Nothing in the graph tracks spend yet.

Step 8: Add a spend counter in graph state

Goal

Make the re-entry test pass by tracking spend in graph state, the way most LangGraph guides recommend.

Why this step

The standard advice for a hard stop is a counter in state, and it is a real improvement over recursion_limit: it counts calls per thread, where the limit counts steps per invocation. The checkpointer saves state, so the counter survives a resume and a new turn. That is exactly what Step 7 tested. Build it and watch it pass, because Step 9 tests it again.

The planner shrinks each round's width to whatever budget is left. When nothing is left, fan_out routes to a budget_exhausted node that records a refusal and ends the run.

Code

Replace agent/graph.py with:

python

# agent/graph.pyimport operatorfrom typing import Annotated, TypedDictfrom langgraph.graph import END, START, StateGraphfrom langgraph.types import SendMAX_FANOUT = 8      # most research tasks one dispatch may createBUDGET = 15         # most research tasks one thread may ever runROLES = ("facts", "policy")class State(TypedDict):    requested_width: int                          # the planner model's choice, stubbed    targets: list[str]    findings: Annotated[list[str], operator.add]    verdicts: Annotated[dict[str, str], operator.or_]   # one key per reviewer role    approved: bool    spent: Annotated[int, operator.add]          # research tasks run on this threaddef build(checkpointer, search, reviewers=("facts_review", "policy_review"),          votes=("approve", "reject")):    def plan(state):        left = BUDGET - state.get("spent", 0)        width = max(0, min(state["requested_width"], left))        return {"targets": [f"source-{i}" for i in range(width)]}    def fan_out(state):        if not state["targets"]:            return "budget_exhausted"        return [Send("research", {"target": t}) for t in state["targets"][:MAX_FANOUT]]    def research(payload):        return {"findings": [search.fetch(payload["target"])], "spent": 1}    def reviewer(role, vote):        return lambda state: {"verdicts": {role: vote}}    def gate(state):        verdicts = state.get("verdicts", {})        return {"approved": all(verdicts.get(r) == "approve" for r in ROLES)}    def decide(state):        return END if state["approved"] else "plan"    g = StateGraph(State)    g.add_node("plan", plan)    g.add_node("research", research)    g.add_node("gate", gate)    g.add_node("budget_exhausted", lambda state: {"approved": False})    g.add_edge(START, "plan")    g.add_conditional_edges("plan", fan_out, ["research", "budget_exhausted"])    g.add_edge("budget_exhausted", END)    for name, role, vote in zip(reviewers, ROLES, votes):        g.add_node(name, reviewer(role, vote))        g.add_edge("research", name)    g.add_edge(list(reviewers), "gate")    g.add_conditional_edges("gate", decide, ["plan", END])    return g.compile(checkpointer=checkpointer)

The changes from Step 6 are BUDGET, the spent key, the width calculation in plan, the empty-targets branch in fan_out, "spent": 1 in research, and the budget_exhausted node with its two edges.

Run it

bash

pytest

Expected output

text

.........                                                                [100%]9 passed in 0.89s

What just happened

The re-entry test's four invocations now make at most 15 searches. Each research task adds 1, the checkpointer saves the total, and the planner reads it on every round of every invocation. spent can use operator.add safely, because integer addition gives the same total in any order and the Step 5 problem does not arise.

A counter in state passes every test you have written. That is exactly why the next test exists.

Step 9: Inject a process crash after a paid call and resume

Goal

Write a test that kills the process right after a search call, resumes the thread, and checks the budget still holds.

Why this step

The counter lives in state, and state reaches disk only when LangGraph saves the node's writes. Suppose the process dies after search.fetch returns but before research returns its "spent": 1. The call has happened, and the counter never heard about it. On resume, LangGraph re-runs the unfinished task, which makes the call a second time.

For this to mean anything, the crash has to be a real process death. An exception is not enough, because it unwinds through LangGraph and LangGraph records it. os._exit ends the process at once with no cleanup, much like a killed container, which is the resume case State Architecture for Agent Networks calls the dangerous part. The test therefore runs the graph in a child process. Its checkpointer must outlive that process, so it uses SqliteSaver on a file; an in-memory checkpointer would die with the child.

durability is an argument to invoke, not to compile, and it controls when checkpoints are written. Under "sync" they are written before the next step starts. Under "async", the default, they are written while that step runs. The test is parametrized over durability="sync" and durability="async" to find out whether either setting protects the budget.

Code

Create tests/crash_child.py, the program the child process runs:

python

# tests/crash_child.py - runs the graph in a process that dies after a search callimport osimport sysfrom pathlib import Pathfrom langgraph.checkpoint.sqlite import SqliteSaverfrom adapter import SPLIT, CountingSearch, make_graph, settleclass CrashingSearch(CountingSearch):    def __init__(self, path, crash_at):        super().__init__(path)        self.crash_at = crash_at    def fetch(self, target):        result = super().fetch(target)          # the paid call has happened        if self.count() == self.crash_at:            os._exit(17)                        # die before the node returns        return resultif __name__ == "__main__":    workdir, durability, crash_at, first = sys.argv[1:]    workdir = Path(workdir)    search = CrashingSearch(workdir / "calls.txt", int(crash_at))    with SqliteSaver.from_conn_string(str(workdir / "checkpoints.sqlite")) as saver:        app = make_graph(workdir, saver, votes=SPLIT, search=search)        inputs = {"requested_width": 1} if first == "yes" else None        settle(app, inputs, "crash", durability=durability)

Create tests/test_crash.py:

python

# tests/test_crash.pyimport osimport subprocessimport sysfrom pathlib import Pathimport pytestfrom langgraph.checkpoint.sqlite import SqliteSaverfrom adapter import BUDGET, SPLIT, calls, make_graph, settleROOT = Path(__file__).resolve().parents[1]CHILD = ROOT / "tests" / "crash_child.py"def crash_once(workdir, durability, first):    crash_at = calls(workdir) + 2              # the 2nd search this process makes    args = [sys.executable, str(CHILD), str(workdir), durability, str(crash_at), first]    env = {**os.environ, "PYTHONPATH": str(ROOT)}    return subprocess.run(args, env=env).returncode@pytest.mark.parametrize("durability", ["sync", "async"])def test_crash_budget(tmp_path, durability):    codes = [crash_once(tmp_path, durability, "yes")]    codes += [crash_once(tmp_path, durability, "no") for _ in range(2)]    assert 17 in codes, "no child crashed, so nothing was tested"    with SqliteSaver.from_conn_string(str(tmp_path / "checkpoints.sqlite")) as saver:        app = make_graph(tmp_path, saver, votes=SPLIT)        settle(app, None, "crash", durability=durability)       # recover the run        for _ in range(4):                                       # then keep working            settle(app, {"requested_width": 1}, "crash", durability=durability)    ran = calls(tmp_path)    assert ran <= BUDGET, f"ran {ran}"

Run it

bash

pytest tests/test_crash.py

Expected output

text

FF                                                                       [100%]=========================== short test summary info ===========================FAILED tests/test_crash.py::test_crash_budget[sync] - AssertionError: ran 18FAILED tests/test_crash.py::test_crash_budget[async] - AssertionError: ran 192 failed in 10.74s

The async count may differ on your machine and between runs. Across five runs on mine, sync printed 18 every time and async printed 17, 19, 19, 20 and 17. Both always exceeded 15.

What just happened

Three child processes each made two searches and died on the second, before research could return. The parent then recovered the run and kept working the thread until the planner saw a spent of 15. Under sync, each crash lost exactly one "spent": 1, so the counter reached 15 while the search API had been called 18 times.

Under async the loss varies, because the checkpoint for the previous step may or may not have reached disk when the process died. LangGraph's documentation says this mode carries "a small risk that LangGraph does not write checkpoints if the process crashes during execution". sync is not fully predictable either: an open issue, langgraph#8039, reports that sync recovery after a hard kill differs between machines. So the test asserts a bound, never an exact count, and it only requires that at least one child crashed.

If the children crash and [sync] still passes on your machine, recovery on your host behaved differently from mine. #8039 shows that recovery after a hard kill can differ between machines, and it is the reason this test never pins an exact number. Watch the [async] case instead, and add a temporary print(ran) above the last assertion if you want to see the count.

The run at the top of this page showed ran 37 for both cases because it used the Step 1 graph, which has no budget at all. There the count is every search the crashing children and the five recovery invocations made. Against the Step 8 counter the overshoot shrinks to a few calls, but not to zero.

A few more details keep the test honest:

The crash happens in research, never in the first node. If you crash before the first checkpoint is saved, there is nothing to resume (langgraph#8764).
crash_at is recomputed for each child, because the call file keeps growing from one child to the next. The child dies on its second search, not its first: the first search is usually the re-run of the task the previous child died in, so each child moves the thread forward before it dies.
The child is a plain python process, not a pytest run, so it does not see pythonpath = . from pytest.ini. crash_once sets PYTHONPATH itself so the child can import agent.
The recovery phase runs one resume and then four new turns. Each turn loops about six rounds under a split vote, so four turns at a width of 1 are enough to spend the whole budget of 15, whatever state the crashes left behind.
The recovery phase starts new turns instead of looping until next == (). After an async crash the saved state can look finished when it is not, and a loop that trusted next would stop early and pass without testing anything. LangGraph Evals Test the Answer, Not the Thread warns about the same field from another angle: .next can be empty for a task that has not completed.

Neither durability setting protects the budget, because the budget lives in the thing that rolls back.

Step 10: Move spend into a ledger the checkpointer cannot roll back

Goal

Make every test pass by charging each search in an external ledger before the call is made.

Why this step

Anything in graph state rolls back with the checkpoint, so the spent total has to live somewhere no checkpointer restores. Order matters too. Charge first and call second, and a crash after the call leaves a charge that the re-run pays again: budget is wasted, but calls can never exceed it. Call first and charge second, and a crash between the two leaves a call nobody paid for. That is the bug you just measured.

The ledger is a SQLite table keyed by thread_id, charged with one conditional UPDATE ... WHERE used < ?. The statement is atomic. It either takes a unit or does nothing, even with several processes writing to the same file. Each method opens and closes its own connection, so a crashed process leaves no lock behind.

Three details in agent/ledger.py look like style and are not:

isolation_level=None puts the connection in autocommit mode, so each statement commits the moment it runs. closing() never calls commit(), so without autocommit the charge would be rolled back when the connection closes.
timeout=30 makes a connection wait up to 30 seconds for another process's write lock instead of failing at once.
INSERT OR IGNORE runs before the UPDATE so the thread has a row to update. Without it, the first charge on a new thread matches no row and is refused.

Do not tidy this into with sqlite3.connect(...) as db:. That form manages a transaction, not the connection, and it does not close the file.

Code

Create agent/ledger.py:

python

# agent/ledger.pyimport sqlite3from contextlib import closingclass SpendLedger:    """Spend lives outside the checkpointer, so no resume can roll it back."""    def __init__(self, path):        self.path = str(path)        with closing(self._connect()) as db:            db.execute("CREATE TABLE IF NOT EXISTS spend "                       "(thread_id TEXT PRIMARY KEY, used INTEGER NOT NULL)")    def _connect(self):        return sqlite3.connect(self.path, timeout=30, isolation_level=None)    def try_charge(self, thread_id, budget):        """Charge one unit if the thread has budget left. Returns True if charged."""        with closing(self._connect()) as db:            db.execute("INSERT OR IGNORE INTO spend VALUES (?, 0)", (thread_id,))            cur = db.execute(                "UPDATE spend SET used = used + 1 WHERE thread_id = ? AND used < ?",                (thread_id, budget),            )            return cur.rowcount == 1    def remaining(self, thread_id, budget):        with closing(self._connect()) as db:            row = db.execute(                "SELECT used FROM spend WHERE thread_id = ?", (thread_id,)            ).fetchone()        return budget - (row[0] if row else 0)

Replace agent/graph.py with:

python

# agent/graph.pyimport operatorfrom typing import Annotated, TypedDictfrom langgraph.graph import END, START, StateGraphfrom langgraph.types import SendMAX_FANOUT = 8      # most research tasks one dispatch may createBUDGET = 15         # most research tasks one thread may ever runROLES = ("facts", "policy")class State(TypedDict):    requested_width: int                          # the planner model's choice, stubbed    targets: list[str]    findings: Annotated[list[str], operator.add]    verdicts: Annotated[dict[str, str], operator.or_]   # one key per reviewer role    approved: booldef build(checkpointer, search, ledger, reviewers=("facts_review", "policy_review"),          votes=("approve", "reject")):    def plan(state, config):        left = ledger.remaining(config["configurable"]["thread_id"], BUDGET)        width = max(0, min(state["requested_width"], left))        return {"targets": [f"source-{i}" for i in range(width)]}    def fan_out(state):        if not state["targets"]:            return "budget_exhausted"        return [Send("research", {"target": t}) for t in state["targets"][:MAX_FANOUT]]    def research(payload, config):        if not ledger.try_charge(config["configurable"]["thread_id"], BUDGET):            return {}                                # no charge, no call        return {"findings": [search.fetch(payload["target"])]}    def reviewer(role, vote):        return lambda state: {"verdicts": {role: vote}}    def gate(state):        verdicts = state.get("verdicts", {})        return {"approved": all(verdicts.get(r) == "approve" for r in ROLES)}    def decide(state):        return END if state["approved"] else "plan"    g = StateGraph(State)    g.add_node("plan", plan)    g.add_node("research", research)    g.add_node("gate", gate)    g.add_node("budget_exhausted", lambda state: {"approved": False})    g.add_edge(START, "plan")    g.add_conditional_edges("plan", fan_out, ["research", "budget_exhausted"])    g.add_edge("budget_exhausted", END)    for name, role, vote in zip(reviewers, ROLES, votes):        g.add_node(name, reviewer(role, vote))        g.add_edge("research", name)    g.add_edge(list(reviewers), "gate")    g.add_conditional_edges("gate", decide, ["plan", END])    return g.compile(checkpointer=checkpointer)

The changes from Step 8: spent is gone from State, build takes a ledger, and plan and research take config so they can read the thread_id. LangGraph passes the run's config to a node function that declares a parameter named config.

build has a new argument, so the adapter must supply it. In tests/adapter.py, add the import below from agent.graph import build:

python

from agent.graph import buildfrom agent.ledger import SpendLedger

And replace make_graph with:

python

def make_graph(workdir, checkpointer, reviewers=REVIEWERS, votes=SPLIT, search=None):    search = search or CountingSearch(Path(workdir) / "calls.txt")    ledger = SpendLedger(Path(workdir) / "ledger.sqlite")    return build(checkpointer, search, ledger, reviewers=reviewers, votes=votes)

No test file changes. Wiring a new graph dependency into the tests is the job the adapter exists to do.

Run it

bash

pytest

Expected output

text

...........                                                              [100%]11 passed in 30.77s

What just happened

All eleven tests pass, including both crash cases. Each research task takes one unit from the ledger before it calls search. A task that re-runs after a crash charges again, so each crash burns a unit of budget with no result to show for it, but calls can never pass 15. The planner reads the same ledger and stops dispatching once the budget is gone.

Before you copy this ledger, know its three limits:

It counts charges, not outcomes. To tell a charge whose call succeeded from one whose result was lost, record intent and outcome separately.
It is keyed on thread_id, so a fork to a new thread gets a fresh budget. If your system forks threads, key the ledger on the business task instead.
A refused charge leaves that source unresearched, and this demo's gate does not check completeness. In production, record the refusal in state and have the gate treat any refusal as a failed review.

Fix common LangGraph and pytest errors in this suite

Landed here from a GraphRecursionError, an InvalidUpdateError at key 'verdicts', or an EmptyInputError? Each has a row in the table. Every error below was produced on the pinned versions while building this tutorial.

What you see	Root cause	Fix
`ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id`	You invoked a graph compiled with a checkpointer without `configurable.thread_id`.	Pass `{"configurable": {"thread_id": "..."}}`. `settle` in the adapter always does.
TypeError: Invalid checkpointer provided. Expected an instance of `BaseCheckpointSaver`, `True`, `False`, or `None`. Received _GeneratorContextManager. Pass a proper saver (e.g., InMemorySaver, AsyncPostgresSaver).	You passed `SqliteSaver.from_conn_string(path)` straight to `compile`. It is a context manager, not a saver.	Use `with SqliteSaver.from_conn_string(path) as saver:` and pass `saver`, as `crash_child.py` does.
langgraph.errors.GraphRecursionError: Recursion limit of 25 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key., followed by a troubleshooting link	A split vote loops until the limit. Expected in this graph.	Call the graph through `settle`, which catches it. Outside tests, treat it as a per-invocation stop, not a budget.
`langgraph.errors.InvalidUpdateError: At key 'verdicts': Can receive only one value per step. Use an Annotated key to handle multiple values.`, followed by a troubleshooting link	In Step 6 you wrote `verdicts: dict[str, str]` without `Annotated[..., operator.or_]`, and two reviewers wrote it in one super-step.	Keep the reducer. `operator.or_` merges the two dictionaries.
`test_crash_budget` fails with `AssertionError: no child crashed, so nothing was tested`, and `pytest -s tests/test_crash.py` shows the child printing `ModuleNotFoundError: No module named 'agent'`	The child process was started without the project root on `PYTHONPATH`, so `adapter` imported but `agent` did not.	Keep `env = {**os.environ, "PYTHONPATH": str(ROOT)}` in `crash_once`.
`langgraph.errors.EmptyInputError: Received no input for __start__`	You resumed with `invoke(None, config)` on a thread with no saved checkpoint: a fresh `thread_id`, a crash before the first checkpoint, or a run under `durability="exit"`, which writes nothing until the graph exits.	Crash later than the first node, use `"sync"` or `"async"`, and check the `thread_id` matches.
`ModuleNotFoundError: No module named 'langgraph.checkpoint.sqlite'`	`SqliteSaver` ships in its own package.	`pip install langgraph-checkpoint-sqlite==3.1.1`.

One failure raises nothing at all: the crash test passing against a graph you have not fixed. If test_crash_budget goes green on a graph that keeps spend in state, first check that the children really crashed (17 in codes). Then check that the recovery phase does real work, since a recovery loop that stops on next == () can end before the thread spends anything.

How the test suite, adapter and spend ledger fit together

The tests reach the graph only through the adapter. Three stores sit outside the graph: the call file the tests count, the checkpointer that can roll back, and the ledger that cannot.

direction: right

tests: "tests/" {
  style: {fill: "#98D8C8"; stroke: "#5FA897"; font-color: "#2C2C2A"}
  props: "test_width\ntest_merge\ntest_reentry" {
    style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}
  }
  crash: "test_crash" {
    style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}
  }
  child: "crash_child.py\n(os._exit after a call)" {
    style: {fill: "#E74C3C"; stroke: "#A93226"; font-color: "#FFFFFF"}
  }
  adapter: "adapter.py\nMAX_FANOUT, BUDGET\nmake_graph, settle, calls" {
    style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}
  }
  props -> adapter
  crash -> child: "subprocess"
  crash -> adapter
  child -> adapter
}

agent: "agent/" {
  style: {fill: "#FFA07A"; stroke: "#C9785A"; font-color: "#2C2C2A"}
  graph: "graph.py\ncap at Send, role keys" {
    style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}
  }
  ledger: "ledger.py\nSpendLedger" {
    style: {fill: "#6BCF7F"; stroke: "#3E9E52"; font-color: "#2C2C2A"}
  }
  graph -> ledger: "charge before call"
}

calls: "calls.txt\n(counted by tests)" {
  style: {fill: "#FFD93D"; stroke: "#C9A800"; font-color: "#2C2C2A"}
}
ckpt: "checkpointer\n(rolls back)" {
  style: {fill: "#95A5A6"; stroke: "#6C7A7B"; font-color: "#2C2C2A"}
}
spend: "ledger.sqlite\n(never rolls back)" {
  style: {fill: "#6BCF7F"; stroke: "#3E9E52"; font-color: "#2C2C2A"}
}

tests.adapter -> agent.graph: "build"
agent.graph -> calls: "search.fetch"
agent.graph -> ckpt: "state"
agent.ledger -> spend

The arrow that matters runs from ledger.py to ledger.sqlite. It is the only path from spend to storage that avoids the checkpointer, and Step 9 showed that anything stored through the checkpointer can come back smaller than it went in.

Complete code for the LangGraph contract test suite

text

contract-tests/├── pytest.ini├── agent/│   ├── __init__.py│   ├── graph.py│   └── ledger.py└── tests/    ├── adapter.py    ├── crash_child.py    ├── test_adapter.py    ├── test_crash.py    ├── test_merge.py    ├── test_reentry.py    └── test_width.py

Every file, in its final form. agent/__init__.py is empty.

pytest.ini:

ini

[pytest]pythonpath = .testpaths = testsaddopts = -q --tb=no -rf

agent/graph.py:

python

# agent/graph.pyimport operatorfrom typing import Annotated, TypedDictfrom langgraph.graph import END, START, StateGraphfrom langgraph.types import SendMAX_FANOUT = 8      # most research tasks one dispatch may createBUDGET = 15         # most research tasks one thread may ever runROLES = ("facts", "policy")class State(TypedDict):    requested_width: int                          # the planner model's choice, stubbed    targets: list[str]    findings: Annotated[list[str], operator.add]    verdicts: Annotated[dict[str, str], operator.or_]   # one key per reviewer role    approved: booldef build(checkpointer, search, ledger, reviewers=("facts_review", "policy_review"),          votes=("approve", "reject")):    def plan(state, config):        left = ledger.remaining(config["configurable"]["thread_id"], BUDGET)        width = max(0, min(state["requested_width"], left))        return {"targets": [f"source-{i}" for i in range(width)]}    def fan_out(state):        if not state["targets"]:            return "budget_exhausted"        return [Send("research", {"target": t}) for t in state["targets"][:MAX_FANOUT]]    def research(payload, config):        if not ledger.try_charge(config["configurable"]["thread_id"], BUDGET):            return {}                                # no charge, no call        return {"findings": [search.fetch(payload["target"])]}    def reviewer(role, vote):        return lambda state: {"verdicts": {role: vote}}    def gate(state):        verdicts = state.get("verdicts", {})        return {"approved": all(verdicts.get(r) == "approve" for r in ROLES)}    def decide(state):        return END if state["approved"] else "plan"    g = StateGraph(State)    g.add_node("plan", plan)    g.add_node("research", research)    g.add_node("gate", gate)    g.add_node("budget_exhausted", lambda state: {"approved": False})    g.add_edge(START, "plan")    g.add_conditional_edges("plan", fan_out, ["research", "budget_exhausted"])    g.add_edge("budget_exhausted", END)    for name, role, vote in zip(reviewers, ROLES, votes):        g.add_node(name, reviewer(role, vote))        g.add_edge("research", name)    g.add_edge(list(reviewers), "gate")    g.add_conditional_edges("gate", decide, ["plan", END])    return g.compile(checkpointer=checkpointer)

agent/ledger.py:

python

# agent/ledger.pyimport sqlite3from contextlib import closingclass SpendLedger:    """Spend lives outside the checkpointer, so no resume can roll it back."""    def __init__(self, path):        self.path = str(path)        with closing(self._connect()) as db:            db.execute("CREATE TABLE IF NOT EXISTS spend "                       "(thread_id TEXT PRIMARY KEY, used INTEGER NOT NULL)")    def _connect(self):        return sqlite3.connect(self.path, timeout=30, isolation_level=None)    def try_charge(self, thread_id, budget):        """Charge one unit if the thread has budget left. Returns True if charged."""        with closing(self._connect()) as db:            db.execute("INSERT OR IGNORE INTO spend VALUES (?, 0)", (thread_id,))            cur = db.execute(                "UPDATE spend SET used = used + 1 WHERE thread_id = ? AND used < ?",                (thread_id, budget),            )            return cur.rowcount == 1    def remaining(self, thread_id, budget):        with closing(self._connect()) as db:            row = db.execute(                "SELECT used FROM spend WHERE thread_id = ?", (thread_id,)            ).fetchone()        return budget - (row[0] if row else 0)

tests/adapter.py:

python

# tests/adapter.py - the only file in tests/ that knows which graph it is testingimport threadingfrom pathlib import Pathfrom langgraph.errors import GraphRecursionErrorfrom agent.graph import buildfrom agent.ledger import SpendLedger# The contract: numbers your team promised, not settings the graph happens to have.MAX_FANOUT = 8      # most research tasks one dispatch may createBUDGET = 15         # most research tasks one thread may ever runREVIEWERS = ("facts_review", "policy_review")SPLIT = ("approve", "reject")UNANIMOUS = ("approve", "approve")class CountingSearch:    """Stands in for a paid search API. Each call appends one line to a file."""    def __init__(self, path):        self.path = Path(path)        self.lock = threading.Lock()    def fetch(self, target):        with self.lock, self.path.open("a") as f:            f.write(target + "\n")        return f"notes on {target}"    def count(self):        return len(self.path.read_text().splitlines()) if self.path.exists() else 0def calls(workdir):    return CountingSearch(Path(workdir) / "calls.txt").count()def make_graph(workdir, checkpointer, reviewers=REVIEWERS, votes=SPLIT, search=None):    search = search or CountingSearch(Path(workdir) / "calls.txt")    ledger = SpendLedger(Path(workdir) / "ledger.sqlite")    return build(checkpointer, search, ledger, reviewers=reviewers, votes=votes)def settle(app, inputs, thread_id, recursion_limit=25, durability=None):    """Run one invocation and return the saved state, however the run stopped."""    config = {"recursion_limit": recursion_limit,              "configurable": {"thread_id": thread_id}}    try:        app.invoke(inputs, config, durability=durability)    except GraphRecursionError:        pass    return app.get_state(config)def approved(snapshot):    return snapshot.next == () and snapshot.values.get("approved") is True

tests/test_adapter.py:

python

# tests/test_adapter.pyfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import UNANIMOUS, approved, calls, make_graph, settledef test_adapter_runs_one_approved_round(tmp_path):    app = make_graph(tmp_path, InMemorySaver(), votes=UNANIMOUS)    snapshot = settle(app, {"requested_width": 3}, "smoke")    assert approved(snapshot)    assert calls(tmp_path) == 3

tests/test_width.py:

python

# tests/test_width.pyfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import MAX_FANOUT, UNANIMOUS, calls, make_graph, settledef test_dispatch_capped(tmp_path):    app = make_graph(tmp_path, InMemorySaver(), votes=UNANIMOUS)    settle(app, {"requested_width": 1000}, "width")    ran = calls(tmp_path)    assert ran <= MAX_FANOUT, f"ran {ran}"

tests/test_merge.py:

python

# tests/test_merge.pyimport pytestfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import SPLIT, UNANIMOUS, approved, make_graph, settleNAMES = [    ("facts_review", "policy_review"),    ("facts_review", "compliance_review"),    ("zz_facts", "aa_policy"),]IDS = ["policy", "compliance", "reversed"]@pytest.mark.parametrize("reviewers", NAMES, ids=IDS)def test_split(tmp_path, reviewers):    app = make_graph(tmp_path, InMemorySaver(), reviewers=reviewers, votes=SPLIT)    result = approved(settle(app, {"requested_width": 1}, "split"))    assert not result, "approved"@pytest.mark.parametrize("reviewers", NAMES, ids=IDS)def test_unanimous(tmp_path, reviewers):    app = make_graph(tmp_path, InMemorySaver(), reviewers=reviewers, votes=UNANIMOUS)    result = approved(settle(app, {"requested_width": 1}, "unanimous"))    assert result, "not approved"

tests/test_reentry.py:

python

# tests/test_reentry.pyfrom langgraph.checkpoint.memory import InMemorySaverfrom adapter import BUDGET, SPLIT, calls, make_graph, settledef test_thread_budget(tmp_path):    app = make_graph(tmp_path, InMemorySaver(), votes=SPLIT)    settle(app, {"requested_width": 3}, "t1")   # first run stops at the limit    settle(app, None, "t1")                     # resume after the limit    settle(app, None, "t1")                     # resume again    settle(app, {"requested_width": 3}, "t1")   # a new turn on the same thread    ran = calls(tmp_path)    assert ran <= BUDGET, f"ran {ran}"

tests/crash_child.py:

python

# tests/crash_child.py - runs the graph in a process that dies after a search callimport osimport sysfrom pathlib import Pathfrom langgraph.checkpoint.sqlite import SqliteSaverfrom adapter import SPLIT, CountingSearch, make_graph, settleclass CrashingSearch(CountingSearch):    def __init__(self, path, crash_at):        super().__init__(path)        self.crash_at = crash_at    def fetch(self, target):        result = super().fetch(target)          # the paid call has happened        if self.count() == self.crash_at:            os._exit(17)                        # die before the node returns        return resultif __name__ == "__main__":    workdir, durability, crash_at, first = sys.argv[1:]    workdir = Path(workdir)    search = CrashingSearch(workdir / "calls.txt", int(crash_at))    with SqliteSaver.from_conn_string(str(workdir / "checkpoints.sqlite")) as saver:        app = make_graph(workdir, saver, votes=SPLIT, search=search)        inputs = {"requested_width": 1} if first == "yes" else None        settle(app, inputs, "crash", durability=durability)

tests/test_crash.py:

python

# tests/test_crash.pyimport osimport subprocessimport sysfrom pathlib import Pathimport pytestfrom langgraph.checkpoint.sqlite import SqliteSaverfrom adapter import BUDGET, SPLIT, calls, make_graph, settleROOT = Path(__file__).resolve().parents[1]CHILD = ROOT / "tests" / "crash_child.py"def crash_once(workdir, durability, first):    crash_at = calls(workdir) + 2              # the 2nd search this process makes    args = [sys.executable, str(CHILD), str(workdir), durability, str(crash_at), first]    env = {**os.environ, "PYTHONPATH": str(ROOT)}    return subprocess.run(args, env=env).returncode@pytest.mark.parametrize("durability", ["sync", "async"])def test_crash_budget(tmp_path, durability):    codes = [crash_once(tmp_path, durability, "yes")]    codes += [crash_once(tmp_path, durability, "no") for _ in range(2)]    assert 17 in codes, "no child crashed, so nothing was tested"    with SqliteSaver.from_conn_string(str(tmp_path / "checkpoints.sqlite")) as saver:        app = make_graph(tmp_path, saver, votes=SPLIT)        settle(app, None, "crash", durability=durability)       # recover the run        for _ in range(4):                                       # then keep working            settle(app, {"requested_width": 1}, "crash", durability=durability)    ran = calls(tmp_path)    assert ran <= BUDGET, f"ran {ran}"

To point the suite at your own graph, rewrite tests/adapter.py. make_graph must build your graph with a counting stand-in for your paid call, and it must accept reviewer-style names for the nodes that write shared state. settle and approved translate "run it" and "did it approve" into your graph's terms. The test files stay as they are.

Extend the suite: Postgres ledger, Hypothesis, a second runtime

Move the ledger to Postgres. The same conditional UPDATE ... WHERE used < $2 is atomic across processes and machines. Run test_crash.py against it with children on separate connections, then add a test where two threads run in parallel against one shared budget.
Fuzz node names with Hypothesis, a property-based testing library for Python. Replace NAMES in test_merge.py with a Hypothesis strategy that generates valid node names (LangGraph rejects | and : in them). The merge test will then find orderings you did not think to write down.
Port the four tests to a second runtime. Rewrite only adapter.py for another agent framework and run the same test files. The graph engineering article found that LangGraph and the OpenAI Agents SDK disagree about what a resume does to a turn budget, and that is the kind of difference this suite exists to catch.

Run the suite on every LangGraph upgrade. Each of these properties depends on runtime behaviour, and no setting's name promises to keep that behaviour the same.

References

LangChain. Checkpointers (super-steps, pending writes, durability modes). LangGraph documentation. https://docs.langchain.com/oss/python/langgraph/checkpointers
LangChain. Use the graph API (map-reduce with Send, parallel update ordering). LangGraph documentation. https://docs.langchain.com/oss/python/langgraph/use-graph-api
LangChain. Graph API overview (recursion limit, reducers). LangGraph documentation. https://docs.langchain.com/oss/python/langgraph/graph-api
LangChain. Test. LangGraph documentation. https://docs.langchain.com/oss/python/langgraph/test
LangChain. GRAPH_RECURSION_LIMIT. https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT
LangChain. INVALID_CONCURRENT_GRAPH_UPDATE. https://docs.langchain.com/oss/python/langgraph/errors/INVALID_CONCURRENT_GRAPH_UPDATE
LangChain. Pregel.invoke (the durability parameter). LangChain reference. https://reference.langchain.com/python/langgraph/pregel/main/Pregel/invoke
LangChain. Send. LangChain reference. https://reference.langchain.com/python/langgraph/types/Send
LangChain. SqliteSaver, langgraph-checkpoint-sqlite 3.1.1 source. GitHub. https://github.com/langchain-ai/langgraph/tree/checkpointsqlite%3D%3D3.1.1/libs/checkpoint-sqlite
LangChain. LangGraph source, tag 1.2.11. GitHub. https://github.com/langchain-ai/langgraph/tree/1.2.11/libs/langgraph
PyPI. langgraph 1.2.11 (2026, August 11). https://pypi.org/project/langgraph/1.2.11/
PyPI. langgraph-checkpoint 4.2.0 (2026, August 7). https://pypi.org/project/langgraph-checkpoint/4.2.0/
PyPI. langgraph-checkpoint-sqlite 3.1.1 (2026, July 30). https://pypi.org/project/langgraph-checkpoint-sqlite/3.1.1/
PyPI. pytest 9.1.1 (2026, June 19). https://pypi.org/project/pytest/9.1.1/
pytest. How to parametrize fixtures and test functions. https://docs.pytest.org/en/stable/how-to/parametrize.html
pytest. How to use temporary directories and files in tests. https://docs.pytest.org/en/stable/how-to/tmp_path.html
pytest. Changelog (9.1.0 parametrize deprecation). https://docs.pytest.org/en/stable/changelog.html
langchain-ai/langgraph issue #8039. (2026, June 10). durability="sync": put_writes/put persistence order is unenforced, so post-crash recovery (replay vs re-execute) is host-dependent. https://github.com/langchain-ai/langgraph/issues/8039
langchain-ai/langgraph issue #8764. (2026, August 30). Crash before first durable checkpoint can silently drop an accepted run with no durable failure record. https://github.com/langchain-ai/langgraph/issues/8764
Stevens, M. crashpoint (crash-injection harness for durable agent runtimes). GitHub. https://github.com/mstevens843/crashpoint
Fowler, M. (2011). Contract Test. https://martinfowler.com/bliki/ContractTest.html
Python. os._exit. Python 3.12 documentation. https://docs.python.org/3.12/library/os.html#os._exit
Python. subprocess: Subprocess management. Python 3.12 documentation. https://docs.python.org/3.12/library/subprocess.html
Python. sqlite3: DB-API 2.0 interface for SQLite databases. Python 3.12 documentation. https://docs.python.org/3.12/library/sqlite3.html
SQLite. The UPDATE statement. https://www.sqlite.org/lang_update.html

Agentic AI

Follow for more technical deep dives on AI/ML systems, production engineering, and building real-world applications:

Prerequisites

How a pytest property test differs from a setting check

Step 1: Build the LangGraph research graph under test

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 2: Write a pytest adapter so the tests never import the graph

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 3: Test fan-out width by requesting 1,000 tasks in one dispatch

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 4: Cap the fan-out where Send creates the tasks

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 5: Test that a reducer merge does not depend on node names

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 6: Give each reviewer its own key and make the gate fail closed

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 7: Test that a spend budget survives resume and a new turn

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 8: Add a spend counter in graph state

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 9: Inject a process crash after a paid call and resume

Goal

Why this step

Code

Run it

Expected output

What just happened

Step 10: Move spend into a ledger the checkpointer cannot roll back

Goal

Why this step

Code

Run it

Expected output

What just happened

Fix common LangGraph and pytest errors in this suite

How the test suite, adapter and spend ledger fit together

Complete code for the LangGraph contract test suite

Extend the suite: Postgres ledger, Hypothesis, a second runtime

References

Related Articles

Get the next article by email

Books by Ranjan Kumar