aituner/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py

#!/usr/bin/env python3
"""Blind Frontier/community-vLLM rank comparison on one fixed prompt cohort.

The earlier grid selected a different ``sampling_u`` subset at every load.  That
confounds offered load with prompt mix and makes binary-search monotonicity an
untested assumption.  This runner selects one length-stratified cohort, changes
load only by scaling its complete arrival timeline, evaluates every discrete
load, freezes Frontier first, and only then permits the community-vLLM run.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import math
import os
import random
import statistics
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import asdict, replace
from pathlib import Path
from typing import Any, Iterable


SCRIPT_DIR = Path(__file__).resolve().parent
REPO_ROOT = SCRIPT_DIR.parents[2]
for search_path in (SCRIPT_DIR, REPO_ROOT / "src"):
    if str(search_path) not in sys.path:
        sys.path.insert(0, str(search_path))

import community_prefill_grid as community_grid  # noqa: E402
import frontier_prefill_grid as frontier_grid  # noqa: E402
from aituner.engine import build_launch_recipe  # noqa: E402
from aituner.spec import ConfigPatch, load_study_spec  # noqa: E402
from aituner.trace import TraceRequest, load_trace_requests  # noqa: E402
from aituner.worker import (  # noqa: E402
    _ignore_sigterm_if_main,
    _install_sigterm_as_keyboardinterrupt,
    _replay_requests,
    _restore_sigterm,
    _terminate_process_tree,
    _wait_for_server_or_exit,
)


PROTOCOL_SCHEMA = "qwen235b-prefill-fixed-cohort-protocol-v1"
FRONTIER_SCHEMA = "frontier-qwen235b-prefill-fixed-cohort-v1"
COMMUNITY_SCHEMA = "community-qwen235b-prefill-fixed-cohort-v1"
COMPARISON_SCHEMA = "frontier-community-qwen235b-rank-comparison-v1"
TRIAL_STABILITY_SCHEMA = "community-qwen235b-trial-stability-v1"
REPEAT_SELECTION_SCHEMA = "community-qwen235b-boundary-repeat-selection-v4"
COHORT_SIZE = 64
COHORT_SEED = 2026071501
CONFIG_ORDER_SEED = 2026071502
RATE_ORDER_SEED = 2026071503
WARMUP_SEED = 2026071504
INPUT_BINS = (0, 1024, 2048, 4096, 8192, 16384, 32769)
OFFERED_RATES = (0.15, 0.25, 0.35, 0.50, 0.75, 1.00, 1.50)
TARGET_PASS_RATE = 0.95
EXPECTED_QUANT_SIGNATURE = "method=fp8|act=dynamic|serialized=True|block=128x128"

SLO_VARIANTS: dict[str, dict[str, Any]] = {
    "linear_8k_primary": {
        "description": "TTFT <= 1000 ms + input_tokens / 8000 tokens/s",
        "kind": "linear_ms",
        "intercept_ms": 1000.0,
        "per_token_ms": 0.125,
    },
    "linear_6k": {
        "description": "TTFT <= 1000 ms + input_tokens / 6000 tokens/s",
        "kind": "linear_ms",
        "intercept_ms": 1000.0,
        "per_token_ms": 1.0 / 6.0,
    },
    "linear_10k": {
        "description": "TTFT <= 1000 ms + input_tokens / 10000 tokens/s",
        "kind": "linear_ms",
        "intercept_ms": 1000.0,
        "per_token_ms": 0.1,
    },
    "legacy_step_1s_2s": {
        "description": "Original strict 1 s/2 s step SLO (null-capacity sensitivity)",
        "kind": "step_ms",
        "buckets": [
            {"max_input_tokens": 8191, "threshold_ms": 1000.0},
            {"threshold_ms": 2000.0},
        ],
    },
}
PRIMARY_SLO = "linear_8k_primary"


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        for chunk in iter(lambda: source.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode()).hexdigest()


def order_hash(values: Iterable[object]) -> str:
    return sha256_text("\n".join(str(value) for value in values))


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
    os.replace(temporary, path)


def rate_key(rate: float) -> str:
    return f"r_{rate:.3f}".replace(".", "p")


def _percentile(values: list[float], fraction: float) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    index = min(len(ordered) - 1, max(0, math.ceil(fraction * len(ordered)) - 1))
    return float(ordered[index])


def _latency_summary(values: list[float]) -> dict[str, Any]:
    return {
        "count": len(values),
        "mean": statistics.fmean(values) if values else None,
        "p50": _percentile(values, 0.50),
        "p95": _percentile(values, 0.95),
        "p99": _percentile(values, 0.99),
        "max": max(values) if values else None,
    }


def slo_threshold_ms(variant: dict[str, Any], input_tokens: int) -> float:
    if variant["kind"] == "linear_ms":
        return float(variant["intercept_ms"]) + float(variant["per_token_ms"]) * input_tokens
    if variant["kind"] != "step_ms":
        raise ValueError(f"unsupported SLO kind: {variant['kind']}")
    for bucket in variant["buckets"]:
        ceiling = bucket.get("max_input_tokens")
        if ceiling is None or input_tokens <= int(ceiling):
            return float(bucket["threshold_ms"])
    raise AssertionError("step SLO must have a terminal bucket")


def _input_bin(input_tokens: int) -> int:
    for index, (low, high) in enumerate(zip(INPUT_BINS, INPUT_BINS[1:])):
        if low <= input_tokens < high:
            return index
    raise ValueError(f"input length outside protocol bins: {input_tokens}")


def _allocate_quotas(counts: list[int], cohort_size: int) -> list[int]:
    total = sum(counts)
    if cohort_size <= 0 or cohort_size > total:
        raise ValueError(f"invalid cohort size {cohort_size} for {total} rows")
    exact = [cohort_size * count / total for count in counts]
    quotas = [min(count, math.floor(value)) for count, value in zip(counts, exact)]
    remaining = cohort_size - sum(quotas)
    order = sorted(
        range(len(counts)),
        key=lambda index: (-(exact[index] - math.floor(exact[index])), index),
    )
    while remaining:
        advanced = False
        for index in order:
            if quotas[index] >= counts[index]:
                continue
            quotas[index] += 1
            remaining -= 1
            advanced = True
            if not remaining:
                break
        if not advanced:
            raise AssertionError("could not allocate complete cohort")
    return quotas


def load_source_rows(trace: Path) -> list[dict[str, Any]]:
    rows = []
    with trace.open(encoding="utf-8") as source:
        for source_index, line in enumerate(source):
            if not line.strip():
                continue
            raw = json.loads(line)
            input_tokens = int(raw["input_length"])
            if not 0 <= input_tokens <= 32768:
                continue
            prompt = raw.get("prompt")
            if not isinstance(prompt, str) or not prompt:
                raise ValueError(f"row {source_index} lacks raw prompt")
            rows.append(
                {
                    "source_row_index": source_index,
                    "source_request_id": str(
                        raw.get("request_id") or raw.get("id") or source_index
                    ),
                    "source_arrival_s": float(raw["timestamp"]),
                    "sampling_u": float(raw["sampling_u"]),
                    "input_tokens": input_tokens,
                    "prompt_sha256": sha256_text(prompt),
                    "input_bin": _input_bin(input_tokens),
                }
            )
    return rows


def select_cohort(
    rows: list[dict[str, Any]], *, cohort_size: int, seed: int
) -> tuple[list[dict[str, Any]], list[int]]:
    by_bin = [[] for _ in range(len(INPUT_BINS) - 1)]
    for row in rows:
        by_bin[int(row["input_bin"])].append(row)
    quotas = _allocate_quotas([len(items) for items in by_bin], cohort_size)
    selected = []
    for bin_index, (items, quota) in enumerate(zip(by_bin, quotas)):
        ranked = sorted(
            items,
            key=lambda row: sha256_text(
                f"{seed}|{bin_index}|{row['source_row_index']}|{row['prompt_sha256']}"
            ),
        )
        selected.extend(ranked[:quota])
    selected.sort(key=lambda row: (row["source_arrival_s"], row["source_row_index"]))
    if len(selected) != cohort_size:
        raise AssertionError("cohort selection returned wrong size")
    return selected, quotas


def prepare_protocol(args: argparse.Namespace) -> Path:
    trace = args.trace.resolve()
    actual_sha = sha256(trace)
    if args.expected_trace_sha256 and actual_sha != args.expected_trace_sha256:
        raise ValueError(
            f"trace SHA256 mismatch: expected={args.expected_trace_sha256}, actual={actual_sha}"
        )
    rows = load_source_rows(trace)
    cohort, quotas = select_cohort(rows, cohort_size=args.cohort_size, seed=args.seed)
    first_arrival = float(cohort[0]["source_arrival_s"])
    source_span = float(cohort[-1]["source_arrival_s"]) - first_arrival
    if source_span <= 0:
        raise ValueError("selected cohort has no arrival span")

    output_root = args.output_root.resolve()
    trace_dir = output_root / "traces"
    trace_dir.mkdir(parents=True, exist_ok=True)
    fields = [
        "arrived_at",
        "num_prefill_tokens",
        "num_decode_tokens",
        "source_row_index",
        "source_request_id",
        "source_arrival_s",
        "sampling_u",
        "slo_ttft_ms",
    ]
    rates = {}
    for rate in args.rate:
        duration_s = len(cohort) / rate
        target = trace_dir / f"{rate_key(rate)}.csv"
        with target.open("w", encoding="utf-8", newline="") as output:
            writer = csv.DictWriter(output, fieldnames=fields)
            writer.writeheader()
            for row in cohort:
                arrival_s = (
                    (float(row["source_arrival_s"]) - first_arrival)
                    / source_span
                    * duration_s
                )
                writer.writerow(
                    {
                        "arrived_at": f"{arrival_s:.12f}",
                        "num_prefill_tokens": row["input_tokens"],
                        "num_decode_tokens": 1,
                        "source_row_index": row["source_row_index"],
                        "source_request_id": row["source_request_id"],
                        "source_arrival_s": f"{row['source_arrival_s']:.12f}",
                        "sampling_u": f"{row['sampling_u']:.12f}",
                        "slo_ttft_ms": f"{slo_threshold_ms(SLO_VARIANTS[PRIMARY_SLO], int(row['input_tokens'])):.6f}",
                    }
                )
        rates[rate_key(rate)] = {
            "offered_request_rate": rate,
            "duration_s": duration_s,
            "path": str(target.resolve()),
            "sha256": sha256(target),
            "request_count": len(cohort),
            "request_rate_recomputed": len(cohort) / duration_s,
            "source_row_order_sha256": order_hash(
                row["source_row_index"] for row in cohort
            ),
            "input_length_order_sha256": order_hash(
                row["input_tokens"] for row in cohort
            ),
        }

    bin_counts = [0] * (len(INPUT_BINS) - 1)
    for row in rows:
        bin_counts[int(row["input_bin"])] += 1
    manifest = {
        "schema": PROTOCOL_SCHEMA,
        "created_unix_s": time.time(),
        "source": {
            "path": str(trace),
            "sha256": actual_sha,
            "eligible_request_count": len(rows),
        },
        "selection": {
            "method": "length_stratified_smallest_sha256",
            "seed": args.seed,
            "cohort_size": len(cohort),
            "input_bin_edges": list(INPUT_BINS),
            "source_bin_counts": bin_counts,
            "cohort_bin_quotas": quotas,
            "stable_order": ["source_arrival_s", "source_row_index"],
            "cohort_source_row_order_sha256": order_hash(
                row["source_row_index"] for row in cohort
            ),
            "source_arrival_span_s": source_span,
        },
        "load_contract": {
            "only_mutated_variable": "uniform_arrival_timeline_scale",
            "request_count_per_point": len(cohort),
            "completion_tokens_override": 1,
            "target_pass_rate": TARGET_PASS_RATE,
            "binary_search": False,
            "monotonicity_assumed": False,
            "offered_rates": list(args.rate),
        },
        "slo_variants": SLO_VARIANTS,
        "primary_slo": PRIMARY_SLO,
        "cohort": cohort,
        "rates": rates,
    }
    manifest_path = output_root / "protocol_manifest.json"
    write_json(manifest_path, manifest)
    audit_protocol(manifest_path)
    print(manifest_path)
    return manifest_path


def audit_protocol(manifest_path: Path) -> dict[str, Any]:
    manifest = json.loads(manifest_path.read_text())
    if manifest.get("schema") != PROTOCOL_SCHEMA:
        raise ValueError(f"unexpected protocol schema: {manifest.get('schema')}")
    expected_ids = [str(row["source_row_index"]) for row in manifest["cohort"]]
    expected_lengths = [int(row["input_tokens"]) for row in manifest["cohort"]]
    rate_checks = []
    for record in manifest["rates"].values():
        path = Path(record["path"])
        if sha256(path) != record["sha256"]:
            raise ValueError(f"trace hash mismatch: {path}")
        with path.open(newline="") as source:
            rows = list(csv.DictReader(source))
        ids = [row["source_row_index"] for row in rows]
        lengths = [int(row["num_prefill_tokens"]) for row in rows]
        arrivals = [float(row["arrived_at"]) for row in rows]
        if ids != expected_ids or lengths != expected_lengths:
            raise ValueError(f"cohort/order drift in {path}")
        if arrivals != sorted(arrivals) or abs(arrivals[0]) > 1e-9:
            raise ValueError(f"invalid arrival sequence in {path}")
        recomputed = len(rows) / (arrivals[-1] - arrivals[0])
        if not math.isclose(
            recomputed, float(record["offered_request_rate"]), rel_tol=1e-9
        ):
            raise ValueError(f"offered-rate mismatch in {path}: {recomputed}")
        rate_checks.append(
            {"rate": record["offered_request_rate"], "recomputed_rate": recomputed}
        )
    result = {
        "status": "passed",
        "protocol_manifest_sha256": sha256(manifest_path),
        "cohort_size": len(expected_ids),
        "rate_checks": rate_checks,
    }
    write_json(manifest_path.parent / "protocol_audit.json", result)
    return result


def _read_trace(path: Path) -> list[dict[str, Any]]:
    with path.open(encoding="utf-8", newline="") as source:
        return list(csv.DictReader(source))


def score_requests(requests: list[dict[str, Any]]) -> dict[str, Any]:
    scores = {}
    for name, variant in SLO_VARIANTS.items():
        passed = 0
        for request in requests:
            threshold = slo_threshold_ms(variant, int(request["input_tokens"]))
            passed += int(
                bool(request["success"])
                and request.get("ttft_ms") is not None
                and float(request["ttft_ms"]) <= threshold
            )
        count = len(requests)
        pass_rate = passed / count if count else 0.0
        scores[name] = {
            "request_count": count,
            "passed_request_count": passed,
            "slo_pass_rate": pass_rate,
            "target_pass_rate": TARGET_PASS_RATE,
            "feasible": pass_rate >= TARGET_PASS_RATE,
        }
    ttfts = [
        float(request["ttft_ms"])
        for request in requests
        if request.get("ttft_ms") is not None
    ]
    return {"scores": scores, "ttft_ms": _latency_summary(ttfts)}


def _capacity_record(
    results: list[dict[str, Any]], *, config: dict[str, Any], slo_name: str
) -> dict[str, Any]:
    ordered = sorted(results, key=lambda item: float(item["offered_request_rate"]))
    feasible = [
        float(item["offered_request_rate"])
        for item in ordered
        if item["scores"][slo_name]["feasible"]
    ]
    vector = [bool(item["scores"][slo_name]["feasible"]) for item in ordered]
    violations = []
    for lower_index in range(len(vector)):
        for higher_index in range(lower_index + 1, len(vector)):
            if not vector[lower_index] and vector[higher_index]:
                violations.append(
                    [
                        float(ordered[lower_index]["offered_request_rate"]),
                        float(ordered[higher_index]["offered_request_rate"]),
                    ]
                )
    capacity = max(feasible) if feasible else None
    tp = int(config["tp"])
    return {
        "config": config,
        "slo": slo_name,
        "tested_rates": [float(item["offered_request_rate"]) for item in ordered],
        "pass_rates": [float(item["scores"][slo_name]["slo_pass_rate"]) for item in ordered],
        "feasibility": vector,
        "maximum_tested_feasible_request_rate": capacity,
        "maximum_tested_feasible_request_rate_per_gpu": (
            capacity / tp if capacity is not None else None
        ),
        "lower_censored": capacity is None,
        "upper_censored": bool(feasible) and capacity == max(item["offered_request_rate"] for item in ordered),
        "monotonicity_violations": violations,
    }


def rank_surface(
    config_results: list[dict[str, Any]], *, slo_name: str
) -> list[dict[str, Any]]:
    records = [
        _capacity_record(item["loads"], config=item["config"], slo_name=slo_name)
        for item in config_results
    ]
    records.sort(
        key=lambda item: (
            -(
                item["maximum_tested_feasible_request_rate_per_gpu"]
                if item["maximum_tested_feasible_request_rate_per_gpu"] is not None
                else -1.0
            ),
            item["config"]["name"],
        )
    )
    previous_value = object()
    previous_rank = 0
    for index, record in enumerate(records, start=1):
        value = record["maximum_tested_feasible_request_rate_per_gpu"]
        if value != previous_value:
            previous_rank = index
            previous_value = value
        record["rank"] = previous_rank
    return records


def _validate_profile_contract(profile_paths: dict[str, Path]) -> dict[str, Any]:
    payload = json.loads(profile_paths["manifest"].read_text())
    contract = payload.get("contract") or {}
    if contract.get("quant_signature") != EXPECTED_QUANT_SIGNATURE:
        raise ValueError(f"unexpected FP8 profile contract: {contract}")
    if contract.get("profiling_precision") != "BF16":
        raise ValueError(f"unexpected output/accumulation precision: {contract}")
    for name, path in (
        ("linear_op.csv", profile_paths["linear"]),
        ("attention.csv", profile_paths["attention"]),
        ("moe.csv", profile_paths["moe"]),
    ):
        expected = payload["outputs"][name]["sha256"]
        if sha256(path) != expected:
            raise ValueError(f"profile output hash mismatch: {path}")
    return contract


def run_frontier_load(
    *,
    args: argparse.Namespace,
    profile_paths: dict[str, Path],
    config: Any,
    rate_record: dict[str, Any],
) -> dict[str, Any]:
    rate = float(rate_record["offered_request_rate"])
    load_dir = args.output_root / "runs" / config.name / rate_key(rate)
    result_path = load_dir / "result.json"
    if result_path.is_file():
        result = json.loads(result_path.read_text())
        if result.get("status") == "completed":
            return result
    load_dir.mkdir(parents=True, exist_ok=True)
    trace = Path(rate_record["path"])
    command = frontier_grid.build_command(
        python=args.python,
        frontier_source=args.frontier_source,
        profile_root=args.profile_root,
        profile_paths=profile_paths,
        trace=trace,
        config=config,
        probe_dir=load_dir,
        run_id=f"{config.name}_{rate_key(rate)}",
        cache_root=args.output_root / "cache" / config.name,
    )
    write_json(load_dir / "command.json", command)
    environment = os.environ.copy()
    environment.update(
        {
            "PYTHONPATH": str(args.frontier_source),
            "WANDB_DISABLED": "true",
            "VIDUR_DISABLE_WANDB": "1",
        }
    )
    started = time.time()
    with (load_dir / "stdout.log").open("w", encoding="utf-8") as output:
        completed = subprocess.run(
            command,
            cwd=args.frontier_source,
            env=environment,
            stdout=output,
            stderr=subprocess.STDOUT,
            check=False,
        )
    if completed.returncode != 0:
        raise RuntimeError(
            f"Frontier failed: config={config.name}, rate={rate}, rc={completed.returncode}"
        )
    request_metrics = frontier_grid.find_request_metrics(load_dir)
    trace_rows = _read_trace(trace)
    with request_metrics.open(encoding="utf-8", newline="") as source:
        metric_rows = list(csv.DictReader(source))
    metrics_by_id = {int(row["Request Id"]): row for row in metric_rows}
    if set(metrics_by_id) != set(range(len(trace_rows))):
        raise ValueError(f"Frontier Request Id mismatch: {request_metrics}")
    requests = [
        {
            "request_id": row["source_row_index"],
            "input_tokens": int(row["num_prefill_tokens"]),
            "success": True,
            "ttft_ms": float(metrics_by_id[index]["ttft"]),
        }
        for index, row in enumerate(trace_rows)
    ]
    score = score_requests(requests)
    result = {
        "status": "completed",
        "config": asdict(config) | {"name": config.name},
        "offered_request_rate": rate,
        "request_rate_per_gpu": rate / config.gpu_count,
        "elapsed_seconds": time.time() - started,
        "trace_path": str(trace),
        "trace_sha256": sha256(trace),
        "request_metrics_path": str(request_metrics),
        "request_metrics_sha256": sha256(request_metrics),
        "requests": requests,
        **score,
    }
    write_json(result_path, result)
    return result


def run_frontier(args: argparse.Namespace) -> Path:
    protocol = json.loads(args.protocol_manifest.read_text())
    if protocol.get("schema") != PROTOCOL_SCHEMA:
        raise ValueError("invalid fixed-cohort protocol manifest")
    audit_protocol(args.protocol_manifest)
    profile_paths = frontier_grid.resolve_profile_paths(args.profile_root)
    profile_contract = _validate_profile_contract(profile_paths)
    ep_equivalence = json.loads(args.ep_equivalence_artifact.read_text())
    if (
        ep_equivalence.get("schema") != "frontier-ep-batched-lane-equivalence-v1"
        or not ep_equivalence.get("byte_identical")
    ):
        raise ValueError("batched EP lane prediction lacks byte-identical evidence")
    if Path(ep_equivalence["new_source"]).resolve() != args.frontier_source.resolve():
        raise ValueError("EP equivalence was measured on a different Frontier source")
    configs = frontier_grid.selected_configs(args.config)
    args.output_root.mkdir(parents=True, exist_ok=True)
    run_manifest = {
        "schema": FRONTIER_SCHEMA,
        "created_unix_s": time.time(),
        "protocol_manifest": {
            "path": str(args.protocol_manifest.resolve()),
            "sha256": sha256(args.protocol_manifest),
        },
        "frontier": {
            "source": str(args.frontier_source.resolve()),
            "python": str(args.python.resolve()),
            "fingerprint": frontier_grid.frontier_source_fingerprint(
                args.frontier_source, args.frontier_commit
            ),
        },
        "profiles": {
            "contract": profile_contract,
            "files": {
                name: {"path": str(path.resolve()), "sha256": sha256(path)}
                for name, path in profile_paths.items()
            },
        },
        "ep_prediction_optimization_equivalence": {
            "path": str(args.ep_equivalence_artifact.resolve()),
            "sha256": sha256(args.ep_equivalence_artifact),
            "request_metrics_byte_identical": True,
            "request_metrics_sha256": ep_equivalence["new_sha256"],
        },
        "kv_capacity_evidence": {
            "tp4": {
                "path": str(args.tp4_capacity_artifact.resolve()),
                "sha256": sha256(args.tp4_capacity_artifact),
                "num_gpu_blocks": 26101,
            },
            "tp8": {
                "path": str(args.tp8_capacity_artifact.resolve()),
                "sha256": sha256(args.tp8_capacity_artifact),
                "num_gpu_blocks": 62351,
            },
        },
        "best_effort_contract": {
            "arrival_semantics": "preserve_materialized_trace_arrivals",
            "quantization": "block_fp8_w8a8_dynamic_with_bf16_output_accumulation",
            "execution_profiles": "community_vllm_0.10.2_serving_entrypoints",
            "moe_ep_prefill": "global_routing_then_slowest_local_ep_lane",
            "moe_ep_critical_lane_cache": "exact_(total_routed_tokens,layer_id)_memoization",
            "moe_ep_lane_inference": "eight_independent_rows_in_one_exact_forest_predict_call",
            "kv_blocks": "measured_community_vllm_and_fixed_per_topology",
            "cpu_overhead_modeling": "disabled_no_native_community_vllm_records",
            "end_to_end_action_specific_calibration": False,
            "real_serving_results_observed_before_freeze": False,
        },
        "orchestration": {
            "max_parallel_configs": args.max_parallel_configs,
            "cache_isolation": "one_cache_root_per_config",
            "loads_within_config": "sequential",
        },
        "configs": [asdict(config) | {"name": config.name} for config in configs],
    }
    write_json(args.output_root / "run_manifest.json", run_manifest)

    rate_records = sorted(
        protocol["rates"].values(), key=lambda item: item["offered_request_rate"]
    )

    def run_config(config: Any) -> dict[str, Any]:
        loads = []
        for record in rate_records:
            result = run_frontier_load(
                args=args,
                profile_paths=profile_paths,
                config=config,
                rate_record=record,
            )
            loads.append(result)
            print(
                json.dumps(
                    {
                        "system": "Frontier",
                        "config": config.name,
                        "rate": result["offered_request_rate"],
                        "pass_rate": result["scores"][PRIMARY_SLO]["slo_pass_rate"],
                        "feasible": result["scores"][PRIMARY_SLO]["feasible"],
                        "elapsed_seconds": result["elapsed_seconds"],
                    },
                    sort_keys=True,
                ),
                flush=True,
            )
        summary = {"config": asdict(config) | {"name": config.name}, "loads": loads}
        write_json(args.output_root / "results" / f"{config.name}.json", summary)
        return summary

    results_by_name = {}
    with ThreadPoolExecutor(max_workers=args.max_parallel_configs) as pool:
        futures = {pool.submit(run_config, config): config for config in configs}
        for future in as_completed(futures):
            config = futures[future]
            results_by_name[config.name] = future.result()
    config_results = [results_by_name[config.name] for config in configs]
    freeze = {
        "schema": FRONTIER_SCHEMA,
        "status": "frozen_before_community_run",
        "created_unix_s": time.time(),
        "run_manifest_sha256": sha256(args.output_root / "run_manifest.json"),
        "protocol_manifest_sha256": sha256(args.protocol_manifest),
        "primary_slo": PRIMARY_SLO,
        "rankings": {
            name: rank_surface(config_results, slo_name=name) for name in SLO_VARIANTS
        },
        "config_results": config_results,
    }
    path = args.output_root / "frontier_ranking_frozen.json"
    write_json(path, freeze)
    print(path)
    return path


def _community_study_payload(
    *,
    tp: int,
    repo: Path,
    python: Path,
    vllm: Path,
    model: Path,
    trace: Path,
    windows: Path,
    port: int,
) -> dict[str, Any]:
    payload = community_grid.study_payload(
        tp=tp,
        repo=repo,
        python=python,
        vllm=vllm,
        model=model,
        trace=trace,
        windows=windows,
        port=port,
    )
    payload["study_id"] = f"community-qwen235b-prefill-fixed-cohort-tp{tp}-v1"
    payload["slo"] = {
        "target_pass_rate": TARGET_PASS_RATE,
        "ttft_rule": {
            key: value
            for key, value in SLO_VARIANTS[PRIMARY_SLO].items()
            if key != "description"
        },
    }
    # The dedicated runner supplies exact fixed-cohort arrivals and disables
    # SLO-based early stopping; these trace values only define raw prompt loading.
    payload["trace"]["early_stop_max_lag_s"] = None
    payload["trace"]["early_stop_max_elapsed_s"] = None
    return payload


def _validate_frontier_freeze(path: Path, protocol_manifest: Path) -> dict[str, Any]:
    freeze = json.loads(path.read_text())
    if freeze.get("schema") != FRONTIER_SCHEMA:
        raise ValueError(f"unexpected Frontier freeze schema: {freeze.get('schema')}")
    if freeze.get("status") != "frozen_before_community_run":
        raise ValueError("Frontier result is not marked frozen")
    if freeze.get("protocol_manifest_sha256") != sha256(protocol_manifest):
        raise ValueError("Frontier and community protocol manifests differ")
    if len(freeze.get("config_results") or []) != len(frontier_grid.GRID):
        raise ValueError("Frontier freeze is incomplete")
    protocol = json.loads(protocol_manifest.read_text())
    expected_rates = {
        float(record["offered_request_rate"])
        for record in protocol["rates"].values()
    }
    for item in freeze["config_results"]:
        rates = {float(load["offered_request_rate"]) for load in item["loads"]}
        if rates != expected_rates:
            raise ValueError(f"Frontier freeze lacks rates for {item['config']['name']}")
    return freeze


def prepare_community(args: argparse.Namespace) -> Path:
    protocol = json.loads(args.protocol_manifest.read_text())
    if protocol.get("schema") != PROTOCOL_SCHEMA:
        raise ValueError("invalid fixed-cohort protocol manifest")
    _validate_frontier_freeze(args.frontier_freeze, args.protocol_manifest)
    repo = args.repo.resolve()
    python = args.python.resolve()
    vllm = args.vllm.resolve()
    model = args.model.resolve()
    windows = args.windows.resolve()
    trace = Path(protocol["source"]["path"])
    for path in (repo, python, vllm, model, windows, trace):
        if not path.exists():
            raise FileNotFoundError(path)
    if sha256(trace) != protocol["source"]["sha256"]:
        raise ValueError("raw prompt trace changed after protocol freeze")
    output_root = args.output_root.resolve()
    output_root.mkdir(parents=True, exist_ok=True)
    studies = {}
    for tp in (4, 8):
        path = output_root / "studies" / f"tp{tp}" / "study.json"
        write_json(
            path,
            _community_study_payload(
                tp=tp,
                repo=repo,
                python=python,
                vllm=vllm,
                model=model,
                trace=trace,
                windows=windows,
                port=args.port,
            ),
        )
        studies[str(tp)] = {"path": str(path), "sha256": sha256(path)}

    configs = list(community_grid.GRID)
    random.Random(args.config_order_seed).shuffle(configs)
    records = []
    for execution_index, config in enumerate(configs, start=1):
        rates = list(OFFERED_RATES)
        random.Random(f"{args.rate_order_seed}|{config.name}").shuffle(rates)
        records.append(
            {
                "execution_index": execution_index,
                "config": asdict(config) | {"name": config.name},
                "rate_order": rates,
                "study_path": studies[str(config.tp)]["path"],
                "result_path": str(output_root / "results" / f"{config.name}.json"),
                "engine_log_path": str(output_root / "runs" / config.name / "engine.log"),
            }
        )
    total_replay_seconds = sum(
        float(record["duration_s"]) for record in protocol["rates"].values()
    ) * len(configs)
    manifest = {
        "schema": COMMUNITY_SCHEMA,
        "created_unix_s": time.time(),
        "repository": community_grid.git_fingerprint(repo),
        "protocol_manifest": {
            "path": str(args.protocol_manifest.resolve()),
            "sha256": sha256(args.protocol_manifest),
        },
        "frontier_freeze": {
            "path": str(args.frontier_freeze.resolve()),
            "sha256": sha256(args.frontier_freeze),
        },
        "runtime": {"python": str(python), "vllm": str(vllm)},
        "model": {
            "path": str(model),
            "config_sha256": sha256(model / "config.json"),
        },
        "studies": studies,
        "execution_contract": {
            "request_mode": "raw_completion",
            "completion_tokens_override": 1,
            "prefix_caching": False,
            "client_max_concurrency": community_grid.CLIENT_MAX_CONCURRENCY,
            "slo_early_stop": False,
            "one_server_launch_per_config": True,
            "warmup": {
                "method": "one_disjoint_request_per_input_length_bin",
                "request_count": len(INPUT_BINS) - 1,
                "seed": WARMUP_SEED,
                "latencies_discarded": True,
            },
            "rate_order_randomized_within_config": True,
            "config_order_seed": args.config_order_seed,
            "rate_order_seed": args.rate_order_seed,
            "estimated_replay_wall_seconds_excluding_8_model_loads": total_replay_seconds,
        },
        "configs": records,
    }
    path = output_root / "run_manifest.json"
    write_json(path, manifest)
    print(path)
    return path


def _boundary_repeat_rates(
    config_result: dict[str, Any], *, slo_name: str = PRIMARY_SLO
) -> list[float]:
    """Select every adjacent load pair whose SLO feasibility label changes."""
    loads = sorted(
        config_result["loads"], key=lambda item: float(item["offered_request_rate"])
    )
    rates = [float(item["offered_request_rate"]) for item in loads]
    if len(rates) != len(set(rates)):
        raise ValueError(f"duplicate offered rate in {config_result['config']['name']}")
    if not rates:
        raise ValueError(f"no loads in {config_result['config']['name']}")
    feasibility = [bool(item["scores"][slo_name]["feasible"]) for item in loads]
    selected: set[float] = set()
    for index in range(len(loads) - 1):
        if feasibility[index] != feasibility[index + 1]:
            selected.update((rates[index], rates[index + 1]))
    if not selected:
        if len(rates) == 1:
            selected.add(rates[0])
        elif all(feasibility):
            selected.update(rates[-2:])
        else:
            selected.update(rates[:2])
    return sorted(selected)


def _slo_has_feasible_load(config_result: dict[str, Any], *, slo_name: str) -> bool:
    return any(
        bool(load["scores"][slo_name]["feasible"])
        for load in config_result["loads"]
    )


def prepare_community_repeat(args: argparse.Namespace) -> Path:
    """Prepare a fresh-server reverse-order repeat of decision-boundary loads."""
    selected_slos = list(getattr(args, "slo", [])) or list(SLO_VARIANTS)
    unknown_slos = sorted(set(selected_slos) - set(SLO_VARIANTS))
    if unknown_slos:
        raise ValueError(f"unknown repeat SLOs: {unknown_slos}")
    selected_configs = set(getattr(args, "config", []))
    source_manifest = json.loads(args.source_manifest.read_text())
    first_pass = json.loads(args.community_freeze.read_text())
    if source_manifest.get("schema") != COMMUNITY_SCHEMA:
        raise ValueError("unexpected source community manifest")
    if (
        first_pass.get("schema") != COMMUNITY_SCHEMA
        or first_pass.get("status") != "completed"
    ):
        raise ValueError("unexpected/incomplete first-pass community freeze")
    if first_pass.get("run_manifest_sha256") != sha256(args.source_manifest):
        raise ValueError("first-pass freeze does not match the source manifest")
    source_protocol_path = Path(source_manifest["protocol_manifest"]["path"])
    if sha256(source_protocol_path) != source_manifest["protocol_manifest"]["sha256"]:
        raise ValueError("source protocol changed before repeat preparation")
    protocol_path = (
        args.protocol_manifest.resolve()
        if args.protocol_manifest is not None
        else source_protocol_path
    )
    frontier_freeze_path = (
        args.frontier_freeze.resolve()
        if args.frontier_freeze is not None
        else Path(source_manifest["frontier_freeze"]["path"])
    )
    _validate_frontier_freeze(frontier_freeze_path, protocol_path)
    protocol = json.loads(protocol_path.read_text())
    duration_by_rate = {
        float(record["offered_request_rate"]): float(record["duration_s"])
        for record in protocol["rates"].values()
    }
    source_records = {
        item["config"]["name"]: item for item in source_manifest["configs"]
    }
    first_results = {
        item["config"]["name"]: item for item in first_pass["config_results"]
    }
    if set(source_records) != set(first_results):
        raise ValueError("source manifest and first-pass config sets differ")

    output_root = args.output_root.resolve()
    output_root.mkdir(parents=True, exist_ok=True)
    first_execution_order = sorted(
        source_manifest["configs"], key=lambda item: int(item["execution_index"])
    )
    if selected_configs:
        unknown_configs = sorted(selected_configs - set(source_records))
        if unknown_configs:
            raise ValueError(f"unknown repeat configs: {unknown_configs}")
        first_execution_order = [
            item
            for item in first_execution_order
            if item["config"]["name"] in selected_configs
        ]
    records = []
    selections = []
    for execution_index, source_record in enumerate(
        reversed(first_execution_order), start=1
    ):
        name = source_record["config"]["name"]
        excluded_unrankable_slos = [
            slo_name
            for slo_name in selected_slos
            if not _slo_has_feasible_load(
                first_results[name], slo_name=slo_name
            )
        ]
        boundary_rates_by_slo = {
            slo_name: (
                []
                if slo_name in excluded_unrankable_slos
                else _boundary_repeat_rates(
                    first_results[name], slo_name=slo_name
                )
            )
            for slo_name in selected_slos
        }
        boundary_rates = sorted(
            {
                rate
                for rates in boundary_rates_by_slo.values()
                for rate in rates
            }
        )
        refinement_rates = sorted(
            {
                rate
                for rates in boundary_rates_by_slo.values()
                for low, high in zip(rates, rates[1:])
                for rate in duration_by_rate
                if low < rate < high and rate not in boundary_rates
            }
        )
        selected_rates = sorted(set(boundary_rates) | set(refinement_rates))
        source_relative_order = [
            float(rate)
            for rate in source_record["rate_order"]
            if float(rate) in boundary_rates
        ]
        if set(source_relative_order) != set(boundary_rates):
            raise ValueError(f"repeat rate is absent from source order: {name}")
        reversed_boundaries = list(reversed(source_relative_order))
        repeat_order = [*reversed_boundaries, *reversed(refinement_rates)]
        records.append(
            {
                "execution_index": execution_index,
                "config": source_record["config"],
                "rate_order": repeat_order,
                "study_path": source_record["study_path"],
                "result_path": str(output_root / "results" / f"{name}.json"),
                "engine_log_path": str(output_root / "runs" / name / "engine.log"),
            }
        )
        selections.append(
            {
                "config": name,
                "excluded_unrankable_slos": excluded_unrankable_slos,
                "boundary_rates_by_slo": boundary_rates_by_slo,
                "boundary_rates": boundary_rates,
                "added_refinement_rates": refinement_rates,
                "selected_rates": selected_rates,
                "source_relative_order": source_relative_order,
                "repeat_order": repeat_order,
            }
        )
    total_replay_seconds = sum(
        duration_by_rate[float(rate)]
        for record in records
        for rate in record["rate_order"]
    )
    manifest = {
        "schema": COMMUNITY_SCHEMA,
        "created_unix_s": time.time(),
        "repository": community_grid.git_fingerprint(args.repo.resolve()),
        "protocol_manifest": {
            "path": str(protocol_path),
            "sha256": sha256(protocol_path),
        },
        "frontier_freeze": {
            "path": str(frontier_freeze_path),
            "sha256": sha256(frontier_freeze_path),
        },
        "runtime": source_manifest["runtime"],
        "model": source_manifest["model"],
        "studies": source_manifest["studies"],
        "repeat_of": {
            "source_manifest": {
                "path": str(args.source_manifest.resolve()),
                "sha256": sha256(args.source_manifest),
            },
            "first_pass_freeze": {
                "path": str(args.community_freeze.resolve()),
                "sha256": sha256(args.community_freeze),
            },
        },
        "execution_contract": {
            "schema": REPEAT_SELECTION_SCHEMA,
            "selection": "all_adjacent_pre_registered_slo_feasibility_transitions_plus_between_rate_refinement",
            "selected_slos": selected_slos,
            "selected_configs": sorted(selected_configs) if selected_configs else "all",
            "unrankable_slo_policy": "exclude_per_config_surfaces_with_no_feasible_load",
            "fallback_when_no_transition": "top_two_if_all_feasible_else_bottom_two",
            "config_order": "reverse_of_first_pass",
            "rate_order": "reverse_first_pass_boundary_order_then_descending_refinement_rates",
            "one_fresh_server_launch_per_config": True,
            "warmup": source_manifest["execution_contract"]["warmup"],
            "estimated_replay_wall_seconds_excluding_model_loads": total_replay_seconds,
            "selections": selections,
        },
        "configs": records,
    }
    path = output_root / "run_manifest.json"
    write_json(path, manifest)
    print(path)
    return path


def _materialize_real_requests(
    *,
    all_requests: list[TraceRequest],
    rate_record: dict[str, Any],
    cohort: list[dict[str, Any]],
) -> list[TraceRequest]:
    by_id = {request.row_id: request for request in all_requests}
    expected_prompt_hash = {
        str(row["source_row_index"]): row["prompt_sha256"] for row in cohort
    }
    trace_rows = _read_trace(Path(rate_record["path"]))
    requests = []
    for row in trace_rows:
        request_id = row["source_row_index"]
        if request_id not in by_id:
            raise ValueError(f"cohort request is absent from raw trace loader: {request_id}")
        request = by_id[request_id]
        prompt = request.body.get("prompt")
        if not isinstance(prompt, str) or sha256_text(prompt) != expected_prompt_hash[request_id]:
            raise ValueError(f"raw prompt fingerprint changed: request={request_id}")
        requests.append(replace(request, arrival_s=float(row["arrived_at"])))
    return requests


def _select_warmup_requests(
    all_requests: list[TraceRequest], *, cohort_ids: set[str]
) -> list[TraceRequest]:
    by_bin = [[] for _ in range(len(INPUT_BINS) - 1)]
    for request in all_requests:
        if request.row_id in cohort_ids or request.prompt_tokens_hint is None:
            continue
        input_tokens = int(request.prompt_tokens_hint)
        if not 0 <= input_tokens <= 32768:
            continue
        by_bin[_input_bin(input_tokens)].append(request)
    selected = []
    for bin_index, candidates in enumerate(by_bin):
        if not candidates:
            raise ValueError(f"no disjoint warmup request in input bin {bin_index}")
        selected.append(
            min(
                candidates,
                key=lambda request: sha256_text(
                    f"{WARMUP_SEED}|{bin_index}|{request.row_id}"
                ),
            )
        )
    return [replace(request, arrival_s=0.0) for request in selected]


def _run_real_replay(
    requests: list[TraceRequest], *, recipe: Any, study: Any, max_elapsed_s: float
) -> tuple[list[Any], bool, str]:
    return _replay_requests(
        requests,
        base_url=recipe.base_url,
        timeout_s=recipe.request_timeout_s,
        max_concurrency=study.trace.max_concurrency,
        # target=0 makes the mathematical SLO early-stop bound unreachable
        # while retaining the existing replay engine.
        target_pass_rate=0.0,
        max_lag_s=None,
        max_elapsed_s=max_elapsed_s,
        evaluate_outcome=lambda outcome: type(
            "NoEarlyStopEvaluation", (), {"passed": bool(outcome.success)}
        )(),
    )


def _real_request_records(
    requests: list[TraceRequest], outcomes: list[Any]
) -> list[dict[str, Any]]:
    outcomes_by_id = {outcome.request_id: outcome for outcome in outcomes}
    records = []
    for request in requests:
        outcome = outcomes_by_id.get(request.row_id)
        records.append(
            {
                "request_id": request.row_id,
                "arrival_s": request.arrival_s,
                "input_tokens": int(request.prompt_tokens_hint or 0),
                "success": bool(outcome and outcome.success),
                "ttft_ms": outcome.ttft_ms if outcome is not None else None,
                "tpot_ms": outcome.tpot_ms if outcome is not None else None,
                "completion_tokens": outcome.completion_tokens if outcome is not None else None,
                "completion_tokens_source": (
                    outcome.completion_tokens_source if outcome is not None else ""
                ),
                "error": outcome.error if outcome is not None else "missing_outcome",
            }
        )
    return records


def run_community_config(
    *,
    manifest: dict[str, Any],
    record: dict[str, Any],
    protocol: dict[str, Any],
) -> dict[str, Any]:
    result_path = Path(record["result_path"])
    if result_path.is_file():
        result = json.loads(result_path.read_text())
        if result.get("status") == "completed":
            return result
    config = record["config"]
    study_path = Path(record["study_path"])
    study = load_study_spec(study_path)
    _, all_requests = load_trace_requests(study, study_spec_path=study_path)
    recipe = build_launch_recipe(
        study.engine,
        ConfigPatch(
            flag_patch={
                "max-num-seqs": int(config["mns"]),
                "max-num-batched-tokens": int(config["mbt"]),
            }
        ),
    )
    run_dir = Path(record["engine_log_path"]).parent
    run_dir.mkdir(parents=True, exist_ok=True)
    write_json(run_dir / "engine_command.json", recipe.argv)
    marker_env = {
        "AITUNER_STUDY_ID": study.study_id,
        "AITUNER_TRIAL_ID": f"fixed-cohort-{config['name']}",
    }
    loads = []
    load_by_rate = {
        float(item["offered_request_rate"]): item
        for item in protocol["rates"].values()
    }
    started = time.time()
    with Path(record["engine_log_path"]).open("w", encoding="utf-8") as engine_log:
        process = subprocess.Popen(  # noqa: S603
            recipe.argv,
            cwd=recipe.cwd,
            env={**recipe.env, **marker_env},
            stdout=engine_log,
            stderr=subprocess.STDOUT,
            text=True,
            start_new_session=True,
        )
        previous_sigterm = _install_sigterm_as_keyboardinterrupt()
        try:
            _wait_for_server_or_exit(
                process,
                base_url=recipe.base_url,
                healthcheck_path=recipe.healthcheck_path,
                ready_timeout_s=recipe.ready_timeout_s,
            )
            cohort_ids = {
                str(row["source_row_index"]) for row in protocol["cohort"]
            }
            warmup_requests = _select_warmup_requests(
                all_requests, cohort_ids=cohort_ids
            )
            warmup_started = time.time()
            warmup_outcomes, warmup_early_stopped, warmup_stop_reason = _run_real_replay(
                warmup_requests,
                recipe=recipe,
                study=study,
                max_elapsed_s=1200.0,
            )
            warmup_records = _real_request_records(warmup_requests, warmup_outcomes)
            warmup_result = {
                "status": "completed"
                if not warmup_early_stopped
                and all(item["success"] for item in warmup_records)
                else "failed",
                "elapsed_seconds": time.time() - warmup_started,
                "early_stopped": warmup_early_stopped,
                "early_stop_reason": warmup_stop_reason,
                "cohort_disjoint": not bool(
                    cohort_ids & {item["request_id"] for item in warmup_records}
                ),
                "requests": warmup_records,
            }
            write_json(run_dir / "warmup.json", warmup_result)
            if warmup_result["status"] != "completed":
                raise RuntimeError(f"server warmup failed: {warmup_result}")
            time.sleep(2.0)
            for rate in record["rate_order"]:
                rate = float(rate)
                load_dir = run_dir / rate_key(rate)
                load_result_path = load_dir / "result.json"
                if load_result_path.is_file():
                    load_result = json.loads(load_result_path.read_text())
                    if load_result.get("status") == "completed":
                        loads.append(load_result)
                        continue
                load_dir.mkdir(parents=True, exist_ok=True)
                rate_record = load_by_rate[rate]
                requests = _materialize_real_requests(
                    all_requests=all_requests,
                    rate_record=rate_record,
                    cohort=protocol["cohort"],
                )
                load_started = time.time()
                outcomes, early_stopped, early_stop_reason = _run_real_replay(
                    requests,
                    recipe=recipe,
                    study=study,
                    max_elapsed_s=max(float(rate_record["duration_s"]) + 900.0, 1200.0),
                )
                request_records = _real_request_records(requests, outcomes)
                score = score_requests(request_records)
                load_result = {
                    "status": "completed",
                    "config": config,
                    "offered_request_rate": rate,
                    "request_rate_per_gpu": rate / int(config["tp"]),
                    "elapsed_seconds": time.time() - load_started,
                    "trace_path": rate_record["path"],
                    "trace_sha256": rate_record["sha256"],
                    "early_stopped": early_stopped,
                    "early_stop_reason": early_stop_reason,
                    "requests": request_records,
                    **score,
                }
                if early_stopped:
                    raise RuntimeError(
                        f"complete replay stopped early: {config['name']} rate={rate}: "
                        f"{early_stop_reason}"
                    )
                write_json(load_result_path, load_result)
                loads.append(load_result)
                print(
                    json.dumps(
                        {
                            "system": "community_vllm",
                            "config": config["name"],
                            "rate": rate,
                            "pass_rate": score["scores"][PRIMARY_SLO]["slo_pass_rate"],
                            "feasible": score["scores"][PRIMARY_SLO]["feasible"],
                            "elapsed_seconds": load_result["elapsed_seconds"],
                        },
                        sort_keys=True,
                    ),
                    flush=True,
                )
                time.sleep(2.0)
        finally:
            _ignore_sigterm_if_main()
            _terminate_process_tree(process, timeout_s=30.0, marker_env=marker_env)
            _restore_sigterm(previous_sigterm)
    loads.sort(key=lambda item: float(item["offered_request_rate"]))
    result = {
        "schema": COMMUNITY_SCHEMA,
        "status": "completed",
        "config": config,
        "elapsed_seconds": time.time() - started,
        "loads": loads,
    }
    write_json(result_path, result)
    return result


def run_community(args: argparse.Namespace) -> Path:
    manifest = json.loads(args.manifest.read_text())
    if manifest.get("schema") != COMMUNITY_SCHEMA:
        raise ValueError(f"unexpected community manifest schema: {manifest.get('schema')}")
    protocol_path = Path(manifest["protocol_manifest"]["path"])
    frontier_freeze = Path(manifest["frontier_freeze"]["path"])
    if sha256(protocol_path) != manifest["protocol_manifest"]["sha256"]:
        raise ValueError("protocol changed after community manifest preparation")
    if sha256(frontier_freeze) != manifest["frontier_freeze"]["sha256"]:
        raise ValueError("Frontier freeze changed after community manifest preparation")
    _validate_frontier_freeze(frontier_freeze, protocol_path)
    protocol = json.loads(protocol_path.read_text())
    config_results = []
    for record in sorted(manifest["configs"], key=lambda item: item["execution_index"]):
        result = run_community_config(
            manifest=manifest,
            record=record,
            protocol=protocol,
        )
        config_results.append(result)
        print(
            json.dumps(
                {
                    "config": result["config"]["name"],
                    "config_completed": True,
                    "elapsed_seconds": result["elapsed_seconds"],
                },
                sort_keys=True,
            ),
            flush=True,
        )
    freeze = {
        "schema": COMMUNITY_SCHEMA,
        "status": "completed",
        "created_unix_s": time.time(),
        "run_manifest_sha256": sha256(args.manifest),
        "protocol_manifest_sha256": sha256(protocol_path),
        "frontier_freeze_sha256": sha256(frontier_freeze),
        "primary_slo": PRIMARY_SLO,
        "rankings": {
            name: rank_surface(config_results, slo_name=name) for name in SLO_VARIANTS
        },
        "config_results": config_results,
    }
    path = args.manifest.parent / "community_ranking_frozen.json"
    write_json(path, freeze)
    print(path)
    return path


def _rank_map(ranking: list[dict[str, Any]]) -> dict[str, int]:
    return {item["config"]["name"]: int(item["rank"]) for item in ranking}


def _capacity_map(
    ranking: list[dict[str, Any]], *, field: str
) -> dict[str, float | None]:
    return {
        item["config"]["name"]: item[field]
        for item in ranking
    }


def _pearson(left: list[float], right: list[float]) -> float | None:
    if len(left) < 2:
        return None
    left_mean = statistics.fmean(left)
    right_mean = statistics.fmean(right)
    numerator = sum((a - left_mean) * (b - right_mean) for a, b in zip(left, right))
    denominator = math.sqrt(
        sum((a - left_mean) ** 2 for a in left)
        * sum((b - right_mean) ** 2 for b in right)
    )
    return numerator / denominator if denominator else None


def _average_ranks(capacities: dict[str, float | None]) -> dict[str, float]:
    ordered = sorted(
        capacities,
        key=lambda name: (
            -(
                capacities[name]
                if capacities[name] is not None
                else float("-inf")
            ),
            name,
        ),
    )
    ranks = {}
    start = 0
    while start < len(ordered):
        value = capacities[ordered[start]]
        end = start + 1
        while end < len(ordered) and capacities[ordered[end]] == value:
            end += 1
        average_rank = ((start + 1) + end) / 2.0
        for name in ordered[start:end]:
            ranks[name] = average_rank
        start = end
    return ranks


def _pairwise_accuracy(
    frontier_capacity: dict[str, float | None], real_capacity: dict[str, float | None]
) -> dict[str, Any]:
    names = sorted(set(frontier_capacity) & set(real_capacity))
    comparable = 0
    correct = 0
    frontier_ties = 0
    real_ties = 0
    for left_index, left in enumerate(names):
        for right in names[left_index + 1 :]:
            f_left = frontier_capacity[left]
            f_right = frontier_capacity[right]
            r_left = real_capacity[left]
            r_right = real_capacity[right]
            if None in (f_left, f_right, r_left, r_right):
                continue
            f_sign = (f_left > f_right) - (f_left < f_right)
            r_sign = (r_left > r_right) - (r_left < r_right)
            frontier_ties += int(f_sign == 0)
            real_ties += int(r_sign == 0)
            if f_sign == 0 or r_sign == 0:
                continue
            comparable += 1
            correct += int(f_sign == r_sign)
    return {
        "comparable_non_tied_pairs": comparable,
        "correct_pairs": correct,
        "accuracy": correct / comparable if comparable else None,
        "frontier_tied_pairs": frontier_ties,
        "real_tied_pairs": real_ties,
    }


def _objective_fidelity(
    frontier_ranking: list[dict[str, Any]],
    real_ranking: list[dict[str, Any]],
    *,
    field: str,
    only_names: set[str] | None = None,
) -> dict[str, Any]:
    frontier_capacity = _capacity_map(frontier_ranking, field=field)
    real_capacity = _capacity_map(real_ranking, field=field)
    if only_names is not None:
        frontier_capacity = {
            name: value for name, value in frontier_capacity.items() if name in only_names
        }
        real_capacity = {
            name: value for name, value in real_capacity.items() if name in only_names
        }
    names = sorted(set(frontier_capacity) & set(real_capacity))
    frontier_valid = [value for value in frontier_capacity.values() if value is not None]
    real_valid = [value for value in real_capacity.values() if value is not None]
    if not frontier_valid or not real_valid:
        return {
            "rankable": False,
            "reason": "one_or_both_surfaces_have_no_feasible_config",
            "frontier_capacity": frontier_capacity,
            "real_capacity": real_capacity,
        }
    frontier_average_ranks = _average_ranks(frontier_capacity)
    real_average_ranks = _average_ranks(real_capacity)
    best_frontier = max(frontier_valid)
    best_real = max(real_valid)
    frontier_top_set = [
        name for name, value in frontier_capacity.items() if value == best_frontier
    ]
    real_top_set = [name for name, value in real_capacity.items() if value == best_real]
    selected_real_capacities = [
        real_capacity[name]
        for name in frontier_top_set
        if real_capacity.get(name) is not None
    ]
    regret_values = [
        (best_real - value) / best_real
        for value in selected_real_capacities
        if best_real != 0
    ]
    return {
        "rankable": True,
        "capacity_field": field,
        "frontier_top1_set": frontier_top_set,
        "real_top1_set": real_top_set,
        "top1_set_intersection": sorted(set(frontier_top_set) & set(real_top_set)),
        "top1_set_match": set(frontier_top_set) == set(real_top_set),
        "top1_regret_fraction_best_tie_break": min(regret_values)
        if regret_values
        else None,
        "top1_regret_fraction_worst_tie_break": max(regret_values)
        if regret_values
        else None,
        "spearman_rank_correlation": _pearson(
            [frontier_average_ranks[name] for name in names],
            [real_average_ranks[name] for name in names],
        ),
        "pairwise_ordering": _pairwise_accuracy(frontier_capacity, real_capacity),
        "frontier_capacity": frontier_capacity,
        "real_capacity": real_capacity,
    }


def _request_residuals(
    frontier_results: list[dict[str, Any]], real_results: list[dict[str, Any]]
) -> dict[str, Any]:
    def index(results: list[dict[str, Any]]) -> dict[tuple[str, float, str], dict[str, Any]]:
        indexed = {}
        for config in results:
            name = config["config"]["name"]
            for load in config["loads"]:
                rate = float(load["offered_request_rate"])
                for request in load["requests"]:
                    indexed[(name, rate, str(request["request_id"]))] = request
        return indexed

    frontier_index = index(frontier_results)
    real_index = index(real_results)
    keys = sorted(set(frontier_index) & set(real_index))
    errors = []
    absolute_percentage_errors = []
    by_bin: dict[str, list[float]] = {}
    for key in keys:
        frontier_request = frontier_index[key]
        real_request = real_index[key]
        if not real_request["success"] or real_request.get("ttft_ms") is None:
            continue
        predicted = float(frontier_request["ttft_ms"])
        observed = float(real_request["ttft_ms"])
        error = predicted - observed
        errors.append(error)
        if observed > 0:
            absolute_percentage_errors.append(abs(error) / observed)
        input_tokens = int(real_request["input_tokens"])
        bin_index = _input_bin(input_tokens)
        label = f"[{INPUT_BINS[bin_index]},{INPUT_BINS[bin_index + 1]})"
        by_bin.setdefault(label, []).append(error)
    return {
        "matched_successful_requests": len(errors),
        "signed_error_ms_mean": statistics.fmean(errors) if errors else None,
        "mae_ms": statistics.fmean(abs(value) for value in errors) if errors else None,
        "mape": (
            statistics.fmean(absolute_percentage_errors)
            if absolute_percentage_errors
            else None
        ),
        "by_input_bin": {
            label: {
                "count": len(values),
                "signed_error_ms_mean": statistics.fmean(values),
                "mae_ms": statistics.fmean(abs(value) for value in values),
            }
            for label, values in sorted(by_bin.items())
        },
    }


def _pass_rate_residuals(
    frontier_results: list[dict[str, Any]], real_results: list[dict[str, Any]]
) -> dict[str, Any]:
    def index(
        results: list[dict[str, Any]], slo_name: str
    ) -> dict[tuple[str, float], tuple[float, bool]]:
        return {
            (config["config"]["name"], float(load["offered_request_rate"])): (
                float(load["scores"][slo_name]["slo_pass_rate"]),
                bool(load["scores"][slo_name]["feasible"]),
            )
            for config in results
            for load in config["loads"]
        }

    variants = {}
    for slo_name in SLO_VARIANTS:
        frontier_index = index(frontier_results, slo_name)
        real_index = index(real_results, slo_name)
        keys = sorted(set(frontier_index) & set(real_index))
        errors = [frontier_index[key][0] - real_index[key][0] for key in keys]
        label_matches = [frontier_index[key][1] == real_index[key][1] for key in keys]
        false_positives = sum(
            frontier_index[key][1] and not real_index[key][1] for key in keys
        )
        false_negatives = sum(
            not frontier_index[key][1] and real_index[key][1] for key in keys
        )
        variants[slo_name] = {
            "matched_config_load_points": len(errors),
            "signed_error_mean": statistics.fmean(errors) if errors else None,
            "mae": statistics.fmean(abs(value) for value in errors) if errors else None,
            "max_absolute_error": max((abs(value) for value in errors), default=None),
            "feasibility_label_accuracy": (
                sum(label_matches) / len(label_matches) if label_matches else None
            ),
            "false_feasible_points": false_positives,
            "false_infeasible_points": false_negatives,
        }
    return variants


def _request_slo_classification(
    frontier_results: list[dict[str, Any]], real_results: list[dict[str, Any]]
) -> dict[str, Any]:
    def index(
        results: list[dict[str, Any]],
    ) -> dict[tuple[str, float, str], dict[str, Any]]:
        return {
            (
                config["config"]["name"],
                float(load["offered_request_rate"]),
                str(request["request_id"]),
            ): request
            for config in results
            for load in config["loads"]
            for request in load["requests"]
        }

    def request_passes(request: dict[str, Any], variant: dict[str, Any]) -> bool:
        return bool(
            request["success"]
            and request.get("ttft_ms") is not None
            and float(request["ttft_ms"])
            <= slo_threshold_ms(variant, int(request["input_tokens"]))
        )

    frontier_index = index(frontier_results)
    real_index = index(real_results)
    keys = sorted(set(frontier_index) & set(real_index))
    variants = {}
    for slo_name, variant in SLO_VARIANTS.items():
        by_load: dict[tuple[str, float], dict[str, Any]] = {}
        frontier_failures: set[tuple[str, float, str]] = set()
        real_failures: set[tuple[str, float, str]] = set()
        for key in keys:
            frontier_passes = request_passes(frontier_index[key], variant)
            real_passes = request_passes(real_index[key], variant)
            if not frontier_passes:
                frontier_failures.add(key)
            if not real_passes:
                real_failures.add(key)
            group = by_load.setdefault(
                key[:2],
                {
                    "matched_requests": 0,
                    "matching_labels": 0,
                    "false_slo_pass_request_ids": [],
                    "false_slo_fail_request_ids": [],
                },
            )
            group["matched_requests"] += 1
            group["matching_labels"] += int(frontier_passes == real_passes)
            if frontier_passes and not real_passes:
                group["false_slo_pass_request_ids"].append(key[2])
            elif not frontier_passes and real_passes:
                group["false_slo_fail_request_ids"].append(key[2])
        intersection = frontier_failures & real_failures
        union = frontier_failures | real_failures
        load_records = []
        for (config, rate), record in sorted(by_load.items()):
            matched = int(record["matched_requests"])
            load_records.append(
                {
                    "config": config,
                    "offered_request_rate": rate,
                    **record,
                    "label_accuracy": record["matching_labels"] / matched,
                }
            )
        matching_labels = sum(item["matching_labels"] for item in by_load.values())
        false_passes = sum(
            len(item["false_slo_pass_request_ids"]) for item in by_load.values()
        )
        false_fails = sum(
            len(item["false_slo_fail_request_ids"]) for item in by_load.values()
        )
        variants[slo_name] = {
            "matched_requests": len(keys),
            "matching_labels": matching_labels,
            "label_accuracy": matching_labels / len(keys) if keys else None,
            "false_slo_pass_requests": false_passes,
            "false_slo_fail_requests": false_fails,
            "frontier_failure_requests": len(frontier_failures),
            "real_failure_requests": len(real_failures),
            "failure_set_intersection": len(intersection),
            "failure_set_union": len(union),
            "failure_set_jaccard": len(intersection) / len(union) if union else 1.0,
            "by_config_load": load_records,
        }
    return variants


def compare(args: argparse.Namespace) -> Path:
    frontier = json.loads(args.frontier_freeze.read_text())
    community = json.loads(args.community_freeze.read_text())
    if frontier.get("schema") != FRONTIER_SCHEMA:
        raise ValueError("unexpected Frontier freeze")
    if community.get("schema") != COMMUNITY_SCHEMA or community.get("status") != "completed":
        raise ValueError("unexpected/incomplete community freeze")
    if frontier["protocol_manifest_sha256"] != community["protocol_manifest_sha256"]:
        raise ValueError("comparison inputs used different protocols")
    variants = {}
    for slo_name in SLO_VARIANTS:
        frontier_ranking = frontier["rankings"][slo_name]
        real_ranking = community["rankings"][slo_name]
        variants[slo_name] = {
            "per_gpu_efficiency": _objective_fidelity(
                frontier_ranking,
                real_ranking,
                field="maximum_tested_feasible_request_rate_per_gpu",
            ),
            "single_replica_raw_capacity": _objective_fidelity(
                frontier_ranking,
                real_ranking,
                field="maximum_tested_feasible_request_rate",
            ),
            "within_topology_raw_capacity": {
                f"tp{tp}": _objective_fidelity(
                    frontier_ranking,
                    real_ranking,
                    field="maximum_tested_feasible_request_rate",
                    only_names={
                        item["config"]["name"]
                        for item in frontier_ranking
                        if int(item["config"]["tp"]) == tp
                    },
                )
                for tp in (4, 8)
            },
            "frontier_ranking": frontier_ranking,
            "real_ranking": real_ranking,
        }
    result = {
        "schema": COMPARISON_SCHEMA,
        "created_unix_s": time.time(),
        "frontier_freeze": {
            "path": str(args.frontier_freeze.resolve()),
            "sha256": sha256(args.frontier_freeze),
        },
        "community_freeze": {
            "path": str(args.community_freeze.resolve()),
            "sha256": sha256(args.community_freeze),
        },
        "primary_slo": PRIMARY_SLO,
        "variants": variants,
        "request_level_ttft_residuals": _request_residuals(
            frontier["config_results"], community["config_results"]
        ),
        "pass_rate_surface_residuals": _pass_rate_residuals(
            frontier["config_results"], community["config_results"]
        ),
        "request_slo_classification": _request_slo_classification(
            frontier["config_results"], community["config_results"]
        ),
    }
    write_json(args.output, result)
    print(args.output)
    return args.output


def _index_loads(
    config_results: list[dict[str, Any]],
) -> dict[tuple[str, float], dict[str, Any]]:
    return {
        (config["config"]["name"], float(load["offered_request_rate"])): load
        for config in config_results
        for load in config["loads"]
    }


def _index_requests(
    config_results: list[dict[str, Any]],
) -> dict[tuple[str, float, str], dict[str, Any]]:
    return {
        (
            config["config"]["name"],
            float(load["offered_request_rate"]),
            str(request["request_id"]),
        ): request
        for config in config_results
        for load in config["loads"]
        for request in load["requests"]
    }


def _request_passes_slo(request: dict[str, Any], variant: dict[str, Any]) -> bool:
    return bool(
        request["success"]
        and request.get("ttft_ms") is not None
        and float(request["ttft_ms"])
        <= slo_threshold_ms(variant, int(request["input_tokens"]))
    )


def compare_trials(args: argparse.Namespace) -> Path:
    frontier = json.loads(args.frontier_freeze.read_text())
    first = json.loads(args.first_community_freeze.read_text())
    repeat = json.loads(args.repeat_community_freeze.read_text())
    if frontier.get("schema") != FRONTIER_SCHEMA:
        raise ValueError("unexpected Frontier freeze")
    for label, payload in (("first", first), ("repeat", repeat)):
        if payload.get("schema") != COMMUNITY_SCHEMA or payload.get("status") != "completed":
            raise ValueError(f"unexpected/incomplete {label} community freeze")

    frontier_loads = _index_loads(frontier["config_results"])
    first_loads = _index_loads(first["config_results"])
    repeat_loads = _index_loads(repeat["config_results"])
    shared_load_keys = sorted(set(frontier_loads) & set(first_loads) & set(repeat_loads))
    if not shared_load_keys:
        raise ValueError("trials have no shared config/load points")
    for key in shared_load_keys:
        trace_hashes = {
            index[key].get("trace_sha256")
            for index in (frontier_loads, first_loads, repeat_loads)
            if index[key].get("trace_sha256") is not None
        }
        if len(trace_hashes) > 1:
            raise ValueError(f"trace mismatch at {key}")

    frontier_requests = _index_requests(frontier["config_results"])
    first_requests = _index_requests(first["config_results"])
    repeat_requests = _index_requests(repeat["config_results"])
    shared_request_keys = sorted(
        set(frontier_requests) & set(first_requests) & set(repeat_requests)
    )
    if not shared_request_keys:
        raise ValueError("trials have no shared requests")
    for key in shared_request_keys:
        lengths = {
            int(index[key]["input_tokens"])
            for index in (frontier_requests, first_requests, repeat_requests)
        }
        if len(lengths) != 1:
            raise ValueError(f"input-token mismatch at {key}")

    trial_ttft_differences = []
    for key in shared_request_keys:
        first_request = first_requests[key]
        repeat_request = repeat_requests[key]
        if (
            first_request["success"]
            and repeat_request["success"]
            and first_request.get("ttft_ms") is not None
            and repeat_request.get("ttft_ms") is not None
        ):
            trial_ttft_differences.append(
                float(repeat_request["ttft_ms"]) - float(first_request["ttft_ms"])
            )

    variants = {}
    for slo_name, variant in SLO_VARIANTS.items():
        request_counts = {
            "matching_real_trial_labels": 0,
            "frontier_correct_first": 0,
            "frontier_correct_repeat": 0,
            "frontier_wrong_both_stable_real_label": 0,
            "real_trial_label_flips": 0,
        }
        first_failures = set()
        repeat_failures = set()
        for key in shared_request_keys:
            predicted = _request_passes_slo(frontier_requests[key], variant)
            first_observed = _request_passes_slo(first_requests[key], variant)
            repeat_observed = _request_passes_slo(repeat_requests[key], variant)
            request_counts["matching_real_trial_labels"] += int(
                first_observed == repeat_observed
            )
            request_counts["frontier_correct_first"] += int(predicted == first_observed)
            request_counts["frontier_correct_repeat"] += int(predicted == repeat_observed)
            request_counts["frontier_wrong_both_stable_real_label"] += int(
                first_observed == repeat_observed and predicted != first_observed
            )
            request_counts["real_trial_label_flips"] += int(
                first_observed != repeat_observed
            )
            if not first_observed:
                first_failures.add(key)
            if not repeat_observed:
                repeat_failures.add(key)

        load_records = []
        matching_load_labels = 0
        frontier_correct_first_loads = 0
        frontier_correct_repeat_loads = 0
        for config_name, rate in shared_load_keys:
            predicted_score = frontier_loads[(config_name, rate)]["scores"][slo_name]
            first_score = first_loads[(config_name, rate)]["scores"][slo_name]
            repeat_score = repeat_loads[(config_name, rate)]["scores"][slo_name]
            predicted_label = bool(predicted_score["feasible"])
            first_label = bool(first_score["feasible"])
            repeat_label = bool(repeat_score["feasible"])
            matching_load_labels += int(first_label == repeat_label)
            frontier_correct_first_loads += int(predicted_label == first_label)
            frontier_correct_repeat_loads += int(predicted_label == repeat_label)
            load_records.append(
                {
                    "config": config_name,
                    "offered_request_rate": rate,
                    "frontier_passed_request_count": int(
                        predicted_score["passed_request_count"]
                    ),
                    "first_passed_request_count": int(first_score["passed_request_count"]),
                    "repeat_passed_request_count": int(
                        repeat_score["passed_request_count"]
                    ),
                    "frontier_feasible": predicted_label,
                    "first_feasible": first_label,
                    "repeat_feasible": repeat_label,
                }
            )
        failure_union = first_failures | repeat_failures
        variants[slo_name] = {
            "shared_requests": len(shared_request_keys),
            **request_counts,
            "real_trial_request_label_accuracy": (
                request_counts["matching_real_trial_labels"] / len(shared_request_keys)
            ),
            "frontier_request_label_accuracy_first": (
                request_counts["frontier_correct_first"] / len(shared_request_keys)
            ),
            "frontier_request_label_accuracy_repeat": (
                request_counts["frontier_correct_repeat"] / len(shared_request_keys)
            ),
            "real_trial_failure_set_jaccard": (
                len(first_failures & repeat_failures) / len(failure_union)
                if failure_union
                else 1.0
            ),
            "shared_config_load_points": len(shared_load_keys),
            "real_trial_load_label_accuracy": matching_load_labels / len(shared_load_keys),
            "frontier_load_label_accuracy_first": (
                frontier_correct_first_loads / len(shared_load_keys)
            ),
            "frontier_load_label_accuracy_repeat": (
                frontier_correct_repeat_loads / len(shared_load_keys)
            ),
            "config_load_records": load_records,
        }

    first_primary_capacity = _capacity_map(
        first["rankings"][PRIMARY_SLO],
        field="maximum_tested_feasible_request_rate",
    )
    first_config = {
        item["config"]["name"]: item["config"] for item in first["config_results"]
    }
    decision_records = []
    repeat_capacity_records = []
    repeated_config_names = {name for name, _ in repeat_loads}
    for config_name in sorted(set(first_primary_capacity) & repeated_config_names):
        old_rates = sorted(
            rate for name, rate in first_loads if name == config_name
        )
        capacity = first_primary_capacity[config_name]
        if capacity is None:
            continue
        higher_rates = [rate for rate in old_rates if rate > capacity]
        next_rate = min(higher_rates) if higher_rates else None
        boundary_rates = [capacity] + ([next_rate] if next_rate is not None else [])
        boundary_repeated = all(
            (config_name, rate) in repeat_loads for rate in boundary_rates
        )
        capacity_still_feasible = bool(
            boundary_repeated
            and repeat_loads[(config_name, capacity)]["scores"][PRIMARY_SLO]["feasible"]
        )
        next_still_infeasible = bool(
            next_rate is None
            or (
                boundary_repeated
                and not repeat_loads[(config_name, next_rate)]["scores"][PRIMARY_SLO][
                    "feasible"
                ]
            )
        )
        stable = boundary_repeated and capacity_still_feasible and next_still_infeasible
        decision_records.append(
            {
                "config": config_name,
                "first_capacity": capacity,
                "first_next_tested_rate": next_rate,
                "boundary_repeated": boundary_repeated,
                "capacity_still_feasible": capacity_still_feasible,
                "next_rate_still_infeasible": next_still_infeasible,
                "stable_original_grid_capacity": stable,
            }
        )
        repeat_capacity = capacity if stable else None
        repeat_capacity_records.append(
            {
                "config": first_config[config_name],
                "maximum_tested_feasible_request_rate": repeat_capacity,
                "maximum_tested_feasible_request_rate_per_gpu": (
                    repeat_capacity / int(first_config[config_name]["tp"])
                    if repeat_capacity is not None
                    else None
                ),
            }
        )

    result = {
        "schema": TRIAL_STABILITY_SCHEMA,
        "created_unix_s": time.time(),
        "inputs": {
            "frontier_freeze": {
                "path": str(args.frontier_freeze.resolve()),
                "sha256": sha256(args.frontier_freeze),
            },
            "first_community_freeze": {
                "path": str(args.first_community_freeze.resolve()),
                "sha256": sha256(args.first_community_freeze),
            },
            "repeat_community_freeze": {
                "path": str(args.repeat_community_freeze.resolve()),
                "sha256": sha256(args.repeat_community_freeze),
            },
        },
        "shared_config_load_points": len(shared_load_keys),
        "shared_requests": len(shared_request_keys),
        "real_trial_ttft_ms": {
            "paired_successful_requests": len(trial_ttft_differences),
            "repeat_minus_first_mean": (
                statistics.fmean(trial_ttft_differences)
                if trial_ttft_differences
                else None
            ),
            "mae": (
                statistics.fmean(abs(value) for value in trial_ttft_differences)
                if trial_ttft_differences
                else None
            ),
            "absolute_error_p95": _percentile(
                [abs(value) for value in trial_ttft_differences], 0.95
            ),
        },
        "variants": variants,
        "primary_original_grid_capacity_stability": {
            "stable_configs": sum(
                item["stable_original_grid_capacity"] for item in decision_records
            ),
            "total_configs": len(decision_records),
            "records": decision_records,
            "per_gpu_efficiency": _objective_fidelity(
                first["rankings"][PRIMARY_SLO],
                repeat_capacity_records,
                field="maximum_tested_feasible_request_rate_per_gpu",
            ),
            "single_replica_raw_capacity": _objective_fidelity(
                first["rankings"][PRIMARY_SLO],
                repeat_capacity_records,
                field="maximum_tested_feasible_request_rate",
            ),
        },
    }
    write_json(args.output, result)
    print(args.output)
    return args.output


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)

    prepare = subparsers.add_parser("prepare-protocol")
    prepare.add_argument("--trace", type=Path, required=True)
    prepare.add_argument("--expected-trace-sha256", default=frontier_grid.TRACE_SHA256)
    prepare.add_argument("--output-root", type=Path, required=True)
    prepare.add_argument("--cohort-size", type=int, default=COHORT_SIZE)
    prepare.add_argument("--seed", type=int, default=COHORT_SEED)
    prepare.add_argument("--rate", action="append", type=float, default=[])

    audit = subparsers.add_parser("audit-protocol")
    audit.add_argument("--manifest", type=Path, required=True)

    frontier = subparsers.add_parser("run-frontier")
    frontier.add_argument("--frontier-source", type=Path, required=True)
    frontier.add_argument(
        "--frontier-commit", default="d9cfeb6d8791fbf2f295dd9744c56a666171776e"
    )
    frontier.add_argument("--python", type=Path, required=True)
    frontier.add_argument("--profile-root", type=Path, required=True)
    frontier.add_argument("--protocol-manifest", type=Path, required=True)
    frontier.add_argument("--output-root", type=Path, required=True)
    frontier.add_argument("--tp4-capacity-artifact", type=Path, required=True)
    frontier.add_argument("--tp8-capacity-artifact", type=Path, required=True)
    frontier.add_argument("--ep-equivalence-artifact", type=Path, required=True)
    frontier.add_argument("--max-parallel-configs", type=int, default=1)
    frontier.add_argument("--config", action="append")

    community = subparsers.add_parser("prepare-community")
    community.add_argument("--repo", type=Path, required=True)
    community.add_argument("--python", type=Path, required=True)
    community.add_argument("--vllm", type=Path, required=True)
    community.add_argument("--model", type=Path, required=True)
    community.add_argument("--windows", type=Path, required=True)
    community.add_argument("--protocol-manifest", type=Path, required=True)
    community.add_argument("--frontier-freeze", type=Path, required=True)
    community.add_argument("--output-root", type=Path, required=True)
    community.add_argument("--port", type=int, default=18918)
    community.add_argument("--config-order-seed", type=int, default=CONFIG_ORDER_SEED)
    community.add_argument("--rate-order-seed", type=int, default=RATE_ORDER_SEED)

    run_real = subparsers.add_parser("run-community")
    run_real.add_argument("--manifest", type=Path, required=True)

    repeat = subparsers.add_parser("prepare-community-repeat")
    repeat.add_argument("--repo", type=Path, required=True)
    repeat.add_argument("--source-manifest", type=Path, required=True)
    repeat.add_argument("--community-freeze", type=Path, required=True)
    repeat.add_argument("--protocol-manifest", type=Path)
    repeat.add_argument("--frontier-freeze", type=Path)
    repeat.add_argument("--output-root", type=Path, required=True)
    repeat.add_argument("--slo", action="append", default=[])
    repeat.add_argument("--config", action="append", default=[])

    compare_parser = subparsers.add_parser("compare")
    compare_parser.add_argument("--frontier-freeze", type=Path, required=True)
    compare_parser.add_argument("--community-freeze", type=Path, required=True)
    compare_parser.add_argument("--output", type=Path, required=True)

    trials = subparsers.add_parser("compare-trials")
    trials.add_argument("--frontier-freeze", type=Path, required=True)
    trials.add_argument("--first-community-freeze", type=Path, required=True)
    trials.add_argument("--repeat-community-freeze", type=Path, required=True)
    trials.add_argument("--output", type=Path, required=True)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if args.command == "prepare-protocol":
        if not args.rate:
            args.rate = list(OFFERED_RATES)
        if sorted(set(args.rate)) != list(args.rate):
            raise ValueError("--rate values must be unique and ascending")
        prepare_protocol(args)
    elif args.command == "audit-protocol":
        print(json.dumps(audit_protocol(args.manifest), indent=2))
    elif args.command == "run-frontier":
        run_frontier(args)
    elif args.command == "prepare-community":
        prepare_community(args)
    elif args.command == "run-community":
        run_community(args)
    elif args.command == "prepare-community-repeat":
        prepare_community_repeat(args)
    elif args.command == "compare":
        compare(args)
    elif args.command == "compare-trials":
        compare_trials(args)
    else:
        raise AssertionError(args.command)


if __name__ == "__main__":
    main()