Time-Travel: State History, Replay, and Forking a Bad Run (LangGraph in Production, Module 08)

This is Module 08 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

The Problem: You Approved the Wrong Plan

Forge just finished a 90-second run against a Tally issue: “Bug: report.py crashes when a category has no expenses.” You watched the Planner produce its output, got the HITL prompt, approved the plan — and the tests passed. Done, right?

Then you look at the PR and realize that storage.py was modified. The Planner emitted a Plan with target_files=["tally/report.py", "tally/storage.py"] and risk="high". You approved it without noticing. The original branch is clean, but Forge has persisted every superstep into the SqliteSaver. The edits are applied. The PR is there.

If only you could go back to the state just before the Planner produced that plan and relaunch from there with a corrected one. Or better: branch off a parallel timeline where storage.py is never touched, without losing the original trajectory.

That is exactly what time-travel means in LangGraph: the checkpoint history is already there, recorded automatically by the persistence layer you wired up in Module 6. This module teaches you to read it, replay from a specific point, and fork an alternate trajectory — without touching a single line of forge/.

In This Module

You’ll learn:

Walk the full checkpoint history of a Forge run with get_state_history (newest-first) to locate exactly where the run went wrong.
Replay from a past checkpoint_id by passing it to graph.invoke(None, config) — re-executing only the nodes after the target checkpoint.
Fork a trajectory with graph.update_state(config, values, as_node=) to inject corrected state and branch a new run without destroying the original.
Combine walk + replay + fork to debug a Forge trajectory end-to-end, deterministically, without any LLM API calls.
Understand how LangGraph Studio (Module 14) visualizes this entire workflow in a graphical UI — one-sentence preview.

You’ll build: A time_travel_demo.py script (and the forge/timetravel.py helper module) that walks a failed Forge run’s checkpoint history, replays from a past checkpoint, and forks a bad Plan into a corrected trajectory — using get_state_history, update_state, and invoke(None, config), with zero changes to forge/graph.py, forge/state.py, or any other part of the existing package.

Concepts covered: CA8 — Time-travel & debugging (core). This module BUILDs CA8.1–CA8.4 of the LangGraph theory inventory; CA8.5 (Studio time-travel) is covered as a theory preview.

Prerequisites: Modules 1–7, the full cumulative forge/ package (M7), SqliteSaver operational (M6), interrupt()/Command(resume=) understood (M7), .env with ANTHROPIC_API_KEY (or Ollama fallback for $0).

Where We Are

M1–M7 ✅ | M8 👉 | M9–M16 ⬜

Before M8: Forge is persisted (M6), human-gated (M7), and survives a process kill. But when a run goes sideways — wrong plan, wrong route, bad HITL decision — you’re blind. The only option has been to restart from scratch.

After M8: you can inspect every superstep’s exact state, replay from any checkpoint in the history, and inject a corrected value to branch an alternative trajectory. None of that requires changing the graph or adding a field to ForgeState. M8 is entirely a read/write API on top of the persistence layer installed in M6.

The production resilience tools — durability modes, RetryPolicy, CachePolicy — arrive in M9. That’s the next step.

The Checkpoint Stream: Walking History with `get_state_history`

You saw get_state(config) in Module 6. It returns a single StateSnapshot — the current state of a thread. get_state_history is different: it’s a generator that yields every snapshot ever recorded for a thread, newest first.

The distinction matters. When you iterate graph.get_state_history(config), the first thing you see is the most recent checkpoint. The last thing you see is the very first checkpoint created when the run started. This newest-first ordering is canonical v1.x behavior — never assume the reverse.

Here is the real timetravel.history helper from the lab’s forge/timetravel.py:

def history(graph, thread_id: str) -> list:
    """All state snapshots for a thread, newest-first."""
    return list(
        graph.get_state_history({"configurable": {"thread_id": thread_id}})
    )

That’s the entire implementation. get_state_history accepts the same config dict as invoke. It streams StateSnapshot objects from the checkpointer; list() materializes them all at once.

What a `StateSnapshot` Carries

StateSnapshot is imported from langgraph.types. Each instance exposes:

values — the complete ForgeState dict at that point in time.
next — a tuple of node names scheduled to run next from this checkpoint. An empty tuple means the run is terminal.
config — the full config dict including checkpoint_id and checkpoint_ns in config["configurable"].
metadata — a dict with step (integer, 0-indexed), source (the node that wrote this checkpoint), writes (the partial state updates from that node), and parents.
created_at — ISO timestamp.
parent_config — the config of the preceding checkpoint (the causal chain).
tasks — any pending writes not yet committed.

Here is how you walk a completed Forge run to print every superstep:

from forge.graph import build_graph
from forge import checkpoint

graph = build_graph(checkpointer=checkpoint.in_memory())
config = {"configurable": {"thread_id": "run-bad-plan-01"}}

# Walk newest → oldest
for snapshot in graph.get_state_history(config):
    step    = snapshot.metadata.get("step", "?")
    source  = snapshot.metadata.get("source", "?")
    next_   = snapshot.next
    ckpt_id = snapshot.config["configurable"]["checkpoint_id"]
    print(f"step={step:>2}  source={source:<16}  next={str(next_):<25}  {ckpt_id[:8]}…")

Running this against a full Forge run produces output like this (newest at the top):

step= 9  source=pr_approval      next=()                     a3f9c1…
step= 8  source=tester           next=('pr_approval',)       b7e204…
step= 7  source=apply            next=('tester',)            c1d83b…
step= 6  source=editor           next=('apply',)             d09f11…
step= 5  source=plan_approval    next=('editor',)            e5a220…
step= 4  source=planner          next=('plan_approval',)     f2b34c…
step= 3  source=triage           next=('planner',)           09cc7a…
step= 2  source=intake           next=('triage',)            1a4f88…
step= 1  source=__start__        next=('intake',)            2c89d0…
step= 0  source=__start__        next=('intake',)            3d12e1…

Notice next=() at step 9: the run is complete. At step 4, source=planner with next=('plan_approval',): that is the checkpoint written just after the Planner node ran and emitted its Plan. That is your fork point.

To locate it programmatically:

snapshots = history(graph, "run-bad-plan-01")

# Find the checkpoint right after the planner wrote its Plan
planner_snap = next(
    s for s in snapshots
    if s.metadata.get("source") == "planner"
)
print("Bad plan:", planner_snap.values["plan"].target_files)
# → ['tally/report.py', 'tally/storage.py']  ← WRONG

get_state_history costs zero API calls. It reads directly from the checkpointer — SQLite or Postgres, same interface. This is my first debugging tool in production.

The Checkpoint Chain (Mermaid)

flowchart TB
    C9["step=9 (newest)\nsource=pr_approval\nnext=()\nvalues: pull_request=PullRequest(...)"]
    C8["step=8\nsource=tester\nnext=('pr_approval',)\nvalues: test_result=TestResult(passed=True)"]
    C7["step=7\nsource=apply\nnext=('tester',)\nvalues: edits=[FileEdit(...)]"]
    C6["step=6\nsource=editor\nnext=('apply',)"]
    C5["step=5\nsource=plan_approval\nnext=('editor',)\nvalues: plan_approved=True"]
    C4["step=4 ← FORK POINT\nsource=planner\nnext=('plan_approval',)\nvalues: plan=Plan(risk='high', target_files=[...])"]
    C3["step=3\nsource=triage\nnext=('planner',)\nvalues: route='bug'"]
    C2["step=2\nsource=intake\nnext=('triage',)"]
    C1["step=1 (oldest)\nsource=__start__\nnext=('intake',)\nvalues: {}"]

    C9 --> C8 --> C7 --> C6 --> C5 --> C4 --> C3 --> C2 --> C1

get_state_history yields C9 first and C1 last. The fork point is C4 — the checkpoint just after the Planner wrote the bad Plan. The parent_config of each snapshot points to the one below it in this chain.

Replay: Pick Any Checkpoint and Run Forward

Replay means taking an existing checkpoint and re-executing the graph from that point forward. The API is the same call you used in Module 7 to resume after an interrupt():

result = graph.invoke(None, snapshot.config)

None as the input tells LangGraph: “do not inject new input; resume from the checkpoint in this config.” The checkpoint_id inside snapshot.config["configurable"] is what tells LangGraph which checkpoint to load.

The real timetravel.replay_from helper is two lines:

def replay_from(graph, snapshot) -> dict:
    """Re-run from a past checkpoint: nodes before it replay from saved state."""
    return graph.invoke(None, snapshot.config)

When LangGraph receives this call, it loads the state at the given checkpoint_id, reads the next tuple from that snapshot, and executes those nodes — and any that follow — in order. Nodes before the target checkpoint do not re-run. Their state is already in the persisted record.

Here is a complete replay of the Planner checkpoint in the context of the test suite (tests/test_pipeline.py):

def test_time_travel_reads_history_of_a_forge_run(monkeypatch, make_routed_model):
    repo = _setup(monkeypatch, make_routed_model)
    graph = build_graph(checkpointer=checkpoint.in_memory())
    out, cfg = _run_to_completion(graph, repo, "tt")
    assert out["test_result"].passed is True

    history = list(graph.get_state_history(cfg))
    assert len(history) >= 6  # many super-steps recorded

    # The plan can be recovered from a past checkpoint.
    with_plan = [s for s in history if s.values.get("plan")]
    assert with_plan and with_plan[-1].values["plan"].summary == "fix empty-category KeyError"

This test verifies that after a complete run, get_state_history contains at least 6 checkpoints and that an old snapshot carries the original Plan in its values. That plan is there forever — checkpoints are immutable records.

What Replay Does NOT Do

⚠️ Common misconception: “Replay re-runs the whole graph from scratch.”

False. Replay resumes exactly from the state recorded at the target checkpoint. Nodes before that checkpoint do not re-execute — their outputs are already in the checkpointer. Only the nodes listed in snapshot.next, and those that follow, run again.

Second confusion (M7 → M8): some engineers assume that calling invoke(None, config) with a past checkpoint_id overwrites or erases the forward history. Also false. The checkpointer keeps every checkpoint. Replaying from a past checkpoint extends the history; it does not delete any existing snapshot. If you want a different outcome, you need a fork (next section).

Freshness trap: there is no replay(checkpoint_id) method, no undo(), no rewind() in LangGraph v1.x. The canonical operation is invoke(None, config_with_checkpoint_id). The term “rewind” implies deletion — avoid it. LangGraph never deletes a checkpoint.

A pure replay from the Planner checkpoint replays the same Plan object, which means the same editors run, the same two files get modified, and the tests may fail again. Replay alone is not a fix. It is useful for reproducing a failure deterministically — same state, same execution path, same result. The fix is a fork.

Fork: `update_state` as a Scalpel on the Timeline

A fork writes a new value into the state at a past checkpoint, creating a brand-new checkpoint that inherits from the target but carries different state. Then you invoke the graph from that new checkpoint. The original trajectory — all of its checkpoints — remains intact.

The operation is graph.update_state:

def fork_from(graph, snapshot, values: dict, as_node: str | None = None) -> dict:
    """Branch an alternate history from a past checkpoint by writing new values.

    Returns the forked checkpoint config; pass it to `graph.invoke(None, config)`
    to run the new branch. The original history is left intact.
    """
    return graph.update_state(snapshot.config, values, as_node=as_node)

update_state takes three arguments:

config — the config of the checkpoint you want to branch from (with its checkpoint_id).
values — a partial state dict with the fields you want to overwrite.
as_node — the name of the node that is logically supposed to have produced these values.

The as_node parameter is critical. LangGraph uses the node name to look up outgoing edges and determine what next should be in the new checkpoint. If you omit as_node, LangGraph cannot derive the correct next tuple, and the fork will not continue to the right node.

The name must match the string used in builder.add_node(...). In Forge, the Planner registers as "planner" (lowercase):

builder.add_node("planner", nodes.planner)

So your update_state call uses as_node="planner" (lowercase string, not the node function itself).

Here is a complete fork that corrects the Plan to touch only report.py:

from forge.graph import build_graph
from forge import checkpoint, timetravel
from forge.models import Plan, PlanStep

graph = build_graph(checkpointer=checkpoint.in_memory())
# ... run a bad Forge run first, get its thread_id ...

snapshots = timetravel.history(graph, "run-bad-plan-01")
planner_snap = next(s for s in snapshots if s.metadata.get("source") == "planner")

corrected_plan = Plan(
    summary="Fix report.py ZeroDivisionError only — storage.py untouched",
    target_files=["tally/report.py"],          # ← the key correction
    steps=[
        PlanStep(
            description="Guard division in report.py",
            kind="edit",
            target_file="tally/report.py",
        ),
    ],
    risk="low",
)

# update_state creates a NEW checkpoint branching from planner_snap.config
forked_cfg = timetravel.fork_from(
    graph,
    planner_snap,
    {"plan": corrected_plan},
    as_node="planner",
)

# forked_cfg is the config of the new checkpoint — invoke resumes from here
forked_result = graph.invoke(None, forked_cfg)
print("Edits in forked run:", [e.path for e in forked_result["edits"]])
# → ['tally/report.py']   ← one file only

The test in tests/test_module_08.py exercises the same pattern on a minimal two-node graph:

def test_history_replay_and_fork():
    graph = _graph(checkpoint.in_memory())
    cfg = {"configurable": {"thread_id": "tt"}}

    final = graph.invoke({"value": "", "log": []}, cfg)
    assert final["value"] == "ab"

    hist = timetravel.history(graph, "tt")
    after_a = next(s for s in hist if s.values.get("value") == "a")
    assert after_a.next == ("b",)

    # Fork: write a different value at the 'a' checkpoint, then run the branch.
    forked_cfg = timetravel.fork_from(graph, after_a, {"value": "X"}, as_node="a")
    branch = graph.invoke(None, forked_cfg)
    assert branch["value"] == "Xb"

    # Forking adds a branch to the history tree; the original "ab" is preserved.
    assert any(s.values.get("value") == "ab" for s in timetravel.history(graph, "tt"))

The last assertion proves the key invariant: forking does not erase history. The original "ab" trajectory is still visible in get_state_history alongside the new "Xb" branch.

What `update_state` Returns

update_state returns the config of the newly created checkpoint. That is what you pass to graph.invoke(None, ...) to continue from the fork. The returned config is a standard config dict with thread_id, checkpoint_id, and checkpoint_ns set correctly.

A thread can accumulate multiple branches. get_state_history yields all snapshots across all branches — you navigate the tree via parent_config.

Putting It Together: Debugging a Forge Trajectory End-to-End

Here is the complete debugging workflow. You will do all of this in time_travel_demo.py:

Run Forge on a known-bad issue (or find a failed historical run by its thread_id).
Walk history with timetravel.history(graph, thread_id) — print each step, source, next, and checkpoint_id[:8] to the console.
Inspect values at the suspect checkpoint: planner_snap.values["plan"], triage_snap.values["route"], etc.
Decide: replay (reproduce same outcome) or fork (inject a fix)?
Fork with timetravel.fork_from(graph, snap, {"plan": corrected_plan}, as_node="planner").
Invoke from the fork: graph.invoke(None, forked_cfg).
Verify: check forked_result["test_result"].passed, forked_result["edits"].

Here is a decision table to keep on hand:

What you need	API	Outcome
Inspect the full run history	`graph.get_state_history(config)` newest-first	Generator of `StateSnapshot`
Get current state only	`graph.get_state(config)`	One `StateSnapshot`
Replay from a past checkpoint (unchanged state)	`graph.invoke(None, snapshot.config)`	Run continues from `snapshot.next`
Inject a correction and branch	`graph.update_state(snap.config, values, as_node=)` + `graph.invoke(None, new_cfg)`	New checkpoint + forked trajectory

The Two Operations Visualized (Mermaid)

flowchart LR
    C0["C0\n__start__"]
    C1["C1\nintake"]
    C2["C2\ntriage"]
    C3["C3\nplanner\nbad plan"]
    C4_orig["C4\nplan_approval\ntests fail"]
    C5_orig["C5\napply\nwrong files"]

    C3_fork["C3'\nupdate_state\ngood plan"]
    C4_fork["C4'\nplan_approval"]
    C5_fork["C5'\napply\ncorrect file"]
    C6_fork["C6'\ntester\npassed ✅"]

    C0 --> C1 --> C2 --> C3 --> C4_orig --> C5_orig

    C3 -. "fork_from\n(update_state)" .-> C3_fork
    C3_fork --> C4_fork --> C5_fork --> C6_fork

    style C3_fork fill:#d4edda,stroke:#28a745
    style C4_fork fill:#d4edda,stroke:#28a745
    style C5_fork fill:#d4edda,stroke:#28a745
    style C6_fork fill:#d4edda,stroke:#28a745
    style C4_orig fill:#f8d7da,stroke:#dc3545
    style C5_orig fill:#f8d7da,stroke:#dc3545

The original trajectory (red) stays in the checkpointer. The forked trajectory (green) branches from C3, carries the corrected Plan, and reaches TestResult(passed=True) without touching storage.py.

In Module 14, LangGraph Studio gives you this entire workflow in a visual UI — click a checkpoint, edit state, replay, all without writing a single line of code.

Hands-on Lab: Build It

Objective: Explore and fix a failed Forge trajectory by walking get_state_history (newest-first), replaying from a past checkpoint, and injecting a corrected Plan via update_state — all in time_travel_demo.py, without modifying a single line of forge/.

What You’re Building

The lab has one new file at the root of module-08/: forge/timetravel.py (the helper module) and time_travel_demo.py (the runnable demo). The forge/ package, tally/, and all tests from M1–M7 are inherited byte-identical.

Step 1: The `forge/timetravel.py` Helper

The complete forge/timetravel.py from the lab (all three functions, all of the module’s logic):

"""Time-travel helpers (Module 8): inspect history, replay, and fork a run.

Persistence (Module 6) records a checkpoint at every super-step. Time-travel
uses those checkpoints to debug a bad Forge trajectory: list the history,
replay from a past point, or fork an alternate branch by writing new state
at an old checkpoint. Requires a checkpointer.
"""
from __future__ import annotations


def history(graph, thread_id: str) -> list:
    """All state snapshots for a thread, newest-first."""
    return list(graph.get_state_history({"configurable": {"thread_id": thread_id}}))


def replay_from(graph, snapshot) -> dict:
    """Re-run from a past checkpoint: nodes before it replay from saved state."""
    return graph.invoke(None, snapshot.config)


def fork_from(graph, snapshot, values: dict, as_node: str | None = None) -> dict:
    """Branch an alternate history from a past checkpoint by writing new values.

    Returns the forked checkpoint config; pass it to `graph.invoke(None, config)`
    to run the new branch. The original history is left intact.
    """
    return graph.update_state(snapshot.config, values, as_node=as_node)

Three functions. Zero new imports beyond what graph already carries. Zero changes to forge/graph.py, forge/state.py, or forge/nodes.py.

Step 2: The Unit Test in `tests/test_module_08.py`

The unit test exercises time-travel on a minimal two-node graph so it stays independent of Forge’s evolving structure:

class _S(TypedDict):
    value: str
    log: Annotated[list, operator.add]

def _graph(saver):
    b = StateGraph(_S)
    b.add_node("a", lambda s: {"value": "a", "log": ["a"]})
    b.add_node("b", lambda s: {"value": s["value"] + "b", "log": ["b"]})
    b.add_edge(START, "a")
    b.add_edge("a", "b")
    b.add_edge("b", END)
    return b.compile(checkpointer=saver)

The graph writes "a" then appends "b", giving "ab". The test forks the "a" checkpoint with {"value": "X"} and asserts the forked result is "Xb" — while the original "ab" is still in history.

This is the core mechanic distilled to its minimum: a deterministic graph, an InMemorySaver, no LLM calls, no network.

Step 3: The Integration Test in `tests/test_pipeline.py`

The integration test runs a complete Forge flow against a real Tally sandbox, then queries the history:

def test_time_travel_reads_history_of_a_forge_run(monkeypatch, make_routed_model):
    repo = _setup(monkeypatch, make_routed_model)
    graph = build_graph(checkpointer=checkpoint.in_memory())
    out, cfg = _run_to_completion(graph, repo, "tt")
    assert out["test_result"].passed is True

    history = list(graph.get_state_history(cfg))
    assert len(history) >= 6  # many super-steps recorded
    with_plan = [s for s in history if s.values.get("plan")]
    assert with_plan and with_plan[-1].values["plan"].summary == "fix empty-category KeyError"

The _setup fixture monkeypatches agents.plan_change to return a canned Plan and monkeypatches config.get_model to return a scripted model — no API keys, no network, fully deterministic. The Tester node (nodes.tester) calls sandbox.run_tests, which shells out to real pytest in a tempdir sandbox. That is your observable signal: TestResult(passed=True) without a single LLM call in the time-travel portion.

Step 4: Building `time_travel_demo.py`

The demo script shows the three trajectories: original run, pure replay, and corrected fork. Here is the core structure:

"""time_travel_demo.py — Module 8 demo: walk, replay, and fork a Forge run."""
from pathlib import Path
from forge import checkpoint, timetravel
from forge.graph import build_graph
from forge.models import Plan, PlanStep
from forge import sandbox
from langgraph.types import Command

def run_bad_forge(graph, repo: str, thread_id: str) -> None:
    """Run Forge with a plan that touches two files (simulated bad run)."""
    cfg = {"configurable": {"thread_id": thread_id}, "recursion_limit": 50}
    graph.invoke({"issue": "Bug: report.py crashes on empty category", "repo_path": repo}, cfg)
    # Approve plan and PR to let the run complete
    graph.invoke(Command(resume={"approved": True}), cfg)
    graph.invoke(Command(resume={"approved": True}), cfg)

def walk_and_print(graph, thread_id: str) -> list:
    """Print the history table and return the snapshots (newest-first)."""
    snaps = timetravel.history(graph, thread_id)
    print(f"\n=== History for thread '{thread_id}' (newest → oldest) ===")
    for s in snaps:
        step   = s.metadata.get("step", "?")
        source = s.metadata.get("source", "?")
        ckpt   = s.config["configurable"]["checkpoint_id"][:8]
        print(f"  step={step:>2}  source={source:<16}  next={str(s.next):<25}  {ckpt}…")
    return snaps

The fork section:

def fork_and_run(graph, snaps, repo: str, thread_id: str) -> dict:
    """Inject a corrected Plan at the planner checkpoint and run the branch."""
    planner_snap = next(s for s in snaps if s.metadata.get("source") == "planner")
    print(f"\n  Bad plan target_files: {planner_snap.values['plan'].target_files}")

    good_plan = Plan(
        summary="Fix report.py KeyError only",
        target_files=["report.py"],
        steps=[PlanStep(description="Guard missing key in report.py",
                        kind="edit", target_file="report.py")],
        risk="low",
    )
    forked_cfg = timetravel.fork_from(
        graph, planner_snap, {"plan": good_plan}, as_node="planner"
    )
    result = graph.invoke(None, forked_cfg)
    files = [e.path for e in result.get("edits", [])]
    passed = result.get("test_result") and result["test_result"].passed
    print(f"  Forked run edits: {files}, tests passed: {passed}")
    return result

Expected Output

Running uv run python time_travel_demo.py prints:

=== History for thread 'run-bad-plan-01' (newest → oldest) ===
  step= 9  source=pr_approval      next=()                     a3f9c1…
  step= 8  source=tester           next=('pr_approval',)       b7e204…
  step= 7  source=apply            next=('tester',)            c1d83b…
  step= 6  source=editor           next=('apply',)             d09f11…
  step= 5  source=plan_approval    next=('editor',)            e5a220…
  step= 4  source=planner          next=('plan_approval',)     f2b34c…
  step= 3  source=triage           next=('planner',)           09cc7a…
  step= 2  source=intake           next=('triage',)            1a4f88…
  step= 1  source=__start__        next=('intake',)            2c89d0…

  Bad plan target_files: ['tally/report.py', 'tally/storage.py']  ← WRONG

=== Replay from planner checkpoint (same state, same outcome) ===
  Replayed edits: ['tally/report.py', 'tally/storage.py']
  (replay ≠ fix — same plan, same two files)

=== Fork at planner checkpoint: injecting corrected Plan ===
  Forked run edits: ['report.py'], tests passed: True  ✅

=== Summary ===
Original trajectory : 2 FileEdits (report.py + storage.py)
Replayed trajectory : 2 FileEdits (same) — tests same result
Forked trajectory   : 1 FileEdit  (report.py only) — TestResult(passed=True) ✅

The TestResult(passed=True) is the observable signal: Tally’s real pytest confirms that the forked trajectory is correct. No LLM API call was needed for the time-travel portion — the corrected Plan is a Python constant.

Try It Yourself

Once you have the demo running, experiment with these extensions:

(a) Fork the triage checkpoint. Find the snapshot where source="triage" and fork it with {"route": "feature"} using as_node="triage". Observe how the downstream routing changes — Forge still goes to the Planner but now carries route="feature" in its state.

(b) Follow the parent_config chain. Add code that starts from the newest snapshot and walks backward via parent_config to reconstruct the full causal chain. This is the lowest-level debugging view.

(c) Count checkpoints vs. attempts. Run Forge with a deliberately failing Fixer (monkeypatch it to always return bad code) and count how many checkpoints are created for 1 attempt vs. 3 attempts. Check the growth of the SQLite file. This gives you an intuition for the checkpoint storage cost in production.

In Production

get_state_history is, in my experience, the most underused LangGraph API in production systems. Engineers reach for logs or LangSmith traces first. But get_state_history lets you inspect the exact typed state at every superstep — no parsing, no string matching. It costs zero API calls.

Here are the three scenarios where I reach for it first:

Approved the wrong plan? Don’t restart from scratch — that costs tokens and time. Instead, find the planner snapshot via get_state_history, build a corrected Plan object, and update_state it in. Then invoke(None, forked_cfg). You skip intake, triage, and the Planner’s full LLM loop. Cost: the Editor + Tester only.

Flaky test on the third attempt? The Tester is deterministic (it shells out to pytest), so the test genuinely failed at attempt 3. To verify that it was the edit that caused the failure and not a fixture issue, replay from the checkpoint just before the Tester. You get the exact same execution with the exact same files — reproducible in under two seconds.

Regression appeared between yesterday and today? Pull both thread_ids from the checkpointer, call get_state_history on each, and compare the plan field at the planner snapshot. You see in one dict comparison whether the Planner’s output changed. No digging through logs.

In production I use update_state as a scalpel, not a hammer. Always set as_node= to the exact node name that should have produced the values — otherwise next in the new checkpoint is wrong and your invoke goes nowhere useful. And as_node="planner" (lowercase) must match builder.add_node("planner", ...) exactly; a capitalization mismatch is silent and maddening to debug.

One more consideration for production: the SqliteSaver keeps every checkpoint indefinitely. There is no default TTL or pruning policy. A thread that runs a 3-attempt fix loop generates 12+ checkpoints per run. Across hundreds of runs, the SQLite file grows steadily. In production you would use PostgresSaver (introduced in Module 6 as theory) with a retention policy — for example, pruning checkpoints older than 30 days or keeping only the last N threads. Module 9 covers the durability stack in detail.

LangGraph Studio (Module 14) automates all of this with a visual UI — click a checkpoint in the timeline, edit any field, click replay.

Mastery Corner

What to Really Understand Here

The time-travel API has three layers:

Reading (get_state_history): a generator that yields every recorded StateSnapshot for a thread, newest-first. Each snapshot is a complete picture of the state at one superstep, with values, next, metadata.step, checkpoint_id, and parent_config.
Resuming (invoke(None, snapshot.config)): re-executes the graph starting from snapshot.next, using snapshot.values as the starting state. Nodes before the checkpoint do not re-run. This is identical to the HITL resumption from Module 7 — the only difference is which checkpoint_id you put in the config.
Forking (update_state(snap.config, values, as_node=) + invoke(None, new_cfg)): creates a new checkpoint that inherits from the target checkpoint but carries different state. The original checkpoint chain is untouched. as_node is the name of the node that is supposed to have produced these values — LangGraph uses it to derive the outgoing edges and set next correctly in the new checkpoint.

CA8.5 (Studio time-travel) is a graphical wrapper around exactly these three operations. Everything Studio does under the hood is get_state_history, update_state, and invoke(None, config).

5 Mastery Questions

Q1. You run for snap in graph.get_state_history(config) on a thread that completed 8 supersteps. Which statement is true?

A) The first snapshot yielded has step=0 and source="__start__".
B) The first snapshot yielded has the highest step value and next=().
C) Snapshots are yielded in insertion order — oldest first, newest last.
D) get_state_history raises an error if the thread has no checkpoints.

Q2. Your colleague writes graph.invoke(Command(resume={"approved": True}), cfg) after an interrupt() to resume a HITL gate. You write graph.invoke(None, past_snapshot.config) to replay a past checkpoint. LangGraph uses the same invoke signature for both. What distinguishes the two operations at the framework level?

A) The type of the first argument: Command vs None.
B) The checkpoint_id in config["configurable"]: pointing to the current (latest) checkpoint causes resumption; pointing to a past checkpoint causes replay.
C) invoke(None, ...) always replays from the beginning of the thread.
D) You must set a replay=True flag in config["configurable"] for replay to work.

Q3. You fork a Forge run at the planner checkpoint by calling graph.update_state(snap.config, {"plan": new_plan}) — but you omit the as_node argument. What is the most likely outcome?

A) LangGraph infers as_node from the node that wrote the original checkpoint and continues normally.
B) The new checkpoint’s next tuple is empty or incorrect, so graph.invoke(None, forked_cfg) does nothing or routes to the wrong node.
C) update_state raises a ValueError immediately because as_node is required.
D) The plan value is silently ignored and the original plan is preserved.

Q4. A Forge run produced three failed fix attempts due to an incorrect initial Plan. A teammate says: “Just replay from the planner checkpoint — the model will generate a better plan this time.” You say: “We should fork and inject a corrected plan manually.” Which approach avoids a new LLM call to the Planner?

A) Replay — because LangGraph caches the Planner’s previous output and uses it again.
B) Fork with update_state — because you write the corrected Plan as a Python constant; the Planner node never re-runs.
C) Both approaches call the Planner again; the difference is only in how many checkpoints are created.
D) Neither approach avoids an LLM call; you must restart from scratch to change the plan.

Q5. After you run update_state + invoke(None, forked_cfg), you call get_state_history(original_config) again. Which statement is true?

A) The original trajectory’s checkpoints have been replaced by the forked trajectory’s checkpoints.
B) get_state_history raises an error because the thread now has conflicting branches.
C) Both the original trajectory and the forked trajectory’s checkpoints are visible in the history; the original is still intact.
D) Only the forked trajectory is returned; the original was deleted when update_state was called.

Answers

Q1 → B. get_state_history yields snapshots newest-first. The first snapshot has the highest step number and next=() (the terminal state). The step=0/source="__start__" snapshot comes last. (CA8.3)

Q2 → B. The invoke call is the same in both cases. What differs is the checkpoint_id in config["configurable"]. When that ID points to the current (latest) checkpoint of the thread, the call resumes after the most recent interrupt(). When it points to a past checkpoint, the call replays from that older state. Answer A is partially relevant (the first arg type) but not what LangGraph uses to distinguish the two semantics. (CA8.1 + CA7.1)

Q3 → B. Without as_node, LangGraph cannot determine which node’s outgoing edges to follow. It cannot set next correctly in the new checkpoint. The result is an empty or wrong next, and invoke(None, forked_cfg) either does nothing or routes unexpectedly. No exception is raised. (CA8.2)

Q4 → B. Replay re-executes the Planner node with its original state — the LLM runs again and may produce the same bad plan or a different one, but you pay an API call either way. A fork with update_state writes the corrected Plan as a Python literal; the Planner node never runs. The replay is useful for reproducing the original failure deterministically, not for correcting it. (CA8.1 + CA8.2)

Q5 → C. update_state creates a new checkpoint that inherits from the target checkpoint. It does not delete or overwrite existing checkpoints. get_state_history yields all checkpoints across all branches of the thread tree, linked by parent_config. Both the original and the forked trajectory are visible. (CA8.2 + CA8.3)

Traps to Avoid

Trap 1 — Confusing replay with correction. Replay re-executes the exact same state. If the Planner wrote a bad Plan, replaying from the Planner checkpoint runs the Planner again with the same inputs — you get the same (or a randomly different) plan, but you do not control it. Correction requires a fork: use update_state to inject the value you want before running the downstream nodes. This is the central confusion of the module and the most common mistake in production debugging.

Trap 2 — Omitting or mistyping as_node. as_node must be the string used in builder.add_node("planner", ...) — not the node function, not a capitalized name, not a display label. as_node="Planner" (capital P) fails silently: LangGraph finds no registered node with that name, sets next=(), and your invoke does nothing. Always check the string against your build_graph() implementation.

Trap 3 (freshness/terminology) — Using “rewind,” “undo,” or “rollback.” These words imply deletion. LangGraph never deletes a checkpoint. The correct terminology is time-travel, replay, and fork. This matters beyond naming: because checkpoints accumulate, your SQLite or Postgres store grows with every fork. In production, design a retention policy. And if a colleague says “let me rewind the agent” — gently correct them: you are not rewinding, you are replaying or forking from a past checkpoint while the original history remains intact.

Key Takeaways

get_state_history(config) is a generator that yields every recorded StateSnapshot for a thread, newest-first. The first snapshot has the highest step and next=(); the last has step=0 and source="__start__".
Each StateSnapshot exposes values (full state), next (scheduled nodes), config (with checkpoint_id), metadata (step, source, writes), created_at, parent_config, and tasks.
Replay = graph.invoke(None, snapshot.config). Resumes from snapshot.next using the snapshot’s state. Nodes before the checkpoint do not re-run. Same idiom as HITL resumption (M7) — the checkpoint_id in config is what selects which checkpoint to load.
Fork = graph.update_state(snap.config, values, as_node=) creates a new checkpoint branching from snap. The original trajectory is untouched. Invoke from the returned config to run the new branch.
as_node must be the exact string used in builder.add_node(...). Wrong name → wrong next → silent no-op. No exception.
M8 = zero graph changes, zero new ForgeState fields. All three APIs — get_state_history, update_state, invoke(None, config) — operate on the persistence layer wired in M6. The only new file is forge/timetravel.py (three thin wrappers).
Time-travel ≠ deletion. LangGraph never removes a checkpoint. Every update_state adds a checkpoint. Avoid the terms “rewind,” “undo,” and “rollback” — the canonical terms are time-travel, replay, and fork.
In production: the SqliteSaver has no default TTL. Use PostgresSaver with a retention policy in any deployment that accumulates many threads.

What’s Next

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

Forge can now replay and fork its own bad runs. But what happens when the process crashes mid-flight, or the Tester flakes on a transient filesystem error? Module 9 hardens Forge with durability modes (durability="exit", "async", "sync"), a RetryPolicy on the Tester node, and CachePolicy on repo analysis — production resilience built in.

Navigate the series:

References

LangGraph — How to view and update past graph state (time-travel): https://langchain-ai.github.io/langgraph/how-tos/time-travel/
LangGraph API reference — get_state_history / StateSnapshot: https://langchain-ai.github.io/langgraph/reference/graphs/
LangGraph API reference — update_state: https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.graph.CompiledGraph.update_state
LangGraph Studio (time-travel visual UI — Module 14 preview): https://smith.langchain.com/studio
Module 08 lab code — langgraph-prod-labs/module-08/: https://github.com/dupuis1212/langgraph-prod-labs/tree/main/module-08

All URLs verified as of June 2026 against LangGraph v1.2.5 / langchain v1.3.9 / langchain-core v1.4.7.