mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-28 08:56:13 +00:00
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
This commit is contained in:
Vanzeren
GitHub
parent
625c07b993
commit
e01173d8b2
@ -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.
|
||||
|
||||
@ -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/`)
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
@ -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),
|
||||
)
|
||||
|
||||
@ -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}")
|
||||
|
||||
@ -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."),
|
||||
|
||||
@ -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()), "<home>")
|
||||
if work_dir is not None:
|
||||
message = message.replace(str(work_dir), "<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:
|
||||
716
backend/scripts/benchmark/checkpoint/bench_production.py
Executable file
716
backend/scripts/benchmark/checkpoint/bench_production.py
Executable file
@ -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())
|
||||
208
backend/scripts/benchmark/checkpoint/checkpoint_bench_common.py
Executable file
208
backend/scripts/benchmark/checkpoint/checkpoint_bench_common.py
Executable file
@ -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()), "<home>")
|
||||
if work_dir is not None:
|
||||
message = message.replace(str(work_dir), "<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
|
||||
@ -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
|
||||
"""
|
||||
|
||||
199
backend/scripts/benchmark/checkpoint/summarize_production.py
Executable file
199
backend/scripts/benchmark/checkpoint/summarize_production.py
Executable file
@ -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())
|
||||
@ -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 \\
|
||||
@ -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",
|
||||
@ -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.
|
||||
|
||||
@ -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)
|
||||
|
||||
237
backend/tests/test_bench_checkpoint_production.py
Normal file
237
backend/tests/test_bench_checkpoint_production.py
Normal file
@ -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
|
||||
@ -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:
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
128
backend/tests/test_summarize_checkpoint_production.py
Normal file
128
backend/tests/test_summarize_checkpoint_production.py
Normal file
@ -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]
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user