backend/checkpointer.py

"""LangGraph checkpointer bootstrap for persistent thread state.

Resolution order:
1. `LANGGRAPH_CHECKPOINT_POSTGRES_URI` -> Postgres checkpointer
2. `LANGGRAPH_CHECKPOINT_SQLITE_PATH`  -> SQLite checkpointer
3. Fallback to `${DEER_FLOW_HOME}/checkpoints.sqlite`
"""

import atexit
import os
from pathlib import Path

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.sqlite import SqliteSaver

_ctx = None


def _default_sqlite_path() -> str:
    deer_flow_home = os.getenv("DEER_FLOW_HOME")
    if deer_flow_home:
        base = Path(deer_flow_home)
    else:
        base = Path.cwd() / ".deer-flow"
    base.mkdir(parents=True, exist_ok=True)
    return str((base / "checkpoints.sqlite").resolve())


def _init_checkpointer():
    global _ctx

    postgres_uri = os.getenv("LANGGRAPH_CHECKPOINT_POSTGRES_URI", "").strip()
    if postgres_uri:
        _ctx = PostgresSaver.from_conn_string(postgres_uri)
        saver = _ctx.__enter__()
        saver.setup()
        return saver

    sqlite_path = os.getenv("LANGGRAPH_CHECKPOINT_SQLITE_PATH", "").strip() or _default_sqlite_path()
    sqlite_file = Path(sqlite_path).expanduser().resolve()
    sqlite_file.parent.mkdir(parents=True, exist_ok=True)
    _ctx = SqliteSaver.from_conn_string(str(sqlite_file))
    saver = _ctx.__enter__()
    saver.setup()
    return saver


def _close_context() -> None:
    global _ctx
    if _ctx is not None:
        try:
            _ctx.__exit__(None, None, None)
        finally:
            _ctx = None


checkpointer = _init_checkpointer()
atexit.register(_close_context)