scripts/summarize_layer17_dynamic_gate.py

#!/usr/bin/env python3
"""Summarize validation-selected Layer-17 dynamic-gating test results."""

from __future__ import annotations

import argparse
import csv
import json
from pathlib import Path
from typing import Any

import matplotlib.pyplot as plt


TARGETS = (4, 6, 8, 10)


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


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    fields = list(rows[0])
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--root", type=Path,
        default=Path("outputs/layer17_dynamic_gate_20260830"),
    )
    args = parser.parse_args()
    root = args.root.resolve()
    test_dir = root / "test"
    summary_rows = read_csv(test_dir / "summary.csv")
    summary = {row["config_name"]: row for row in summary_rows}
    selected = json.loads((root / "validation" / "selected.json").read_text())
    selected_by_target = {
        int(row["target_accepts"]): row for row in selected["selected_dynamic"]
    }
    ffff_time = float(summary["ffff"]["generation_time_s"])

    comparisons: list[dict[str, Any]] = []
    acceptance_rows: list[dict[str, Any]] = []
    for target in TARGETS:
        selected_row = selected_by_target[target]
        dynamic_name = selected_row["config_name"]
        dynamic = summary[dynamic_name]
        static = summary[f"static_late_k{target:02d}"]
        dynamic_lpips = float(dynamic["tail_lpips"])
        static_lpips = float(static["tail_lpips"])
        comparisons.append(
            {
                "target_accepts": target,
                "dynamic_config": dynamic_name,
                "beta": float(dynamic["beta"]),
                "threshold": float(dynamic["threshold"]),
                "dynamic_actual_accepts": float(dynamic["accepted_predictor_calls"]),
                "static_actual_accepts": float(static["accepted_predictor_calls"]),
                "dynamic_full_calls": float(dynamic["full_calls"]),
                "static_full_calls": float(static["full_calls"]),
                "dynamic_tail_lpips": dynamic_lpips,
                "static_tail_lpips": static_lpips,
                "tail_lpips_reduction_percent": 100.0 * (static_lpips - dynamic_lpips) / static_lpips,
                "dynamic_generation_time_s": float(dynamic["generation_time_s"]),
                "static_generation_time_s": float(static["generation_time_s"]),
                "dynamic_overhead_vs_static_percent": 100.0 * (
                    float(dynamic["generation_time_s"])
                    / float(static["generation_time_s"])
                    - 1.0
                ),
                "dynamic_speedup_vs_ffff_percent": 100.0 * (
                    1.0 - float(dynamic["generation_time_s"]) / ffff_time
                ),
                "dynamic_lpips": float(dynamic["lpips"]),
                "static_lpips": float(static["lpips"]),
                "dynamic_latent_tail_nrmse": float(dynamic["latent_tail_nrmse"]),
                "static_latent_tail_nrmse": float(static["latent_tail_nrmse"]),
            }
        )

        decision_files = sorted((test_dir / "per_run" / dynamic_name).glob("*.json"))
        decisions = [
            decision
            for path in decision_files
            for decision in json.loads(path.read_text())["decisions"]
        ]
        for chunk in range(1, 7):
            for step in (1, 2):
                cell = [
                    row for row in decisions
                    if int(row["chunk"]) == chunk and int(row["step"]) == step
                ]
                acceptance_rows.append(
                    {
                        "target_accepts": target,
                        "dynamic_config": dynamic_name,
                        "chunk": chunk,
                        "step": step,
                        "acceptance_ratio": sum(bool(row["accepted"]) for row in cell)
                        / len(cell),
                    }
                )
    write_csv(test_dir / "dynamic_vs_static.csv", comparisons)
    write_csv(test_dir / "acceptance_by_chunk_step.csv", acceptance_rows)

    fig, axes = plt.subplots(1, 2, figsize=(11, 4.2))
    dynamic_rows = [summary[selected_by_target[target]["config_name"]] for target in TARGETS]
    static_rows = [summary[f"static_late_k{target:02d}"] for target in TARGETS]
    for axis, x_field, label in (
        (axes[0], "full_calls", "Mean Full calls"),
        (axes[1], "generation_time_s", "Generation time (s)"),
    ):
        axis.plot(
            [float(row[x_field]) for row in dynamic_rows],
            [float(row["tail_lpips"]) for row in dynamic_rows],
            "o-", label="Dynamic confidence", color="#d64b40", linewidth=2,
        )
        axis.plot(
            [float(row[x_field]) for row in static_rows],
            [float(row["tail_lpips"]) for row in static_rows],
            "s--", label="Static late-first", color="#3977b8", linewidth=2,
        )
        axis.scatter(
            [float(summary["ffff"][x_field])],
            [float(summary["ffff"]["tail_lpips"])],
            marker="*", s=100, color="#333333", label="FFFF",
        )
        axis.scatter(
            [float(summary["fppf"][x_field])],
            [float(summary["fppf"]["tail_lpips"])],
            marker="X", s=80, color="#777777", label="FPPF",
        )
        axis.set_xlabel(label)
        axis.set_ylabel("Tail LPIPS")
        axis.grid(alpha=0.25)
    axes[0].legend(frameon=False)
    fig.suptitle("Layer-17 Predictor: quality-compute frontier on prompts 90–99")
    fig.tight_layout()
    fig.savefig(root / "quality_compute_pareto.png", dpi=180)
    plt.close(fig)

    report = [
        "# Layer-17 dynamic confidence gating",
        "",
        "Thresholds and beta were selected only on prompts 80–89. The table below "
        "reports the frozen configurations on prompts 90–99.",
        "",
        "| Target P | Beta | Actual P | Full | Tail LPIPS dynamic | Static | Reduction | Gen speedup vs FFFF |",
        "|---:|---:|---:|---:|---:|---:|---:|---:|",
    ]
    for row in comparisons:
        report.append(
            f"| {row['target_accepts']} | {row['beta']:.1f} | "
            f"{row['dynamic_actual_accepts']:.1f} | {row['dynamic_full_calls']:.1f} | "
            f"{row['dynamic_tail_lpips']:.5f} | {row['static_tail_lpips']:.5f} | "
            f"{row['tail_lpips_reduction_percent']:.1f}% | "
            f"{row['dynamic_speedup_vs_ffff_percent']:.1f}% |"
        )
    report.extend(
        [
            "",
            f"FFFF generation time: {float(summary['ffff']['generation_time_s']):.3f}s. "
            f"FPPF generation time: {float(summary['fppf']['generation_time_s']):.3f}s; "
            f"tail LPIPS: {float(summary['fppf']['tail_lpips']):.5f}.",
            "",
            "Dynamic gating evaluates the Predictor at all 12 candidate decisions, "
            "including rejected calls. Its generation-time overhead relative to the "
            "budget-matched static policies is 0–4.4%, and is included in the table/plot.",
            "",
            "The K≈6 point is the recommended balanced operating point: beta=1.0, "
            "threshold=0.333097, 5.8 accepted Predictor calls, 22.2 Full calls, "
            "tail LPIPS 0.03449, and 16.3% generation speedup versus FFFF.",
        ]
    )
    (root / "REPORT.md").write_text("\n".join(report) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()