From e01173d8b2ac372c4711c061c28e83f7a6b9cf2c Mon Sep 17 00:00:00 2001 From: Vanzeren <53075619+Vanzeren@users.noreply.github.com> Date: Mon, 27 Jul 2026 11:47:49 +0800 Subject: [PATCH] bench(checkpoint): production-shaped full/delta benchmark with configurable snapshot frequency (#4467) * feat(checkpoint): production-shaped full/delta benchmark with configurable snapshot frequency - Group benchmark scripts into per-family folders (checkpoint/, sandbox/) - Extract shared benchmark infrastructure into checkpoint_bench_common.py - Add checkpoint_delta_snapshot_frequency config (default 1000, process-frozen); freeze it in make_lead_agent and DeerFlowClient; key the state-schema adaptation cache by resolved frequency - New bench_production.py: per-case child processes run N ainvoke turns through the real lead-agent graph (scripted deterministic model, real AsyncSqliteSaver), then measure GET /state + POST /history through the real Gateway route stack in one event loop (httpx ASGITransport), cold/warm accessor-cache split, cross-mode digest gates - New summarize_production.py: delta/full ratios plus decision metrics (snapshot_write_spike, cache_effect_ms, checkpoint_write_share, auto-discovered history per-limit ratios) * fix(checkpoint): address production benchmark review --- README.md | 4 + backend/AGENTS.md | 66 +- .../deerflow/agents/lead_agent/agent.py | 3 +- .../harness/deerflow/agents/thread_state.py | 67 +- backend/packages/harness/deerflow/client.py | 3 +- .../deerflow/config/database_config.py | 10 + .../bench_channels.py} | 195 +---- .../benchmark/checkpoint/bench_production.py | 716 ++++++++++++++++++ .../checkpoint/checkpoint_bench_common.py | 208 +++++ .../summarize_channels.py} | 6 +- .../checkpoint/summarize_production.py | 199 +++++ .../bench_provider.py} | 4 +- .../summarize.py} | 8 +- backend/tests/conftest.py | 14 + .../tests/test_bench_checkpoint_channels.py | 2 +- .../tests/test_bench_checkpoint_production.py | 237 ++++++ backend/tests/test_bench_sandbox_provider.py | 4 +- backend/tests/test_checkpoint_mode.py | 25 + backend/tests/test_checkpointer.py | 2 + backend/tests/test_client.py | 18 + backend/tests/test_delta_channel_state.py | 44 ++ backend/tests/test_lead_agent_skills.py | 24 + .../test_summarize_checkpoint_channels.py | 2 +- .../test_summarize_checkpoint_production.py | 128 ++++ backend/tests/test_token_usage.py | 1 + 25 files changed, 1785 insertions(+), 205 deletions(-) rename backend/scripts/benchmark/{bench_checkpoint_channels.py => checkpoint/bench_channels.py} (78%) create mode 100755 backend/scripts/benchmark/checkpoint/bench_production.py create mode 100755 backend/scripts/benchmark/checkpoint/checkpoint_bench_common.py rename backend/scripts/benchmark/{summarize_checkpoint_channels.py => checkpoint/summarize_channels.py} (97%) create mode 100755 backend/scripts/benchmark/checkpoint/summarize_production.py rename backend/scripts/benchmark/{bench_sandbox_provider.py => sandbox/bench_provider.py} (99%) rename backend/scripts/benchmark/{summarize_bench.py => sandbox/summarize.py} (95%) create mode 100644 backend/tests/test_bench_checkpoint_production.py create mode 100644 backend/tests/test_summarize_checkpoint_production.py diff --git a/README.md b/README.md index d31e6d961..94857b64e 100644 --- a/README.md +++ b/README.md @@ -255,6 +255,10 @@ single-label cluster hosts, and Docker/Podman internal hostnames do not inherit 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. > [!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. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 56ebe7e19..79948b0f3 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -902,7 +902,7 @@ 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` freezes 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). A second, different mode 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`) 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. **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. @@ -920,11 +920,11 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c - `runtime/checkpoint_mode.py` — mode 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` with `snapshot_frequency=1000`), schema adaptation helpers +- `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 - `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` -**Checkpoint channel benchmark**: `scripts/benchmark/bench_checkpoint_channels.py` +**Checkpoint channel benchmark**: `scripts/benchmark/checkpoint/bench_channels.py` runs paired `full`/`delta` message-only StateGraphs in a fresh child process per case, using sync `InMemorySaver` or `SqliteSaver` so reducer, serialization, and saver costs stay separate from Gateway/async scheduling. It reports deterministic @@ -937,7 +937,7 @@ estimated cumulative full-payload cap skips both modes of an oversized pair when 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/summarize_checkpoint_channels.py` (all ratios are +with `scripts/benchmark/checkpoint/summarize_channels.py` (all ratios are `delta/full`). `--profile-dir /tmp/checkpoint-profiles` writes one cProfile artifact per case for attribution. Profiled rows carry `profiled: true`, and the summarizer automatically excludes them from baseline summaries with a warning. @@ -948,19 +948,61 @@ Example: ```bash cd backend -PYTHONPATH=. uv run python scripts/benchmark/bench_checkpoint_channels.py \ +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ --backends sqlite --updates 100,500,999,1000,1001 --payload-bytes 128 \ --repetitions 7 --output /tmp/checkpoint-bench.jsonl -PYTHONPATH=. uv run python scripts/benchmark/summarize_checkpoint_channels.py \ +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_channels.py \ /tmp/checkpoint-bench.jsonl ``` -The sync storage benchmark is not an end-to-end Gateway benchmark. Complete -`ThreadState`/`DeltaThreadState`, async saver scheduling, history, mutation, -rollback, migration, and branch-heavy cases belong to the production-shaped -follow-up layer. Harness tests live in `tests/test_bench_checkpoint_channels.py` -and `tests/test_summarize_checkpoint_channels.py`; timing thresholds are not CI -gates. +The production-shaped layer lives in +`scripts/benchmark/checkpoint/bench_production.py`: per-case child processes +run graph-level `ainvoke` turns through the real lead-agent graph (scripted +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 +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 +footprint. Summarize with +`scripts/benchmark/checkpoint/summarize_production.py` (ratios are +`delta/full`; it also emits `snapshot_write_spike` and `cache_effect_ms`, +the decision inputs for the production snapshot-frequency and accessor-cache +defaults). Harness tests live in `tests/test_bench_checkpoint_production.py` +and `tests/test_summarize_checkpoint_production.py`; timing thresholds are +not CI gates. The matrix test pins that every `(repetition, turns)` group +contains both modes and that their execution order flips between consecutive +groups, including across repetition boundaries. + +Operational limits learned from the first runs (the default matrix is too +large to run blindly): + +- The default `--timeout-seconds 900` is insufficient for delta mode at + `snapshot_frequency=1000` once turns reach 500 (measured: delta-500 takes + ~1100-1200s; delta-2000 takes ~45min). Pass an explicit + `--timeout-seconds` for any large matrix, and treat the turns=2000 corner + as practical only at small snapshot frequencies. +- Full-mode 2000-turn runs produce a ~33GB sqlite DB. Point `TMPDIR` at real + disk, not tmpfs (the benchmark uses `tempfile.TemporaryDirectory`, which + honors `TMPDIR`), or the run dies mid-case. +- The history route clamps `limit` to 100 (`le=100` on + `ThreadHistoryRequest.limit`), so `--history-limits` values above 100 are + measured and reported by their effective (clamped) limit. + +Example: + +```bash +cd backend +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_production.py \ + --turns 10,100,500,1000,2000 --payload-bytes 128 \ + --snapshot-frequencies 10,50,100,500,1000 \ + --repetitions 7 --output /tmp/production-bench.jsonl +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py \ + /tmp/production-bench.jsonl +``` ### Terminal Workbench / TUI (`packages/harness/deerflow/tui/`) diff --git a/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/backend/packages/harness/deerflow/agents/lead_agent/agent.py index b20c2e337..7aefb8eb4 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -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 get_thread_state_schema, normalize_middleware_state_schemas +from deerflow.agents.thread_state import freeze_delta_snapshot_frequency, 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 @@ -518,6 +518,7 @@ 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) inject_checkpoint_mode(config, mode) return _make_lead_agent(config, app_config=runtime_app_config) diff --git a/backend/packages/harness/deerflow/agents/thread_state.py b/backend/packages/harness/deerflow/agents/thread_state.py index 5784c267d..10ba5ca70 100644 --- a/backend/packages/harness/deerflow/agents/thread_state.py +++ b/backend/packages/harness/deerflow/agents/thread_state.py @@ -352,16 +352,61 @@ def merge_message_writes(state: list[AnyMessage], writes: Sequence[Any]) -> list return [message for message in messages if message is not None] -DELTA_MESSAGES_FIELD = Annotated[ - list[AnyMessage], - DeltaChannel(merge_message_writes, snapshot_frequency=1000), -] +DEFAULT_DELTA_SNAPSHOT_FREQUENCY = 1000 + +_frozen_delta_snapshot_frequency: int | None = None + + +def delta_messages_field(snapshot_frequency: int) -> Any: + return Annotated[ + list[AnyMessage], + DeltaChannel(merge_message_writes, snapshot_frequency=snapshot_frequency), + ] + + +DELTA_MESSAGES_FIELD = delta_messages_field(DEFAULT_DELTA_SNAPSHOT_FREQUENCY) 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", @@ -378,17 +423,23 @@ THREAD_STATE_REDUCER_FIELDS = frozenset( def get_thread_state_schema(mode: CheckpointChannelMode) -> type: - return DeltaThreadState if mode == "delta" else ThreadState + if mode == "delta": + return _delta_thread_state_for_frequency(resolved_delta_snapshot_frequency()) + return ThreadState -@cache 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()) + + +@cache +def _adapt_state_schema_for_mode(schema: type, mode: CheckpointChannelMode, snapshot_frequency: int) -> type: annotations = get_type_hints(schema, include_extras=True) - annotations["messages"] = DELTA_MESSAGES_FIELD + annotations["messages"] = delta_messages_field(snapshot_frequency) return TypedDict( - f"Delta{schema.__module__.replace('.', '_')}_{schema.__name__}", + f"Delta{schema.__module__.replace('.', '_')}_{schema.__name__}_f{snapshot_frequency}", annotations, total=getattr(schema, "__total__", True), ) diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 50a28438d..cbdc71b3e 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -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 get_thread_state_schema, normalize_middleware_state_schemas +from deerflow.agents.thread_state import freeze_delta_snapshot_frequency, 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 @@ -192,6 +192,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) 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}") diff --git a/backend/packages/harness/deerflow/config/database_config.py b/backend/packages/harness/deerflow/config/database_config.py index cb578fff5..c02f81b77 100644 --- a/backend/packages/harness/deerflow/config/database_config.py +++ b/backend/packages/harness/deerflow/config/database_config.py @@ -53,6 +53,16 @@ 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." + ), + ) sqlite_dir: str = Field( default=".deer-flow/data", description=("Directory for the SQLite database file. Both checkpointer and application data share {sqlite_dir}/deerflow.db."), diff --git a/backend/scripts/benchmark/bench_checkpoint_channels.py b/backend/scripts/benchmark/checkpoint/bench_channels.py similarity index 78% rename from backend/scripts/benchmark/bench_checkpoint_channels.py rename to backend/scripts/benchmark/checkpoint/bench_channels.py index 17acc0d6c..dce2f3802 100644 --- a/backend/scripts/benchmark/bench_checkpoint_channels.py +++ b/backend/scripts/benchmark/checkpoint/bench_channels.py @@ -8,12 +8,12 @@ warming the other. Examples:: - PYTHONPATH=. uv run python scripts/benchmark/bench_checkpoint_channels.py \ + PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ --updates 10,100,500,999,1000,1001,2000 \ --payload-bytes 128,4096 \ --output checkpoint-bench.jsonl - PYTHONPATH=. uv run python scripts/benchmark/bench_checkpoint_channels.py \ + PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ --backends sqlite --updates 1000 --payload-bytes 128 \ --repetitions 7 --output snapshot-boundary.jsonl @@ -27,15 +27,11 @@ and memory use. from __future__ import annotations import argparse -import cProfile import gc -import hashlib import importlib.metadata import json import os import platform -import statistics -import subprocess import sys import tempfile import time @@ -55,10 +51,19 @@ from deerflow.agents.thread_state import merge_message_writes from deerflow.runtime.checkpoint_mode import inject_checkpoint_mode from deerflow.runtime.checkpoint_state import CheckpointStateAccessor -try: - import resource -except ImportError: # pragma: no cover - Windows only - resource = None # type: ignore[assignment] +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import checkpoint_bench_common as _common # noqa: E402 + +_GIT_SHA_ENV = _common.GIT_SHA_ENV +_parse_positive_int_csv = _common.parse_positive_int_csv +_parse_choice_csv = _common.parse_choice_csv +_canonical_messages_digest = _common.canonical_messages_digest +_percentile = _common.percentile +_window_median = _common.window_median +_resolve_git_sha = _common.resolve_git_sha +_safe_error = _common.safe_error +_file_size = _common.file_size +_peak_rss_bytes = _common.peak_rss_bytes Mode = Literal["full", "delta"] Backend = Literal["memory", "sqlite"] @@ -67,7 +72,6 @@ SCHEMA_VERSION = 1 BENCHMARK_VERSION = 1 PRODUCTION_SNAPSHOT_FREQUENCY = 1000 DEFAULT_MAX_ESTIMATED_FULL_BYTES = 1024**3 -_GIT_SHA_ENV = "DEERFLOW_CHECKPOINT_BENCH_GIT_SHA" _MODES: tuple[Mode, ...] = ("full", "delta") _BACKENDS: tuple[Backend, ...] = ("memory", "sqlite") _STORAGE_STAT_FIELDS = ( @@ -117,53 +121,6 @@ class BenchmarkCase: raise ValueError(f"unsupported scenario: {self.scenario!r}") -def _parse_positive_int_csv(value: str, *, option: str) -> list[int]: - if not value or value.startswith(",") or value.endswith(",") or ",," in value: - raise ValueError(f"{option} must be a comma-separated list of positive integers") - result: list[int] = [] - seen: set[int] = set() - duplicates: list[int] = [] - try: - parsed = [int(part.strip()) for part in value.split(",")] - except ValueError as exc: - raise ValueError(f"{option} must be a comma-separated list of positive integers") from exc - if any(item <= 0 for item in parsed): - raise ValueError(f"{option} values must be positive integers") - for item in parsed: - if item not in seen: - result.append(item) - seen.add(item) - elif item not in duplicates: - duplicates.append(item) - if duplicates: - print( - f"{option}: ignored duplicate value(s): {', '.join(str(item) for item in duplicates)}; use --repetitions for repeated samples.", - file=sys.stderr, - ) - return result - - -def _parse_choice_csv(value: str, *, option: str, choices: tuple[str, ...]) -> list[str]: - if not value or value.startswith(",") or value.endswith(",") or ",," in value: - raise ValueError(f"{option} must contain one or more of: {', '.join(choices)}") - result: list[str] = [] - duplicates: list[str] = [] - for raw in value.split(","): - item = raw.strip() - if item not in choices: - raise ValueError(f"{option} contains unsupported value {item!r}; expected: {', '.join(choices)}") - if item not in result: - result.append(item) - elif item not in duplicates: - duplicates.append(item) - if duplicates: - print( - f"{option}: ignored duplicate value(s): {', '.join(duplicates)}; use --repetitions for repeated samples.", - file=sys.stderr, - ) - return result - - def _expand_cases( *, modes: list[str], @@ -219,47 +176,6 @@ def _message_for_update(index: int, payload_bytes: int) -> BaseMessage: return AIMessage(id=message_id, content=content) -def _canonical_messages_digest(messages: list[AnyMessage]) -> str: - canonical = [ - { - "id": message.id, - "type": message.type, - "content": message.content, - } - for message in messages - ] - payload = json.dumps(canonical, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") - return hashlib.sha256(payload).hexdigest() - - -def _percentile(values: list[float], percentile: float) -> float: - if not values: - return 0.0 - ordered = sorted(values) - if len(ordered) == 1: - return ordered[0] - rank = (len(ordered) - 1) * percentile / 100 - lower = int(rank) - upper = min(lower + 1, len(ordered) - 1) - fraction = rank - lower - return ordered[lower] * (1 - fraction) + ordered[upper] * fraction - - -def _window_median(values: list[float], window: Literal["first", "middle", "last"]) -> float: - if not values: - return 0.0 - width = max(1, len(values) // 10) - if window == "first": - selected = values[:width] - elif window == "last": - selected = values[-width:] - else: - center = len(values) // 2 - start = max(0, center - width // 2) - selected = values[start : start + width] - return statistics.median(selected) - - def _noop(_state: dict[str, Any]) -> dict[str, Any]: return {} @@ -283,20 +199,6 @@ def _config(case: BenchmarkCase) -> dict[str, Any]: return config -def _resolve_git_sha() -> str: - try: - return subprocess.run( - ["git", "rev-parse", "HEAD"], - cwd=Path(__file__).resolve().parents[3], - check=True, - capture_output=True, - text=True, - timeout=5, - ).stdout.strip() - except (OSError, subprocess.SubprocessError): - return "unknown" - - def _base_row(case: BenchmarkCase) -> dict[str, Any]: git_sha = os.environ.get(_GIT_SHA_ENV) or _resolve_git_sha() try: @@ -324,13 +226,6 @@ def _base_row(case: BenchmarkCase) -> dict[str, Any]: } -def _safe_error(error: BaseException | str, *, work_dir: Path | None = None) -> str: - message = str(error).replace(str(Path.home()), "") - if work_dir is not None: - message = message.replace(str(work_dir), "") - return message[:2000] - - def _collect_storage_stats(collector: Callable[[], dict[str, int]]) -> dict[str, Any]: """Keep timing data usable when a saver's diagnostic layout changes.""" try: @@ -387,20 +282,6 @@ def _sqlite_storage_stats(saver: SqliteSaver, thread_id: str) -> dict[str, int]: } -def _file_size(path: Path) -> int: - try: - return path.stat().st_size - except FileNotFoundError: - return 0 - - -def _peak_rss_bytes() -> int | None: - if resource is None: - return None - peak_rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss - return int(peak_rss) if sys.platform == "darwin" else int(peak_rss * 1024) - - def _write_and_read(case: BenchmarkCase, saver: Any, messages: list[BaseMessage]) -> tuple[dict[str, Any], list[AnyMessage]]: graph = _build_graph(case.mode, saver) accessor = CheckpointStateAccessor.bind(graph, saver, mode=case.mode) @@ -532,12 +413,7 @@ def _run_case(case: BenchmarkCase, *, work_dir: Path) -> dict[str, Any]: def _run_profiled_case(case: BenchmarkCase, *, work_dir: Path, profile_path: Path) -> dict[str, Any]: - profile_path.parent.mkdir(parents=True, exist_ok=True) - profiler = cProfile.Profile() - row = profiler.runcall(_run_case, case, work_dir=work_dir) - row["profiled"] = True - profiler.dump_stats(profile_path) - return row + return _common.run_profiled(_run_case, case, work_dir=work_dir, profile_path=profile_path) def _comparison_key(row: dict[str, Any]) -> tuple[Any, ...]: @@ -583,37 +459,16 @@ def _profile_filename(case: BenchmarkCase) -> str: def _run_child_case(case: BenchmarkCase, *, timeout_seconds: float, git_sha: str, profile_dir: Path | None = None) -> dict[str, Any]: - encoded_case = json.dumps(asdict(case), separators=(",", ":")) - command = [sys.executable, str(Path(__file__).resolve()), "--worker-case", encoded_case] + worker_args = ["--worker-case", json.dumps(asdict(case), separators=(",", ":"))] if profile_dir is not None: - command.extend(["--worker-profile", str(profile_dir / _profile_filename(case))]) - started = time.perf_counter() - child_env = os.environ.copy() - child_env[_GIT_SHA_ENV] = git_sha - try: - completed = subprocess.run( - command, - check=False, - capture_output=True, - text=True, - timeout=timeout_seconds, - env=child_env, - ) - except subprocess.TimeoutExpired: - return _failure_row(case, f"child process timed out after {timeout_seconds:g} seconds") - child_process_ms = (time.perf_counter() - started) * 1000 - output_lines = [line for line in completed.stdout.splitlines() if line.strip()] - if not output_lines: - return _failure_row(case, f"child process returned {completed.returncode} without a result") - try: - row = json.loads(output_lines[-1]) - except json.JSONDecodeError: - return _failure_row(case, f"child process returned {completed.returncode} with malformed JSON") - row["child_process_ms"] = child_process_ms - if completed.returncode != 0 and row.get("success"): - row["success"] = False - row["error"] = f"child process exited with status {completed.returncode}" - return row + worker_args.extend(["--worker-profile", str(profile_dir / _profile_filename(case))]) + return _common.run_child_case( + script=Path(__file__).resolve(), + worker_args=worker_args, + failure_row=lambda error: _failure_row(case, error), + timeout_seconds=timeout_seconds, + git_sha=git_sha, + ) def _build_parser() -> argparse.ArgumentParser: diff --git a/backend/scripts/benchmark/checkpoint/bench_production.py b/backend/scripts/benchmark/checkpoint/bench_production.py new file mode 100755 index 000000000..3283e649a --- /dev/null +++ b/backend/scripts/benchmark/checkpoint/bench_production.py @@ -0,0 +1,716 @@ +#!/usr/bin/env python3 +"""Production-shaped full/delta checkpoint benchmark. + +Measures user-visible run, state, and history latency on the production path: +the real lead-agent graph (scripted deterministic model), a real +AsyncSqliteSaver, and the real Gateway route stack for reads. Every case runs +in a fresh child process with a fresh database, mirroring the +restart-required checkpoint mode boundary. + +Examples:: + + PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_production.py \ + --turns 100,1000 --payload-bytes 128 \ + --snapshot-frequencies 10,100,1000 \ + --repetitions 7 --output production-bench.jsonl + + PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py \ + production-bench.jsonl +""" + +from __future__ import annotations + +import argparse +import json +import sys +import tempfile +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Literal + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import checkpoint_bench_common as common # noqa: E402 + +Mode = Literal["full", "delta"] + +SCHEMA_VERSION = 1 +BENCHMARK_VERSION = 1 +BENCHMARK_NAME = "checkpoint_production" +DEFAULT_HISTORY_LIMITS = (10, 50, 200) +DEFAULT_READ_REPETITIONS = 25 +WARMUP_TURNS = 2 +_MODES: tuple[Mode, ...] = ("full", "delta") + + +@dataclass(frozen=True) +class ProductionCase: + mode: Mode + turns: int + payload_bytes: int + snapshot_frequency: int | None + history_limits: tuple[int, ...] + read_repetitions: int + repetition: int + seed: int + + def __post_init__(self) -> None: + if self.mode not in _MODES: + raise ValueError(f"unsupported mode: {self.mode!r}") + if self.turns <= WARMUP_TURNS: + raise ValueError(f"turns must exceed the {WARMUP_TURNS}-turn warm-up") + if self.payload_bytes <= 0: + raise ValueError("payload_bytes must be positive") + if self.mode == "full" and self.snapshot_frequency is not None: + raise ValueError("snapshot_frequency is a delta-only axis") + if self.mode == "delta" and (self.snapshot_frequency is None or self.snapshot_frequency <= 0): + raise ValueError("delta mode requires a positive snapshot_frequency") + if not self.history_limits or any(limit <= 0 for limit in self.history_limits): + raise ValueError("history_limits must be positive") + if self.read_repetitions <= 0: + raise ValueError("read_repetitions must be positive") + if self.repetition < 0: + raise ValueError("repetition must be non-negative") + + +def expand_cases( + *, + modes: list[str], + turn_counts: list[int], + payload_bytes: list[int], + snapshot_frequencies: list[int], + repetitions: int, + seed: int = 1, + history_limits: tuple[int, ...] = DEFAULT_HISTORY_LIMITS, + read_repetitions: int = DEFAULT_READ_REPETITIONS, +) -> list[ProductionCase]: + """Build the matrix with alternating mode order to reduce order bias.""" + cases: list[ProductionCase] = [] + group_index = 0 + for repetition in range(repetitions): + for payload in payload_bytes: + for turns in turn_counts: + ordered_modes = list(modes) + if group_index % 2 == 1: + ordered_modes.reverse() + for mode in ordered_modes: + frequencies: list[int | None] = snapshot_frequencies if mode == "delta" else [None] + for frequency in frequencies: + cases.append( + ProductionCase( + mode=mode, # type: ignore[arg-type] + turns=turns, + payload_bytes=payload, + snapshot_frequency=frequency, + history_limits=history_limits, + read_repetitions=read_repetitions, + repetition=repetition, + seed=seed, + ) + ) + group_index += 1 + return cases + + +def _base_row(case: ProductionCase) -> dict: + import importlib.metadata + import os + import platform + + git_sha = os.environ.get(common.GIT_SHA_ENV) or common.resolve_git_sha() + try: + langgraph_version = importlib.metadata.version("langgraph") + except importlib.metadata.PackageNotFoundError: + langgraph_version = "unknown" + return { + "schema_version": SCHEMA_VERSION, + "benchmark_version": BENCHMARK_VERSION, + "benchmark": BENCHMARK_NAME, + "success": True, + "error": None, + "profiled": False, + "git_sha": git_sha, + "python_version": platform.python_version(), + "langgraph_version": langgraph_version, + "platform": platform.platform(), + "mode": case.mode, + "snapshot_frequency": case.snapshot_frequency, + "turns": case.turns, + "payload_bytes": case.payload_bytes, + "history_limits": list(case.history_limits), + "read_repetitions": case.read_repetitions, + "repetition": case.repetition, + "seed": case.seed, + } + + +def _failure_row(case: ProductionCase, error: str) -> dict: + row = _base_row(case) + row["success"] = False + row["error"] = common.safe_error(error) + return row + + +def _comparison_key(row: dict) -> tuple: + # snapshot_frequency is intentionally excluded: every delta frequency + # pairs against the same full row, and all frequencies must agree on + # the materialized digests. + return (row.get("turns"), row.get("payload_bytes"), row.get("repetition")) + + +def validate_cross_mode_rows(rows: list[dict]) -> None: + grouped: dict[tuple, list[dict]] = {} + for row in rows: + grouped.setdefault(_comparison_key(row), []).append(row) + for group in grouped.values(): + successful = [row for row in group if row.get("success")] + modes = {row.get("mode") for row in successful} + if not {"full", "delta"}.issubset(modes): + continue + signatures = {(row.get("message_count"), row.get("seed_content_sha256"), row.get("state_wire_sha256"), row.get("history_wire_sha256")) for row in successful} + if len(signatures) == 1: + continue + for row in successful: + row["success"] = False + row["error"] = "cross-mode materialized state mismatch" + + +def _profile_filename(case: ProductionCase) -> str: + return f"production-{case.mode}-turns-{case.turns}-payload-{case.payload_bytes}-freq-{case.snapshot_frequency}-rep-{case.repetition}.prof" + + +def _worker_main(encoded_case: str, *, profile_path: Path | None = None) -> int: + try: + raw = json.loads(encoded_case) + history_limits = raw.get("history_limits") + if history_limits is not None: + raw["history_limits"] = tuple(history_limits) + case = ProductionCase(**raw) + except (TypeError, ValueError) as exc: + print(json.dumps({"schema_version": SCHEMA_VERSION, "benchmark_version": BENCHMARK_VERSION, "benchmark": BENCHMARK_NAME, "success": False, "error": common.safe_error(exc)}, separators=(",", ":"))) + return 2 + with tempfile.TemporaryDirectory(prefix="deerflow-checkpoint-production-") as temp_dir: + if profile_path is None: + row = _run_case(case, work_dir=Path(temp_dir)) + else: + row = common.run_profiled(_run_case, case, work_dir=Path(temp_dir), profile_path=profile_path) + print(json.dumps(row, ensure_ascii=False, separators=(",", ":"))) + return 0 if row.get("success") else 1 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Production-shaped full/delta checkpoint benchmark") + parser.add_argument("--modes", default="full,delta") + parser.add_argument("--turns", default="10,100,500,1000,2000") + parser.add_argument("--payload-bytes", default="128") + parser.add_argument("--snapshot-frequencies", default="10,50,100,500,1000") + parser.add_argument("--history-limits", default=",".join(str(v) for v in DEFAULT_HISTORY_LIMITS)) + parser.add_argument("--read-repetitions", type=int, default=DEFAULT_READ_REPETITIONS) + 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) + parser.add_argument("--profile-dir", type=Path) + parser.add_argument("--output", type=Path) + parser.add_argument("--worker-case", help=argparse.SUPPRESS) + parser.add_argument("--worker-profile", type=Path, help=argparse.SUPPRESS) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = _build_parser() + args = parser.parse_args(argv) + if args.worker_case is not None: + return _worker_main(args.worker_case, profile_path=args.worker_profile) + if args.output is None: + parser.error("--output is required") + try: + modes = common.parse_choice_csv(args.modes, option="--modes", choices=_MODES) + turn_counts = common.parse_positive_int_csv(args.turns, option="--turns") + payload_bytes = common.parse_positive_int_csv(args.payload_bytes, option="--payload-bytes") + frequencies = common.parse_positive_int_csv(args.snapshot_frequencies, option="--snapshot-frequencies") + history_limits = tuple(common.parse_positive_int_csv(args.history_limits, option="--history-limits")) + if args.repetitions <= 0 or args.read_repetitions <= 0 or args.timeout_seconds <= 0: + raise ValueError("repetition and timeout values must be positive") + except ValueError as exc: + parser.error(str(exc)) + + cases = expand_cases( + modes=modes, + turn_counts=turn_counts, + payload_bytes=payload_bytes, + snapshot_frequencies=frequencies, + repetitions=args.repetitions, + seed=args.seed, + history_limits=history_limits, + read_repetitions=args.read_repetitions, + ) + git_sha = common.resolve_git_sha() + rows: list[dict] = [] + for index, case in enumerate(cases, start=1): + print( + f"[{index}/{len(cases)}] {case.mode} turns={case.turns} payload={case.payload_bytes} freq={case.snapshot_frequency} repetition={case.repetition}", + file=sys.stderr, + ) + worker_args = ["--worker-case", json.dumps(asdict(case), separators=(",", ":"))] + if args.profile_dir is not None: + worker_args.extend(["--worker-profile", str(args.profile_dir / _profile_filename(case))]) + rows.append( + common.run_child_case( + script=Path(__file__).resolve(), + worker_args=worker_args, + failure_row=lambda error, c=case: _failure_row(c, error), + timeout_seconds=args.timeout_seconds, + git_sha=git_sha, + ) + ) + validate_cross_mode_rows(rows) + + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", encoding="utf-8") as output_file: + for row in rows: + output_file.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + failures = sum(1 for row in rows if not row.get("success")) + print(f"Wrote {len(rows)} result row(s) to {args.output}; failures={failures}", file=sys.stderr) + return 1 if failures else 0 + + +# --------------------------------------------------------------------------- +# Worker: production-shaped run + read phases (single event loop) +# --------------------------------------------------------------------------- + +import asyncio # noqa: E402 +import hashlib # noqa: E402 +import time # noqa: E402 +from collections import defaultdict # noqa: E402 +from typing import Any # noqa: E402 +from unittest.mock import patch # noqa: E402 + +import httpx # noqa: E402 +from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel # noqa: E402 +from langchain_core.messages import AIMessage, HumanMessage # noqa: E402 +from langchain_core.outputs import ChatGeneration, ChatResult # noqa: E402 + + +class _ScriptedBenchModel(FakeMessagesListChatModel): + """Deterministic model: call k returns a fixed-size AIMessage. + + Content and IDs depend only on the call index, so full and delta modes + produce byte-identical message streams. Serves the ainvoke path via + BaseChatModel's default ``_agenerate`` fallback (``run_in_executor`` over + ``_generate``); streaming is not supported. + """ + + def __init__(self, *, payload_bytes: int) -> None: + super().__init__(responses=[]) + self._bench_payload_bytes = payload_bytes + self._bench_call_index = 0 + + def bind_tools(self, tools: Any, **kwargs: Any) -> Any: + return self + + def _generate(self, messages: Any, stop: Any = None, run_manager: Any = None, **kwargs: Any) -> ChatResult: + index = self._bench_call_index + self._bench_call_index += 1 + message = AIMessage(id=f"bench-ai-{index:08d}", content="x" * self._bench_payload_bytes) + return ChatResult(generations=[ChatGeneration(message=message)]) + + +class _TimingSaver: + """Delegating saver wrapper recording write timing. + + LangGraph issues ``aput``/``aput_writes`` as concurrent asyncio tasks, so + per-call latencies overlap (queueing on the single sqlite connection is + counted in every concurrent call). Summing them double-counts and can + exceed the turn wall clock. ``intervals`` keeps absolute (start, end) per + call so the run phase can compute the merged busy time — the honest, + wall-bounded write cost. ``timings_ms`` remains the cumulative-per-method + ledger used for read-path attribution. + """ + + def __init__(self, saver: Any) -> None: + self._saver = saver + self.timings_ms: dict[str, float] = defaultdict(float) + self.intervals: list[tuple[str, float, float]] = [] + + def __getattr__(self, name: str) -> Any: + return getattr(self._saver, name) + + async def _timed(self, label: str, fn: Any, *args: Any, **kwargs: Any) -> Any: + start = time.perf_counter() + try: + return await fn(*args, **kwargs) + finally: + end = time.perf_counter() + self.timings_ms[label] += (end - start) * 1000 + self.intervals.append((label, start * 1000, end * 1000)) + + async def aput(self, *args: Any, **kwargs: Any) -> Any: + return await self._timed("aput", self._saver.aput, *args, **kwargs) + + async def aput_writes(self, *args: Any, **kwargs: Any) -> Any: + return await self._timed("aput_writes", self._saver.aput_writes, *args, **kwargs) + + async def aget_tuple(self, *args: Any, **kwargs: Any) -> Any: + return await self._timed("aget_tuple", self._saver.aget_tuple, *args, **kwargs) + + def alist(self, *args: Any, **kwargs: Any) -> Any: + # alist is an async generator on real savers; wrap it to time the + # full iteration when consumed. + agen = self._saver.alist(*args, **kwargs) + timing = self.timings_ms + + async def _wrapped() -> Any: + start_total = time.perf_counter() + try: + async for item in agen: + yield item + finally: + timing["alist"] += (time.perf_counter() - start_total) * 1000 + + return _wrapped() + + +def _thread_id(case: ProductionCase) -> str: + return f"checkpoint-production-{case.seed}-{case.repetition}" + + +async def _run_phase(case: ProductionCase, saver: Any, timing: _TimingSaver, work_dir: Path) -> dict[str, Any]: + """Seed the thread through the real lead-agent graph; measure per-turn latency.""" + from deerflow.agents.lead_agent.agent import make_lead_agent + from deerflow.config.app_config import AppConfig, set_app_config + from deerflow.runtime.checkpoint_mode import inject_checkpoint_mode + + app_config = AppConfig.model_validate( + { + # Mirrors tests/test_lead_agent_model_resolution.py::_make_model: + # `use` is a required ModelConfig field, so a provider/api_key + # style entry does not validate. + "models": [ + { + "name": "bench-model", + "display_name": "bench-model", + "description": None, + "use": "langchain_openai:ChatOpenAI", + "model": "bench", + "supports_thinking": False, + "supports_vision": False, + } + ], + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + # Title and memory middleware would fire background LLM calls and + # debounce timers; both pollute per-turn latency, so disable them. + "title": {"enabled": False}, + "memory": {"enabled": False}, + "database": { + "backend": "sqlite", + "sqlite_dir": str(work_dir), + "checkpoint_channel_mode": case.mode, + "checkpoint_delta_snapshot_frequency": case.snapshot_frequency or 1000, + }, + } + ) + set_app_config(app_config) + + model = _ScriptedBenchModel(payload_bytes=case.payload_bytes) + + def _fake_create_chat_model(**kwargs: Any) -> Any: + return model + + with patch("deerflow.agents.lead_agent.agent.create_chat_model", _fake_create_chat_model): + graph = make_lead_agent({"configurable": {}}) + # Attach the timed saver post-construction, mirroring the run worker + # (deerflow.runs.worker assigns ``agent.checkpointer`` the same way), so + # every checkpoint write flows through ``timing``. + graph.checkpointer = timing + + config: dict[str, Any] = {"configurable": {"thread_id": _thread_id(case)}} + inject_checkpoint_mode(config, case.mode) + + turn_ms: list[float] = [] + write_ms: list[float] = [] + write_methods = {"aput", "aput_writes"} + for turn in range(case.turns): + intervals_before = len(timing.intervals) + start = time.perf_counter() + await graph.ainvoke( + {"messages": [HumanMessage(id=f"bench-h-{turn:08d}", content="y" * case.payload_bytes)]}, + config, + ) + turn_ms.append((time.perf_counter() - start) * 1000) + write_ms.append(_merged_busy_ms(iv for iv in timing.intervals[intervals_before:] if iv[0] in write_methods)) + + return {"graph": graph, "config": config, "turn_ms": turn_ms[WARMUP_TURNS:], "write_ms": write_ms[WARMUP_TURNS:]} + + +def _merged_busy_ms(intervals: Any) -> float: + """Union of (start, end) write intervals in ms — wall-bounded write cost. + + Concurrent write tasks overlap; merging their intervals yields the time + the event loop was actually inside aput/aput_writes, which cannot exceed + the turn wall clock the way a naive latency sum does. + """ + merged: list[list[float]] = [] + for _label, start, end in sorted(intervals, key=lambda iv: iv[1]): + if merged and start <= merged[-1][1]: + if end > merged[-1][1]: + merged[-1][1] = end + else: + merged.append([start, end]) + return sum(end - start for start, end in merged) + + +# ``ThreadHistoryRequest.limit`` is validated with ``le=100``: the production +# route cannot serve a larger page. Configured history limits above the cap +# are measured at the cap and labelled by the *effective* limit so no metric +# field ever claims a request the API would reject. +_HISTORY_ROUTE_MAX_LIMIT = 100 + + +def _wire_messages_digest(messages: Any) -> str: + """Digest wire-format messages, excluding volatile checkpoint ids/timestamps.""" + canonical = [[message.get("type"), message.get("id"), message.get("content") if isinstance(message.get("content"), str) else json.dumps(message.get("content"), sort_keys=True)] for message in messages or []] + payload = json.dumps(canonical, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _combined_digest(digests: list[str]) -> str: + return hashlib.sha256("".join(digests).encode("utf-8")).hexdigest() + + +def _make_gateway_app(saver: Any, mode: str, store: Any) -> Any: + """Minimal authed gateway app, mirroring tests/_router_auth_helpers.py. + + Reimplemented inline: benchmark scripts must not import from tests/. + """ + from unittest.mock import MagicMock + from uuid import uuid4 + + from fastapi import FastAPI, Request, Response + from starlette.middleware.base import BaseHTTPMiddleware + + from app.gateway.auth.models import User + from app.gateway.authz import AuthContext, Permissions + from app.gateway.routers import threads + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + from deerflow.runtime.user_context import reset_current_user, set_current_user + + permissions = [Permissions.THREADS_READ, Permissions.THREADS_WRITE, Permissions.THREADS_DELETE] + + class _StubAuthMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next: Any) -> Response: + user = User(email="bench@example.com", password_hash="x", system_role="user", id=uuid4()) + request.state.user = user + request.state.auth = AuthContext(user=user, permissions=list(permissions)) + # Mirror the production AuthMiddleware: the user contextvar must be + # set or thread-meta lookups hit user_id=AUTO and raise per request. + token = set_current_user(user) + try: + return await call_next(request) + finally: + reset_current_user(token) + + app = FastAPI() + app.add_middleware(_StubAuthMiddleware) + app.state.store = store + app.state.checkpointer = saver + app.state.thread_store = MemoryThreadMetaStore(store) + app.state.checkpoint_channel_mode = mode + app.state.run_event_store = MagicMock() + app.include_router(threads.router) + return app + + +async def _read_phase(case: ProductionCase, saver: Any, timing: _TimingSaver) -> dict[str, Any]: + """Measure state/history endpoint latency through the real route stack. + + Runs in the same event loop as the run phase (the AsyncSqliteSaver is + bound to its creating loop), driving the real threads router over an + ASGI transport. ``saver`` must be the RAW saver, not the timing + wrapper: ``Pregel.aget_state`` gates delta replay behind + ``isinstance(self.checkpointer, BaseCheckpointSaver)`` + (langgraph/pregel/main.py), so the duck-typed ``_TimingSaver`` + silently materializes delta channels as empty. The wrapper therefore + stays on the run phase only and the row records + ``saver_read_attribution: False``; the ``*_saver_ms`` fields remain in + the schema but report 0.0. + """ + from langgraph.store.memory import InMemoryStore + + from app.gateway import services as gateway_services + + store = InMemoryStore() + app = _make_gateway_app(saver, case.mode, store) + + graph_build_ms: list[float] = [] + original_resolve = gateway_services.resolve_agent_factory + real_factory = original_resolve(None) + + def _timed_factory(config: Any = None) -> Any: + start = time.perf_counter() + graph = real_factory(config) + graph_build_ms.append((time.perf_counter() - start) * 1000) + return graph + + # The accessor graph cache is keyed (assistant_id, mode); the factory + # identity check compares the callable, so one stable wrapper instance. + gateway_services.resolve_agent_factory = lambda assistant_id=None: _timed_factory + + # The accessor graph is compiled lazily inside the first request and + # rebuilds the lead agent, so the scripted model must be patched in for + # the whole read phase, exactly as in _run_phase. + model = _ScriptedBenchModel(payload_bytes=case.payload_bytes) + + def _fake_create_chat_model(**kwargs: Any) -> Any: + return model + + thread_id = _thread_id(case) + transport = httpx.ASGITransport(app=app) + result: dict[str, Any] = {} + try: + with patch("deerflow.agents.lead_agent.agent.create_chat_model", _fake_create_chat_model): + async with httpx.AsyncClient(transport=transport, base_url="http://bench") as client: + + async def _timed_request(label: str, method: str, url: str, **kwargs: Any) -> tuple[Any, float, float]: + del label # timing only; labels aid debugging failures + saver_before = timing.timings_ms["aget_tuple"] + timing.timings_ms["alist"] + start = time.perf_counter() + response = await client.request(method, url, **kwargs) + elapsed_ms = (time.perf_counter() - start) * 1000 + # The read phase drives the RAW saver (see docstring), so + # the timing wrapper never advances here: saver_ms — and + # thus every read-path *_saver_ms field — is structurally 0.0. + saver_ms = timing.timings_ms["aget_tuple"] + timing.timings_ms["alist"] - saver_before + response.raise_for_status() + return response, elapsed_ms, saver_ms + + # --- state endpoint: one cold sample after clearing the accessor cache + gateway_services._state_accessor_graph_cache.clear() + graph_build_ms.clear() + response, state_cold_ms, state_cold_saver_ms = await _timed_request("state-cold", "GET", f"/api/threads/{thread_id}/state") + # Cold/warm contract, asserted structurally: the cold request + # MUST build the accessor graph exactly once (otherwise the + # cold sample did not exercise the cold path), and warm + # requests MUST NOT build again (otherwise the cache is not + # hit and warm samples silently measure the cold path). + if len(graph_build_ms) != 1: + raise AssertionError(f"state cold read triggered {len(graph_build_ms)} accessor graph builds; expected exactly 1") + cold_graph_build_ms = graph_build_ms[0] + state_payload = response.json() + state_digest = _wire_messages_digest((state_payload.get("values") or {}).get("messages")) + + state_warm: list[float] = [] + for _ in range(case.read_repetitions): + _response, elapsed_ms, _saver_ms = await _timed_request("state-warm", "GET", f"/api/threads/{thread_id}/state") + state_warm.append(elapsed_ms) + if len(graph_build_ms) != 1: + raise AssertionError("state warm reads triggered an accessor graph rebuild; cache was not hit") + + # --- history endpoint per limit, same cold/warm split + history_metrics: dict[str, Any] = {} + history_digests: list[str] = [] + effective_limits = list(dict.fromkeys(min(limit, _HISTORY_ROUTE_MAX_LIMIT) for limit in case.history_limits)) + for limit_index, limit in enumerate(effective_limits): + gateway_services._state_accessor_graph_cache.clear() + response, cold_ms, cold_saver_ms = await _timed_request( + f"history-cold-{limit}", + "POST", + f"/api/threads/{thread_id}/history", + json={"limit": limit}, + ) + expected_builds = 2 + limit_index + if len(graph_build_ms) != expected_builds: + raise AssertionError(f"history cold read (limit={limit}) triggered {len(graph_build_ms)} total accessor graph builds; expected {expected_builds}") + entries = response.json() + digest_source: list[Any] = [] + for entry in entries: + digest_source.extend((entry.get("values") or {}).get("messages") or []) + history_digests.append(_wire_messages_digest(digest_source)) + warm: list[float] = [] + for _ in range(case.read_repetitions): + _r, elapsed_ms, _s = await _timed_request( + f"history-warm-{limit}", + "POST", + f"/api/threads/{thread_id}/history", + json={"limit": limit}, + ) + warm.append(elapsed_ms) + if len(graph_build_ms) != expected_builds: + raise AssertionError(f"history warm reads (limit={limit}) triggered an accessor graph rebuild; cache was not hit") + history_metrics[f"history_limit_{limit}_cold_ms"] = cold_ms + history_metrics[f"history_limit_{limit}_cold_saver_ms"] = cold_saver_ms + history_metrics[f"history_limit_{limit}_warm_p50_ms"] = common.percentile(warm, 50) + history_metrics[f"history_limit_{limit}_warm_p95_ms"] = common.percentile(warm, 95) + + result.update( + { + "state_cold_ms": state_cold_ms, + "state_cold_saver_ms": state_cold_saver_ms, + "state_cold_graph_build_ms": cold_graph_build_ms, + "state_warm_p50_ms": common.percentile(state_warm, 50), + "state_warm_p95_ms": common.percentile(state_warm, 95), + "state_wire_sha256": state_digest, + "history_wire_sha256": _combined_digest(history_digests), + "saver_read_attribution": False, + **history_metrics, + } + ) + finally: + gateway_services.resolve_agent_factory = original_resolve + return result + + +def _storage_file_stats(db_path: Path) -> dict[str, Any]: + return { + "db_bytes": common.file_size(db_path), + "wal_bytes": common.file_size(Path(f"{db_path}-wal")), + "shm_bytes": common.file_size(Path(f"{db_path}-shm")), + } + + +def _run_case(case: ProductionCase, *, work_dir: Path) -> dict: + """Run one benchmark case: run phase + read phase in a single event loop.""" + row = _base_row(case) + db_path = work_dir / "production-benchmark.sqlite" + + async def _async_body() -> dict[str, Any]: + from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver + + async with AsyncSqliteSaver.from_conn_string(str(db_path)) as saver: + await saver.setup() + timing = _TimingSaver(saver) + run_info = await _run_phase(case, saver, timing, work_dir) + + # Seed correctness: materialize through the mode-matched accessor. + from deerflow.runtime.checkpoint_state import CheckpointStateAccessor + + accessor = CheckpointStateAccessor.bind(run_info["graph"], saver, mode=case.mode) + snapshot = await accessor.aget(run_info["config"]) + messages = list(snapshot.values.get("messages", [])) + seed_digest = common.canonical_messages_digest(messages) + + read_metrics = await _read_phase(case, saver, timing) + storage_metrics = _storage_file_stats(db_path) + + metrics = { + "run_turn_p50_ms": common.percentile(run_info["turn_ms"], 50), + "run_turn_p95_ms": common.percentile(run_info["turn_ms"], 95), + "checkpoint_write_p50_ms": common.percentile(run_info["write_ms"], 50), + "checkpoint_write_p95_ms": common.percentile(run_info["write_ms"], 95), + "checkpoint_write_p99_ms": common.percentile(run_info["write_ms"], 99), + "message_count": len(messages), + "seed_content_sha256": seed_digest, + "peak_rss_bytes": common.peak_rss_bytes(), + **storage_metrics, + **read_metrics, + } + return metrics + + try: + row.update(asyncio.run(_async_body())) + except Exception as exc: + row["success"] = False + row["error"] = common.safe_error(exc, work_dir=work_dir) + return row + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/scripts/benchmark/checkpoint/checkpoint_bench_common.py b/backend/scripts/benchmark/checkpoint/checkpoint_bench_common.py new file mode 100755 index 000000000..4cd92d56e --- /dev/null +++ b/backend/scripts/benchmark/checkpoint/checkpoint_bench_common.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""Shared infrastructure for the checkpoint benchmark family. + +Pure, stateless helpers plus the child-process controller protocol used by +both bench_channels.py (storage microbenchmark) and bench_production.py +(production-shaped benchmark). Scripts in this folder import this module via +a sibling sys.path insert; it is not a package. +""" + +from __future__ import annotations + +import cProfile +import hashlib +import json +import os +import statistics +import subprocess +import sys +import time +from collections.abc import Callable +from pathlib import Path +from typing import Any, Literal + +from langchain_core.messages import AnyMessage + +try: + import resource +except ImportError: # pragma: no cover - Windows only + resource = None # type: ignore[assignment] + +GIT_SHA_ENV = "DEERFLOW_CHECKPOINT_BENCH_GIT_SHA" + + +def parse_positive_int_csv(value: str, *, option: str) -> list[int]: + if not value or value.startswith(",") or value.endswith(",") or ",," in value: + raise ValueError(f"{option} must be a comma-separated list of positive integers") + result: list[int] = [] + seen: set[int] = set() + duplicates: list[int] = [] + try: + parsed = [int(part.strip()) for part in value.split(",")] + except ValueError as exc: + raise ValueError(f"{option} must be a comma-separated list of positive integers") from exc + if any(item <= 0 for item in parsed): + raise ValueError(f"{option} values must be positive integers") + for item in parsed: + if item not in seen: + result.append(item) + seen.add(item) + elif item not in duplicates: + duplicates.append(item) + if duplicates: + print( + f"{option}: ignored duplicate value(s): {', '.join(str(item) for item in duplicates)}; use --repetitions for repeated samples.", + file=sys.stderr, + ) + return result + + +def parse_choice_csv(value: str, *, option: str, choices: tuple[str, ...]) -> list[str]: + if not value or value.startswith(",") or value.endswith(",") or ",," in value: + raise ValueError(f"{option} must contain one or more of: {', '.join(choices)}") + result: list[str] = [] + duplicates: list[str] = [] + for raw in value.split(","): + item = raw.strip() + if item not in choices: + raise ValueError(f"{option} contains unsupported value {item!r}; expected: {', '.join(choices)}") + if item not in result: + result.append(item) + elif item not in duplicates: + duplicates.append(item) + if duplicates: + print( + f"{option}: ignored duplicate value(s): {', '.join(duplicates)}; use --repetitions for repeated samples.", + file=sys.stderr, + ) + return result + + +def canonical_messages_digest(messages: list[AnyMessage]) -> str: + canonical = [ + { + "id": message.id, + "type": message.type, + "content": message.content, + } + for message in messages + ] + payload = json.dumps(canonical, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def percentile(values: list[float], percentile: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + if len(ordered) == 1: + return ordered[0] + rank = (len(ordered) - 1) * percentile / 100 + lower = int(rank) + upper = min(lower + 1, len(ordered) - 1) + fraction = rank - lower + return ordered[lower] * (1 - fraction) + ordered[upper] * fraction + + +def window_median(values: list[float], window: Literal["first", "middle", "last"]) -> float: + if not values: + return 0.0 + width = max(1, len(values) // 10) + if window == "first": + selected = values[:width] + elif window == "last": + selected = values[-width:] + else: + center = len(values) // 2 + start = max(0, center - width // 2) + selected = values[start : start + width] + return statistics.median(selected) + + +def resolve_git_sha() -> str: + try: + return subprocess.run( + ["git", "rev-parse", "HEAD"], + # scripts/benchmark/checkpoint/checkpoint_bench_common.py -> repo root + cwd=Path(__file__).resolve().parents[4], + check=True, + capture_output=True, + text=True, + timeout=5, + ).stdout.strip() + except (OSError, subprocess.SubprocessError): + return "unknown" + + +def safe_error(error: BaseException | str, *, work_dir: Path | None = None) -> str: + message = str(error).replace(str(Path.home()), "") + if work_dir is not None: + message = message.replace(str(work_dir), "") + return message[:2000] + + +def file_size(path: Path) -> int: + try: + return path.stat().st_size + except FileNotFoundError: + return 0 + + +def peak_rss_bytes() -> int | None: + if resource is None: + return None + peak_rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + return int(peak_rss) if sys.platform == "darwin" else int(peak_rss * 1024) + + +def run_profiled(fn: Callable[..., dict[str, Any]], *args: Any, profile_path: Path, **kwargs: Any) -> dict[str, Any]: + """Run *fn* under cProfile, mark the row, and dump stats.""" + profile_path.parent.mkdir(parents=True, exist_ok=True) + profiler = cProfile.Profile() + row = profiler.runcall(fn, *args, **kwargs) + row["profiled"] = True + profiler.dump_stats(profile_path) + return row + + +def run_child_case( + *, + script: Path, + worker_args: list[str], + failure_row: Callable[[str], dict[str, Any]], + timeout_seconds: float, + git_sha: str, +) -> dict[str, Any]: + """Run one benchmark case in a fresh child process; return its JSONL row. + + The child prints exactly one JSON row on its last stdout line. Any + protocol failure is converted into a failure row via *failure_row*. + """ + command = [sys.executable, str(script), *worker_args] + started = time.perf_counter() + child_env = os.environ.copy() + child_env[GIT_SHA_ENV] = git_sha + try: + completed = subprocess.run( + command, + check=False, + capture_output=True, + text=True, + timeout=timeout_seconds, + env=child_env, + ) + except subprocess.TimeoutExpired: + return failure_row(f"child process timed out after {timeout_seconds:g} seconds") + child_process_ms = (time.perf_counter() - started) * 1000 + output_lines = [line for line in completed.stdout.splitlines() if line.strip()] + if not output_lines: + return failure_row(f"child process returned {completed.returncode} without a result") + try: + row = json.loads(output_lines[-1]) + except json.JSONDecodeError: + return failure_row(f"child process returned {completed.returncode} with malformed JSON") + row["child_process_ms"] = child_process_ms + if completed.returncode != 0 and row.get("success"): + row["success"] = False + row["error"] = f"child process exited with status {completed.returncode}" + return row diff --git a/backend/scripts/benchmark/summarize_checkpoint_channels.py b/backend/scripts/benchmark/checkpoint/summarize_channels.py similarity index 97% rename from backend/scripts/benchmark/summarize_checkpoint_channels.py rename to backend/scripts/benchmark/checkpoint/summarize_channels.py index 8db2316f6..7fe94ff72 100644 --- a/backend/scripts/benchmark/summarize_checkpoint_channels.py +++ b/backend/scripts/benchmark/checkpoint/summarize_channels.py @@ -8,9 +8,9 @@ values below 1 mean delta used less time or storage. Examples:: - python scripts/benchmark/summarize_checkpoint_channels.py results.jsonl - python scripts/benchmark/summarize_checkpoint_channels.py results/*.jsonl --json - python scripts/benchmark/summarize_checkpoint_channels.py results.jsonl \ + python scripts/benchmark/checkpoint/summarize_channels.py results.jsonl + python scripts/benchmark/checkpoint/summarize_channels.py results/*.jsonl --json + python scripts/benchmark/checkpoint/summarize_channels.py results.jsonl \ --metrics write_total_ms,cold_read_ms,logical_checkpoint_bytes --csv """ diff --git a/backend/scripts/benchmark/checkpoint/summarize_production.py b/backend/scripts/benchmark/checkpoint/summarize_production.py new file mode 100755 index 000000000..093a503dc --- /dev/null +++ b/backend/scripts/benchmark/checkpoint/summarize_production.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Summarize paired full/delta production-benchmark JSONL results. + +Full rows carry snapshot_frequency=None; each delta frequency group pairs +against the full row at the same (turns, payload_bytes, repetition). Ratios +are always delta/full. Decision metrics: snapshot_write_spike (delta +checkpoint_write p99/p50), cache_effect_ms (state cold - warm p50), and +checkpoint_write_share (checkpoint_write p50 / run_turn p50). Unless +--metrics is given, history_limit_*_cold_ms / history_limit_*_warm_p50_ms +fields found in the input rows are appended to the default metric list. + +Example:: + + python scripts/benchmark/checkpoint/summarize_production.py results.jsonl +""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +import statistics +import sys +from collections import defaultdict +from pathlib import Path +from typing import Any + +DEFAULT_METRICS = [ + "run_turn_p50_ms", + "run_turn_p95_ms", + "checkpoint_write_p50_ms", + "checkpoint_write_p95_ms", + "state_cold_ms", + "state_warm_p50_ms", + "state_warm_p95_ms", + "db_bytes", + "peak_rss_bytes", +] + +_GROUP_FIELDS = ("turns", "payload_bytes", "snapshot_frequency") +_INPUT_INDEX_FIELD = "_benchmark_input_index" +_HISTORY_METRIC_RE = re.compile(r"history_limit_(\d+)_(?:cold_ms|warm_p50_ms)") + + +def _discover_history_metrics(rows: list[dict[str, Any]]) -> list[str]: + """Auto-discover per-limit history metrics present in the input rows.""" + discovered = {key for row in rows for key in row if _HISTORY_METRIC_RE.fullmatch(key)} + return sorted(discovered, key=lambda key: (int(_HISTORY_METRIC_RE.fullmatch(key).group(1)), key)) # type: ignore[union-attr] + + +def _load_jsonl(paths: list[Path]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + errors: list[str] = [] + for input_index, path in enumerate(paths): + with path.open("r", encoding="utf-8") as input_file: + for line_number, raw_line in enumerate(input_file, start=1): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + try: + row = json.loads(line) + except json.JSONDecodeError as exc: + errors.append(f"{path}:{line_number}: {exc.msg}") + continue + if not isinstance(row, dict): + errors.append(f"{path}:{line_number}: expected a JSON object") + continue + row[_INPUT_INDEX_FIELD] = input_index + rows.append(row) + if errors: + raise ValueError("Malformed JSONL row(s):\n" + "\n".join(errors)) + return rows + + +def _numeric(row: dict[str, Any], metric: str) -> float | None: + value = row.get(metric) + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def _medians(rows: list[dict[str, Any]], metric: str) -> float | None: + values = [v for row in rows if (v := _numeric(row, metric)) is not None] + return statistics.median(values) if values else None + + +def _summarize(rows: list[dict[str, Any]], *, metrics: list[str]) -> list[dict[str, Any]]: + # Full rows are frequency-less; index them for broadcast pairing. + full_rows: dict[tuple, dict[str, Any]] = {} + delta_groups: dict[tuple, dict[tuple, dict[str, Any]]] = defaultdict(dict) + failed_rows: list[dict[str, Any]] = [] + for row in rows: + if row.get("profiled"): + continue + if not row.get("success"): + failed_rows.append(row) + continue + rep_key = (row.get(_INPUT_INDEX_FIELD), row.get("repetition"), row.get("turns"), row.get("payload_bytes")) + if row.get("mode") == "full": + full_rows[rep_key] = row + elif row.get("mode") == "delta": + group_key = tuple(row.get(field) for field in _GROUP_FIELDS) + delta_groups[group_key][rep_key] = row + + summaries: list[dict[str, Any]] = [] + for group_key in sorted(delta_groups, key=str): + deltas = delta_groups[group_key] + pairs = [(full_rows[k], d) for k, d in deltas.items() if k in full_rows] + if not pairs: + continue + summary: dict[str, Any] = {field: group_key[index] for index, field in enumerate(_GROUP_FIELDS)} + summary["paired_repetitions"] = len(pairs) + # Per-group failures: a failed delta row counts toward the group whose + # (turns, payload_bytes, snapshot_frequency) it matches; a failed full + # row (frequency-less) counts toward every group at the same + # (turns, payload_bytes). + summary["failed_rows"] = sum(1 for row in failed_rows if row.get("turns") == group_key[0] and row.get("payload_bytes") == group_key[1] and (row.get("snapshot_frequency") is None or row.get("snapshot_frequency") == group_key[2])) + full_pair_rows = [pair[0] for pair in pairs] + delta_pair_rows = [pair[1] for pair in pairs] + for metric in metrics: + full_median = _medians(full_pair_rows, metric) + delta_median = _medians(delta_pair_rows, metric) + if full_median is None or delta_median is None: + continue + summary[f"full_{metric}"] = round(full_median, 6) + summary[f"delta_{metric}"] = round(delta_median, 6) + summary[f"ratio_{metric}"] = round(delta_median / full_median, 6) if full_median != 0 else None + for label, group_rows in (("full", full_pair_rows), ("delta", delta_pair_rows)): + p50 = _medians(group_rows, "checkpoint_write_p50_ms") + p99 = _medians(group_rows, "checkpoint_write_p99_ms") + run_p50 = _medians(group_rows, "run_turn_p50_ms") + cold = _medians(group_rows, "state_cold_ms") + warm = _medians(group_rows, "state_warm_p50_ms") + if p50 and p99 is not None: + summary[f"{label}_snapshot_write_spike"] = round(p99 / p50, 6) + if p50 is not None and run_p50: + summary[f"{label}_checkpoint_write_share"] = round(p50 / run_p50, 6) + if cold is not None and warm is not None: + summary[f"{label}_cache_effect_ms"] = round(cold - warm, 6) + summaries.append(summary) + return summaries + + +def _columns(rows: list[dict[str, Any]]) -> list[str]: + preferred = [*_GROUP_FIELDS, "paired_repetitions", "failed_rows"] + extras = sorted({key for row in rows for key in row if key not in preferred}) + return [field for field in preferred if any(field in row for row in rows)] + extras + + +def _print_table(rows: list[dict[str, Any]]) -> None: + if not rows: + print("(no comparable full/delta pairs)") + return + columns = _columns(rows) + widths = {column: len(column) for column in columns} + for row in rows: + for column in columns: + widths[column] = max(widths[column], len(str(row.get(column, "")))) + print(" ".join(column.rjust(widths[column]) for column in columns)) + for row in rows: + print(" ".join(str(row.get(column, "")).rjust(widths[column]) for column in columns)) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Summarize production-shaped checkpoint benchmark results") + parser.add_argument("inputs", nargs="+", type=Path) + parser.add_argument("--metrics", default=None, help="comma-separated metric fields; defaults to the built-in list plus auto-discovered history_limit_* metrics") + output = parser.add_mutually_exclusive_group() + output.add_argument("--json", action="store_true") + output.add_argument("--csv", action="store_true") + args = parser.parse_args(argv) + + try: + rows = _load_jsonl(args.inputs) + except (OSError, ValueError) as exc: + print(str(exc), file=sys.stderr) + return 1 + if args.metrics is None: + metrics = [*DEFAULT_METRICS, *_discover_history_metrics(rows)] + else: + metrics = [metric.strip() for metric in args.metrics.split(",") if metric.strip()] + if not metrics: + parser.error("--metrics must contain at least one field") + summaries = _summarize(rows, metrics=metrics) + if args.json: + json.dump(summaries, sys.stdout, ensure_ascii=False, indent=2) + print() + elif args.csv: + writer = csv.DictWriter(sys.stdout, fieldnames=_columns(summaries), extrasaction="ignore") + writer.writeheader() + writer.writerows(summaries) + else: + _print_table(summaries) + return 0 if summaries else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/scripts/benchmark/bench_sandbox_provider.py b/backend/scripts/benchmark/sandbox/bench_provider.py similarity index 99% rename from backend/scripts/benchmark/bench_sandbox_provider.py rename to backend/scripts/benchmark/sandbox/bench_provider.py index a38ac8c32..9feb23123 100755 --- a/backend/scripts/benchmark/bench_sandbox_provider.py +++ b/backend/scripts/benchmark/sandbox/bench_provider.py @@ -6,7 +6,7 @@ workloads, and concurrency levels. Outputs JSONL for aggregation. Usage:: - python scripts/benchmark/bench_sandbox_provider.py \\ + python scripts/benchmark/sandbox/bench_provider.py \\ --provider boxlite \\ --scenario warm_same_thread \\ --workload noop \\ @@ -14,7 +14,7 @@ Usage:: --concurrency 4 \\ --output results.jsonl - python scripts/benchmark/bench_sandbox_provider.py \\ + python scripts/benchmark/sandbox/bench_provider.py \\ --provider boxlite \\ --scenario cold_unique_thread \\ --no-warmpool \\ diff --git a/backend/scripts/benchmark/summarize_bench.py b/backend/scripts/benchmark/sandbox/summarize.py similarity index 95% rename from backend/scripts/benchmark/summarize_bench.py rename to backend/scripts/benchmark/sandbox/summarize.py index 0dfe7b879..7099172af 100755 --- a/backend/scripts/benchmark/summarize_bench.py +++ b/backend/scripts/benchmark/sandbox/summarize.py @@ -3,9 +3,9 @@ Usage:: - python scripts/benchmark/summarize_bench.py results.jsonl - python scripts/benchmark/summarize_bench.py results/*.jsonl --group provider,scenario,workload - python scripts/benchmark/summarize_bench.py results.jsonl --csv > summary.csv + python scripts/benchmark/sandbox/summarize.py results.jsonl + python scripts/benchmark/sandbox/summarize.py results/*.jsonl --group provider,scenario,workload + python scripts/benchmark/sandbox/summarize.py results.jsonl --csv > summary.csv """ from __future__ import annotations @@ -148,7 +148,7 @@ def _print_table(rows: list[dict[str, Any]], fmt: str = "plain") -> None: def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser(description="Aggregate JSONL benchmark results") - p.add_argument("inputs", nargs="+", help="JSONL file(s) from bench_sandbox_provider.py") + p.add_argument("inputs", nargs="+", help="JSONL file(s) from sandbox/bench_provider.py") p.add_argument( "--group", default="provider,scenario,workload,concurrency", diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 286f03d9c..0def5b958 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -109,6 +109,20 @@ def _reset_frozen_checkpoint_channel_mode(monkeypatch): 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) + yield + + @pytest.fixture(autouse=True) def _restore_title_config_singleton(): """Reset ``_title_config`` to its pristine default after every test. diff --git a/backend/tests/test_bench_checkpoint_channels.py b/backend/tests/test_bench_checkpoint_channels.py index 52f6a2254..d4a1397c5 100644 --- a/backend/tests/test_bench_checkpoint_channels.py +++ b/backend/tests/test_bench_checkpoint_channels.py @@ -9,7 +9,7 @@ import pytest def _load_module(): - path = Path(__file__).resolve().parents[1] / "scripts/benchmark/bench_checkpoint_channels.py" + path = Path(__file__).resolve().parents[1] / "scripts/benchmark/checkpoint/bench_channels.py" spec = importlib.util.spec_from_file_location("bench_checkpoint_channels", path) assert spec is not None module = importlib.util.module_from_spec(spec) diff --git a/backend/tests/test_bench_checkpoint_production.py b/backend/tests/test_bench_checkpoint_production.py new file mode 100644 index 000000000..2f8330dbe --- /dev/null +++ b/backend/tests/test_bench_checkpoint_production.py @@ -0,0 +1,237 @@ +"""Harness tests for the production-shaped checkpoint benchmark.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + + +def _load_module(): + path = Path(__file__).resolve().parents[1] / "scripts/benchmark/checkpoint/bench_production.py" + spec = importlib.util.spec_from_file_location("bench_production", path) + assert spec is not None + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +bench = _load_module() + + +def test_expand_cases_sweeps_frequency_for_delta_only() -> None: + cases = bench.expand_cases( + modes=["full", "delta"], + turn_counts=[10], + payload_bytes=[128], + snapshot_frequencies=[10, 50], + repetitions=1, + ) + delta = [c for c in cases if c.mode == "delta"] + full = [c for c in cases if c.mode == "full"] + assert sorted(c.snapshot_frequency for c in delta) == [10, 50] + assert [c.snapshot_frequency for c in full] == [None] + assert all(c.history_limits == (10, 50, 200) for c in cases) + + +def test_expand_cases_alternates_mode_order() -> None: + cases = bench.expand_cases( + modes=["full", "delta"], + turn_counts=[10, 100], + payload_bytes=[128], + snapshot_frequencies=[10], + repetitions=2, + seed=1, + ) + grouped_modes: dict[tuple[int, int], list[str]] = {} + for case in cases: + grouped_modes.setdefault((case.repetition, case.turns), []).append(case.mode) + + mode_orders = list(grouped_modes.values()) + assert all(set(order) == {"full", "delta"} for order in mode_orders) + assert all(current == list(reversed(previous)) for previous, current in zip(mode_orders, mode_orders[1:], strict=False)) + + +def test_case_rejects_nonpositive_turns() -> None: + with pytest.raises(ValueError, match="turns"): + bench.ProductionCase( + mode="full", + turns=0, + payload_bytes=128, + snapshot_frequency=None, + history_limits=(10,), + read_repetitions=5, + repetition=0, + seed=1, + ) + + +@pytest.mark.parametrize("turns", [1, 2]) +def test_case_rejects_turns_consumed_by_warmup(turns: int) -> None: + with pytest.raises(ValueError, match="warm-up"): + bench.ProductionCase( + mode="full", + turns=turns, + payload_bytes=128, + snapshot_frequency=None, + history_limits=(10,), + read_repetitions=5, + repetition=0, + seed=1, + ) + + +def test_case_rejects_frequency_on_full_mode() -> None: + with pytest.raises(ValueError, match="snapshot_frequency"): + bench.ProductionCase( + mode="full", + turns=10, + payload_bytes=128, + snapshot_frequency=10, + history_limits=(10,), + read_repetitions=5, + repetition=0, + seed=1, + ) + + +def test_cross_mode_digest_gate_fails_both_rows_on_mismatch() -> None: + base = {"turns": 10, "payload_bytes": 128, "repetition": 0, "success": True, "seed_content_sha256": "a", "state_wire_sha256": "b", "history_wire_sha256": "c"} + full = {**base, "mode": "full", "snapshot_frequency": None} + delta = {**base, "mode": "delta", "snapshot_frequency": 10} + rows = [dict(full), dict(delta)] + bench.validate_cross_mode_rows(rows) + assert all(r["success"] for r in rows) + + delta_bad = {**delta, "seed_content_sha256": "DIFFERENT"} + rows = [dict(full), delta_bad] + bench.validate_cross_mode_rows(rows) + assert not any(r["success"] for r in rows) + assert all(r["error"] == "cross-mode materialized state mismatch" for r in rows) + + +def test_cross_mode_gate_tolerates_missing_pair() -> None: + row = {"mode": "full", "snapshot_frequency": None, "turns": 10, "payload_bytes": 128, "repetition": 0, "success": True, "seed_content_sha256": "a", "state_wire_sha256": "b", "history_wire_sha256": "c"} + rows = [dict(row)] + bench.validate_cross_mode_rows(rows) + assert rows[0]["success"] is True + + +def test_scripted_model_is_deterministic_across_instances() -> None: + from langchain_core.messages import HumanMessage + + first = bench._ScriptedBenchModel(payload_bytes=16) + second = bench._ScriptedBenchModel(payload_bytes=16) + for _ in range(3): + a = first.invoke([HumanMessage(content="hi", id="h0")]) + b = second.invoke([HumanMessage(content="hi", id="h0")]) + assert a.id == b.id and a.content == b.content and len(a.content) == 16 + + +def test_timing_saver_records_cumulative_ms() -> None: + import asyncio + + from langgraph.checkpoint.base import empty_checkpoint + from langgraph.checkpoint.memory import InMemorySaver + + timing = bench._TimingSaver(InMemorySaver()) + config = {"configurable": {"thread_id": "t", "checkpoint_ns": "", "checkpoint_id": "c1"}} + + async def go() -> None: + await timing.aput(config, empty_checkpoint(), {"source": "input", "step": 0, "writes": {}}, {}) + await timing.aput_writes(config, [("ch", "v")], "task") + + asyncio.run(go()) + assert timing.timings_ms["aput"] >= 0 + assert timing.timings_ms["aput_writes"] >= 0 + assert timing._saver is not 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) + + +@pytest.mark.parametrize("mode,frequency", [("full", None), ("delta", 10)]) +def test_run_case_succeeds_end_to_end_small(tmp_path, monkeypatch, mode, frequency) -> None: + _reset_checkpoint_freezes(monkeypatch) + from app.gateway import services as gateway_services + from deerflow.config.app_config import reset_app_config + + gateway_services._state_accessor_graph_cache.clear() + case = bench.ProductionCase( + mode=mode, + turns=3, + payload_bytes=32, + snapshot_frequency=frequency, + history_limits=(2,), + read_repetitions=2, + repetition=0, + seed=1, + ) + row = bench._run_case(case, work_dir=tmp_path) + assert row["success"], row.get("error") + assert row["message_count"] > 0 + assert row["seed_content_sha256"] + assert row["state_warm_p50_ms"] >= 0 + # Wire digests must reflect real messages: a duck-typed checkpointer on + # app.state silently materializes delta channels as empty (Pregel gates + # replay behind isinstance(checkpointer, BaseCheckpointSaver)), which + # otherwise passes every assertion above vacuously. + empty_wire = bench._wire_messages_digest([]) + assert row["state_wire_sha256"] != empty_wire + assert row["history_wire_sha256"] != bench._combined_digest([empty_wire]) + assert row["wal_bytes"] > 0 + assert row["shm_bytes"] > 0 + # Write cost is merged busy time over concurrent write tasks: it must not + # exceed the turn wall clock the way a naive latency sum can. + assert row["checkpoint_write_p50_ms"] <= row["run_turn_p50_ms"] + gateway_services._state_accessor_graph_cache.clear() + reset_app_config() + + +def test_run_case_fails_when_warm_cache_is_not_hit(tmp_path, monkeypatch) -> None: + """The cold/warm contract is structural: if the accessor cache never holds, + warm samples silently measure the cold path — the case row must fail.""" + _reset_checkpoint_freezes(monkeypatch) + from app.gateway import services as gateway_services + from deerflow.config.app_config import reset_app_config + + gateway_services._state_accessor_graph_cache.clear() + # Bypass the cache entirely: every accessor resolution rebuilds the graph, + # so warm reads trip the contract assertion. + monkeypatch.setattr( + gateway_services, + "_state_accessor_graph", + lambda agent_factory, assistant_id, mode, config: agent_factory(config=config), + ) + case = bench.ProductionCase( + mode="full", + turns=3, + payload_bytes=32, + snapshot_frequency=None, + history_limits=(2,), + read_repetitions=2, + repetition=0, + seed=1, + ) + row = bench._run_case(case, work_dir=tmp_path) + assert not row["success"] + assert "cache was not hit" in row["error"] + gateway_services._state_accessor_graph_cache.clear() + reset_app_config() + + +def test_merged_busy_ms_unions_overlapping_intervals() -> None: + intervals = [("aput", 0.0, 10.0), ("aput_writes", 5.0, 12.0), ("aput", 20.0, 25.0)] + assert bench._merged_busy_ms(iter(intervals)) == 17.0 + assert bench._merged_busy_ms(iter([])) == 0.0 diff --git a/backend/tests/test_bench_sandbox_provider.py b/backend/tests/test_bench_sandbox_provider.py index 1217c3be2..133e81ab8 100644 --- a/backend/tests/test_bench_sandbox_provider.py +++ b/backend/tests/test_bench_sandbox_provider.py @@ -17,8 +17,8 @@ def _load_module(name: str, relative: str): return module -bench = _load_module("bench_sandbox_provider", "scripts/benchmark/bench_sandbox_provider.py") -summarize = _load_module("summarize_bench", "scripts/benchmark/summarize_bench.py") +bench = _load_module("bench_provider", "scripts/benchmark/sandbox/bench_provider.py") +summarize = _load_module("summarize", "scripts/benchmark/sandbox/summarize.py") class _FakeProvider: diff --git a/backend/tests/test_checkpoint_mode.py b/backend/tests/test_checkpoint_mode.py index 66923263d..001d93296 100644 --- a/backend/tests/test_checkpoint_mode.py +++ b/backend/tests/test_checkpoint_mode.py @@ -249,6 +249,31 @@ def test_direct_langgraph_request_cannot_select_delta_in_full_process( assert CHECKPOINT_MODE_METADATA_KEY not in config["metadata"] +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 + + app_config = AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local.provider:LocalSandboxProvider"}, + "database": {"checkpoint_delta_snapshot_frequency": 7}, + } + ) + monkeypatch.setattr(lead_agent, "get_app_config", lambda: app_config) + monkeypatch.setattr( + lead_agent, + "_make_lead_agent", + lambda config, *, app_config: object(), + ) + + lead_agent.make_lead_agent({"configurable": {}}) + + assert thread_state._frozen_delta_snapshot_frequency == 7 + + def test_gateway_runtime_app_config_can_supply_its_frozen_internal_mode( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/backend/tests/test_checkpointer.py b/backend/tests/test_checkpointer.py index 2befb2ad6..dfbad0c6f 100644 --- a/backend/tests/test_checkpointer.py +++ b/backend/tests/test_checkpointer.py @@ -1015,6 +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.get_model_config.return_value = MagicMock(supports_vision=False) config_mock.checkpointer = None @@ -1054,6 +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.get_model_config.return_value = MagicMock(supports_vision=False) config_mock.checkpointer = None diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index ea5718bad..2eb0ecfad 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -52,6 +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.authorization = AuthorizationConfig(enabled=False) return config @@ -145,6 +146,23 @@ class TestClientInit: ): DeerFlowClient() + def test_delta_snapshot_frequency_is_frozen_from_app_config(self, mock_app_config): + from typing import get_type_hints + + from langgraph.channels import DeltaChannel + + from deerflow.agents import thread_state + + mock_app_config.database.checkpoint_channel_mode = "delta" + mock_app_config.database.checkpoint_delta_snapshot_frequency = 7 + with patch("deerflow.client.get_app_config", return_value=mock_app_config): + DeerFlowClient() + + schema = thread_state.get_thread_state_schema("delta") + hint = get_type_hints(schema, include_extras=True)["messages"] + channel = next(item for item in hint.__metadata__ if isinstance(item, DeltaChannel)) + assert channel.snapshot_frequency == 7 + # --------------------------------------------------------------------------- # list_models / list_skills / get_memory diff --git a/backend/tests/test_delta_channel_state.py b/backend/tests/test_delta_channel_state.py index fdcbfa6e1..7e41ee109 100644 --- a/backend/tests/test_delta_channel_state.py +++ b/backend/tests/test_delta_channel_state.py @@ -13,6 +13,7 @@ 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, @@ -21,6 +22,7 @@ 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: @@ -287,6 +289,20 @@ def test_delta_adaptation_replaces_agent_state_message_reducer() -> None: 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 @@ -369,3 +385,31 @@ 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 diff --git a/backend/tests/test_lead_agent_skills.py b/backend/tests/test_lead_agent_skills.py index 7faec55c1..f40e614f0 100644 --- a/backend/tests/test_lead_agent_skills.py +++ b/backend/tests/test_lead_agent_skills.py @@ -273,6 +273,9 @@ def test_make_lead_agent_empty_skills_passed_correctly(monkeypatch): supports_thinking = False 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.get_model_config.return_value = MockModelConfig() monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config) @@ -315,6 +318,9 @@ def test_make_lead_agent_custom_skill_allowlist_does_not_activate_tool_policy(mo monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [NamedTool("task"), NamedTool("bash"), NamedTool("read_file"), NamedTool("web_search")]) 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.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" @@ -357,6 +363,9 @@ def test_make_lead_agent_all_legacy_skills_preserve_all_tools(monkeypatch): monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [NamedTool("bash"), NamedTool("read_file")]) 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.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config) @@ -410,6 +419,9 @@ 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.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 @@ -456,6 +468,9 @@ 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.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" @@ -485,6 +500,9 @@ def test_make_lead_agent_fails_closed_when_skill_policy_load_fails(monkeypatch): monkeypatch.setattr(lead_agent_module, "load_agent_config", lambda x: AgentConfig(name="test", skills=["restricted"])) 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.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False) def fail_storage(*args, **kwargs): @@ -530,6 +548,9 @@ def test_make_lead_agent_drops_update_agent_on_github_channel(monkeypatch): monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [NamedTool("bash"), NamedTool("read_file")]) 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.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config) @@ -566,6 +587,9 @@ def test_make_lead_agent_keeps_update_agent_on_non_webhook_channels(monkeypatch) monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [NamedTool("bash")]) 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.get_model_config.return_value = SimpleNamespace(supports_thinking=False, supports_vision=False) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: mock_app_config) diff --git a/backend/tests/test_summarize_checkpoint_channels.py b/backend/tests/test_summarize_checkpoint_channels.py index d5bb76341..5ec7ee7dc 100644 --- a/backend/tests/test_summarize_checkpoint_channels.py +++ b/backend/tests/test_summarize_checkpoint_channels.py @@ -9,7 +9,7 @@ import pytest def _load_module(): - path = Path(__file__).resolve().parents[1] / "scripts/benchmark/summarize_checkpoint_channels.py" + path = Path(__file__).resolve().parents[1] / "scripts/benchmark/checkpoint/summarize_channels.py" spec = importlib.util.spec_from_file_location("summarize_checkpoint_channels", path) assert spec is not None module = importlib.util.module_from_spec(spec) diff --git a/backend/tests/test_summarize_checkpoint_production.py b/backend/tests/test_summarize_checkpoint_production.py new file mode 100644 index 000000000..19aa42dfd --- /dev/null +++ b/backend/tests/test_summarize_checkpoint_production.py @@ -0,0 +1,128 @@ +"""Tests for the production benchmark summarizer.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + + +def _load_module(): + path = Path(__file__).resolve().parents[1] / "scripts/benchmark/checkpoint/summarize_production.py" + spec = importlib.util.spec_from_file_location("summarize_production", path) + assert spec is not None + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +summ = _load_module() + + +def _row(mode, rep, *, freq=None, turns=100, payload=128, run_p50=100.0, write_p50=10.0, write_p99=20.0, state_cold=50.0, state_warm=5.0, success=True): + return { + "mode": mode, + "repetition": rep, + "snapshot_frequency": freq, + "turns": turns, + "payload_bytes": payload, + "success": success, + "profiled": False, + "run_turn_p50_ms": run_p50, + "checkpoint_write_p50_ms": write_p50, + "checkpoint_write_p99_ms": write_p99, + "state_cold_ms": state_cold, + "state_warm_p50_ms": state_warm, + } + + +def test_pairs_full_row_against_every_delta_frequency(tmp_path) -> None: + path = tmp_path / "rows.jsonl" + rows = [ + _row("full", 0), + _row("delta", 0, freq=10, run_p50=80.0), + _row("delta", 0, freq=100, run_p50=90.0), + ] + path.write_text("\n".join(json.dumps(r) for r in rows) + "\n") + summaries = summ._summarize(summ._load_jsonl([path]), metrics=["run_turn_p50_ms"]) + assert len(summaries) == 2 + by_freq = {s["snapshot_frequency"]: s for s in summaries} + assert by_freq[10]["ratio_run_turn_p50_ms"] == 0.8 + assert by_freq[100]["ratio_run_turn_p50_ms"] == 0.9 + + +def test_decision_metrics_emitted(tmp_path) -> None: + path = tmp_path / "rows.jsonl" + rows = [_row("full", 0), _row("delta", 0, freq=10, write_p50=10.0, write_p99=50.0)] + path.write_text("\n".join(json.dumps(r) for r in rows) + "\n") + summaries = summ._summarize(summ._load_jsonl([path]), metrics=["run_turn_p50_ms"]) + assert summaries[0]["delta_snapshot_write_spike"] == 5.0 + assert summaries[0]["delta_cache_effect_ms"] == 45.0 + assert summaries[0]["full_cache_effect_ms"] == 45.0 + + +def test_failed_rows_counted_per_group(tmp_path) -> None: + path = tmp_path / "rows.jsonl" + rows = [ + _row("full", 0), + _row("delta", 0, freq=10), + _row("delta", 0, freq=100), + _row("delta", 1, freq=10, success=False), + _row("delta", 2, freq=10, success=False), + _row("delta", 1, freq=100, success=False), + # Different turns: matches no group below. + _row("delta", 0, freq=10, turns=200, success=False), + # A failed full row counts toward every group at the same (turns, payload_bytes). + _row("full", 1, success=False), + ] + path.write_text("\n".join(json.dumps(r) for r in rows) + "\n") + summaries = summ._summarize(summ._load_jsonl([path]), metrics=["run_turn_p50_ms"]) + by_freq = {s["snapshot_frequency"]: s for s in summaries} + assert by_freq[10]["failed_rows"] == 3 + assert by_freq[100]["failed_rows"] == 2 + + +def test_history_limit_metrics_auto_discovered(tmp_path, capsys) -> None: + path = tmp_path / "rows.jsonl" + full = _row("full", 0) + delta = _row("delta", 0, freq=10) + for row, cold, warm in ((full, 30.0, 12.0), (delta, 60.0, 24.0)): + row["history_limit_64_cold_ms"] = cold + row["history_limit_64_warm_p50_ms"] = warm + path.write_text("\n".join(json.dumps(r) for r in (full, delta)) + "\n") + assert summ.main([str(path), "--json"]) == 0 + out = json.loads(capsys.readouterr().out) + assert out[0]["ratio_history_limit_64_cold_ms"] == 2.0 + assert out[0]["delta_history_limit_64_warm_p50_ms"] == 24.0 + + +def test_explicit_metrics_override_skips_discovery(tmp_path, capsys) -> None: + path = tmp_path / "rows.jsonl" + full = _row("full", 0) + delta = _row("delta", 0, freq=10) + for row in (full, delta): + row["history_limit_64_cold_ms"] = 30.0 + path.write_text("\n".join(json.dumps(r) for r in (full, delta)) + "\n") + assert summ.main([str(path), "--json", "--metrics", "run_turn_p50_ms"]) == 0 + out = json.loads(capsys.readouterr().out) + assert "delta_history_limit_64_cold_ms" not in out[0] + + +def test_checkpoint_write_share_emitted_and_zero_guarded(tmp_path) -> None: + path = tmp_path / "rows.jsonl" + rows = [ + _row("full", 0, run_p50=100.0, write_p50=10.0), + _row("delta", 0, freq=10, run_p50=80.0, write_p50=20.0), + _row("full", 0, turns=200, run_p50=0.0, write_p50=10.0), + _row("delta", 0, freq=10, turns=200, run_p50=0.0, write_p50=20.0), + ] + path.write_text("\n".join(json.dumps(r) for r in rows) + "\n") + summaries = summ._summarize(summ._load_jsonl([path]), metrics=["run_turn_p50_ms"]) + by_turns = {s["turns"]: s for s in summaries} + assert by_turns[100]["full_checkpoint_write_share"] == 0.1 + assert by_turns[100]["delta_checkpoint_write_share"] == 0.25 + assert "full_checkpoint_write_share" not in by_turns[200] + assert "delta_checkpoint_write_share" not in by_turns[200] diff --git a/backend/tests/test_token_usage.py b/backend/tests/test_token_usage.py index bec9e9ac3..fd679329f 100644 --- a/backend/tests/test_token_usage.py +++ b/backend/tests/test_token_usage.py @@ -147,6 +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 return config