Persistence: store and checkpointer

The wrong first question for DeerFlow persistence is only “what gets saved?” In an agent system, persistence is closer to a fact boundary: after an agent spans turns, calls tools, writes files, emits events, and creates side effects, the system must know what can resume, what can be queried, and what can be audited.

This stop is not about memorizing table names. It is about drawing one boundary:

Persistence is not “saving chat history”. It is the recoverable, queryable, auditable fact layer for agent runs. The checkpointer owns resumable graph state; Store owns long-lived key-value state; RunStore, RunEventStore, and ThreadMetaStore own DeerFlow Gateway runtime records for queries, display, recovery, and audit.

flowchart TD
REQ["HTTP run request"] --> RM["RunManager / RunStore"]
REQ --> TM["ThreadMetaStore"]
RM --> WORKER["run_agent()"]
WORKER --> CP["LangGraph Checkpointer"]
WORKER --> STORE["LangGraph Store"]
WORKER --> JOURNAL["RunJournal"]
JOURNAL --> EV["RunEventStore"]
CP --> STATE["/state / history / wait"]
TM --> LIST["/threads/search"]
RM --> RUNS["/runs / token-usage"]
EV --> MSG["/messages / events"]

Startup Assembles Runtime Infrastructure

Persistence components are not created per request. During Gateway startup, langgraph_runtime() ( ⌖ deps.py::langgraph_runtime ) uses a startup snapshot of AppConfig to create long-lived objects:

make_stream_bridge(config)
init_engine_from_config(config.database)
make_checkpointer(config)
make_store(config)
RunRepository or MemoryRunStore
ThreadMetaRepository or MemoryThreadMetaStore
make_run_event_store(config.run_events)
RunManager(store=run_store)

These objects live on app.state. When a request arrives, the router does not create a new database pool or a new checkpointer. It reads the already-created runtime objects from app.state.

This explains an important behavior: DeerFlow supports some per-run configuration flexibility, but database, checkpointer, store, and run-event infrastructure are startup-bound. If you change database or run_events in config.yaml, you usually need to restart the Gateway before the backend uses the new setting.

Schema Bootstrap Happens At Startup

Persistence also includes how application tables are created and upgraded. DeerFlow does not only call Base.metadata.create_all() unconditionally. init_engine_from_config() enters bootstrap_schema() ( ⌖ bootstrap.py::bootstrap_schema ):

empty database
  create_all()
  alembic stamp head

legacy database: DeerFlow tables exist but alembic_version does not
  ensure baseline tables exist
  stamp 0001_baseline
  upgrade head

versioned database: alembic_version exists
  alembic upgrade head

This hybrid bootstrap chooses a practical split: an empty database uses SQLAlchemy metadata to create tables, avoiding a hand-written baseline that can drift; after the baseline, changes go through Alembic migrations. 0001_baseline ( ⌖ 0001_baseline.py ) mainly acts as the root of the Alembic chain and the stamp target for legacy databases. In the usual new-database path, DeerFlow does create_all + stamp head; it does not execute every baseline DDL statement one by one.

0002_runs_token_usage ( ⌖ 0002_runs_token_usage.py ) shows how post-baseline changes work: it adds token_usage_by_model to the runs table. It uses _helpers.safe_add_column() ( ⌖ _helpers.py::safe_add_column ) for idempotent column creation: if the column already exists, it is not added again; nullable, default, and type drift are still checked against the model definition.

Concurrency is handled in layers:

Postgres
  uses an advisory lock to serialize bootstrap.

SQLite
  uses an asyncio.Lock inside one process.
  across processes, it relies on SQLite file locking plus a 30-second busy_timeout.
  this is best-effort, not a distributed lock.

migrations
  column-level changes use idempotent helpers as the final retry/concurrency guard.

Startup bootstrap solves a concrete problem: an old database should not wait until the first request to fail with “missing column” or “missing table”. Schema bootstrap is a startup responsibility. It is related to the checkpointer/store boundary below, while remaining a separate concern.

Five Storage Surfaces

Put the names on the table first:

Checkpointer
  LangGraph checkpoint storage.

Store
  LangGraph long-lived KV storage, exposed as Runtime.store.

RunStore
  DeerFlow run metadata storage.

RunEventStore
  DeerFlow run event stream storage.

ThreadMetaStore
  DeerFlow thread list and ownership metadata storage.

Checkpointer: Graph State And Resume Points

The checkpointer comes from LangGraph. The async Gateway creates it with make_checkpointer() ( ⌖ async_provider.py::make_checkpointer ).

It stores thread checkpoints. The core pieces are:

channel_values
  messages
  title
  thread_data
  sandbox
  other ThreadState fields

metadata
  step
  source
  writes
  parents
  created_at / updated_at

tasks / pending_writes
  interrupts, errors, next tasks, and other execution details

Its main consumers are:

/threads/{thread_id}/state
/threads/{thread_id}/history
/runs/wait final channel_values
cancel(action=rollback)
run_agent finally reads title and syncs it into thread_meta

So the checkpointer is what decides whether the next turn can continue the conversation. Whether message history, sandbox_id, and thread_data can be restored depends on whether the checkpointer is persistent.

Store: LangGraph Long-Lived KV

Store is created by make_store() ( ⌖ async_provider.py::make_store ) and is passed into:

Runtime(context=runtime_ctx, store=store)
agent.store = store

It is not the checkpoint store. It does not save every graph step. It is the long-lived key-value storage capability that LangGraph exposes to the agent runtime.

DeerFlow also has a compatibility use: when there is no SQL session factory, MemoryThreadMetaStore stores thread metadata under the Store namespace ("threads",). So in memory mode, the thread list may borrow Store; in sqlite or postgres mode, thread listing mainly uses the threads_meta table.

RunStore: The External Lifecycle Of A Run

The SQL implementation of RunStore is RunRepository ( ⌖ sql.py::RunRepository ), used by RunManager.

It stores the external run record:

run_id
thread_id
assistant_id
user_id
status
model_name
multitask_strategy
metadata / kwargs
error
created_at / updated_at
token usage
token_usage_by_model
message_count
first_human_message
last_ai_message

These fields serve:

list runs for a thread
query current run status
check status before cancel
aggregate token usage
recognize orphaned inflight runs after restart

RunStore does not save the full conversation and does not restore LangGraph state.

token_usage_by_model on the runs table ( ⌖ model.py::RunRow ) fixes the accounting shape for multi-model runs. One run may include lead-agent model calls, subagent calls, and middleware calls. Those cannot be accurately grouped by the run’s single model_name. token_usage_by_model stores input, output, and total tokens for each model used.

RunEventStore: Messages And Execution Events

RunEventStore is the event stream interface ( ⌖ base.py::RunEventStore ). The SQL implementation is DbRunEventStore ( ⌖ db.py::DbRunEventStore ).

One event roughly contains:

thread_id
run_id
event_type
category
content
metadata
seq
created_at

category matters:

message
  user-visible messages.

trace / middleware / outputs / error
  debugging, audit, and lifecycle records.

seq is an increasing sequence within the same thread, used for pagination and ordering. The SQL backend enforces uniqueness for (thread_id, seq).

RunEventStore consumers include:

/threads/{thread_id}/messages
/threads/{thread_id}/runs/{run_id}/messages
/threads/{thread_id}/runs/{run_id}/events

It is good for asking “what happened?” It is not the authority for graph state. Restoring graph state still goes through the checkpointer.

ThreadMetaStore: Thread List Metadata

The SQL implementation is ThreadMetaRepository ( ⌖ sql.py::ThreadMetaRepository ).

It stores:

thread_id
assistant_id
user_id
display_name
status
metadata
created_at / updated_at

This is not the conversation body and not a checkpoint. It is the information needed for thread lists, ownership checks, and search.

For example, /threads/search does not scan every checkpoint. It queries ThreadMetaStore. When a run finishes, the worker reads title from the latest checkpoint and syncs it into threads_meta.display_name, so the list page can show a title quickly.

How One Run Writes Them

start_run() ( ⌖ services.py::start_run ) and run_agent() ( ⌖ worker.py::run_agent ) connect the main path:

1. start_run()
   RunManager.create_or_reject()
   -> create a RunRecord
   -> write a pending row to RunStore

2. start_run()
   upsert thread_meta
   -> create if missing
   -> set status=running if present

3. run_agent()
   RunManager.set_status(running)
   -> update RunStore

4. run_agent()
   capture the pre-run checkpoint
   -> used later for rollback

5. agent.astream(...)
   LangGraph executes
   -> checkpointer keeps writing checkpoints
   -> StreamBridge pushes SSE

6. RunJournal callback
   -> converts LLM / tool / middleware callbacks into run events
   -> aggregates tokens and message summary

7. run_agent() ends
   -> success / error / interrupted
   -> update RunStore status

8. finally
   -> journal.flush()
   -> update_run_completion()
   -> sync checkpoint.title into thread_meta.display_name
   -> set thread_meta.status to idle/error/interrupted

9. bridge.publish_end()
   -> end SSE
   -> bridge cleanup

That path gives three different meanings to “messages”:

SSE
  online events visible to the current connection.

RunEventStore
  event stream that can be queried later.

Checkpointer
  graph state needed by the next turn.

They can all contain message-shaped data, but their contracts are different.

RunJournal

RunJournal ( ⌖ journal.py::RunJournal ) is a LangChain callback handler. It does not run the agent and does not decide state transitions. It observes LLM, tool, and chain callbacks, then writes events into RunEventStore.

It also builds two useful aggregates:

event stream
  llm.human.input
  llm.ai.response
  llm.tool.result
  middleware:...
  run.start / run.end / run.error

run summary
  total_input_tokens
  total_output_tokens
  total_tokens
  llm_call_count
  lead_agent_tokens
  subagent_tokens
  middleware_tokens
  token_usage_by_model
  message_count
  first_human_message
  last_ai_message

Why write the summary back to RunStore? Because list pages and token statistics should not scan the full event stream every time. RunEventStore keeps the details; RunStore keeps the common summary fields. That is a read-path optimization.

The accounting is grouped by the model that produced each LLM response. RunJournal accumulates total tokens, extracts the model name from each response’s response_metadata.model_name or model field, and updates a per-model accumulator. Subagents and middleware contribute extra token records back to the parent run. At completion, all of this lands in RunStore.update_run_completion(). Therefore /threads/{thread_id}/token-usage by_model means “how many tokens did each model produce under this thread”; it is broader than “how many tokens did the lead model use”.

RunRepository.aggregate_tokens_by_thread() ( ⌖ sql.py::RunRepository.aggregate_tokens_by_thread ) prefers each row’s token_usage_by_model JSON. If older rows do not have that content, it falls back to the legacy model_name + total_tokens shape. Note that by_model[model].runs is not a mutually exclusive count: if one run used multiple models, that run can count under multiple models.

LangGraph State And DeerFlow Runtime State

To answer “what recovers after restart?”, first separate two kinds of state.

LangGraph state enters the checkpointer as graph channel values:

messages
title
thread_data
sandbox
artifacts
promoted tools
other ThreadState fields

Those fields serve:

the next conversation turn
state/history APIs
rollback
tool runtime context

DeerFlow Gateway runtime state is managed separately:

RunRecord
  asyncio.Task, abort_event, and status in the current process.

RunRow
  queryable run metadata and token summary.

RunEventRow
  messages, execution events, and audit records.

ThreadMetaRow
  thread list, title, owner, and status.

StreamBridge buffer
  current-run SSE subscribers and short-lived buffering.

app.state singletons
  checkpointer, store, event_store, run_manager, and other startup objects.

These are for cancellation, queries, display, authorization, and online transport. They should not be mixed into ThreadState.

What Recovers After Restart

Recovery depends on the backend for each surface:

persistent checkpointer
  can restore thread state, history, messages, title, thread_data, sandbox_id, and other graph state.

persistent run_store
  can restore run lists, run status, and token summary.

persistent run_event_store
  db/jsonl can restore messages and events.
  memory loses them.

persistent thread_meta
  SQL backend can restore thread lists and owners.
  MemoryThreadMetaStore can recover only if the LangGraph Store behind it is persistent.

StreamBridge
  in-process transport layer.
  cannot recover after restart.

asyncio.Task / abort_event
  in-process objects.
  cannot recover after restart.

So a persisted run row does not mean the run is still executing. The Python task that was executing it disappeared with the process.

RunManager.reconcile_orphaned_inflight_runs() ( ⌖ manager.py::RunManager.reconcile_orphaned_inflight_runs ) handles exactly that:

database still has pending/running runs
but the current process has no matching task
=> mark them as error

The goal is not to resume execution. It is to turn an uncertain state into a clear failure so the UI does not display running forever.

Cancel And Rollback

RunManager.cancel() can only cancel a task in the current process:

record.abort_event.set()
record.task.cancel()
record.status = interrupted

That makes cancellation a process-local capability. Another process may see the run row in RunStore, but it cannot directly cancel this process’s asyncio.Task.

Rollback is about graph state. Before a run starts, run_agent() captures a pre-run checkpoint snapshot. If the user chooses action=rollback, the worker restores the checkpoint to the state before the run began.

That is not a database transaction rollback. It writes a new recovery point into the checkpointer:

pre-run checkpoint
  -> copy content
  -> assign a new checkpoint id / ts
  -> write it back to the checkpointer

So rollback reverts LangGraph state. It does not undo every external side effect. For example, files written in the sandbox or side effects in external services need separate handling.

API Serialization

Checkpoints store LangChain and LangGraph objects. The frontend cannot receive those objects directly. runtime/serialization.py ( ⌖ serialization.py::serialize_channel_values_for_api ) converts them into JSON-ready data.

It also protects one practical API boundary. ViewImageMiddleware may place base64 image blocks inside hidden messages for model use. When REST state/history is returned, serialize_channel_values_for_api() removes data: image blocks from messages marked hide_from_ui, avoiding large internal context from leaking into frontend responses.

Serialization is therefore not only format conversion. It also enforces part of the API boundary.

Keep Schema Migration Separate From Runtime Store Consistency

In the current code, application-table creation and upgrade already have an explicit path: persistence.bootstrap, the Alembic baseline, 0002_runs_token_usage, and the idempotent column helper are all on the Gateway startup path.

Schema bootstrap answers “how are tables created and upgraded?” The split-brain risk below answers a different question: do the checkpointer, store, run store, run event store, and thread metadata all consume the same persistence config? Keep those problems separate.

Remaining Split-Brain Risk

The code already has a unified config shape:

database:
  backend: sqlite | postgres | memory

DatabaseConfig ( ⌖ database_config.py::DatabaseConfig ) intends one backend to govern both the checkpointer and DeerFlow application data.

The startup path is not fully unified yet:

make_checkpointer(config)
  supports legacy checkpointer
  also supports the new database config

init_engine_from_config(config.database)
  supports the new database config

make_store(config)
  only reads legacy checkpointer
  falls back to InMemoryStore when checkpointer is absent

That creates a concrete risk:

only database.sqlite is configured, no legacy checkpointer

checkpointer
  may be SQLite-persistent.

RunRepository / ThreadMetaRepository
  are SQL-persistent.

LangGraph Store
  may still be InMemoryStore.

This is a split-brain shape: in the same Gateway, different state surfaces choose persistence backend from different config sources. Some state can recover and some cannot, while the user may assume “I configured database.sqlite, so persistence is on”.

This is not necessarily fatal for conversation recovery if the app mostly relies on checkpoint state. But it is semantically inconsistent for LangGraph Store, memory, future extensions, and embedded/sync paths.

Final Rule

Do not describe persistence as one big promise like “turn on the database and everything recovers”. Ask per object:

Can conversation state recover?
  check the checkpointer.

Can the thread list recover?
  check ThreadMetaStore.

Can run lists and token statistics recover?
  check RunStore.

Can messages and audit events recover?
  check RunEventStore.

Can an executing task recover?
  no. asyncio.Task is process-local.

Can an SSE connection recover?
  no. StreamBridge is an online transport layer.

That is the most important mental model of this chapter: DeerFlow persistence is a set of layered contracts, not a single switch. For every layer, ask what it stores, who consumes it, and how much recovery it promises after restart.