diff --git a/README.md b/README.md index 5b7fffaad..bb66faf08 100644 --- a/README.md +++ b/README.md @@ -253,6 +253,11 @@ make down # Stop and remove containers Access: http://localhost:2026 +For persistent deployments, configure `database.backend` as `sqlite` or +`postgres`. The selected backend is shared by the LangGraph checkpointer, +LangGraph Store, and DeerFlow application data. The deprecated `checkpointer` +section, when present, overrides the first two for backward compatibility. + The unified nginx endpoint is same-origin by default and does not emit browser CORS headers. If you run a split-origin or port-forwarded browser client, set `GATEWAY_CORS_ORIGINS` to comma-separated exact origins such as `http://localhost:3000`; the Gateway then applies the CORS allowlist and matching CSRF origin checks. > [!IMPORTANT] diff --git a/backend/AGENTS.md b/backend/AGENTS.md index b7dfc40f2..63b39c1cc 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -250,6 +250,12 @@ Setup: Copy `config.example.yaml` to `config.yaml` in the **project root** direc Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig`, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `database`, `checkpointer`, `run_events`, `stream_bridge`, `sandbox`, `log_level`, `channels`, `channel_connections`. Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. +**Persistence backend resolution**: the unified `database` section selects the +Gateway's LangGraph checkpointer, LangGraph Store, and DeerFlow SQL repositories. +The deprecated `checkpointer` section remains backward compatible and, when +present, overrides `database` for the LangGraph checkpointer and Store only; +application repositories continue to use `database`. + Configuration priority: 1. Explicit `config_path` argument 2. `DEER_FLOW_CONFIG_PATH` environment variable diff --git a/backend/packages/harness/deerflow/runtime/store/async_provider.py b/backend/packages/harness/deerflow/runtime/store/async_provider.py index bc1f07eba..035a1a9c4 100644 --- a/backend/packages/harness/deerflow/runtime/store/async_provider.py +++ b/backend/packages/harness/deerflow/runtime/store/async_provider.py @@ -1,11 +1,11 @@ -"""Async Store factory — backend mirrors the configured checkpointer. +"""Async Store factory — backend mirrors runtime persistence configuration. -The store and checkpointer share the same ``checkpointer`` section in -*config.yaml* so they always use the same persistence backend: +The deprecated ``checkpointer`` section takes precedence when present; +otherwise Store follows the unified ``database`` section in *config.yaml*: -- ``type: memory`` → :class:`langgraph.store.memory.InMemoryStore` -- ``type: sqlite`` → :class:`langgraph.store.sqlite.aio.AsyncSqliteStore` -- ``type: postgres`` → :class:`langgraph.store.postgres.aio.AsyncPostgresStore` +- ``memory`` → :class:`langgraph.store.memory.InMemoryStore` +- ``sqlite`` → :class:`langgraph.store.sqlite.aio.AsyncSqliteStore` +- ``postgres`` → :class:`langgraph.store.postgres.aio.AsyncPostgresStore` Usage (e.g. FastAPI lifespan):: @@ -17,6 +17,7 @@ Usage (e.g. FastAPI lifespan):: from __future__ import annotations +import asyncio import contextlib import logging from collections.abc import AsyncIterator @@ -24,7 +25,14 @@ from collections.abc import AsyncIterator from langgraph.store.base import BaseStore from deerflow.config.app_config import AppConfig, get_app_config -from deerflow.runtime.store.provider import POSTGRES_CONN_REQUIRED, POSTGRES_STORE_INSTALL, SQLITE_STORE_INSTALL, ensure_sqlite_parent_dir, resolve_sqlite_conn_str +from deerflow.runtime.store.provider import ( + POSTGRES_CONN_REQUIRED, + POSTGRES_STORE_INSTALL, + SQLITE_STORE_INSTALL, + _resolve_store_config, + ensure_sqlite_parent_dir, + resolve_sqlite_conn_str, +) logger = logging.getLogger(__name__) @@ -54,7 +62,7 @@ async def _async_store(config) -> AsyncIterator[BaseStore]: raise ImportError(SQLITE_STORE_INSTALL) from exc conn_str = resolve_sqlite_conn_str(config.connection_string or "store.db") - ensure_sqlite_parent_dir(conn_str) + await asyncio.to_thread(ensure_sqlite_parent_dir, conn_str) async with AsyncSqliteStore.from_conn_string(conn_str) as store: await store.setup() @@ -87,28 +95,21 @@ async def _async_store(config) -> AsyncIterator[BaseStore]: @contextlib.asynccontextmanager async def make_store(app_config: AppConfig | None = None) -> AsyncIterator[BaseStore]: - """Async context manager that yields a Store whose backend matches the - configured checkpointer. + """Yield a Store selected from legacy or unified persistence config. - Reads from the same ``checkpointer`` section of *config.yaml* used by - :func:`deerflow.runtime.checkpointer.async_provider.make_checkpointer` so - that both singletons always use the same persistence technology:: + The legacy ``checkpointer`` section takes precedence when configured; + otherwise the unified ``database`` section selects the backend, matching + :func:`deerflow.runtime.checkpointer.async_provider.make_checkpointer`:: async with make_store(app_config) as store: app.state.store = store - Yields an :class:`~langgraph.store.memory.InMemoryStore` when no - ``checkpointer`` section is configured (emits a WARNING in that case). + An :class:`~langgraph.store.memory.InMemoryStore` is returned only when the + resolved backend is explicitly ``memory``. """ if app_config is None: app_config = get_app_config() - if app_config.checkpointer is None: - from langgraph.store.memory import InMemoryStore - - logger.warning("No 'checkpointer' section in config.yaml — using InMemoryStore for the store. Thread list will be lost on server restart. Configure a sqlite or postgres backend for persistence.") - yield InMemoryStore() - return - - async with _async_store(app_config.checkpointer) as store: + config = _resolve_store_config(app_config) + async with _async_store(config) as store: yield store diff --git a/backend/packages/harness/deerflow/runtime/store/provider.py b/backend/packages/harness/deerflow/runtime/store/provider.py index 7e2ae563d..2175749f4 100644 --- a/backend/packages/harness/deerflow/runtime/store/provider.py +++ b/backend/packages/harness/deerflow/runtime/store/provider.py @@ -3,8 +3,9 @@ Provides a **sync singleton** and a **sync context manager** for CLI tools and the embedded :class:`~deerflow.client.DeerFlowClient`. -The backend mirrors the configured checkpointer so that both always use the -same persistence technology. Supported backends: memory, sqlite, postgres. +The deprecated ``checkpointer`` section takes precedence when present; +otherwise Store follows the unified ``database`` section. Supported backends: +memory, sqlite, postgres. Usage:: @@ -27,8 +28,8 @@ from collections.abc import Iterator from langgraph.store.base import BaseStore -from deerflow.config.app_config import get_app_config -from deerflow.config.checkpointer_config import ensure_config_loaded +from deerflow.config.app_config import AppConfig, get_app_config +from deerflow.config.checkpointer_config import CheckpointerConfig, ensure_config_loaded, get_checkpointer_config from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir, resolve_sqlite_conn_str logger = logging.getLogger(__name__) @@ -43,6 +44,44 @@ POSTGRES_STORE_INSTALL = ( ) POSTGRES_CONN_REQUIRED = "checkpointer.connection_string is required for the postgres backend" + +def _resolve_store_config(app_config: AppConfig) -> CheckpointerConfig: + """Resolve the Store backend from legacy or unified application config. + + The legacy ``checkpointer`` section remains authoritative when present so + Store and Checkpointer continue to use the same backend. Otherwise the + unified ``database`` section drives the Store as documented. + """ + if app_config.checkpointer is not None: + return app_config.checkpointer + + database = app_config.database + if database.backend == "memory": + return CheckpointerConfig(type="memory") + if database.backend == "sqlite": + return CheckpointerConfig(type="sqlite", connection_string=database.checkpointer_sqlite_path) + if database.backend == "postgres": + if not database.postgres_url: + raise ValueError("database.postgres_url is required for the postgres backend") + return CheckpointerConfig(type="postgres", connection_string=database.postgres_url) + raise ValueError(f"Unknown database backend: {database.backend!r}") + + +def _get_store_config() -> CheckpointerConfig: + """Load Store config without holding the provider singleton lock.""" + ensure_config_loaded() + + # Preserve callers that initialise the legacy config singleton directly. + legacy_config = get_checkpointer_config() + if legacy_config is not None: + return legacy_config + try: + app_config = get_app_config() + except FileNotFoundError: + return CheckpointerConfig(type="memory") + return _resolve_store_config(app_config) + + # --------------------------------------------------------------------------- # Sync factory # --------------------------------------------------------------------------- @@ -108,37 +147,26 @@ _store_lock = threading.Lock() def get_store() -> BaseStore: """Return the global sync Store singleton, creating it on first call. - Returns an :class:`~langgraph.store.memory.InMemoryStore` when no - checkpointer is configured in *config.yaml* (emits a WARNING in that case). + The legacy ``checkpointer`` section takes precedence when configured; + otherwise the unified ``database`` section selects the backend. Raises: ImportError: If the required package for the configured backend is not installed. - ValueError: If ``connection_string`` is missing for a backend that requires it. + ValueError: If the selected backend is missing its required connection value. """ global _store, _store_ctx if _store is not None: return _store - # Config loading can reset both persistence singletons. Keep it outside - # this provider lock to avoid cross-provider lock-order inversion. - ensure_config_loaded() + # Config loading can reset both persistence singletons. Resolve the full + # config outside this provider lock to avoid lock-order inversion. + config = _get_store_config() with _store_lock: if _store is not None: return _store - from deerflow.config.checkpointer_config import get_checkpointer_config - - config = get_checkpointer_config() - - if config is None: - from langgraph.store.memory import InMemoryStore - - logger.warning("No 'checkpointer' section in config.yaml — using InMemoryStore for the store. Thread list will be lost on server restart. Configure a sqlite or postgres backend for persistence.") - _store = InMemoryStore() - return _store - store_ctx = _sync_store_cm(config) store = store_ctx.__enter__() _store_ctx = store_ctx @@ -179,16 +207,9 @@ def store_context() -> Iterator[BaseStore]: with store_context() as store: store.put(("threads",), thread_id, {...}) - Yields an :class:`~langgraph.store.memory.InMemoryStore` when no - checkpointer is configured in *config.yaml*. + The legacy ``checkpointer`` section takes precedence when configured; + otherwise the unified ``database`` section selects the backend. """ - config = get_app_config() - if config.checkpointer is None: - from langgraph.store.memory import InMemoryStore - - logger.warning("No 'checkpointer' section in config.yaml — using InMemoryStore for the store. Thread list will be lost on server restart. Configure a sqlite or postgres backend for persistence.") - yield InMemoryStore() - return - - with _sync_store_cm(config.checkpointer) as store: + config = _resolve_store_config(get_app_config()) + with _sync_store_cm(config) as store: yield store diff --git a/backend/tests/test_checkpointer.py b/backend/tests/test_checkpointer.py index a9ac227ce..c089ddc93 100644 --- a/backend/tests/test_checkpointer.py +++ b/backend/tests/test_checkpointer.py @@ -3,8 +3,10 @@ import sys import tomllib from concurrent.futures import ThreadPoolExecutor +from contextlib import nullcontext from pathlib import Path from threading import Barrier, Event, Lock +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -17,6 +19,7 @@ from deerflow.config.checkpointer_config import ( load_checkpointer_config_from_dict, set_checkpointer_config, ) +from deerflow.config.database_config import DatabaseConfig from deerflow.runtime.checkpointer import get_checkpointer, reset_checkpointer from deerflow.runtime.checkpointer.provider import POSTGRES_INSTALL from deerflow.runtime.store import get_store, reset_store @@ -702,6 +705,156 @@ class TestAsyncCheckpointer: mock_saver.setup.assert_awaited_once() +class TestStoreDatabaseConfig: + def test_sync_store_falls_back_to_memory_when_config_file_is_missing(self): + """The sync Store keeps its no-config fallback for embedded callers.""" + from langgraph.store.memory import InMemoryStore + + with ( + patch("deerflow.runtime.store.provider.ensure_config_loaded"), + patch("deerflow.runtime.store.provider.get_checkpointer_config", return_value=None), + patch("deerflow.runtime.store.provider.get_app_config", side_effect=FileNotFoundError), + ): + assert isinstance(get_store(), InMemoryStore) + + @pytest.mark.anyio + async def test_async_postgres_store_uses_database_config(self, caplog): + """Unified database postgres config must not fall back to InMemoryStore.""" + from deerflow.runtime.store.async_provider import make_store + + caplog.set_level("WARNING", logger="deerflow.runtime.store.async_provider") + app_config = SimpleNamespace( + checkpointer=None, + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + mock_store = AsyncMock() + mock_cm = AsyncMock() + mock_cm.__aenter__.return_value = mock_store + mock_cm.__aexit__.return_value = False + mock_store_cls = MagicMock() + mock_store_cls.from_conn_string.return_value = mock_cm + mock_module = MagicMock(AsyncPostgresStore=mock_store_cls) + + with patch.dict(sys.modules, {"langgraph.store.postgres.aio": mock_module}): + async with make_store(app_config) as store: + assert store is mock_store + + mock_store_cls.from_conn_string.assert_called_once_with("postgresql://localhost/db") + mock_store.setup.assert_awaited_once() + assert "No 'checkpointer' section" not in caplog.text + + @pytest.mark.anyio + async def test_async_sqlite_store_uses_unified_database_path(self, tmp_path): + """Unified database SQLite config must use the shared deerflow.db path.""" + from deerflow.runtime.store.async_provider import make_store + from deerflow.runtime.store.provider import ensure_sqlite_parent_dir + + db_config = DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)) + app_config = SimpleNamespace(checkpointer=None, database=db_config) + mock_store = AsyncMock() + mock_cm = AsyncMock() + mock_cm.__aenter__.return_value = mock_store + mock_cm.__aexit__.return_value = False + mock_store_cls = MagicMock() + mock_store_cls.from_conn_string.return_value = mock_cm + mock_module = MagicMock(AsyncSqliteStore=mock_store_cls) + + with ( + patch.dict(sys.modules, {"langgraph.store.sqlite.aio": mock_module}), + patch( + "deerflow.runtime.store.async_provider.asyncio.to_thread", + new_callable=AsyncMock, + ) as mock_to_thread, + ): + async with make_store(app_config) as store: + assert store is mock_store + + mock_to_thread.assert_awaited_once() + called_fn, called_path = mock_to_thread.await_args.args + assert called_fn is ensure_sqlite_parent_dir + assert called_path == db_config.checkpointer_sqlite_path + mock_store_cls.from_conn_string.assert_called_once_with(db_config.checkpointer_sqlite_path) + mock_store.setup.assert_awaited_once() + + def test_sync_store_context_uses_database_config(self): + """The one-shot sync Store factory must follow unified database config.""" + from deerflow.runtime.store.provider import store_context + + app_config = SimpleNamespace( + checkpointer=None, + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + expected_store = object() + factory = MagicMock(return_value=nullcontext(expected_store)) + + with ( + patch("deerflow.runtime.store.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.store.provider._sync_store_cm", factory), + store_context() as store, + ): + assert store is expected_store + + resolved = factory.call_args.args[0] + assert resolved.type == "postgres" + assert resolved.connection_string == "postgresql://localhost/db" + + def test_sync_store_singleton_uses_database_config(self): + """The cached sync Store factory must resolve database config before locking.""" + app_config = SimpleNamespace( + checkpointer=None, + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + expected_store = object() + factory = MagicMock(return_value=nullcontext(expected_store)) + + with ( + patch("deerflow.runtime.store.provider.ensure_config_loaded"), + patch("deerflow.runtime.store.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.store.provider._sync_store_cm", factory), + ): + assert get_store() is expected_store + + resolved = factory.call_args.args[0] + assert resolved.type == "postgres" + assert resolved.connection_string == "postgresql://localhost/db" + + def test_legacy_checkpointer_config_takes_precedence_for_store(self): + """Backward-compatible checkpointer config must override database for Store.""" + from deerflow.runtime.store.provider import store_context + + app_config = SimpleNamespace( + checkpointer=CheckpointerConfig(type="memory"), + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + expected_store = object() + factory = MagicMock(return_value=nullcontext(expected_store)) + + with ( + patch("deerflow.runtime.store.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.store.provider._sync_store_cm", factory), + store_context() as store, + ): + assert store is expected_store + + resolved = factory.call_args.args[0] + assert resolved.type == "memory" + assert resolved.connection_string is None + + def test_explicit_memory_database_uses_in_memory_store(self): + """Explicit memory mode remains an intentional non-persistent Store.""" + from langgraph.store.memory import InMemoryStore + + from deerflow.runtime.store.provider import store_context + + app_config = SimpleNamespace( + checkpointer=None, + database=DatabaseConfig(backend="memory"), + ) + with patch("deerflow.runtime.store.provider.get_app_config", return_value=app_config): + with store_context() as store: + assert isinstance(store, InMemoryStore) + + # --------------------------------------------------------------------------- # app_config.py integration # --------------------------------------------------------------------------- diff --git a/backend/tests/test_runtime_lifecycle_e2e.py b/backend/tests/test_runtime_lifecycle_e2e.py index 508cc69d8..c093a2187 100644 --- a/backend/tests/test_runtime_lifecycle_e2e.py +++ b/backend/tests/test_runtime_lifecycle_e2e.py @@ -274,6 +274,15 @@ def isolated_app(isolated_deer_flow_home: Path, monkeypatch: pytest.MonkeyPatch) return create_app() +def test_lifespan_uses_sqlite_store_from_database_config(isolated_app): + """Gateway startup must bind LangGraph Store to the unified database backend.""" + from langgraph.store.sqlite.aio import AsyncSqliteStore + from starlette.testclient import TestClient + + with TestClient(isolated_app): + assert isinstance(isolated_app.state.store, AsyncSqliteStore) + + @pytest.fixture def isolated_app_with_title(isolated_deer_flow_home: Path, monkeypatch: pytest.MonkeyPatch): config_path = isolated_deer_flow_home.parent / "config-title-enabled.yaml" diff --git a/config.example.yaml b/config.example.yaml index 522cb83d7..a4bf3127c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1299,12 +1299,12 @@ skill_evolution: # Checkpointer Configuration (DEPRECATED — use `database` instead) # ============================================================================ # Legacy standalone checkpointer config. Kept for backward compatibility. -# Prefer the unified `database` section below, which drives BOTH the -# LangGraph checkpointer AND DeerFlow application data (runs, feedback, -# events) from a single backend setting. +# Prefer the unified `database` section below, which drives the LangGraph +# checkpointer, LangGraph Store, and DeerFlow application data (runs, +# feedback, events) from a single backend setting. # # If both `checkpointer` and `database` are present, `checkpointer` -# takes precedence for LangGraph state persistence only. +# takes precedence for the LangGraph checkpointer and Store only. # # checkpointer: # type: sqlite @@ -1317,8 +1317,8 @@ skill_evolution: # ============================================================================ # Database # ============================================================================ -# Unified storage backend for LangGraph checkpointer and DeerFlow -# application data (runs, threads metadata, feedback, etc.). +# Unified storage backend for the LangGraph checkpointer, LangGraph Store, +# and DeerFlow application data (runs, threads metadata, feedback, etc.). # # backend: memory -- No persistence, data lost on restart # backend: sqlite -- Single-node deployment, files in sqlite_dir @@ -1329,7 +1329,7 @@ skill_evolution: # sqlite_dir: .deer-flow/data # # SQLite mode uses a single deerflow.db file with WAL journal mode -# for both checkpointer and application data. +# for the checkpointer, Store, and application data. # # Postgres mode: put your connection URL in .env as DATABASE_URL, # then reference it here with $DATABASE_URL. @@ -1354,7 +1354,8 @@ skill_evolution: # (--all-packages propagates the extra into workspace members — see PR #2584) # # NOTE: When both `checkpointer` and `database` are configured, -# `checkpointer` takes precedence for LangGraph state persistence. +# `checkpointer` takes precedence for the LangGraph checkpointer and Store; +# `database` still controls DeerFlow application data. # If you use `database`, you can remove the `checkpointer` section. # database: # backend: sqlite