github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-28 17:06:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(checkpoint): make delta snapshot_frequency configurable (#4516)

Browse Source 
* feat(checkpoint): make delta snapshot_frequency configurable

* fix(config): carry legacy checkpoint_delta_snapshot_frequency with warning

Addresses review on #4516: the rename from the flat
database.checkpoint_delta_snapshot_frequency key to nested
database.checkpoint_delta.snapshot_frequency silently dropped the old
value (pydantic extra="ignore"). Add a before-validator that maps the
legacy key onto the nested one with a deprecation warning (nested key
wins when both are set), plus a CHANGELOG breaking-change note covering
the rename and the 1000 -> 10 default change.

* fix(checkpoint): validate frozen snapshot frequency
This commit is contained in:
Vanzeren 2026-07-28 23:21:23 +08:00 committed by GitHub
parent 9fa4debae1
commit c48de5e70b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
31 changed files with 731 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGELOG.md
View File

@ -62,6 +62,13 @@ This section accumulates work toward the **2.1.0** milestone
intentional -- silent empties are worse than a loud startup error. Fix: switch
to `mode: middleware` or override `search()` (and set `supports_search=True`).
([#4324])
- **config:** `database.checkpoint_delta_snapshot_frequency` moved to
`database.checkpoint_delta.snapshot_frequency` and its default changed from
`1000` to `10`. A legacy top-level value is still honored with a deprecation
warning and mapped onto the nested key (an explicitly set nested key wins).
Deployments that relied on the old default now snapshot 100x more often in
delta mode -- set `database.checkpoint_delta.snapshot_frequency: 1000`
explicitly to keep the previous cadence. ([#4516])
### Added
@ -1343,3 +1350,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
[#4468]: https://github.com/bytedance/deer-flow/pull/4468
[#4469]: https://github.com/bytedance/deer-flow/pull/4469
[#4471]: https://github.com/bytedance/deer-flow/pull/4471
[#4516]: https://github.com/bytedance/deer-flow/pull/4516

6
README.md
View File

@ -256,9 +256,9 @@ honor environment proxy settings.
Backend processes automatically pick up `config.yaml` changes on the next config access, so model metadata updates do not require a manual restart during development.
The checkpoint storage settings `database.checkpoint_channel_mode` and
`database.checkpoint_delta_snapshot_frequency` are exceptions: both are frozen
when the process first builds an agent (including through `DeerFlowClient`) and
require a process restart to change safely.
`database.checkpoint_delta.snapshot_frequency` (default `10`) are exceptions:
both are frozen when the process first builds an agent (including through
`DeerFlowClient`) and require a process restart to change safely.
> [!TIP]
> On Linux, if Docker-based commands fail with `permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock`, add your user to the `docker` group and re-login before retrying. See [CONTRIBUTING.md](CONTRIBUTING.md#linux-docker-daemon-permission-denied) for the full fix.

19
backend/AGENTS.md
View File

@ -943,7 +943,11 @@ This invokes `alembic revision --autogenerate` against the live ORM models. Revi
Checkpointer storage runs in one of two channel modes, selected by `checkpoint_channel_mode` in `config.yaml` (default `full`). `delta` mode adopts LangGraph 1.2's `DeltaChannel` for `messages`: checkpoints store a sentinel + per-step writes instead of the full message list, so storage/serde grows O(N) instead of O(N²) in turns. All checkpointer backends (memory/sqlite/postgres) serve both modes unchanged — the semantics live in the compiled graph's channel table, not in the saver.
**Mode is process-frozen and restart-required.** `make_lead_agent` and the embedded `DeerFlowClient` freeze the resolved mode (`runtime/checkpoint_mode.py::freeze_checkpoint_channel_mode`) and the delta snapshot frequency (`agents/thread_state.py::freeze_delta_snapshot_frequency`, from the `checkpoint_delta_snapshot_frequency` config knob, default 1000 — restart-required like the mode) before compiling the graph with the mode-matched schema (`agents/thread_state.py::get_thread_state_schema`, plus `adapt_state_schema_for_mode` / `normalize_middleware_state_schemas` for middleware state). Adapted middleware schemas are cached by schema, mode, and resolved snapshot frequency so a pre-freeze ephemeral graph cannot leave a stale default-frequency schema behind. A second, different mode or frequency in the same process raises `CheckpointModeReconfigurationError`. To switch: edit config, restart.
**Mode is process-frozen and restart-required.** `make_lead_agent` and the embedded `DeerFlowClient` freeze the resolved mode (`runtime/checkpoint_mode.py::freeze_checkpoint_channel_mode`) before compiling the graph with the mode-matched schema (`agents/thread_state.py::get_thread_state_schema`, plus `adapt_state_schema_for_mode` / `normalize_middleware_state_schemas` for middleware state). Adapted middleware schemas are cached by schema, mode, and resolved snapshot frequency so a pre-freeze ephemeral graph cannot leave a stale default-frequency schema behind. A second, different mode or frequency in the same process raises `CheckpointModeReconfigurationError`. To switch: edit config, restart.
**Delta snapshot cadence is configurable but frozen with the mode.** `database.checkpoint_delta.snapshot_frequency` (default `10`) sets the `DeltaChannel` snapshot cadence. It is frozen alongside the mode (`freeze_checkpoint_snapshot_frequency`; non-positive direct inputs raise `ValueError`, while a frozen-value mismatch raises `CheckpointModeReconfigurationError`), restart-required, and must match across every process sharing one checkpoint database — the cadence lives in each compiled graph's channel table and is deliberately NOT stamped into checkpoint metadata, so the mode-compatibility marker and full -> delta migration semantics are unchanged. Schema helpers resolve it explicit-arg -> frozen -> default, and every schema/graph cache (`_delta_thread_state_schema`, `_adapt_state_schema_for_delta`, the client agent-config key, the gateway accessor-graph cache) keys on the resolved value.
**Compiled-graph cache cap is configurable and hot-reloadable.** `database.checkpoint_graph_cache.accessor_graph_max` (default `64`) bounds the gateway accessor-graph cache, which clears wholesale at the cap. The cap is re-read on every eviction check (`resolve_checkpoint_graph_cache_max`), so a config.yaml reload takes effect without a restart — a size change never affects graph semantics, only eviction timing.
**Compatibility is asymmetric and fail-closed.** Every checkpoint written in delta mode carries metadata marker `deerflow_checkpoint_channel_mode: "delta"` (injected via `inject_checkpoint_mode`; absence of marker = full, so pre-feature checkpoints need no migration). Before any state read/write, `ensure_checkpoint_mode_compatible` rejects a full-mode process opening a delta thread with `CheckpointModeMismatchError` (surfaced as HTTP 409 with the cause and thread id by the threads router; `CheckpointModeReconfigurationError` maps to 503) — a full-mode raw read of a delta blob would silently return empty/partial `messages`. The reverse direction is allowed: delta-mode processes read full checkpoints transparently (old full checkpoints seed the delta channel), so full → delta is the smooth migration path; delta → full requires materializing/converting the data first. Detection also honors upstream's `counters_since_delta_snapshot.messages` metadata, and an explicit config marker takes precedence over any ambient context value.
@ -958,10 +962,10 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
**Run rollback flow** (`runtime/runs/worker.py`): `_capture_rollback_point` materializes the complete pre-run state via the accessor and captures raw `pending_writes` via `aget_tuple` into an immutable `RollbackPoint` before the run starts — capture failure disables rollback (fail-closed), never restores partial state. In `full` mode, cancel-with-rollback forks from the pre-run checkpoint via the mutation graph and inherits non-message channels from that parent. In `delta` mode, forking is unsafe once the cancelled path has attached sibling writes to the pre-run checkpoint, so rollback replaces every captured channel on the current head, using `Overwrite` for reducers and schema defaults for current-head-only channels. Both modes reattach only the captured pre-run pending writes to the restored checkpoint. Edit replay runs (`metadata.replay_kind="edit"`) also restore the pre-run checkpoint on failed, timed-out, or interrupted completion and publish the restored `values` snapshot to the stream before `end`, so clients do not remain on a transient edited branch when the replay did not produce a successful replacement.
**Where things live**:
- `runtime/checkpoint_mode.py` — mode freeze, marker injection, delta detection, compatibility gate, both error types
- `runtime/checkpoint_mode.py` — mode + snapshot-frequency freeze, marker injection, delta detection, compatibility gate, both error types
- `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint`
- `checkpoint_patches.py` (package root) — checkpoint-machinery patches: delta-history folding for `InMemorySaver` (delegating to the base walk), stable message IDs across materialization, upstream first-write drop fix, and `BinaryOperatorAggregate` unwrapping an `Overwrite` first write into an empty (MISSING) channel — Union-typed reducer channels (`sandbox`/`goal`/`todos`/`promoted`) have no constructible default, so a replace-style write into a fresh branch thread or a never-written channel stored the wrapper literally and crashed the next consumer (#4380; probe-guarded, stands down if upstream fixes it)
- `agents/thread_state.py` — `ThreadState`/`DeltaThreadState`, `DELTA_MESSAGES_FIELD` (`DeltaChannel` whose `snapshot_frequency` resolves from the `checkpoint_delta_snapshot_frequency` config knob via `freeze_delta_snapshot_frequency`/`resolved_delta_snapshot_frequency`; 1000 is only the default), schema adaptation helpers
- `agents/thread_state.py` — `ThreadState`/`DeltaThreadState`, `delta_messages_field` / `DELTA_MESSAGES_FIELD` (`DeltaChannel` at the configured `snapshot_frequency`, default 10), schema adaptation helpers
- `runtime/context_compaction.py` — compaction via accessor + mutation graph (reference consumer)
- Tests: `tests/test_checkpoint_mode.py` (freeze/detect/gate), `tests/test_checkpoint_state.py` (accessor/mutation graph), `tests/test_delta_channel_checkpointers.py` (saver parity), `tests/test_threads_checkpoint_mode.py`, `tests/test_gateway_checkpoint_mode.py` (dual-mode e2e parity), `tests/test_context_compaction.py` (mutation-graph write, no scheduling), `tests/test_run_worker_rollback.py`
@ -974,8 +978,9 @@ logical checkpoint/write bytes, SQLite DB/WAL/SHM footprint, reducer replay time
and peak RSS as versioned JSONL. The controller alternates mode order and rejects
performance data when paired modes materialize different state. Its default 1 GiB
estimated cumulative full-payload cap skips both modes of an oversized pair when
`full` is selected; intentional `--modes delta` diagnostics bypass this
full-payload cap, so size those runs explicitly. Use `--allow-large-cases` only
`full` is selected, including every delta cadence in a `--snapshot-frequencies`
sweep; intentional `--modes delta` diagnostics bypass this full-payload cap, so
size those runs explicitly. Use `--allow-large-cases` only
on a provisioned machine. Duplicate CSV matrix values are ignored with a warning;
use `--repetitions` for repeated samples. Summarize paired successful repetitions
with `scripts/benchmark/checkpoint/summarize_channels.py` (all ratios are
@ -1003,8 +1008,8 @@ deterministic model, real `AsyncSqliteSaver`), then measure
`GET /threads/{id}/state` and `POST /threads/{id}/history` through the real
Gateway route stack in the same event loop (httpx ASGITransport), split into
cold/warm accessor-graph-cache samples. It sweeps `snapshot_frequency`
(config: `checkpoint_delta_snapshot_frequency`, process-frozen like the
mode), pairs every delta frequency against the same full row, and fails both
(config: `checkpoint_delta.snapshot_frequency`, process-frozen like the mode),
pairs every delta frequency against the same full row, and fails both
rows of a pair when materialized or wire digests diverge. Each case must have
more than the two discarded warm-up turns, and SQLite DB/WAL/SHM sizes are
captured while the saver is still open so they represent the online storage

4
backend/app/gateway/deps.py
View File

@ -371,7 +371,7 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
"""
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
from deerflow.runtime import make_store, make_stream_bridge
from deerflow.runtime.checkpoint_mode import freeze_checkpoint_channel_mode
from deerflow.runtime.checkpoint_mode import freeze_checkpoint_channel_mode, freeze_checkpoint_snapshot_frequency
from deerflow.runtime.checkpointer.async_provider import make_checkpointer
from deerflow.runtime.events.store import make_run_event_store
@ -387,6 +387,7 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
async with AsyncExitStack() as stack:
config = startup_config
app.state.checkpoint_channel_mode = freeze_checkpoint_channel_mode(config.database.checkpoint_channel_mode)
app.state.checkpoint_snapshot_frequency = freeze_checkpoint_snapshot_frequency(config.database.checkpoint_delta.snapshot_frequency)
app.state.stream_bridge = await stack.enter_async_context(make_stream_bridge(config))
@ -592,6 +593,7 @@ def get_run_context(request: Request) -> RunContext:
event_store=get_run_event_store(request),
run_events_config=getattr(request.app.state, "run_events_config", None),
checkpoint_channel_mode=getattr(request.app.state, "checkpoint_channel_mode", "full"),
checkpoint_snapshot_frequency=getattr(request.app.state, "checkpoint_snapshot_frequency", None),
thread_store=get_thread_store(request),
app_config=get_config(),
on_run_completed=getattr(request.app.state, "scheduled_task_service", None).handle_run_completion if getattr(request.app.state, "scheduled_task_service", None) is not None else None,

27
backend/app/gateway/services.py
View File

@ -34,6 +34,7 @@ from app.gateway.utils import sanitize_log_param
from deerflow.agents.middlewares.dynamic_context_middleware import _DYNAMIC_CONTEXT_REMINDER_KEY, _REMINDER_DATE_KEY
from deerflow.agents.middlewares.view_image_middleware import _IMAGE_CONTEXT_MESSAGE_MARKER_KEY
from deerflow.config.app_config import get_app_config
from deerflow.config.database_config import resolve_checkpoint_graph_cache_max
from deerflow.runtime import (
END_SENTINEL,
HEARTBEAT_SENTINEL,
@ -690,23 +691,35 @@ def build_checkpoint_state_mutation_accessor(
# Cache of factory-built accessor graphs. Accessor operations (aget_state /
# aupdate_state) never execute graph nodes or middleware, so per-request
# variations (user, model, skills) cannot affect materialization semantics;
# the compiled graph is stable per (assistant_id, mode, app_config). The
# the compiled graph is stable per (assistant_id, mode, snapshot_frequency,
# app_config). The
# factory and app_config identities are re-validated on every call so patched
# factories take effect immediately and a config.yaml hot-reload (which
# rebuilds the AppConfig object) never serves a stale compiled graph — the
# cached reference keeps the old config alive, so id-reuse cannot produce a
# false hit. Bounded: cleared when too many distinct assistants appear.
# false hit. Bounded: cleared when too many distinct assistants appear. The
# cap is configurable (database.checkpoint_graph_cache.accessor_graph_max)
# and re-read on every eviction check, so a hot-reload takes effect without
# a restart.
_STATE_ACCESSOR_GRAPH_CACHE_MAX = 64
_state_accessor_graph_cache: dict[tuple[str | None, str], tuple[Any, Any, Any]] = {}
_state_accessor_graph_cache: dict[tuple[str | None, str, int | None], tuple[Any, Any, Any]] = {}
def _state_accessor_graph(agent_factory: Any, assistant_id: str | None, mode: str, config: dict[str, Any]) -> Any:
def _accessor_graph_cache_max(app_config: Any) -> int:
return resolve_checkpoint_graph_cache_max(
getattr(app_config, "database", None),
"accessor_graph_max",
_STATE_ACCESSOR_GRAPH_CACHE_MAX,
)
def _state_accessor_graph(agent_factory: Any, assistant_id: str | None, mode: str, snapshot_frequency: int | None, config: dict[str, Any]) -> Any:
app_config = (config.get("context") or {}).get("app_config")
key = (assistant_id, mode)
key = (assistant_id, mode, snapshot_frequency)
cached = _state_accessor_graph_cache.get(key)
if cached is not None and cached[0] is agent_factory and cached[1] is app_config:
return cached[2]
if len(_state_accessor_graph_cache) >= _STATE_ACCESSOR_GRAPH_CACHE_MAX:
if len(_state_accessor_graph_cache) >= _accessor_graph_cache_max(app_config):
_state_accessor_graph_cache.clear()
graph = agent_factory(config=config)
_state_accessor_graph_cache[key] = (agent_factory, app_config, graph)
@ -811,7 +824,7 @@ def build_checkpoint_state_accessor(
agent_factory = resolve_agent_factory(assistant_id)
try:
graph = _state_accessor_graph(agent_factory, assistant_id, ctx.checkpoint_channel_mode, config)
graph = _state_accessor_graph(agent_factory, assistant_id, ctx.checkpoint_channel_mode, getattr(ctx, "checkpoint_snapshot_frequency", None), config)
except Exception:
if ctx.checkpoint_channel_mode != "full":
# Delta materialization needs the graph's channel table; there is

8
backend/packages/harness/deerflow/agents/factory.py
View File

@ -72,6 +72,7 @@ def create_deerflow_agent(
plan_mode: bool = False,
state_schema: type | None = None,
checkpoint_channel_mode: CheckpointChannelMode = "full",
checkpoint_snapshot_frequency: int | None = None,
checkpointer: BaseCheckpointSaver | None = None,
name: str = "default",
) -> CompiledStateGraph:
@ -107,6 +108,10 @@ def create_deerflow_agent(
persistence paths (mode markers + compatibility gate) and is therefore
rejected when combined with *checkpointer* in this factory; without a
checkpointer the graph is ephemeral and delta is allowed.
checkpoint_snapshot_frequency:
DeltaChannel snapshot cadence for ``"delta"`` mode. ``None`` uses the
process-frozen value, falling back to the config default. Ignored in
``"full"`` mode.
checkpointer:
Optional persistence backend.
name:
@ -135,7 +140,7 @@ def create_deerflow_agent(
raise TypeError(f"extra_middleware items must be AgentMiddleware instances, got {type(mw).__name__}")
effective_tools: list[BaseTool] = list(tools or [])
effective_state = get_thread_state_schema(checkpoint_channel_mode) if state_schema is None else adapt_state_schema_for_mode(state_schema, checkpoint_channel_mode)
effective_state = get_thread_state_schema(checkpoint_channel_mode, checkpoint_snapshot_frequency) if state_schema is None else adapt_state_schema_for_mode(state_schema, checkpoint_channel_mode, checkpoint_snapshot_frequency)
if middleware is not None:
effective_middleware = list(middleware)
@ -157,6 +162,7 @@ def create_deerflow_agent(
effective_middleware = normalize_middleware_state_schemas(
effective_middleware,
checkpoint_channel_mode,
checkpoint_snapshot_frequency,
)
return create_agent(

8
backend/packages/harness/deerflow/agents/lead_agent/agent.py
View File

@ -46,7 +46,7 @@ from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
from deerflow.agents.middlewares.token_usage_middleware import TokenUsageMiddleware
from deerflow.agents.middlewares.tool_error_handling_middleware import build_lead_runtime_middlewares
from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
from deerflow.agents.thread_state import freeze_delta_snapshot_frequency, get_thread_state_schema, normalize_middleware_state_schemas
from deerflow.agents.thread_state import get_thread_state_schema, normalize_middleware_state_schemas
from deerflow.authz.tool_filter import apply_tool_authorization
from deerflow.config.agents_config import load_agent_config, validate_agent_name
from deerflow.config.app_config import AppConfig, get_app_config
@ -56,6 +56,7 @@ from deerflow.models import create_chat_model
from deerflow.runtime.checkpoint_mode import (
INTERNAL_CHECKPOINT_MODE_KEY,
freeze_checkpoint_channel_mode,
freeze_checkpoint_snapshot_frequency,
frozen_checkpoint_channel_mode,
inject_checkpoint_mode,
)
@ -518,7 +519,10 @@ def make_lead_agent(config: RunnableConfig):
runtime_app_config.database.checkpoint_channel_mode,
)
mode = freeze_checkpoint_channel_mode(requested_mode)
freeze_delta_snapshot_frequency(runtime_app_config.database.checkpoint_delta_snapshot_frequency)
# The snapshot cadence travels with the mode: restart-required, frozen
# from the app config, and deliberately not client-injectable (a forged
# configurable key must not recompile the channel table either).
freeze_checkpoint_snapshot_frequency(runtime_app_config.database.checkpoint_delta.snapshot_frequency)
inject_checkpoint_mode(config, mode)
return _make_lead_agent(config, app_config=runtime_app_config)

101
backend/packages/harness/deerflow/agents/thread_state.py
View File

@ -17,10 +17,21 @@ from langgraph.graph.message import REMOVE_ALL_MESSAGES
import deerflow.checkpoint_patches as _checkpoint_patches # noqa: F401 - import-time saver fixes
from deerflow.agents.goal_state import GoalState
from deerflow.config.database_config import CheckpointChannelMode
from deerflow.config.database_config import DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY, CheckpointChannelMode
from deerflow.subagents.status_contract import SUBAGENT_STATUS_VALUES
def _resolve_snapshot_frequency(snapshot_frequency: int | None) -> int:
"""Resolve the effective cadence: explicit value, else process-frozen,
else default. Imported lazily — ``deerflow.runtime.__init__`` reaches this
module via ``checkpoint_state``, so a top-level import would cycle."""
if snapshot_frequency is not None:
return snapshot_frequency
from deerflow.runtime.checkpoint_mode import resolve_checkpoint_snapshot_frequency
return resolve_checkpoint_snapshot_frequency()
class SandboxState(TypedDict):
sandbox_id: NotRequired[str | None]
@ -352,61 +363,21 @@ def merge_message_writes(state: list[AnyMessage], writes: Sequence[Any]) -> list
return [message for message in messages if message is not None]
DEFAULT_DELTA_SNAPSHOT_FREQUENCY = 1000
_frozen_delta_snapshot_frequency: int | None = None
def delta_messages_field(snapshot_frequency: int) -> Any:
def delta_messages_field(snapshot_frequency: int = DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY) -> Any:
"""Messages field annotation with a ``DeltaChannel`` at the given cadence."""
return Annotated[
list[AnyMessage],
DeltaChannel(merge_message_writes, snapshot_frequency=snapshot_frequency),
]
DELTA_MESSAGES_FIELD = delta_messages_field(DEFAULT_DELTA_SNAPSHOT_FREQUENCY)
DELTA_MESSAGES_FIELD = delta_messages_field()
class DeltaThreadState(ThreadState):
messages: DELTA_MESSAGES_FIELD
def freeze_delta_snapshot_frequency(frequency: int) -> int:
"""Freeze the process-wide delta snapshot frequency (restart-required)."""
# Lazy import: deerflow.runtime.__init__ imports checkpoint_state, which
# imports this module — a top-level import would be circular.
from deerflow.runtime.checkpoint_mode import CheckpointModeReconfigurationError
global _frozen_delta_snapshot_frequency
if frequency <= 0:
raise ValueError("snapshot frequency must be positive")
if _frozen_delta_snapshot_frequency is None:
_frozen_delta_snapshot_frequency = frequency
elif _frozen_delta_snapshot_frequency != frequency:
raise CheckpointModeReconfigurationError(f"checkpoint delta snapshot frequency is frozen at {_frozen_delta_snapshot_frequency}; refusing to reconfigure to {frequency} — restart the process to change it")
return _frozen_delta_snapshot_frequency
def resolved_delta_snapshot_frequency() -> int:
return _frozen_delta_snapshot_frequency or DEFAULT_DELTA_SNAPSHOT_FREQUENCY
def reset_delta_snapshot_frequency_for_tests() -> None:
global _frozen_delta_snapshot_frequency
_frozen_delta_snapshot_frequency = None
@cache
def _delta_thread_state_for_frequency(snapshot_frequency: int) -> type:
if snapshot_frequency == DEFAULT_DELTA_SNAPSHOT_FREQUENCY:
# Preserve the canonical class so identity checks (and its public
# re-export) keep holding at the default frequency.
return DeltaThreadState
annotations = get_type_hints(ThreadState, include_extras=True)
annotations["messages"] = delta_messages_field(snapshot_frequency)
return TypedDict(f"DeltaThreadState_f{snapshot_frequency}", annotations, total=True)
THREAD_STATE_REDUCER_FIELDS = frozenset(
{
"messages",
@ -422,20 +393,35 @@ THREAD_STATE_REDUCER_FIELDS = frozenset(
)
def get_thread_state_schema(mode: CheckpointChannelMode) -> type:
if mode == "delta":
return _delta_thread_state_for_frequency(resolved_delta_snapshot_frequency())
return ThreadState
def adapt_state_schema_for_mode(schema: type, mode: CheckpointChannelMode) -> type:
if mode == "full":
return schema
return _adapt_state_schema_for_mode(schema, mode, resolved_delta_snapshot_frequency())
def get_thread_state_schema(mode: CheckpointChannelMode, snapshot_frequency: int | None = None) -> type:
if mode != "delta":
return ThreadState
return _delta_thread_state_schema(_resolve_snapshot_frequency(snapshot_frequency))
@cache
def _adapt_state_schema_for_mode(schema: type, mode: CheckpointChannelMode, snapshot_frequency: int) -> type:
def _delta_thread_state_schema(snapshot_frequency: int) -> type:
"""Delta thread schema keyed by cadence; the default keeps the static
``DeltaThreadState`` identity so existing type checks keep holding."""
if snapshot_frequency == DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY:
return DeltaThreadState
annotations = get_type_hints(ThreadState, include_extras=True)
annotations["messages"] = delta_messages_field(snapshot_frequency)
return TypedDict(
f"DeltaThreadState_f{snapshot_frequency}",
annotations,
total=getattr(ThreadState, "__total__", True),
)
def adapt_state_schema_for_mode(schema: type, mode: CheckpointChannelMode, snapshot_frequency: int | None = None) -> type:
if mode == "full":
return schema
return _adapt_state_schema_for_delta(schema, _resolve_snapshot_frequency(snapshot_frequency))
@cache
def _adapt_state_schema_for_delta(schema: type, snapshot_frequency: int) -> type:
annotations = get_type_hints(schema, include_extras=True)
annotations["messages"] = delta_messages_field(snapshot_frequency)
return TypedDict(
@ -445,9 +431,10 @@ def _adapt_state_schema_for_mode(schema: type, mode: CheckpointChannelMode, snap
)
def normalize_middleware_state_schemas(middleware: Sequence[Any], mode: CheckpointChannelMode) -> list[Any]:
def normalize_middleware_state_schemas(middleware: Sequence[Any], mode: CheckpointChannelMode, snapshot_frequency: int | None = None) -> list[Any]:
if mode == "full":
return list(middleware)
resolved_frequency = _resolve_snapshot_frequency(snapshot_frequency)
normalized = []
for item in middleware:
schema = getattr(item, "state_schema", None)
@ -455,6 +442,6 @@ def normalize_middleware_state_schemas(middleware: Sequence[Any], mode: Checkpoi
normalized.append(item)
continue
adapted = copy.copy(item)
adapted.state_schema = adapt_state_schema_for_mode(schema, mode)
adapted.state_schema = adapt_state_schema_for_mode(schema, mode, resolved_frequency)
normalized.append(adapted)
return normalized

9
backend/packages/harness/deerflow/client.py
View File

@ -37,7 +37,7 @@ from langchain_core.runnables import RunnableConfig
from deerflow.agents.lead_agent.agent import build_middlewares
from deerflow.agents.lead_agent.prompt import apply_prompt_template, get_enabled_skills_for_config
from deerflow.agents.thread_state import freeze_delta_snapshot_frequency, get_thread_state_schema, normalize_middleware_state_schemas
from deerflow.agents.thread_state import get_thread_state_schema, normalize_middleware_state_schemas
from deerflow.authz.principal import build_principal_from_context
from deerflow.config.agents_config import AGENT_NAME_PATTERN
from deerflow.config.app_config import get_app_config, is_trace_correlation_enabled, reload_app_config
@ -48,6 +48,7 @@ from deerflow.runtime import CheckpointStateAccessor
from deerflow.runtime.checkpoint_mode import (
ensure_checkpoint_mode_compatible,
freeze_checkpoint_channel_mode,
freeze_checkpoint_snapshot_frequency,
inject_checkpoint_mode,
)
from deerflow.runtime.goal import DEFAULT_MAX_GOAL_CONTINUATIONS, build_goal_state, goal_thread_lock, read_thread_goal, write_thread_goal
@ -192,7 +193,7 @@ class DeerFlowClient:
reload_app_config(config_path)
self._app_config = get_app_config()
self._checkpoint_channel_mode = freeze_checkpoint_channel_mode(self._app_config.database.checkpoint_channel_mode)
freeze_delta_snapshot_frequency(self._app_config.database.checkpoint_delta_snapshot_frequency)
self._checkpoint_snapshot_frequency = freeze_checkpoint_snapshot_frequency(self._app_config.database.checkpoint_delta.snapshot_frequency)
if agent_name is not None and not AGENT_NAME_PATTERN.match(agent_name):
raise ValueError(f"Invalid agent name '{agent_name}'. Must match pattern: {AGENT_NAME_PATTERN.pattern}")
@ -288,6 +289,7 @@ class DeerFlowClient:
self._agent_name,
frozenset(self._available_skills) if self._available_skills is not None else None,
self._checkpoint_channel_mode,
self._checkpoint_snapshot_frequency,
authorization_identity,
)
@ -359,6 +361,7 @@ class DeerFlowClient:
authorization_provider=_authz_provider,
),
self._checkpoint_channel_mode,
self._checkpoint_snapshot_frequency,
),
"system_prompt": apply_prompt_template(
subagent_enabled=subagent_enabled,
@ -372,7 +375,7 @@ class DeerFlowClient:
user_id=effective_user_id,
skill_names=skill_setup.skill_names or None,
),
"state_schema": get_thread_state_schema(self._checkpoint_channel_mode),
"state_schema": get_thread_state_schema(self._checkpoint_channel_mode, self._checkpoint_snapshot_frequency),
}
checkpointer = self._checkpointer
if checkpointer is None:

119
backend/packages/harness/deerflow/config/database_config.py
View File

@ -31,13 +31,72 @@ need to do any environment variable processing.
from __future__ import annotations
import logging
import os
from typing import Literal
from typing import Any, Literal
from pydantic import BaseModel, Field, model_validator
logger = logging.getLogger(__name__)
def resolve_checkpoint_graph_cache_max(database_config: Any, field_name: str, default: int) -> int:
"""Read a graph-cache cap from a database config-ish object.
Tolerates stub configs (SimpleNamespace/MagicMock in tests, missing
section): anything that is not a plain int >= 1 falls back to
``default``.
"""
section = getattr(database_config, "checkpoint_graph_cache", None)
value = getattr(section, field_name, None)
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
return default
return value
from pydantic import BaseModel, Field
CheckpointChannelMode = Literal["full", "delta"]
DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY = 10
class CheckpointDeltaConfig(BaseModel):
"""Tuning knobs for ``checkpoint_channel_mode: delta``.
Ignored in ``full`` mode. Like the mode itself, these values are
restart-required and must match across every process sharing one
checkpoint database: the snapshot cadence is baked into each compiled
graph's channel table, not stored in the checkpoint, so a mismatched
process would apply a different cadence to the same threads.
"""
snapshot_frequency: int = Field(
default=DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY,
ge=1,
description=(
"DeltaChannel snapshot cadence: a full messages snapshot is stored "
"every N per-step writes (higher = smaller checkpoints, slower "
"materialization). Restart is required, and all processes sharing "
"one checkpoint database must use the same value."
),
)
class CheckpointGraphCacheConfig(BaseModel):
"""Size cap for the process-local compiled checkpoint graph cache.
Unlike the mode and snapshot cadence, this is NOT restart-required:
a larger/smaller cap only changes when the cache evicts, never graph
semantics, so a hot-reloaded value takes effect on the next eviction
check. The cache is keyed by (assistant, mode, cadence, app_config) and
cleared wholesale at the cap.
"""
accessor_graph_max: int = Field(
default=64,
ge=1,
description=("Max compiled thread-state accessor graphs cached by the gateway (keyed per assistant, channel mode, and snapshot cadence)."),
)
class DatabaseConfig(BaseModel):
backend: Literal["memory", "sqlite", "postgres"] = Field(
@ -53,15 +112,13 @@ class DatabaseConfig(BaseModel):
"processes sharing one checkpoint database must use the same value."
),
)
checkpoint_delta_snapshot_frequency: int = Field(
default=1000,
ge=1,
description=(
"DeltaChannel snapshot frequency used in checkpoint 'delta' mode: "
"every N message-channel writes the saver stores a full snapshot "
"instead of a delta sentinel. Restart-required, and all processes "
"sharing one checkpoint database must use the same value."
),
checkpoint_delta: CheckpointDeltaConfig = Field(
default_factory=CheckpointDeltaConfig,
description="Delta-mode checkpoint tuning. Only applies when checkpoint_channel_mode is 'delta'.",
)
checkpoint_graph_cache: CheckpointGraphCacheConfig = Field(
default_factory=CheckpointGraphCacheConfig,
description="Size caps for the compiled checkpoint graph caches. Hot-reloadable; not restart-required.",
)
sqlite_dir: str = Field(
default=".deer-flow/data",
@ -95,6 +152,46 @@ class DatabaseConfig(BaseModel):
description="Timeout in seconds for app ORM PostgreSQL commands. Set to null to disable the command timeout.",
)
# -- Legacy key migration (not user-configured) --
@model_validator(mode="before")
@classmethod
def _migrate_legacy_snapshot_frequency(cls, data: Any) -> Any:
"""Carry the pre-rename top-level ``checkpoint_delta_snapshot_frequency``
key onto ``checkpoint_delta.snapshot_frequency``.
``DatabaseConfig`` ignores unknown keys (pydantic ``extra="ignore"``),
so without this shim a config.yaml written against the old flat key
would silently fall back to the new default cadence instead of the
operator's chosen value. An explicitly set nested key always wins.
"""
if not isinstance(data, dict) or "checkpoint_delta_snapshot_frequency" not in data:
return data
data = dict(data)
legacy_value = data.pop("checkpoint_delta_snapshot_frequency")
nested = data.get("checkpoint_delta")
if isinstance(nested, dict):
if "snapshot_frequency" in nested:
logger.warning(
"Both database.checkpoint_delta_snapshot_frequency (deprecated) and database.checkpoint_delta.snapshot_frequency are set; the nested key wins.",
)
return data
data["checkpoint_delta"] = {**nested, "snapshot_frequency": legacy_value}
elif nested is None:
data["checkpoint_delta"] = {"snapshot_frequency": legacy_value}
else:
# Programmatically constructed CheckpointDeltaConfig instance: the
# explicit object wins over the legacy scalar.
logger.warning(
"Ignoring deprecated database.checkpoint_delta_snapshot_frequency because database.checkpoint_delta is already set.",
)
return data
logger.warning(
"database.checkpoint_delta_snapshot_frequency is deprecated; use database.checkpoint_delta.snapshot_frequency instead. Carried the legacy value (%r) forward.",
legacy_value,
)
return data
# -- Derived helpers (not user-configured) --
@property

36
backend/packages/harness/deerflow/runtime/checkpoint_mode.py
View File

@ -13,7 +13,7 @@ from __future__ import annotations
from typing import Any
from deerflow.config.database_config import CheckpointChannelMode
from deerflow.config.database_config import DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY, CheckpointChannelMode
INTERNAL_CHECKPOINT_MODE_KEY = "__deerflow_checkpoint_channel_mode"
CHECKPOINT_MODE_METADATA_KEY = "deerflow_checkpoint_channel_mode"
@ -28,6 +28,7 @@ class CheckpointModeReconfigurationError(RuntimeError):
_frozen_checkpoint_channel_mode: CheckpointChannelMode | None = None
_frozen_checkpoint_snapshot_frequency: int | None = None
def frozen_checkpoint_channel_mode() -> CheckpointChannelMode | None:
@ -44,6 +45,39 @@ def freeze_checkpoint_channel_mode(mode: CheckpointChannelMode) -> CheckpointCha
return _frozen_checkpoint_channel_mode
def frozen_checkpoint_snapshot_frequency() -> int | None:
"""Return the process-frozen delta snapshot frequency, if already frozen."""
return _frozen_checkpoint_snapshot_frequency
def freeze_checkpoint_snapshot_frequency(snapshot_frequency: int) -> int:
"""Freeze the delta snapshot cadence alongside the channel mode.
The cadence is compiled into each graph's channel table, so like the
mode it is restart-required and must match across every process sharing
one checkpoint database. It is deliberately NOT stamped into checkpoint
metadata: the mode marker contract (absence = full) and the full ->
delta migration semantics are unchanged by the frequency value.
"""
global _frozen_checkpoint_snapshot_frequency
if snapshot_frequency <= 0:
raise ValueError("snapshot frequency must be positive")
if _frozen_checkpoint_snapshot_frequency is None:
_frozen_checkpoint_snapshot_frequency = snapshot_frequency
elif _frozen_checkpoint_snapshot_frequency != snapshot_frequency:
raise CheckpointModeReconfigurationError("checkpoint_delta.snapshot_frequency is restart-required and cannot change in a running process")
return _frozen_checkpoint_snapshot_frequency
def resolve_checkpoint_snapshot_frequency(snapshot_frequency: int | None = None) -> int:
"""Resolve the effective snapshot frequency: explicit value, else the
process-frozen value, else the config default."""
if snapshot_frequency is not None:
return snapshot_frequency
frozen = _frozen_checkpoint_snapshot_frequency
return frozen if frozen is not None else DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY
def inject_checkpoint_mode(config: dict[str, Any], mode: CheckpointChannelMode) -> None:
configurable = config.setdefault("configurable", {})
configurable[INTERNAL_CHECKPOINT_MODE_KEY] = mode

13
backend/packages/harness/deerflow/runtime/checkpoint_state.py
View File

@ -33,7 +33,13 @@ def _finish_state_mutation(_state: dict[str, Any]) -> dict[str, Any]:
return {}
def build_state_mutation_graph(as_node: str, mode: CheckpointChannelMode, state_schema: Any | None = None) -> Any:
def build_state_mutation_graph(
as_node: str,
mode: CheckpointChannelMode,
state_schema: Any | None = None,
*,
snapshot_frequency: int | None = None,
) -> Any:
"""Compile a state-only graph whose single writer node finishes immediately.
``update_state(..., as_node=...)`` requires the node to be registered in
@ -45,12 +51,15 @@ def build_state_mutation_graph(as_node: str, mode: CheckpointChannelMode, state_
assistant graph was compiled with) whenever the write carries materialized
state: the base ThreadState fallback does not know channels contributed by
custom middleware, and writes to unknown channels are silently discarded.
The fallback resolves the delta snapshot cadence explicit arg ->
process-frozen -> config default; an explicit ``state_schema`` already
carries its cadence in its identity.
"""
if not as_node:
raise ValueError("as_node is required for checkpoint state mutation")
from langgraph.graph import StateGraph
builder = StateGraph(state_schema if state_schema is not None else get_thread_state_schema(mode))
builder = StateGraph(state_schema if state_schema is not None else get_thread_state_schema(mode, snapshot_frequency))
builder.add_node(as_node, _finish_state_mutation)
builder.set_entry_point(as_node)
builder.set_finish_point(as_node)

3
backend/packages/harness/deerflow/runtime/runs/worker.py
View File

@ -389,6 +389,9 @@ class RunContext:
thread_store: Any | None = field(default=None)
app_config: AppConfig | None = field(default=None)
checkpoint_channel_mode: CheckpointChannelMode = "full"
# Delta snapshot cadence frozen at startup; ``None`` means "not frozen in
# this process" (embedded/tests) and resolves to the config default.
checkpoint_snapshot_frequency: int | None = None
on_run_completed: Any | None = field(default=None)

87
backend/scripts/benchmark/checkpoint/bench_channels.py
View File

@ -17,9 +17,10 @@ Examples::
--backends sqlite --updates 1000 --payload-bytes 128 \
--repetitions 7 --output snapshot-boundary.jsonl
The controller suppresses matrix pairs whose estimated cumulative full-mode
message payload exceeds ``--max-estimated-full-bytes``. Both modes are skipped
as a pair so every emitted full/delta result remains comparable. Use
The controller suppresses matrix cells whose estimated cumulative full-mode
message payload exceeds ``--max-estimated-full-bytes``. Full mode and every
swept delta cadence are skipped together so every emitted result remains
comparable and subject to the same safety cap. Use
``--allow-large-cases`` only on a machine provisioned for the resulting disk
and memory use.
"""
@ -37,6 +38,7 @@ import tempfile
import time
from collections.abc import Callable
from dataclasses import asdict, dataclass
from functools import cache
from pathlib import Path
from typing import Annotated, Any, Literal, TypedDict
@ -48,6 +50,7 @@ from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from deerflow.agents.thread_state import merge_message_writes
from deerflow.config.database_config import DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY
from deerflow.runtime.checkpoint_mode import inject_checkpoint_mode
from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
@ -70,7 +73,7 @@ Backend = Literal["memory", "sqlite"]
SCHEMA_VERSION = 1
BENCHMARK_VERSION = 1
PRODUCTION_SNAPSHOT_FREQUENCY = 1000
PRODUCTION_SNAPSHOT_FREQUENCY = DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY
DEFAULT_MAX_ESTIMATED_FULL_BYTES = 1024**3
_MODES: tuple[Mode, ...] = ("full", "delta")
_BACKENDS: tuple[Backend, ...] = ("memory", "sqlite")
@ -86,11 +89,13 @@ class _FullBenchmarkState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
class _DeltaBenchmarkState(TypedDict):
messages: Annotated[
list[AnyMessage],
DeltaChannel(merge_message_writes, snapshot_frequency=PRODUCTION_SNAPSHOT_FREQUENCY),
]
@cache
def _delta_benchmark_state(snapshot_frequency: int) -> type:
"""Delta benchmark schema for one snapshot cadence (cached per value)."""
return TypedDict(
f"_DeltaBenchmarkState_f{snapshot_frequency}",
{"messages": Annotated[list[AnyMessage], DeltaChannel(merge_message_writes, snapshot_frequency=snapshot_frequency)]},
)
@dataclass(frozen=True)
@ -115,8 +120,8 @@ class BenchmarkCase:
raise ValueError("payload_bytes must be positive")
if self.repetition < 0:
raise ValueError("repetition must be non-negative")
if self.snapshot_frequency != PRODUCTION_SNAPSHOT_FREQUENCY:
raise ValueError(f"Phase 1 supports only production snapshot_frequency={PRODUCTION_SNAPSHOT_FREQUENCY}")
if self.snapshot_frequency <= 0:
raise ValueError("snapshot_frequency must be positive")
if self.scenario != "append":
raise ValueError(f"unsupported scenario: {self.scenario!r}")
@ -129,8 +134,15 @@ def _expand_cases(
payload_bytes: list[int],
repetitions: int,
seed: int,
snapshot_frequencies: list[int] | None = None,
) -> list[BenchmarkCase]:
"""Build a matrix with alternating mode order to reduce order bias."""
"""Build a matrix with alternating mode order to reduce order bias.
``snapshot_frequencies`` sweeps delta cases across cadences. Full-mode
cases ignore the cadence and run once per cell at the production default,
so a sweep costs no duplicate full-mode measurements.
"""
frequencies = snapshot_frequencies or [PRODUCTION_SNAPSHOT_FREQUENCY]
cases: list[BenchmarkCase] = []
for repetition in range(repetitions):
for backend in backends:
@ -140,16 +152,19 @@ def _expand_cases(
if (repetition + update_index) % 2 == 1:
ordered_modes.reverse()
for mode in ordered_modes:
cases.append(
BenchmarkCase(
mode=mode, # type: ignore[arg-type]
backend=backend, # type: ignore[arg-type]
update_count=update_count,
payload_bytes=payload,
repetition=repetition,
seed=seed,
mode_frequencies = frequencies if mode == "delta" else [PRODUCTION_SNAPSHOT_FREQUENCY]
for snapshot_frequency in mode_frequencies:
cases.append(
BenchmarkCase(
mode=mode, # type: ignore[arg-type]
backend=backend, # type: ignore[arg-type]
update_count=update_count,
payload_bytes=payload,
repetition=repetition,
seed=seed,
snapshot_frequency=snapshot_frequency,
)
)
)
return cases
@ -161,7 +176,10 @@ def _estimated_full_payload_bytes(case: BenchmarkCase) -> int:
def _filter_oversized_pairs(cases: list[BenchmarkCase], *, max_bytes: int | None) -> tuple[list[BenchmarkCase], list[BenchmarkCase]]:
if max_bytes is None:
return cases, []
group_fields = ("backend", "scenario", "snapshot_frequency", "update_count", "payload_bytes", "repetition")
# Cadence is a delta-only sweep dimension: full mode runs once per benchmark
# cell at the production default. Excluding it ensures an oversized full
# case suppresses every delta cadence in that same cell.
group_fields = ("backend", "scenario", "update_count", "payload_bytes", "repetition")
oversized_keys = {tuple(getattr(case, field) for field in group_fields) for case in cases if case.mode == "full" and _estimated_full_payload_bytes(case) > max_bytes}
kept = [case for case in cases if tuple(getattr(case, field) for field in group_fields) not in oversized_keys]
skipped = [case for case in cases if tuple(getattr(case, field) for field in group_fields) in oversized_keys]
@ -180,8 +198,8 @@ def _noop(_state: dict[str, Any]) -> dict[str, Any]:
return {}
def _build_graph(mode: Mode, saver: Any) -> Any:
schema = _DeltaBenchmarkState if mode == "delta" else _FullBenchmarkState
def _build_graph(mode: Mode, saver: Any, snapshot_frequency: int) -> Any:
schema = _delta_benchmark_state(snapshot_frequency) if mode == "delta" else _FullBenchmarkState
builder = StateGraph(schema)
builder.add_node("noop", _noop)
builder.set_entry_point("noop")
@ -283,7 +301,7 @@ def _sqlite_storage_stats(saver: SqliteSaver, thread_id: str) -> dict[str, int]:
def _write_and_read(case: BenchmarkCase, saver: Any, messages: list[BaseMessage]) -> tuple[dict[str, Any], list[AnyMessage]]:
graph = _build_graph(case.mode, saver)
graph = _build_graph(case.mode, saver, case.snapshot_frequency)
accessor = CheckpointStateAccessor.bind(graph, saver, mode=case.mode)
config = _config(case)
update_latencies: list[float] = []
@ -312,7 +330,7 @@ def _write_and_read(case: BenchmarkCase, saver: Any, messages: list[BaseMessage]
def _cold_read(case: BenchmarkCase, saver: Any) -> tuple[float, list[AnyMessage]]:
graph = _build_graph(case.mode, saver)
graph = _build_graph(case.mode, saver, case.snapshot_frequency)
accessor = CheckpointStateAccessor.bind(graph, saver, mode=case.mode)
gc.collect()
start = time.perf_counter()
@ -455,7 +473,7 @@ def _failure_row(case: BenchmarkCase, error: str) -> dict[str, Any]:
def _profile_filename(case: BenchmarkCase) -> str:
return f"{case.backend}-{case.mode}-updates-{case.update_count}-payload-{case.payload_bytes}-rep-{case.repetition}.prof"
return f"{case.backend}-{case.mode}-freq-{case.snapshot_frequency}-updates-{case.update_count}-payload-{case.payload_bytes}-rep-{case.repetition}.prof"
def _run_child_case(case: BenchmarkCase, *, timeout_seconds: float, git_sha: str, profile_dir: Path | None = None) -> dict[str, Any]:
@ -477,6 +495,16 @@ def _build_parser() -> argparse.ArgumentParser:
parser.add_argument("--backends", default="memory,sqlite", help="Comma-separated backends (default: memory,sqlite)")
parser.add_argument("--updates", default="10,100", help="Comma-separated message update counts (default: 10,100)")
parser.add_argument("--payload-bytes", default="128", help="Comma-separated exact message content sizes (default: 128)")
parser.add_argument(
"--snapshot-frequencies",
default=str(PRODUCTION_SNAPSHOT_FREQUENCY),
help=(
"Comma-separated DeltaChannel snapshot cadences for delta cases "
f"(default: {PRODUCTION_SNAPSHOT_FREQUENCY}). Full-mode cases ignore "
"the cadence and run once per cell at the production default. "
"Sweep 100,250,500,1000 to compare cadence tradeoffs."
),
)
parser.add_argument("--repetitions", type=int, default=3)
parser.add_argument("--seed", type=int, default=1)
parser.add_argument("--timeout-seconds", type=float, default=900)
@ -521,6 +549,7 @@ def main(argv: list[str] | None = None) -> int:
backends = _parse_choice_csv(args.backends, option="--backends", choices=_BACKENDS)
updates = _parse_positive_int_csv(args.updates, option="--updates")
payload_bytes = _parse_positive_int_csv(args.payload_bytes, option="--payload-bytes")
snapshot_frequencies = _parse_positive_int_csv(args.snapshot_frequencies, option="--snapshot-frequencies")
if args.repetitions <= 0:
raise ValueError("--repetitions must be positive")
if args.timeout_seconds <= 0:
@ -537,6 +566,7 @@ def main(argv: list[str] | None = None) -> int:
payload_bytes=payload_bytes,
repetitions=args.repetitions,
seed=args.seed,
snapshot_frequencies=snapshot_frequencies,
)
cases, skipped = _filter_oversized_pairs(cases, max_bytes=None if args.allow_large_cases else args.max_estimated_full_bytes)
if skipped:
@ -552,8 +582,9 @@ def main(argv: list[str] | None = None) -> int:
git_sha = _resolve_git_sha()
rows: list[dict[str, Any]] = []
for index, case in enumerate(cases, start=1):
cadence = f" freq={case.snapshot_frequency}" if case.mode == "delta" else ""
print(
f"[{index}/{len(cases)}] {case.backend} {case.mode} updates={case.update_count} payload={case.payload_bytes} repetition={case.repetition}",
f"[{index}/{len(cases)}] {case.backend} {case.mode}{cadence} updates={case.update_count} payload={case.payload_bytes} repetition={case.repetition}",
file=sys.stderr,
)
rows.append(_run_child_case(case, timeout_seconds=args.timeout_seconds, git_sha=git_sha, profile_dir=args.profile_dir))

4
backend/scripts/benchmark/checkpoint/bench_production.py
View File

@ -28,6 +28,8 @@ from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Literal
from deerflow.config.database_config import DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY
sys.path.insert(0, str(Path(__file__).resolve().parent))
import checkpoint_bench_common as common # noqa: E402
@ -404,7 +406,7 @@ async def _run_phase(case: ProductionCase, saver: Any, timing: _TimingSaver, wor
"backend": "sqlite",
"sqlite_dir": str(work_dir),
"checkpoint_channel_mode": case.mode,
"checkpoint_delta_snapshot_frequency": case.snapshot_frequency or 1000,
"checkpoint_delta": {"snapshot_frequency": case.snapshot_frequency or DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY},
},
}
)

18
backend/tests/conftest.py
View File

@ -97,7 +97,8 @@ def _reset_skill_storage_singleton():
def _reset_frozen_checkpoint_channel_mode(monkeypatch):
"""Reset the process-global frozen checkpoint channel mode between tests.
Production treats ``checkpoint_channel_mode`` as restart-required: the
Production treats ``checkpoint_channel_mode`` (and the delta
``snapshot_frequency`` frozen alongside it) as restart-required: the
first client/app freezes it for the process. The test suite builds many
clients and apps with different modes in one process, so the freeze must
not leak across tests. Mirrors the per-test ``monkeypatch.setattr``
@ -106,20 +107,7 @@ def _reset_frozen_checkpoint_channel_mode(monkeypatch):
from deerflow.runtime import checkpoint_mode
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_channel_mode", None)
yield
@pytest.fixture(autouse=True)
def _reset_frozen_delta_snapshot_frequency(monkeypatch):
"""Reset the process-global frozen delta snapshot frequency between tests.
Same restart-required semantics as the checkpoint channel mode freeze
above: ``make_lead_agent`` freezes it from the app config, and the suite
builds many agents with different configs in one process.
"""
from deerflow.agents import thread_state
monkeypatch.setattr(thread_state, "_frozen_delta_snapshot_frequency", None)
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", None)
yield

64
backend/tests/test_app_config_reload.py
View File

@ -119,6 +119,70 @@ def test_checkpoint_channel_mode_rejects_unknown_value() -> None:
DatabaseConfig(checkpoint_channel_mode="auto")
def test_checkpoint_delta_defaults_to_production_cadence() -> None:
assert DatabaseConfig().checkpoint_delta.snapshot_frequency == 10
def test_checkpoint_delta_accepts_custom_snapshot_frequency() -> None:
assert DatabaseConfig(checkpoint_delta={"snapshot_frequency": 250}).checkpoint_delta.snapshot_frequency == 250
@pytest.mark.parametrize("value", [0, -1])
def test_checkpoint_delta_rejects_non_positive_snapshot_frequency(value: int) -> None:
with pytest.raises(ValidationError):
DatabaseConfig(checkpoint_delta={"snapshot_frequency": value})
def test_legacy_snapshot_frequency_maps_to_nested_key(caplog: pytest.LogCaptureFixture) -> None:
with caplog.at_level("WARNING"):
config = DatabaseConfig(checkpoint_delta_snapshot_frequency=1000)
assert config.checkpoint_delta.snapshot_frequency == 1000
assert "checkpoint_delta_snapshot_frequency is deprecated" in caplog.text
def test_nested_snapshot_frequency_wins_over_legacy_key(caplog: pytest.LogCaptureFixture) -> None:
with caplog.at_level("WARNING"):
config = DatabaseConfig(
checkpoint_delta_snapshot_frequency=1000,
checkpoint_delta={"snapshot_frequency": 250},
)
assert config.checkpoint_delta.snapshot_frequency == 250
assert "the nested key wins" in caplog.text
@pytest.mark.parametrize("value", [0, -1])
def test_legacy_snapshot_frequency_rejects_non_positive_value(value: int) -> None:
with pytest.raises(ValidationError):
DatabaseConfig(checkpoint_delta_snapshot_frequency=value)
def test_checkpoint_graph_cache_defaults() -> None:
assert DatabaseConfig().checkpoint_graph_cache.accessor_graph_max == 64
def test_checkpoint_graph_cache_accepts_custom_cap() -> None:
assert DatabaseConfig(checkpoint_graph_cache={"accessor_graph_max": 8}).checkpoint_graph_cache.accessor_graph_max == 8
@pytest.mark.parametrize("value", [0, -1])
def test_checkpoint_graph_cache_rejects_non_positive_cap(value: int) -> None:
with pytest.raises(ValidationError):
DatabaseConfig(checkpoint_graph_cache={"accessor_graph_max": value})
def test_resolve_checkpoint_graph_cache_max_tolerates_stub_configs() -> None:
from types import SimpleNamespace
from unittest.mock import MagicMock
from deerflow.config.database_config import resolve_checkpoint_graph_cache_max
assert resolve_checkpoint_graph_cache_max(None, "accessor_graph_max", 64) == 64
assert resolve_checkpoint_graph_cache_max(SimpleNamespace(), "accessor_graph_max", 64) == 64
assert resolve_checkpoint_graph_cache_max(MagicMock(), "accessor_graph_max", 64) == 64
stub = SimpleNamespace(checkpoint_graph_cache=SimpleNamespace(accessor_graph_max=3))
assert resolve_checkpoint_graph_cache_max(stub, "accessor_graph_max", 64) == 3
def test_config_example_does_not_enable_empty_extensions_block_by_default():
config_example_path = Path(__file__).resolve().parents[2] / "config.example.yaml"

71
backend/tests/test_bench_checkpoint_channels.py
View File

@ -86,6 +86,28 @@ def test_oversized_filter_skips_full_and_delta_as_a_comparable_pair() -> None:
assert {(case.update_count, case.mode) for case in skipped} == {(100, "full"), (100, "delta")}
def test_oversized_filter_applies_full_cap_to_every_swept_delta_cadence() -> None:
cases = bench._expand_cases(
modes=["full", "delta"],
backends=["memory"],
update_counts=[100],
payload_bytes=[128],
repetitions=1,
seed=1,
snapshot_frequencies=[1, 250, 1000],
)
kept, skipped = bench._filter_oversized_pairs(cases, max_bytes=100_000)
assert kept == []
assert {(case.mode, case.snapshot_frequency) for case in skipped} == {
("full", bench.PRODUCTION_SNAPSHOT_FREQUENCY),
("delta", 1),
("delta", 250),
("delta", 1000),
}
def test_oversized_filter_does_not_suppress_a_delta_only_diagnostic() -> None:
case = bench.BenchmarkCase(
mode="delta",
@ -106,6 +128,55 @@ def test_help_explains_delta_only_runs_bypass_full_payload_cap() -> None:
assert "delta-only" in bench._build_parser().format_help()
def test_expand_cases_sweeps_delta_frequencies_without_duplicating_full_cases() -> None:
cases = bench._expand_cases(
modes=["full", "delta"],
backends=["memory"],
update_counts=[10],
payload_bytes=[128],
repetitions=1,
seed=1,
snapshot_frequencies=[100, 250, 500, 1000],
)
full_cases = [case for case in cases if case.mode == "full"]
delta_cases = [case for case in cases if case.mode == "delta"]
assert len(full_cases) == 1
assert full_cases[0].snapshot_frequency == bench.PRODUCTION_SNAPSHOT_FREQUENCY
assert sorted(case.snapshot_frequency for case in delta_cases) == [100, 250, 500, 1000]
def test_case_rejects_non_positive_snapshot_frequency() -> None:
with pytest.raises(ValueError, match="snapshot_frequency"):
bench.BenchmarkCase(
mode="delta",
backend="memory",
update_count=4,
payload_bytes=64,
repetition=0,
seed=1,
snapshot_frequency=0,
)
def test_memory_smoke_case_materializes_at_low_snapshot_frequency(tmp_path: Path) -> None:
case = bench.BenchmarkCase(
mode="delta",
backend="memory",
update_count=6,
payload_bytes=64,
repetition=0,
seed=1,
snapshot_frequency=2,
)
row = bench._run_case(case, work_dir=tmp_path)
assert row["success"] is True
assert row["snapshot_frequency"] == 2
assert row["actual_message_count"] == 6
def test_cross_mode_validation_rejects_materialized_state_mismatch() -> None:
rows = [
{

5
backend/tests/test_bench_checkpoint_production.py
View File

@ -152,13 +152,12 @@ def test_timing_saver_records_cumulative_ms() -> None:
def _reset_checkpoint_freezes(monkeypatch: pytest.MonkeyPatch) -> None:
from deerflow.agents import thread_state
from deerflow.config.app_config import reset_app_config
from deerflow.runtime import checkpoint_mode
reset_app_config()
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_channel_mode", None)
monkeypatch.setattr(thread_state, "_frozen_delta_snapshot_frequency", None)
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", None)
@pytest.mark.parametrize("mode,frequency", [("full", None), ("delta", 10)])
@ -212,7 +211,7 @@ def test_run_case_fails_when_warm_cache_is_not_hit(tmp_path, monkeypatch) -> Non
monkeypatch.setattr(
gateway_services,
"_state_accessor_graph",
lambda agent_factory, assistant_id, mode, config: agent_factory(config=config),
lambda agent_factory, assistant_id, mode, snapshot_frequency, config: agent_factory(config=config),
)
case = bench.ProductionCase(
mode="full",

84
backend/tests/test_checkpoint_mode.py
View File

@ -23,6 +23,41 @@ def test_process_mode_change_requires_restart(monkeypatch: pytest.MonkeyPatch) -
checkpoint_mode.freeze_checkpoint_channel_mode("delta")
def test_process_snapshot_frequency_change_requires_restart(monkeypatch: pytest.MonkeyPatch) -> None:
from deerflow.runtime import checkpoint_mode
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", None)
assert checkpoint_mode.freeze_checkpoint_snapshot_frequency(250) == 250
assert checkpoint_mode.frozen_checkpoint_snapshot_frequency() == 250
assert checkpoint_mode.freeze_checkpoint_snapshot_frequency(250) == 250
with pytest.raises(checkpoint_mode.CheckpointModeReconfigurationError, match="restart"):
checkpoint_mode.freeze_checkpoint_snapshot_frequency(500)
@pytest.mark.parametrize("snapshot_frequency", [0, -1])
def test_process_snapshot_frequency_must_be_positive(
monkeypatch: pytest.MonkeyPatch,
snapshot_frequency: int,
) -> None:
from deerflow.runtime import checkpoint_mode
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", None)
with pytest.raises(ValueError, match="snapshot frequency must be positive"):
checkpoint_mode.freeze_checkpoint_snapshot_frequency(snapshot_frequency)
assert checkpoint_mode.frozen_checkpoint_snapshot_frequency() is None
def test_resolve_snapshot_frequency_prefers_explicit_then_frozen_then_default(monkeypatch: pytest.MonkeyPatch) -> None:
from deerflow.runtime import checkpoint_mode
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", None)
assert checkpoint_mode.resolve_checkpoint_snapshot_frequency() == 10
assert checkpoint_mode.resolve_checkpoint_snapshot_frequency(100) == 100
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", 250)
assert checkpoint_mode.resolve_checkpoint_snapshot_frequency() == 250
assert checkpoint_mode.resolve_checkpoint_snapshot_frequency(100) == 100
def _config() -> dict:
return {"configurable": {"thread_id": "thread-1", "checkpoint_ns": ""}}
@ -153,6 +188,49 @@ def test_yaml_mode_change_is_rejected_when_graph_is_reconstructed(tmp_path, monk
reset_app_config()
def test_yaml_snapshot_frequency_change_is_rejected_when_graph_is_reconstructed(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:
from deerflow.agents.lead_agent import agent as lead_agent
from deerflow.config.app_config import reset_app_config
from deerflow.runtime import checkpoint_mode
config_path = tmp_path / "config.yaml"
def write_config(snapshot_frequency: int) -> None:
config_path.write_text(
"\n".join(
(
"sandbox:",
" use: deerflow.sandbox.local.provider:LocalSandboxProvider",
"database:",
" checkpoint_channel_mode: delta",
" checkpoint_delta:",
f" snapshot_frequency: {snapshot_frequency}",
)
)
+ "\n",
encoding="utf-8",
)
write_config(250)
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path))
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_channel_mode", None)
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", None)
monkeypatch.setattr(lead_agent, "_make_lead_agent", lambda config, *, app_config: object())
reset_app_config()
try:
lead_agent.make_lead_agent({"configurable": {}})
assert checkpoint_mode.frozen_checkpoint_snapshot_frequency() == 250
write_config(500)
future_mtime = config_path.stat().st_mtime + 5
os.utime(config_path, (future_mtime, future_mtime))
with pytest.raises(checkpoint_mode.CheckpointModeReconfigurationError, match="restart"):
lead_agent.make_lead_agent({"configurable": {}})
finally:
reset_app_config()
@pytest.mark.asyncio
async def test_gateway_runtime_rejects_mode_different_from_frozen_process(
monkeypatch: pytest.MonkeyPatch,
@ -252,14 +330,14 @@ def test_direct_langgraph_request_cannot_select_delta_in_full_process(
def test_make_lead_agent_freezes_delta_snapshot_frequency_from_app_config(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from deerflow.agents import thread_state
from deerflow.agents.lead_agent import agent as lead_agent
from deerflow.config.app_config import AppConfig
from deerflow.runtime import checkpoint_mode
app_config = AppConfig.model_validate(
{
"sandbox": {"use": "deerflow.sandbox.local.provider:LocalSandboxProvider"},
"database": {"checkpoint_delta_snapshot_frequency": 7},
"database": {"checkpoint_delta": {"snapshot_frequency": 7}},
}
)
monkeypatch.setattr(lead_agent, "get_app_config", lambda: app_config)
@ -271,7 +349,7 @@ def test_make_lead_agent_freezes_delta_snapshot_frequency_from_app_config(
lead_agent.make_lead_agent({"configurable": {}})
assert thread_state._frozen_delta_snapshot_frequency == 7
assert checkpoint_mode.frozen_checkpoint_snapshot_frequency() == 7
def test_gateway_runtime_app_config_can_supply_its_frozen_internal_mode(

13
backend/tests/test_checkpoint_state.py
View File

@ -82,6 +82,19 @@ def _assert_delta_config_is_copied(original: dict[str, Any], forwarded: dict[str
assert forwarded["metadata"][CHECKPOINT_MODE_METADATA_KEY] == "delta"
def test_mutation_graph_bakes_snapshot_frequency_into_fallback_schema() -> None:
from langgraph.channels import DeltaChannel
from deerflow.runtime.checkpoint_state import build_state_mutation_graph
default_graph = build_state_mutation_graph("compact", "delta")
assert isinstance(default_graph.channels["messages"], DeltaChannel)
assert default_graph.channels["messages"].snapshot_frequency == 10
low_cadence = build_state_mutation_graph("compact", "delta", snapshot_frequency=250)
assert low_cadence.channels["messages"].snapshot_frequency == 250
def test_sync_accessor_binds_persistence_guards_operations_and_preserves_input() -> None:
graph = FakeGraph()
saver = FakeCheckpointer()

4
backend/tests/test_checkpointer.py
View File

@ -1015,7 +1015,7 @@ class TestClientCheckpointerFallback:
model_mock = MagicMock()
config_mock = MagicMock()
config_mock.models = [model_mock]
config_mock.database.checkpoint_delta_snapshot_frequency = 1000
config_mock.database.checkpoint_delta.snapshot_frequency = 10
config_mock.get_model_config.return_value = MagicMock(supports_vision=False)
config_mock.checkpointer = None
@ -1055,7 +1055,7 @@ class TestClientCheckpointerFallback:
model_mock = MagicMock()
config_mock = MagicMock()
config_mock.models = [model_mock]
config_mock.database.checkpoint_delta_snapshot_frequency = 1000
config_mock.database.checkpoint_delta.snapshot_frequency = 10
config_mock.get_model_config.return_value = MagicMock(supports_vision=False)
config_mock.checkpointer = None

6
backend/tests/test_client.py
View File

@ -52,7 +52,7 @@ def mock_app_config():
config.skills.container_path = "/mnt/skills"
config.tool_search.enabled = False
config.database.checkpoint_channel_mode = "full"
config.database.checkpoint_delta_snapshot_frequency = 1000
config.database.checkpoint_delta.snapshot_frequency = 10
config.authorization = AuthorizationConfig(enabled=False)
return config
@ -154,7 +154,7 @@ class TestClientInit:
from deerflow.agents import thread_state
mock_app_config.database.checkpoint_channel_mode = "delta"
mock_app_config.database.checkpoint_delta_snapshot_frequency = 7
mock_app_config.database.checkpoint_delta.snapshot_frequency = 7
with patch("deerflow.client.get_app_config", return_value=mock_app_config):
DeerFlowClient()
@ -1334,7 +1334,7 @@ class TestEnsureAgent:
"""_ensure_agent does not recreate if config key unchanged."""
mock_agent = MagicMock()
client._agent = mock_agent
client._agent_config_key = (None, True, False, False, None, None, None, None, "full", None)
client._agent_config_key = (None, True, False, False, None, None, None, None, "full", 10, None)
config = client._get_runnable_config("t1")
client._ensure_agent(config)

88
backend/tests/test_delta_channel_state.py
View File

@ -13,7 +13,6 @@ from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages
from deerflow.agents import thread_state
from deerflow.agents.thread_state import (
DeltaThreadState,
ThreadState,
@ -22,7 +21,6 @@ from deerflow.agents.thread_state import (
merge_message_writes,
normalize_middleware_state_schemas,
)
from deerflow.runtime.checkpoint_mode import CheckpointModeReconfigurationError
def _fold(state: list, writes: list) -> list:
@ -283,26 +281,56 @@ def test_mode_selects_expected_state_schema() -> None:
assert any(isinstance(item, DeltaChannel) for item in message_hint.__metadata__)
def _delta_channel(schema: type) -> DeltaChannel:
hint = get_type_hints(schema, include_extras=True)["messages"]
return next(item for item in hint.__metadata__ if isinstance(item, DeltaChannel))
def test_snapshot_frequency_parametrizes_delta_schema() -> None:
schema = get_thread_state_schema("delta", 250)
assert _delta_channel(schema).snapshot_frequency == 250
# Cached per frequency, and the default keeps DeltaThreadState identity.
assert get_thread_state_schema("delta", 250) is schema
assert get_thread_state_schema("delta") is DeltaThreadState
assert get_thread_state_schema("delta", 10) is DeltaThreadState
def test_frozen_snapshot_frequency_drives_default_delta_schema(monkeypatch: pytest.MonkeyPatch) -> None:
from deerflow.runtime import checkpoint_mode
monkeypatch.setattr(checkpoint_mode, "_frozen_checkpoint_snapshot_frequency", 500)
assert _delta_channel(get_thread_state_schema("delta")).snapshot_frequency == 500
# An explicit argument always wins over the frozen value.
assert _delta_channel(get_thread_state_schema("delta", 250)).snapshot_frequency == 250
def test_delta_adaptation_cache_keys_on_snapshot_frequency() -> None:
slow = adapt_state_schema_for_mode(AgentState, "delta", 250)
fast = adapt_state_schema_for_mode(AgentState, "delta", 500)
assert slow is not fast
assert _delta_channel(slow).snapshot_frequency == 250
assert _delta_channel(fast).snapshot_frequency == 500
assert adapt_state_schema_for_mode(AgentState, "delta", 250) is slow
def test_compiled_delta_graph_bakes_configured_snapshot_frequency() -> None:
graph = create_agent(
model=_FakeModel(responses=[AIMessage(id="response", content="done")]),
tools=None,
middleware=[],
state_schema=get_thread_state_schema("delta", 2),
)
channel = graph.channels["messages"]
assert isinstance(channel, DeltaChannel)
assert channel.snapshot_frequency == 2
def test_delta_adaptation_replaces_agent_state_message_reducer() -> None:
adapted = adapt_state_schema_for_mode(AgentState, "delta")
hint = get_type_hints(adapted, include_extras=True)["messages"]
assert any(isinstance(item, DeltaChannel) for item in hint.__metadata__)
def test_delta_adaptation_cache_varies_by_resolved_snapshot_frequency(monkeypatch) -> None:
monkeypatch.setattr(thread_state, "_frozen_delta_snapshot_frequency", None)
thread_state._adapt_state_schema_for_mode.cache_clear()
default_adapted = adapt_state_schema_for_mode(AgentState, "delta")
thread_state.freeze_delta_snapshot_frequency(7)
configured_adapted = adapt_state_schema_for_mode(AgentState, "delta")
assert configured_adapted is not default_adapted
hint = get_type_hints(configured_adapted, include_extras=True)["messages"]
channel = next(item for item in hint.__metadata__ if isinstance(item, DeltaChannel))
assert channel.snapshot_frequency == 7
def test_agents_package_exports_delta_thread_state() -> None:
from deerflow.agents import DeltaThreadState as ExportedDeltaThreadState
@ -385,31 +413,3 @@ def test_production_message_forms_keep_assigned_ids_across_delta_replay(write: o
assert first[0].id is not None
assert second[0].id == first[0].id
def test_delta_snapshot_frequency_defaults_to_1000(monkeypatch) -> None:
monkeypatch.setattr(thread_state, "_frozen_delta_snapshot_frequency", None)
assert thread_state.resolved_delta_snapshot_frequency() == 1000
def test_freeze_delta_snapshot_frequency_is_process_frozen(monkeypatch) -> None:
monkeypatch.setattr(thread_state, "_frozen_delta_snapshot_frequency", None)
assert thread_state.freeze_delta_snapshot_frequency(50) == 50
assert thread_state.freeze_delta_snapshot_frequency(50) == 50
with pytest.raises(CheckpointModeReconfigurationError):
thread_state.freeze_delta_snapshot_frequency(10)
def test_get_thread_state_schema_honors_frozen_frequency(monkeypatch) -> None:
monkeypatch.setattr(thread_state, "_frozen_delta_snapshot_frequency", None)
thread_state.freeze_delta_snapshot_frequency(7)
hint = get_type_hints(thread_state.get_thread_state_schema("delta"), include_extras=True)["messages"]
channel = next(item for item in hint.__metadata__ if isinstance(item, DeltaChannel))
assert channel.snapshot_frequency == 7
def test_database_config_default_snapshot_frequency() -> None:
from deerflow.config.database_config import DatabaseConfig
assert DatabaseConfig().checkpoint_delta_snapshot_frequency == 1000
assert DatabaseConfig(checkpoint_delta_snapshot_frequency=25).checkpoint_delta_snapshot_frequency == 25

2
backend/tests/test_gateway_run_drain_shutdown.py
View File

@ -196,7 +196,7 @@ async def test_langgraph_runtime_drains_runs_before_closing_checkpointer(monkeyp
monkeypatch.setattr(RunManager, "shutdown", spy_shutdown, raising=False)
app = FastAPI()
startup_config = SimpleNamespace(database=SimpleNamespace(backend="memory", checkpoint_channel_mode="full"), run_events=None)
startup_config = SimpleNamespace(database=SimpleNamespace(backend="memory", checkpoint_channel_mode="full", checkpoint_delta=SimpleNamespace(snapshot_frequency=10)), run_events=None)
async with langgraph_runtime(app, startup_config):
pass

4
backend/tests/test_gateway_run_recovery.py
View File

@ -221,7 +221,7 @@ async def test_sqlite_runtime_reconciles_orphaned_runs_on_startup(monkeypatch):
"""SQLite startup should recover stale active runs before serving requests."""
app = FastAPI()
config = SimpleNamespace(
database=SimpleNamespace(backend="sqlite", checkpoint_channel_mode="full"),
database=SimpleNamespace(backend="sqlite", checkpoint_channel_mode="full", checkpoint_delta=SimpleNamespace(snapshot_frequency=10)),
run_events=SimpleNamespace(backend="memory"),
stream_bridge=SimpleNamespace(recovered_stream_cleanup_delay_seconds=60.0),
)
@ -266,7 +266,7 @@ async def test_sqlite_runtime_does_not_mark_thread_error_when_newer_run_is_succe
"""Startup recovery should not let an old orphaned run overwrite a newer terminal thread state."""
app = FastAPI()
config = SimpleNamespace(
database=SimpleNamespace(backend="sqlite", checkpoint_channel_mode="full"),
database=SimpleNamespace(backend="sqlite", checkpoint_channel_mode="full", checkpoint_delta=SimpleNamespace(snapshot_frequency=10)),
run_events=SimpleNamespace(backend="memory"),
stream_bridge=SimpleNamespace(recovered_stream_cleanup_delay_seconds=60.0),
)

56
backend/tests/test_gateway_services.py
View File

@ -726,6 +726,62 @@ def test_build_checkpoint_state_accessor_uses_frozen_mode_and_binds_runtime_pers
assert config["configurable"]["checkpoint_id"] == checkpoint_id
def test_state_accessor_graph_cache_keys_on_snapshot_frequency():
"""The accessor-graph cache must not serve a graph compiled at a different
delta snapshot cadence."""
from app.gateway import services as gateway_services
builds = []
def fake_factory(*, config):
graph = object()
builds.append(graph)
return graph
gateway_services._state_accessor_graph_cache.clear()
try:
first = gateway_services._state_accessor_graph(fake_factory, None, "delta", 1000, {})
again = gateway_services._state_accessor_graph(fake_factory, None, "delta", 1000, {})
assert again is first
assert len(builds) == 1
other_cadence = gateway_services._state_accessor_graph(fake_factory, None, "delta", 250, {})
assert other_cadence is not first
assert len(builds) == 2
finally:
gateway_services._state_accessor_graph_cache.clear()
def test_state_accessor_graph_cache_honors_configured_cap():
"""database.checkpoint_graph_cache.accessor_graph_max bounds the cache;
it is re-read per eviction check (hot-reloadable)."""
from types import SimpleNamespace
from app.gateway import services as gateway_services
builds = []
def fake_factory(*, config):
graph = object()
builds.append(graph)
return graph
app_config = SimpleNamespace(database=SimpleNamespace(checkpoint_graph_cache=SimpleNamespace(accessor_graph_max=2)))
config = {"context": {"app_config": app_config}}
gateway_services._state_accessor_graph_cache.clear()
try:
gateway_services._state_accessor_graph(fake_factory, "a", "full", None, config)
gateway_services._state_accessor_graph(fake_factory, "b", "full", None, config)
assert len(builds) == 2
# Third distinct key exceeds the configured cap of 2: wholesale clear.
gateway_services._state_accessor_graph(fake_factory, "c", "full", None, config)
assert len(gateway_services._state_accessor_graph_cache) == 1
assert len(builds) == 3
finally:
gateway_services._state_accessor_graph_cache.clear()
def test_build_run_config_configurable_custom_agent_dual_writes_agent_name():
"""Regression for issue #3549: even when the caller uses the legacy
``configurable`` path, ``agent_name`` must also land in

16
backend/tests/test_lead_agent_skills.py
View File

@ -275,7 +275,7 @@ def test_make_lead_agent_empty_skills_passed_correctly(monkeypatch):
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = MockModelConfig()
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config)
@ -320,7 +320,7 @@ def test_make_lead_agent_custom_skill_allowlist_does_not_activate_tool_policy(mo
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
mock_app_config.tool_search.enabled = True
mock_app_config.skills.container_path = "/mnt/skills"
@ -365,7 +365,7 @@ def test_make_lead_agent_all_legacy_skills_preserve_all_tools(monkeypatch):
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config)
@ -421,7 +421,7 @@ def test_make_lead_agent_passive_empty_skill_policy_preserves_mcp_and_other_tool
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
mock_app_config.tool_search.enabled = True
mock_app_config.tool_search.auto_promote_top_k = 3
@ -470,7 +470,7 @@ def test_default_lead_agent_does_not_apply_installed_skill_allowlists(monkeypatc
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
mock_app_config.tool_search.enabled = True
mock_app_config.skills.container_path = "/mnt/skills"
@ -502,7 +502,7 @@ def test_make_lead_agent_fails_closed_when_skill_policy_load_fails(monkeypatch):
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
def fail_storage(*args, **kwargs):
@ -550,7 +550,7 @@ def test_make_lead_agent_drops_update_agent_on_github_channel(monkeypatch):
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config)
@ -589,7 +589,7 @@ def test_make_lead_agent_keeps_update_agent_on_non_webhook_channels(monkeypatch)
mock_app_config = MagicMock()
# make_lead_agent freezes the delta snapshot frequency from the app config;
# a bare MagicMock attribute cannot survive the freeze's positivity check.
mock_app_config.database.checkpoint_delta_snapshot_frequency = 1000
mock_app_config.database.checkpoint_delta.snapshot_frequency = 10
mock_app_config.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False)
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config)

39
backend/tests/test_threads_checkpoint_mode.py
View File

@ -145,3 +145,42 @@ async def test_delta_thread_avoids_per_step_full_message_blobs(tmp_path) -> None
# message-writing checkpoint does.
assert delta_blobs <= 1, f"delta checkpoints re-serialized messages: {delta_blobs}/{delta_total}"
assert full_blobs >= 4, f"full mode should blob messages per step: {full_blobs}/{full_total}"
async def test_delta_snapshot_frequency_controls_snapshot_cadence(tmp_path) -> None:
"""``checkpoint_delta.snapshot_frequency`` must reach the compiled channel
table: a low cadence snapshots the message list far more often than the
default, while materialized state stays identical."""
async def _run(snapshot_frequency: int, db_name: str) -> tuple[int, list[tuple[str, str, str]]]:
async with AsyncSqliteSaver.from_conn_string(str(tmp_path / db_name)) as saver:
await saver.setup()
async def _reply(state: dict[str, Any]) -> dict[str, Any]:
n = len(state.get("messages") or [])
return {"messages": [AIMessage(content=f"answer-{n}", id=f"a{n}")]}
builder = StateGraph(get_thread_state_schema("delta", snapshot_frequency))
builder.add_node("reply", _reply)
builder.set_entry_point("reply")
builder.set_finish_point("reply")
graph = builder.compile(checkpointer=saver)
accessor = CheckpointStateAccessor.bind(graph, saver, mode="delta")
config: dict[str, Any] = {"configurable": {"thread_id": "thread-cadence"}}
inject_checkpoint_mode(config, "delta")
for i in range(6):
await graph.ainvoke({"messages": [HumanMessage(content=f"q{i}", id=f"h{i}")]}, config)
blobs = 0
async for checkpoint_tuple in saver.alist(config):
if "messages" in checkpoint_tuple.checkpoint.get("channel_values", {}):
blobs += 1
latest = await accessor.aget(config)
return blobs, _normalize_messages(latest.values)
low_cadence_blobs, low_cadence_messages = await _run(2, "low-cadence.sqlite3")
default_blobs, default_messages = await _run(10, "default-cadence.sqlite3")
assert low_cadence_blobs >= 4, f"frequency=2 should snapshot repeatedly: {low_cadence_blobs}"
assert default_blobs < low_cadence_blobs
assert low_cadence_messages == default_messages

2
backend/tests/test_token_usage.py
View File

@ -147,7 +147,7 @@ def _mock_app_config():
model.model_dump.return_value = {"name": "test-model", "use": "langchain_openai:ChatOpenAI"}
config = MagicMock()
config.models = [model]
config.database.checkpoint_delta_snapshot_frequency = 1000
config.database.checkpoint_delta.snapshot_frequency = 10
return config

13
config.example.yaml
View File

@ -1866,6 +1866,19 @@ database:
# Restart required. Use one value across every process sharing this database.
# full: current full-message checkpoints; delta: DeltaChannel for messages.
checkpoint_channel_mode: full
# Delta-mode tuning (only applies when checkpoint_channel_mode is delta).
# Restart required, like the mode itself: the cadence is compiled into each
# graph's channel table. All processes sharing this database must agree.
# snapshot_frequency: full messages snapshot every N per-step writes
# (higher = smaller checkpoints, slower materialization).
checkpoint_delta:
snapshot_frequency: 10
# Size caps for the compiled checkpoint graph caches (per process).
# Hot-reloadable, no restart needed: the value is re-read on each eviction
# check and only changes when a cache evicts, never graph semantics.
checkpoint_graph_cache:
# Gateway thread-state accessor graphs (per assistant/mode/cadence).
accessor_graph_max: 64
# ============================================================================
# Run Events Configuration