tests/test_cross_codec_experiment.py

"""Cross-codec Task 2 runnerの厳格TDD契約。"""
from __future__ import annotations

import hashlib
import json
from argparse import Namespace
from pathlib import Path

import numpy as np
import pytest
from PIL import Image

import pixelmodel_robustness.cross_codec as codecs
from pixelmodel_robustness.prompts import CATEGORIES


@pytest.fixture
def fixture_inputs(tmp_path: Path) -> dict[str, Path]:
    values = np.arange(16, dtype=np.float16)
    pixels = np.zeros((4, 4, 3), dtype=np.uint8)
    bits = values.view(np.uint16)
    pixels.reshape(-1, 3)[:, 0] = (bits >> 8).astype(np.uint8)
    pixels.reshape(-1, 3)[:, 1] = bits.astype(np.uint8)
    image = tmp_path / "model.png"
    Image.fromarray(pixels, "RGB").save(image)
    manifest = tmp_path / "model_png.json"
    manifest.write_text(json.dumps({"total_parameters": 16, "params": [{"name": "a", "numel": 8}, {"name": "b", "numel": 8}]}), encoding="utf-8")
    config = tmp_path / "config.json"
    config.write_text(json.dumps({"architecture": "fixture"}), encoding="utf-8")
    model = tmp_path / "model.safetensors"
    model.write_bytes(b"fixture")
    prompts = tmp_path / "prompts.json"
    rows = []
    for index, category in enumerate(CATEGORIES):
        rows.append({"prompt_id": f"first-{category}", "category": category, "prompt": f"first {category}", "seed": index, "initial_latent_seed": index})
    for index in range(92):
        category = CATEGORIES[index % len(CATEGORIES)]
        rows.append({"prompt_id": f"p-{index}", "category": category, "prompt": f"prompt {index}", "seed": 100 + index, "initial_latent_seed": 100 + index})
    prompts.write_text(json.dumps(rows), encoding="utf-8")
    return {"image": image, "manifest": manifest, "config": config, "model": model, "prompts": prompts}


def make_args(inputs: dict[str, Path], output: Path, **overrides: object) -> Namespace:
    values = {"profile": "pilot", "image": str(inputs["image"]), "manifest": str(inputs["manifest"]), "config": str(inputs["config"]), "model": str(inputs["model"]), "prompts_manifest": str(inputs["prompts"]), "prompts": 8, "steps": 50, "cfg": 6.0, "device": "mps", "output": str(output), "conditions": "", "recover_lock": False, "fixture": True, "fake_runtime": False, "vae": "vae", "clip": "clip", "pilot_manifest": None}
    values.update(overrides)
    return Namespace(**values)


def preflight() -> tuple[codecs.CodecPreflight, ...]:
    return tuple(codecs.CodecPreflight(name, True, "fixture", None, ("fixture", name), {"fixture": True}) for name in codecs.CODEC_SPECS)


def payload_factory(image: Image.Image, spec: codecs.CodecSpec) -> tuple[bytes, Image.Image]:
    from io import BytesIO
    stream = BytesIO()
    image.convert("RGB").save(stream, format="PNG")
    payload = stream.getvalue()
    return payload, Image.open(BytesIO(payload)).convert("RGB")


class Runtime:
    def generate(self, prompt: str, seed: int) -> tuple[Image.Image, float]:
        return Image.new("RGB", (4, 4), seed % 255), 0.5


def runtime_factory(args: Namespace, device: str) -> tuple[object, object]:
    return (lambda path: object()), Runtime()


def test_pilot_selection_is_category_order_and_full_is_exact() -> None:
    from pixelmodel_robustness.cross_codec_experiment import select_prompts
    prompts = [{"prompt_id": f"{category}-{index}-{duplicate}", "category": category, "prompt": f"{category}-{index}-{duplicate}", "seed": index, "initial_latent_seed": index} for index, category in enumerate(CATEGORIES) for duplicate in range(2)]
    selected = select_prompts(prompts, "pilot")
    assert [item["prompt_id"] for item in selected] == [next(item["prompt_id"] for item in prompts if item["category"] == category) for category in CATEGORIES]
    with pytest.raises(ValueError, match="exactly 100"):
        select_prompts(prompts, "full")


def test_cli_contract_rejects_before_output(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import build_parser, run
    args = build_parser().parse_args(["--profile", "pilot", "--image", str(fixture_inputs["image"]), "--manifest", str(fixture_inputs["manifest"]), "--config", str(fixture_inputs["config"]), "--model", str(fixture_inputs["model"]), "--prompts-manifest", str(fixture_inputs["prompts"]), "--output", str(tmp_path / "out"), "--prompts", "7"])
    with pytest.raises(ValueError, match="8"):
        run(args)
    assert not Path(args.output).exists()


def test_fixture_e2e_has_complete_rows_and_artifacts(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    weights = [row for row in rows if row["row_type"] == "weight"]
    layers = [row for row in rows if row["row_type"] == "layer"]
    generations = [row for row in rows if row["row_type"] == "generation"]
    assert len(weights) == 17 and len(layers) == 34 and len(generations) == 136
    assert all(row["attempted"] and row["completed"] and row["success"] for row in generations if not row["blocked"])
    assert all((result / row["sample_path"]).is_file() and row["sample_bytes"] == (result / row["sample_path"]).stat().st_size and row["sample_hash"] == hashlib.sha256((result / row["sample_path"]).read_bytes()).hexdigest() for row in generations if row["success"])
    baseline = next(row for row in weights if row["condition"] == "png_baseline")
    assert (result / baseline["materialized"]["path"]).read_bytes() == fixture_inputs["image"].read_bytes()
    assert json.loads((result / "pilot_gate.json").read_text())["pass"] is True


def test_save_sample_supports_pil_chw_and_rejects_nonfinite(tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import _save_sample
    class Tensor:
        def __init__(self, value: np.ndarray): self.value = value
        def detach(self): return self
        def cpu(self): return self
        def clamp(self, low: float, high: float): return Tensor(np.clip(self.value, low, high))
        def permute(self, *axes: int): return Tensor(np.transpose(self.value, axes))
        def numpy(self): return self.value
    path = tmp_path / "nested" / "sample.png"
    artifact = _save_sample(Tensor(np.ones((3, 4, 4), dtype=np.float32)), path, base=tmp_path)
    assert path.is_file() and artifact["sha256"] == hashlib.sha256(path.read_bytes()).hexdigest()
    with pytest.raises((ValueError, FloatingPointError)):
        _save_sample(Tensor(np.full((3, 4, 4), np.nan, dtype=np.float32)), tmp_path / "bad.png", base=tmp_path)


def test_loader_called_once_per_condition_and_baseline_path_exists(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    calls: list[str] = []
    def loader(path: str) -> object:
        calls.append(path)
        assert Path(path).is_file()
        return object()
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=lambda a, d: (loader, Runtime()))
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    nonblocked = {row["condition"] for row in rows if row["row_type"] == "weight" and not row["blocked"]}
    assert len(calls) == len(nonblocked)
    assert any(path.endswith("png_baseline.png") for path in calls)


def test_model_loader_failure_is_one_attempt_and_gate_fails(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    calls: list[str] = []
    def loader(path: str) -> object:
        calls.append(path)
        raise ValueError("load failed")
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=lambda a, d: (loader, Runtime()))
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    assert len(calls) == 17
    assert all(not row["attempted"] and row["blocked"] and row["failure_stage"] == "model_load" and row["sample_path"] is None for row in rows if row["row_type"] == "generation")
    assert json.loads((result / "pilot_gate.json").read_text())["pass"] is False


def test_raw_block_repair_count_and_shared_payload(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    def candidate(condition: str, values: np.ndarray) -> np.ndarray:
        result = values.copy()
        if condition == "jpeg_q100_raw": result[:3] = [np.nan, np.inf, -np.inf]
        if condition == "jpeg_q80_raw": result[:3] = [np.nan, np.inf, -np.inf]
        return result
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory, candidate_factory=candidate)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    weights = {row["condition"]: row for row in rows if row["row_type"] == "weight"}
    assert weights["jpeg_q100_raw"]["blocked"] is True
    assert weights["jpeg_q100_repair_zero"]["repaired_count"] == 3
    assert weights["jpeg_q100_raw"]["payload_sha256"] == weights["jpeg_q100_repair_zero"]["payload_sha256"]


def test_payload_storage_metrics_and_protected_artifacts(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    weights = {row["condition"]: row for row in rows if row["row_type"] == "weight"}
    for row in weights.values():
        assert {"payload_bytes", "payload_over_png", "png_over_payload", "size_saving_percent"} <= row.keys()
    protected = weights["jpeg_q100_protected_high"]
    assert {"high", "low"} <= protected["protected"].keys()
    assert protected["payload_bytes"] == protected["protected"]["high"]["bytes"] + protected["protected"]["low"]["bytes"]


def test_strict_rows_and_condition_completion_reject_tamper(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import condition_complete, load_rows_strict, run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest = json.loads((result / "manifest.json").read_text())
    rows = load_rows_strict(result / "results" / "raw_rows.jsonl", manifest["run_identity_hash"])
    sample = next(row for row in rows if row["row_type"] == "generation" and row["success"])
    (result / sample["sample_path"]).write_bytes(b"tampered")
    assert condition_complete(rows, sample["condition"], set(manifest["prompt_ids"]), {"a", "b"}, result) is False
    with pytest.raises(ValueError, match="identity"):
        load_rows_strict(result / "results" / "raw_rows.jsonl", "wrong")


def test_strict_rows_reject_duplicate_canonical_identity(tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import load_rows_strict
    path = tmp_path / "rows.jsonl"
    row = {"row_type": "weight", "stage": "weight", "condition": "png_baseline", "prompt_id": None, "run_identity_hash": "identity"}
    path.write_text(json.dumps(row) + "\n" + json.dumps(row) + "\n", encoding="utf-8")
    with pytest.raises(ValueError, match="duplicate"):
        load_rows_strict(path, "identity")


def test_resume_skips_completed_and_rejects_completed_resume(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    args = make_args(fixture_inputs, result)
    with pytest.raises(ValueError, match="completed"):
        run(args, preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)


def test_full_gate_validation_rejects_tampered_gate_before_output(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    pilot = run(make_args(fixture_inputs, tmp_path / "pilot"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    gate = pilot / "pilot_gate.json"
    value = json.loads(gate.read_text())
    value["prompt_ids"] = ["tampered"]
    gate.write_text(json.dumps(value), encoding="utf-8")
    args = make_args(fixture_inputs, tmp_path / "full", profile="full", prompts=100, fixture=False, pilot_manifest=str(gate))
    with pytest.raises(ValueError, match="pilot gate"):
        run(args, preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    assert not Path(args.output).exists()


def test_invalid_prompt_category_and_duplicate_ids_rejected_before_output(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    prompts = json.loads(fixture_inputs["prompts"].read_text())
    prompts[0]["category"] = "invalid"
    fixture_inputs["prompts"].write_text(json.dumps(prompts), encoding="utf-8")
    with pytest.raises(ValueError, match="category"):
        run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight)


def test_preflight_identity_order_and_duplicates_rejected(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    statuses = list(preflight())
    statuses[0], statuses[1] = statuses[1], statuses[0]
    with pytest.raises(RuntimeError, match="preflight"):
        run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=lambda: tuple(statuses))


def test_resource_sample_is_measured_and_three_violation_guard(fixture_inputs: dict[str, Path], tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    from pixelmodel_robustness import cross_codec_experiment as experiment
    snapshot = experiment.sample_resource(tmp_path)
    assert snapshot["rss_bytes"] > 0 and snapshot["system_available_percent"] > 0 and snapshot["disk_free_bytes"] > 0
    monkeypatch.setattr(experiment, "sample_resource", lambda output: {"rss_bytes": 13 * 1024**3, "system_available_percent": 5.0, "disk_free_bytes": 1})
    with pytest.raises(experiment.ResourceGuardError):
        experiment.run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight)
    assert not (tmp_path / "out").exists()


def test_generation_failure_has_no_fake_sample(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    class Failing(Runtime):
        def generate(self, prompt: str, seed: int): raise FloatingPointError("nonfinite image")
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=lambda a, d: (lambda path: object(), Failing()))
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    assert all(not row["success"] and row["sample_path"] is None and row["clip_score"] is None and row["first_error"] for row in rows if row["row_type"] == "generation")


def test_generation_row_schema_and_layer_rows_have_no_prompt_hack(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    generation = next(row for row in rows if row["row_type"] == "generation")
    assert {"category", "prompt", "seed", "initial_latent_seed", "attempt_count", "completed", "elapsed_seconds", "failure_stage", "failure_class", "first_error", "sample_path", "sample_hash", "sample_bytes"} <= generation.keys()
    assert all(not str(row.get("prompt_id", "")).startswith("layer:") for row in rows if row["row_type"] == "layer")


def test_full_fixture_gate_binding_unit_does_not_create_gate(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    pilot = run(make_args(fixture_inputs, tmp_path / "pilot"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    gate = pilot / "pilot_gate.json"
    args = make_args(fixture_inputs, tmp_path / "full", profile="full", prompts=100, fixture=False, pilot_manifest=str(gate))
    with pytest.raises(ValueError):
        run(args, preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    assert not (Path(args.output) / "pilot_gate.json").exists()


def test_preflight_unavailable_fails_before_output(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    def unavailable() -> tuple[codecs.CodecPreflight, ...]:
        return tuple(codecs.CodecPreflight(name, name != "avif_q70", "fixture", None if name != "avif_q70" else "missing", ("fixture", name), {}) for name in codecs.CODEC_SPECS)
    with pytest.raises(RuntimeError, match="preflight"):
        run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=unavailable)
    assert not (tmp_path / "out").exists()


@pytest.mark.parametrize("artifact_key", ["sample", "payload", "materialized", "protected_high", "protected_low"])
def test_validate_pilot_gate_rehashes_every_persisted_artifact(fixture_inputs: dict[str, Path], tmp_path: Path, artifact_key: str) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run, validate_pilot_gate
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    rows_path = result / "results" / "raw_rows.jsonl"
    rows = [json.loads(line) for line in rows_path.read_text().splitlines()]
    if artifact_key == "sample":
        target = next(row for row in rows if row["row_type"] == "generation" and row["success"])
        path = result / target["sample_path"]
    elif artifact_key == "payload":
        target = next(row for row in rows if row["row_type"] == "weight")
        path = result / target["payload"]["path"]
    elif artifact_key == "materialized":
        target = next(row for row in rows if row["row_type"] == "weight" and row["materialized"])
        path = result / target["materialized"]["path"]
    else:
        target = next(row for row in rows if row["row_type"] == "weight" and row["protected"])
        path = result / target["protected"]["high" if artifact_key.endswith("high") else "low"]["path"]
    path.write_bytes(path.read_bytes() + b"tamper")
    with pytest.raises(ValueError, match="pilot gate"):
        validate_pilot_gate(result / "pilot_gate.json")


@pytest.mark.parametrize("status", ["running", "resource_guard_failed"])
@pytest.mark.parametrize("tree_key", ["payload", "protected", "materialized"])
def test_resume_revalidates_existing_weight_artifact_tree(fixture_inputs: dict[str, Path], tmp_path: Path, status: str, tree_key: str) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest_path = result / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["status"] = status
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    rows_path = result / "results" / "raw_rows.jsonl"
    rows = [json.loads(line) for line in rows_path.read_text().splitlines()]
    weight = next(row for row in rows if row["row_type"] == "weight" and row[tree_key])
    artifact = weight[tree_key]
    if tree_key == "protected":
        artifact = artifact["high"]
    (result / artifact["path"]).write_bytes(b"tampered")
    with pytest.raises(ValueError, match="artifact tamper"):
        run(make_args(fixture_inputs, result, recover_lock=True), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)


def test_codec_encode_failure_is_canonical_and_gate_fails(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    def failing_payload(image: Image.Image, spec: codecs.CodecSpec) -> tuple[bytes, Image.Image]:
        if spec.name == "jpeg_q80":
            raise OSError("encoder unavailable")
        return payload_factory(image, spec)
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=failing_payload, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    failed = [row for row in rows if row["row_type"] == "weight" and row["condition"].startswith("jpeg_q80")]
    assert len(failed) == 3 and all(row["status"] == "codec_pipeline_failed" and row["blocked"] for row in failed)
    assert all(row["failure_stage"] == "encode" for row in rows if row["row_type"] == "generation" and row["condition"].startswith("jpeg_q80"))
    assert json.loads((result / "manifest.json").read_text())["status"] == "completed_with_pipeline_failures"
    assert json.loads((result / "pilot_gate.json").read_text())["pass"] is False


def test_codec_decode_failure_is_canonical_and_gate_fails(fixture_inputs: dict[str, Path], tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    import pixelmodel_robustness.cross_codec as module
    from pixelmodel_robustness.cross_codec_experiment import run
    original = module.decode_rgb
    def failing_decode(payload: bytes, spec: codecs.CodecSpec, **kwargs: object) -> Image.Image:
        if spec.name == "jpeg_q80":
            raise ValueError("decoder rejected payload")
        return original(payload, spec, **kwargs)
    monkeypatch.setattr(module, "decode_rgb", failing_decode)
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    failed = [row for row in rows if row["row_type"] == "generation" and row["condition"].startswith("jpeg_q80")]
    assert len(failed) == 8 * 3 and all(not row["attempted"] and row["failure_stage"] == "decode" for row in failed)
    assert json.loads((result / "pilot_gate.json").read_text())["pass"] is False


def test_resource_preflight_is_persisted_once(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "resource_monitor.jsonl").read_text().splitlines()]
    preflight_rows = [row for row in rows if row["row_type"] == "resource_guard" and row["stage"] == "preflight"]
    assert len(preflight_rows) == 1 and preflight_rows[0]["rss_bytes"] > 0


def test_protected_payload_is_read_back_before_generation(fixture_inputs: dict[str, Path], tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    import pixelmodel_robustness.cross_codec as module
    from pixelmodel_robustness.cross_codec_experiment import run
    calls = 0
    original = module.decode_protected_high
    def counted(*args: object, **kwargs: object):
        nonlocal calls
        calls += 1
        return original(*args, **kwargs)
    monkeypatch.setattr(module, "decode_protected_high", counted)
    run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    assert calls == 5


def test_scientific_gate_identity_projection_is_immutable_and_profile_neutral() -> None:
    from pixelmodel_robustness.cross_codec_experiment import SCIENTIFIC_GATE_IDENTITY_KEYS, scientific_identity_hash
    assert isinstance(SCIENTIFIC_GATE_IDENTITY_KEYS, tuple)
    base = {key: f"value-{key}" for key in SCIENTIFIC_GATE_IDENTITY_KEYS}
    base["mode"] = "real"
    pilot = {**base, "profile": "pilot", "prompt_ids": ["p1"], "prompt_selection_hash": "pilot", "completed_at": "old"}
    full = {**base, "profile": "full", "prompt_ids": ["p1", "p2"], "prompt_selection_hash": "full", "completed_at": "new"}
    assert scientific_identity_hash(pilot) == scientific_identity_hash(full)
    changed = {**full, "code_hashes": "tampered"}
    assert scientific_identity_hash(full) != scientific_identity_hash(changed)


def test_manifest_binds_jpeg_repair_module_hash_to_run_identity(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness import jpeg_repair
    from pixelmodel_robustness.cross_codec_experiment import run, scientific_identity_hash, sha256_file

    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest = json.loads((result / "manifest.json").read_text())
    expected = sha256_file(jpeg_repair.__file__)

    assert manifest["run_identity"]["code_hashes"]["jpeg_repair"] == expected
    assert manifest["scientific_identity_hash"] == scientific_identity_hash(manifest["run_identity"])
    changed = {**manifest["run_identity"], "code_hashes": {**manifest["run_identity"]["code_hashes"], "jpeg_repair": "tampered"}}
    assert scientific_identity_hash(changed) != manifest["scientific_identity_hash"]


def test_webp_lossless_pixel_mismatch_is_pipeline_failure(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run
    def mismatching_payload(image: Image.Image, spec: codecs.CodecSpec) -> tuple[bytes, Image.Image]:
        payload, decoded = payload_factory(image, spec)
        if spec.name == "webp_lossless":
            decoded = decoded.copy()
            decoded.putpixel((0, 0), (255, 0, 0))
        return payload, decoded
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=mismatching_payload, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    affected = [row for row in rows if row["condition"] == "webp_lossless"]
    assert len([row for row in affected if row["row_type"] == "weight"]) == 1
    assert all(row["failure_stage"] == "decode_invariant" for row in affected)
    assert json.loads((result / "manifest.json").read_text())["status"] == "completed_with_pipeline_failures"
    assert json.loads((result / "pilot_gate.json").read_text())["pass"] is False


def test_protected_decode_failure_is_condition_local(fixture_inputs: dict[str, Path], tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    import pixelmodel_robustness.cross_codec as module
    from pixelmodel_robustness.cross_codec_experiment import run
    original = module.decode_protected_high
    def failing_decode(artifact: object, shape: object, values: object, **kwargs: object):
        if getattr(artifact, "low_codec", None) == "jpeg_q80":
            raise RuntimeError("protected decoder failed")
        return original(artifact, shape, values, **kwargs)
    monkeypatch.setattr(module, "decode_protected_high", failing_decode)
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    failed_weights = [row for row in rows if row["row_type"] == "weight" and row["status"] == "codec_pipeline_failed"]
    assert [row["condition"] for row in failed_weights] == ["jpeg_q80_protected_high"]
    assert all(row["failure_stage"] == "protected_invariant" for row in failed_weights)
    assert json.loads((result / "pilot_gate.json").read_text())["pass"] is False


def test_incomplete_failed_generation_rejects_any_sample_artifact_field(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import condition_complete, load_rows_strict, run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest = json.loads((result / "manifest.json").read_text())
    rows = load_rows_strict(result / "results" / "raw_rows.jsonl", manifest["run_identity_hash"])
    target = next(row for row in rows if row["row_type"] == "generation")
    target["success"] = False
    target["attempted"] = True
    target["completed"] = True
    target["failure_stage"] = "generation"
    target["sample_bytes"] = 1
    assert condition_complete(rows, target["condition"], set(manifest["prompt_ids"]), set(manifest["layer_ids"]), result) is False


def test_completed_manifest_records_raw_rows_bytes(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run

    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest = json.loads((result / "manifest.json").read_text())
    raw = result / "results" / "raw_rows.jsonl"
    assert manifest["raw_rows_bytes"] == raw.stat().st_size


def test_resource_monitor_is_bound_and_gate_rejects_resource_tamper(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import run, validate_pilot_gate
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest = json.loads((result / "manifest.json").read_text())
    metadata = manifest["resource_monitor"]
    assert metadata["binding_method"] == "run_completion" and metadata["bytes"] > 0 and metadata["sha256"]
    resource = result / metadata["path"]
    resource.write_bytes(resource.read_bytes() + b'{}\n')
    with pytest.raises(ValueError, match="pilot gate|resource"):
        validate_pilot_gate(result / "pilot_gate.json")


def test_resource_monitor_migration_preserves_data_and_is_idempotent(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import (
        bind_completed_resource_monitor,
        run,
        validate_pilot_gate,
        validate_resource_monitor,
    )
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest_path = result / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    original_identity = {key: manifest[key] for key in ("source_sha256", "run_identity", "run_identity_hash", "scientific_identity_hash")}
    manifest.pop("resource_monitor")
    manifest["mode"] = "real"
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    tracked = [result / "results" / "raw_rows.jsonl", result / "results" / "resource_monitor.jsonl"]
    tracked += [path for path in result.rglob("*") if path.is_file() and ("payloads" in path.parts or "samples" in path.parts)]
    before = {str(path.relative_to(result)): (path.stat().st_size, hashlib.sha256(path.read_bytes()).hexdigest()) for path in tracked}
    bind_completed_resource_monitor(result, fixture_inputs["image"])
    bind_completed_resource_monitor(result, fixture_inputs["image"])
    after = {str(path.relative_to(result)): (path.stat().st_size, hashlib.sha256(path.read_bytes()).hexdigest()) for path in tracked}
    updated = json.loads(manifest_path.read_text())
    assert before == after
    assert {key: updated[key] for key in original_identity} == original_identity
    assert updated["resource_monitor"]["binding_method"] == "post_run_verified"
    gate_data = validate_pilot_gate(result / "pilot_gate.json")
    assert gate_data["gate"]["resource_monitor"] == updated["resource_monitor"]
    actual = validate_resource_monitor(result, updated, [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()])
    assert all(updated["resource_monitor"][key] == actual[key] for key in actual)


def test_migration_gate_rejects_forged_resource_annotations(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import (
        _hash,
        bind_completed_resource_monitor,
        run,
        sha256_file,
        validate_pilot_gate,
    )
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest_path = result / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest["mode"] = "real"
    manifest.pop("resource_monitor")
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    bind_completed_resource_monitor(result, fixture_inputs["image"])
    gate_path = result / "pilot_gate.json"
    gate = json.loads(gate_path.read_text())
    gate["resource_monitor"]["binding_method"] = "unknown"
    gate_path.write_text(json.dumps(gate), encoding="utf-8")
    with pytest.raises(ValueError, match="pilot gate|tampered"):
        validate_pilot_gate(gate_path)
    manifest = json.loads(manifest_path.read_text())
    manifest["resource_monitor"]["binding_method"] = "unknown"
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    gate = json.loads(gate_path.read_text())
    gate["resource_monitor"] = manifest["resource_monitor"]
    gate["manifest_sha256"] = sha256_file(manifest_path)
    gate["gate_hash"] = _hash({key: value for key, value in gate.items() if key != "gate_hash"})
    gate_path.write_text(json.dumps(gate), encoding="utf-8")
    with pytest.raises(ValueError, match="resource|pilot gate"):
        validate_pilot_gate(gate_path)


def test_bound_resource_rejects_invalid_post_run_annotations(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import _validate_bound_resource, run, validate_resource_monitor
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest = json.loads((result / "manifest.json").read_text())
    rows = [json.loads(line) for line in (result / "results" / "raw_rows.jsonl").read_text().splitlines()]
    actual = validate_resource_monitor(result, manifest, rows)
    for annotation in ({"bound_at": "not-utc", "binding_code_hash": "a" * 64}, {"bound_at": "2026-01-01T00:00:00+00:00", "binding_code_hash": "A" * 64}):
        recorded = {**actual, "binding_method": "post_run_verified", **annotation}
        with pytest.raises((ValueError, TypeError), match="bound_at|binding_code_hash"):
            _validate_bound_resource(recorded, actual)


def test_resource_monitor_migration_rejects_wrong_source(fixture_inputs: dict[str, Path], tmp_path: Path) -> None:
    from pixelmodel_robustness.cross_codec_experiment import bind_completed_resource_monitor, run
    result = run(make_args(fixture_inputs, tmp_path / "out"), preflight_factory=preflight, payload_factory=payload_factory, runtime_factory=runtime_factory)
    manifest_path = result / "manifest.json"
    manifest = json.loads(manifest_path.read_text())
    manifest.pop("resource_monitor")
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    with pytest.raises(ValueError, match="source"):
        bind_completed_resource_monitor(result, tmp_path / "wrong.png")