tests/test_cross_codec_report.py

from __future__ import annotations

import csv
import hashlib
import json
from pathlib import Path
from types import MappingProxyType
from typing import ClassVar

import numpy as np
import pytest
from PIL import Image

import cross_codec_report as report
from pixelmodel_robustness import cross_codec
from pixelmodel_robustness.cross_codec import CONDITION_IDS
from pixelmodel_robustness.cross_codec_experiment import _hash, scientific_identity_hash, validate_resource_monitor


def _artifact(root: Path, relative: str, data: bytes) -> dict[str, object]:
    path = root / relative
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    return {"path": relative, "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}


def _run(tmp_path: Path, *, status: str = "completed", mode: str = "real", prompt_count: int = 2, prompt_projection: str = "base") -> Path:
    run = tmp_path / "run"
    run.mkdir()
    source = np.zeros((4, 4, 3), dtype=np.uint8)
    Image.fromarray(source, "RGB").save(run / "source.png")
    _artifact(run, "source.png", (run / "source.png").read_bytes())
    prompts = [
        {"prompt_id": f"p-{i}", "category": f"cat-{i}", "prompt": f"prompt {i}", "seed": i, "initial_latent_seed": i + 10}
        for i in range(prompt_count)
    ]
    rows: list[dict[str, object]] = []
    for index, condition in enumerate(CONDITION_IDS):
        family = condition.replace("_repair_zero", "_raw").replace("_protected_high", "_raw")
        payload = _artifact(run, f"payloads/{family}.bin", bytes([CONDITION_IDS.index(family) + 1, 2, 3]))
        weight: dict[str, object] = {
            "row_type": "weight", "stage": "weight", "condition": condition, "run_identity_hash": "run-hash",
            "status": "completed", "failure_stage": None, "payload": payload, "protected": None,
            "payload_bytes": payload["bytes"], "payload_sha256": payload["sha256"], "source_bytes": 3,
            "payload_definition": "source payload; protected=high+low; materialized excluded",
            "nan_count": index % 2, "posinf_count": 0, "neginf_count": 0,
            "sign_mismatch_count": index, "exponent_mismatch_count": index * 2, "mantissa_mismatch_count": index * 3,
            "repaired_count": 1 if "repair" in condition else 0, "mse": float(index), "mae": float(index) / 2,
            "cosine": 1.0, "finite_pair_label": "all_elements", "observed": True,
        }
        if "protected_high" in condition:
            high = _artifact(run, f"payloads/{condition}-high.png", b"high")
            low = _artifact(run, f"payloads/{condition}-low.bin", b"low")
            weight["protected"] = {"high": high, "low": low, "bytes": 7, "sha256": hashlib.sha256(b"highlow").hexdigest()}
            weight["payload"] = {"high": high, "low": low, "bytes": 7, "sha256": hashlib.sha256(b"highlow").hexdigest()}
            weight["payload_bytes"] = 7
            weight["payload_sha256"] = hashlib.sha256(b"highlow").hexdigest()
        rows.append(weight)
        rows.append({"row_type": "layer", "stage": "layer", "condition": condition, "layer": "layer0", "prompt_id": None, "run_identity_hash": "run-hash", "finite": True, "mse": 0.0, "mae": 0.0, "cosine": 1.0})
        for prompt in prompts:
            success = not (condition == "jpeg_q80_raw" and prompt["prompt_id"] == "p-1")
            sample = None
            score = None
            if success:
                sample_path = f"samples/{condition}/{prompt['prompt_id']}.png"
                sample = _artifact(run, sample_path, b"sample")
                score = 0.5 + index / 100 + int(prompt["seed"]) / 1000
            rows.append({
                "row_type": "generation", "stage": "generation", "condition": condition,
                "prompt_id": prompt["prompt_id"], "category": prompt["category"], "prompt": prompt["prompt"],
                "seed": prompt["seed"], "initial_latent_seed": prompt["initial_latent_seed"],
                "run_identity_hash": "run-hash", "attempted": True, "completed": True,
                "success": success, "blocked": False, "failure_stage": None if success else "generation",
                "first_error": None if success else "CLIP failed", "clip_score": score,
                "sample": sample, "sample_path": sample["path"] if sample else None,
                "sample_hash": sample["sha256"] if sample else None, "sample_bytes": sample["bytes"] if sample else None,
            })
    raw = run / "results" / "raw_rows.jsonl"
    raw.parent.mkdir()
    raw.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows), encoding="utf-8")
    resource = run / "results" / "resource_monitor.jsonl"
    resource_rows = [{"row_type": "resource_guard", "stage": "preflight", "condition": None, "prompt_id": None, "run_identity_hash": "placeholder", "rss_bytes": 1, "system_available_percent": 90.0, "mps_allocated_bytes": None, "mps_driver_allocated_bytes": None, "violation_reason": None}]
    resource_rows += [{"row_type": "resource", "stage": "resource", "condition": row["condition"], "prompt_id": row["prompt_id"], "run_identity_hash": "placeholder", "rss_bytes": 1, "system_available_percent": 90.0, "mps_allocated_bytes": None, "mps_driver_allocated_bytes": None, "violation_reason": None} for row in rows if row["row_type"] == "generation" and row["attempted"] and not row["blocked"]]
    resource.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in resource_rows), encoding="utf-8")
    prompt_ids = [p["prompt_id"] for p in prompts]
    selected_prompts = prompts if prompt_projection == "base" else [{**prompt, "paired_seed": prompt["seed"]} for prompt in prompts]
    prompt_hash = _hash(selected_prompts)
    identity = {"source_sha256": hashlib.sha256((run / "source.png").read_bytes()).hexdigest(), "source_bytes": (run / "source.png").stat().st_size, "model_sha256": None, "config_sha256": "config", "prompt_sha256": "prompts", "upstream_revision": "upstream", "code_hashes": {}, "runtime_git": {}, "runtime_platform": {}, "package_versions": {}, "vae": {}, "clip": {}, "requested_device": "mps", "resolved_device": "mps", "mode": mode, "steps": 50, "cfg": 6.0, "profile": "pilot", "prompt_ids": prompt_ids, "prompt_selection_hash": prompt_hash, "condition_registry_hash": cross_codec.semantic_hash(cross_codec.CONDITION_REGISTRY), "codec_preflight": []}
    run_hash = _hash(identity)
    for row in rows: row["run_identity_hash"] = run_hash
    raw.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows), encoding="utf-8")
    resource_rows = [{**row, "run_identity_hash": run_hash} for row in resource_rows]
    resource.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in resource_rows), encoding="utf-8")
    resource_metadata = validate_resource_monitor(run, {"run_identity_hash": run_hash}, rows)
    manifest = {
        "status": status, "mode": mode, "profile": "pilot", "run_identity": identity,
        "run_identity_hash": run_hash, "scientific_identity_hash": scientific_identity_hash(identity), "raw_rows_sha256": hashlib.sha256(raw.read_bytes()).hexdigest(),
        "raw_rows_bytes": raw.stat().st_size, "source_sha256": identity["source_sha256"], "source_bytes": identity["source_bytes"],
        "condition_registry_hash": cross_codec.semantic_hash(cross_codec.CONDITION_REGISTRY), "conditions": list(CONDITION_IDS), "prompt_ids": prompt_ids, "layer_ids": ["layer0"],
        "bootstrap": {"seed": 20260727, "resamples": 10000}, "prompt_selection_hash": prompt_hash, "resource_monitor": {**resource_metadata, "binding_method": "run_completion"},
    }
    (run / "manifest.json").write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
    return run


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


def test_condition_registry_is_exactly_ordered() -> None:
    assert len(report.CONDITION_IDS) == 17
    assert report.CONDITION_IDS == CONDITION_IDS


def test_bootstrap_is_sorted_and_known_fixture() -> None:
    pairs = [("p-2", 2.0), ("p-0", 0.0), ("p-1", 1.0)]
    assert report.paired_bootstrap(pairs, seed=1, samples=1000) == report.paired_bootstrap(list(reversed(pairs)), seed=1, samples=1000)
    result = report.paired_bootstrap(pairs, seed=1, samples=1000)
    assert result["n"] == 3 and result["mean_delta"] == 1.0 and result["ci_low"] <= 1.0 <= result["ci_high"]


def test_bootstrap_empty_is_na() -> None:
    assert report.paired_bootstrap([], seed=1, samples=10)["n"] == 0
    assert report.paired_bootstrap([], seed=1, samples=10)["ci_low"] is None


def test_generate_exact_csv_schemas_and_counts(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    assert {p.name for p in output.iterdir()} >= {"codec_conditions.csv", "weight_metrics.csv", "generation_metrics.csv", "storage_metrics.csv", "summary.md", "final_conclusion.md", "index.html", "report_manifest.json"}
    assert len(_csv(output / "codec_conditions.csv")) == len(_csv(output / "weight_metrics.csv")) == len(_csv(output / "storage_metrics.csv")) == 17
    assert len(_csv(output / "generation_metrics.csv")) == 34
    assert list(_csv(output / "codec_conditions.csv")[0]) == list(report.CONDITION_COLUMNS)


def test_failed_clip_is_na_not_zero(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    failed = next(row for row in _csv(output / "generation_metrics.csv") if row["condition_id"] == "jpeg_q80_raw" and row["prompt_id"] == "p-1")
    assert failed["clip_score"] in {"", "N/A"} and failed["success"] == "0"


def test_condition_aggregate_has_fixed_denominator_and_failure_stage(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    summary = _csv(output / "condition_summary.csv")
    row = next(item for item in summary if item["condition_id"] == "jpeg_q80_raw")
    assert row["attempted"] == "2" and row["completed"] == "2" and row["denominator"] == "2" and row["first_failure_stage"] == "generation"


def test_payload_protected_raw_repair_share_and_materialized_exclusion(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    rows = _csv(output / "storage_metrics.csv")
    assert all(row["payload_definition"].startswith("encoded payload") for row in rows)
    assert all(row["payload_bytes"] != "" for row in rows)
    assert report.validate_protected_payload_rows(rows)


def test_grid_has_thirteen_columns_and_q80_grid(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    assert report.GRID_COLUMNS == ("PNG", "JPEG Q100 Raw", "JPEG Q100 Repair", "JPEG Q100 Protected", "WebP Q80 Raw", "WebP Q80 Repair", "WebP Q80 Protected", "AVIF Q70 Raw", "AVIF Q70 Repair", "AVIF Q70 Protected", "JXL d1 Raw", "JXL d1 Repair", "JXL d1 Protected")
    for name in ("generation_comparison_grid.png", "generation_comparison_grid_jpeg_q80.png"):
        assert (output / "plots" / name).is_file()


def test_all_plots_are_decodable_and_nonempty(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    for path in (output / "plots").glob("*.png"):
        with Image.open(path) as image:
            assert image.width >= 400 and image.height >= 200


def test_identity_binds_required_inputs(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    manifest = json.loads((output / "report_manifest.json").read_text())
    for key in ("run_identity_hash", "raw_rows_sha256", "report_code_hash", "condition_registry_hash", "profile", "prompt_selection_hash"):
        assert key in manifest["report_identity"]


def test_same_identity_is_deterministic_except_generated_at(tmp_path: Path) -> None:
    run = _run(tmp_path)
    first = report.generate_report(run, fixture=True)
    before = {str(p.relative_to(first)): p.read_bytes() for p in first.rglob("*") if p.is_file() and p.name != "report_manifest.json"}
    second = report.generate_report(run, fixture=True)
    after = {str(p.relative_to(second)): p.read_bytes() for p in second.rglob("*") if p.is_file() and p.name != "report_manifest.json"}
    assert before == after


def test_identity_mismatch_rejected_without_force(tmp_path: Path) -> None:
    run = _run(tmp_path)
    report.generate_report(run, fixture=True)
    value = json.loads((run / "manifest.json").read_text())
    value["prompt_selection_hash"] = "changed"
    (run / "manifest.json").write_text(json.dumps(value), encoding="utf-8")
    with pytest.raises(ValueError, match="identity|hash"):
        report.generate_report(run, fixture=True)


def test_force_report_allows_replacement(tmp_path: Path) -> None:
    run = _run(tmp_path)
    report.generate_report(run, fixture=True, report_options={"version": "A"})
    assert report.generate_report(run, fixture=True, report_options={"version": "B"}, force_report=True).is_dir()


def test_report_manifest_contains_relative_rehashed_artifacts(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    manifest = json.loads((output / "report_manifest.json").read_text())
    for artifact in manifest["artifacts"]:
        path = output / artifact["path"]
        assert not path.is_absolute() or str(path).startswith(str(output))
        assert path.stat().st_size == artifact["bytes"]
        assert hashlib.sha256(path.read_bytes()).hexdigest() == artifact["sha256"]


def test_raw_tamper_rejected(tmp_path: Path) -> None:
    run = _run(tmp_path)
    (run / "results" / "raw_rows.jsonl").write_text("tampered\n", encoding="utf-8")
    with pytest.raises(ValueError, match="raw"):
        report.generate_report(run, fixture=True)


def test_completed_pipeline_failure_is_rejected(tmp_path: Path) -> None:
    run = _run(tmp_path, status="completed_with_pipeline_failures")
    with pytest.raises(ValueError, match="pipeline|scientific"):
        report.generate_report(run, fixture=True)


def test_fixture_mode_is_not_scientific_conclusion(tmp_path: Path) -> None:
    run = _run(tmp_path, mode="fixture")
    output = report.generate_report(run, fixture=True)
    text = (output / "final_conclusion.md").read_text()
    assert "fixture" in text.lower() and "not final" in text.lower()


def test_summary_and_conclusion_have_observations_questions_and_fid_na(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    summary = (output / "summary.md").read_text()
    conclusion = (output / "final_conclusion.md").read_text()
    assert "Observations" in summary and "Conclusions" not in summary
    assert "Questions" in conclusion and "FID" in conclusion and "N/A" in conclusion
    assert all(f"Question {i}" in conclusion for i in range(1, 7))


def test_weight_map_helper_uses_injected_decoder() -> None:
    values = np.array([0, 1, 2, 3], dtype=np.uint8)
    decoded = report.compute_weight_error_map(values, values + 1, decoder=lambda payload: payload)
    assert decoded.shape == (2, 2) and float(decoded.mean()) == 1.0


def test_missing_sample_is_annotated_not_synthesized(tmp_path: Path) -> None:
    run = _run(tmp_path)
    sample = run / "samples/png_baseline/p-0.png"
    sample.unlink()
    with pytest.raises(ValueError, match="sample|artifact|condition_complete"):
        report.generate_report(run, fixture=True)


def test_cli_parser_has_required_options() -> None:
    parser = report.build_parser()
    args = parser.parse_args(["--run-dir", "run", "--force-report"])
    assert args.run_dir == "run" and args.force_report is True and args.output is None


def test_paired_quality_csv_has_exact_order_and_png_zero(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    rows = _csv(output / "paired_quality_metrics.csv")
    assert [row["condition_id"] for row in rows] == list(CONDITION_IDS)
    assert list(rows[0]) == list(report.PAIRED_COLUMNS)
    assert rows[0]["paired_n"] == "2" and rows[0]["mean_delta"] == "0.0" and rows[0]["ci_low"] == "0.0" and rows[0]["ci_high"] == "0.0"


def test_required_artifact_allowlist_and_public_manifest_verifier(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    assert {item["path"] for item in json.loads((output / "report_manifest.json").read_text())["artifacts"]} == report.REQUIRED_ARTIFACTS
    assert report.verify_report_manifest(output)


def test_report_records_and_rehashes_source_and_resource_provenance(tmp_path: Path) -> None:
    run = _run(tmp_path)
    output = report.generate_report(run, fixture=True)
    manifest = json.loads((output / "report_manifest.json").read_text())
    provenance = manifest["input_provenance"]
    assert provenance["source_sha256"] == json.loads((run / "manifest.json").read_text())["source_sha256"]
    assert provenance["source_bytes"] == json.loads((run / "manifest.json").read_text())["source_bytes"]
    assert "source_path" not in provenance and "source_path" not in manifest["report_identity"]
    assert provenance["baseline_source_path"] == "payloads/png_baseline.bin"
    assert provenance["baseline_source_sha256"] and provenance["baseline_source_bytes"] == 3
    assert provenance["resource_monitor_path"] == "results/resource_monitor.jsonl"
    assert provenance["resource_monitor_sha256"] and provenance["resource_monitor_bytes"] > 0
    assert report.verify_report_manifest(output)
    (run / provenance["baseline_source_path"]).write_bytes(b"tampered")
    assert report.verify_report_manifest(output) is False


def test_storage_ratios_use_baseline_payload_bytes_not_scientific_source_bytes(tmp_path: Path) -> None:
    run = _run(tmp_path)
    manifest_path = run / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["source_bytes"] = 10
    manifest_path.write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
    rows = [json.loads(line) for line in (run / "results" / "raw_rows.jsonl").read_text().splitlines()]
    target = next(row for row in rows if row.get("row_type") == "weight" and row.get("condition") == "jpeg_q100_raw")
    payload = run / target["payload"]["path"]
    payload.write_bytes(b"123456")
    payload_hash = hashlib.sha256(payload.read_bytes()).hexdigest()
    for row in rows:
        if row.get("row_type") == "weight" and row.get("condition") in {"jpeg_q100_raw", "jpeg_q100_repair_zero"}:
            row["payload_bytes"] = 6
            row["payload_sha256"] = payload_hash
            row["payload"]["bytes"] = 6
            row["payload"]["sha256"] = payload_hash
    raw = run / "results" / "raw_rows.jsonl"
    raw.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows), encoding="utf-8")
    manifest["raw_rows_sha256"] = hashlib.sha256(raw.read_bytes()).hexdigest()
    manifest["raw_rows_bytes"] = raw.stat().st_size
    manifest_path.write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")
    output = report.generate_report(run, fixture=True)
    storage = {row["condition_id"]: row for row in _csv(output / "storage_metrics.csv")}
    assert storage["png_baseline"]["source_bytes"] == "3"
    assert storage["png_baseline"]["payload_over_png"] == "1.0"
    assert storage["jpeg_q100_raw"]["payload_over_png"] == "2.0"
    assert storage["jpeg_q100_raw"]["png_over_payload"] == "0.5"
    assert storage["jpeg_q100_raw"]["size_saving_percent"] == "-100.0"


def test_report_rejects_run_dir_as_output(tmp_path: Path) -> None:
    run = _run(tmp_path)
    with pytest.raises(ValueError, match="run directory"):
        report.generate_report(run, run, fixture=True)


def test_task2_fixture_run_integrates_with_report(tmp_path: Path) -> None:
    from test_cross_codec_experiment import fixture_inputs, make_args, payload_factory, preflight, runtime_factory
    (tmp_path / "task2").mkdir(parents=True)
    inputs = fixture_inputs.__wrapped__(tmp_path / "task2")
    from pixelmodel_robustness.cross_codec_experiment import run
    run_dir = run(make_args(inputs, tmp_path / "task2-run"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    output = report.generate_report(run_dir, fixture=True)
    assert len(_csv(output / "generation_metrics.csv")) == 17 * 8


def test_actual_exponent_map_is_nonzero_and_plot_calls_helper() -> None:
    source = np.array([0x3C, 0x00, 0, 0], dtype=np.uint8)
    candidate = source.copy(); candidate[0] ^= 0x7C
    mapping = report.compute_exponent_density_map(source, candidate)
    assert float(np.nansum(mapping)) > 0.0


def _rewrite_raw(run: Path, mutator: object) -> None:
    raw = run / "results" / "raw_rows.jsonl"
    rows = [json.loads(line) for line in raw.read_text().splitlines()]
    mutator(rows)
    raw.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows), encoding="utf-8")
    manifest_path = run / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["raw_rows_sha256"] = hashlib.sha256(raw.read_bytes()).hexdigest()
    manifest["raw_rows_bytes"] = raw.stat().st_size
    resource_path = run / "results" / "resource_monitor.jsonl"
    resource_rows = [json.loads(line) for line in resource_path.read_text().splitlines()]
    expected = {(row["condition"], str(row["prompt_id"])) for row in rows if row.get("row_type") == "generation" and row.get("attempted") is True and not row.get("blocked")}
    resource_rows = [row for row in resource_rows if row.get("row_type") == "resource_guard" or (row.get("condition"), str(row.get("prompt_id"))) in expected]
    resource_path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in resource_rows), encoding="utf-8")
    manifest["resource_monitor"] = {**validate_resource_monitor(run, manifest, rows), "binding_method": "run_completion"}
    manifest_path.write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8")


@pytest.mark.parametrize("field", ["payload_bytes", "payload_sha256"])
def test_top_level_payload_metadata_mismatch_is_rejected(tmp_path: Path, field: str) -> None:
    run = _run(tmp_path)
    rows = [json.loads(line) for line in (run / "results" / "raw_rows.jsonl").read_text().splitlines()]
    weight = next(row for row in rows if row["row_type"] == "weight" and row["condition"] == "jpeg_q100_raw")
    weight[field] = weight[field] + 1 if field == "payload_bytes" else "bad"
    with pytest.raises(ValueError, match="payload metadata"):
        report._verify_payload_contract(run, rows)


def test_protected_nested_payload_mismatch_is_rejected(tmp_path: Path) -> None:
    run = _run(tmp_path)
    rows = [json.loads(line) for line in (run / "results" / "raw_rows.jsonl").read_text().splitlines()]
    weight = next(row for row in rows if row["row_type"] == "weight" and row["condition"] == "jpeg_q100_protected_high")
    weight["payload"]["bytes"] = 6
    with pytest.raises(ValueError, match="payload metadata|protected"):
        report._verify_payload_contract(run, rows)


def test_all_blocked_has_explicit_n_zero_na_text(tmp_path: Path) -> None:
    run = _run(tmp_path)
    def block(rows: list[dict[str, object]]) -> None:
        for row in rows:
            if row.get("row_type") == "generation":
                row.update(success=False, blocked=True, completed=True, clip_score=None, sample=None, sample_path=None, sample_hash=None, sample_bytes=None, failure_stage="blocked")
    _rewrite_raw(run, block)
    output = report.generate_report(run, fixture=True)
    paired = _csv(output / "paired_quality_metrics.csv")
    assert all(row["paired_n"] == "0" and row["mean_delta"] == "" and row["ci_low"] == "" and row["ci_high"] == "" and row["na_reason"] == "no successful finite prompt intersection" for row in paired)
    summary = (output / "summary.md").read_text()
    assert "N/A (no successful finite prompt intersection; N=0)" in summary and "None" not in summary
    assert "None" not in (output / "final_conclusion.md").read_text()


def test_lossless_sample_sha_rates_are_condition_specific(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    rows = _csv(output / "condition_summary.csv")
    by = {row["condition_id"]: row for row in rows}
    assert by["png_baseline"]["sample_sha_match_rate"] == "1.0"
    assert by["webp_lossless"]["sample_sha_match_rate"] == "1.0"
    assert all(by[condition]["sample_sha_match_rate"] == "N/A" for condition in CONDITION_IDS if condition not in {"png_baseline", "webp_lossless"})


def test_hypothesis_status_uses_observed_association_rule() -> None:
    summaries = [{"condition_id": name, "nan_count": 1, "success": 0} for name in ("jpeg_q100_raw", "jpeg_q80_raw", "webp_q80_raw", "avif_q70_raw", "jxl_d1_raw")]
    summaries += [{"condition_id": name, "success": 0} for name in ("jpeg_q100_repair_zero", "jpeg_q80_repair_zero", "webp_q80_repair_zero", "avif_q70_repair_zero", "jxl_d1_repair_zero")]
    summaries += [{"condition_id": name, "success": 1, "exponent_mismatch_count": 0} for name in ("jpeg_q100_protected_high", "jpeg_q80_protected_high", "webp_q80_protected_high", "avif_q70_protected_high", "jxl_d1_protected_high")]
    assert report.hypothesis_status(summaries) == "supported (5/5 associated lossy raw codecs)"


def test_json_normalize_handles_immutable_nested_preflight_metadata() -> None:
    value = MappingProxyType({"dims": (8, 8), "nested": MappingProxyType({"x": 1})})
    assert report._json_normalize(value) == {"dims": [8, 8], "nested": {"x": 1}}


def test_jxl_preflight_json_normalization_invokes_verified_decode(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    payload = _artifact(tmp_path, "payload.jxl", b"jxl")
    class Preflight:
        name = "jxl_d1"
        version = "1"
        metadata: ClassVar[dict[str, object]] = {"levels": (1, 2)}
    called: dict[str, object] = {}
    monkeypatch.setattr(report.cross_codec, "preflight_codecs", lambda: [Preflight()])
    monkeypatch.setattr(report.cross_codec, "decode_rgb", lambda data, spec, verified_preflight=None: called.update(data=data, verified=verified_preflight) or np.zeros((1, 1, 3), dtype=np.uint8))
    decoded, reason = report._decode_sample_payload(tmp_path, {"payload": payload}, "jxl_d1_raw", {"codec_preflight": [{"name": "jxl_d1", "version": "1", "metadata": {"levels": [1, 2]}}]}, fixture=False)
    assert decoded is not None and reason is None and called["verified"]


def test_jxl_preflight_changed_identity_rejects(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    payload = _artifact(tmp_path, "payload.jxl", b"jxl")
    class Preflight:
        name = "jxl_d1"
        version = "2"
        metadata: ClassVar[dict[str, object]] = {"levels": (1, 2)}
    monkeypatch.setattr(report.cross_codec, "preflight_codecs", lambda: [Preflight()])
    with pytest.raises(ValueError, match="JXL preflight identity mismatch"):
        report._decode_sample_payload(tmp_path, {"payload": payload}, "jxl_d1_raw", {"codec_preflight": [{"name": "jxl_d1", "version": "1", "metadata": {"levels": [1, 2]}}]}, fixture=False)


def test_pilot_conclusion_uses_concrete_codec_semantics(tmp_path: Path) -> None:
    summaries = []
    for condition in CONDITION_IDS:
        raw = condition.endswith("_raw")
        protected = condition.endswith("_protected_high")
        summaries.append({"condition_id": condition, "success": 0 if raw else 8 if protected or condition in {"png_baseline", "webp_lossless", "jpeg_q100_repair_zero"} else 0, "denominator": 8, "nan_count": 1 if raw else 0, "exponent_mismatch_count": 1 if raw else 0, "payload_over_png": 0.8 if protected else 1.0, "clip_mean": 0.9 if not raw else None, "sample_sha_match_rate": 1.0 if condition in {"png_baseline", "webp_lossless"} else "N/A"})
    paired = [{"condition_id": condition, "paired_n": 8, "mean_delta": 0.01, "ci_low": -0.01, "ci_high": 0.02} for condition in CONDITION_IDS]
    report._documents(tmp_path, {"profile": "pilot"}, summaries, paired, fixture=False)
    conclusion = (tmp_path / "final_conclusion.md").read_text()
    assert "Hypothesis status: supported (5/5 associated lossy raw codecs)" in conclusion
    assert "JPEG is not uniquely fragile in pilot" in conclusion
    assert "Repair Zero recovered only JPEG Q100 (1/5)" in conclusion
    assert "WebP Lossless" in conclusion and "WebP Q80 Protected" in conclusion
    assert "JXL" in conclusion and "Question 6" in conclusion
    assert "partially supported (1/5)" not in conclusion


def test_full_conclusion_uses_dynamic_scope_and_full_run_language(tmp_path: Path) -> None:
    summaries = []
    for condition in CONDITION_IDS:
        raw = condition.endswith("_raw")
        protected = condition.endswith("_protected_high")
        success = 100 if condition == "png_baseline" or protected or condition == "jpeg_q100_repair_zero" else 0
        summaries.append({
            "condition_id": condition,
            "success": success,
            "denominator": 100,
            "nan_count": 1 if raw else 0,
            "exponent_mismatch_count": 1 if raw else 0,
            "payload_over_png": 0.8 if protected else 1.0,
            "clip_mean": 0.9 if success else None,
            "sample_sha_match_rate": 1.0 if condition in {"png_baseline", "webp_lossless"} else "N/A",
        })
    paired = [{"condition_id": condition, "paired_n": 100, "mean_delta": 0.01, "ci_low": -0.01, "ci_high": 0.02} for condition in CONDITION_IDS]

    report._documents(tmp_path, {"profile": "full"}, summaries, paired, fixture=False)
    conclusion = (tmp_path / "final_conclusion.md").read_text()

    assert "JPEG is not uniquely fragile in full run" in conclusion
    assert "all had 0/100 blocked" in conclusion
    assert "other four 0/100" in conclusion
    assert "protected candidates maintained baseline success" in conclusion
    assert "Overall observed best" in conclusion
    assert "pilot" not in conclusion.lower()
    assert "limited to this pilot" not in conclusion


def test_report_options_bind_identity_and_force_replacement(tmp_path: Path) -> None:
    run = _run(tmp_path)
    output = report.generate_report(run, fixture=True, report_options={"label": "A"})
    first = json.loads((output / "report_manifest.json").read_text())["report_identity_hash"]
    with pytest.raises(ValueError, match="identity"):
        report.generate_report(run, fixture=True, report_options={"label": "B"})
    report.generate_report(run, fixture=True, report_options={"label": "B"}, force_report=True)
    second = json.loads((output / "report_manifest.json").read_text())["report_identity_hash"]
    assert first != second


def test_manifest_verifier_rejects_traversal_sha_tamper_and_duplicate(tmp_path: Path) -> None:
    output = report.generate_report(_run(tmp_path), fixture=True)
    manifest_path = output / "report_manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["artifacts"][0]["path"] = "../escape"
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    assert report.verify_report_manifest(output) is False
    (tmp_path / "again").mkdir()
    output = report.generate_report(_run(tmp_path / "again"), fixture=True)
    manifest_path = output / "report_manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["artifacts"][0]["sha256"] = "bad"
    manifest["artifacts"].append(dict(manifest["artifacts"][1]))
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    assert report.verify_report_manifest(output) is False


def test_same_run_report_manifest_is_byte_deterministic(tmp_path: Path) -> None:
    run = _run(tmp_path)
    output = report.generate_report(run, fixture=True)
    first = {str(path.relative_to(output)): path.read_bytes() for path in output.rglob("*") if path.is_file()}
    report.generate_report(run, fixture=True)
    second = {str(path.relative_to(output)): path.read_bytes() for path in output.rglob("*") if path.is_file()}
    assert first == second


def test_real_runner_paired_seed_hash_projection_is_resolved(tmp_path: Path) -> None:
    run = _run(tmp_path, prompt_projection="paired")
    output = report.generate_report(run, fixture=True)
    manifest = json.loads((output / "report_manifest.json").read_text())
    assert manifest["report_identity"]["prompt_projection"] == "paired_seed_equals_seed"
    assert manifest["input_provenance"]["prompt_projection"] == "paired_seed_equals_seed"
    assert [row["prompt_id"] for row in _csv(output / "generation_metrics.csv")[:2]] == ["p-0", "p-1"]


def test_prompt_projection_rejects_unmatched_hash(tmp_path: Path) -> None:
    prompts = [{"prompt_id": "p-0", "category": "cat", "prompt": "x", "seed": 1, "initial_latent_seed": 2}]
    with pytest.raises(ValueError, match="prompt selection hash"):
        report._resolve_prompt_projection(prompts, "not-a-candidate")


def test_prompt_projection_fails_closed_on_ambiguous_hash(monkeypatch: pytest.MonkeyPatch) -> None:
    prompts = [{"prompt_id": "p-0", "category": "cat", "prompt": "x", "seed": 1, "initial_latent_seed": 2}]
    monkeypatch.setattr(report, "_hash", lambda value: "same-hash")
    with pytest.raises(ValueError, match="ambiguous"):
        report._resolve_prompt_projection(prompts, "same-hash")