diff --git a/backend/AGENTS.md b/backend/AGENTS.md index ec7f9f985..22d66d271 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -821,6 +821,44 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c - `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` +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 +correctness digests, write windows/percentiles, warm and graph-rebuilt cold reads, +logical checkpoint/write bytes, SQLite DB/WAL/SHM footprint, reducer replay time, +and peak RSS as versioned JSONL. The controller alternates mode order and rejects +performance data when paired modes materialize different state. Its default 1 GiB +estimated cumulative full-payload cap skips both modes of an oversized pair when +`full` is selected; intentional `--modes delta` diagnostics bypass this +full-payload cap, so size those runs explicitly. Use `--allow-large-cases` only +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 +`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. +Storage-size collection relies on saver-specific diagnostic layouts; if those +layouts change, the timing/correctness row remains successful while storage +fields become `null` and `storage_stats_error` records the diagnostic failure. +Example: + +```bash +cd backend +PYTHONPATH=. uv run python scripts/benchmark/bench_checkpoint_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 \ + /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. + ### Terminal Workbench / TUI (`packages/harness/deerflow/tui/`) A terminal-native UI over the embedded harness, exposed as the `deerflow` console script (`[project.scripts]` in `packages/harness/pyproject.toml`). It is a UI shell over `DeerFlowClient` and does **not** fork agent behavior. `textual` is an optional dependency (`deerflow-harness[tui]`; also in the backend dev group); the console script degrades to headless help when it is absent. Full guide: [docs/TUI.md](docs/TUI.md). diff --git a/backend/scripts/benchmark/bench_checkpoint_channels.py b/backend/scripts/benchmark/bench_checkpoint_channels.py new file mode 100644 index 000000000..17acc0d6c --- /dev/null +++ b/backend/scripts/benchmark/bench_checkpoint_channels.py @@ -0,0 +1,717 @@ +#!/usr/bin/env python3 +"""Benchmark DeerFlow's full and DeltaChannel checkpoint message storage. + +The public CLI is a controller. Every benchmark case runs in a fresh child +process and, for SQLite, a fresh database. This mirrors the restart-required +checkpoint mode boundary and prevents one mode's graph/channel caches from +warming the other. + +Examples:: + + PYTHONPATH=. uv run python scripts/benchmark/bench_checkpoint_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 \ + --backends sqlite --updates 1000 --payload-bytes 128 \ + --repetitions 7 --output snapshot-boundary.jsonl + +The controller suppresses matrix pairs whose estimated cumulative full-mode +message payload exceeds ``--max-estimated-full-bytes``. Both modes are skipped +as a pair so every emitted full/delta result remains comparable. Use +``--allow-large-cases`` only on a machine provisioned for the resulting disk +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 +from collections.abc import Callable +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Annotated, Any, Literal, TypedDict + +from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage +from langgraph.channels import DeltaChannel +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.checkpoint.sqlite import SqliteSaver +from langgraph.graph import StateGraph +from langgraph.graph.message import add_messages + +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] + +Mode = Literal["full", "delta"] +Backend = Literal["memory", "sqlite"] + +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 = ( + "logical_checkpoint_bytes", + "logical_write_bytes", + "checkpoint_rows", + "write_rows", +) + + +class _FullBenchmarkState(TypedDict): + messages: Annotated[list[AnyMessage], add_messages] + + +class _DeltaBenchmarkState(TypedDict): + messages: Annotated[ + list[AnyMessage], + DeltaChannel(merge_message_writes, snapshot_frequency=PRODUCTION_SNAPSHOT_FREQUENCY), + ] + + +@dataclass(frozen=True) +class BenchmarkCase: + mode: Mode + backend: Backend + update_count: int + payload_bytes: int + repetition: int + seed: int + scenario: str = "append" + snapshot_frequency: int = PRODUCTION_SNAPSHOT_FREQUENCY + + def __post_init__(self) -> None: + if self.mode not in _MODES: + raise ValueError(f"unsupported mode: {self.mode!r}") + if self.backend not in _BACKENDS: + raise ValueError(f"unsupported backend: {self.backend!r}") + if self.update_count <= 0: + raise ValueError("update_count must be positive") + if self.payload_bytes <= 0: + raise ValueError("payload_bytes must be positive") + if self.repetition < 0: + raise ValueError("repetition must be non-negative") + if self.snapshot_frequency != PRODUCTION_SNAPSHOT_FREQUENCY: + raise ValueError(f"Phase 1 supports only production snapshot_frequency={PRODUCTION_SNAPSHOT_FREQUENCY}") + if self.scenario != "append": + 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], + backends: list[str], + update_counts: list[int], + payload_bytes: list[int], + repetitions: int, + seed: int, +) -> list[BenchmarkCase]: + """Build a matrix with alternating mode order to reduce order bias.""" + cases: list[BenchmarkCase] = [] + for repetition in range(repetitions): + for backend in backends: + for payload in payload_bytes: + for update_index, update_count in enumerate(update_counts): + ordered_modes = list(modes) + if (repetition + update_index) % 2 == 1: + ordered_modes.reverse() + for mode in ordered_modes: + cases.append( + BenchmarkCase( + mode=mode, # type: ignore[arg-type] + backend=backend, # type: ignore[arg-type] + update_count=update_count, + payload_bytes=payload, + repetition=repetition, + seed=seed, + ) + ) + return cases + + +def _estimated_full_payload_bytes(case: BenchmarkCase) -> int: + """Lower-bound cumulative message content serialized by full mode.""" + return case.payload_bytes * case.update_count * (case.update_count + 1) // 2 + + +def _filter_oversized_pairs(cases: list[BenchmarkCase], *, max_bytes: int | None) -> tuple[list[BenchmarkCase], list[BenchmarkCase]]: + if max_bytes is None: + return cases, [] + group_fields = ("backend", "scenario", "snapshot_frequency", "update_count", "payload_bytes", "repetition") + oversized_keys = {tuple(getattr(case, field) for field in group_fields) for case in cases if case.mode == "full" and _estimated_full_payload_bytes(case) > max_bytes} + kept = [case for case in cases if tuple(getattr(case, field) for field in group_fields) not in oversized_keys] + skipped = [case for case in cases if tuple(getattr(case, field) for field in group_fields) in oversized_keys] + return kept, skipped + + +def _message_for_update(index: int, payload_bytes: int) -> BaseMessage: + content = "x" * payload_bytes + message_id = f"bench-message-{index:08d}" + if index % 2 == 0: + return HumanMessage(id=message_id, content=content) + 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 {} + + +def _build_graph(mode: Mode, saver: Any) -> Any: + schema = _DeltaBenchmarkState if mode == "delta" else _FullBenchmarkState + builder = StateGraph(schema) + builder.add_node("noop", _noop) + builder.set_entry_point("noop") + builder.set_finish_point("noop") + return builder.compile(checkpointer=saver) + + +def _config(case: BenchmarkCase) -> dict[str, Any]: + config: dict[str, Any] = { + "configurable": { + "thread_id": f"checkpoint-bench-{case.seed}-{case.repetition}", + } + } + inject_checkpoint_mode(config, case.mode) + 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: + langgraph_version = importlib.metadata.version("langgraph") + except importlib.metadata.PackageNotFoundError: + langgraph_version = "unknown" + return { + "schema_version": SCHEMA_VERSION, + "benchmark_version": BENCHMARK_VERSION, + "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, + "backend": case.backend, + "scenario": case.scenario, + "snapshot_frequency": case.snapshot_frequency, + "update_count": case.update_count, + "payload_bytes": case.payload_bytes, + "repetition": case.repetition, + "seed": case.seed, + } + + +def _safe_error(error: BaseException | str, *, work_dir: Path | None = None) -> str: + message = str(error).replace(str(Path.home()), "") + if work_dir is not None: + message = message.replace(str(work_dir), "") + return message[:2000] + + +def _collect_storage_stats(collector: Callable[[], dict[str, int]]) -> dict[str, Any]: + """Keep timing data usable when a saver's diagnostic layout changes.""" + try: + return {**collector(), "storage_stats_error": None} + except Exception as exc: + return { + **dict.fromkeys(_STORAGE_STAT_FIELDS), + "storage_stats_error": _safe_error(exc), + } + + +def _memory_storage_stats(saver: InMemorySaver, thread_id: str) -> dict[str, int]: + checkpoint_rows = 0 + checkpoint_bytes = 0 + for namespace in saver.storage.get(thread_id, {}).values(): + for checkpoint, metadata, _parent_id in namespace.values(): + checkpoint_rows += 1 + checkpoint_bytes += len(checkpoint[1]) + len(metadata[1]) + for (stored_thread_id, _namespace, _channel, _version), (_type_tag, blob) in saver.blobs.items(): + if stored_thread_id == thread_id: + checkpoint_bytes += len(blob) + + write_rows = 0 + write_bytes = 0 + for (stored_thread_id, _namespace, _checkpoint_id), writes in saver.writes.items(): + if stored_thread_id != thread_id: + continue + for _task_id, _channel, (_type_tag, blob), _task_path in writes.values(): + write_rows += 1 + write_bytes += len(blob) + return { + "logical_checkpoint_bytes": checkpoint_bytes, + "logical_write_bytes": write_bytes, + "checkpoint_rows": checkpoint_rows, + "write_rows": write_rows, + } + + +def _sqlite_storage_stats(saver: SqliteSaver, thread_id: str) -> dict[str, int]: + with saver.cursor(transaction=False) as cursor: + checkpoint_rows, checkpoint_bytes = cursor.execute( + "SELECT COUNT(*), COALESCE(SUM(length(checkpoint) + length(metadata)), 0) FROM checkpoints WHERE thread_id = ?", + (thread_id,), + ).fetchone() + write_rows, write_bytes = cursor.execute( + "SELECT COUNT(*), COALESCE(SUM(length(value)), 0) FROM writes WHERE thread_id = ?", + (thread_id,), + ).fetchone() + return { + "logical_checkpoint_bytes": int(checkpoint_bytes), + "logical_write_bytes": int(write_bytes), + "checkpoint_rows": int(checkpoint_rows), + "write_rows": int(write_rows), + } + + +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) + config = _config(case) + update_latencies: list[float] = [] + write_start = time.perf_counter() + for message in messages: + update_start = time.perf_counter() + graph.invoke({"messages": [message]}, config) + update_latencies.append((time.perf_counter() - update_start) * 1000) + write_total_ms = (time.perf_counter() - write_start) * 1000 + + warm_start = time.perf_counter() + snapshot = accessor.get(config) + warm_read_ms = (time.perf_counter() - warm_start) * 1000 + materialized = list(snapshot.values.get("messages", [])) + metrics = { + "write_total_ms": write_total_ms, + "write_p50_ms": _percentile(update_latencies, 50), + "write_p95_ms": _percentile(update_latencies, 95), + "write_p99_ms": _percentile(update_latencies, 99), + "write_first_window_ms": _window_median(update_latencies, "first"), + "write_middle_window_ms": _window_median(update_latencies, "middle"), + "write_last_window_ms": _window_median(update_latencies, "last"), + "warm_read_ms": warm_read_ms, + } + return metrics, materialized + + +def _cold_read(case: BenchmarkCase, saver: Any) -> tuple[float, list[AnyMessage]]: + graph = _build_graph(case.mode, saver) + accessor = CheckpointStateAccessor.bind(graph, saver, mode=case.mode) + gc.collect() + start = time.perf_counter() + snapshot = accessor.get(_config(case)) + elapsed_ms = (time.perf_counter() - start) * 1000 + return elapsed_ms, list(snapshot.values.get("messages", [])) + + +def _validate_materialized(case: BenchmarkCase, expected: list[BaseMessage], warm: list[AnyMessage], cold: list[AnyMessage]) -> tuple[int, str]: + expected_digest = _canonical_messages_digest(expected) + warm_digest = _canonical_messages_digest(warm) + cold_digest = _canonical_messages_digest(cold) + if len(warm) != case.update_count or len(cold) != case.update_count: + raise AssertionError(f"expected {case.update_count} messages, materialized warm={len(warm)} cold={len(cold)}") + if warm_digest != expected_digest or cold_digest != expected_digest: + raise AssertionError("materialized message content or ordering differs from deterministic input") + return len(cold), cold_digest + + +def _run_memory_case(case: BenchmarkCase, messages: list[BaseMessage]) -> dict[str, Any]: + saver = InMemorySaver() + metrics, warm = _write_and_read(case, saver, messages) + stats = _collect_storage_stats(lambda: _memory_storage_stats(saver, _config(case)["configurable"]["thread_id"])) + cold_read_ms, cold = _cold_read(case, saver) + actual_count, digest = _validate_materialized(case, messages, warm, cold) + return { + **metrics, + **stats, + "cold_read_ms": cold_read_ms, + # InMemorySaver has no durable external storage to reopen. The cold + # sample rebuilds the graph/channel table over the same saver only. + "saver_reopen_ms": 0.0, + "db_bytes": None, + "wal_bytes": None, + "shm_bytes": None, + "durable_db_bytes": None, + "expected_message_count": case.update_count, + "actual_message_count": actual_count, + "content_sha256": digest, + } + + +def _run_sqlite_case(case: BenchmarkCase, messages: list[BaseMessage], db_path: Path) -> dict[str, Any]: + with SqliteSaver.from_conn_string(str(db_path)) as saver: + saver.setup() + metrics, warm = _write_and_read(case, saver, messages) + stats = _collect_storage_stats(lambda: _sqlite_storage_stats(saver, _config(case)["configurable"]["thread_id"])) + db_bytes = _file_size(db_path) + wal_bytes = _file_size(Path(f"{db_path}-wal")) + shm_bytes = _file_size(Path(f"{db_path}-shm")) + + durable_db_bytes = _file_size(db_path) + reopen_start = time.perf_counter() + with SqliteSaver.from_conn_string(str(db_path)) as reopened: + reopened.setup() + saver_reopen_ms = (time.perf_counter() - reopen_start) * 1000 + cold_read_ms, cold = _cold_read(case, reopened) + + actual_count, digest = _validate_materialized(case, messages, warm, cold) + return { + **metrics, + **stats, + "cold_read_ms": cold_read_ms, + "saver_reopen_ms": saver_reopen_ms, + "db_bytes": db_bytes, + "wal_bytes": wal_bytes, + "shm_bytes": shm_bytes, + "durable_db_bytes": durable_db_bytes, + "expected_message_count": case.update_count, + "actual_message_count": actual_count, + "content_sha256": digest, + } + + +def _run_case(case: BenchmarkCase, *, work_dir: Path) -> dict[str, Any]: + row = _base_row(case) + messages = [_message_for_update(index, case.payload_bytes) for index in range(case.update_count)] + try: + if case.backend == "memory": + measured = _run_memory_case(case, messages) + else: + measured = _run_sqlite_case(case, messages, work_dir / "checkpoint-benchmark.sqlite") + + reducer_writes = [[message] for message in messages] + reducer_start = time.perf_counter() + reduced = merge_message_writes([], reducer_writes) + reducer_replay_ms = (time.perf_counter() - reducer_start) * 1000 + if _canonical_messages_digest(reduced) != _canonical_messages_digest(messages): + raise AssertionError("standalone reducer diagnostic produced incorrect state") + + row.update(measured) + row["reducer_replay_ms"] = reducer_replay_ms + row["peak_rss_bytes"] = _peak_rss_bytes() + except Exception as exc: + row["success"] = False + row["error"] = _safe_error(exc, work_dir=work_dir) + return row + + +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 + + +def _comparison_key(row: dict[str, Any]) -> tuple[Any, ...]: + return tuple( + row.get(field) + for field in ( + "backend", + "scenario", + "snapshot_frequency", + "update_count", + "payload_bytes", + "repetition", + ) + ) + + +def _validate_cross_mode_rows(rows: list[dict[str, Any]]) -> None: + grouped: dict[tuple[Any, ...], list[dict[str, Any]]] = {} + 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("actual_message_count"), row.get("content_sha256")) for row in successful if row.get("mode") in {"full", "delta"}} + if len(signatures) == 1: + continue + for row in successful: + row["success"] = False + row["error"] = "cross-mode materialized state mismatch" + + +def _failure_row(case: BenchmarkCase, error: str) -> dict[str, Any]: + row = _base_row(case) + row["success"] = False + row["error"] = _safe_error(error) + return row + + +def _profile_filename(case: BenchmarkCase) -> str: + return f"{case.backend}-{case.mode}-updates-{case.update_count}-payload-{case.payload_bytes}-rep-{case.repetition}.prof" + + +def _run_child_case(case: BenchmarkCase, *, timeout_seconds: float, git_sha: str, profile_dir: Path | None = None) -> dict[str, Any]: + encoded_case = json.dumps(asdict(case), separators=(",", ":")) + command = [sys.executable, str(Path(__file__).resolve()), "--worker-case", encoded_case] + 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 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Benchmark full and delta checkpoint message channels") + parser.add_argument("--modes", default="full,delta", help="Comma-separated modes (default: full,delta)") + parser.add_argument("--backends", default="memory,sqlite", help="Comma-separated backends (default: memory,sqlite)") + parser.add_argument("--updates", default="10,100", help="Comma-separated message update counts (default: 10,100)") + parser.add_argument("--payload-bytes", default="128", help="Comma-separated exact message content sizes (default: 128)") + parser.add_argument("--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( + "--max-estimated-full-bytes", + type=int, + default=DEFAULT_MAX_ESTIMATED_FULL_BYTES, + help=("Skip comparable pairs whose estimated cumulative full-mode payload exceeds this value. The cap applies only when full mode is selected; delta-only diagnostics bypass it."), + ) + parser.add_argument("--allow-large-cases", action="store_true", help="Disable the estimated cumulative full-payload safety cap") + parser.add_argument("--profile-dir", type=Path, help="Write one cProfile file per case; profiling inflates timings") + 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 _worker_main(encoded_case: str, *, profile_path: Path | None = None) -> int: + try: + case = BenchmarkCase(**json.loads(encoded_case)) + except (TypeError, ValueError, json.JSONDecodeError) as exc: + print(json.dumps({"schema_version": SCHEMA_VERSION, "benchmark_version": BENCHMARK_VERSION, "success": False, "error": _safe_error(exc)}, separators=(",", ":"))) + return 2 + with tempfile.TemporaryDirectory(prefix="deerflow-checkpoint-benchmark-") as temp_dir: + if profile_path is None: + row = _run_case(case, work_dir=Path(temp_dir)) + else: + row = _run_profiled_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 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 = _parse_choice_csv(args.modes, option="--modes", choices=_MODES) + backends = _parse_choice_csv(args.backends, option="--backends", choices=_BACKENDS) + updates = _parse_positive_int_csv(args.updates, option="--updates") + payload_bytes = _parse_positive_int_csv(args.payload_bytes, option="--payload-bytes") + if args.repetitions <= 0: + raise ValueError("--repetitions must be positive") + if args.timeout_seconds <= 0: + raise ValueError("--timeout-seconds must be positive") + if not args.allow_large_cases and args.max_estimated_full_bytes <= 0: + raise ValueError("--max-estimated-full-bytes must be positive") + except ValueError as exc: + parser.error(str(exc)) + + cases = _expand_cases( + modes=modes, + backends=backends, + update_counts=updates, + payload_bytes=payload_bytes, + repetitions=args.repetitions, + seed=args.seed, + ) + cases, skipped = _filter_oversized_pairs(cases, max_bytes=None if args.allow_large_cases else args.max_estimated_full_bytes) + if skipped: + skipped_pairs = len(skipped) // max(1, len(modes)) + print( + f"Skipping {skipped_pairs} oversized comparable case pair(s); use --allow-large-cases to run them.", + file=sys.stderr, + ) + if not cases: + print("No benchmark cases remain after applying the safety cap.", file=sys.stderr) + return 2 + + git_sha = _resolve_git_sha() + rows: list[dict[str, Any]] = [] + for index, case in enumerate(cases, start=1): + print( + f"[{index}/{len(cases)}] {case.backend} {case.mode} updates={case.update_count} payload={case.payload_bytes} repetition={case.repetition}", + file=sys.stderr, + ) + rows.append(_run_child_case(case, timeout_seconds=args.timeout_seconds, git_sha=git_sha, profile_dir=args.profile_dir)) + _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 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/scripts/benchmark/summarize_checkpoint_channels.py b/backend/scripts/benchmark/summarize_checkpoint_channels.py new file mode 100644 index 000000000..8db2316f6 --- /dev/null +++ b/backend/scripts/benchmark/summarize_checkpoint_channels.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""Summarize paired full/delta checkpoint benchmark JSONL results. + +Only repetitions with one successful ``full`` row and one successful ``delta`` +row enter metric medians. This prevents a failed or missing mode from turning +different workloads into a misleading ratio. Ratios are always ``delta/full``: +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 \ + --metrics write_total_ms,cold_read_ms,logical_checkpoint_bytes --csv +""" + +from __future__ import annotations + +import argparse +import csv +import json +import statistics +import sys +from collections import defaultdict +from pathlib import Path +from typing import Any + +DEFAULT_METRICS = [ + "write_total_ms", + "write_p50_ms", + "write_p95_ms", + "write_last_window_ms", + "warm_read_ms", + "cold_read_ms", + "logical_checkpoint_bytes", + "logical_write_bytes", + "durable_db_bytes", + "reducer_replay_ms", + "peak_rss_bytes", +] + +_GROUP_FIELDS = ( + "backend", + "scenario", + "snapshot_frequency", + "update_count", + "payload_bytes", +) +_INPUT_INDEX_FIELD = "_benchmark_input_index" + + +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 + # Repetition numbers restart at zero for every benchmark + # invocation. Keep the source input in the in-memory row so + # separate result files cannot overwrite or pair each other. + row[_INPUT_INDEX_FIELD] = input_index + rows.append(row) + if errors: + raise ValueError("Malformed JSONL row(s):\n" + "\n".join(errors)) + return rows + + +def _group_key(row: dict[str, Any]) -> tuple[Any, ...]: + return tuple(row.get(field) for field in _GROUP_FIELDS) + + +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 _group_sort_key(key: tuple[Any, ...]) -> tuple[tuple[int, Any], ...]: + return tuple((0, value) if isinstance(value, (int, float)) and not isinstance(value, bool) else (1, str(value)) for value in key) + + +def _summarize(rows: list[dict[str, Any]], *, metrics: list[str]) -> list[dict[str, Any]]: + groups: dict[tuple[Any, ...], list[dict[str, Any]]] = defaultdict(list) + for row in rows: + if row.get("profiled"): + continue + groups[_group_key(row)].append(row) + + summaries: list[dict[str, Any]] = [] + for key, group in sorted(groups.items(), key=lambda item: _group_sort_key(item[0])): + by_repetition: dict[Any, dict[str, dict[str, Any]]] = defaultdict(dict) + for row in group: + if row.get("success") and row.get("mode") in {"full", "delta"}: + repetition_key = (row.get(_INPUT_INDEX_FIELD), row.get("repetition")) + by_repetition[repetition_key][row["mode"]] = row + pairs = [pair for pair in by_repetition.values() if "full" in pair and "delta" in pair] + if not pairs: + continue + + summary: dict[str, Any] = {field: key[index] for index, field in enumerate(_GROUP_FIELDS)} + summary["paired_repetitions"] = len(pairs) + summary["failed_rows"] = sum(1 for row in group if not row.get("success")) + for metric in metrics: + full_values: list[float] = [] + delta_values: list[float] = [] + for pair in pairs: + full_value = _numeric(pair["full"], metric) + delta_value = _numeric(pair["delta"], metric) + if full_value is None or delta_value is None: + continue + full_values.append(full_value) + delta_values.append(delta_value) + if not full_values: + continue + full_median = statistics.median(full_values) + delta_median = statistics.median(delta_values) + 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 + 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 _print_csv(rows: list[dict[str, Any]]) -> None: + if not rows: + return + writer = csv.DictWriter(sys.stdout, fieldnames=_columns(rows), extrasaction="ignore") + writer.writeheader() + writer.writerows(rows) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Summarize paired full/delta checkpoint benchmark results") + parser.add_argument("inputs", nargs="+", type=Path) + parser.add_argument("--metrics", default=",".join(DEFAULT_METRICS), help="Comma-separated numeric result fields") + 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) + + metrics = [metric.strip() for metric in args.metrics.split(",") if metric.strip()] + if not metrics: + parser.error("--metrics must contain at least one field") + try: + rows = _load_jsonl(args.inputs) + except (OSError, ValueError) as exc: + print(str(exc), file=sys.stderr) + return 1 + profiled_rows = sum(1 for row in rows if row.get("profiled")) + if profiled_rows: + print(f"Skipping {profiled_rows} profiled row(s); profiled timings are excluded from baseline summaries.", file=sys.stderr) + summaries = _summarize(rows, metrics=metrics) + if args.json: + json.dump(summaries, sys.stdout, ensure_ascii=False, indent=2) + print() + elif args.csv: + _print_csv(summaries) + else: + _print_table(summaries) + return 0 if summaries else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_bench_checkpoint_channels.py b/backend/tests/test_bench_checkpoint_channels.py new file mode 100644 index 000000000..52f6a2254 --- /dev/null +++ b/backend/tests/test_bench_checkpoint_channels.py @@ -0,0 +1,281 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + + +def _load_module(): + path = Path(__file__).resolve().parents[1] / "scripts/benchmark/bench_checkpoint_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) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +bench = _load_module() + + +def test_parse_positive_int_csv_deduplicates_in_input_order(capsys: pytest.CaptureFixture[str]) -> None: + assert bench._parse_positive_int_csv("100,10,100,500", option="--updates") == [100, 10, 500] + assert "ignored duplicate value(s): 100" in capsys.readouterr().err + + +def test_parse_choice_csv_reports_duplicate_values(capsys: pytest.CaptureFixture[str]) -> None: + assert bench._parse_choice_csv("full,full,delta", option="--modes", choices=("full", "delta")) == ["full", "delta"] + assert "ignored duplicate value(s): full" in capsys.readouterr().err + + +@pytest.mark.parametrize("value", ["", "0", "-1", "one", "1,,2"]) +def test_parse_positive_int_csv_rejects_invalid_values(value: str) -> None: + with pytest.raises(ValueError, match="--updates"): + bench._parse_positive_int_csv(value, option="--updates") + + +def test_deterministic_message_has_exact_payload_bytes_and_stable_identity() -> None: + first = bench._message_for_update(3, 128) + second = bench._message_for_update(3, 128) + + assert first == second + assert first.id == "bench-message-00000003" + assert first.type == "ai" + assert len(first.content.encode("utf-8")) == 128 + + +def test_expand_cases_alternates_modes_without_cross_product_reordering() -> None: + cases = bench._expand_cases( + modes=["full", "delta"], + backends=["sqlite"], + update_counts=[10, 100], + payload_bytes=[128], + repetitions=2, + seed=7, + ) + + assert [(case.repetition, case.update_count, case.mode) for case in cases] == [ + (0, 10, "full"), + (0, 10, "delta"), + (0, 100, "delta"), + (0, 100, "full"), + (1, 10, "delta"), + (1, 10, "full"), + (1, 100, "full"), + (1, 100, "delta"), + ] + + +def test_oversized_filter_skips_full_and_delta_as_a_comparable_pair() -> None: + cases = bench._expand_cases( + modes=["full", "delta"], + backends=["memory"], + update_counts=[10, 100], + payload_bytes=[128], + repetitions=1, + seed=1, + ) + + kept, skipped = bench._filter_oversized_pairs(cases, max_bytes=100_000) + + assert {(case.update_count, case.mode) for case in kept} == {(10, "full"), (10, "delta")} + assert {(case.update_count, case.mode) for case in skipped} == {(100, "full"), (100, "delta")} + + +def test_oversized_filter_does_not_suppress_a_delta_only_diagnostic() -> None: + case = bench.BenchmarkCase( + mode="delta", + backend="memory", + update_count=2000, + payload_bytes=4096, + repetition=0, + seed=1, + ) + + kept, skipped = bench._filter_oversized_pairs([case], max_bytes=1) + + assert kept == [case] + assert skipped == [] + + +def test_help_explains_delta_only_runs_bypass_full_payload_cap() -> None: + assert "delta-only" in bench._build_parser().format_help() + + +def test_cross_mode_validation_rejects_materialized_state_mismatch() -> None: + rows = [ + { + "success": True, + "mode": "full", + "backend": "sqlite", + "scenario": "append", + "snapshot_frequency": 1000, + "update_count": 10, + "payload_bytes": 128, + "repetition": 0, + "actual_message_count": 10, + "content_sha256": "full-digest", + }, + { + "success": True, + "mode": "delta", + "backend": "sqlite", + "scenario": "append", + "snapshot_frequency": 1000, + "update_count": 10, + "payload_bytes": 128, + "repetition": 0, + "actual_message_count": 10, + "content_sha256": "delta-digest", + }, + ] + + bench._validate_cross_mode_rows(rows) + + assert all(row["success"] is False for row in rows) + assert all("cross-mode" in row["error"] for row in rows) + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +def test_memory_smoke_case_materializes_expected_state(mode: str, tmp_path: Path) -> None: + case = bench.BenchmarkCase( + mode=mode, + backend="memory", + update_count=4, + payload_bytes=64, + repetition=0, + seed=1, + ) + + row = bench._run_case(case, work_dir=tmp_path) + + assert row["success"] is True + assert row["expected_message_count"] == 4 + assert row["actual_message_count"] == 4 + assert row["warm_read_ms"] >= 0 + assert row["cold_read_ms"] >= 0 + assert row["saver_reopen_ms"] == 0 + assert len(row["content_sha256"]) == 64 + assert row["db_bytes"] is None + assert row["logical_checkpoint_bytes"] is not None + + +def test_memory_case_keeps_timings_when_private_storage_stats_change(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + case = bench.BenchmarkCase( + mode="delta", + backend="memory", + update_count=2, + payload_bytes=32, + repetition=0, + seed=1, + ) + + def fail_storage_stats(*_args, **_kwargs): + raise AttributeError("private saver layout changed") + + monkeypatch.setattr(bench, "_memory_storage_stats", fail_storage_stats) + + row = bench._run_case(case, work_dir=tmp_path) + + assert row["success"] is True + assert row["write_total_ms"] >= 0 + assert row["logical_checkpoint_bytes"] is None + assert row["logical_write_bytes"] is None + assert row["checkpoint_rows"] is None + assert row["write_rows"] is None + assert "private saver layout changed" in row["storage_stats_error"] + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +def test_sqlite_smoke_case_reports_durable_and_logical_storage(mode: str, tmp_path: Path) -> None: + case = bench.BenchmarkCase( + mode=mode, + backend="sqlite", + update_count=3, + payload_bytes=64, + repetition=0, + seed=2, + ) + + row = bench._run_case(case, work_dir=tmp_path) + + assert row["success"] is True + assert row["db_bytes"] > 0 + assert row["durable_db_bytes"] > 0 + assert row["logical_checkpoint_bytes"] > 0 + assert row["logical_write_bytes"] > 0 + assert row["checkpoint_rows"] > 0 + assert row["write_rows"] > 0 + + +def test_controller_writes_versioned_jsonl_without_sensitive_case_paths(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + output = tmp_path / "result.jsonl" + child_git_shas = [] + + def fake_run_child(case, *, timeout_seconds, git_sha, profile_dir=None): + child_git_shas.append(git_sha) + return { + "schema_version": 1, + "benchmark_version": 1, + "success": True, + "error": None, + "mode": case.mode, + "backend": case.backend, + "scenario": case.scenario, + "snapshot_frequency": case.snapshot_frequency, + "update_count": case.update_count, + "payload_bytes": case.payload_bytes, + "repetition": case.repetition, + "actual_message_count": case.update_count, + "content_sha256": "same", + } + + monkeypatch.setattr(bench, "_run_child_case", fake_run_child) + monkeypatch.setattr(bench, "_resolve_git_sha", lambda: "controller-sha") + + rc = bench.main( + [ + "--modes", + "full,delta", + "--backends", + "memory", + "--updates", + "2", + "--payload-bytes", + "32", + "--repetitions", + "1", + "--output", + str(output), + ] + ) + + assert rc == 0 + rows = [json.loads(line) for line in output.read_text(encoding="utf-8").splitlines()] + assert [row["mode"] for row in rows] == ["full", "delta"] + assert all(row["schema_version"] == 1 for row in rows) + assert all("work_dir" not in row and "database_path" not in row for row in rows) + assert child_git_shas == ["controller-sha", "controller-sha"] + + +def test_profiled_case_writes_loadable_stats(tmp_path: Path) -> None: + case = bench.BenchmarkCase( + mode="delta", + backend="memory", + update_count=2, + payload_bytes=32, + repetition=0, + seed=3, + ) + profile_path = tmp_path / bench._profile_filename(case) + + row = bench._run_profiled_case(case, work_dir=tmp_path / "work", profile_path=profile_path) + + assert row["success"] is True + assert row["profiled"] is True + assert profile_path.is_file() + assert profile_path.stat().st_size > 0 diff --git a/backend/tests/test_summarize_checkpoint_channels.py b/backend/tests/test_summarize_checkpoint_channels.py new file mode 100644 index 000000000..d5bb76341 --- /dev/null +++ b/backend/tests/test_summarize_checkpoint_channels.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + + +def _load_module(): + path = Path(__file__).resolve().parents[1] / "scripts/benchmark/summarize_checkpoint_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) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +summarize = _load_module() + + +def _row(mode: str, repetition: int, write_ms: float, checkpoint_bytes: int, *, success: bool = True) -> dict: + return { + "success": success, + "mode": mode, + "backend": "sqlite", + "scenario": "append", + "snapshot_frequency": 1000, + "update_count": 100, + "payload_bytes": 128, + "repetition": repetition, + "write_total_ms": write_ms, + "logical_checkpoint_bytes": checkpoint_bytes, + } + + +def test_summarize_uses_only_successful_paired_repetitions() -> None: + rows = [ + _row("full", 0, 10, 1000), + _row("delta", 0, 5, 200), + _row("full", 1, 30, 3000), + _row("delta", 1, 15, 600), + _row("full", 2, 999, 9999), + _row("delta", 2, 1, 1, success=False), + ] + + result = summarize._summarize(rows, metrics=["write_total_ms", "logical_checkpoint_bytes"]) + + assert result == [ + { + "backend": "sqlite", + "scenario": "append", + "snapshot_frequency": 1000, + "update_count": 100, + "payload_bytes": 128, + "paired_repetitions": 2, + "failed_rows": 1, + "full_write_total_ms": 20.0, + "delta_write_total_ms": 10.0, + "ratio_write_total_ms": 0.5, + "full_logical_checkpoint_bytes": 2000.0, + "delta_logical_checkpoint_bytes": 400.0, + "ratio_logical_checkpoint_bytes": 0.2, + } + ] + + +def test_summarize_omits_group_without_a_successful_pair() -> None: + assert summarize._summarize([_row("full", 0, 10, 1000)], metrics=["write_total_ms"]) == [] + + +def test_summarize_excludes_profiled_pairs_from_baseline_medians() -> None: + rows = [ + _row("full", 0, 10, 1000), + _row("delta", 0, 5, 200), + {**_row("full", 1, 1000, 1000), "profiled": True}, + {**_row("delta", 1, 1000, 200), "profiled": True}, + ] + + result = summarize._summarize(rows, metrics=["write_total_ms"]) + + assert result[0]["paired_repetitions"] == 1 + assert result[0]["full_write_total_ms"] == 10.0 + assert result[0]["delta_write_total_ms"] == 5.0 + + +def test_summarize_sorts_numeric_update_counts_numerically() -> None: + rows = [] + for update_count in (10, 2): + full = _row("full", 0, 10, 1000) + delta = _row("delta", 0, 5, 200) + full["update_count"] = update_count + delta["update_count"] = update_count + rows.extend([full, delta]) + + result = summarize._summarize(rows, metrics=["write_total_ms"]) + + assert [row["update_count"] for row in result] == [2, 10] + + +def test_load_jsonl_reports_file_and_line_for_malformed_input(tmp_path: Path) -> None: + path = tmp_path / "results.jsonl" + path.write_text('{"success": true}\nnot-json\n', encoding="utf-8") + + with pytest.raises(ValueError, match=r"results\.jsonl:2"): + summarize._load_jsonl([path]) + + +def test_multiple_inputs_keep_same_numbered_repetitions_separate(tmp_path: Path) -> None: + first = tmp_path / "first.jsonl" + second = tmp_path / "second.jsonl" + first.write_text( + "".join(json.dumps(row) + "\n" for row in [_row("full", 0, 10, 1000), _row("delta", 0, 5, 200)]), + encoding="utf-8", + ) + second.write_text( + "".join(json.dumps(row) + "\n" for row in [_row("full", 0, 30, 3000), _row("delta", 0, 15, 600)]), + encoding="utf-8", + ) + + result = summarize._summarize(summarize._load_jsonl([first, second]), metrics=["write_total_ms"]) + + assert result[0]["paired_repetitions"] == 2 + assert result[0]["full_write_total_ms"] == 20.0 + assert result[0]["delta_write_total_ms"] == 10.0 + + +def test_multiple_inputs_do_not_cross_pair_single_mode_results(tmp_path: Path) -> None: + full_only = tmp_path / "full.jsonl" + delta_only = tmp_path / "delta.jsonl" + full_only.write_text(json.dumps(_row("full", 0, 10, 1000)) + "\n", encoding="utf-8") + delta_only.write_text(json.dumps(_row("delta", 0, 5, 200)) + "\n", encoding="utf-8") + + rows = summarize._load_jsonl([full_only, delta_only]) + + assert summarize._summarize(rows, metrics=["write_total_ms"]) == [] + + +def test_main_writes_json_summary(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + path = tmp_path / "results.jsonl" + rows = [_row("full", 0, 10, 1000), _row("delta", 0, 5, 200)] + path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + + rc = summarize.main([str(path), "--metrics", "write_total_ms", "--json"]) + + assert rc == 0 + output = json.loads(capsys.readouterr().out) + assert output[0]["ratio_write_total_ms"] == 0.5 + + +def test_main_warns_when_profiled_rows_are_skipped(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + path = tmp_path / "results.jsonl" + rows = [ + _row("full", 0, 10, 1000), + _row("delta", 0, 5, 200), + {**_row("full", 1, 1000, 1000), "profiled": True}, + {**_row("delta", 1, 1000, 200), "profiled": True}, + ] + path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + + rc = summarize.main([str(path), "--metrics", "write_total_ms", "--json"]) + + captured = capsys.readouterr() + assert rc == 0 + assert "Skipping 2 profiled row(s)" in captured.err