deer-flow/backend/tests/test_config_version.py
Vanzeren 42baed8c8c
feat(checkpoint): dual-mode checkpoint storage with LangGraph DeltaChannel (#4292)
* feat(checkpoint): dual-mode checkpoint storage with LangGraph DeltaChannel

Add a restart-required database.checkpoint_channel_mode ("full" default,
"delta") that stores the messages channel via LangGraph 1.2 DeltaChannel,
cutting checkpoint storage from O(n^2) to O(n) for append-only history.
Existing full checkpoints seed delta state transparently; no data migration.

- config: mode schema + freeze-on-first-use with
  CheckpointModeReconfigurationError; mode marker persisted in checkpoint
  metadata; unsafe delta->full downgrade rejected fail-closed with
  CheckpointModeMismatchError (run-level error, failed state read)
- state: delta message state schema; CheckpointStateAccessor centralizes
  materialized reads for all consumers (threads API, branches,
  regeneration, compaction, state updates, memory, goal workers)
- runtime: raw writers (run durations, interrupted title, thread goal)
  parent their checkpoints to the checkpoint they derive from, preserving
  delta ancestry; rollback forks the pre-run lineage through a state
  mutation graph with Overwrite restores; InMemorySaver delta-history
  override delegates to the base walk (fixes dropped first write after
  migration, also present upstream)
- tests: conformance suite over {memory, sqlite, postgres} covering
  migration replay, stable message IDs, storage shape and writer
  preservation; conftest fixture isolates the frozen mode between tests;
  stale config fakes refreshed
- ci: backend unit tests gain a postgres service

* fix(checkpoint): close materialization gaps in goal flow, guard public factory

- Route goal-continuation message reads through CheckpointStateAccessor:
  raw channel_values reads see the delta sentinel in delta mode, which
  disabled goal continuation (stand_down=no_durable_end_of_turn) after
  durable assistant turns. Raw tuples remain for tuple-only metadata
  (checkpoint id, pending_writes).
- Reject checkpoint_channel_mode='delta' + checkpointer in
  create_deerflow_agent at construction: factory-built persisted graphs
  bypass mode-marker injection and the fail-closed gate, reproducing
  silent mixed-mode state loss. Delta without persistence stays allowed.
- Import the postgres saver lazily (pytest.importorskip in the fixture)
  so the documented default install collects the suite; add a CI job
  running pytest --collect-only on uv sync --group dev without extras.
- Fix test_checkpointer fallback test to patch get_app_config at its
  use site (provider module), making it deterministic when a local
  config.yaml selects a persistent backend.

* fix(gateway): preserve extension-owned channels in state mutations, bump config version

- build_state_mutation_graph / build_checkpoint_state_mutation_accessor
  accept an explicit state_schema; branch and POST /state now compile the
  mutation graph from the thread's effective schema
  (graph_state_schema on the assistant graph). The base-ThreadState
  fallback silently discarded channels contributed by custom
  AgentMiddleware.state_schema on branch (data loss) and returned a
  false-success 200 on POST /state.
- POST /state validates values keys against the mutation graph's
  channels and rejects unknown fields with 422 instead of ignoring
  them; reducer detection covers extension channels
  (BinaryOperatorAggregate or DeltaChannel) so Overwrite replace
  semantics work for middleware reducers in both modes.
- Endpoint regression: custom AgentMiddleware.state_schema value
  survives branch, updates through POST /state, and an unknown field
  receives 422.
- config_version 26 -> 27 for the new database.checkpoint_channel_mode
  (example, Helm chart values + README, support-bundle fixture), so
  existing installs get the outdated-config warning and
  make config-upgrade merges the field; covered by a test driving the
  real example file and the real config-upgrade script.

* fix(gateway): resolve assistant schema via one boundary, copy branch reducer values with Overwrite

GET /threads/{id}/state now resolves the thread's assistant_id through a
single reusable boundary (thread metadata -> assistant_id -> effective
graph), so channels contributed by a custom AgentMiddleware.state_schema
are materialized instead of dropped by the default lead schema. POST
/state uses the same boundary instead of resolving the schema ad hoc.

Branch writes wrap every copied reducer channel in Overwrite (derived
from the effective mutation graph: BinaryOperatorAggregate + DeltaChannel),
not just messages, so already-aggregated values are never re-merged.

Regression tests use a real AgentMiddleware.state_schema with a
non-identity reducer in both full and delta modes: GET /state returns the
extension value, POST /state replaces it, branch preserves it
byte-for-byte; the unknown-field 422 is a separate assertion.

* refactor(checkpoint): collapse read-path round-trips and ship dual-mode parity tests

Address review round 4 on PR #4292:

- Push ahistory/history limit through Pregel into checkpointer.alist
  (SQL LIMIT) instead of materializing all rows and breaking in Python
- Fold the read-side mode-compat gate onto the returned snapshot's
  metadata; only writes keep the pre-write tuple fetch (fail-closed)
- Cache factory-built accessor graphs per (assistant_id, mode) with
  factory-identity revalidation; state reads no longer build a lead
  agent per request
- get_thread: one snapshot fetch + one raw pending_writes fetch on the
  resolved checkpoint (post-checkpoint __error__ writes never surface
  in snapshot.tasks; verified empirically)
- DeerFlowClient.get_thread: single checkpointer.list walk collects
  pending_writes per checkpoint instead of N get_tuple calls
- InMemorySaver delta-history patch: stand-down when the upstream
  override disappears, try/except guard, validated-version warning,
  guard tests
- make_lead_agent mode precedence: first freeze is owned by app_config
  (client-supplied configurable key ignored); once frozen, injected
  key/app_config must match or fail closed
- Rollback: lock in non-message channel restoration via fork
  inheritance with a dedicated reducer-channel test
- Add tests/test_threads_checkpoint_mode.py and
  tests/test_gateway_checkpoint_mode.py referenced by AGENTS.md and
  the PR validation section: lifecycle parity (memory + sqlite),
  per-step blob-count storage guard, gateway endpoint parity

Counted-saver tests pin checkpoint round-trips for aget/ahistory so
these regressions cannot silently return.

* fix(checkpoint): precise mode-mismatch HTTP mapping, gate E2E, and accessor resilience

- threads router: map CheckpointModeMismatchError to 409 (with cause and
  thread id) and CheckpointModeReconfigurationError to 503 across all state
  endpoints instead of swallowing both into a generic 500
- gate coverage: seed a real delta checkpoint into AsyncSqliteSaver and
  assert aget/aupdate/ahistory fail closed; assert 409 at the HTTP boundary
  through the real route stack
- rollback: compile the restore mutation graph with the thread's effective
  state schema per the build_state_mutation_graph contract
- inheritance contract locks: rollback and manual compaction preserve
  middleware-contributed channels via checkpoint fork cloning
- services: revalidate the accessor-graph cache against app_config identity
  so config.yaml hot-reloads never serve a stale compiled graph
- services: degrade full-mode state reads to raw checkpointer reads when the
  agent factory is unavailable (delta gate still applies; delta mode has no
  fallback)
- deps: override websockets==16.0 (langgraph-sdk 0.4.2's <16 pin silently
  downgraded 16.0 -> 15.0.1; pin is not grounded in any API incompatibility)
  and bump the langchain lower bound to what the lockfile actually resolves

* fix(checkpoint): include anchor checkpoint in degraded history walk + cover get_thread

- _RawCheckpointReadAccessor.ahistory: alist(before=...) is exclusive while
  pregel's get_state_history treats config.checkpoint_id as the inclusive
  start; fetch the anchor explicitly so both read paths paginate identically
- extend the degraded-path gateway test: GET /thread returns raw values, and
  POST /history with before starts at the anchor checkpoint

* fix(gateway): preserve degraded checkpoint timestamps

* fix(gateway): harden degraded checkpoint access

* fix(gateway): resolve assistants for checkpoint reads

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-22 08:33:29 +08:00

239 lines
9.4 KiB
Python

"""Tests for config version check and upgrade logic."""
from __future__ import annotations
import logging
import os
import tempfile
from pathlib import Path
import yaml
from deerflow.config.app_config import AppConfig
def _make_config_files(tmpdir: Path, user_config: dict, example_config: dict) -> Path:
"""Write user config.yaml and config.example.yaml to a temp dir, return config path."""
config_path = tmpdir / "config.yaml"
example_path = tmpdir / "config.example.yaml"
# Minimal valid config needs sandbox
defaults = {
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
}
for cfg in (user_config, example_config):
for k, v in defaults.items():
cfg.setdefault(k, v)
with open(config_path, "w", encoding="utf-8") as f:
yaml.dump(user_config, f)
with open(example_path, "w", encoding="utf-8") as f:
yaml.dump(example_config, f)
return config_path
def test_missing_version_treated_as_zero(caplog):
"""Config without config_version should be treated as version 0."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = _make_config_files(
Path(tmpdir),
user_config={}, # no config_version
example_config={"config_version": 1},
)
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version(
{"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}},
config_path,
)
assert "outdated" in caplog.text
assert "version 0" in caplog.text
assert "version is 1" in caplog.text
def test_matching_version_no_warning(caplog):
"""Config with matching version should not emit a warning."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = _make_config_files(
Path(tmpdir),
user_config={"config_version": 1},
example_config={"config_version": 1},
)
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version(
{"config_version": 1},
config_path,
)
assert "outdated" not in caplog.text
def test_outdated_version_emits_warning(caplog):
"""Config with lower version should emit a warning."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = _make_config_files(
Path(tmpdir),
user_config={"config_version": 1},
example_config={"config_version": 2},
)
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version(
{"config_version": 1},
config_path,
)
assert "outdated" in caplog.text
assert "version 1" in caplog.text
assert "version is 2" in caplog.text
def test_no_example_file_no_warning(caplog):
"""If config.example.yaml doesn't exist, no warning should be emitted."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = Path(tmpdir) / "config.yaml"
with open(config_path, "w", encoding="utf-8") as f:
yaml.dump({"sandbox": {"use": "test"}}, f)
# No config.example.yaml created
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version({}, config_path)
assert "outdated" not in caplog.text
def test_string_config_version_does_not_raise_type_error(caplog):
"""config_version stored as a YAML string should not raise TypeError on comparison."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = _make_config_files(
Path(tmpdir),
user_config={"config_version": "1"}, # string, as YAML can produce
example_config={"config_version": 2},
)
# Must not raise TypeError: '<' not supported between instances of 'str' and 'int'
AppConfig._check_config_version({"config_version": "1"}, config_path)
def test_newer_user_version_no_warning(caplog):
"""If user has a newer version than example (edge case), no warning."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = _make_config_files(
Path(tmpdir),
user_config={"config_version": 3},
example_config={"config_version": 2},
)
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version(
{"config_version": 3},
config_path,
)
assert "outdated" not in caplog.text
def test_version_26_config_upgrades_to_checkpoint_channel_mode(tmp_path, caplog):
"""A v26 user config must be flagged outdated and merge the new persisted field.
`database.checkpoint_channel_mode` shipped with config_version 27; the
upgrade path must add it with the safe default (``full``) without touching
the user's existing database backend settings. Uses the repository's real
config.example.yaml and the real config-upgrade script.
"""
import subprocess
repo_root = Path(__file__).resolve().parents[2]
example_src = repo_root / "config.example.yaml"
example_data = yaml.safe_load(example_src.read_text(encoding="utf-8"))
expected_version = example_data["config_version"]
assert expected_version > 26, "config.example.yaml must be bumped past 26 for checkpoint_channel_mode"
config_path = tmp_path / "config.yaml"
(tmp_path / "config.example.yaml").write_text(example_src.read_text(encoding="utf-8"), encoding="utf-8")
user_config = {
"config_version": 26,
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"database": {"backend": "sqlite", "sqlite_dir": "custom-data"},
}
config_path.write_text(yaml.dump(user_config), encoding="utf-8")
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version(dict(user_config), config_path)
assert "outdated" in caplog.text
assert "(version 26)" in caplog.text
env = {**os.environ, "DEER_FLOW_CONFIG_PATH": str(config_path)}
result = subprocess.run(
["bash", str(repo_root / "scripts" / "config-upgrade.sh")],
env=env,
capture_output=True,
text=True,
timeout=120,
)
assert result.returncode == 0, result.stderr
upgraded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
assert upgraded["config_version"] == expected_version
assert upgraded["database"]["checkpoint_channel_mode"] == "full"
assert upgraded["database"]["backend"] == "sqlite"
assert upgraded["database"]["sqlite_dir"] == "custom-data"
def _load_repo_example() -> dict:
"""Load the real repo config.example.yaml (first-run template)."""
example_path = Path(__file__).resolve().parents[2] / "config.example.yaml"
with open(example_path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
def _merge_missing(target: dict, source: dict) -> None:
"""Add-missing-keys-only recursive merge mirroring scripts/config-upgrade.sh."""
for key, value in source.items():
if key not in target:
import copy
target[key] = copy.deepcopy(value)
elif isinstance(value, dict) and isinstance(target[key], dict):
_merge_missing(target[key], value)
def test_security_fail_closed_bumped_config_version():
"""The example must ship security_fail_closed under a version > 26 so v26 configs upgrade."""
example = _load_repo_example()
assert example.get("config_version", 0) >= 27
assert example["skill_evolution"]["security_fail_closed"] is True
def test_version_26_config_reported_outdated_against_example(caplog):
"""A version-26 user config is flagged outdated against the real example version."""
example = _load_repo_example()
example_version = example["config_version"]
with tempfile.TemporaryDirectory() as tmpdir:
config_path = _make_config_files(
Path(tmpdir),
user_config={"config_version": 26},
example_config=example,
)
with caplog.at_level(logging.WARNING, logger="deerflow.config.app_config"):
AppConfig._check_config_version({"config_version": 26}, config_path)
assert "outdated" in caplog.text
assert "version 26" in caplog.text
assert f"version is {example_version}" in caplog.text
def test_config_upgrade_adds_security_fail_closed_preserving_user_values():
"""config-upgrade merges security_fail_closed: true without touching existing skill_evolution values."""
example = _load_repo_example()
# A version-26 user who customized skill_evolution but predates the new field.
user = {
"config_version": 26,
"skill_evolution": {
"enabled": True,
"moderation_model_name": "custom-moderation-model",
},
}
_merge_missing(user, example)
user["config_version"] = example["config_version"]
# New persisted field is merged in with the example's fail-closed default.
assert user["skill_evolution"]["security_fail_closed"] is True
# The user's existing skill_evolution values are preserved unchanged.
assert user["skill_evolution"]["enabled"] is True
assert user["skill_evolution"]["moderation_model_name"] == "custom-moderation-model"
assert user["config_version"] == example["config_version"]