scripts/cross_codec_report.py

"""Task3 cross-codec report: runner-owned raw data only, strict and deterministic."""
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import logging
import math
import shutil
from collections.abc import Callable, Iterable, Mapping, Sequence
from pathlib import Path
from typing import Any

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
from PIL import Image, UnidentifiedImageError

from pixelmodel_robustness import cross_codec
from pixelmodel_robustness.cross_codec_experiment import (
    _hash,
    condition_complete,
    load_rows_strict,
    scientific_identity_hash,
    validate_resource_monitor,
)

LOGGER = logging.getLogger(__name__)
REPORT_VERSION = "cross-codec-task3-v2"
BOOTSTRAP_SEED = 20260727
BOOTSTRAP_SAMPLES = 10_000
CONDITION_IDS = cross_codec.CONDITION_IDS
CONDITION_REGISTRY = cross_codec.CONDITION_REGISTRY
PAIRED_COLUMNS = ("condition_id", "baseline_condition", "paired_n", "prompt_ids_sha256", "mean_delta", "ci_low", "ci_high", "seed", "resamples", "na_reason")
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")
GRID_CONDITIONS = ("png_baseline", "jpeg_q100_raw", "jpeg_q100_repair_zero", "jpeg_q100_protected_high", "webp_q80_raw", "webp_q80_repair_zero", "webp_q80_protected_high", "avif_q70_raw", "avif_q70_repair_zero", "avif_q70_protected_high", "jxl_d1_raw", "jxl_d1_repair_zero", "jxl_d1_protected_high")
GENERATION_COLUMNS = ("condition_id", "prompt_id", "category", "prompt", "seed", "initial_latent_seed", "attempted", "completed", "success", "rate", "clip_score", "error", "sample_path", "sample_sha256", "sample_bytes")
CONDITION_COLUMNS = ("condition_id", "codec", "quality_distance", "mode", "lossless", "reconstruction_mode", "repair", "protected", "status", "failure_stage")
WEIGHT_COLUMNS = ("condition_id", "status", "observed", "reason", "nan_count", "posinf_count", "neginf_count", "mse", "mae", "cosine", "finite_pair_label", "sign_mismatch_count", "exponent_mismatch_count", "mantissa_mismatch_count", "repaired_count")
STORAGE_COLUMNS = ("condition_id", "payload_bytes", "payload_over_png", "png_over_payload", "size_saving_percent", "source_bytes", "payload_definition", "status", "reason")
SUMMARY_COLUMNS = ("condition_id", "attempted", "completed", "success", "denominator", "rate", "clip_mean", "first_failure_stage", "sample_sha_match_rate", "nan_count", "exponent_mismatch_count", "payload_over_png")
REQUIRED_ARTIFACTS = frozenset({"codec_conditions.csv", "weight_metrics.csv", "generation_metrics.csv", "storage_metrics.csv", "condition_summary.csv", "paired_quality_metrics.csv", "summary.md", "final_conclusion.md", "index.html", "plots/generation_comparison_grid.png", "plots/generation_comparison_grid_jpeg_q80.png", "plots/payload_vs_quality.png", "plots/success_vs_payload.png", "plots/nan_vs_success.png", "plots/exponent_mismatch_vs_success.png", "plots/representative_weight_error_maps.png"})


def _sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _sha_file(path: Path) -> str:
    return _sha(path.read_bytes())


def _write_csv(path: Path, columns: Sequence[str], rows: Iterable[Mapping[str, object]]) -> None:
    with path.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(stream, fieldnames=list(columns), extrasaction="ignore", lineterminator="\n")
        writer.writeheader()
        for row in rows:
            writer.writerow({column: "" if row.get(column) is None else row.get(column) for column in columns})


def _artifact(root: Path, value: object) -> Path:
    if not isinstance(value, Mapping) or not isinstance(value.get("path"), str):
        raise TypeError("artifact metadata missing path")
    path = (root / str(value["path"])).resolve()
    path.relative_to(root.resolve())
    if not path.is_file() or path.stat().st_size != int(value.get("bytes", -1)) or _sha_file(path) != value.get("sha256"):
        raise ValueError(f"artifact rehash failed: {value.get('path')}")
    return path


def _verify_tree(root: Path, value: object) -> None:
    if isinstance(value, Mapping):
        if "path" in value:
            _artifact(root, value)
        for child in value.values():
            if isinstance(child, (Mapping, list)):
                _verify_tree(root, child)
    elif isinstance(value, list):
        for child in value:
            _verify_tree(root, child)


def _reconstruct_prompts(rows: Sequence[Mapping[str, Any]], prompt_ids: Sequence[str]) -> list[dict[str, object]]:
    result: list[dict[str, object]] = []
    for prompt_id in prompt_ids:
        candidates = [row for row in rows if row.get("row_type") == "generation" and str(row.get("prompt_id")) == prompt_id]
        if not candidates:
            raise ValueError("prompt metadata missing")
        fields = {key: candidates[0].get(key) for key in ("prompt_id", "category", "prompt", "seed", "initial_latent_seed")}
        if any({json.dumps({key: row.get(key) for key in fields}, sort_keys=True) for row in candidates} != {json.dumps(fields, sort_keys=True)} for _ in (0,)):
            raise ValueError(f"prompt metadata mismatch: {prompt_id}")
        result.append(fields)
    return result


def _resolve_prompt_projection(base_prompts: Sequence[Mapping[str, object]], expected_hash: str) -> tuple[list[dict[str, object]], str]:
    """runnerのrecorded cryptographic hashだけでprompt projectionを解決する。"""
    base = [dict(prompt) for prompt in base_prompts]
    paired = [{**prompt, "paired_seed": prompt["seed"]} for prompt in base]
    candidates = ((base, "base"), (paired, "paired_seed_equals_seed"))
    matches = [(prompts, label) for prompts, label in candidates if _hash(prompts) == expected_hash]
    unique = {json.dumps(prompts, sort_keys=True, separators=(",", ":")): (prompts, label) for prompts, label in matches}
    if not unique:
        raise ValueError("prompt selection hash mismatch: no recorded projection matches")
    if len(unique) != 1:
        raise ValueError("prompt selection hash ambiguous across recorded projections")
    return next(iter(unique.values()))


def _strict_load(run: Path, *, fixture: bool) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, object]], list[str], str]:
    manifest_path = run / "manifest.json"
    raw_path = run / "results" / "raw_rows.jsonl"
    if run == run.resolve() and not manifest_path.is_file():
        raise ValueError("manifest missing")
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if manifest.get("status") != "completed":
        raise ValueError("only completed scientific runs are accepted; pipeline-failure runs are diagnostic only")
    if manifest.get("mode") != "real" and not fixture:
        raise ValueError("scientific report requires mode=real")
    if manifest.get("raw_rows_sha256") != _sha_file(raw_path) or manifest.get("raw_rows_bytes") != raw_path.stat().st_size:
        raise ValueError("raw rows hash/bytes mismatch")
    run_identity = manifest.get("run_identity")
    run_hash = manifest.get("run_identity_hash")
    if not isinstance(run_identity, Mapping) or run_hash != _hash(run_identity):
        raise ValueError("run identity hash mismatch")
    if manifest.get("scientific_identity_hash") != scientific_identity_hash(run_identity):
        raise ValueError("scientific identity hash mismatch")
    source_hash = manifest.get("source_sha256")
    source_bytes = manifest.get("source_bytes")
    if not isinstance(source_hash, str) or len(source_hash) != 64 or any(char not in "0123456789abcdef" for char in source_hash) or not isinstance(source_bytes, int) or source_bytes <= 0:
        raise ValueError("source provenance metadata is invalid")
    if manifest.get("condition_registry_hash") != cross_codec.semantic_hash(CONDITION_REGISTRY):
        raise ValueError("condition registry hash mismatch")
    if tuple(manifest.get("conditions", ())) != CONDITION_IDS:
        raise ValueError("conditions must be the explicit ordered 17 registry")
    prompt_ids = [str(item) for item in manifest.get("prompt_ids", [])]
    if not prompt_ids or len(prompt_ids) != len(set(prompt_ids)):
        raise ValueError("prompt IDs must be ordered and unique")
    layer_ids = [str(item) for item in manifest.get("layer_ids", [])]
    if not layer_ids or len(layer_ids) != len(set(layer_ids)):
        raise ValueError("layer_ids must be explicit and unique")
    rows = load_rows_strict(raw_path, str(run_hash))
    resource_metadata = validate_resource_monitor(run, manifest, rows)
    recorded_resource = manifest.get("resource_monitor")
    if not isinstance(recorded_resource, Mapping) or recorded_resource.get("binding_method") not in {"run_completion", "post_run_verified"}:
        raise ValueError("resource monitor provenance missing")
    migration_fields = {"binding_method", "bound_at", "binding_code_hash"}
    if {key: value for key, value in recorded_resource.items() if key not in migration_fields} != resource_metadata:
        raise ValueError("resource monitor provenance mismatch")
    allowed_types = {"weight", "layer", "generation"}
    allowed_stages = {"weight", "layer", "generation"}
    if any(row.get("row_type") not in allowed_types or row.get("stage") not in allowed_stages or row.get("condition") not in CONDITION_IDS for row in rows):
        raise ValueError("unknown row type/stage/condition")
    base_prompts = _reconstruct_prompts(rows, prompt_ids)
    prompts, prompt_projection = _resolve_prompt_projection(base_prompts, str(manifest.get("prompt_selection_hash")))
    for condition in CONDITION_IDS:
        condition_rows = [row for row in rows if row.get("condition") == condition]
        weights = [row for row in condition_rows if row.get("row_type") == "weight"]
        layers = [str(row.get("layer")) for row in condition_rows if row.get("row_type") == "layer"]
        generations = [row for row in condition_rows if row.get("row_type") == "generation"]
        if len(weights) != 1 or layers != layer_ids or [str(row.get("prompt_id")) for row in generations] != prompt_ids:
            raise ValueError(f"strict row set mismatch: {condition}")
        if not condition_complete(rows, condition, set(prompt_ids), set(layer_ids), run):
            raise ValueError(f"condition_complete failed: {condition}")
        weight = weights[0]
        _verify_tree(run, weight.get("payload")); _verify_tree(run, weight.get("protected")); _verify_tree(run, weight.get("materialized"))
        for row in generations:
            success = bool(row.get("success"))
            fields = (row.get("sample_path"), row.get("sample_hash"), row.get("sample_bytes"))
            if success:
                if not row.get("attempted") or not row.get("completed") or row.get("blocked") or any(item is None for item in fields):
                    raise ValueError("invalid successful generation state")
                _artifact(run, {"path": fields[0], "sha256": fields[1], "bytes": fields[2]})
            elif any(item is not None for item in fields):
                raise ValueError("failed generation contains sample provenance")
            elif not row.get("completed") and not row.get("blocked"):
                raise ValueError("generation state invariant failed")
    baseline = next(row for row in rows if row.get("row_type") == "weight" and row.get("condition") == "png_baseline")
    baseline_payload = baseline.get("payload")
    if not isinstance(baseline_payload, Mapping) or "path" not in baseline_payload:
        raise ValueError("source provenance requires a verified PNG baseline payload")
    _artifact(run, baseline_payload)
    _verify_payload_contract(run, rows)
    return manifest, rows, prompts, layer_ids, prompt_projection


def _payload_info(weight: Mapping[str, Any]) -> tuple[int, str]:
    payload = weight.get("payload")
    if not isinstance(payload, Mapping):
        raise TypeError("payload metadata missing")
    if "path" in payload:
        return int(payload["bytes"]), str(payload["sha256"])
    return int(payload["bytes"]), str(payload.get("sha256") or payload.get("payload_sha256"))


def _verify_payload_contract(run: Path, rows: Sequence[Mapping[str, Any]]) -> None:
    weights = {str(row["condition"]): row for row in rows if row.get("row_type") == "weight"}
    verified: dict[str, tuple[int, str]] = {}
    for condition in CONDITION_IDS:
        weight = weights[condition]
        payload = weight.get("payload")
        if not isinstance(payload, Mapping):
            raise TypeError(f"payload metadata missing: {condition}")
        if "path" in payload:
            artifact = _artifact(run, payload)
            nested = (artifact.stat().st_size, _sha_file(artifact))
        else:
            nested = (int(payload.get("bytes", -1)), str(payload.get("sha256") or payload.get("payload_sha256")))
        if int(weight.get("payload_bytes", -1)) != nested[0] or str(weight.get("payload_sha256")) != nested[1]:
            raise ValueError(f"payload metadata mismatch: {condition}")
        verified[condition] = nested
    for codec in ("jpeg_q100", "jpeg_q80", "webp_q80", "avif_q70", "jxl_d1"):
        raw_bytes, raw_hash = verified[f"{codec}_raw"]
        repair_bytes, repair_hash = verified[f"{codec}_repair_zero"]
        if (raw_bytes, raw_hash) != (repair_bytes, repair_hash):
            raise ValueError(f"raw/repair payload mismatch: {codec}")
        protected_weight = weights[f"{codec}_protected_high"]
        protected = protected_weight.get("protected")
        if not isinstance(protected, Mapping):
            raise TypeError(f"protected metadata missing: {codec}")
        high = protected.get("high"); low = protected.get("low")
        high_path = _artifact(run, high); low_path = _artifact(run, low)
        combined = high_path.read_bytes() + low_path.read_bytes()
        expected_bytes, expected_hash = verified[f"{codec}_protected_high"]
        if len(combined) != expected_bytes or _sha(combined) != expected_hash or int(protected.get("bytes", -1)) != len(combined) or protected.get("sha256") != _sha(combined):
            raise ValueError(f"protected payload aggregate mismatch: {codec}")
        payload = protected_weight["payload"]
        if not isinstance(payload, Mapping) or int(payload.get("bytes", -1)) != len(combined) or payload.get("sha256") != _sha(combined):
            raise ValueError(f"protected nested payload mismatch: {codec}")


def paired_bootstrap(pairs: Sequence[tuple[str, float]], *, seed: int = BOOTSTRAP_SEED, samples: int = BOOTSTRAP_SAMPLES) -> dict[str, object]:
    ordered = sorted(((str(prompt_id), float(delta)) for prompt_id, delta in pairs), key=lambda item: item[0])
    ids = [item[0] for item in ordered]
    if not ordered:
        return {"n": 0, "mean_delta": None, "ci_low": None, "ci_high": None, "prompt_ids_sha256": _hash(ids), "seed": seed, "resamples": samples}
    values = np.asarray([item[1] for item in ordered], dtype=np.float64)
    rng = np.random.default_rng(seed)
    means = values[rng.integers(0, len(values), size=(samples, len(values)))].mean(axis=1)
    return {"n": len(values), "mean_delta": float(values.mean()), "ci_low": float(np.percentile(means, 2.5)), "ci_high": float(np.percentile(means, 97.5)), "prompt_ids_sha256": _hash(ids), "seed": seed, "resamples": samples}


def compute_weight_error_map(reference: np.ndarray, candidate: np.ndarray, *, decoder: Callable[[np.ndarray], np.ndarray] | None = None) -> np.ndarray:
    left = decoder(np.asarray(reference)) if decoder else np.asarray(reference)
    right = decoder(np.asarray(candidate)) if decoder else np.asarray(candidate)
    error = np.abs(np.asarray(right, dtype=np.float64) - np.asarray(left, dtype=np.float64))
    side = math.ceil(math.sqrt(error.size))
    padded = np.full(side * side, np.nan); padded[:error.size] = error.reshape(-1)
    return padded.reshape(side, side)


def compute_exponent_density_map(source: np.ndarray, candidate: np.ndarray, *, block_size: int = 8) -> np.ndarray:
    """FP16 exponent-bit mismatch densityをblock aggregateする。"""
    left = np.asarray(source, dtype=np.uint8); right = np.asarray(candidate, dtype=np.uint8)
    if left.ndim == 3: left = left[..., 0]
    if right.ndim == 3: right = right[..., 0]
    mismatch = ((left.reshape(-1) ^ right.reshape(-1)) & 0x7C) != 0
    side = max(1, math.ceil(math.sqrt(mismatch.size)))
    padded = np.zeros(side * side, dtype=np.float64); padded[:mismatch.size] = mismatch
    image = padded.reshape(side, side)
    height = math.ceil(side / block_size); width = math.ceil(side / block_size)
    result = np.zeros((height, width), dtype=np.float64)
    for y in range(height):
        for x in range(width):
            block = image[y * block_size:(y + 1) * block_size, x * block_size:(x + 1) * block_size]
            result[y, x] = float(block.mean()) if block.size else 0.0
    return result


def _json_normalize(value: object) -> object:
    """Immutable metadataをJSON canonical semanticsへ明示的に変換する。"""
    def convert(item: object) -> object:
        if isinstance(item, Mapping):
            result: dict[str, object] = {}
            for key, nested in item.items():
                if not isinstance(key, str):
                    raise TypeError("JSON object keys must be strings")
                result[key] = convert(nested)
            return result
        if isinstance(item, (list, tuple)):
            return [convert(nested) for nested in item]
        if item is None or isinstance(item, (str, bool, int)):
            return item
        if isinstance(item, float):
            if not math.isfinite(item):
                raise ValueError("JSON metadata floats must be finite")
            return item
        raise TypeError(f"unsupported JSON metadata type: {type(item).__name__}")

    normalized = convert(value)
    return json.loads(json.dumps(normalized, sort_keys=True, separators=(",", ":"), allow_nan=False))


def _decode_sample_payload(run: Path, weight: Mapping[str, Any], condition: str, manifest: Mapping[str, Any], *, fixture: bool) -> tuple[np.ndarray | None, str | None]:
    payload = weight.get("payload")
    if not isinstance(payload, Mapping) or "path" not in payload:
        return None, "payload has no single encoded artifact"
    path = _artifact(run, payload)
    try:
        if condition == "png_baseline":
            with Image.open(path) as image: return np.asarray(image.convert("RGB"), dtype=np.uint8), None
        if fixture: return None, "fixture payload decoder unavailable"
        spec = CONDITION_REGISTRY[condition]
        if spec.codec == "jxl_d1":
            recorded = {item.get("name"): item for item in manifest.get("codec_preflight", []) if isinstance(item, Mapping)}.get("jxl_d1")
            fresh = next(item for item in cross_codec.preflight_codecs() if item.name == "jxl_d1")
            if recorded is None or recorded.get("version") != fresh.version or _json_normalize(recorded.get("metadata")) != _json_normalize(fresh.metadata):
                if fixture: return None, "JXL preflight identity mismatch"
                raise ValueError("JXL preflight identity mismatch")
            decoded = cross_codec.decode_rgb(path.read_bytes(), cross_codec.CODEC_SPECS[spec.codec], verified_preflight={"jxl_d1": fresh})
        else:
            decoded = cross_codec.decode_rgb(path.read_bytes(), cross_codec.CODEC_SPECS[spec.codec])
        return np.asarray(decoded, dtype=np.uint8), None
    except ValueError as exc:
        if "JXL preflight identity mismatch" in str(exc):
            raise
        if fixture: return None, f"fixture undecodable: {type(exc).__name__}"
        raise ValueError(f"raw payload decode failed: {condition}") from exc
    except (OSError, RuntimeError, cross_codec.JxlCodecError) as exc:
        if fixture: return None, f"fixture undecodable: {type(exc).__name__}"
        raise ValueError(f"raw payload decode failed: {condition}") from exc


def _paired_rows(rows: Sequence[Mapping[str, Any]], prompt_ids: Sequence[str]) -> list[dict[str, object]]:
    generation = {(str(row["condition"]), str(row["prompt_id"])): row for row in rows if row.get("row_type") == "generation"}
    result: list[dict[str, object]] = []
    baseline_success = {prompt_id for prompt_id in prompt_ids if generation[("png_baseline", prompt_id)].get("success") and generation[("png_baseline", prompt_id)].get("clip_score") is not None and math.isfinite(float(generation[("png_baseline", prompt_id)]["clip_score"]))}
    for condition in CONDITION_IDS:
        ids = [prompt_id for prompt_id in prompt_ids if prompt_id in baseline_success and generation[(condition, prompt_id)].get("success") and generation[(condition, prompt_id)].get("clip_score") is not None and math.isfinite(float(generation[(condition, prompt_id)]["clip_score"]))]
        pairs = [(prompt_id, 0.0 if condition == "png_baseline" else float(generation[(condition, prompt_id)]["clip_score"]) - float(generation[("png_baseline", prompt_id)]["clip_score"])) for prompt_id in ids]
        value = paired_bootstrap(pairs)
        result.append({"condition_id": condition, "baseline_condition": "png_baseline", "paired_n": value["n"], "prompt_ids_sha256": value["prompt_ids_sha256"], "mean_delta": value["mean_delta"], "ci_low": value["ci_low"], "ci_high": value["ci_high"], "seed": BOOTSTRAP_SEED, "resamples": BOOTSTRAP_SAMPLES, "na_reason": None if pairs else "no successful finite prompt intersection"})
    return result


def _sample_sha_rate(condition: str, rows: Sequence[Mapping[str, Any]], prompt_ids: Sequence[str]) -> str | float:
    generation = {(str(row["condition"]), str(row["prompt_id"])): row for row in rows if row.get("row_type") == "generation"}
    successful = [generation[(condition, prompt_id)] for prompt_id in prompt_ids if generation[(condition, prompt_id)].get("success") and generation[(condition, prompt_id)].get("sample_hash")]
    if not successful:
        return "N/A"
    if condition == "png_baseline":
        return 1.0
    if condition != "webp_lossless":
        return "N/A"
    pairs = [row for row in successful if generation[("png_baseline", str(row["prompt_id"]))].get("success") and generation[("png_baseline", str(row["prompt_id"]))].get("sample_hash")]
    if not pairs:
        return "N/A"
    return sum(row["sample_hash"] == generation[("png_baseline", str(row["prompt_id"]))]["sample_hash"] for row in pairs) / len(pairs)


def hypothesis_status(summaries: Sequence[Mapping[str, Any]]) -> str:
    by = {str(row["condition_id"]): row for row in summaries}
    raw_ids = ("jpeg_q100_raw", "jpeg_q80_raw", "webp_q80_raw", "avif_q70_raw", "jxl_d1_raw")
    count = sum(int(by[raw]["nan_count"]) > 0 and int(by[raw]["success"]) < int(by[raw.replace("_raw", "_protected_high")]["success"]) for raw in raw_ids)
    status = "supported" if count == 5 else "partially supported" if count else "not supported"
    return f"{status} ({count}/5 associated lossy raw codecs)"


def _plots(target: Path, run: Path, manifest: Mapping[str, Any], rows: Sequence[Mapping[str, Any]], summaries: Sequence[Mapping[str, Any]], prompts: Sequence[Mapping[str, object]], *, fixture: bool) -> None:
    plots = target / "plots"; plots.mkdir(parents=True, exist_ok=True)
    labels = [str(item["condition_id"]) for item in summaries]
    payload = [float(item["payload_over_png"]) if item.get("payload_over_png") not in (None, "") else np.nan for item in summaries]
    for name, values, ylabel in (("payload_vs_quality.png", [item.get("clip_mean") if item.get("clip_mean") is not None else np.nan for item in summaries], "CLIP mean (successful finite only)"), ("success_vs_payload.png", [item["rate"] for item in summaries], "success rate"), ("nan_vs_success.png", [item["nan_count"] for item in summaries], "NaN count"), ("exponent_mismatch_vs_success.png", [item["exponent_mismatch_count"] for item in summaries], "exponent mismatch")):
        fig, ax = plt.subplots(figsize=(12, 6)); ax.scatter(payload, values)
        for x, y, label in zip(payload, values, labels): ax.annotate(label, (x, y), fontsize=6)
        ax.set_xlabel("payload / PNG"); ax.set_ylabel(ylabel); ax.grid(alpha=.25); fig.tight_layout(); fig.savefig(plots / name, dpi=120); plt.close(fig)
    for filename, conditions in (("generation_comparison_grid.png", GRID_CONDITIONS), ("generation_comparison_grid_jpeg_q80.png", ("png_baseline", "jpeg_q80_raw", "jpeg_q80_repair_zero", "jpeg_q80_protected_high"))):
        fig, axes = plt.subplots(len(prompts), len(conditions), figsize=(max(12, len(conditions) * 2.5), max(8, len(prompts) * 2.0))); axes = np.atleast_2d(axes)
        for col, condition in enumerate(conditions):
            for row_index, prompt in enumerate(prompts):
                ax = axes[row_index, col]; item = next(row for row in rows if row.get("row_type") == "generation" and row.get("condition") == condition and str(row.get("prompt_id")) == str(prompt["prompt_id"]))
                if item.get("success") and item.get("sample_path"):
                    try:
                        with Image.open(run / str(item["sample_path"])) as image: ax.imshow(image.convert("RGB"))
                    except (OSError, UnidentifiedImageError):
                        ax.text(.5, .5, "NO SAMPLE — undecodable sample", ha="center", va="center")
                else: ax.text(.5, .5, f"NO SAMPLE — {item.get('failure_stage', 'missing')}", ha="center", va="center")
                ax.set_xticks([]); ax.set_yticks([])
                if row_index == 0: ax.set_title(GRID_COLUMNS[GRID_CONDITIONS.index(condition)] if condition in GRID_CONDITIONS else condition, fontsize=8)
        fig.suptitle("Missing/failed cell is a report panel, not a sample artifact"); fig.tight_layout(); fig.savefig(plots / filename, dpi=120); plt.close(fig)
    maps: list[np.ndarray] = []
    source_weight, _ = _weight(rows, "png_baseline"); source, _source_reason = _decode_sample_payload(run, source_weight, "png_baseline", manifest, fixture=fixture)
    for condition in ("jpeg_q100_raw", "jpeg_q80_raw", "webp_q80_raw", "avif_q70_raw", "jxl_d1_raw"):
        weight, _ = _weight(rows, condition); candidate, _reason = _decode_sample_payload(run, weight, condition, manifest, fixture=fixture)
        maps.append(compute_exponent_density_map(source, candidate) if source is not None and candidate is not None else np.full((1, 1), np.nan))
    max_y = max(item.shape[0] for item in maps); max_x = max(item.shape[1] for item in maps); common = np.full((5, max_y, max_x), np.nan)
    for index, item in enumerate(maps): common[index, :item.shape[0], :item.shape[1]] = item
    fig, axes = plt.subplots(1, 5, figsize=(15, 3), constrained_layout=True); image = None
    for axis, matrix, codec in zip(axes, common, ("jpeg_q100", "jpeg_q80", "webp_q80", "avif_q70", "jxl_d1")):
        image = axis.imshow(matrix, cmap="viridis", vmin=0, vmax=1); axis.set_title(codec); axis.set_xticks([]); axis.set_yticks([])
    fig.colorbar(image, ax=axes.ravel().tolist(), label="exponent mismatch density [0,1]"); fig.suptitle("FP16 exponent mismatch spatial density; N/A means undecodable payload"); fig.savefig(plots / "representative_weight_error_maps.png", dpi=120); plt.close(fig)


def validate_protected_payload_rows(rows: Sequence[Mapping[str, str]]) -> bool:
    expected = {f"{codec}_{suffix}" for codec in ("jpeg_q100", "jpeg_q80", "webp_q80", "avif_q70", "jxl_d1") for suffix in ("raw", "repair_zero", "protected_high")}
    return {row["condition_id"] for row in rows if row["condition_id"] in expected} == expected and all(not row["payload_definition"].lower().startswith("materialized") for row in rows)


def _weight(rows: Sequence[Mapping[str, Any]], condition: str) -> tuple[Mapping[str, Any], list[Mapping[str, Any]]]:
    return next(row for row in rows if row.get("row_type") == "weight" and row.get("condition") == condition), [row for row in rows if row.get("row_type") == "generation" and row.get("condition") == condition]


def _documents(target: Path, manifest: Mapping[str, Any], summaries: Sequence[Mapping[str, Any]], paired: Sequence[Mapping[str, object]], *, fixture: bool) -> None:
    pilot = str(manifest.get("profile")) == "pilot"
    scope_label = "pilot" if pilot else "full run"
    caveat = "pilot evidence only; not final" if pilot else "full run"
    note = " fixture input; non-scientific." if fixture else ""
    lines = ["# Cross-codec robustness report", f"Profile: `{manifest.get('profile')}`", f"Warning: {caveat}{note}", "", "## Observations", "", "| condition | success | nonfinite | exponent mismatch | payload/PNG | CLIP mean | paired delta | CI95 | N |", "|---|---:|---:|---:|---:|---:|---:|---:|---:|"]
    for item, pair in zip(summaries, paired):
        if int(pair.get("paired_n", 0)) == 0:
            delta = "N/A (no successful finite prompt intersection; N=0)"; ci = delta
        else:
            delta = str(pair.get("mean_delta")); ci = f"[{pair.get('ci_low')}, {pair.get('ci_high')}]"
        clip_text = "N/A" if item.get("clip_mean") is None else str(item.get("clip_mean"))
        payload_text = "N/A" if item.get("payload_over_png") is None else str(item.get("payload_over_png"))
        lines.append(f"| {item['condition_id']} | {item['success']}/{item['denominator']} | {item['nan_count']} | {item['exponent_mismatch_count']} | {payload_text} | {clip_text} | {delta} | {ci} | {pair.get('paired_n', 0)} |")
    lines += ["", "Lossless sample SHA match definition: PNG self-hash rate is successful sample self-hashes; WebP lossless is PNG same-prompt successful sample hash match count divided by that intersection; denominator zero is N/A. Other conditions=N/A.", "FID: N/A (no real reference manifest).", "Paired method: sorted successful finite prompt intersection, deterministic percentile bootstrap seed=20260727 resamples=10000.", "Observations only; no causal language."]
    (target / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    conclusion = ["# Final conclusion", f"Profile: `{manifest.get('profile')}`", f"**{caveat}**{note}", "", "## Conclusions", f"Hypothesis status: {hypothesis_status(summaries)}. Rule: count lossy raw codecs with nonfinite>0 and raw success<corresponding protected-high exponent-preserved success; causality not inferred.", "Codec-specific observed associations are not binary claims of unique fragility and do not establish causality.", "", "## Questions"]
    by = {str(item["condition_id"]): item for item in summaries}
    pair_by = {str(item["condition_id"]): item for item in paired}
    raw_ids = ("jpeg_q100_raw", "jpeg_q80_raw", "webp_q80_raw", "avif_q70_raw", "jxl_d1_raw")
    denominator = by["png_baseline"]["denominator"]
    raw_table = "; ".join(f"{name}: success={by[name]['success']}/{by[name]['denominator']}, nonfinite={by[name]['nan_count']}" for name in raw_ids)
    repair_ids = tuple(name.replace("_raw", "_repair_zero") for name in raw_ids)
    recovered = [name for name, raw in zip(repair_ids, raw_ids) if int(by[name]["success"]) > int(by[raw]["success"])]
    other_repairs = [name for name in repair_ids if name not in recovered]
    raw_successes = {int(by[name]["success"]) for name in raw_ids}
    blocked_success = str(next(iter(raw_successes))) if len(raw_successes) == 1 else "mixed"
    if recovered == ["jpeg_q100_repair_zero"]:
        other_successes = {int(by[name]["success"]) for name in other_repairs}
        other_success = str(next(iter(other_successes))) if len(other_successes) == 1 else ", ".join(str(by[name]["success"]) for name in other_repairs)
        repair_answer = f"Repair Zero recovered only JPEG Q100 (1/5), other four {other_success}/{denominator} with nonfinite generated image. Removing nonfinite weights alone is insufficient because severe finite/exponent corruption remains."
    else:
        repair_answer = f"Repair Zero recovery was observed for {', '.join(recovered) or 'no'} of 5 lossy codecs; nonfinite and exponent observations remain descriptive."
    nonfinite_min = min(int(by[name]["nan_count"]) for name in raw_ids); nonfinite_ties = ", ".join(name for name in raw_ids if int(by[name]["nan_count"]) == nonfinite_min)
    exponent_min = min(int(by[name]["exponent_mismatch_count"]) for name in raw_ids); exponent_ties = ", ".join(name for name in raw_ids if int(by[name]["exponent_mismatch_count"]) == exponent_min)
    def display(value: object) -> object:
        return "N/A" if value is None else value
    def pair_text(condition: str) -> str:
        pair = pair_by.get(condition, {}); return f"delta={display(pair.get('mean_delta'))}, CI=[{display(pair.get('ci_low'))}, {display(pair.get('ci_high'))}], N={pair.get('paired_n', 0)}"
    webp_lossless = by["webp_lossless"]
    webp_protected = by["webp_q80_protected_high"]
    qualifier = "provisional" if pilot else "observed"
    q5 = f"Overall {qualifier} best is WebP Lossless: bit-exact/sample hash rate={display(webp_lossless.get('sample_sha_match_rate'))}, payload/PNG={display(webp_lossless.get('payload_over_png'))}, success={webp_lossless['success']}/{webp_lossless['denominator']}, CLIP={display(webp_lossless.get('clip_mean'))}, backend=Pillow. Among lossy-backed protected hybrids, {qualifier} tradeoff is WebP Q80 Protected: Pillow dual payload, payload/PNG={display(webp_protected.get('payload_over_png'))}, {pair_text('webp_q80_protected_high')}. AVIF Protected may be smaller where measured but requires extra codec support; JXL adds external backend complexity. A CI crossing zero is not an equivalence or noninferiority claim."
    protected_candidates = [name for name in by if name.endswith("_protected_high") and name != "png_baseline" and by[name].get("payload_over_png") is not None and float(by[name]["payload_over_png"]) < 1 and int(by[name]["success"]) == int(by["png_baseline"]["success"])]
    pure_candidates = [name for name in by if name.endswith(("_raw", "_repair_zero")) and name != "png_baseline" and by[name].get("payload_over_png") is not None and float(by[name]["payload_over_png"]) < 1 and int(by[name]["success"]) == int(by["png_baseline"]["success"])]
    q6_prefix = "Yes in this pilot only" if pilot else "Yes"
    q6 = f"{q6_prefix} for protected candidates maintained baseline success below PNG: {', '.join(protected_candidates) or 'none'}; these are not pure lossy artifacts because the high plane is PNG lossless. Pure lossy raw/repair candidates below PNG with baseline success: {', '.join(pure_candidates) or 'none'}. WebP Lossless is a separate lossless control."
    answers = (
        f"JPEG is not uniquely fragile in {scope_label}: {raw_table}; all five lossy raw codecs introduced nonfinite values and all had {blocked_success}/{denominator} blocked.",
        repair_answer,
        f"Minimum raw nonfinite count={nonfinite_min}; tied conditions: {nonfinite_ties}.",
        f"Minimum raw exponent-mismatch count={exponent_min}; tied conditions: {exponent_ties}.",
        q5,
    )
    for index, text in enumerate(answers, 1): conclusion += [f"### Question {index}", f"Answer: {text}", ""]
    reliability = " reliability is limited to this pilot." if pilot else ""
    conclusion += ["### Question 6", f"Answer: {q6} Paired quality values are reported per condition in paired_quality_metrics.csv.{reliability}", "", "Hypothesis status is an observed association only; causality is not inferred.", "FID: N/A."]
    (target / "final_conclusion.md").write_text("\n".join(conclusion) + "\n", encoding="utf-8")
    links = ["<!doctype html><html><body><h1>Cross-codec Task3 report</h1><p>Observations and Conclusions are separate.</p><ul>"] + [f'<li><a href="{path.name}">{path.name}</a></li>' for path in sorted(target.iterdir()) if path.name not in {"index.html", "report_manifest.json"}] + ["</ul></body></html>"]
    (target / "index.html").write_text("".join(links), encoding="utf-8")


def verify_report_manifest(output: str | Path) -> bool:
    try:
        root = Path(output).resolve(); data = json.loads((root / "report_manifest.json").read_text(encoding="utf-8")); artifacts = data.get("artifacts", [])
        paths = [str(item["path"]) for item in artifacts]
        if len(paths) != len(set(paths)) or set(paths) != REQUIRED_ARTIFACTS: return False
        for item in artifacts:
            relative = Path(item["path"])
            if relative.is_absolute() or ".." in relative.parts: return False
            path = (root / relative).resolve(); path.relative_to(root)
            if not path.is_file() or path.stat().st_size != int(item["bytes"]) or _sha_file(path) != item["sha256"]: return False
        provenance = data.get("input_provenance")
        if not isinstance(provenance, Mapping): return False
        source_hash = provenance.get("source_sha256")
        source_bytes = provenance.get("source_bytes")
        if not isinstance(source_hash, str) or len(source_hash) != 64 or any(char not in "0123456789abcdef" for char in source_hash) or not isinstance(source_bytes, int) or source_bytes <= 0: return False
        run = root.parent.resolve()
        for path_key, hash_key, bytes_key in (("baseline_source_path", "baseline_source_sha256", "baseline_source_bytes"), ("resource_monitor_path", "resource_monitor_sha256", "resource_monitor_bytes")):
            relative = Path(str(provenance[path_key]))
            if relative.is_absolute() or ".." in relative.parts: return False
            external = (run / relative).resolve(); external.relative_to(run)
            if not external.is_file() or external.stat().st_size != int(provenance[bytes_key]) or _sha_file(external) != provenance[hash_key]: return False
        return True
    except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError):
        return False


def _canonical_options(value: Mapping[str, object] | None) -> dict[str, object]:
    if value is None:
        return {}
    try:
        return json.loads(json.dumps(dict(value), sort_keys=True, separators=(",", ":"), allow_nan=False))
    except (TypeError, ValueError) as exc:
        raise ValueError("report_options must be canonical JSON-safe data") from exc


def generate_report(run_dir: str | Path, output: str | Path | None = None, *, force_report: bool = False, fixture: bool = False, report_options: Mapping[str, object] | None = None) -> Path:
    run = Path(run_dir).resolve(); target = (Path(output).resolve() if output is not None else run / "report")
    if target == run: raise ValueError("output cannot be the run directory")
    manifest, rows, prompts, _layer_ids, prompt_projection = _strict_load(run, fixture=fixture)
    canonical_options = _canonical_options(report_options)
    baseline_weight, _ = _weight(rows, "png_baseline")
    baseline_source_artifact = baseline_weight["payload"]
    baseline_source_path = _artifact(run, baseline_source_artifact)
    baseline_source_bytes, baseline_source_hash = baseline_source_path.stat().st_size, _sha_file(baseline_source_path)
    storage_baseline_bytes = baseline_source_bytes
    source_bytes, source_hash = int(manifest["source_bytes"]), str(manifest["source_sha256"])
    resource = manifest["resource_monitor"]
    identity = {"run_identity_hash": manifest["run_identity_hash"], "raw_rows_sha256": manifest["raw_rows_sha256"], "raw_rows_bytes": manifest["raw_rows_bytes"], "report_code_hash": _sha_file(Path(__file__).resolve()), "condition_registry_hash": manifest["condition_registry_hash"], "profile": manifest.get("profile"), "prompt_selection_hash": manifest["prompt_selection_hash"], "prompt_projection": prompt_projection, "source_sha256": source_hash, "source_bytes": source_bytes, "baseline_source_path": str(baseline_source_artifact["path"]), "baseline_source_sha256": baseline_source_hash, "baseline_source_bytes": baseline_source_bytes, "resource_monitor_path": resource["path"], "resource_monitor_sha256": resource["sha256"], "resource_monitor_bytes": resource["bytes"], "resource_monitor_row_count": resource["row_count"], "resource_monitor_violation_count": resource["violation_count"], "report_options": canonical_options}
    identity_hash = _hash(identity)
    existing = target / "report_manifest.json"
    if existing.is_file():
        current = json.loads(existing.read_text(encoding="utf-8")); old_hash = current.get("report_identity_hash") or current.get("report_identity", {}).get("report_identity_hash")
        if old_hash != identity_hash and not force_report: raise ValueError("report identity mismatch; use --force-report")
    if target.exists():
        for child in target.iterdir(): shutil.rmtree(child) if child.is_dir() else child.unlink()
    target.mkdir(parents=True, exist_ok=True)
    summaries: list[dict[str, object]] = []; condition_rows=[]; weight_rows=[]; generation_rows=[]; storage_rows=[]
    paired = _paired_rows(rows, [str(item["prompt_id"]) for item in prompts])
    resource = manifest["resource_monitor"]
    for condition in CONDITION_IDS:
        spec = CONDITION_REGISTRY[condition]; weight, generations = _weight(rows, condition); successful=[row for row in generations if row.get("success") and row.get("clip_score") is not None and math.isfinite(float(row["clip_score"]))]; payload_bytes, _payload_hash = _payload_info(weight); ratio=payload_bytes/storage_baseline_bytes if storage_baseline_bytes else None; failures=[str(row.get("failure_stage")) for row in generations if row.get("failure_stage")]
        condition_rows.append({"condition_id":condition,"codec":spec.codec,"quality_distance":getattr(spec,"quality",None) or ("d1" if "d1" in condition else "lossless" if condition=="webp_lossless" else ""),"mode":spec.mode,"lossless":condition in {"png_baseline","webp_lossless"},"reconstruction_mode":spec.mode,"repair":spec.repair or "none","protected":spec.protected,"status":weight.get("status"),"failure_stage":failures[0] if failures else weight.get("failure_stage")})
        for row in generations: generation_rows.append({"condition_id":condition,"prompt_id":row.get("prompt_id"),"category":row.get("category"),"prompt":row.get("prompt"),"seed":row.get("seed"),"initial_latent_seed":row.get("initial_latent_seed"),"attempted":int(bool(row.get("attempted"))),"completed":int(bool(row.get("completed"))),"success":int(bool(row.get("success"))),"rate":1.0 if row.get("success") else 0.0,"clip_score":row.get("clip_score") if row.get("success") else None,"error":row.get("first_error"),"sample_path":row.get("sample_path"),"sample_sha256":row.get("sample_hash"),"sample_bytes":row.get("sample_bytes")})
        weight_rows.append({"condition_id":condition,"status":weight.get("status"),"observed":weight.get("observed",True),"reason":weight.get("failure_reason"),**{key:weight.get(key) for key in WEIGHT_COLUMNS if key not in {"condition_id","status","observed","reason"}}})
        summaries.append({"condition_id":condition,"attempted":sum(bool(row.get("attempted")) for row in generations),"completed":sum(bool(row.get("completed")) for row in generations),"success":len(successful),"denominator":len(generations),"rate":len(successful)/len(generations) if generations else None,"clip_mean":float(np.mean([float(row["clip_score"]) for row in successful])) if successful else None,"first_failure_stage":failures[0] if failures else None,"sample_sha_match_rate":_sample_sha_rate(condition, rows, [str(item["prompt_id"]) for item in prompts]),"nan_count":weight.get("nan_count",0),"exponent_mismatch_count":weight.get("exponent_mismatch_count",0),"payload_over_png":ratio})
        storage_rows.append({"condition_id":condition,"payload_bytes":payload_bytes,"payload_over_png":ratio,"png_over_payload":storage_baseline_bytes/payload_bytes if payload_bytes else None,"size_saving_percent":(1-ratio)*100 if ratio is not None else None,"source_bytes":storage_baseline_bytes,"payload_definition":"encoded payload; protected=high PNG + low codec bytes; materialized inference excluded","status":weight.get("status"),"reason":weight.get("failure_reason")})
    _write_csv(target/"codec_conditions.csv", CONDITION_COLUMNS, condition_rows); _write_csv(target/"weight_metrics.csv", WEIGHT_COLUMNS, weight_rows); _write_csv(target/"generation_metrics.csv", GENERATION_COLUMNS, generation_rows); _write_csv(target/"storage_metrics.csv", STORAGE_COLUMNS, storage_rows); _write_csv(target/"condition_summary.csv", SUMMARY_COLUMNS, summaries); _write_csv(target/"paired_quality_metrics.csv", PAIRED_COLUMNS, paired)
    _plots(target, run, manifest, rows, summaries, prompts, fixture=fixture); _documents(target, manifest, summaries, paired, fixture=fixture)
    generated_at = manifest.get("completed_at") or manifest.get("started_at") or "unknown"
    artifacts=[{"path":str(path.relative_to(target)),"bytes":path.stat().st_size,"sha256":_sha_file(path)} for path in sorted(target.rglob("*")) if path.is_file()]
    report_manifest={"status":"completed","report_identity_hash":identity_hash,"report_identity":identity,"input_provenance":{"source_sha256":source_hash,"source_bytes":source_bytes,"baseline_source_path":str(baseline_source_artifact["path"]),"baseline_source_sha256":baseline_source_hash,"baseline_source_bytes":baseline_source_path.stat().st_size,"resource_monitor_path":resource["path"],"resource_monitor_sha256":resource["sha256"],"resource_monitor_bytes":resource["bytes"],"resource_monitor_row_count":resource["row_count"],"resource_monitor_violation_count":resource["violation_count"],"raw_rows_sha256":manifest["raw_rows_sha256"],"raw_rows_bytes":manifest["raw_rows_bytes"],"prompt_projection":prompt_projection},"bootstrap":{"seed":BOOTSTRAP_SEED,"resamples":BOOTSTRAP_SAMPLES,"order":"sorted prompt_id"},"generated_at":generated_at,"artifacts":artifacts}
    (target/"report_manifest.json").write_text(json.dumps(report_manifest,ensure_ascii=False,sort_keys=True,indent=2)+"\n",encoding="utf-8")
    return target


def build_parser() -> argparse.ArgumentParser:
    parser=argparse.ArgumentParser(description=__doc__); parser.add_argument("--run-dir",required=True); parser.add_argument("--output",default=None); parser.add_argument("--force-report",action="store_true"); return parser


def main() -> None:
    args=build_parser().parse_args(); generate_report(args.run_dir,args.output,force_report=args.force_report)


if __name__ == "__main__": main()