> ## Documentation Index
> Fetch the complete documentation index at: https://notes.kodekloud.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Demo Time Travel with State Injection
> Demonstrating workflow checkpointing and state injection to rewind execution, modify checkpoints, and resume for faster debugging, testing, and exploring alternate outcomes
In this lesson we show a practical debugging pattern for LangGraph-style workflows: time travel via state injection. Using simple checkpointing, you can rewind a workflow to an earlier step, modify the saved state, and resume execution from that point. This enables fast experimentation, targeted debugging, and verification of alternate decision paths without re-running the entire pipeline.
Why this is useful
* Debug agent reasoning and decision logic by exploring alternate histories.
* Reproduce and test edge cases by changing specific values at a chosen checkpoint.
* Save time when only a later-stage decision needs to be inspected or changed.
Keywords: time travel, state injection, checkpointing, LangGraph, workflow state, debugging
## Overview
We’ll build a minimal example in Python that demonstrates:
1. Creating and storing checkpoints as deep copies of workflow state.
2. Rewinding to a specific checkpoint.
3. Injecting a modified value into that checkpointed state.
4. Resuming execution from the rewound checkpoint to observe a different outcome.
Code for the demo is organized into discrete blocks so you can run each section independently.
```python theme={null}
# Imports used throughout the examples
from typing import TypedDict, List, Dict
import copy
from pprint import pprint
```
Define a compact `TypedDict` to represent the workflow state used in the demo:
```python theme={null}
class WorkflowState(TypedDict):
input_text: str
confidence: float
decision: str
status: str
```
Next, define the workflow nodes. This tiny workflow has two phases:
1. Evaluate (marks the state as evaluated)
2. Route (chooses a path based on confidence)
```python theme={null}
def evaluate_node(state: WorkflowState) -> WorkflowState:
return {
**state,
"status": "evaluated"
}
def route_node(state: WorkflowState) -> WorkflowState:
if state["confidence"] >= 0.7:
return {
**state,
"decision": "main_path",
"status": "completed"
}
return {
**state,
"decision": "fallback_path",
"status": "completed"
}
```
Build a tiny checkpoint engine to simulate LangGraph's checkpointing. Each checkpoint stores a deep copy of the state in a history list so it can be inspected or restored later.
```python theme={null}
def save_checkpoint(history: List[Dict], step_name: str, state: WorkflowState):
history.append({
"step": step_name,
"state": copy.deepcopy(state)
})
def run_workflow(initial_state: WorkflowState):
history: List[Dict] = []
state = copy.deepcopy(initial_state)
save_checkpoint(history, "start", state)
state = evaluate_node(state)
save_checkpoint(history, "after_evaluate", state)
state = route_node(state)
save_checkpoint(history, "after_route", state)
return state, history
```
Run the workflow once with a low-confidence input so the route chooses the fallback path. We print both the final state and the saved checkpoints for inspection.
```python theme={null}
initial_state: WorkflowState = {
"input_text": "Classify this request and decide whether to continue automatically.",
"confidence": 0.4,
"decision": "",
"status": "new"
}
final_state, history = run_workflow(initial_state)
print("Final state from first run:")
pprint(final_state)
print("\nCheckpoints:")
for i, checkpoint in enumerate(history):
print(f"Checkpoint {i}: {checkpoint['step']}")
pprint(checkpoint["state"])
print("-" * 60)
```
```plaintext theme={null}
Final state from first run:
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.4,
'decision': 'fallback_path',
'status': 'completed'}
Checkpoints:
Checkpoint 0: start
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.4,
'decision': '',
'status': 'new'}
------------------------------------------------------------
Checkpoint 1: after_evaluate
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.4,
'decision': '',
'status': 'evaluated'}
------------------------------------------------------------
Checkpoint 2: after_route
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.4,
'decision': 'fallback_path',
'status': 'completed'}
------------------------------------------------------------
```
Each saved checkpoint represents a snapshot of execution at a particular step. Time travel means selecting one of those snapshots, changing the state, and continuing from there to observe how the workflow would behave under the modified conditions.
Time travel is useful for debugging and experimentation: instead of rerunning the entire pipeline, you can jump to the step you care about, change a variable, and see what would have happened.
## Inspecting checkpoints
Here’s a concise table that summarizes the checkpoints created in the example:
| Index | Step name | Description |
| ----- | ---------------- | ------------------------------------------------------------------ |
| 0 | `start` | Initial input state before any processing. |
| 1 | `after_evaluate` | State after the evaluate node marks the request as `evaluated`. |
| 2 | `after_route` | Final state after routing (decision chosen based on `confidence`). |
## Rewind, inject, and resume
Now rewind to the checkpoint saved immediately after evaluation (`after_evaluate`), inject a higher confidence value, and resume routing from that checkpoint so the workflow follows the alternate path.
```python theme={null}
# Choose the checkpoint to rewind to (after_evaluate)
rewind_checkpoint = next(c for c in history if c["step"] == "after_evaluate")
rewind_state: WorkflowState = copy.deepcopy(rewind_checkpoint["state"])
# Inject a modified value
rewind_state["confidence"] = 0.9
print("Injected state (after rewind):")
pprint(rewind_state)
# Continue execution from the rewound checkpoint
continued_state = route_node(rewind_state)
print("\nFinal state after time travel and injection:")
pprint(continued_state)
```
```plaintext theme={null}
Injected state (after rewind):
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.9,
'decision': '',
'status': 'evaluated'}
Final state after time travel and injection:
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.9,
'decision': 'main_path',
'status': 'completed'}
```
Compare the original run outcome with the time-travel outcome:
```python theme={null}
print("Original run outcome:")
pprint(final_state)
print("\nAfter rewind + injection:")
pprint(continued_state)
```
```plaintext theme={null}
Original run outcome:
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.4,
'decision': 'fallback_path',
'status': 'completed'}
After rewind + injection:
{'input_text': 'Classify this request and decide whether to continue automatically.',
'confidence': 0.9,
'decision': 'main_path',
'status': 'completed'}
```
Because we injected a higher confidence before routing, the resumed execution followed the `main_path` instead of the `fallback_path`. This demonstrates how state injection lets you explore alternate outcomes without re-running the entire workflow.
When modifying checkpoints, always work on deep copies (as shown) to avoid mutating historical records. Preserving immutable checkpoints ensures reproducibility and accurate auditing of prior runs.
## When to use time travel with state injection
* Reproduce a bug that occurs only with a specific intermediate value.
* Test how downstream logic responds to different upstream outputs.
* Perform A/B style experiments by changing one variable at a checkpoint and comparing outcomes.
## References and further reading
* Python `TypedDict` and typing: [https://docs.python.org/3/library/typing.html#typing.TypedDict](https://docs.python.org/3/library/typing.html#typing.TypedDict)
* Checkpointing concepts (computing): [https://en.wikipedia.org/wiki/Checkpoint\_(computing)](https://en.wikipedia.org/wiki/Checkpoint_\(computing\))
* LangGraph concepts and docs (search for LangGraph checkpoints and debugging in your project docs)
Key takeaway: State injection (time travel) lets you rewind a workflow to a chosen checkpoint, modify the saved state, and continue execution from there—making debugging, testing, and scenario exploration faster and more targeted.