6. Time-travel Debugging

Overview

In the previous capsules you learned to save an agent's state with checkpointing (03), persist it in PostgreSQL (04), and build long-term memory across sessions (05). Every time a node runs, LangGraph saves a checkpoint — a complete snapshot of the state at that instant. So far you used those checkpoints to resume interrupted runs. But there's a far more powerful use: navigating an agent's complete execution history, going back to any point, modifying the state, and re-running from there.

That's time-travel debugging. Think of git: you can see the commit history, check out an earlier commit, create a branch from that point, and explore a different path. LangGraph gives you exactly the same thing for agent execution. Did the agent make a bad decision at step 3? Go back to step 2, change the state, and re-run to see what would have happened. Want to try the same research with a different parameter? Replay from the initial checkpoint with different values.

In this capsule you'll master three fundamental operations: navigating the history with get_state_history(), replaying from any checkpoint with invoke(None, checkpoint_config), and modifying + re-running with update_state(). You'll also see how to combine time-travel with human-in-the-loop to create flows where a human can inspect, correct, and resume an agent's execution in real time.

What Time-travel Debugging Is

The git analogy

When you work with git, each commit is a snapshot of your code at a specific moment. You can:

• git log → see every commit in chronological order
• git checkout abc123 → go back to the code's state at that commit
• git checkout -b fix-branch abc123 → create a new branch from that point and take a different path

Time-travel debugging in LangGraph works the same, but for agent execution:

Why it's extraordinary for agents

An agent isn't a linear program. It's a system that makes decisions: which tool to call, which query to use, when to stop searching, how to synthesize results. When something goes wrong — the agent picked the wrong tool, used a bad query, or got confused by a result — you need to understand where exactly it went off track.

Without time-travel, your only option is to re-run everything from the start and hope to reproduce the problem. With time-travel:

1. Navigate the history and find the checkpoint where everything was fine
2. Identify the next checkpoint where something went wrong
3. Go back to the good checkpoint
4. Modify the state (fix the query, change a parameter)
5. Re-run and verify the result is different

It's surgical debugging: you don't repeat everything — you only repeat from the exact point where you need to change something.

Anatomy of a checkpoint (quick recap)

Each checkpoint contains all the information needed to restore the agent:

state = app.get_state(config)

state.values          # The complete state: messages, custom variables, everything
state.next            # The next node(s) to run — () if the graph finished
state.config          # Config with thread_id and checkpoint_id
state.metadata        # Metadata: the node that created it, timestamp
state.parent_config   # The previous checkpoint's config (to navigate backward)

The state.config field is the key to time-travel: it contains the exact checkpoint_id. If you pass that config to invoke, the graph restores that state and continues from there.

Navigating the State History

get_state_history: the complete timeline

get_state_history() returns an iterator with every checkpoint of a thread, newest to oldest:

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.tools import tool
from typing import TypedDict, Annotated

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    sources: list[str]
    current_step: str
    steps_completed: int

model = init_chat_model("openai:gpt-4.1-mini")

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': relevant information found."

def search_node(state: ResearchState) -> dict:
    step = state.get("steps_completed", 0) + 1
    source = f"source-{step}.example.com"
    return {
        "messages": [AIMessage(content=f"I searched {source}")],
        "sources": state.get("sources", []) + [source],
        "current_step": "searching",
        "steps_completed": step
    }

def analyze_node(state: ResearchState) -> dict:
    n = len(state.get("sources", []))
    return {
        "messages": [AIMessage(content=f"Analysis of {n} sources complete.")],
        "current_step": "analyzed"
    }

def report_node(state: ResearchState) -> dict:
    return {
        "messages": [AIMessage(content="Final report generated.")],
        "current_step": "completed"
    }

def route(state: ResearchState) -> str:
    if state.get("steps_completed", 0) < 3:
        return "search_more"
    if state.get("current_step") != "analyzed":
        return "analyze"
    return "report"

graph = StateGraph(ResearchState)
graph.add_node("search", search_node)
graph.add_node("analyze", analyze_node)
graph.add_node("report", report_node)
graph.add_edge(START, "search")
graph.add_conditional_edges("search", route, {
    "search_more": "search",
    "analyze": "analyze",
    "report": "report"
})
graph.add_edge("analyze", "report")
graph.add_edge("report", END)

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

Run the graph and then navigate its history:

config = {"configurable": {"thread_id": "research-tt-1"}}

result = app.invoke(
    {"messages": [("user", "Research AI agents in production")]},
    config
)

print("=== State History ===\n")
for i, state in enumerate(app.get_state_history(config)):
    step = state.values.get("steps_completed", 0)
    current = state.values.get("current_step", "initial")
    n_msgs = len(state.values.get("messages", []))
    sources = state.values.get("sources", [])
    next_nodes = state.next

    print(f"Checkpoint {i}:")
    print(f"  Steps: {step} | Current step: {current}")
    print(f"  Messages: {n_msgs} | Sources: {len(sources)}")
    print(f"  Next node: {next_nodes}")
    print(f"  Checkpoint ID: {state.config['configurable']['checkpoint_id'][:16]}...")
    print()

Typical output:

Checkpoint 0:                          ← Most recent (final report)
  Steps: 3 | Current step: completed
  Messages: 6 | Sources: 3
  Next node: ()

Checkpoint 1:                          ← After the analysis
  Steps: 3 | Current step: analyzed
  Messages: 5 | Sources: 3
  Next node: ('report',)

Checkpoint 2:                          ← After the 3rd search
  Steps: 3 | Current step: searching
  Messages: 4 | Sources: 3
  Next node: ('analyze',)

Checkpoint 3:                          ← After the 2nd search
  Steps: 2 | Current step: searching
  Messages: 3 | Sources: 2
  Next node: ('search',)

Checkpoint 4:                          ← After the 1st search
  Steps: 1 | Current step: searching
  Messages: 2 | Sources: 1
  Next node: ('search',)

Checkpoint 5:                          ← Initial state
  Steps: 0 | Current step: initial
  Messages: 1 | Sources: 0
  Next node: ('search',)

Each line is a point in time you can go back to. It's your agent's git log.

Inspecting a specific checkpoint

To examine in detail what's at a particular point of the run:

config = {"configurable": {"thread_id": "research-tt-1"}}

checkpoints = list(app.get_state_history(config))

target = checkpoints[3]  # After the 2nd search
print(f"=== Inspecting Checkpoint 3 ===\n")
print(f"State:")
print(f"  steps_completed: {target.values.get('steps_completed')}")
print(f"  current_step: {target.values.get('current_step')}")
print(f"  sources: {target.values.get('sources')}")

print(f"\nMessages:")
for j, msg in enumerate(target.values.get("messages", [])):
    role = msg.__class__.__name__
    content = msg.content[:80] if msg.content else "(empty)"
    print(f"  [{j}] {role}: {content}")

print(f"\nMetadata:")
print(f"  Next node: {target.next}")
print(f"  Checkpoint ID: {target.config['configurable']['checkpoint_id']}")
if target.parent_config:
    print(f"  Parent ID: {target.parent_config['configurable']['checkpoint_id']}")

The ability to inspect the exact content of every message, every variable, at every point of the run is what makes time-travel so powerful for debugging.

Replaying from a Checkpoint

Going back in time and re-running

The replay is time-travel's central operation: you take an earlier checkpoint's config and run invoke(None, config). LangGraph restores that checkpoint's exact state and continues execution from there.

config = {"configurable": {"thread_id": "research-tt-1"}}
checkpoints = list(app.get_state_history(config))

# Go back to the checkpoint after the 1st search (checkpoint 4)
target_config = checkpoints[4].config

print(f"State at the selected checkpoint:")
print(f"  Steps: {checkpoints[4].values.get('steps_completed')}")
print(f"  Sources: {checkpoints[4].values.get('sources')}")
print(f"  Next node: {checkpoints[4].next}")

print(f"\nRe-running from checkpoint 4...\n")
result = app.invoke(None, target_config)

print(f"Replay result:")
print(f"  Steps: {result['steps_completed']}")
print(f"  Sources: {result['sources']}")
print(f"  Current step: {result['current_step']}")
print(f"  Total messages: {len(result['messages'])}")

What happens internally:

1. LangGraph reads the checkpoint with the specified ID
2. It restores the state: 1 step completed, 1 source, next node = search
3. It runs search (step 2), then search (step 3), then analyze, then report
4. The agent completes the research from step 1, as if steps 2-5 had never happened

It's like git checkout abc123 — you go back to that point and everything that came after gets "forgotten" in this new line of execution.

Replay vs Resume

It's essential to distinguish between replay and resume:

config = {"configurable": {"thread_id": "research-tt-1"}}
checkpoints = list(app.get_state_history(config))

# Resume: continues from the last state (which already finished)
# If the graph already completed, there's nothing to do
state = app.get_state(config)
print(f"Resume — next node: {state.next}")  # () — already finished

# Replay: go back to an earlier point and re-run from there
replay_config = checkpoints[3].config  # After the 2nd search
state_replay = app.get_state(replay_config)
print(f"Replay — next node: {state_replay.next}")  # ('search',) — there's work to do

result = app.invoke(None, replay_config)
print(f"Replay complete: {result['steps_completed']} steps")

Resume says "continue where you left off." Replay says "go back to this point and do it again."

Modifying State and Re-running

update_state: the agent's "what if"

Replay re-runs from an earlier point with the same state. But time-travel's real power comes when you modify the state before re-running. That lets you ask "what if" questions:

• What would have happened if the agent had a different source?
• What result would it get with a different query?
• How would the decision change if the confidence score were higher?

update_state lets you inject changes into a checkpoint's state:

from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "research-tt-1"}}
checkpoints = list(app.get_state_history(config))

# Take the checkpoint after the 1st search
target_config = checkpoints[4].config

print("Original state:")
print(f"  Sources: {checkpoints[4].values.get('sources')}")
print(f"  Steps: {checkpoints[4].values.get('steps_completed')}")

# Modify: inject a different source and an extra step
app.update_state(
    target_config,
    values={
        "sources": ["arxiv.org/ai-agents-survey"],
        "steps_completed": 2,
        "messages": [AIMessage(content="I found a high-quality academic survey.")]
    },
    as_node="search"
)

# Check the modified state
modified = app.get_state(target_config)
print(f"\nModified state:")
print(f"  Sources: {modified.values.get('sources')}")
print(f"  Steps: {modified.values.get('steps_completed')}")

# Re-run from the modified state
result = app.invoke(None, target_config)
print(f"\nResult after re-running:")
print(f"  Final steps: {result['steps_completed']}")
print(f"  Final sources: {result['sources']}")

The as_node parameter

The as_node parameter in update_state matters: it tells LangGraph which node "wrote" this update. That affects which node runs next according to the graph's edges.

# If you say as_node="search", LangGraph evaluates the edges leaving "search"
# to decide the next node (in our case, route() decides whether to search more,
# analyze, or generate the report)
app.update_state(config, values={"steps_completed": 3}, as_node="search")

# If you say as_node="analyze", LangGraph follows the "analyze" → "report" edge
app.update_state(config, values={"current_step": "analyzed"}, as_node="analyze")

Choosing the right as_node is crucial. If you pick the wrong one, the graph can take an unexpected path after your modification.

A practical scenario: fixing a faulty search

Imagine your agent searched "AI agents" but should have searched "AI agents in production deployment." You can fix it without re-running everything:

config = {"configurable": {"thread_id": "research-fix-1"}}

result = app.invoke(
    {"messages": [("user", "Research AI agents in production")]},
    config
)

# Review the history and find where the search went wrong
checkpoints = list(app.get_state_history(config))

# Suppose checkpoint 4 (after the 1st search) has a generic source
print(f"Original source: {checkpoints[4].values.get('sources')}")

# Fix it: inject a more relevant search result
target_config = checkpoints[4].config
app.update_state(
    target_config,
    values={
        "sources": ["production-deployment-guide.dev"],
        "messages": [AIMessage(content="I found a specific production deployment guide.")],
    },
    as_node="search"
)

# Re-run: the agent continues with the corrected source
corrected_result = app.invoke(None, target_config)
print(f"Sources in the corrected result: {corrected_result['sources']}")

Practical Use Cases

Case 1: Debugging — finding where the agent failed

The most common case. The agent produced an incorrect result and you need to find the exact step where it went off track.

config = {"configurable": {"thread_id": "debug-session"}}

result = app.invoke(
    {"messages": [("user", "Research RAG pipelines")]},
    config
)

# The result isn't good — where did it go wrong?
print("=== Debugging: looking for the failure point ===\n")
for i, state in enumerate(app.get_state_history(config)):
    step = state.values.get("current_step", "initial")
    sources = state.values.get("sources", [])
    msgs = state.values.get("messages", [])

    last_msg = msgs[-1].content[:60] if msgs else "(empty)"
    print(f"[{i}] Step: {step} | Sources: {len(sources)} | Last msg: {last_msg}")

    # Look for anomalies
    if sources and any("irrelevant" in s for s in sources):
        print(f"  ⚠ Suspicious source found at checkpoint {i}")
    if state.values.get("steps_completed", 0) > 5:
        print(f"  ⚠ Too many steps — possible loop")

The pattern is: iterate the history, inspect each checkpoint, look for the point where the data went from "correct" to "incorrect."

Case 2: A/B testing — comparing the agent's decisions

Would the agent make a better decision with different information? Time-travel lets you A/B test reasoning:

config = {"configurable": {"thread_id": "ab-test-session"}}

app.invoke(
    {"messages": [("user", "Research vector databases")]},
    config
)

checkpoints = list(app.get_state_history(config))
# The checkpoint after the 1st search
first_search = checkpoints[4]

# Variant A: re-run with the original source
result_a = app.invoke(None, first_search.config)
print(f"Variant A — sources: {result_a['sources']}")

# Variant B: modify the source and inject an academic one
app.update_state(
    first_search.config,
    values={
        "sources": ["arxiv.org/vector-db-comparison-2025"],
        "messages": [AIMessage(content="An academic paper comparing Pinecone, Weaviate, and Chroma.")]
    },
    as_node="search"
)
result_b = app.invoke(None, first_search.config)
print(f"Variant B — sources: {result_b['sources']}")

Case 3: Recovery — undoing an agent action

If the agent executed an action it shouldn't have (wrote wrong data, sent the wrong message), you can go back to the previous state:

config = {"configurable": {"thread_id": "recovery-session"}}

app.invoke(
    {"messages": [("user", "Research and generate a report")]},
    config
)

# The generated report has errors. Go back to the pre-report state.
checkpoints = list(app.get_state_history(config))
for i, cp in enumerate(checkpoints):
    if cp.values.get("current_step") == "analyzed":
        pre_report = cp.config
        print(f"Found: checkpoint {i} (pre-report)")
        print(f"  Next node: {cp.next}")  # ('report',)
        break

# Re-run just the report (the report node runs with the analyzed state)
recovered = app.invoke(None, pre_report)
print(f"Report regenerated. Status: {recovered['current_step']}")

Human-in-the-loop with Time-travel

interrupt_before and interrupt_after

LangGraph lets you pause a graph's execution before or after a specific node. Combined with time-travel, this creates a flow where a human can:

1. Pause the agent before a critical action
2. Inspect the current state
3. Modify the state if necessary
4. Resume the execution

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from typing import TypedDict, Annotated

class ReviewState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    draft_report: str
    approved: bool
    steps_completed: int

def research(state: ReviewState) -> dict:
    return {
        "messages": [AIMessage(content="Research complete.")],
        "steps_completed": state.get("steps_completed", 0) + 1
    }

def generate_draft(state: ReviewState) -> dict:
    draft = "DRAFT: AI agents are transforming software development..."
    return {
        "messages": [AIMessage(content=f"Draft generated: {draft[:50]}...")],
        "draft_report": draft
    }

def publish_report(state: ReviewState) -> dict:
    return {
        "messages": [AIMessage(content="Report published.")],
        "approved": True
    }

graph = StateGraph(ReviewState)
graph.add_node("research", research)
graph.add_node("draft", generate_draft)
graph.add_node("publish", publish_report)
graph.add_edge(START, "research")
graph.add_edge("research", "draft")
graph.add_edge("draft", "publish")
graph.add_edge("publish", END)

memory = MemorySaver()

# interrupt_before stops BEFORE running "publish"
app_with_review = graph.compile(
    checkpointer=memory,
    interrupt_before=["publish"]
)

The complete flow: pause, inspect, modify, resume

config = {"configurable": {"thread_id": "review-session"}}

# Run it — it will stop before "publish"
result = app_with_review.invoke(
    {"messages": [("user", "Generate a report on AI agents")]},
    config
)

# Inspect where it stopped
state = app_with_review.get_state(config)
print(f"Next node: {state.next}")            # ('publish',)
print(f"Draft: {state.values.get('draft_report', '')[:60]}...")

# Option A: the human approves — resume with no changes
# result = app_with_review.invoke(None, config)

# Option B: the human wants to modify the draft
app_with_review.update_state(
    config,
    values={
        "draft_report": "REVISED DRAFT: AI agents are transforming software development. Data updated to 2026.",
        "messages": [AIMessage(content="Draft corrected by the human reviewer.")]
    },
    as_node="draft"
)

# Resume with the corrected draft
final = app_with_review.invoke(None, config)
print(f"\nReport published with draft: {final.get('draft_report', '')[:60]}...")
print(f"Approved: {final.get('approved')}")

interrupt_after: inspecting a node's results

interrupt_after stops the execution after a node finishes. Useful when you want to see what a node produced before the next one consumes it:

app_after = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_after=["draft"]
)

config = {"configurable": {"thread_id": "after-review"}}
result = app_after.invoke(
    {"messages": [("user", "Research and generate a draft")]},
    config
)

state = app_after.get_state(config)
print(f"Draft generated: {state.values.get('draft_report', '')[:80]}")
print(f"Next node: {state.next}")  # ('publish',)

# Decide whether to continue or modify before publishing
final = app_after.invoke(None, config)

The difference: interrupt_before=["publish"] stops before publishing (the draft is already ready). interrupt_after=["draft"] stops after generating the draft (the same point, different semantics). Choose based on which action you want to "protect."

Connection to the Project

Module 6 — Research Agent with Persistent Memory

In capsule 08 you'll integrate time-travel debugging into the Research Agent. The direct applications:

• Debugging research runs: When the Research Agent produces a low-quality report, you'll use get_state_history() to find which research step went off track — did it search the wrong term? Did it misread a result? Did it synthesize while ignoring key sources?
• Replay with different parameters: If the user says "the research didn't cover academic sources," you can replay from the search step with a modified query that prioritizes academic sources — without repeating the steps that already worked
• Human review before publishing: Using interrupt_before on the final synthesis node, the Research Agent will pause so the user can review the report's draft before generating it for real

The next capsules in this module

Troubleshooting

Problem 1: get_state_history returns an empty iterator

Symptom: You call get_state_history(config) and get no checkpoints. The loop doesn't iterate.

Cause: The thread_id doesn't exist in the checkpointer. It could be a typo, or the agent ran with a different thread_id, or the MemorySaver got re-initialized.

Solution: Verify the thread_id matches exactly the one you used when running the graph:

# Check whether the thread has checkpoints
config = {"configurable": {"thread_id": "my-thread"}}
history = list(app.get_state_history(config))
print(f"Checkpoints found: {len(history)}")

if not history:
    print("Thread not found. Check:")
    print("  1. Is it the same thread_id? (case-sensitive, hyphens vs underscores)")
    print("  2. Is it the same MemorySaver object?")
    print("  3. Did the process restart? (MemorySaver loses data)")

Problem 2: update_state doesn't change the expected next node

Symptom: After update_state, the graph runs a different node than you expected.

Cause: The as_node you used determines which edges get evaluated to decide the next node. If you use the wrong as_node, the routing function can take an unexpected path.

Solution: Check the resulting state and the next node before invoking:

app.update_state(config, values={"steps_completed": 3}, as_node="search")

state = app.get_state(config)
print(f"Next node after the update: {state.next}")

# If it isn't the node you expected, try a different as_node
# that has the right edge toward where you want to go

Problem 3: The replay produces different results than the original

Symptom: You re-run from a checkpoint and the result differs from the one you got originally, even though you didn't modify anything.

Cause: If your nodes call an LLM, the responses are non-deterministic (temperature > 0). If they call external APIs, the data may have changed. The checkpoint saves the state, not the LLM's future responses.

Solution: This is expected behavior. Replay guarantees that the starting state is identical — not that the future results will be the same. For exact reproducibility, use temperature=0 in your model or mock the tools in tests:

model = init_chat_model("openai:gpt-4.1-mini", temperature=0)

Problem 4: "Cannot update state of a completed graph"

Symptom: When you try update_state on a checkpoint where state.next is empty, you get an error or the update has no useful effect.

Cause: The checkpoint is the graph's final state — there's no next node to run. Updating the state makes no sense if you're not going to re-run.

Solution: Use a checkpoint that has a pending next node:

config = {"configurable": {"thread_id": "my-thread"}}
checkpoints = list(app.get_state_history(config))

# Find the first checkpoint that has a next node
for cp in checkpoints:
    if cp.next:
        print(f"Usable checkpoint: next={cp.next}")
        target_config = cp.config
        break
else:
    print("Every checkpoint is in a terminal state.")
    print("Use a checkpoint from before the end to run update_state.")

Problem 5: The history is very long and hard to navigate

Symptom: An agent that ran many steps has dozens of checkpoints. Finding the exact point is tedious.

Cause: Every node generates a checkpoint. A cyclic agent that iterates 10 times creates 10+ checkpoints.

Solution: Filter the history by relevant criteria instead of iterating everything:

config = {"configurable": {"thread_id": "long-session"}}

# Filter by a state field
def find_checkpoint(app, config, condition):
    for state in app.get_state_history(config):
        if condition(state.values):
            return state
    return None

# Find where steps_completed == 2
target = find_checkpoint(
    app, config,
    lambda v: v.get("steps_completed") == 2
)

if target:
    print(f"Found: {target.config['configurable']['checkpoint_id'][:16]}...")

Exercises

Exercise 1: Navigate and list the history (Easy)

Run the research graph with 3 searches from this capsule's setup. Use get_state_history to list every checkpoint. For each one, print: the number of steps completed, current_step, the number of sources, and whether it has a pending next node.

# Use the graph defined in the "Navigating the State History" section
# Run it with a thread_id, then iterate get_state_history
# Your code here...

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from typing import TypedDict, Annotated

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    sources: list[str]
    current_step: str
    steps_completed: int

def search_node(state: ResearchState) -> dict:
    step = state.get("steps_completed", 0) + 1
    return {
        "messages": [AIMessage(content=f"I searched source-{step}.example.com")],
        "sources": state.get("sources", []) + [f"source-{step}.example.com"],
        "current_step": "searching",
        "steps_completed": step
    }

def analyze_node(state: ResearchState) -> dict:
    return {
        "messages": [AIMessage(content="Analysis complete.")],
        "current_step": "analyzed"
    }

def report_node(state: ResearchState) -> dict:
    return {
        "messages": [AIMessage(content="Report generated.")],
        "current_step": "completed"
    }

def route(state: ResearchState) -> str:
    if state.get("steps_completed", 0) < 3:
        return "search_more"
    if state.get("current_step") != "analyzed":
        return "analyze"
    return "report"

graph = StateGraph(ResearchState)
graph.add_node("search", search_node)
graph.add_node("analyze", analyze_node)
graph.add_node("report", report_node)
graph.add_edge(START, "search")
graph.add_conditional_edges("search", route, {
    "search_more": "search",
    "analyze": "analyze",
    "report": "report"
})
graph.add_edge("analyze", "report")
graph.add_edge("report", END)

memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "exercise-1"}}

app.invoke({"messages": [("user", "Research AI agents")]}, config)

print("=== Complete History ===\n")
for i, state in enumerate(app.get_state_history(config)):
    steps = state.values.get("steps_completed", 0)
    current = state.values.get("current_step", "initial")
    sources = len(state.values.get("sources", []))
    has_next = "Yes" if state.next else "No"
    
    print(f"Checkpoint {i}: steps={steps}, current_step={current}, sources={sources}, pending={has_next}")

print("\n✓ Every checkpoint is a restore point in the agent's execution")

Exercise 2: Replay from an intermediate point (Easy)

Run the complete graph and then replay from the checkpoint where only 1 search is completed. Verify that the final result has the same 3 searches (the agent re-ran from that point) and that the status is "completed."

# 1. Run the complete graph
# 2. Find the checkpoint with steps_completed == 1
# 3. Do invoke(None, checkpoint_config)
# 4. Verify the final result is correct
# Your code here...

config = {"configurable": {"thread_id": "exercise-2"}}

original = app.invoke({"messages": [("user", "Research AI agents")]}, config)
print(f"Original: {original['steps_completed']} steps, status={original['current_step']}")

checkpoints = list(app.get_state_history(config))

target = None
for cp in checkpoints:
    if cp.values.get("steps_completed") == 1:
        target = cp
        break

if target:
    print(f"\nReplay from the checkpoint with 1 step (ID: {target.config['configurable']['checkpoint_id'][:16]}...)")
    print(f"  Sources at that point: {target.values.get('sources')}")
    print(f"  Next node: {target.next}")
    
    replayed = app.invoke(None, target.config)
    
    print(f"\nReplay result:")
    print(f"  Steps: {replayed['steps_completed']}")
    print(f"  Sources: {replayed['sources']}")
    print(f"  Status: {replayed['current_step']}")
    
    assert replayed["steps_completed"] == 3
    assert replayed["current_step"] == "completed"
    print("\n✓ The replay completed the research from step 1")

Exercise 3: Modify the state and re-run (Medium)

Run the complete graph. Then find the checkpoint after the 2nd search, use update_state to change steps_completed to 3 and add a custom source ("custom-paper.arxiv.org"). Re-run and verify that the agent jumps straight to analyze (because it already has 3 steps) and that the custom source appears in the result.

# 1. Run the graph
# 2. Find the checkpoint with steps_completed == 2
# 3. update_state to "skip" the 3rd search
# 4. Re-run and verify
# Your code here...

config = {"configurable": {"thread_id": "exercise-3"}}

app.invoke({"messages": [("user", "Research AI agents")]}, config)

checkpoints = list(app.get_state_history(config))
target = None
for cp in checkpoints:
    if cp.values.get("steps_completed") == 2:
        target = cp
        break

print(f"Original state (2 steps):")
print(f"  Sources: {target.values.get('sources')}")
print(f"  Next node: {target.next}")

app.update_state(
    target.config,
    values={
        "sources": target.values.get("sources", []) + ["custom-paper.arxiv.org"],
        "steps_completed": 3,
        "messages": [AIMessage(content="Custom paper injected via time-travel.")]
    },
    as_node="search"
)

modified = app.get_state(target.config)
print(f"\nModified state:")
print(f"  Steps: {modified.values.get('steps_completed')}")
print(f"  Sources: {modified.values.get('sources')}")
print(f"  Next node: {modified.next}")  # Should be ('analyze',)

result = app.invoke(None, target.config)
print(f"\nResult:")
print(f"  Status: {result['current_step']}")
print(f"  Sources: {result['sources']}")

assert "custom-paper.arxiv.org" in result["sources"]
assert result["current_step"] == "completed"
print("\n✓ The agent skipped the 3rd search and used the custom source")

Exercise 4: Human-in-the-loop with interrupt (Hard)

Build a graph with 3 nodes: research → draft → publish. Use interrupt_before=["publish"] to stop before publishing. Run the graph, inspect the draft, modify it with update_state, and then resume. Verify the published report contains your modification.

# 1. Define the graph with 3 nodes
# 2. Compile with interrupt_before=["publish"]
# 3. Run it — it stops before publish
# 4. Inspect the draft
# 5. Modify the draft with update_state
# 6. Resume with invoke(None, config)
# 7. Verify the modification in the result
# Your code here...

from dotenv import load_dotenv
load_dotenv()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from typing import TypedDict, Annotated

class PublishState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    draft: str
    published_report: str

def research(state: PublishState) -> dict:
    return {
        "messages": [AIMessage(content="Research complete.")],
    }

def draft(state: PublishState) -> dict:
    text = "AI agents let you automate complex tasks using LLMs."
    return {
        "messages": [AIMessage(content=f"Draft: {text}")],
        "draft": text
    }

def publish(state: PublishState) -> dict:
    final = f"PUBLISHED: {state.get('draft', '')}"
    return {
        "messages": [AIMessage(content=final)],
        "published_report": final
    }

graph = StateGraph(PublishState)
graph.add_node("research", research)
graph.add_node("draft", draft)
graph.add_node("publish", publish)
graph.add_edge(START, "research")
graph.add_edge("research", "draft")
graph.add_edge("draft", "publish")
graph.add_edge("publish", END)

memory = MemorySaver()
app_review = graph.compile(checkpointer=memory, interrupt_before=["publish"])
config = {"configurable": {"thread_id": "hitl-exercise"}}

# Run it — it stops before publish
app_review.invoke(
    {"messages": [("user", "Generate a report on AI agents")]},
    config
)

# Inspect
state = app_review.get_state(config)
print(f"Stopped before: {state.next}")
print(f"Current draft: {state.values.get('draft')}")

# Modify the draft
app_review.update_state(
    config,
    values={
        "draft": "AI agents let you automate complex tasks using LLMs. Key data point: 78% of companies plan to adopt agents in 2026.",
        "messages": [AIMessage(content="Draft modified by a human.")]
    },
    as_node="draft"
)

# Resume
final = app_review.invoke(None, config)
print(f"\nPublished report: {final.get('published_report', '')[:80]}...")

assert "78%" in final.get("published_report", "")
print("\n✓ The human modification made it into the published report")

Exercise 5: An automated debugging function (Hard)

Implement a debug_agent_run(app, config) function that: (1) gets the complete history, (2) for each pair of consecutive checkpoints identifies what changed (which state fields were modified), and (3) generates a "diff" report showing the transitions. Test it with a run of the research graph.

# Implement debug_agent_run(app, config) that returns a report
# of every transition between checkpoints
# Your code here...

def debug_agent_run(app, config) -> str:
    checkpoints = list(app.get_state_history(config))
    checkpoints.reverse()  # Oldest first
    
    report_lines = ["=== Debug Report ===", f"Thread: {config['configurable']['thread_id']}", f"Total checkpoints: {len(checkpoints)}", ""]
    
    for i in range(1, len(checkpoints)):
        prev = checkpoints[i - 1].values
        curr = checkpoints[i].values
        next_nodes = checkpoints[i].next
        
        report_lines.append(f"--- Transition {i-1} → {i} ---")
        report_lines.append(f"  Next node: {next_nodes}")
        
        all_keys = set(list(prev.keys()) + list(curr.keys()))
        changes = []
        
        for key in all_keys:
            if key == "messages":
                prev_count = len(prev.get("messages", []))
                curr_count = len(curr.get("messages", []))
                if prev_count != curr_count:
                    changes.append(f"  messages: {prev_count} → {curr_count} (+{curr_count - prev_count})")
                    new_msgs = curr.get("messages", [])[prev_count:]
                    for m in new_msgs:
                        content = m.content[:50] if m.content else "(empty)"
                        changes.append(f"    + [{m.__class__.__name__}] {content}")
            else:
                prev_val = prev.get(key)
                curr_val = curr.get(key)
                if prev_val != curr_val:
                    changes.append(f"  {key}: {prev_val} → {curr_val}")
        
        if changes:
            report_lines.extend(changes)
        else:
            report_lines.append("  (no changes)")
        report_lines.append("")
    
    report = "\n".join(report_lines)
    return report


config = {"configurable": {"thread_id": "debug-exercise"}}
app.invoke({"messages": [("user", "Research AI agents")]}, config)

report = debug_agent_run(app, config)
print(report)

=== Debug Report ===
Thread: debug-exercise
Total checkpoints: 6

--- Transition 0 → 1 ---
  Next node: ('search',)
  messages: 1 → 2 (+1)
    + [AIMessage] I searched source-1.example.com
  sources: None → ['source-1.example.com']
  current_step: None → searching
  steps_completed: None → 1

--- Transition 1 → 2 ---
  Next node: ('search',)
  messages: 2 → 3 (+1)
    + [AIMessage] I searched source-2.example.com
  sources: ['source-1.example.com'] → ['source-1.example.com', 'source-2.example.com']
  steps_completed: 1 → 2

...

Summary

In this capsule you learned:

• What time-travel debugging is: the ability to navigate an agent's complete state history, go back to any point, modify the state, and re-run. It's like git for agent execution — every checkpoint is a commit you can return to
• Navigating the history with get_state_history: getting every checkpoint of a thread in reverse chronological order, inspecting each one in detail (state, messages, metadata, next node), and finding specific points in the run
• Replaying from a checkpoint: using invoke(None, checkpoint_config) to return to an earlier state and re-run the graph from there. The agent doesn't repeat the steps before the checkpoint — it only runs what comes after
• Modifying state with update_state: injecting changes into a checkpoint (new sources, corrected values, extra messages) and re-running with those changes. The as_node parameter controls which edges get evaluated to determine the next step
• Three key use cases: debugging (finding where it failed), A/B testing (comparing decisions with different inputs), and recovery (undoing agent actions by returning to an earlier state)
• Human-in-the-loop with interrupt: combining interrupt_before/interrupt_after with update_state to create flows where a human can pause, inspect, modify, and resume an agent's execution

Next capsule: Conversation Management and Memory Patterns. How to handle long conversations that pile up too many messages — window trimming, summarization, and memory patterns that keep the context manageable without exploding in tokens.

Additional Resources

1. LangGraph Time Travel — Conceptual documentation on time-travel debugging, replay, and forking
2. LangGraph How-to: Time Travel — Step-by-step tutorial with replay and update_state examples
3. LangGraph How-to: Edit Graph State — A guide to update_state for human-in-the-loop
4. LangGraph Human-in-the-loop — Concepts of interrupt_before, interrupt_after, and approval flows
5. LangGraph Persistence Concepts — The checkpointing foundations that make time-travel possible