codebase-memory-mcp/scripts/benchmark-incremental-speed.py

#!/usr/bin/env python3
"""Measure fast-mode exact incremental indexing against a fresh full rebuild.

This is an explicit opt-in performance gate. It creates a synthetic Go repo in
a temporary work root, uses an isolated CBM_CACHE_DIR, enables disk incremental
indexing only for that cache, and removes only paths it created.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import queue
import re
import shutil
import sqlite3
import subprocess
import sys
import tempfile
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


DEFAULT_FILE_COUNT = 240
DEFAULT_FUNCTIONS_PER_FILE = 12
DEFAULT_CHANGED_FILES = 2
DEFAULT_MIN_SPEEDUP = 10.0
DEFAULT_TIMEOUT_SECONDS = 240
DEFAULT_RANK_REFRESH = "stale_on_exact"
DEFAULT_OVERHEAD_PROBES = 0
DEFAULT_OVERHEAD_TOOL = "index_status"
DEFAULT_FRONTIER_FILES = 16
DEFAULT_FASTAPI_URL = "https://github.com/fastapi/fastapi.git"
CONFIG_PROFILE_DEFAULT = "default"
CONFIG_PROFILE_RANK_DISABLED = "rank_disabled"
CONFIG_PROFILE_OPTIONAL_GRAPH_DISABLED = "optional_graph_disabled"
CONFIG_PROFILE_MINIMAL_INDEXING = "minimal_indexing"
CONFIG_PROFILES: dict[str, dict[str, str]] = {
    CONFIG_PROFILE_DEFAULT: {},
    CONFIG_PROFILE_RANK_DISABLED: {"rank_enabled": "false"},
    CONFIG_PROFILE_OPTIONAL_GRAPH_DISABLED: {
        "githistory_enabled": "false",
        "httplinks_enabled": "false",
        "rank_enabled": "false",
        "semantic_edges_enabled": "false",
        "similarity_enabled": "false",
    },
    CONFIG_PROFILE_MINIMAL_INDEXING: {
        "auto_index_deps": "false",
        "githistory_enabled": "false",
        "httplinks_enabled": "false",
        "rank_enabled": "false",
        "semantic_edges_enabled": "false",
        "similarity_enabled": "false",
    },
}
PROJECT_DB_SUFFIX = ".db"
CONFIG_DB_NAME = "_config.db"
LOG_TAIL_LINES = 24
FAILURE_TAIL_LINES = 80
FAILURE_ARTIFACT_DIRNAME = "failures"
FAILURE_FALLBACK_DIRNAME = "cbm-benchmark-failures"
FAILURE_TIMESTAMP_FORMAT = "%Y%m%dT%H%M%SZ"
MCP_INIT_PROTOCOL_VERSION = "2024-11-05"
MATRIX_SCENARIOS_DEFAULT = "go_modify_1,go_modify_2,go_create,go_delete,go_rename,go_new_folder,route_decorator,python_reexport"
MATRIX_REAL_REPO_SCENARIOS = frozenset({"fastapi_insert_probe"})
CROSS_FILE_RESOLVER_LANGUAGES = (
    "go",
    "c",
    "cpp",
    "cuda",
    "python",
    "javascript",
    "typescript",
    "tsx",
    "php",
    "csharp",
    "java",
    "kotlin",
    "rust",
)
SCOPED_EXACT_FRONTIER_LANGUAGES = frozenset({"go", "c", "cpp", "cuda", "python"})
MATRIX_FRONTIER_SCENARIOS = {
    "go_inbound_frontier": "go",
    "python_inbound_frontier": "python",
    "c_header_inbound_frontier": "c_header",
    "cpp_inbound_frontier": "cpp",
    "cuda_inbound_frontier": "cuda",
    "javascript_inbound_frontier": "javascript",
    "typescript_inbound_frontier": "typescript",
    "tsx_inbound_frontier": "tsx",
    "php_inbound_frontier": "php",
    "csharp_inbound_frontier": "csharp",
    "java_inbound_frontier": "java",
    "kotlin_inbound_frontier": "kotlin",
    "rust_inbound_frontier": "rust",
}
SELF_DOGFOOD_SCENARIOS_DEFAULT = "noop,one_source_file,route_handler,store_pipeline_batch,multi_file_small"
SELF_DOGFOOD_MARKER_PREFIX = "cbm_pan4_oracle"
SELF_DOGFOOD_REPO_SUBDIR = "repo"
SELF_DOGFOOD_CACHE_SUBDIR = "cache"
FASTAPI_PROBE_REL_PATH = "fastapi/routing.py"
FASTAPI_PROBE_INSERT_BEFORE = "\n    def add_api_route(\n"
FASTAPI_PROBE_RETURN_VALUE = 64
PUBLISH_FULL = "full"
PUBLISH_INCREMENTAL_NOOP = "incremental_noop"
PUBLISH_INCREMENTAL_EXACT = "incremental_exact"
PUBLISH_INCREMENTAL_OVERLAY = "incremental_overlay"
PUBLISH_INCREMENTAL_CONTAINMENT = "incremental_containment"
OVERLAY_STATUS_READY = "overlay_ready"  # CBM_STORE_OVERLAY_STATUS_READY
OVERLAY_TOMBSTONE_FILE = "file"  # CBM_STORE_OVERLAY_TOMBSTONE_FILE
OVERLAY_TOMBSTONE_ACTIVE = 1  # STORE_OVERLAY_TOMBSTONE_ACTIVE
OVERLAY_ROW_OWNED = 1  # STORE_OVERLAY_ROW_OWNED
SOURCE_SPAN_LABELS = frozenset(
    {
        "Function",
        "Method",
        "Class",
        "Struct",
        "Interface",
        "Enum",
        "Type",
        "Trait",
        "Module",
    }
)
LOG_MARKER_PIPELINE_DONE = "pipeline.done"
LOG_MARKER_INCREMENTAL_DONE = "incremental.done"
LOG_MARKER_EXACT_DONE = "incremental.exact.done"
LOG_MARKER_EXACT_FRONTIER = "incremental.exact.frontier"
LOG_MARKER_EXACT_FALLBACK = "incremental.exact.fallback"
LOG_MARKER_EXACT_DELETE_FALLBACK = "incremental.exact.delete.fallback"
LOG_MARKER_EXACT_SKIP = "incremental.exact.skip"


class BenchmarkCommandError(RuntimeError):
    def __init__(self, message: str, detail: dict[str, Any]) -> None:
        super().__init__(message)
        self.detail = detail


def now_ms() -> float:
    return time.perf_counter() * 1000.0


def write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def go_file_content(index: int, revision: int, funcs_per_file: int) -> str:
    lines = ["package main", ""]
    for func_index in range(funcs_per_file):
        value = index * funcs_per_file + func_index + revision
        lines.extend(
            [
                f"func Func{index:04d}_{func_index:02d}() int {{",
                f"\treturn {value}",
                "}",
                "",
            ]
        )
    return "\n".join(lines)


def create_repo(repo_dir: Path, file_count: int, funcs_per_file: int) -> None:
    write_text(repo_dir / "go.mod", "module example.com/cbmbench\n\ngo 1.22\n")
    write_text(repo_dir / "main.go", "package main\n\nfunc main() {}\n")
    for index in range(file_count):
        write_text(repo_dir / f"pkg/file_{index:04d}.go", go_file_content(index, 0, funcs_per_file))


def modify_existing_files(repo_dir: Path, changed_files: int, funcs_per_file: int) -> list[str]:
    changed: list[str] = []
    for index in range(changed_files):
        rel = Path("pkg") / f"file_{index:04d}.go"
        write_text(repo_dir / rel, go_file_content(index, 1000, funcs_per_file))
        changed.append(rel.as_posix())
    return changed


def create_python_reexport_repo(repo_dir: Path) -> None:
    write_text(repo_dir / "fastapi" / "__init__.py", "from .param_functions import Header\n")
    write_text(repo_dir / "fastapi" / "param_functions.py", "def Header(default=None):\n    return default\n")
    write_text(repo_dir / "fastapi" / "openapi" / "models.py", "class Header:\n    pass\n")
    write_text(
        repo_dir / "docs_src" / "app" / "main.py",
        "from fastapi import Header\n\n"
        "def create_item():\n"
        "    return Header(None)\n",
    )


def create_route_repo(repo_dir: Path, route_path: str) -> None:
    write_text(
        repo_dir / "routes.py",
        "from fastapi import FastAPI\n\n"
        "app = FastAPI()\n\n"
        f"@app.get('{route_path}')\n"
        "def orders():\n"
        "    return {'ok': True}\n",
    )


def create_inbound_frontier_repo(
    repo_dir: Path, language: str, dependent_files: int
) -> dict[str, Any]:
    """Create one definition file with a requested number of inbound dependents."""
    if dependent_files <= 0:
        raise ValueError("frontier files must be positive")
    dependent_paths: list[str] = []
    if language == "go":
        write_text(repo_dir / "go.mod", "module example.com/cbmfrontier\n\ngo 1.22\n")
        write_text(repo_dir / "leaf.go", "package frontier\n\nfunc Leaf() int { return 1 }\n")
        for index in range(dependent_files):
            relative = f"caller_{index:04d}.go"
            write_text(
                repo_dir / relative,
                "package frontier\n\n"
                f"func Caller{index:04d}() int {{ return Leaf() + {index} }}\n",
            )
            dependent_paths.append(relative)
        changed_path = "leaf.go"
    elif language == "python":
        write_text(repo_dir / "leaf.py", "def leaf():\n    return 1\n")
        for index in range(dependent_files):
            relative = f"caller_{index:04d}.py"
            write_text(
                repo_dir / relative,
                "from leaf import leaf\n\n"
                f"def caller_{index:04d}():\n    return leaf() + {index}\n",
            )
            dependent_paths.append(relative)
        changed_path = "leaf.py"
    elif language == "c_header":
        write_text(
            repo_dir / "shared.h",
            "#ifndef SHARED_H\n"
            "#define SHARED_H\n"
            "static int shared_value(void) { return 1; }\n"
            "#endif\n",
        )
        for index in range(dependent_files):
            relative = f"consumer_{index:04d}.c"
            write_text(
                repo_dir / relative,
                '#include "shared.h"\n\n'
                f"int consumer_{index:04d}(void) {{ return shared_value() + {index}; }}\n",
            )
            dependent_paths.append(relative)
        changed_path = "shared.h"
    elif language in {"cpp", "cuda"}:
        header_ext, source_ext = ("hpp", "cpp") if language == "cpp" else ("cuh", "cu")
        changed_path = f"shared.{header_ext}"
        write_text(repo_dir / changed_path, "inline int shared_value() { return 1; }\n")
        for index in range(dependent_files):
            relative = f"consumer_{index:04d}.{source_ext}"
            write_text(
                repo_dir / relative,
                f'#include "{changed_path}"\n\n'
                f"int consumer_{index:04d}() {{ return shared_value() + {index}; }}\n",
            )
            dependent_paths.append(relative)
    elif language in {"javascript", "typescript", "tsx"}:
        extension = {"javascript": "js", "typescript": "ts", "tsx": "tsx"}[language]
        changed_path = f"leaf.{extension}"
        return_type = "" if language == "javascript" else ": number"
        write_text(repo_dir / changed_path, f"export function leaf(){return_type} {{ return 1; }}\n")
        for index in range(dependent_files):
            relative = f"caller_{index:04d}.{extension}"
            import_suffix = ".js" if language == "javascript" else ""
            write_text(
                repo_dir / relative,
                f"import {{ leaf }} from './leaf{import_suffix}';\n\n"
                f"export function caller{index:04d}(){return_type} "
                f"{{ return leaf() + {index}; }}\n",
            )
            dependent_paths.append(relative)
    elif language == "php":
        changed_path = "Leaf.php"
        write_text(
            repo_dir / changed_path,
            "<?php\nnamespace Frontier;\nfunction leaf_value(): int { return 1; }\n",
        )
        for index in range(dependent_files):
            relative = f"Caller{index:04d}.php"
            write_text(
                repo_dir / relative,
                "<?php\nnamespace Frontier;\nrequire_once __DIR__ . '/Leaf.php';\n"
                f"function caller_{index:04d}(): int {{ return leaf_value() + {index}; }}\n",
            )
            dependent_paths.append(relative)
    elif language == "csharp":
        changed_path = "Leaf.cs"
        write_text(
            repo_dir / changed_path,
            "namespace Frontier { public static class LeafApi { "
            "public static int Value() { return 1; } } }\n",
        )
        for index in range(dependent_files):
            relative = f"Caller{index:04d}.cs"
            write_text(
                repo_dir / relative,
                "namespace Frontier { "
                f"public class Caller{index:04d} {{ public int Call() "
                f"{{ return LeafApi.Value() + {index}; }} }} }}\n",
            )
            dependent_paths.append(relative)
    elif language == "java":
        changed_path = "Leaf.java"
        write_text(
            repo_dir / changed_path,
            "package frontier;\npublic class Leaf { public static int value() { return 1; } }\n",
        )
        for index in range(dependent_files):
            relative = f"Caller{index:04d}.java"
            write_text(
                repo_dir / relative,
                "package frontier;\n"
                f"class Caller{index:04d} {{ int call() {{ return Leaf.value() + {index}; }} }}\n",
            )
            dependent_paths.append(relative)
    elif language == "kotlin":
        changed_path = "Leaf.kt"
        write_text(repo_dir / changed_path, "package frontier\n\nfun leafValue(): Int = 1\n")
        for index in range(dependent_files):
            relative = f"Caller{index:04d}.kt"
            write_text(
                repo_dir / relative,
                f"package frontier\n\nfun caller{index:04d}(): Int = leafValue() + {index}\n",
            )
            dependent_paths.append(relative)
    elif language == "rust":
        changed_path = "leaf.rs"
        write_text(repo_dir / "Cargo.toml", "[package]\nname='cbm-frontier'\nversion='0.1.0'\n")
        write_text(repo_dir / changed_path, "pub fn leaf_value() -> i32 { 1 }\n")
        modules = ["mod leaf;"]
        for index in range(dependent_files):
            module = f"caller_{index:04d}"
            relative = f"{module}.rs"
            modules.append(f"mod {module};")
            write_text(
                repo_dir / relative,
                "use crate::leaf::leaf_value;\n"
                f"pub fn caller_{index:04d}() -> i32 {{ leaf_value() + {index} }}\n",
            )
            dependent_paths.append(relative)
        write_text(repo_dir / "lib.rs", "\n".join(modules) + "\n")
    else:
        raise ValueError(f"unsupported frontier language: {language}")
    resolver_language = "c" if language == "c_header" else language
    metadata = {
        "source": "synthetic_inbound_frontier",
        "language": language,
        "cross_file_resolver_language": resolver_language,
        "changed_path": changed_path,
        "requested_inbound_dependents": dependent_files,
        "dependent_paths": dependent_paths,
    }
    if resolver_language in SCOPED_EXACT_FRONTIER_LANGUAGES:
        metadata.update(
            {
                "incremental_contract": "exact_frontier",
                "expected_minimum_affected_files": dependent_files + 1,
            }
        )
    else:
        metadata.update(
            {
                "incremental_contract": "safe_full_rebuild",
                "expected_publish_kind": PUBLISH_FULL,
                "expected_reason": "scoped_lsp_gap",
            }
        )
    return metadata


def mutate_inbound_frontier_repo(repo_dir: Path, language: str) -> list[str]:
    if language == "go":
        changed_path = "leaf.go"
        content = (
            "package frontier\n\n"
            "func Leaf() int { return 2 }\n\n"
            "func LeafExtra() int { return Leaf() + 1 }\n"
        )
    elif language == "python":
        changed_path = "leaf.py"
        content = (
            "def leaf():\n    return 2\n\n"
            "def leaf_extra():\n    return leaf() + 1\n"
        )
    elif language == "c_header":
        changed_path = "shared.h"
        content = (
            "#ifndef SHARED_H\n"
            "#define SHARED_H\n"
            "static int shared_value(void) { return 2; }\n"
            "static int shared_extra(void) { return shared_value() + 1; }\n"
            "#endif\n"
        )
    elif language in {"cpp", "cuda"}:
        header_ext = "hpp" if language == "cpp" else "cuh"
        changed_path = f"shared.{header_ext}"
        content = (
            "inline int shared_value() { return 2; }\n"
            "inline int shared_extra() { return shared_value() + 1; }\n"
        )
    elif language in {"javascript", "typescript", "tsx"}:
        extension = {"javascript": "js", "typescript": "ts", "tsx": "tsx"}[language]
        changed_path = f"leaf.{extension}"
        return_type = "" if language == "javascript" else ": number"
        content = (
            f"export function leaf(){return_type} {{ return 2; }}\n"
            f"export function leafExtra(){return_type} {{ return leaf() + 1; }}\n"
        )
    elif language == "php":
        changed_path = "Leaf.php"
        content = (
            "<?php\nnamespace Frontier;\nfunction leaf_value(): int { return 2; }\n"
            "function leaf_extra(): int { return leaf_value() + 1; }\n"
        )
    elif language == "csharp":
        changed_path = "Leaf.cs"
        content = (
            "namespace Frontier { public static class LeafApi { "
            "public static int Value() { return 2; } "
            "public static int Extra() { return Value() + 1; } } }\n"
        )
    elif language == "java":
        changed_path = "Leaf.java"
        content = (
            "package frontier;\npublic class Leaf { public static int value() { return 2; } "
            "public static int extra() { return value() + 1; } }\n"
        )
    elif language == "kotlin":
        changed_path = "Leaf.kt"
        content = (
            "package frontier\n\nfun leafValue(): Int = 2\n"
            "fun leafExtra(): Int = leafValue() + 1\n"
        )
    elif language == "rust":
        changed_path = "leaf.rs"
        content = (
            "pub fn leaf_value() -> i32 { 2 }\n"
            "pub fn leaf_extra() -> i32 { leaf_value() + 1 }\n"
        )
    else:
        raise ValueError(f"unsupported frontier language: {language}")
    write_text(repo_dir / changed_path, content)
    return [changed_path]


def command_result(
    cmd: list[str],
    env: dict[str, str],
    timeout: int,
    cwd: Path | None = None,
) -> tuple[subprocess.CompletedProcess[str], float]:
    start = now_ms()
    proc = subprocess.run(
        cmd,
        cwd=str(cwd) if cwd else None,
        env=env,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    return proc, now_ms() - start


def text_tail(text: str, max_lines: int = FAILURE_TAIL_LINES) -> list[str]:
    lines = text.splitlines()
    return lines[-max_lines:]


def failure_artifact_dir(env: dict[str, str]) -> Path:
    cache_dir = env.get("CBM_CACHE_DIR")
    if cache_dir:
        return Path(cache_dir).expanduser().parent / FAILURE_ARTIFACT_DIRNAME
    return Path(tempfile.gettempdir()) / FAILURE_FALLBACK_DIRNAME


def command_failure(
    label: str,
    cmd: list[str],
    env: dict[str, str],
    proc: subprocess.CompletedProcess[str],
    elapsed_ms: float,
) -> BenchmarkCommandError:
    safe_label = re.sub(r"[^A-Za-z0-9_.-]+", "_", label).strip("_") or "command"
    stamp = datetime.now(timezone.utc).strftime(FAILURE_TIMESTAMP_FORMAT)
    prefix = failure_artifact_dir(env) / f"{stamp}-{safe_label}"
    stdout_path = Path(f"{prefix}.stdout.txt")
    stderr_path = Path(f"{prefix}.stderr.txt")
    meta_path = Path(f"{prefix}.meta.json")

    write_text(stdout_path, proc.stdout)
    write_text(stderr_path, proc.stderr)
    detail: dict[str, Any] = {
        "label": label,
        "returncode": proc.returncode,
        "elapsed_ms": round(elapsed_ms, 3),
        "stdout_bytes": len(proc.stdout.encode("utf-8")),
        "stderr_bytes": len(proc.stderr.encode("utf-8")),
        "stdout_tail": text_tail(proc.stdout),
        "stderr_tail": text_tail(proc.stderr),
        "artifacts": {
            "stdout": str(stdout_path),
            "stderr": str(stderr_path),
            "meta": str(meta_path),
        },
    }
    write_text(
        meta_path,
        json.dumps({"cmd": cmd, **detail}, indent=2, sort_keys=True) + "\n",
    )
    return BenchmarkCommandError(
        f"{label} failed with rc={proc.returncode}; artifacts={detail['artifacts']}",
        detail,
    )


def record_report_error(report: dict[str, Any], exc: Exception) -> None:
    report["error"] = f"{type(exc).__name__}: {exc}"
    if isinstance(exc, BenchmarkCommandError):
        report["error_detail"] = exc.detail


def command_stdout(cmd: list[str], timeout: int, cwd: Path | None = None) -> str:
    proc, _ = command_result(cmd, dict(os.environ), timeout, cwd)
    if proc.returncode != 0:
        rendered = " ".join(cmd)
        raise RuntimeError(f"{rendered} failed: {proc.stderr.strip()}")
    return proc.stdout.strip()


def command_stdout_bytes(cmd: list[str], timeout: int, cwd: Path | None = None) -> bytes:
    proc = subprocess.run(
        cmd,
        cwd=str(cwd) if cwd else None,
        env=dict(os.environ),
        capture_output=True,
        timeout=timeout,
    )
    if proc.returncode != 0:
        rendered = " ".join(cmd)
        stderr = proc.stderr.decode("utf-8", "replace").strip()
        raise RuntimeError(f"{rendered} failed: {stderr}")
    return proc.stdout


def append_text(path: Path, text: str) -> None:
    current = path.read_text(encoding="utf-8")
    path.write_text(current + text, encoding="utf-8")


def unwrap_cli_json(stdout: str) -> dict[str, Any]:
    outer = json.loads(stdout)
    if "content" in outer:
        return json.loads(outer["content"][0]["text"])
    return outer


def unwrap_mcp_result(response: dict[str, Any]) -> dict[str, Any]:
    result = response.get("result", {})
    if "content" in result:
        return json.loads(result["content"][0]["text"])
    return result


def cli_result_text(stdout: str) -> str:
    outer = json.loads(stdout)
    if "content" in outer:
        return str(outer["content"][0]["text"])
    return json.dumps(outer, separators=(",", ":"), sort_keys=True)


def mcp_result_text(response: dict[str, Any]) -> str:
    result = response.get("result", {})
    if "content" in result:
        return str(result["content"][0]["text"])
    return json.dumps(result, separators=(",", ":"), sort_keys=True)


TOKEN_ESTIMATOR = "utf8_bytes_div_4_ceil"


def canonical_response_bytes(data: dict[str, Any]) -> bytes:
    """Serialize the tool payload independently of CLI/MCP envelopes."""
    return json.dumps(data, separators=(",", ":"), sort_keys=True).encode("utf-8")


def estimate_response_tokens(payload: bytes) -> int:
    """Return a deterministic, dependency-free byte/4 token estimate."""
    return (len(payload) + 3) // 4


class McpClient:
    def __init__(self, binary: Path, env: dict[str, str], timeout: int) -> None:
        self.binary = binary
        self.env = env
        self.timeout = timeout
        self.next_id = 1
        self.stderr_lines: list[str] = []
        self.stderr_lock = threading.Lock()
        self.stdout_queue: queue.Queue[str | None] = queue.Queue()
        self.proc: subprocess.Popen[str] | None = None
        self.stdout_thread: threading.Thread | None = None
        self.stderr_thread: threading.Thread | None = None

    def __enter__(self) -> "McpClient":
        self.proc = subprocess.Popen(
            [str(self.binary)],
            env=self.env,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )
        self.stdout_thread = threading.Thread(target=self._read_stdout, daemon=True)
        self.stderr_thread = threading.Thread(target=self._read_stderr, daemon=True)
        self.stdout_thread.start()
        self.stderr_thread.start()
        self._initialize()
        return self

    def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
        if not self.proc:
            return
        if self.proc.stdin:
            self.proc.stdin.close()
        try:
            self.proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.proc.kill()
                self.proc.wait(timeout=5)

    def _read_stdout(self) -> None:
        assert self.proc and self.proc.stdout
        for line in self.proc.stdout:
            self.stdout_queue.put(line)
        self.stdout_queue.put(None)

    def _read_stderr(self) -> None:
        assert self.proc and self.proc.stderr
        for line in self.proc.stderr:
            with self.stderr_lock:
                self.stderr_lines.append(line.rstrip("\n"))

    def _stderr_mark(self) -> int:
        with self.stderr_lock:
            return len(self.stderr_lines)

    def _stderr_since(self, mark: int) -> str:
        with self.stderr_lock:
            return "\n".join(self.stderr_lines[mark:])

    def _send(self, message: dict[str, Any]) -> None:
        if not self.proc or not self.proc.stdin:
            raise RuntimeError("MCP server is not running")
        self.proc.stdin.write(json.dumps(message, separators=(",", ":")) + "\n")
        self.proc.stdin.flush()

    def _request(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        req_id = self.next_id
        self.next_id += 1
        message: dict[str, Any] = {"jsonrpc": "2.0", "id": req_id, "method": method}
        if params is not None:
            message["params"] = params
        self._send(message)

        deadline = time.monotonic() + self.timeout
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError(f"MCP request timed out: {method}")
            line = self.stdout_queue.get(timeout=remaining)
            if line is None:
                raise RuntimeError(f"MCP server exited before response: {method}")
            try:
                response = json.loads(line)
            except json.JSONDecodeError as exc:
                raise RuntimeError(f"non-JSON MCP stdout line: {line[:200]!r}") from exc
            if response.get("id") == req_id:
                if "error" in response:
                    raise RuntimeError(f"MCP request failed: {response['error']}")
                return response

    def _notification(self, method: str, params: dict[str, Any] | None = None) -> None:
        message: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
        if params is not None:
            message["params"] = params
        self._send(message)

    def _initialize(self) -> None:
        self._request(
            "initialize",
            {
                "protocolVersion": MCP_INIT_PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "cbm-incr-speed", "version": "1.0"},
            },
        )
        self._notification("notifications/initialized")

    def call_tool(
        self, name: str, arguments: dict[str, Any]
    ) -> tuple[dict[str, Any], str, int, float]:
        text, stderr, stdout_bytes, elapsed_ms = self.call_tool_text(name, arguments)
        return json.loads(text), stderr, stdout_bytes, elapsed_ms

    def call_tool_text(
        self, name: str, arguments: dict[str, Any]
    ) -> tuple[str, str, int, float]:
        mark = self._stderr_mark()
        start = now_ms()
        response = self._request("tools/call", {"name": name, "arguments": arguments})
        elapsed_ms = now_ms() - start
        stderr = self._stderr_since(mark)
        stdout_bytes = len(json.dumps(response, separators=(",", ":")).encode("utf-8"))
        return mcp_result_text(response), stderr, stdout_bytes, elapsed_ms


def log_tail(stderr: str) -> list[str]:
    lines = stderr.splitlines()
    return lines[-LOG_TAIL_LINES:]


def log_has(stderr: str, marker: str) -> bool:
    return marker in stderr


def response_publish_kind(data: dict[str, Any]) -> str:
    publish_kind = data.get("publish_kind")
    return publish_kind if isinstance(publish_kind, str) else ""


def response_publish_reason(data: dict[str, Any]) -> str:
    publish_reason = data.get("publish_reason")
    return publish_reason if isinstance(publish_reason, str) else ""


def response_freshness(data: dict[str, Any]) -> dict[str, Any] | None:
    freshness = data.get("freshness")
    return freshness if isinstance(freshness, dict) else None


def response_freshness_state(data: dict[str, Any]) -> str:
    freshness = response_freshness(data)
    if not freshness:
        return ""
    state = freshness.get("state")
    return state if isinstance(state, str) else ""


def is_incremental_publish_kind(publish_kind: str) -> bool:
    return publish_kind in {
        PUBLISH_INCREMENTAL_NOOP,
        PUBLISH_INCREMENTAL_EXACT,
        PUBLISH_INCREMENTAL_OVERLAY,
        PUBLISH_INCREMENTAL_CONTAINMENT,
    }


def is_explicit_incremental_route(publish_kind: str | None, reason: str | None = None) -> bool:
    return is_incremental_publish_kind(publish_kind or "") or bool(reason)


def parse_logged_elapsed_ms(stderr: str, marker: str) -> int | None:
    return parse_log_int_field(stderr, marker, "elapsed_ms")


def parse_log_int_field(stderr: str, marker: str, field: str) -> int | None:
    prefix = f"{field}="
    for line in stderr.splitlines():
        if marker not in line:
            continue
        for item in line.split():
            if item.startswith(prefix):
                try:
                    return int(item.split("=", 1)[1])
                except ValueError:
                    return None
    return None


def parse_log_max_int_field(stderr: str, marker: str, field: str) -> int | None:
    prefix = f"{field}="
    maximum: int | None = None
    for line in stderr.splitlines():
        if marker not in line:
            continue
        for item in line.split():
            if not item.startswith(prefix):
                continue
            try:
                value = int(item.split("=", 1)[1])
            except ValueError:
                continue
            maximum = value if maximum is None else max(maximum, value)
    return maximum


def parse_log_text_field(stderr: str, marker: str, field: str) -> str | None:
    prefix = f"{field}="
    for line in reversed(stderr.splitlines()):
        if marker not in line:
            continue
        for item in line.split():
            if item.startswith(prefix):
                return item[len(prefix) :]
    return None


def parse_exact_reason(stderr: str) -> str | None:
    detail = parse_exact_route_detail(stderr)
    reason = detail.get("reason")
    return reason if isinstance(reason, str) and reason else None


def parse_exact_route_detail(stderr: str) -> dict[str, Any]:
    detail: dict[str, Any] = {
        "frontier_changed_files": parse_log_int_field(stderr, LOG_MARKER_EXACT_FRONTIER, "changed"),
        "frontier_expanded_files": parse_log_int_field(stderr, LOG_MARKER_EXACT_FRONTIER, "expanded"),
        "exact_done_files": parse_log_int_field(stderr, LOG_MARKER_EXACT_DONE, "files"),
        "event": None,
        "reason": None,
    }
    reason_markers = (
        (LOG_MARKER_EXACT_FALLBACK, "fallback"),
        (LOG_MARKER_EXACT_DELETE_FALLBACK, "delete_fallback"),
        (LOG_MARKER_EXACT_SKIP, "skip"),
    )
    for line in stderr.splitlines():
        for marker, event in reason_markers:
            prefix = f"msg={marker} reason="
            if prefix not in line:
                continue
            reason = line.split(prefix, 1)[1].split()[0]
            detail["event"] = event
            detail["reason"] = reason or None
            return detail
    if detail["exact_done_files"] is not None:
        detail["event"] = "exact"
    elif detail["frontier_expanded_files"] is not None:
        detail["event"] = "frontier_observed"
    return detail


def response_exact_delta(data: dict[str, Any]) -> dict[str, Any]:
    exact_delta = data.get("exact_delta")
    return exact_delta if isinstance(exact_delta, dict) else {}


def merge_exact_route_detail(
    detail: dict[str, Any],
    data: dict[str, Any],
    publish_kind: str,
    publish_reason: str,
) -> dict[str, Any]:
    exact_delta = response_exact_delta(data)
    field_map = {
        "changed_paths": "frontier_changed_files",
        "affected_paths": "frontier_expanded_files",
        "published_paths": "exact_done_files",
    }
    for response_key, detail_key in field_map.items():
        value = exact_delta.get(response_key)
        if detail.get(detail_key) is None and isinstance(value, int):
            detail[detail_key] = value
    if not detail.get("reason") and publish_reason:
        detail["reason"] = publish_reason
    if not detail.get("event"):
        published = detail.get("exact_done_files")
        if isinstance(published, int) and published > 0:
            detail["event"] = "exact"
        elif isinstance(published, int) and published == 0:
            detail["event"] = "noop"
        elif publish_reason:
            detail["event"] = "fallback"
        elif publish_kind == PUBLISH_INCREMENTAL_EXACT:
            detail["event"] = "exact"
        elif publish_kind == PUBLISH_INCREMENTAL_OVERLAY:
            detail["event"] = "overlay"
        elif publish_kind == PUBLISH_INCREMENTAL_NOOP:
            detail["event"] = "noop"
    return detail


def indexed_work_elapsed_ms(logged_elapsed_ms: dict[str, int | None]) -> int | None:
    incremental_ms = logged_elapsed_ms.get("incremental_done")
    if incremental_ms is not None:
        return incremental_ms
    return logged_elapsed_ms.get("pipeline_done")


def run_config_set(binary: Path, env: dict[str, str], key: str, value: str, timeout: int) -> None:
    cmd = [str(binary), "config", "set", key, value]
    proc, elapsed_ms = command_result(cmd, env, timeout)
    if proc.returncode != 0:
        raise command_failure(f"config_set_{key}", cmd, env, proc, elapsed_ms)


def parse_config_overrides(items: list[str]) -> dict[str, str]:
    overrides: dict[str, str] = {}
    for item in items:
        key, sep, value = item.partition("=")
        if not sep or not key or not value:
            raise SystemExit(f"error: --config must be key=value, got {item!r}")
        overrides[key] = value
    return overrides


def resolve_config_overrides(profile: str, items: list[str]) -> dict[str, str]:
    """Return one explicit benchmark profile plus higher-priority per-key overrides."""
    if profile not in CONFIG_PROFILES:
        raise ValueError(f"unknown config profile: {profile}")
    overrides = dict(CONFIG_PROFILES[profile])
    overrides.update(parse_config_overrides(items))
    return overrides


def apply_config_overrides(
    binary: Path, env: dict[str, str], overrides: dict[str, str], timeout: int
) -> None:
    for key, value in overrides.items():
        run_config_set(binary, env, key, value, timeout)


def build_index_result(
    data: dict[str, Any],
    stderr: str,
    stdout_bytes: int,
    elapsed_ms: float,
    include_logs: bool,
) -> dict[str, Any]:
    measurement_log_markers: list[str] = []
    logfiles: list[str] = []
    supervisor_log = parse_log_text_field(stderr, "index.supervisor.profile_log", "log")
    if supervisor_log:
        logfiles.append(supervisor_log)
    response_log = data.get("logfile")
    if isinstance(response_log, str) and response_log and response_log not in logfiles:
        logfiles.append(response_log)
    for logfile in logfiles:
        try:
            with Path(logfile).open(encoding="utf-8", errors="replace") as stream:
                for line in stream:
                    if any(
                        marker in line
                        for marker in (
                            "msg=mem.phase",
                            "msg=pipeline.done",
                            "msg=incremental.done",
                        )
                    ):
                        measurement_log_markers.append(line.rstrip("\n"))
                        if len(measurement_log_markers) >= 512:
                            break
        except OSError:
            continue
        if measurement_log_markers:
            break
    measurement_text = "\n".join((stderr, *measurement_log_markers))
    elapsed_ms_int = int(elapsed_ms)
    publish_kind = response_publish_kind(data)
    logged_elapsed_ms = {
        "pipeline_done": parse_logged_elapsed_ms(measurement_text, LOG_MARKER_PIPELINE_DONE),
        "incremental_done": parse_logged_elapsed_ms(measurement_text, LOG_MARKER_INCREMENTAL_DONE),
    }
    indexed_ms = indexed_work_elapsed_ms(logged_elapsed_ms)
    publish_reason = response_publish_reason(data)
    exact_route_detail = merge_exact_route_detail(
        parse_exact_route_detail(stderr), data, publish_kind, publish_reason
    )
    freshness = response_freshness(data)
    freshness_state = response_freshness_state(data)
    peak_candidates = [
        parse_log_max_int_field(measurement_text, marker, "peak_mb")
        for marker in ("mem.phase", LOG_MARKER_PIPELINE_DONE, LOG_MARKER_INCREMENTAL_DONE)
    ]
    peak_rss_mb = max((value for value in peak_candidates if value is not None), default=None)
    result: dict[str, Any] = {
        "elapsed_ms": elapsed_ms_int,
        "peak_rss_mb": peak_rss_mb,
        "measurement_log_markers": measurement_log_markers,
        "indexed_work_elapsed_ms": indexed_ms,
        "unlogged_overhead_ms": (elapsed_ms_int - indexed_ms) if indexed_ms is not None else None,
        "response": data,
        "publish_kind": publish_kind or None,
        "freshness_state": freshness_state or None,
        "freshness": freshness,
        "stdout_bytes": stdout_bytes,
        "markers": {
            "incremental_exact_done": log_has(stderr, LOG_MARKER_EXACT_DONE)
            or publish_kind == PUBLISH_INCREMENTAL_EXACT,
            "incremental_done": log_has(stderr, LOG_MARKER_INCREMENTAL_DONE)
            or is_incremental_publish_kind(publish_kind),
            "pagerank_done": log_has(stderr, "pagerank.done"),
            "pagerank_defer": log_has(stderr, "pagerank.defer"),
            "full_route": log_has(stderr, "pipeline.route path=full")
            or publish_kind == PUBLISH_FULL,
            "incremental_route": log_has(stderr, "pipeline.route path=incremental")
            or is_incremental_publish_kind(publish_kind),
        },
        "logged_elapsed_ms": logged_elapsed_ms,
        "exact_reason": publish_reason or exact_route_detail.get("reason"),
        "exact_route_detail": exact_route_detail,
        "stderr_tail": log_tail(stderr),
    }
    if include_logs:
        result["stderr"] = stderr
    return result


def run_index(
    binary: Path,
    env: dict[str, str],
    repo_dir: Path,
    timeout: int,
    include_logs: bool,
) -> dict[str, Any]:
    args = json.dumps({"repo_path": str(repo_dir), "mode": "fast"})
    cmd = [str(binary), "cli", "--json", "index_repository", args]
    proc, elapsed_ms = command_result(
        cmd,
        env,
        timeout,
    )
    if proc.returncode != 0:
        raise command_failure("index_repository", cmd, env, proc, elapsed_ms)
    data = unwrap_cli_json(proc.stdout)
    return build_index_result(
        data, proc.stderr, len(proc.stdout.encode("utf-8")), elapsed_ms, include_logs
    )


def run_index_mcp(client: McpClient, repo_dir: Path, include_logs: bool) -> dict[str, Any]:
    data, stderr, stdout_bytes, elapsed_ms = client.call_tool(
        "index_repository", {"repo_path": str(repo_dir), "mode": "fast"}
    )
    return build_index_result(data, stderr, stdout_bytes, elapsed_ms, include_logs)


def build_tool_probe_result(
    data: dict[str, Any],
    stderr: str,
    stdout_bytes: int,
    elapsed_ms: float,
    include_logs: bool,
) -> dict[str, Any]:
    elapsed_ms_value = round(elapsed_ms, 3)
    result: dict[str, Any] = {
        "elapsed_ms": elapsed_ms_value,
        "stdout_bytes": stdout_bytes,
        "response_keys": sorted(str(key) for key in data.keys()),
        "stderr_tail": log_tail(stderr),
    }
    if include_logs:
        result["stderr"] = stderr
    return result


def run_cli_tool_probe(
    binary: Path,
    env: dict[str, str],
    tool_name: str,
    timeout: int,
    include_logs: bool,
) -> dict[str, Any]:
    cmd = [str(binary), "cli", "--json", tool_name, "{}"]
    proc, elapsed_ms = command_result(cmd, env, timeout)
    if proc.returncode != 0:
        raise command_failure(f"{tool_name}_probe", cmd, env, proc, elapsed_ms)
    data = unwrap_cli_json(proc.stdout)
    return build_tool_probe_result(
        data, proc.stderr, len(proc.stdout.encode("utf-8")), elapsed_ms, include_logs
    )


def run_mcp_tool_probe(
    client: McpClient,
    tool_name: str,
    include_logs: bool,
) -> dict[str, Any]:
    data, stderr, stdout_bytes, elapsed_ms = client.call_tool(tool_name, {})
    return build_tool_probe_result(data, stderr, stdout_bytes, elapsed_ms, include_logs)


def build_tool_call_result(
    data: dict[str, Any],
    stderr: str,
    stdout_bytes: int,
    elapsed_ms: float,
    include_logs: bool,
    response_payload: bytes | None = None,
) -> dict[str, Any]:
    quality_payload = canonical_response_bytes(data)
    payload = response_payload if response_payload is not None else quality_payload
    result: dict[str, Any] = {
        "elapsed_ms": round(elapsed_ms, 3),
        # Preserve the historical field while separating transport framing from
        # the canonical payload used for cross-transport comparisons.
        "stdout_bytes": stdout_bytes,
        "transport_response_bytes": stdout_bytes,
        "response_bytes": len(payload),
        "response_token_estimate": estimate_response_tokens(payload),
        "token_estimator": TOKEN_ESTIMATOR,
        "response_encoding": "tool_default" if response_payload is not None else "canonical_json",
        "quality_response_bytes": len(quality_payload),
        "response": data,
        "freshness_state": response_freshness_state(data) or None,
        "freshness": response_freshness(data),
        "stderr_tail": log_tail(stderr),
    }
    if include_logs:
        result["stderr"] = stderr
    return result


def run_cli_tool_call(
    binary: Path,
    env: dict[str, str],
    tool_name: str,
    arguments: dict[str, Any],
    timeout: int,
    include_logs: bool,
) -> dict[str, Any]:
    encoded = json.dumps(arguments, separators=(",", ":"))
    cmd = [str(binary), "cli", "--json", tool_name, encoded]
    proc, elapsed_ms = command_result(cmd, env, timeout)
    if proc.returncode != 0:
        raise command_failure(f"{tool_name}_call", cmd, env, proc, elapsed_ms)
    raw_payload = cli_result_text(proc.stdout).encode("utf-8")
    quality_arguments = dict(arguments)
    quality_arguments["format"] = "json"
    quality_cmd = [
        str(binary), "cli", "--json", tool_name,
        json.dumps(quality_arguments, separators=(",", ":")),
    ]
    quality_proc, quality_elapsed_ms = command_result(quality_cmd, env, timeout)
    if quality_proc.returncode != 0:
        raise command_failure(
            f"{tool_name}_quality_call", quality_cmd, env, quality_proc, quality_elapsed_ms
        )
    data = unwrap_cli_json(quality_proc.stdout)
    result = build_tool_call_result(
        data, proc.stderr, len(proc.stdout.encode("utf-8")), elapsed_ms, include_logs, raw_payload
    )
    result["quality_probe_elapsed_ms"] = round(quality_elapsed_ms, 3)
    return result


def run_mcp_tool_call(
    client: McpClient,
    tool_name: str,
    arguments: dict[str, Any],
    include_logs: bool,
) -> dict[str, Any]:
    raw_text, stderr, stdout_bytes, elapsed_ms = client.call_tool_text(tool_name, arguments)
    quality_arguments = dict(arguments)
    quality_arguments["format"] = "json"
    data, _, _, quality_elapsed_ms = client.call_tool(tool_name, quality_arguments)
    result = build_tool_call_result(
        data, stderr, stdout_bytes, elapsed_ms, include_logs, raw_text.encode("utf-8")
    )
    result["quality_probe_elapsed_ms"] = round(quality_elapsed_ms, 3)
    return result


def run_tool_call_for_transport(
    transport: str,
    binary: Path,
    env: dict[str, str],
    tool_name: str,
    arguments: dict[str, Any],
    timeout: int,
    include_logs: bool,
    client: McpClient | None = None,
) -> dict[str, Any]:
    if transport == "mcp":
        if client is None:
            raise RuntimeError("MCP transport requires an active client")
        return run_mcp_tool_call(client, tool_name, arguments, include_logs)
    return run_cli_tool_call(binary, env, tool_name, arguments, timeout, include_logs)


def summarize_elapsed_ms(probes: list[dict[str, Any]]) -> dict[str, Any]:
    elapsed = sorted(float(probe["elapsed_ms"]) for probe in probes)
    if not elapsed:
        return {"count": 0}
    return {
        "count": len(elapsed),
        "min_ms": elapsed[0],
        "median_ms": elapsed[len(elapsed) // 2],
        "max_ms": elapsed[-1],
    }


def measure_cli_overhead_probes(
    binary: Path,
    env: dict[str, str],
    tool_name: str,
    count: int,
    timeout: int,
    include_logs: bool,
) -> dict[str, Any] | None:
    if count <= 0:
        return None
    probes = [
        run_cli_tool_probe(binary, env, tool_name, timeout, include_logs)
        for _ in range(count)
    ]
    return {"tool": tool_name, "trials": probes, "summary": summarize_elapsed_ms(probes)}


def measure_mcp_overhead_probes(
    client: McpClient,
    tool_name: str,
    count: int,
    include_logs: bool,
) -> dict[str, Any] | None:
    if count <= 0:
        return None
    probes = [run_mcp_tool_probe(client, tool_name, include_logs) for _ in range(count)]
    return {"tool": tool_name, "trials": probes, "summary": summarize_elapsed_ms(probes)}


def remove_project_dbs(cache_dir: Path) -> list[str]:
    removed: list[str] = []
    for path in cache_dir.iterdir():
        if not path.is_file():
            continue
        if path.name == CONFIG_DB_NAME or not path.name.endswith(PROJECT_DB_SUFFIX):
            continue
        path.unlink()
        removed.append(path.name)
        for suffix in ("-wal", "-shm"):
            sidecar = cache_dir / f"{path.name}{suffix}"
            if sidecar.exists():
                sidecar.unlink()
                removed.append(sidecar.name)
    return removed


def find_project_db(cache_dir: Path) -> Path:
    dbs = sorted(
        path
        for path in cache_dir.iterdir()
        if path.is_file() and path.name != CONFIG_DB_NAME and path.name.endswith(PROJECT_DB_SUFFIX)
    )
    if len(dbs) != 1:
        names = ", ".join(path.name for path in dbs)
        raise RuntimeError(f"expected one project DB in {cache_dir}, found {len(dbs)}: {names}")
    return dbs[0]


def remove_sqlite_sidecars(path: Path) -> None:
    for suffix in ("-wal", "-shm"):
        sidecar = Path(f"{path}{suffix}")
        if sidecar.exists():
            sidecar.unlink()


def copy_sqlite_snapshot(source: Path, destination: Path) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    if destination.exists():
        destination.unlink()
    remove_sqlite_sidecars(destination)
    uri = f"{source.resolve().as_uri()}?mode=ro"
    with sqlite3.connect(uri, uri=True) as src, sqlite3.connect(str(destination)) as dst:
        src.backup(dst)


def decode_sqlite_text(data: bytes) -> str:
    return data.decode("utf-8", "surrogateescape")


def sqlite_cbm_source_span_label(label: str | None) -> int:
    return int(label in SOURCE_SPAN_LABELS)


def query_rows(db_path: Path, sql: str, params: tuple[Any, ...]) -> list[str]:
    con = sqlite3.connect(str(db_path))
    con.text_factory = decode_sqlite_text
    con.create_function("cbm_source_span_label", 1, sqlite_cbm_source_span_label)
    try:
        rows = [str(row[0]) for row in con.execute(sql, params)]
    finally:
        con.close()
    return rows


def canonical_query_rows(db_path: Path, project: str, sql: str) -> list[str]:
    return query_rows(db_path, sql, (project,))


def compare_query_rows(
    left_db: Path,
    right_db: Path,
    kind: str,
    left_sql: str,
    left_params: tuple[Any, ...],
    right_sql: str,
    right_params: tuple[Any, ...],
) -> dict[str, Any]:
    left = query_rows(left_db, left_sql, left_params)
    right = query_rows(right_db, right_sql, right_params)
    if left != right:
        left_set = set(left)
        right_set = set(right)
        left_only = next((row for row in left if row not in right_set), None)
        right_only = next((row for row in right if row not in left_set), None)
        return {
            "equal": False,
            "kind": kind,
            "left_count": len(left),
            "right_count": len(right),
            "left_only": left_only,
            "right_only": right_only,
        }
    return {"equal": True}


CANONICAL_NODES_SQL = (
    "SELECT quote(label) || char(9) || quote(name) || char(9) || "
    "quote(qualified_name) || char(9) || quote(coalesce(file_path,'')) || char(9) || "
    "start_line || char(9) || end_line || char(9) || "
    "COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(n.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '') "
    "FROM nodes n WHERE project = ?1 "
    "ORDER BY label, name, qualified_name, coalesce(file_path,''), start_line, end_line, "
    "COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(n.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '')"
)

CANONICAL_EDGES_SQL = (
    "SELECT quote(s.label) || char(9) || quote(s.qualified_name) || char(9) || "
    "quote(coalesce(s.file_path,'')) || char(9) || s.start_line || char(9) || "
    "s.end_line || char(9) || quote(t.label) || char(9) || quote(t.qualified_name) || "
    "char(9) || quote(coalesce(t.file_path,'')) || char(9) || t.start_line || char(9) || "
    "t.end_line || char(9) || quote(e.type) || char(9) || "
    "COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(e.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '') "
    "FROM edges e "
    "JOIN nodes s ON s.id = e.source_id "
    "JOIN nodes t ON t.id = e.target_id "
    "WHERE e.project = ?1 "
    "ORDER BY s.label, s.qualified_name, coalesce(s.file_path,''), s.start_line, s.end_line, "
    "t.label, t.qualified_name, coalesce(t.file_path,''), t.start_line, t.end_line, "
    "e.type, COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(e.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '')"
)

CANONICAL_HASHES_SQL = (
    "SELECT quote(rel_path) || char(9) || quote(sha256) || char(9) || mtime_ns || char(9) || "
    "size FROM file_hashes WHERE project = ?1 ORDER BY rel_path"
)

ACTIVE_OVERLAY_CTE_SQL = (
    "WITH active_overlay_files AS ("
    "  SELECT project, rel_path, MAX(overlay_generation) AS overlay_generation"
    "  FROM ("
    "    SELECT n.project, n.rel_path, n.overlay_generation"
    "    FROM overlay_nodes n"
    "    JOIN overlay_generations g"
    "      ON g.project = n.project AND g.overlay_generation = n.overlay_generation"
    "    WHERE g.status = ?1 AND n.project = ?4"
    "    UNION"
    "    SELECT e.project, e.rel_path, e.overlay_generation"
    "    FROM overlay_edges e"
    "    JOIN overlay_generations g"
    "      ON g.project = e.project AND g.overlay_generation = e.overlay_generation"
    "    WHERE g.status = ?1 AND e.project = ?4"
    "    UNION"
    "    SELECT t.project, t.rel_path, t.overlay_generation"
    "    FROM overlay_tombstones t"
    "    JOIN overlay_generations g"
    "      ON g.project = t.project AND g.overlay_generation = t.overlay_generation"
    "    WHERE g.status = ?1 AND t.active = ?3 AND t.project = ?4"
    "  ) overlay_files"
    "  GROUP BY project, rel_path"
    "), active_file_tombstones AS ("
    "  SELECT t.project, t.rel_path, MAX(t.overlay_generation) AS overlay_generation"
    "  FROM overlay_tombstones t"
    "  JOIN overlay_generations g"
    "    ON g.project = t.project AND g.overlay_generation = t.overlay_generation"
    "  WHERE g.status = ?1 AND t.entity_kind = ?2 AND t.active = ?3 AND t.project = ?4"
    "  GROUP BY t.project, t.rel_path"
    "), active_node_candidates AS ("
    "  SELECT 0 AS overlay_row, n.project, n.label, n.name, n.qualified_name, n.file_path,"
    "         n.start_line, n.end_line, n.properties"
    "  FROM nodes n"
    "  WHERE n.project = ?4"
    "    AND NOT EXISTS (SELECT 1 FROM active_file_tombstones af"
    "                    WHERE af.project = n.project AND af.rel_path = n.file_path)"
    "  UNION ALL"
    "  SELECT 1 AS overlay_row, n.project, n.label, n.name, n.qualified_name, n.file_path,"
    "         n.start_line, n.end_line, n.properties"
    "  FROM overlay_nodes n"
    "  JOIN active_overlay_files af"
    "    ON af.project = n.project AND af.rel_path = n.rel_path"
    "   AND af.overlay_generation = n.overlay_generation"
    "  WHERE n.owned = ?5"
    "), active_nodes AS ("
    "  SELECT project, label, name, qualified_name, file_path, start_line, end_line, properties"
    "  FROM ("
    "    SELECT c.*, ROW_NUMBER() OVER ("
    "      PARTITION BY c.project, c.qualified_name"
    "      ORDER BY cbm_source_span_label(c.label) DESC,"
    "               CASE WHEN cbm_source_span_label(c.label) = 1"
    "                    THEN CASE WHEN c.file_path <> '' THEN 1 ELSE 0 END"
    "                    ELSE c.overlay_row END DESC,"
    "               CASE WHEN cbm_source_span_label(c.label) = 1"
    "                         AND c.start_line > 0 AND c.end_line >= c.start_line"
    "                    THEN c.end_line - c.start_line + 1 ELSE 0 END DESC,"
    "               CASE WHEN cbm_source_span_label(c.label) = 1 THEN c.start_line ELSE 0 END ASC,"
    "               CASE WHEN cbm_source_span_label(c.label) = 1 THEN c.end_line ELSE 0 END DESC,"
    "               CASE WHEN cbm_source_span_label(c.label) = 1 THEN c.file_path ELSE '' END ASC,"
    "               c.overlay_row DESC"
    "    ) AS rn"
    "    FROM active_node_candidates c"
    "  ) ranked_nodes"
    "  WHERE rn = 1"
    "), active_edges AS ("
    "  SELECT e.project, s.qualified_name AS source_qn, t.qualified_name AS target_qn,"
    "         e.type, e.properties"
    "  FROM edges e"
    "  JOIN nodes s ON s.id = e.source_id"
    "  JOIN nodes t ON t.id = e.target_id"
    "  WHERE e.project = ?4"
    "    AND NOT EXISTS (SELECT 1 FROM active_file_tombstones af"
    "                    WHERE af.project = s.project AND af.rel_path = s.file_path)"
    "    AND NOT EXISTS (SELECT 1 FROM active_file_tombstones af"
    "                    WHERE af.project = t.project AND af.rel_path = t.file_path)"
    "  UNION"
    "  SELECT e.project, e.source_qn, e.target_qn, e.type, e.properties"
    "  FROM overlay_edges e"
    "  JOIN active_overlay_files af"
    "    ON af.project = e.project AND af.rel_path = e.rel_path"
    "   AND af.overlay_generation = e.overlay_generation"
    "  WHERE e.owned = ?5"
    ") "
)

ACTIVE_OVERLAY_NODES_SQL = (
    ACTIVE_OVERLAY_CTE_SQL
    + "SELECT quote(label) || char(9) || quote(name) || char(9) || "
    "quote(qualified_name) || char(9) || quote(coalesce(file_path,'')) || char(9) || "
    "start_line || char(9) || end_line || char(9) || "
    "COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(n.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '') "
    "FROM active_nodes n WHERE project = ?4 "
    "ORDER BY label, name, qualified_name, coalesce(file_path,''), start_line, end_line, "
    "COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(n.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '')"
)

ACTIVE_OVERLAY_EDGES_SQL = (
    ACTIVE_OVERLAY_CTE_SQL
    + "SELECT quote(s.label) || char(9) || quote(s.qualified_name) || char(9) || "
    "quote(coalesce(s.file_path,'')) || char(9) || s.start_line || char(9) || "
    "s.end_line || char(9) || quote(t.label) || char(9) || quote(t.qualified_name) || "
    "char(9) || quote(coalesce(t.file_path,'')) || char(9) || t.start_line || char(9) || "
    "t.end_line || char(9) || quote(e.type) || char(9) || "
    "COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(e.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '') "
    "FROM active_edges e "
    "JOIN active_nodes s ON s.project = e.project AND s.qualified_name = e.source_qn "
    "JOIN active_nodes t ON t.project = e.project AND t.qualified_name = e.target_qn "
    "WHERE e.project = ?4 "
    "ORDER BY s.label, s.qualified_name, coalesce(s.file_path,''), s.start_line, s.end_line, "
    "t.label, t.qualified_name, coalesce(t.file_path,''), t.start_line, t.end_line, "
    "e.type, COALESCE((SELECT group_concat(item, char(30)) FROM ("
    "SELECT quote(je.key) || '=' || je.type || '=' || "
    "COALESCE(quote(CAST(je.value AS TEXT)), 'NULL') AS item "
    "FROM json_each(e.properties) AS je "
    "ORDER BY je.key, je.type, CAST(je.value AS TEXT)"
    ")), '')"
)


def compare_canonical_graph(left_db: Path, right_db: Path, project: str) -> dict[str, Any]:
    for kind, sql in (
        ("canonical nodes", CANONICAL_NODES_SQL),
        ("canonical edges", CANONICAL_EDGES_SQL),
        ("file hashes", CANONICAL_HASHES_SQL),
    ):
        result = compare_query_rows(left_db, right_db, kind, sql, (project,), sql, (project,))
        if not result["equal"]:
            return result
    return {"equal": True}


def compare_active_overlay_graph(left_db: Path, right_db: Path, project: str) -> dict[str, Any]:
    left_params = (
        OVERLAY_STATUS_READY,
        OVERLAY_TOMBSTONE_FILE,
        OVERLAY_TOMBSTONE_ACTIVE,
        project,
        OVERLAY_ROW_OWNED,
    )
    for kind, left_sql, right_sql in (
        ("active overlay nodes", ACTIVE_OVERLAY_NODES_SQL, CANONICAL_NODES_SQL),
        ("active overlay edges", ACTIVE_OVERLAY_EDGES_SQL, CANONICAL_EDGES_SQL),
    ):
        result = compare_query_rows(
            left_db, right_db, kind, left_sql, left_params, right_sql, (project,)
        )
        if not result["equal"]:
            return result
    return {"equal": True}


def graph_gate_for_publish_kind(
    canonical: dict[str, Any],
    publish_kind: str | None,
    oracle_passed: bool | None = None,
    active_overlay: dict[str, Any] | None = None,
) -> dict[str, Any]:
    canonical_equal = bool(canonical.get("equal"))
    active_overlay_equal = bool(active_overlay and active_overlay.get("equal"))
    if publish_kind == PUBLISH_INCREMENTAL_OVERLAY and active_overlay is not None:
        return {
            "passed": active_overlay_equal,
            "policy": "overlay_active_graph",
            "canonical_equal": canonical_equal,
            "active_overlay_equal": active_overlay_equal,
            "reason": (
                "overlay publish leaves canonical rows unchanged; validate active overlay "
                "nodes and edges against a fresh full graph"
            ),
        }
    if publish_kind == PUBLISH_INCREMENTAL_OVERLAY and oracle_passed is not None:
        return {
            "passed": bool(oracle_passed),
            "policy": "overlay_active_oracles",
            "canonical_equal": canonical_equal,
            "reason": (
                "overlay publish leaves canonical rows unchanged; self-dogfood gates "
                "on active read oracles and freshness metadata"
            ),
        }
    return {
        "passed": canonical_equal,
        "policy": "canonical_graph",
        "canonical_equal": canonical_equal,
    }


def frontier_coverage_gate(
    scenario_metadata: dict[str, Any],
    incremental: dict[str, Any],
    exact_cap: int | None = None,
) -> dict[str, Any]:
    expected_publish_kind = scenario_metadata.get("expected_publish_kind")
    expected_reason = scenario_metadata.get("expected_reason")
    if isinstance(expected_publish_kind, str) and isinstance(expected_reason, str):
        observed_publish_kind = incremental.get("publish_kind")
        observed_reason = incremental.get("exact_reason")
        passed = observed_publish_kind == expected_publish_kind and observed_reason == expected_reason
        result = {
            "passed": passed,
            "applicable": True,
            "contract": "safe_full_rebuild",
            "expected_publish_kind": expected_publish_kind,
            "observed_publish_kind": observed_publish_kind,
            "expected_reason": expected_reason,
            "observed_reason": observed_reason,
        }
        if not passed:
            result["reason"] = "observed fallback route does not match the fixture contract"
        return result
    expected = scenario_metadata.get("expected_minimum_affected_files")
    if not isinstance(expected, int):
        return {"passed": True, "applicable": False}
    exact_delta = incremental.get("response", {}).get("exact_delta", {})
    observed = exact_delta.get("affected_paths")
    if not isinstance(observed, int):
        observed = incremental.get("exact_route_detail", {}).get("frontier_expanded_files")
    if isinstance(exact_cap, int) and exact_cap < expected:
        observed_publish_kind = incremental.get("publish_kind")
        observed_reason = incremental.get("exact_reason")
        truncated = exact_delta.get("affected_paths_truncated") is True
        passed = (
            observed_publish_kind in {PUBLISH_FULL, PUBLISH_INCREMENTAL_CONTAINMENT}
            and observed_reason == "frontier_too_large"
            and truncated
        )
        result = {
            "passed": passed,
            "applicable": True,
            "contract": "configured_cap_fallback",
            "configured_exact_cap": exact_cap,
            "expected_minimum_affected_files": expected,
            "observed_affected_files": observed,
            "observed_publish_kind": observed_publish_kind,
            "observed_reason": observed_reason,
            "affected_paths_truncated": truncated,
        }
        if not passed:
            result["reason"] = (
                "configured cap fallback requires containment/full publication, "
                "frontier_too_large, and truncation evidence"
            )
        return result
    passed = isinstance(observed, int) and observed >= expected
    result = {
        "passed": passed,
        "applicable": True,
        "contract": "exact_frontier",
        "expected_minimum_affected_files": expected,
        "observed_affected_files": observed,
    }
    if not passed:
        result["reason"] = "observed frontier is smaller than the fixture contract"
    return result


def build_env(cache_dir: Path) -> dict[str, str]:
    env = dict(os.environ)
    env["CBM_CACHE_DIR"] = str(cache_dir)
    env["CBM_AUTO_INDEX"] = "false"
    env["CBM_CONTEXT_INJECTION"] = "false"
    # The supervisor retains successful worker logs only in profile mode. The
    # harness streams their exact memory/timing markers before cleaning the cache.
    env["CBM_PROFILE"] = "1"
    return env


def prepare_matrix_scenario(
    name: str,
    repo_dir: Path,
    files: int,
    funcs_per_file: int,
    args: argparse.Namespace,
    case_root: Path,
) -> dict[str, Any]:
    frontier_language = MATRIX_FRONTIER_SCENARIOS.get(name)
    if frontier_language:
        return create_inbound_frontier_repo(repo_dir, frontier_language, args.frontier_files)
    if name in {
        "go_modify_1",
        "go_modify_2",
        "go_create",
        "go_delete",
        "go_rename",
        "go_new_folder",
    }:
        create_repo(repo_dir, files, funcs_per_file)
        return {"source": "synthetic_go"}
    if name == "route_decorator":
        create_route_repo(repo_dir, "/api/orders")
        return {"source": "synthetic_route"}
    if name == "python_reexport":
        create_python_reexport_repo(repo_dir)
        return {"source": "synthetic_python_reexport"}
    if name == "fastapi_insert_probe":
        return copy_fastapi_head_to_case(args, repo_dir, case_root)
    raise ValueError(f"unknown matrix scenario: {name}")


def mutate_matrix_scenario(name: str, repo_dir: Path, funcs_per_file: int) -> list[str]:
    frontier_language = MATRIX_FRONTIER_SCENARIOS.get(name)
    if frontier_language:
        return mutate_inbound_frontier_repo(repo_dir, frontier_language)
    if name == "go_modify_1":
        return modify_existing_files(repo_dir, 1, funcs_per_file)
    if name == "go_modify_2":
        return modify_existing_files(repo_dir, 2, funcs_per_file)
    if name == "go_create":
        rel = Path("pkg") / "file_created.go"
        write_text(repo_dir / rel, go_file_content(9999, 1, funcs_per_file))
        return [rel.as_posix()]
    if name == "go_delete":
        rel = Path("pkg") / "file_0000.go"
        (repo_dir / rel).unlink()
        return [rel.as_posix()]
    if name == "go_rename":
        old_rel = Path("pkg") / "file_0000.go"
        new_rel = Path("pkg") / "file_renamed.go"
        (repo_dir / old_rel).unlink()
        write_text(repo_dir / new_rel, go_file_content(9998, 1, funcs_per_file))
        return [old_rel.as_posix(), new_rel.as_posix()]
    if name == "go_new_folder":
        rel = Path("newpkg") / "leaf.go"
        write_text(repo_dir / rel, "package newpkg\n\nfunc NewFolderLeaf() int {\n\treturn 23\n}\n")
        return [rel.as_posix()]
    if name == "route_decorator":
        create_route_repo(repo_dir, "/api/items")
        return ["routes.py"]
    if name == "python_reexport":
        rel = Path("fastapi") / "__init__.py"
        write_text(repo_dir / rel, "from .openapi.models import Header\n")
        return [rel.as_posix()]
    if name == "fastapi_insert_probe":
        rel = Path(FASTAPI_PROBE_REL_PATH)
        path = repo_dir / rel
        source = path.read_text(encoding="utf-8")
        insert = (
            "\n"
            "    def cbm_frontier_noop_mask_probe(self) -> int:\n"
            f"        return {FASTAPI_PROBE_RETURN_VALUE}\n"
        )
        if FASTAPI_PROBE_INSERT_BEFORE not in source:
            raise RuntimeError(f"FastAPI probe insertion point not found: {rel.as_posix()}")
        mutated = source.replace(FASTAPI_PROBE_INSERT_BEFORE, insert + FASTAPI_PROBE_INSERT_BEFORE, 1)
        try:
            compile(mutated, rel.as_posix(), "exec")
        except SyntaxError as exc:
            raise RuntimeError(f"FastAPI probe mutation produced invalid Python: {exc}") from exc
        path.write_text(mutated, encoding="utf-8")
        return [rel.as_posix()]
    raise ValueError(f"unknown matrix scenario: {name}")


def resolve_git_repo_root(repo_root: Path, timeout: int) -> Path:
    root = repo_root.expanduser().resolve()
    return Path(command_stdout(["git", "rev-parse", "--show-toplevel"], timeout, root)).resolve()


def git_metadata(repo_root: Path, timeout: int) -> dict[str, Any]:
    def maybe(args: list[str]) -> str:
        try:
            return command_stdout(["git", *args], timeout, repo_root)
        except Exception as exc:  # noqa: BLE001 - metadata should not abort benchmark execution.
            return f"<error: {type(exc).__name__}: {exc}>"

    return {
        "repo_root": str(repo_root),
        "head": maybe(["rev-parse", "HEAD"]),
        "short_head": maybe(["rev-parse", "--short", "HEAD"]),
        "branch": maybe(["branch", "--show-current"]),
        "dirty_status_short": maybe(["status", "--short"]),
    }


def binary_metadata(binary: Path) -> dict[str, Any]:
    digest = hashlib.sha256()
    with binary.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    stat = binary.stat()
    return {
        "path": str(binary.resolve()),
        "size_bytes": stat.st_size,
        "sha256": digest.hexdigest(),
    }


def clone_real_repo(url: str, target: Path, timeout: int) -> Path:
    target.parent.mkdir(parents=True, exist_ok=True)
    proc, _ = command_result(
        ["git", "clone", "--depth=1", url, str(target)],
        dict(os.environ),
        timeout,
    )
    if proc.returncode != 0:
        raise RuntimeError(f"git clone failed for {url}: {proc.stderr.strip()}")
    return target


def resolve_fastapi_source(args: argparse.Namespace, case_root: Path) -> Path:
    candidates: list[Path] = []
    if args.fastapi_repo:
        candidates.append(Path(args.fastapi_repo).expanduser())
    env_repo = os.environ.get("CBM_FASTAPI_REPO")
    if env_repo:
        candidates.append(Path(env_repo).expanduser())
    candidates.extend(
        [
            Path.home() / "source" / "fastapi",
            Path.home() / ".cache" / "codebase-memory-mcp" / "bench-repos" / "fastapi",
        ]
    )
    for candidate in candidates:
        if (candidate / FASTAPI_PROBE_REL_PATH).is_file():
            return resolve_git_repo_root(candidate, args.timeout)
    if not args.clone_missing_real_repos:
        searched = ", ".join(str(path) for path in candidates)
        raise RuntimeError(
            "fastapi_insert_probe requires --fastapi-repo, CBM_FASTAPI_REPO, "
            f"or --clone-missing-real-repos; searched: {searched}"
        )
    return clone_real_repo(args.fastapi_url, case_root / "source-fastapi", args.timeout)


def copy_git_head_to_dir(source_repo: Path, dest: Path, timeout: int) -> None:
    if dest.exists() and any(dest.iterdir()):
        raise RuntimeError(f"destination is not empty: {dest}")
    dest.mkdir(parents=True, exist_ok=True)
    raw = command_stdout_bytes(
        ["git", "ls-tree", "-r", "--name-only", "-z", "HEAD"], timeout, source_repo
    )
    rel_paths = [
        item.decode("utf-8", "surrogateescape")
        for item in raw.split(b"\0")
        if item
    ]
    for rel_path in rel_paths:
        blob = command_stdout_bytes(["git", "show", f"HEAD:{rel_path}"], timeout, source_repo)
        target = dest / rel_path
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(blob)


def copy_fastapi_head_to_case(
    args: argparse.Namespace, repo_dir: Path, case_root: Path
) -> dict[str, Any]:
    source_repo = resolve_fastapi_source(args, case_root)
    copy_git_head_to_dir(source_repo, repo_dir, args.timeout)
    return {
        "source_repo": str(source_repo),
        "source_git": git_metadata(source_repo, args.timeout),
        "copy_policy": "git_tracked_files_from_HEAD",
    }


def create_self_dogfood_worktree(source_repo: Path, case_root: Path, timeout: int) -> Path:
    repo_dir = case_root / SELF_DOGFOOD_REPO_SUBDIR
    if repo_dir.exists():
        raise RuntimeError(f"self-dogfood worktree already exists: {repo_dir}")
    proc, _ = command_result(
        ["git", "worktree", "add", "--detach", str(repo_dir), "HEAD"],
        dict(os.environ),
        timeout,
        source_repo,
    )
    if proc.returncode != 0:
        raise RuntimeError(f"git worktree add failed: {proc.stderr.strip()}")
    return repo_dir


def remove_self_dogfood_worktree(source_repo: Path, repo_dir: Path, timeout: int) -> dict[str, Any]:
    cleanup: dict[str, Any] = {"requested": True, "path": str(repo_dir), "removed": False}
    proc, _ = command_result(
        ["git", "worktree", "remove", "--force", str(repo_dir)],
        dict(os.environ),
        timeout,
        source_repo,
    )
    if proc.returncode != 0:
        cleanup["git_worktree_remove_error"] = proc.stderr.strip()
        shutil.rmtree(repo_dir, ignore_errors=True)
    cleanup["removed"] = not repo_dir.exists()
    return cleanup


def self_dogfood_marker(name: str) -> str:
    return f"{SELF_DOGFOOD_MARKER_PREFIX}_{name}"


def append_c_marker_function(repo_dir: Path, rel_path: str, marker: str, value: int) -> str:
    append_text(
        repo_dir / rel_path,
        (
            "\n"
            f"static int {marker}(void) {{\n"
            f"    return {value};\n"
            "}\n"
        ),
    )
    return rel_path


def mutate_self_dogfood_scenario(name: str, repo_dir: Path) -> dict[str, Any]:
    marker = self_dogfood_marker(name)
    changed: list[str] = []
    if name == "noop":
        return {"marker": None, "changed_paths": changed, "description": "no source mutation"}
    if name == "one_source_file":
        changed.append(
            append_c_marker_function(repo_dir, "src/pipeline/pipeline_internal.h", marker, 4101)
        )
        return {"marker": marker, "changed_paths": changed, "description": "single C header edit"}
    if name == "route_handler":
        append_text(
            repo_dir / "src/ui/http_server.c",
            (
                "\n"
                f"static int {marker}(const char *path) {{\n"
                '    return cbm_http_path_match(path, "/api/pan4-oracle");\n'
                "}\n"
            ),
        )
        changed.append("src/ui/http_server.c")
        return {
            "marker": marker,
            "changed_paths": changed,
            "description": "HTTP UI handler source edit with route literal oracle",
        }
    if name == "store_pipeline_batch":
        changed.append(append_c_marker_function(repo_dir, "src/store/store.h", marker, 4103))
        second_marker = f"{marker}_pipeline"
        changed.append(
            append_c_marker_function(repo_dir, "src/pipeline/pipeline_internal.h", second_marker, 4104)
        )
        return {
            "marker": marker,
            "secondary_marker": second_marker,
            "changed_paths": changed,
            "description": "small store plus pipeline header batch",
        }
    if name == "multi_file_small":
        changed.append(append_c_marker_function(repo_dir, "src/mcp/mcp.c", marker, 4105))
        second_marker = f"{marker}_test"
        changed.append(append_c_marker_function(repo_dir, "tests/test_mcp.c", second_marker, 4106))
        return {
            "marker": marker,
            "secondary_marker": second_marker,
            "changed_paths": changed,
            "description": "small production plus test source batch",
        }
    raise ValueError(f"unknown self-dogfood scenario: {name}")


def oracle_passed(tool_result: dict[str, Any], marker: str | None) -> bool:
    if not marker:
        return True
    response = tool_result.get("response")
    return marker in json.dumps(response, sort_keys=True)


def score_quality_oracles(
    oracles: dict[str, Any],
    expectations: dict[str, tuple[str | None, str]],
) -> dict[str, Any]:
    """Attach auditable per-oracle verdicts and summarize applicable checks."""
    applicable_count = 0
    passed_count = 0
    reciprocal_rank_total = 0.0
    hit_at_1_count = 0
    hit_at_5_count = 0
    for name, result in oracles.items():
        if not isinstance(result, dict):
            continue
        expected, criterion = expectations.get(name, (None, "no quality criterion"))
        applicable = expected is not None
        passed = False
        rank: int | None = None
        returned_count: int | None = None
        if applicable:
            applicable_count += 1
            response = result.get("response")
            passed = expected in json.dumps(response, separators=(",", ":"), sort_keys=True)
            passed_count += int(passed)
            ranked_items = (
                response.get("results")
                if isinstance(response, dict) and isinstance(response.get("results"), list)
                else response if isinstance(response, list) else [response]
            )
            returned_count = len(ranked_items)
            for position, item in enumerate(ranked_items, start=1):
                if expected in json.dumps(item, separators=(",", ":"), sort_keys=True):
                    rank = position
                    break
            reciprocal_rank = 1.0 / rank if rank is not None else 0.0
            reciprocal_rank_total += reciprocal_rank
            hit_at_1_count += int(rank == 1)
            hit_at_5_count += int(rank is not None and rank <= 5)
        else:
            reciprocal_rank = None
        result["quality"] = {
            "applicable": applicable,
            "passed": passed if applicable else None,
            "criterion": criterion,
            "expected_substring": expected,
            "rank": rank,
            "returned_count": returned_count,
            "reciprocal_rank": reciprocal_rank,
            "hit_at_1": rank == 1 if applicable else None,
            "hit_at_5": rank is not None and rank <= 5 if applicable else None,
        }
    mean_reciprocal_rank = (
        reciprocal_rank_total / applicable_count if applicable_count else None
    )
    return {
        "passed": passed_count == applicable_count,
        "passed_count": passed_count,
        "applicable_count": applicable_count,
        "binary_pass_rate": round(passed_count / applicable_count, 6) if applicable_count else None,
        "mean_reciprocal_rank": (
            round(mean_reciprocal_rank, 6) if mean_reciprocal_rank is not None else None
        ),
        "hit_at_1": round(hit_at_1_count / applicable_count, 6) if applicable_count else None,
        "hit_at_5": round(hit_at_5_count / applicable_count, 6) if applicable_count else None,
        "score": round(mean_reciprocal_rank, 6) if mean_reciprocal_rank is not None else None,
    }


def run_self_dogfood_oracles(
    transport: str,
    binary: Path,
    env: dict[str, str],
    project: str,
    mutation: dict[str, Any],
    args: argparse.Namespace,
    client: McpClient | None = None,
) -> dict[str, Any]:
    marker = mutation.get("marker")
    changed_paths = list(mutation.get("changed_paths") or [])
    first_changed = changed_paths[0] if changed_paths else ""
    oracles: dict[str, Any] = {}
    expectations: dict[str, tuple[str | None, str]] = {}
    if marker:
        search_code_args: dict[str, Any] = {"project": project, "pattern": marker, "limit": 5}
        if first_changed:
            search_code_args["file_pattern"] = Path(first_changed).name
            search_code_args["path_filter"] = f"^{re.escape(first_changed)}$"
        oracles["marker_search_graph"] = run_tool_call_for_transport(
            transport,
            binary,
            env,
            "search_graph",
            {"project": project, "name_pattern": marker, "limit": 5},
            args.timeout,
            args.include_logs,
            client,
        )
        expectations["marker_search_graph"] = (marker, "mutated symbol appears in graph search")
        oracles["marker_search_code"] = run_tool_call_for_transport(
            transport,
            binary,
            env,
            "search_code",
            search_code_args,
            args.timeout,
            args.include_logs,
            client,
        )
        expectations["marker_search_code"] = (marker, "mutated symbol appears in source search")
    if first_changed:
        oracles["changed_file_query_graph"] = run_tool_call_for_transport(
            transport,
            binary,
            env,
            "query_graph",
            {
                "project": project,
                "query": (
                    "MATCH (n) WHERE n.file_path CONTAINS "
                    f"'{first_changed}' RETURN n.name, n.label, n.file_path LIMIT 10"
                ),
            },
            args.timeout,
            args.include_logs,
            client,
        )
        expectations["changed_file_query_graph"] = (
            first_changed,
            "changed file path appears in graph query",
        )
        oracles["scoped_architecture"] = run_tool_call_for_transport(
            transport,
            binary,
            env,
            "get_architecture",
            {"project": project, "path": first_changed, "aspects": ["all"]},
            args.timeout,
            args.include_logs,
            client,
        )
        expectations["scoped_architecture"] = (
            first_changed,
            "changed file path appears in scoped architecture",
        )
    route_expected = "/api/pan4-oracle" if mutation.get("description", "").startswith(
        "HTTP UI handler"
    ) else None
    route_arguments: dict[str, Any] = {"project": project, "label": "Route", "limit": 5}
    if route_expected:
        route_arguments["name_pattern"] = "pan4-oracle"
    oracles["route_freshness_probe"] = run_tool_call_for_transport(
        transport,
        binary,
        env,
        "search_graph",
        route_arguments,
        args.timeout,
        args.include_logs,
        client,
    )
    expectations["route_freshness_probe"] = (
        route_expected,
        "new route literal appears in route search" if route_expected else "route mutation not applicable",
    )
    quality = score_quality_oracles(oracles, expectations)
    oracles["quality"] = quality
    oracles["passed"] = quality["passed"]
    return oracles


def run_index_for_transport(
    transport: str,
    binary: Path,
    env: dict[str, str],
    repo_dir: Path,
    timeout: int,
    include_logs: bool,
    client: McpClient | None = None,
) -> dict[str, Any]:
    if transport == "mcp":
        if client is None:
            raise RuntimeError("MCP transport requires an active client")
        return run_index_mcp(client, repo_dir, include_logs)
    return run_index(binary, env, repo_dir, timeout, include_logs)


def run_matrix_case(
    scenario: str,
    binary: Path,
    env: dict[str, str],
    case_root: Path,
    args: argparse.Namespace,
) -> dict[str, Any]:
    repo_dir = case_root / "repo"
    cache_dir = case_root / "cache"
    case_root.mkdir(parents=True, exist_ok=True)
    if scenario not in MATRIX_REAL_REPO_SCENARIOS:
        repo_dir.mkdir(parents=True, exist_ok=True)
    cache_dir.mkdir(parents=True, exist_ok=True)
    case_env = dict(env)
    case_env["CBM_CACHE_DIR"] = str(cache_dir)
    run_config_set(binary, case_env, "incremental_reindex", "always", args.timeout)
    run_config_set(binary, case_env, "rank_refresh", args.rank_refresh, args.timeout)
    apply_config_overrides(binary, case_env, args.config_overrides, args.timeout)

    scenario_metadata = prepare_matrix_scenario(
        scenario, repo_dir, args.files, args.functions_per_file, args, case_root
    )
    if args.transport == "mcp":
        with McpClient(binary, case_env, args.timeout) as client:
            initial = run_index_for_transport(
                args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs, client
            )
            changed_paths = mutate_matrix_scenario(scenario, repo_dir, args.functions_per_file)
            incremental = run_index_for_transport(
                args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs, client
            )
    else:
        initial = run_index_for_transport(
            args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs
        )
        changed_paths = mutate_matrix_scenario(scenario, repo_dir, args.functions_per_file)
        incremental = run_index_for_transport(
            args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs
        )

    project_db = find_project_db(cache_dir)
    project = str(incremental.get("response", {}).get("project") or project_db.stem)
    incremental_snapshot = case_root / "incremental.db"
    copy_sqlite_snapshot(project_db, incremental_snapshot)
    removed_dbs = remove_project_dbs(cache_dir)

    if args.transport == "mcp":
        with McpClient(binary, case_env, args.timeout) as client:
            full_rebuild = run_index_for_transport(
                args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs, client
            )
    else:
        full_rebuild = run_index_for_transport(
            args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs
        )

    full_db = find_project_db(cache_dir)
    canonical = compare_canonical_graph(incremental_snapshot, full_db, project)
    incremental_reason = incremental.get("exact_reason")
    publish_kind = incremental.get("publish_kind")
    active_overlay = None
    if publish_kind == PUBLISH_INCREMENTAL_OVERLAY:
        active_overlay = compare_active_overlay_graph(incremental_snapshot, full_db, project)
    graph_gate = graph_gate_for_publish_kind(
        canonical, str(publish_kind or ""), active_overlay=active_overlay
    )
    configured_cap = args.config_overrides.get("incremental_exact_max_affected_paths")
    try:
        exact_cap = int(configured_cap) if configured_cap is not None else None
    except ValueError:
        exact_cap = None
    frontier_gate = frontier_coverage_gate(scenario_metadata, incremental, exact_cap=exact_cap)
    explicit_route = is_explicit_incremental_route(publish_kind, incremental_reason)
    passed = bool(graph_gate.get("passed")) and bool(frontier_gate.get("passed")) and explicit_route
    speedup = max(1, int(full_rebuild["elapsed_ms"])) / max(1, int(incremental["elapsed_ms"]))
    return {
        "scenario": scenario,
        "project": project,
        "changed_paths": changed_paths,
        "scenario_metadata": scenario_metadata,
        "removed_project_dbs": removed_dbs,
        "initial_fast_full": initial,
        "incremental": incremental,
        "fresh_fast_full_after_change": full_rebuild,
        "canonical_graph": canonical,
        "active_overlay_graph": active_overlay,
        "graph_gate": graph_gate,
        "frontier_coverage_gate": frontier_gate,
        "explicit_exact_or_fallback": explicit_route,
        "explicit_incremental_route": explicit_route,
        "exact_reason": incremental_reason,
        "speedup_full_rebuild_over_incremental": speedup,
        "passed": passed,
    }


def run_matrix(args: argparse.Namespace, binary: Path) -> tuple[dict[str, Any], int]:
    auto_root = not bool(args.work_root)
    work_root = Path(args.work_root).expanduser() if args.work_root else Path(
        tempfile.mkdtemp(prefix="cbm-incr-matrix-")
    )
    work_root.mkdir(parents=True, exist_ok=True)
    scenarios = [item.strip() for item in args.matrix_scenarios.split(",") if item.strip()]
    report: dict[str, Any] = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "binary": str(binary),
        "binary_metadata": binary_metadata(binary),
        "work_root": str(work_root),
        "mode": "matrix",
        "parameters": {
            "files": args.files,
            "functions_per_file": args.functions_per_file,
            "frontier_files": args.frontier_files,
            "rank_refresh": args.rank_refresh,
            "config_profile": args.config_profile,
            "config_overrides": args.config_overrides,
            "timeout": args.timeout,
            "transport": args.transport,
            "scenarios": scenarios,
        },
        "cleanup": {"requested": auto_root and not args.keep_work_root, "removed": False},
        "cases": [],
    }
    exit_code = 1
    try:
        base_env = build_env(work_root / "cache-base")
        for scenario in scenarios:
            case = run_matrix_case(scenario, binary, base_env, work_root / scenario, args)
            report["cases"].append(case)
        report["derived"] = {
            "passed": all(bool(case.get("passed")) for case in report["cases"]),
            "case_count": len(report["cases"]),
        }
        exit_code = 0 if report["derived"]["passed"] else 1
    except Exception as exc:
        record_report_error(report, exc)
        exit_code = 1
    finally:
        if auto_root and not args.keep_work_root:
            shutil.rmtree(work_root, ignore_errors=True)
            report["cleanup"]["removed"] = not work_root.exists()
        if args.out:
            out_path = Path(args.out).expanduser()
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        print(json.dumps(report, indent=2, sort_keys=True))
    return report, exit_code


def run_self_dogfood_case(
    scenario: str,
    source_repo: Path,
    binary: Path,
    case_root: Path,
    args: argparse.Namespace,
) -> dict[str, Any]:
    cache_dir = case_root / SELF_DOGFOOD_CACHE_SUBDIR
    cache_dir.mkdir(parents=True, exist_ok=True)
    repo_dir = create_self_dogfood_worktree(source_repo, case_root, args.timeout)
    case_env = build_env(cache_dir)
    cleanup: dict[str, Any] = {"requested": not args.keep_work_root, "removed": False}
    result: dict[str, Any] | None = None
    try:
        run_config_set(binary, case_env, "incremental_reindex", "always", args.timeout)
        run_config_set(binary, case_env, "rank_refresh", args.rank_refresh, args.timeout)
        apply_config_overrides(binary, case_env, args.config_overrides, args.timeout)
        if args.transport == "mcp":
            with McpClient(binary, case_env, args.timeout) as client:
                initial = run_index_for_transport(
                    args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs, client
                )
                mutation = mutate_self_dogfood_scenario(scenario, repo_dir)
                incremental = run_index_for_transport(
                    args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs, client
                )
                project_db = find_project_db(cache_dir)
                project = str(incremental.get("response", {}).get("project") or project_db.stem)
                oracles = run_self_dogfood_oracles(
                    args.transport, binary, case_env, project, mutation, args, client
                )
        else:
            initial = run_index_for_transport(
                args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs
            )
            mutation = mutate_self_dogfood_scenario(scenario, repo_dir)
            incremental = run_index_for_transport(
                args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs
            )
            project_db = find_project_db(cache_dir)
            project = str(incremental.get("response", {}).get("project") or project_db.stem)
            oracles = run_self_dogfood_oracles(
                args.transport, binary, case_env, project, mutation, args
            )

        incremental_snapshot = case_root / "incremental.db"
        copy_sqlite_snapshot(project_db, incremental_snapshot)
        removed_dbs = remove_project_dbs(cache_dir)
        if args.transport == "mcp":
            with McpClient(binary, case_env, args.timeout) as client:
                full_rebuild = run_index_for_transport(
                    args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs, client
                )
        else:
            full_rebuild = run_index_for_transport(
                args.transport, binary, case_env, repo_dir, args.timeout, args.include_logs
            )
        full_db = find_project_db(cache_dir)
        canonical = compare_canonical_graph(incremental_snapshot, full_db, project)
        publish_kind = incremental.get("publish_kind")
        incremental_reason = incremental.get("exact_reason")
        active_overlay = None
        if publish_kind == PUBLISH_INCREMENTAL_OVERLAY:
            active_overlay = compare_active_overlay_graph(incremental_snapshot, full_db, project)
        explicit_route = is_explicit_incremental_route(publish_kind, incremental_reason)
        speedup = max(1, int(full_rebuild["elapsed_ms"])) / max(1, int(incremental["elapsed_ms"]))
        graph_gate = graph_gate_for_publish_kind(
            canonical,
            str(publish_kind or ""),
            bool(oracles.get("passed")),
            active_overlay=active_overlay,
        )
        passed = bool(graph_gate.get("passed")) and explicit_route and bool(oracles.get("passed"))
        result = {
            "scenario": scenario,
            "project": project,
            "repo_dir": str(repo_dir),
            "mutation": mutation,
            "removed_project_dbs": removed_dbs,
            "initial_fast_full": initial,
            "incremental": incremental,
            "fresh_fast_full_after_change": full_rebuild,
            "canonical_graph": canonical,
            "active_overlay_graph": active_overlay,
            "graph_gate": graph_gate,
            "oracles": oracles,
            "explicit_incremental_route": explicit_route,
            "exact_reason": incremental_reason,
            "speedup_full_rebuild_over_incremental": speedup,
            "passed": passed,
        }
    finally:
        if not args.keep_work_root:
            cleanup = remove_self_dogfood_worktree(source_repo, repo_dir, args.timeout)
        if cache_dir.exists() and not args.keep_work_root:
            shutil.rmtree(cache_dir, ignore_errors=True)
            cleanup["cache_removed"] = not cache_dir.exists()
        cleanup["case_root_removed"] = False
        if not args.keep_work_root and case_root.exists():
            shutil.rmtree(case_root, ignore_errors=True)
            cleanup["case_root_removed"] = not case_root.exists()
    if result is None:
        raise RuntimeError(f"self-dogfood case did not produce a result: {scenario}")
    result["cleanup"] = cleanup
    return result


def run_self_dogfood(args: argparse.Namespace, binary: Path) -> tuple[dict[str, Any], int]:
    auto_root = not bool(args.work_root)
    work_root = Path(args.work_root).expanduser() if args.work_root else Path(
        tempfile.mkdtemp(prefix="cbm-self-dogfood-")
    )
    work_root.mkdir(parents=True, exist_ok=True)
    source_repo = resolve_git_repo_root(Path(args.repo_root), args.timeout)
    scenarios = [item.strip() for item in args.self_dogfood_scenarios.split(",") if item.strip()]
    report: dict[str, Any] = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "binary": str(binary),
        "binary_metadata": binary_metadata(binary),
        "work_root": str(work_root),
        "source_repo": str(source_repo),
        "source_git": git_metadata(source_repo, args.timeout),
        "mode": "self_dogfood",
        "parameters": {
            "rank_refresh": args.rank_refresh,
            "config_profile": args.config_profile,
            "config_overrides": args.config_overrides,
            "timeout": args.timeout,
            "transport": args.transport,
            "scenarios": scenarios,
        },
        "cleanup": {"requested": auto_root and not args.keep_work_root, "removed": False},
        "cases": [],
    }
    exit_code = 1
    try:
        for scenario in scenarios:
            case = run_self_dogfood_case(
                scenario, source_repo, binary, work_root / scenario, args
            )
            report["cases"].append(case)
        report["derived"] = {
            "passed": all(bool(case.get("passed")) for case in report["cases"]),
            "case_count": len(report["cases"]),
        }
        exit_code = 0 if report["derived"]["passed"] else 1
    except Exception as exc:
        record_report_error(report, exc)
        exit_code = 1
    finally:
        if auto_root and not args.keep_work_root:
            shutil.rmtree(work_root, ignore_errors=True)
            report["cleanup"]["removed"] = not work_root.exists()
        if args.out:
            out_path = Path(args.out).expanduser()
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        print(json.dumps(report, indent=2, sort_keys=True))
    return report, exit_code


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Gate exact fast-mode incremental indexing against a fresh full rebuild."
    )
    parser.add_argument("--binary", default="build/c/codebase-memory-mcp")
    parser.add_argument("--work-root", default="")
    parser.add_argument("--repo-root", default=".")
    parser.add_argument("--out", default="")
    parser.add_argument("--files", type=int, default=DEFAULT_FILE_COUNT)
    parser.add_argument("--functions-per-file", type=int, default=DEFAULT_FUNCTIONS_PER_FILE)
    parser.add_argument("--changed-files", type=int, default=DEFAULT_CHANGED_FILES)
    parser.add_argument("--min-speedup", type=float, default=DEFAULT_MIN_SPEEDUP)
    parser.add_argument(
        "--rank-refresh",
        choices=("eager", "stale_on_exact", "stale_on_incremental"),
        default=DEFAULT_RANK_REFRESH,
    )
    parser.add_argument(
        "--config-profile",
        choices=tuple(CONFIG_PROFILES),
        default=CONFIG_PROFILE_DEFAULT,
        help=(
            "Named, auditable capability profile. minimal_indexing disables dependency "
            "indexing plus every optional graph/rank pass; repeated --config KEY=VALUE "
            "arguments take priority over the profile."
        ),
    )
    parser.add_argument(
        "--config",
        action="append",
        default=[],
        metavar="KEY=VALUE",
        help="Additional config override; repeat to set multiple keys. Applied after built-in settings.",
    )
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SECONDS)
    parser.add_argument("--keep-work-root", action="store_true")
    parser.add_argument("--include-logs", action="store_true")
    parser.add_argument("--matrix", action="store_true", help="Run the affected-frontier scenario matrix.")
    parser.add_argument(
        "--self-dogfood",
        action="store_true",
        help="Run isolated edit-loop scenarios against a detached worktree of --repo-root.",
    )
    parser.add_argument(
        "--matrix-scenarios",
        default=MATRIX_SCENARIOS_DEFAULT,
        help="Comma-separated matrix scenarios to run.",
    )
    parser.add_argument(
        "--frontier-files",
        type=int,
        default=DEFAULT_FRONTIER_FILES,
        help=(
            "Number of inbound-dependent source files created by each *_inbound_frontier "
            "matrix scenario. The changed definition file is additional."
        ),
    )
    parser.add_argument(
        "--fastapi-repo",
        default="",
        help=(
            "Existing FastAPI checkout for matrix scenario fastapi_insert_probe. "
            "Defaults also check CBM_FASTAPI_REPO and common local cache/source paths."
        ),
    )
    parser.add_argument(
        "--fastapi-url",
        default=DEFAULT_FASTAPI_URL,
        help="Clone URL used only with --clone-missing-real-repos.",
    )
    parser.add_argument(
        "--clone-missing-real-repos",
        action="store_true",
        help="Clone missing real benchmark repos into the isolated work root.",
    )
    parser.add_argument(
        "--self-dogfood-scenarios",
        default=SELF_DOGFOOD_SCENARIOS_DEFAULT,
        help="Comma-separated real-repo edit-loop scenarios to run.",
    )
    parser.add_argument(
        "--transport",
        choices=("cli", "mcp"),
        default="cli",
        help="Measure cold CLI subprocess calls or persistent MCP tool-call latency.",
    )
    parser.add_argument(
        "--overhead-probes",
        type=int,
        default=DEFAULT_OVERHEAD_PROBES,
        help=(
            "Run N cheap tool-call probes before indexing to estimate invocation overhead; "
            "0 preserves the historical gate behavior."
        ),
    )
    parser.add_argument(
        "--overhead-tool",
        default=DEFAULT_OVERHEAD_TOOL,
        help="Existing MCP tool used by --overhead-probes.",
    )
    args = parser.parse_args()
    args.config_overrides = resolve_config_overrides(args.config_profile, args.config)
    return args


def resolve_binary_path(binary_arg: str) -> Path:
    binary = Path(binary_arg).expanduser()
    if binary.is_absolute():
        return binary.resolve()
    cwd_candidate = (Path.cwd() / binary).resolve()
    if cwd_candidate.is_file():
        return cwd_candidate
    script_candidate = (Path(__file__).resolve().parents[1] / binary).resolve()
    return script_candidate


def main() -> int:
    args = parse_args()
    binary = resolve_binary_path(args.binary)
    if not binary.is_file():
        print(f"error: binary not found: {binary}", file=sys.stderr)
        return 2
    if args.matrix:
        _, matrix_exit_code = run_matrix(args, binary)
        return matrix_exit_code
    if args.self_dogfood:
        _, self_dogfood_exit_code = run_self_dogfood(args, binary)
        return self_dogfood_exit_code

    auto_root = not bool(args.work_root)
    work_root = Path(args.work_root).expanduser() if args.work_root else Path(
        tempfile.mkdtemp(prefix="cbm-incr-speed-")
    )
    work_root.mkdir(parents=True, exist_ok=True)
    repo_dir = work_root / "repo"
    cache_dir = work_root / "cache"
    repo_dir.mkdir(parents=True, exist_ok=True)
    cache_dir.mkdir(parents=True, exist_ok=True)

    report: dict[str, Any] = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "binary": str(binary),
        "binary_metadata": binary_metadata(binary),
        "work_root": str(work_root),
        "parameters": {
            "files": args.files,
            "functions_per_file": args.functions_per_file,
            "changed_files": args.changed_files,
            "min_speedup": args.min_speedup,
            "rank_refresh": args.rank_refresh,
            "config_profile": args.config_profile,
            "config_overrides": args.config_overrides,
            "timeout": args.timeout,
            "transport": args.transport,
            "overhead_probes": args.overhead_probes,
            "overhead_tool": args.overhead_tool,
        },
        "cleanup": {"requested": auto_root and not args.keep_work_root, "removed": False},
    }

    exit_code = 1
    try:
        create_repo(repo_dir, args.files, args.functions_per_file)
        env = build_env(cache_dir)
        run_config_set(binary, env, "incremental_reindex", "always", args.timeout)
        run_config_set(binary, env, "rank_refresh", args.rank_refresh, args.timeout)
        apply_config_overrides(binary, env, args.config_overrides, args.timeout)

        if args.transport == "mcp":
            with McpClient(binary, env, args.timeout) as client:
                overhead_probe = measure_mcp_overhead_probes(
                    client, args.overhead_tool, args.overhead_probes, args.include_logs
                )
                initial = run_index_mcp(client, repo_dir, args.include_logs)
                changed_paths = modify_existing_files(
                    repo_dir, args.changed_files, args.functions_per_file
                )
                incremental = run_index_mcp(client, repo_dir, args.include_logs)
            removed_dbs = remove_project_dbs(cache_dir)
            with McpClient(binary, env, args.timeout) as client:
                full_rebuild = run_index_mcp(client, repo_dir, args.include_logs)
        else:
            overhead_probe = measure_cli_overhead_probes(
                binary,
                env,
                args.overhead_tool,
                args.overhead_probes,
                args.timeout,
                args.include_logs,
            )
            initial = run_index(binary, env, repo_dir, args.timeout, args.include_logs)
            changed_paths = modify_existing_files(
                repo_dir, args.changed_files, args.functions_per_file
            )
            incremental = run_index(binary, env, repo_dir, args.timeout, args.include_logs)
            removed_dbs = remove_project_dbs(cache_dir)
            full_rebuild = run_index(binary, env, repo_dir, args.timeout, args.include_logs)

        incr_ms = max(1, int(incremental["elapsed_ms"]))
        full_ms = max(1, int(full_rebuild["elapsed_ms"]))
        speedup = full_ms / incr_ms
        incremental_markers = incremental["markers"]
        explicit_incremental_route = is_incremental_publish_kind(
            str(incremental.get("publish_kind") or "")
        )
        defer_marker = bool(incremental_markers["pagerank_defer"])
        passed = speedup >= args.min_speedup and explicit_incremental_route

        report.update(
            {
                "changed_paths": changed_paths,
                "removed_project_dbs": removed_dbs,
                "measurements": {
                    "overhead_probe": overhead_probe,
                    "initial_fast_full": initial,
                    "incremental_exact": incremental,
                    "incremental": incremental,
                    "fresh_fast_full_after_change": full_rebuild,
                },
                "derived": {
                    "speedup_full_rebuild_over_incremental": speedup,
                    "exact_incremental_marker_seen": explicit_incremental_route,
                    "explicit_incremental_route_seen": explicit_incremental_route,
                    "rank_defer_marker_seen": defer_marker,
                    "passed": passed,
                },
            }
        )
        exit_code = 0 if passed else 1
    except Exception as exc:
        record_report_error(report, exc)
        exit_code = 1
    finally:
        if auto_root and not args.keep_work_root:
            shutil.rmtree(work_root, ignore_errors=True)
            report["cleanup"]["removed"] = not work_root.exists()
        if args.out:
            out_path = Path(args.out).expanduser()
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        print(json.dumps(report, indent=2, sort_keys=True))

    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())