bench(checkpoint): add channel mode benchmark (#4395)

* bench(checkpoint): add channel mode benchmark

* bench(checkpoint): harden benchmark reporting

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).

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()), "<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:
+        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())

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())

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

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