From c143c0415bed7b1b4641def8cb505ac7812d5e26 Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Sat, 11 Jul 2026 17:15:32 -0700 Subject: [PATCH] fix(config): Make the sync checkpointer honor the unified database config (#3994) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Make the sync checkpointer honor the unified database config The sync checkpointer factory (`get_checkpointer` and `checkpointer_context`) read only the legacy `checkpointer:` config section and fell back to `InMemorySaver` when it was absent — it never consulted `database:`. The async `make_checkpointer` factory and both sync/async Store providers already resolve the unified `database:` section (legacy `checkpointer:` takes precedence, otherwise `database:` drives the backend), so the sync checkpointer was the lone outlier. Consequence: with `database: {backend: sqlite|postgres}` and no legacy `checkpointer:` section, the sync checkpointer silently returned `InMemorySaver` while the Store — same process, same config — correctly persisted to sqlite/postgres. Embedded callers (`DeerFlowClient`) and the TUI hit this: e.g. the TUI writes `threads_meta` rows to sqlite (thread appears in the Web UI) but its checkpoints went to memory and were lost on exit. This also contradicts backend/AGENTS.md ("the unified `database` section selects the Gateway's LangGraph checkpointer, LangGraph Store, and DeerFlow SQL repositories"). Mirror the sync Store provider: add `_resolve_checkpointer_config` / `_get_checkpointer_config` (legacy precedence, else translate `database:` into a CheckpointerConfig) and route both the singleton and the context manager through them. The Gateway is unaffected — it uses the async path. Adds a TestCheckpointerDatabaseConfig suite mirroring the Store's database tests (singleton + context-manager honor `database:`, legacy precedence, explicit-memory, missing-config fallback). The two `uses_database_config` tests fail (return InMemorySaver) before the fix and pass after. * Handle database=None in sync checkpointer resolution The unified-config change made `checkpointer_context` / `get_checkpointer` consult `app_config.database` when no legacy `checkpointer:` section is set. `AppConfig.database` is always a `DatabaseConfig` in production, but the existing regression test for issue #1016 (`test_checkpointer_none_fix.py::test_sync_checkpointer_context_returns_in_memory_saver_when_not_configured`) mocks the app config with `database` left unset, exercising a `database=None` path that now raised `ValueError: Unknown database backend`. Mirror the async `make_checkpointer` factory, which already tolerates `database=None` (falls back to memory), by treating a `None` database as the memory backend in `_resolve_checkpointer_config`. Update the sync test to set `mock_config.database = None`, matching its async sibling in the same file. * Address review: mirror None-guard in store resolver, add sqlite coverage --- .../deerflow/runtime/checkpointer/provider.py | 84 ++++++++---- .../deerflow/runtime/store/provider.py | 2 +- backend/tests/test_checkpointer.py | 122 ++++++++++++++++++ backend/tests/test_checkpointer_none_fix.py | 3 +- 4 files changed, 183 insertions(+), 28 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/checkpointer/provider.py b/backend/packages/harness/deerflow/runtime/checkpointer/provider.py index 226454eb3..1545ab03f 100644 --- a/backend/packages/harness/deerflow/runtime/checkpointer/provider.py +++ b/backend/packages/harness/deerflow/runtime/checkpointer/provider.py @@ -26,8 +26,8 @@ from collections.abc import Iterator from langgraph.types import Checkpointer -from deerflow.config.app_config import get_app_config -from deerflow.config.checkpointer_config import CheckpointerConfig, 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__) @@ -42,6 +42,51 @@ POSTGRES_INSTALL = ( ) POSTGRES_CONN_REQUIRED = "checkpointer.connection_string is required for the postgres backend" + +# --------------------------------------------------------------------------- +# Config resolution +# --------------------------------------------------------------------------- + + +def _resolve_checkpointer_config(app_config: AppConfig) -> CheckpointerConfig: + """Resolve the checkpointer backend from legacy or unified application config. + + The legacy ``checkpointer`` section remains authoritative when present so + Checkpointer and Store keep using the same backend. Otherwise the unified + ``database`` section drives the checkpointer, matching the async + :func:`~deerflow.runtime.checkpointer.async_provider.make_checkpointer` + factory and the sync Store provider's ``_resolve_store_config``. + """ + if app_config.checkpointer is not None: + return app_config.checkpointer + + database = app_config.database + if database is None or 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_checkpointer_config() -> CheckpointerConfig: + """Load checkpointer 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_checkpointer_config(app_config) + + # --------------------------------------------------------------------------- # Sync factory # --------------------------------------------------------------------------- @@ -107,7 +152,9 @@ _checkpointer_lock = threading.Lock() def get_checkpointer() -> Checkpointer: """Return the global sync checkpointer singleton, creating it on first call. - Returns an ``InMemorySaver`` when no checkpointer is configured in *config.yaml*. + The legacy ``checkpointer`` section takes precedence when configured; + otherwise the unified ``database`` section selects the backend. Returns an + ``InMemorySaver`` when neither selects a persistent backend. Raises: ImportError: If the required package for the configured backend is not installed. @@ -118,25 +165,14 @@ def get_checkpointer() -> Checkpointer: if _checkpointer is not None: return _checkpointer - # 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 cross-provider lock-order inversion. + config = _get_checkpointer_config() with _checkpointer_lock: if _checkpointer is not None: return _checkpointer - from deerflow.config.checkpointer_config import get_checkpointer_config - - config = get_checkpointer_config() - - if config is None: - from langgraph.checkpoint.memory import InMemorySaver - - logger.info("Checkpointer: using InMemorySaver (in-process, not persistent)") - _checkpointer = InMemorySaver() - return _checkpointer - checkpointer_ctx = _sync_checkpointer_cm(config) checkpointer = checkpointer_ctx.__enter__() _checkpointer_ctx = checkpointer_ctx @@ -178,15 +214,11 @@ def checkpointer_context() -> Iterator[Checkpointer]: with checkpointer_context() as cp: graph.invoke(input, config={"configurable": {"thread_id": "1"}}) - Yields an ``InMemorySaver`` when no checkpointer is configured in *config.yaml*. + The legacy ``checkpointer`` section takes precedence when configured; + otherwise the unified ``database`` section selects the backend. Yields an + ``InMemorySaver`` when neither selects a persistent backend. """ - config = get_app_config() - if config.checkpointer is None: - from langgraph.checkpoint.memory import InMemorySaver - - yield InMemorySaver() - return - - with _sync_checkpointer_cm(config.checkpointer) as saver: + config = _resolve_checkpointer_config(get_app_config()) + with _sync_checkpointer_cm(config) as saver: yield saver diff --git a/backend/packages/harness/deerflow/runtime/store/provider.py b/backend/packages/harness/deerflow/runtime/store/provider.py index 2175749f4..37269995c 100644 --- a/backend/packages/harness/deerflow/runtime/store/provider.py +++ b/backend/packages/harness/deerflow/runtime/store/provider.py @@ -56,7 +56,7 @@ def _resolve_store_config(app_config: AppConfig) -> CheckpointerConfig: return app_config.checkpointer database = app_config.database - if database.backend == "memory": + if database is None or database.backend == "memory": return CheckpointerConfig(type="memory") if database.backend == "sqlite": return CheckpointerConfig(type="sqlite", connection_string=database.checkpointer_sqlite_path) diff --git a/backend/tests/test_checkpointer.py b/backend/tests/test_checkpointer.py index f305afc0a..753f1053f 100644 --- a/backend/tests/test_checkpointer.py +++ b/backend/tests/test_checkpointer.py @@ -705,6 +705,128 @@ class TestAsyncCheckpointer: mock_saver.setup.assert_awaited_once() +class TestCheckpointerDatabaseConfig: + """The sync checkpointer must follow the unified ``database`` section when no + legacy ``checkpointer`` section is configured — matching the async + ``make_checkpointer`` factory and the sync Store provider. + + Regression: ``get_checkpointer`` / ``checkpointer_context`` previously read + only the legacy ``checkpointer`` section and fell back to ``InMemorySaver``, + silently ignoring ``database``. Embedded callers (``DeerFlowClient``) and the + TUI then persisted Store rows to sqlite/postgres while checkpoints went to an + in-memory saver and were lost on exit. + """ + + def test_sync_checkpointer_context_uses_database_config(self): + """The one-shot sync checkpointer factory must follow unified database config.""" + from deerflow.runtime.checkpointer.provider import checkpointer_context + + app_config = SimpleNamespace( + checkpointer=None, + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + expected = object() + factory = MagicMock(return_value=nullcontext(expected)) + + with ( + patch("deerflow.runtime.checkpointer.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.checkpointer.provider._sync_checkpointer_cm", factory), + checkpointer_context() as cp, + ): + assert cp is expected + + resolved = factory.call_args.args[0] + assert resolved.type == "postgres" + assert resolved.connection_string == "postgresql://localhost/db" + + def test_sync_checkpointer_context_uses_sqlite_database_config(self, tmp_path): + """The one-shot sync checkpointer factory must resolve the sqlite branch too, not just postgres.""" + from deerflow.runtime.checkpointer.provider import checkpointer_context + + db_config = DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)) + app_config = SimpleNamespace(checkpointer=None, database=db_config) + expected = object() + factory = MagicMock(return_value=nullcontext(expected)) + + with ( + patch("deerflow.runtime.checkpointer.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.checkpointer.provider._sync_checkpointer_cm", factory), + checkpointer_context() as cp, + ): + assert cp is expected + + resolved = factory.call_args.args[0] + assert resolved.type == "sqlite" + assert resolved.connection_string == db_config.checkpointer_sqlite_path + + def test_sync_checkpointer_singleton_uses_database_config(self): + """The cached sync checkpointer factory must resolve database config before locking.""" + app_config = SimpleNamespace( + checkpointer=None, + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + expected = object() + factory = MagicMock(return_value=nullcontext(expected)) + + with ( + patch("deerflow.runtime.checkpointer.provider.ensure_config_loaded"), + patch("deerflow.runtime.checkpointer.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.checkpointer.provider._sync_checkpointer_cm", factory), + ): + assert get_checkpointer() is expected + + resolved = factory.call_args.args[0] + assert resolved.type == "postgres" + assert resolved.connection_string == "postgresql://localhost/db" + + def test_sync_checkpointer_falls_back_to_memory_when_config_file_is_missing(self): + """The sync checkpointer keeps its no-config fallback for embedded callers.""" + from langgraph.checkpoint.memory import InMemorySaver + + with ( + patch("deerflow.runtime.checkpointer.provider.ensure_config_loaded"), + patch("deerflow.runtime.checkpointer.provider.get_checkpointer_config", return_value=None), + patch("deerflow.runtime.checkpointer.provider.get_app_config", side_effect=FileNotFoundError), + ): + assert isinstance(get_checkpointer(), InMemorySaver) + + def test_legacy_checkpointer_config_takes_precedence(self): + """Backward-compatible checkpointer config must override database.""" + from deerflow.runtime.checkpointer.provider import checkpointer_context + + app_config = SimpleNamespace( + checkpointer=CheckpointerConfig(type="memory"), + database=DatabaseConfig(backend="postgres", postgres_url="postgresql://localhost/db"), + ) + expected = object() + factory = MagicMock(return_value=nullcontext(expected)) + + with ( + patch("deerflow.runtime.checkpointer.provider.get_app_config", return_value=app_config), + patch("deerflow.runtime.checkpointer.provider._sync_checkpointer_cm", factory), + checkpointer_context() as cp, + ): + assert cp is expected + + resolved = factory.call_args.args[0] + assert resolved.type == "memory" + assert resolved.connection_string is None + + def test_explicit_memory_database_uses_in_memory_saver(self): + """Explicit memory mode remains an intentional non-persistent checkpointer.""" + from langgraph.checkpoint.memory import InMemorySaver + + from deerflow.runtime.checkpointer.provider import checkpointer_context + + app_config = SimpleNamespace(checkpointer=None, database=DatabaseConfig(backend="memory")) + + with ( + patch("deerflow.runtime.checkpointer.provider.get_app_config", return_value=app_config), + checkpointer_context() as cp, + ): + assert isinstance(cp, InMemorySaver) + + 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.""" diff --git a/backend/tests/test_checkpointer_none_fix.py b/backend/tests/test_checkpointer_none_fix.py index 3c7a25fa1..34a3b75fe 100644 --- a/backend/tests/test_checkpointer_none_fix.py +++ b/backend/tests/test_checkpointer_none_fix.py @@ -38,9 +38,10 @@ class TestCheckpointerNoneFix: """checkpointer_context should return InMemorySaver when config.checkpointer is None.""" from deerflow.runtime.checkpointer.provider import checkpointer_context - # Mock get_app_config to return a config with checkpointer=None + # Mock get_app_config to return a config with checkpointer=None and database=None mock_config = MagicMock() mock_config.checkpointer = None + mock_config.database = None with patch("deerflow.runtime.checkpointer.provider.get_app_config", return_value=mock_config): with checkpointer_context() as checkpointer: