How to Debug LangGraph Workflow State Machines and Checkpoints in Open Notebook

Open Notebook uses LangGraph with SQLite checkpoint persistence, allowing developers to inspect state transitions in data/sqlite-db/checkpoints.sqlite, replay workflows from any saved checkpoint, and unit-test individual nodes to isolate failures.

The Open Notebook repository orchestrates asynchronous AI processing through LangGraph state machines that power content ingestion, chat interactions, and document transformations. When these workflows fail or produce unexpected results, the built-in checkpoint system provides full visibility into every state transition. Understanding how to query and manipulate these checkpoints transforms debugging from guesswork into a deterministic process.

Understanding the Checkpoint Architecture

Open Notebook implements LangGraph workflows using StateGraph instances compiled with persistent storage. Each graph node writes its complete state to a SQLite database after execution, creating a recoverable trail of the entire workflow.

State Definitions

The data flowing through the graph is strictly typed using TypedDict definitions. In open_notebook/graphs/chat.py, the ThreadState defines the schema for chat workflows, while open_notebook/graphs/source_chat.py contains SourceChatState for source-specific interactions. These state objects carry messages, context, notebook references, and model configurations between nodes.

The SqliteSaver Implementation

Checkpoint persistence relies on SqliteSaver from LangGraph's SQLite integration. The implementation appears in both chat and source-chat graphs:

# open_notebook/graphs/chat.py

from langgraph.checkpoint.sqlite import SqliteSaver
from open_notebook.config import LANGGRAPH_CHECKPOINT_FILE
import sqlite3

conn = sqlite3.connect(
    LANGGRAPH_CHECKPOINT_FILE,
    check_same_thread=False,
)
memory = SqliteSaver(conn)  # Checkpoint store instance

# ... graph definition ...

graph = agent_state.compile(checkpointer=memory)  # Attaches persistence

The memory object intercepts every state transition and serializes the full TypedDict to the database.

Configuration and File Location

The checkpoint database location is centralized in open_notebook/config.py:

LANGGRAPH_CHECKPOINT_FILE = f"{sqlite_folder}/checkpoints.sqlite"

By default, sqlite_folder resolves to data/sqlite-db within the project directory. Each node automatically persists its output to this file, creating a row for every step in the workflow.

Inspecting Checkpoint Data

The checkpoint database is a standard SQLite file queryable through any database tool or Python script.

Querying Checkpoints via SQL

The checkpoints table contains checkpoint_id, state, metadata, and created_at columns. Use JSON extraction functions to inspect specific state fields:

sqlite3 data/sqlite-db/checkpoints.sqlite \
    "SELECT checkpoint_id, 
            json_extract(state, '$.messages') AS msgs,
            json_extract(state, '$.source') AS src,
            created_at 
     FROM checkpoints 
     ORDER BY created_at DESC 
     LIMIT 10;"

Inspecting Checkpoints from Python

For programmatic analysis, connect directly to the database:

import json
import sqlite3

db_path = "data/sqlite-db/checkpoints.sqlite"
conn = sqlite3.connect(db_path)
cur = conn.cursor()
cur.execute(
    "SELECT checkpoint_id, state FROM checkpoints "
    "ORDER BY created_at DESC LIMIT 5"
)
for chk_id, state_json in cur.fetchall():
    state = json.loads(state_json)
    print(f"--- {chk_id} ---")
    print("Messages:", state.get("messages"))
    print("Source:", state.get("source"))

Replaying and Resetting Workflows

Checkpoints enable time-travel debugging by allowing you to resume execution from any historical state.

Replaying from a Specific Checkpoint

To restart a workflow from a saved checkpoint, retrieve the state using the checkpoint ID and invoke the graph:

from open_notebook.graphs.source_chat import source_chat_graph, memory

# Load saved state by checkpoint ID

saved_state = await memory.get("checkpoint-id-from-query")

# Resume execution from that point

result = await source_chat_graph.ainvoke(saved_state)
print(result)

This technique is invaluable when fixing a failing node—you can replay the exact same input state without rebuilding the entire conversation history.

Clearing Stale Checkpoints

During iterative development, stale checkpoints may contain obsolete schema versions or corrupted states. Remove the database file to start fresh:

import os

os.remove("data/sqlite-db/checkpoints.sqlite")

The next graph compilation automatically recreates the file with an empty schema.

Unit Testing Individual Nodes

Isolate specific failures by testing graph nodes outside the workflow context. This bypasses complex state dependencies and provides immediate feedback.

import pytest
from open_notebook.graphs.chat import call_model_with_messages

@pytest.mark.asyncio
async def test_call_model_with_messages():
    state = {
        "messages": [],
        "notebook": None,
        "context": None,
        "model_override": "gpt-4o-mini",
    }
    config = {"configurable": {"model_id": "gpt-4o-mini"}}
    out = await call_model_with_messages(state, config)
    assert "messages" in out
    assert out["messages"].content

Similar unit tests can target call_model_with_source_context from open_notebook/graphs/source_chat.py or transformation nodes in open_notebook/graphs/source.py.

Debugging Workflow Failures

When a LangGraph workflow fails, follow this systematic approach using the checkpoint system:

1. Enable verbose logging by setting LOG_LEVEL=DEBUG or calling logger.enable("open_notebook") to capture execution traces.

2. Identify the failing checkpoint by querying the database for the most recent entries before the error timestamp. Look for missing keys, empty strings, or malformed data in the state column.

3. Check exception metadata—LangGraph automatically records exceptions in the checkpoint metadata column when nodes raise errors. Query this field to retrieve the original traceback without reproducing the failure.

4. Replay the checkpoint locally using the memory.get() pattern to confirm the failure is reproducible with the exact state.

5. Fix and iterate by updating the node logic and replaying from the same checkpoint until the state transitions correctly.

Summary

• Open Notebook persists all LangGraph workflow states to data/sqlite-db/checkpoints.sqlite using SqliteSaver attached during graph compilation.
• Query the checkpoint database directly with SQL or Python to inspect ThreadState and SourceChatState values at any execution step.
• Replay workflows from specific checkpoint IDs using memory.get() and graph.ainvoke() to debug failures without rebuilding context.
• Clear checkpoints by deleting the SQLite file when schema changes or state corruption occurs during development.
• Unit test nodes in isolation by calling async functions like call_model_with_messages with crafted state dictionaries.

Frequently Asked Questions

How do I find which checkpoint corresponds to a specific workflow error?

Query the metadata column in the checkpoints table for entries containing exception information. LangGraph automatically writes exception details to the checkpoint metadata when a node fails, allowing you to correlate timestamps and error messages with specific checkpoint IDs.

Can I share checkpoint databases between different Open Notebook instances?

Yes, since LANGGRAPH_CHECKPOINT_FILE in open_notebook/config.py defines a central path, multiple instances can point to the same SQLite file. Ensure file permissions allow concurrent read/write access, or use a shared volume when running in containerized environments.

What is the difference between chat.py and source_chat.py checkpoints?

Both use the same underlying SqliteSaver and database file, but they operate on different state schemas. open_notebook/graphs/chat.py uses ThreadState for general notebook conversations, while open_notebook/graphs/source_chat.py uses SourceChatState which includes source-specific context. The checkpoint table stores both types, distinguished by the JSON structure in the state column.

How do I debug a node that silently produces incorrect output?

Enable debug logging to confirm the node execution, then query the checkpoint database for the node's output state. Compare the actual state against expected values using the Python inspection pattern. If the state is correct but the application behaves incorrectly, the issue likely lies in downstream nodes or state interpretation rather than the checkpointed node itself.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}