LangGraph Testing and Debugging: A Practical Guide
Tom PiaggioCo-Founder at Autonoma
LangGraph testing covers a compiled agent graph at three levels: node-level unit tests that invoke a single node in isolation through the checkpointer, trajectory tests that assert the graph took the expected path through nodes and tools, and multi-turn simulations that drive several turns over one thread and check the accumulated state. Paired with LangGraph's built-in state inspection and time-travel replay, it is also the fastest way to debug why an agent went down the wrong branch.
Search "LangGraph testing" today and the official docs give you a few paragraphs on node tests and a mention of the checkpointer, then stop. The rest of what you need is scattered across four or five posts written at different times against different LangGraph versions: one on unit-testing nodes with interrupt_before and update_state, one on trajectory evaluation, one on multi-turn conversation testing, one on time-travel debugging with get_state_history. Each is fine on its own. None treats these as one connected workflow, which means every team shipping a LangGraph agent ends up rebuilding the connective tissue between them from scratch.
This is that connective tissue, built once. Everything below runs against a current LangGraph install, verified against the current docs rather than copied from whichever post happened to rank first. We'll build a small support-ticket triage agent (classify the ticket, pull account context, draft a reply, then escalate or resolve) and test it the way a LangGraph agent actually needs to be tested: one node at a time, the path the graph takes through those nodes, and the outcome accumulated across a multi-turn thread. Then we'll reuse the same checkpointer to debug it when a run goes sideways.
Each layer verifies something the one below it cannot: a single node's logic, the route the graph took to get there, and the outcome accumulated across a whole conversation.
Unit Testing a Single LangGraph Node in Isolation
The graph under test is a StateGraph with four nodes: classify_ticket, fetch_account_context, draft_response, and a conditional split into escalate or resolve. Each node is a plain function that reads the graph's state and returns the fields it changes, which is the point: nothing about testing that logic requires a real model call. Here's the graph, compiled once with a checkpointer so every invocation is tied to a thread_id:
"""Support-ticket triage graph built with LangGraph.The graph classifies an incoming support ticket, fetches deterministic"account context", drafts a response, and then either escalates or resolvesthe ticket. Every node is a *deterministic stand-in* for what would, in a realdeployment, be a model call. Because the nodes never touch a network or an LLM,the whole graph runs with zero API keys and produces byte-stable trajectories,which is exactly what makes the accompanying pytest suite able to assert onexact node sequences.Run it directly to see a full triage in action: python src/graph.py"""from typing import TypedDictfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import END, START, StateGraphclass TicketState(TypedDict, total=False): """State channels threaded through the triage graph. ``total=False`` lets tests seed a partial state (e.g. pretend only ``classify_ticket`` has run) without having to populate every key. """ ticket_text: str account_id: str category: str priority: str account_context: dict draft: str status: str escalation_count: int# A tiny deterministic "database" of accounts. In production this would be a# real lookup against a CRM or billing system.MOCK_ACCOUNTS = { "ACC-1001": {"plan": "enterprise", "mrr": 4200, "tenure_months": 26}, "ACC-1002": {"plan": "starter", "mrr": 49, "tenure_months": 3},}DEFAULT_ACCOUNT = {"plan": "unknown", "mrr": 0, "tenure_months": 0}# Keyword heuristics standing in for a real classifier.BILLING_KEYWORDS = ("refund", "charge", "invoice", "billing", "overcharged", "double charged")URGENT_KEYWORDS = ("urgent", "asap", "immediately", "escalate", "angry", "cancel")def classify_ticket(state: TicketState) -> dict: """Assign a category and priority from the ticket text. A real deployment would replace this body with an LLM call pinned to ``temperature=0`` (e.g. a structured-output classification prompt). The keyword heuristic here is a deterministic stand-in so the trajectory is stable and no API key is required. """ text = state.get("ticket_text", "").lower() if any(word in text for word in BILLING_KEYWORDS): category = "billing_dispute" else: category = "general" if category == "billing_dispute" or any(word in text for word in URGENT_KEYWORDS): priority = "high" else: priority = "normal" return {"category": category, "priority": priority}def fetch_account_context(state: TicketState) -> dict: """Deterministically look up mock account data for the ticket's account.""" account_id = state.get("account_id", "") context = dict(MOCK_ACCOUNTS.get(account_id, DEFAULT_ACCOUNT)) context["account_id"] = account_id return {"account_context": context}def draft_response(state: TicketState) -> dict: """Build a draft reply that references the category and account context.""" category = state.get("category", "general") context = state.get("account_context", {}) or {} plan = context.get("plan", "unknown") if category == "billing_dispute": body = ( f"Thanks for reaching out about your billing concern. As a {plan} " "customer, your case is being reviewed by our billing specialists." ) else: body = ( f"Thanks for contacting support. As a {plan} customer, here is some " "guidance to help resolve your question." ) return {"draft": body}def route_after_draft(state: TicketState) -> str: """Decide whether the drafted ticket should escalate or resolve.""" if state.get("category") == "billing_dispute" or state.get("priority") == "high": return "escalate" return "resolve"def escalate(state: TicketState) -> dict: """Terminal node: mark the ticket escalated. Idempotent by design: a ticket thread can be invoked multiple times (a multi-turn conversation reuses one ``thread_id``), but the escalate path must never be counted more than once for the same thread. If the ticket is already escalated we leave ``escalation_count`` untouched. """ if state.get("status") == "escalated": return {"status": "escalated"} return { "status": "escalated", "escalation_count": state.get("escalation_count", 0) + 1, }def resolve(state: TicketState) -> dict: """Terminal node: mark the ticket resolved.""" return {"status": "resolved"}def build_graph() -> StateGraph: """Assemble and return the *uncompiled* graph builder. Tests import this so they can compile a fresh graph with their own checkpointer (and their own interrupt configuration) per test, keeping every test fully isolated. """ builder = StateGraph(TicketState) builder.add_node("classify_ticket", classify_ticket) builder.add_node("fetch_account_context", fetch_account_context) builder.add_node("draft_response", draft_response) builder.add_node("escalate", escalate) builder.add_node("resolve", resolve) builder.add_edge(START, "classify_ticket") builder.add_edge("classify_ticket", "fetch_account_context") builder.add_edge("fetch_account_context", "draft_response") builder.add_conditional_edges( "draft_response", route_after_draft, {"escalate": "escalate", "resolve": "resolve"}, ) builder.add_edge("escalate", END) builder.add_edge("resolve", END) return builder# Uncompiled builder, exported for tests.builder = build_graph()# Default compiled graph with an in-memory checkpointer, ready to run.graph = builder.compile(checkpointer=InMemorySaver())if __name__ == "__main__": import uuid config = {"configurable": {"thread_id": str(uuid.uuid4())}} sample_ticket = { "ticket_text": "I was double charged on my last invoice, please refund ASAP.", "account_id": "ACC-1001", } final_state = graph.invoke(sample_ticket, config) print("=== Support-ticket triage: final state ===") for key in ( "ticket_text", "account_id", "category", "priority", "account_context", "draft", "status", "escalation_count", ): print(f"{key:>17}: {final_state.get(key)}")
To unit test fetch_account_context on its own, you do not need to run classify_ticket first. Compile the graph once with interrupt_after set to ["fetch_account_context"], a static setting passed to .compile(), not something you flip per invocation. Then call update_state with as_node set to classify_ticket, seeding the state exactly as if that node had already run and fetch_account_context were next in line. Resuming with graph.invoke(None, config) runs exactly that one node and then stops, because the compiled interrupt fires right after it, so nothing downstream executes by accident.
The test itself has three moves: compile with interrupt_after set to the node under test, seed state via update_state with as_node pointing at the node upstream of it, then resume with graph.invoke(None, config) and call graph.get_state(config) to assert on the fields that node is supposed to add, account_context in this case. Nothing about that assertion cares what classify_ticket or draft_response do; it is a true unit test of one function, with the checkpointer standing in for the setup code you would otherwise write by hand.
"""Layer 1 - node-level unit testing via checkpointer isolation.The idea: seed the graph's state to look as though the upstream nodes have*already* run (using ``update_state(..., as_node=...)``), then execute exactlyone more node and assert on its output. This is a true unit test - neither testbelow depends on the upstream node's real logic being correct, only on the onenode under test.To run a single node we compile the graph with an ``interrupt_after`` on thenode we want, so execution pauses the moment that node finishes and neverreaches the following node. In current LangGraph, ``interrupt_before`` /``interrupt_after`` are static and passed to ``compile()`` (not to ``invoke``),which is the form used here."""from uuid import uuid4from langgraph.checkpoint.memory import InMemorySaverfrom src.graph import builderdef _fresh_config(): """A config with a unique thread_id so every test is fully isolated.""" return {"configurable": {"thread_id": str(uuid4())}}def test_fetch_account_context_in_isolation(): # Compile a private graph that pauses right after fetch_account_context. graph = builder.compile( checkpointer=InMemorySaver(), interrupt_after=["fetch_account_context"], ) config = _fresh_config() # Seed state as if classify_ticket already produced a classification. We do # NOT run classify_ticket's real logic - we assert the values directly. graph.update_state( config, values={ "ticket_text": "seeded ticket", "account_id": "ACC-1001", "category": "billing_dispute", "priority": "high", }, as_node="classify_ticket", ) # Resume: only fetch_account_context runs, then execution interrupts. graph.invoke(None, config) values = graph.get_state(config).values # fetch_account_context ran and populated account_context... assert "account_context" in values assert values["account_context"]["account_id"] == "ACC-1001" assert values["account_context"]["plan"] == "enterprise" # ...and draft_response never ran, so there is no draft yet. assert "draft" not in valuesdef test_draft_response_in_isolation(): # Same pattern, one node further downstream: pause after draft_response. graph = builder.compile( checkpointer=InMemorySaver(), interrupt_after=["draft_response"], ) config = _fresh_config() # Seed state as if fetch_account_context already ran, with a distinctive # account_context we can look for in the generated draft. graph.update_state( config, values={ "ticket_text": "seeded ticket", "category": "billing_dispute", "priority": "high", "account_context": {"account_id": "ACC-1001", "plan": "enterprise"}, }, as_node="fetch_account_context", ) # Resume: only draft_response runs, then execution interrupts (the terminal # escalate/resolve node never runs). graph.invoke(None, config) values = graph.get_state(config).values assert "draft" in values # The draft references the seeded account_context, proving draft_response # consumed exactly the state we injected. assert "enterprise" in values["draft"] # The terminal node has not run, so status is still unset. assert "status" not in values
Verifying the Trajectory: Did the Graph Take the Right Path?
A node-level test tells you fetch_account_context handles one input correctly. It says nothing about whether the graph actually calls that node, in that order, for a given ticket. That's a routing problem, and add_conditional_edges is exactly where routing bugs hide: a comparison flipped, a threshold off by one category, a boolean the model hands back as a string instead of an actual bool. Node tests never catch it, because the node itself was correct. What you need is a trajectory assertion: run the full graph once, then check the sequence of nodes it actually visited against the sequence you expected.
LangGraph makes the executed sequence cheap to recover. Call graph.stream with stream_mode="updates" and read the node name off each yielded update, or call graph.get_state_history(config) after the run and walk the checkpoints in order (they come back most-recent-first, so reverse the list). Either way you get a list of node names, and the assertion is a plain list comparison: for a high-priority billing dispute, the expected path is classify_ticket, fetch_account_context, draft_response, escalate; for a routine ticket it's the same first three nodes followed by resolve instead.
The trajectory test reads this exact sequence back out of get_state_history and asserts it against the expected path for a high-priority billing ticket.
This is the same idea our companion piece on testing AI agent tool calls covers in general terms: the sequence of steps an agent chooses matters as much as its final answer. Applied to LangGraph specifically, the sequence is nodes and edges instead of an abstract tool-call log, and get_state_history hands it to you for free, because the checkpointer was already recording it for durability, not for testing.
"""Layer 2 - trajectory / route assertions via get_state_history.Instead of only checking the final answer, we assert on the *path* the graphtook: which nodes ran, in what order. We reconstruct that path from thecheckpoint history.Note on the current LangGraph API: ``get_state_history`` yields one``StateSnapshot`` per super-step in reverse-chronological order. In thisversion the per-node ``writes`` map is no longer carried in snapshot metadata,so we reconstruct the executed order from each snapshot's ``.next`` field - thenode that was about to run at that checkpoint. Reading those in chronologicalorder (skipping the initial ``__start__`` input snapshot and the final snapshotwhose ``.next`` is empty) yields the exact sequence of executed nodes.Because the nodes in src/graph.py are deterministic stand-ins (no real modelcall), this trajectory is stable across repeated runs, which is what makes anexact list-equality assertion appropriate here."""from uuid import uuid4from langgraph.checkpoint.memory import InMemorySaverfrom src.graph import builderdef _fresh_config(): return {"configurable": {"thread_id": str(uuid4())}}def _executed_trajectory(graph, config): """Reconstruct the chronological list of executed nodes from history.""" history = list(graph.get_state_history(config)) # reverse-chronological trajectory = [] for snapshot in reversed(history): # walk chronologically # Each snapshot's `.next` is the node about to run at that checkpoint. # Skip the input snapshot (next == ('__start__',)) and the final # snapshot (next == ()). if len(snapshot.next) == 1 and snapshot.next[0] != "__start__": trajectory.append(snapshot.next[0]) return trajectorydef test_billing_dispute_trajectory_escalates(): graph = builder.compile(checkpointer=InMemorySaver()) config = _fresh_config() graph.invoke( { "ticket_text": "I was double charged, please refund my invoice ASAP.", "account_id": "ACC-1001", }, config, ) assert _executed_trajectory(graph, config) == [ "classify_ticket", "fetch_account_context", "draft_response", "escalate", ]def test_routine_ticket_trajectory_resolves(): graph = builder.compile(checkpointer=InMemorySaver()) config = _fresh_config() graph.invoke( { "ticket_text": "How do I change the email address on my account?", "account_id": "ACC-1002", }, config, ) assert _executed_trajectory(graph, config) == [ "classify_ticket", "fetch_account_context", "draft_response", "resolve", ]
How Autonoma Covers What LangGraph's Own Tests Can't See
Node and trajectory tests both run entirely inside the graph. A node test confirms fetch_account_context transforms state correctly. A trajectory test confirms escalate ran instead of resolve, in the right order. Neither confirms that escalating the ticket did anything to the product wrapped around the graph: that the support queue a human agent looks at actually shows the ticket as high-priority, that the customer's status widget updated, that the on-call alert actually fired. The graph can be exactly right and the feature can still be broken, because the code wiring the graph's decision to the rest of the application is something no graph-level test ever touches.
That's the layer we built Autonoma to cover. It's an open-source, codebase-first testing product that runs behavioral end-to-end tests against your running application, not against the graph in isolation. Our Planner agent reads the actual routes, components, and schema your ticket-triage feature touches (the queue page, the status widget, the alert handler) and plans test cases against what the product is supposed to do once that graph decision fires. Our Execution Agent drives the real UI in a live preview environment the way a support engineer would, checking that the escalated ticket actually landed in the queue instead of reading a state dict that claims it did. Our GenerationReviewer separates a genuine regression from a flaky run, and our Diffs Agent keeps that suite aligned every time someone reworks the escalation endpoint, so the coverage doesn't rot the week after you write it.
Think of it as a layer sitting above everything a graph-level test can see. Node tests, trajectory tests, and the multi-turn simulation below all verify the graph produced the right decision. Autonoma verifies the rest of the product actually acted on it, which is the part a passing pytest run has no way to check.
Multi-Turn Simulation Over a Compiled Graph
Node tests and trajectory tests both run the graph once, start to finish. Production traffic rarely looks like that. A real support ticket is a thread: the customer replies, the agent re-classifies, the account context gets re-fetched with new information, and the final status only makes sense in light of everything that came before it. Testing that means driving several turns through the samethread_id, not a fresh one each time, so the checkpointer accumulates state across calls the way it would in production.
The pattern is a loop: build one config with a fixed thread_id, call graph.invoke once per turn against that same config, and after the last turn call graph.get_state(config) to read the accumulated state. What you assert matters more than how. Exact wording is the wrong target, two runs of the same conversation can phrase a draft differently and both be correct, but an ID or a status is not: pin temperature=0 on the underlying model call where you can to cut incidental variance, then assert on structure and invariants rather than text. The final status is one of escalated or resolved. Escalation happens at most once per thread. Every account fact the agent repeats back matches the fact it was given, exactly, every turn.
If a scenario genuinely needs to check that a reply means the right thing rather than contains the right token, that's the one case worth reaching for an LLM-as-judge, and only there. Save it for semantic equivalence checks specifically; using it as the default assertion for everything trades a fast, deterministic test for a slow, non-deterministic one grading another non-deterministic one.
"""Layer 3 - multi-turn simulation over a single thread_id.A support ticket is a conversation, not a one-shot call. Here we reuse ONEconfig (one ``thread_id``) across three sequential ``invoke`` calls so thecheckpointer threads state from turn to turn, exactly as a real multi-turnsession would.We assert on *invariants* rather than exact wording. Free text (the ``draft``)is intentionally not asserted verbatim: if a scenario ever needed to check thesemantic meaning of generated prose, an LLM-as-judge would be the right tool -reserved for that case only. And if these nodes called a real LLM instead ofthe deterministic stand-ins used here, pinning ``temperature=0`` would be theequivalent move to keep the structured fields stable enough to assert on."""from langgraph.checkpoint.memory import InMemorySaverfrom src.graph import builderdef test_multi_turn_ticket_thread_invariants(): graph = builder.compile(checkpointer=InMemorySaver()) # ONE config reused across every turn: the whole point of the test. config = {"configurable": {"thread_id": "support-ticket-42"}} # Turn 1: an ordinary question with no account attached. graph.invoke( {"ticket_text": "I have a general question about my account settings."}, config, ) # Turn 2: the customer introduces an account_id and escalating urgency. account_id = "ACC-1001" graph.invoke( { "ticket_text": "Actually I was overcharged and need a refund urgently!", "account_id": account_id, }, config, ) # Turn 3: a follow-up nudge on the same thread. graph.invoke( {"ticket_text": "Any update on this? I'll cancel otherwise."}, config, ) final = graph.get_state(config).values # Invariant 1: the ticket reaches a valid terminal status. assert final["status"] in {"escalated", "resolved"} # Invariant 2: the escalate path is counted at most once per thread, even # though routing chose "escalate" on more than one turn. The idempotent # guard in the escalate node enforces this bound. assert final.get("escalation_count", 0) <= 1 # Invariant 3: the account_id introduced in turn 2 survives unchanged into # the final state - state persists verbatim across turns on one thread. assert final["account_id"] == account_id
Put the three layers side by side and each one is catching a different class of bug, which is why none of them is optional once an agent ships:
Layer
Catches
Needs an LLM call
Cost
Node isolation
Wrong output from one function
No
Milliseconds
Trajectory
Wrong routing decision
Only if the node calls one
One graph run
Multi-turn simulation
Wrong accumulated outcome
Yes, several per test
Seconds per turn
LangGraph Debugging: State Inspection and Time-Travel Replay
Testing catches the bugs you thought to check for. Debugging is what you need when a run does something you didn't anticipate, and the checkpointer is already recording everything required for it, whether or not you set out to test that particular run.
Start with graph.get_state(config). It returns the graph's current snapshot for a thread_id: every state field, plus next, the node or nodes that would run on the next invocation. That single call answers the question you ask first when a run looks stuck or wrong: what does the graph currently believe, and what is it about to do next.
When the current snapshot isn't enough, graph.get_state_history(config) returns every checkpoint for that thread, most recent first, each carrying its own config with a checkpoint_id. That's a full audit trail of every node execution in order, and it's the same data the trajectory test earlier reads for its assertion, meaning the debugging tool and the test are, structurally, the same call pointed at a different question.
Time-travel is what turns that history from readable into actionable. Pick an earlier checkpoint (say, the one right before draft_response ran), pass its config straight into graph.invoke(None, config), and the graph resumes from exactly that point: every node before it is skipped, everything after re-executes. Pair that with update_state on the earlier checkpoint, changing category before resuming, for instance, and the replay forks into a new timeline built from the corrected input, without re-running the parts of the graph that were never in question.
One detail that trips people up the first time: if any node in your graph is itself a compiled subgraph, get_state only returns the parent graph's fields by default. Pass subgraphs=True and the returned snapshot's tasks include each subgraph's own state and its own config, so you can drill into a subgraph's checkpoint history the same way you'd drill into the parent's. Skipping that flag is the usual reason a debugging session stalls on "the state looks right, so why did the subgraph node behave differently," when the subgraph's internal state was never inspected in the first place.
"""Layer 4 - time-travel debugging via checkpoint replay.This is a standalone script (not a pytest file). It shows how to: 1. Run a ticket to completion and inspect the final state. 2. Walk the full checkpoint history. 3. Rewind to an earlier checkpoint, change one field to fork the timeline, and replay forward from that point - watching the outcome change.Run it directly (no API keys required): python scripts/debug_replay.py"""import osimport sysfrom uuid import uuid4# Make the repo root importable so `from src.graph import ...` works when this# script is run directly as `python scripts/debug_replay.py`.sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))from langgraph.checkpoint.memory import InMemorySaverfrom src.graph import builderdef main(): graph = builder.compile(checkpointer=InMemorySaver()) config = {"configurable": {"thread_id": str(uuid4())}} # 1. Run a routine ticket that ends up resolved. graph.invoke( { "ticket_text": "How do I change the email address on my account?", "account_id": "ACC-1002", }, config, ) current = graph.get_state(config) print("=== Original run ===") print("final status :", current.values["status"]) print("state .next :", current.next) # empty -> the run is complete print() # 2. Walk the checkpoint history, newest first, noting which node was about # to run at each checkpoint and its checkpoint_id. print("=== Checkpoint history (newest first) ===") target_config = None for snapshot in graph.get_state_history(config): checkpoint_id = snapshot.config["configurable"]["checkpoint_id"] about_to_run = snapshot.next or ("<end>",) print(f"checkpoint {checkpoint_id} next={about_to_run}") # 3. Find the checkpoint right before draft_response ran. if "draft_response" in snapshot.next: target_config = snapshot.config print() if target_config is None: raise RuntimeError("Could not find a checkpoint before draft_response") # 4. Fork the timeline: rewind to that checkpoint and reclassify the ticket # as a billing dispute, then replay forward from that point. forked_config = graph.update_state( target_config, values={"category": "billing_dispute"}, ) forked_final = graph.invoke(None, forked_config) print("=== Forked replay (category changed to billing_dispute) ===") print("forked status:", forked_final["status"]) print(f"Original resolved -> forked {forked_final['status']} by changing one") print("field at an earlier checkpoint and replaying forward.") # If any node here were itself a compiled subgraph, graph.get_state(config, # subgraphs=True) is the call to reach for to inspect nested state. This # example uses no subgraphs, so we don't need it.if __name__ == "__main__": main()
Wiring the Three Layers Into CI
Node tests are cheap enough to run on every commit: a few hundred milliseconds each, no LLM calls if the nodes are pure functions. Trajectory tests cost one graph run and belong on every pull request. Multi-turn simulations are the expensive ones, several LLM calls per test, so they belong on a slower cadence: a one-scenario smoke pass on every PR, the full multi-turn suite nightly, and a failing nightly run treated as an incident to triage that day rather than a stat to note in passing.
None of the three layers above touches the product the graph is embedded in, and that is deliberate scope, not an oversight. Catching whether the right decision actually reached the customer's screen is a job for the layer above the graph, which is where Autonoma and behavioral end-to-end testing come in, not where LangGraph's own testing surface was ever trying to compete.
If your stack is CrewAI instead of LangGraph, the same three-layer arc applies almost unchanged: isolate a single agent's role logic, verify the crew's task routing, then simulate a multi-turn run. The specifics of how you seed and inspect state differ enough to matter, and our CrewAI evaluation guide walks through those specifics end to end.
Frequently Asked Questions
LangGraph testing means verifying a compiled agent graph at three levels: node-level unit tests that exercise a single node in isolation via the checkpointer, trajectory tests that assert the graph visited the expected sequence of nodes and tools, and multi-turn simulations that drive several turns over one thread and check the accumulated state. It also includes debugging with state inspection and time-travel replay, since both rely on the same checkpointer data the tests already use.
Compile the graph with an InMemorySaver checkpointer and interrupt_after set to the node under test, a static setting passed to compile(), not invoke(). Use update_state with as_node set to the upstream node's name to seed the state as if that node had already run. Resume with invoke(None, config); the graph runs exactly that one node and stops, because the compiled interrupt fires right after it. Then call get_state and assert on the fields the node is supposed to have added.
Run the graph once, then recover the executed node sequence either by streaming with stream_mode="updates" and reading the node name off each update, or by calling get_state_history after the run and reversing the most-recent-first list of checkpoints. Compare that sequence to the list of node names you expected for the given input.
Call get_state to see the graph's current snapshot and what it will run next. If that is not enough, call get_state_history to see every checkpoint for the thread in order. Pick the checkpoint right before the point things went wrong, pass its config into invoke to replay from there, and optionally call update_state first to fork the replay with corrected input.
No. InMemorySaver keeps checkpoints in process memory, so state disappears on restart and never shares across processes. It is meant for tests and local development. Production deployments need a durable checkpointer, such as a Postgres or SQLite-backed saver, so a thread's state survives restarts and scales across multiple workers.
