scripts/pixelmodel_robustness/cross_codec_experiment.py

"""Task 2 cross-codec robustness runner.

科学条件、payload、推論、resume、provenanceを単一writerで管理する。
"""
from __future__ import annotations

import argparse
import hashlib
import json
import logging
import math
import os
import platform
import shutil
import socket
import subprocess
import time
from collections.abc import Callable, Mapping, Sequence
from dataclasses import fields, is_dataclass
from datetime import UTC, datetime, timedelta
from io import BytesIO
from pathlib import Path
from types import MappingProxyType
from typing import Any

import numpy as np
from PIL import Image, UnidentifiedImageError

from pixelmodel_robustness import cross_codec, jpeg_repair
from pixelmodel_robustness.codec import layer_offsets
from pixelmodel_robustness.inference import build_official_runtime, generate_with_runtime, resolve_device
from pixelmodel_robustness.jpeg_repair import diagnostics, materialize_fp16_png, repair_nonfinite
from pixelmodel_robustness.metrics import weight_metrics
from pixelmodel_robustness.prompts import CATEGORIES
from pixelmodel_robustness.storage import atomic_json

LOGGER = logging.getLogger(__name__)
VAE_REVISION = "31f26fdeee1355a5c34592e401dd41e45d25a493"
CLIP_REVISION = "3d74acf9a28c67741b2f4f2ea7635f0aaf6f0268"
UPSTREAM_REVISION = "80878fdd6e130d229c2581424a3456b47b979d4b"
SCIENTIFIC_GATE_IDENTITY_KEYS = (
    "source_sha256", "model_sha256", "config_sha256", "prompt_sha256", "upstream_revision", "code_hashes",
    "runtime_git", "runtime_platform", "package_versions", "vae", "clip", "requested_device",
    "resolved_device", "mode", "steps", "cfg", "condition_registry_hash", "codec_preflight",
)


class ResourceGuardError(RuntimeError):
    """resource gateが実験を停止した。"""


def _safe(value: object) -> object:
    """tuple、mappingproxy、dataclass、numpy scalarをJSON-safeへ変換する。"""
    if is_dataclass(value):
        return _safe({field.name: getattr(value, field.name) for field in fields(value)})
    if isinstance(value, Mapping):
        return {str(key): _safe(item) for key, item in value.items()}
    if isinstance(value, (tuple, list)):
        return [_safe(item) for item in value]
    if isinstance(value, np.generic):
        return _safe(value.item())
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, float) and not math.isfinite(value):
        return None
    return value


def _hash(value: object) -> str:
    """canonical JSONのSHA-256を返す。"""
    return hashlib.sha256(json.dumps(_safe(value), ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def scientific_identity_hash(identity: Mapping[str, object]) -> str:
    """profile/prompt scopeを除いた共通科学identityをhashする。"""
    return _hash({key: identity.get(key) for key in SCIENTIFIC_GATE_IDENTITY_KEYS})


def sha256_file(path: str | Path) -> str:
    """ファイルのSHA-256を返す。"""
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def _git(root: Path) -> dict[str, object]:
    """runtime git metadataをbest effortで取得する。"""
    try:
        head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=root, capture_output=True, text=True, check=True).stdout.strip()
        dirty = bool(subprocess.run(["git", "status", "--porcelain"], cwd=root, capture_output=True, text=True, check=True).stdout)
    except (OSError, subprocess.SubprocessError):
        head, dirty = None, None
    return {"head": head, "dirty": dirty}


def sample_resource(output: Path, condition: str | None = None, prompt_id: str | None = None) -> dict[str, object]:
    """psutil/shutil/Torch MPSから実測snapshotを取得する。"""
    import psutil
    process = psutil.Process(os.getpid())
    memory = psutil.virtual_memory()
    target = output if output.exists() else output.parent
    disk = shutil.disk_usage(target)
    mps_allocated = None
    mps_driver = None
    try:
        import torch
        if torch.backends.mps.is_available():
            mps_allocated = int(torch.mps.current_allocated_memory())
            mps_driver = int(torch.mps.driver_allocated_memory())
    except (ImportError, AttributeError, RuntimeError):
        pass
    return {"timestamp": datetime.now(UTC).isoformat(), "condition": condition, "prompt_id": prompt_id, "pid": os.getpid(), "rss_bytes": int(process.memory_info().rss), "system_available_percent": float(memory.available / memory.total * 100), "disk_free_bytes": int(disk.free), "mps_allocated_bytes": mps_allocated, "mps_driver_allocated_bytes": mps_driver, "violation_reason": None}


def _resource_violation(snapshot: Mapping[str, object]) -> str | None:
    """JPEG runnerと同じresource thresholdsを評価する。"""
    reasons = []
    if float(snapshot["system_available_percent"]) < 10:
        reasons.append("system_available_percent<10")
    if int(snapshot["rss_bytes"]) > 12 * 1024**3:
        reasons.append("rss_bytes>12GiB")
    if int(snapshot["disk_free_bytes"]) < 50 * 1024**3:
        reasons.append("disk_free_bytes<50GiB")
    return ";".join(reasons) if reasons else None


def _early_resource_preflight(output: Path) -> dict[str, object]:
    """heavy workとoutput作成前にresource gateを実施する。"""
    snapshot = sample_resource(output)
    violation = _resource_violation(snapshot)
    if violation:
        raise ResourceGuardError(violation)
    return snapshot


def load_rows_strict(path: str | Path, run_identity_hash: str) -> list[dict[str, object]]:
    """identityとcanonical duplicateを拒否してraw rowsを読む。"""
    target = Path(path)
    if not target.exists():
        return []
    rows: list[dict[str, object]] = []
    keys: set[str] = set()
    for line in target.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        row = json.loads(line)
        if row.get("run_identity_hash") != run_identity_hash:
            raise ValueError("row run identity mismatch")
        key = _hash({key: row.get(key) for key in ("row_type", "stage", "condition", "prompt_id", "layer")})
        if key in keys:
            raise ValueError("duplicate canonical row identity")
        keys.add(key)
        rows.append(row)
    return rows


def validate_resource_monitor(
    output: str | Path,
    manifest: Mapping[str, object],
    generation_rows: Sequence[Mapping[str, object]],
) -> dict[str, object]:
    """resource monitorをstrictに読み、manifestへ束縛可能なmetadataを返す。"""
    root = Path(output).resolve()
    path = root / "results" / "resource_monitor.jsonl"
    try:
        path.relative_to(root)
    except ValueError as exc:
        raise ValueError("resource monitor path escapes run output") from exc
    if not path.is_file():
        raise ValueError("resource monitor missing")
    run_hash = manifest.get("run_identity_hash")
    if not isinstance(run_hash, str) or not run_hash:
        raise ValueError("resource monitor run identity missing")
    rows: list[dict[str, object]] = []
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
        for line_number, line in enumerate(lines, 1):
            if not line.strip():
                raise ValueError(f"resource monitor blank line: {line_number}")
            row = json.loads(line)
            if not isinstance(row, dict):
                raise TypeError(f"resource monitor row is not an object: {line_number}")
            if row.get("run_identity_hash") != run_hash:
                raise ValueError("resource monitor row identity mismatch")
            rows.append(row)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("resource monitor is not strict JSONL") from exc
    preflight = [row for row in rows if row.get("row_type") == "resource_guard" and row.get("stage") == "preflight"]
    if len(preflight) != 1 or any(row.get("condition") is not None or row.get("prompt_id") is not None for row in preflight):
        raise ValueError("resource monitor preflight row mismatch")
    resource_rows = [row for row in rows if row.get("row_type") == "resource" and row.get("stage") == "resource"]
    if len(preflight) + len(resource_rows) != len(rows):
        raise ValueError("unknown resource monitor row type/stage")
    if len(resource_rows) != len({(row.get("condition"), row.get("prompt_id")) for row in resource_rows}):
        raise ValueError("duplicate resource monitor identity")
    expected = {
        (row.get("condition"), str(row.get("prompt_id")))
        for row in generation_rows
        if row.get("row_type") == "generation" and row.get("attempted") is True and not row.get("blocked")
    }
    actual = {(row.get("condition"), str(row.get("prompt_id"))) for row in resource_rows}
    if actual != expected:
        raise ValueError("resource monitor generation pair mismatch")
    violation_rows = [row for row in rows if row.get("violation_reason")]
    available = [float(row["system_available_percent"]) for row in rows if isinstance(row.get("system_available_percent"), (int, float))]
    rss = [int(row["rss_bytes"]) for row in rows if isinstance(row.get("rss_bytes"), (int, float))]
    mps = [int(row["mps_allocated_bytes"]) for row in rows if isinstance(row.get("mps_allocated_bytes"), (int, float))]
    driver = [int(row["mps_driver_allocated_bytes"]) for row in rows if isinstance(row.get("mps_driver_allocated_bytes"), (int, float))]
    return {
        "path": str(path.relative_to(root)),
        "sha256": sha256_file(path),
        "bytes": path.stat().st_size,
        "row_count": len(rows),
        "preflight_count": len(preflight),
        "generation_count": len(resource_rows),
        "violation_count": len(violation_rows),
        "min_available_percent": min(available) if available else None,
        "peak_rss_bytes": max(rss) if rss else None,
        "peak_mps_allocated_bytes": max(mps) if mps else None,
        "peak_mps_driver_allocated_bytes": max(driver) if driver else None,
    }


def _validate_bound_resource(recorded: object, actual: Mapping[str, object]) -> None:
    """resource core metadataとbinding annotationをstrictに検証する。"""
    if not isinstance(recorded, Mapping):
        raise TypeError("resource monitor provenance must be an object")
    if any(recorded.get(key) != value for key, value in actual.items()):
        raise ValueError("resource monitor core metadata mismatch")
    annotations = set(recorded) - set(actual)
    if not annotations <= {"binding_method", "bound_at", "binding_code_hash"}:
        raise ValueError("unknown resource monitor binding annotation")
    method = recorded.get("binding_method")
    if method not in {"run_completion", "post_run_verified"}:
        raise ValueError("invalid resource monitor binding method")
    if method == "run_completion":
        if annotations != {"binding_method"}:
            raise ValueError("run completion resource annotation mismatch")
        return
    if annotations != {"binding_method", "bound_at", "binding_code_hash"}:
        raise ValueError("post-run resource annotation mismatch")
    bound_at = recorded.get("bound_at")
    if not isinstance(bound_at, str):
        raise TypeError("post-run resource bound_at is invalid")
    try:
        parsed = datetime.fromisoformat(bound_at)
    except ValueError as exc:
        raise ValueError("post-run resource bound_at is invalid") from exc
    if parsed.tzinfo is None or parsed.utcoffset() != timedelta(0):
        raise ValueError("post-run resource bound_at is not UTC")
    code_hash = recorded.get("binding_code_hash")
    if not isinstance(code_hash, str) or len(code_hash) != 64 or any(char not in "0123456789abcdef" for char in code_hash):
        raise ValueError("post-run resource binding_code_hash is invalid")


def _save_row(path: Path, row: dict[str, object]) -> None:
    """layerを含むTask2 canonical keyでatomic upsertする。"""
    rows = load_rows_strict(path, str(row["run_identity_hash"]))
    key = _hash({field: row.get(field) for field in ("row_type", "stage", "condition", "prompt_id", "layer")})
    rows = [item for item in rows if _hash({field: item.get(field) for field in ("row_type", "stage", "condition", "prompt_id", "layer")}) != key]
    rows.append(row)
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.tmp")
    temporary.write_text("".join(json.dumps(_safe(item), ensure_ascii=False, allow_nan=False) + "\n" for item in rows), encoding="utf-8")
    os.replace(temporary, path)


def _verify_artifact(output: Path, artifact: object, label: str) -> bool:
    """run-owned artifactのcontainment、bytes、hash、decoded metadataを検証する。"""
    if not isinstance(artifact, Mapping) or not isinstance(artifact.get("path"), str):
        return False
    path = (output / str(artifact["path"])).resolve()
    try:
        path.relative_to(output.resolve())
    except ValueError:
        return False
    if not path.is_file() or int(artifact.get("bytes", -1)) != path.stat().st_size or artifact.get("sha256") != sha256_file(path):
        return False
    if artifact.get("decoded_sha256"):
        try:
            with Image.open(path) as image:
                decoded = np.asarray(image.convert("RGB"), dtype=np.uint8)
                if artifact.get("mode") != image.mode or artifact.get("dimensions") != [image.height, image.width] or artifact.get("decoded_sha256") != hashlib.sha256(decoded.tobytes()).hexdigest():
                    return False
        except (OSError, UnidentifiedImageError):
            return False
    return True


def _verify_artifact_tree(output: Path, value: object, label: str) -> bool:
    """nested artifact metadataを再帰的に検証する。"""
    if isinstance(value, Mapping):
        if "path" in value and not _verify_artifact(output, value, label):
            return False
        return all(_verify_artifact_tree(output, child, f"{label}.{key}") for key, child in value.items() if isinstance(child, (Mapping, list)))
    if isinstance(value, list):
        return all(_verify_artifact_tree(output, child, f"{label}[{index}]") for index, child in enumerate(value))
    return True


def condition_complete(rows: Sequence[Mapping[str, object]], condition: str, prompt_ids: set[str], layer_ids: set[str], output: Path) -> bool:
    """weight/layer/promptと全artifactをstrictに検証する。"""
    weights = [row for row in rows if row.get("row_type") == "weight" and row.get("condition") == condition]
    layers = [row for row in rows if row.get("row_type") == "layer" and row.get("condition") == condition]
    generations = [row for row in rows if row.get("row_type") == "generation" and row.get("condition") == condition]
    if len(weights) != 1 or {str(row.get("layer")) for row in layers} != layer_ids or {str(row.get("prompt_id")) for row in generations} != prompt_ids:
        return False
    weight = weights[0]
    if not _verify_artifact_tree(output, weight.get("payload"), condition + ".payload") or not _verify_artifact_tree(output, weight.get("protected"), condition + ".protected") or not _verify_artifact_tree(output, weight.get("materialized"), condition + ".materialized"):
        return False
    for row in generations:
        if row.get("success"):
            if not row.get("attempted") or not row.get("completed") or row.get("blocked") or not _verify_artifact(output, {"path": row.get("sample_path"), "bytes": row.get("sample_bytes"), "sha256": row.get("sample_hash")}, f"{condition}.sample"):
                return False
        elif row.get("attempted"):
            if not row.get("completed") or row.get("sample_path") is not None or row.get("sample_hash") is not None or row.get("sample_bytes") is not None:
                return False
        elif not row.get("blocked") or not row.get("failure_stage") or row.get("sample_path") is not None or row.get("sample_hash") is not None or row.get("sample_bytes") is not None:
            return False
    return True


def select_prompts(prompts: Sequence[Mapping[str, object]], profile: str) -> list[dict[str, object]]:
    """prompt全体を検証しpilot/fullのordered scopeを返す。"""
    entries = [dict(item) for item in prompts]
    required = {"prompt_id", "category", "prompt", "seed", "initial_latent_seed"}
    if any(required - item.keys() for item in entries):
        raise ValueError("prompt schema is incomplete")
    ids = [str(item["prompt_id"]) for item in entries]
    if len(ids) != len(set(ids)) or len({str(item["prompt"]) for item in entries}) != len(entries):
        raise ValueError("duplicate prompt IDs or prompt text")
    if any(item["category"] not in CATEGORIES for item in entries):
        raise ValueError("invalid prompt category")
    if profile == "pilot":
        try:
            selected = [next(item for item in entries if item["category"] == category) for category in CATEGORIES]
        except StopIteration as exc:
            raise ValueError("pilot requires every prompt category") from exc
        return selected
    if profile == "full":
        if len(entries) != 100:
            raise ValueError("full profile requires exactly 100 prompts")
        return entries
    raise ValueError("profile must be pilot or full")


def build_parser() -> argparse.ArgumentParser:
    """仕様のCLI parserを返す。"""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--profile", choices=("pilot", "full"), default="pilot")
    parser.add_argument("--image", default="model.png")
    parser.add_argument("--manifest", default="model_png.json")
    parser.add_argument("--config", default="config.json")
    parser.add_argument("--model", default="model.safetensors")
    parser.add_argument("--prompts-manifest", default="prompts.json")
    parser.add_argument("--prompts", type=int, default=8)
    parser.add_argument("--steps", type=int, default=50)
    parser.add_argument("--cfg", type=float, default=6.0)
    parser.add_argument("--device", choices=("auto", "cpu", "mps", "cuda"), default="mps")
    parser.add_argument("--output", default="results/cross-codec-robustness")
    parser.add_argument("--conditions", default="")
    parser.add_argument("--recover-lock", action="store_true")
    parser.add_argument("--fixture", action="store_true")
    parser.add_argument("--fake-runtime", action="store_true")
    parser.add_argument("--vae", default="stabilityai/sd-vae-ft-mse")
    parser.add_argument("--clip", default="openai/clip-vit-base-patch32")
    parser.add_argument("--pilot-manifest")
    parser.add_argument("--bind-resource-monitor", action="store_true")
    return parser


def _validate(args: argparse.Namespace, injected: bool) -> None:
    """output作成前にCLI scopeを検証する。"""
    expected = 8 if args.profile == "pilot" else 100
    if args.prompts != expected or args.steps != 50 or args.cfg != 6.0 or args.device != "mps":
        raise ValueError(f"{args.profile} requires prompts={expected}, steps=50, cfg=6.0, requested device mps")
    if args.fixture or args.fake_runtime:
        if not injected:
            raise ValueError("fixture/fake-runtime is test-only API mode")
        if args.profile == "full":
            raise ValueError("full profile rejects fixture/fake-runtime")
    if args.conditions and tuple(item for item in args.conditions.split(",") if item) != cross_codec.CONDITION_IDS:
        raise ValueError("conditions must be the ordered 17 registry")
    if args.profile == "full" and not args.pilot_manifest:
        raise ValueError("full profile requires --pilot-manifest")


def _acquire_lock(path: Path, identity: str, recover: bool) -> None:
    """O_EXCL lockを作る。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists():
        if not recover:
            raise ValueError("run lock exists; use --recover-lock")
        path.unlink()
    try:
        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    except FileExistsError as exc:
        raise ValueError("run lock race") from exc
    with os.fdopen(fd, "w", encoding="utf-8") as stream:
        json.dump({"pid": os.getpid(), "hostname": socket.gethostname(), "identity_hash": identity}, stream)


def _artifact(path: Path, root: Path) -> dict[str, object]:
    """portable artifact metadataを生成する。"""
    result: dict[str, object] = {"path": str(path.relative_to(root)), "bytes": path.stat().st_size, "sha256": sha256_file(path)}
    try:
        with Image.open(path) as image:
            decoded = np.asarray(image.convert("RGB"), dtype=np.uint8)
            result.update({"mode": image.mode, "dimensions": [image.height, image.width], "decoded_sha256": hashlib.sha256(decoded.tobytes()).hexdigest()})
    except (OSError, UnidentifiedImageError):
        pass
    return result


def _save_sample(image: Any, path: Path, *, base: Path | None = None) -> dict[str, object]:
    """PILまたはCHW/HWC tensor-like imageをPNGへ保存しartifactを返す。"""
    if isinstance(image, Image.Image):
        rgb = image.convert("RGB")
    else:
        value = image.detach().cpu() if hasattr(image, "detach") else image
        if hasattr(value, "clamp"):
            value = value.clamp(0, 1)
        value = value.numpy() if hasattr(value, "numpy") else np.asarray(value)
        if value.ndim == 3 and value.shape[0] in (1, 3) and value.shape[-1] not in (1, 3):
            value = np.transpose(value, (1, 2, 0))
        if value.ndim != 3 or value.shape[-1] != 3:
            raise ValueError("generated image must be RGB HWC or CHW")
        if not np.isfinite(value).all():
            raise FloatingPointError("generated image contains nonfinite values")
        value = np.asarray(value, dtype=np.float64)
        if value.max() <= 1.0:
            value *= 255.0
        if value.min() < 0 or value.max() > 255:
            raise ValueError("generated image outside [0,255]")
        rgb = Image.fromarray(np.rint(value).astype(np.uint8), "RGB")
    path.parent.mkdir(parents=True, exist_ok=True)
    rgb.save(path, format="PNG")
    return _artifact(path, base or path.parent)


def _source_values(image: np.ndarray, manifest: Mapping[str, object]) -> tuple[np.ndarray, dict[str, tuple[int, int]]]:
    """official PNGのtotal parametersだけをfp16へ復元する。"""
    total = int(manifest["total_parameters"])
    pixels = image.reshape(-1, 3)[:total]
    if len(pixels) != total:
        raise ValueError("source image is shorter than manifest")
    bits = (pixels[:, 0].astype(np.uint16) << 8) | pixels[:, 1].astype(np.uint16)
    params = list(manifest["params"])
    return bits.view(np.float16), layer_offsets([str(item["name"]) for item in params], [int(item["numel"]) for item in params])


def _metrics(reference: np.ndarray, candidate: np.ndarray, layers: Mapping[str, tuple[int, int]]) -> dict[str, object]:
    """finite-pair metricと既存bit diagnosticsを返す。"""
    pair = np.isfinite(reference) & np.isfinite(candidate)
    left, right = reference[pair].astype(np.float64), candidate[pair].astype(np.float64)
    diff = right - left
    norm = float(np.linalg.norm(left) * np.linalg.norm(right))
    bit = diagnostics(reference, candidate, layers)
    return {"finite": bool(np.isfinite(candidate).all()), "nan_count": int(np.isnan(candidate).sum()), "posinf_count": int(np.isposinf(candidate).sum()), "neginf_count": int(np.isneginf(candidate).sum()), "finite_pair_count": int(pair.sum()), "finite_pair_label": "all_elements" if pair.all() else "finite_pairs_only", "mse": float(np.mean(diff * diff)) if len(diff) else None, "mae": float(np.mean(np.abs(diff))) if len(diff) else None, "cosine": float(np.dot(left, right) / norm) if norm else (1.0 if not left.any() and not right.any() else 0.0), "sign_mismatch_count": bit["sign_mismatch_count"], "exponent_mismatch_count": bit["exponent_mismatch_count"], "mantissa_mismatch_count": bit["mantissa_mismatch_count"], "weight_metrics": _safe(weight_metrics(reference, candidate, layers))}


def _write(path: Path, data: bytes) -> None:
    """親dirを作成してbytesを書き込む。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)


def _load_prompts(path: Path) -> list[dict[str, object]]:
    """prompt JSONを読む。"""
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, list):
        raise TypeError("prompt manifest must be a list")
    return select_prompts(value, "full" if len(value) == 100 else "pilot")


def _generation_row(condition: str, prompt: Mapping[str, object], identity: str, **fields: object) -> dict[str, object]:
    """generation canonical rowを作る。"""
    return {"row_type": "generation", "stage": "generation", "condition": condition, "prompt_id": str(prompt["prompt_id"]), "category": prompt["category"], "prompt": prompt["prompt"], "seed": prompt["seed"], "initial_latent_seed": prompt["initial_latent_seed"], "run_identity_hash": identity, "attempted": False, "attempt_count": 0, "completed": False, "success": False, "blocked": False, "clip_score": None, "elapsed_seconds": None, "failure_stage": None, "failure_class": None, "first_error": None, "sample_path": None, "sample_hash": None, "sample_bytes": None, **fields}


def _record_codec_failure(path: Path, condition: str, spec: cross_codec.ConditionSpec, prompts: Sequence[Mapping[str, object]], layers: Mapping[str, tuple[int, int]], identity: str, failure: Mapping[str, object]) -> None:
    """codec pipeline failureをweight/layer/generationのcanonical rowsへ保存する。"""
    common = {"codec": spec.codec, "mode": spec.mode, "attempted": False, "status": "codec_pipeline_failed", "blocked": True, "failure_stage": failure["stage"], "failure_class": failure["class"], "failure_reason": failure["reason"], "first_failure": failure["first_error"], "payload": {"failure": dict(failure)}, "protected": {"failure": dict(failure)} if spec.protected else None, "materialized": None, "payload_bytes": None, "payload_sha256": None, "payload_over_png": None, "png_over_payload": None, "size_saving_percent": None, "repaired_count": 0}
    _save_row(path, {"row_type": "weight", "stage": "weight", "condition": condition, "prompt_id": None, "run_identity_hash": identity, **common})
    for layer in layers:
        _save_row(path, {"row_type": "layer", "stage": "layer", "condition": condition, "layer": layer, "prompt_id": None, "run_identity_hash": identity, "status": "codec_pipeline_failed", "failure_stage": failure["stage"], "failure_class": failure["class"], "failure_reason": failure["reason"], "finite": None, "mse": None, "mae": None, "cosine": None})
    for prompt in prompts:
        _save_row(path, _generation_row(condition, prompt, identity, attempted=False, completed=False, blocked=True, failure_stage=failure["stage"], failure_class=failure["class"], first_error=failure["first_error"]))


def _protected(values: np.ndarray, shape: tuple[int, int], spec: cross_codec.CodecSpec, payload_factory: Callable[[Image.Image, cross_codec.CodecSpec], tuple[bytes, Image.Image]] | None, output: Path, verified_preflight: Mapping[str, cross_codec.CodecPreflight] | None = None) -> tuple[np.ndarray, dict[str, object], dict[str, object]]:
    """high PNGとlossy lowを個別artifact化し、high exactnessを検証する。"""
    bits = np.zeros(int(np.prod(shape)), dtype=np.uint16)
    bits[: values.size] = values.view(np.uint16)
    high = (bits >> 8).astype(np.uint8).reshape(shape)
    low = bits.astype(np.uint8).reshape(shape)
    high_stream = BytesIO()
    Image.fromarray(high, "L").save(high_stream, format="PNG")
    if payload_factory is None:
        artifact = cross_codec.encode_l_plane(low, spec, verified_preflight=verified_preflight)
        decoded_low = cross_codec.decode_l_plane(artifact, spec, verified_preflight=verified_preflight)
    else:
        artifact, decoded_image = payload_factory(Image.fromarray(low, "L"), spec)
        decoded_low = np.asarray(decoded_image.convert("L"), dtype=np.uint8)
    if decoded_low.shape != shape or not np.array_equal(high, (bits >> 8).astype(np.uint8).reshape(shape)):
        raise ValueError("protected plane invariant failed")
    high_path, low_path = output / "payloads" / f"protected_{spec.name}_high.png", output / "payloads" / f"protected_{spec.name}_low.bin"
    _write(high_path, high_stream.getvalue())
    _write(low_path, artifact)
    persisted = cross_codec.ProtectedArtifact(high_path.read_bytes(), low_path.read_bytes(), spec.name)
    try:
        _, persisted_metadata = cross_codec.decode_protected_high(persisted, shape, values, verified_preflight=verified_preflight)
    except (ValueError, cross_codec.JxlCodecError):
        if payload_factory is None:
            raise
        with Image.open(high_path) as high_image:
            if high_image.format != "PNG" or high_image.mode != "L" or high_image.size != (shape[1], shape[0]):
                raise ValueError("persisted protected high artifact gate failed")
            if not np.array_equal(np.asarray(high_image), high):
                raise ValueError("persisted protected high plane is not exact")
        persisted_metadata = {"decoded_high_exact": True, "fixture_readback": True}
    high_info, low_info = _artifact(high_path, output), _artifact(low_path, output)
    candidate = ((high.astype(np.uint16) << 8) | decoded_low.astype(np.uint16)).reshape(-1).view(np.float16)[: values.size]
    protected = {"high": high_info, "low": low_info, **persisted_metadata, "decoded_high_exact": True, "high_mode": "L", "low_mode": "L", "dimensions": list(shape), "bytes": high_info["bytes"] + low_info["bytes"], "sha256": hashlib.sha256(high_stream.getvalue() + artifact).hexdigest()}
    payload = {"high": high_info, "low": low_info, "bytes": protected["bytes"], "sha256": protected["sha256"]}
    return candidate, payload, protected


def _gate(output: Path, manifest: Mapping[str, object], rows: Sequence[Mapping[str, object]]) -> dict[str, object]:
    """pilot completionをstrict検証してgateを生成する。"""
    reasons: list[str] = []
    prompt_ids = {str(item) for item in manifest["prompt_ids"]}
    layer_ids = {str(item) for item in manifest["layer_ids"]}
    for condition in cross_codec.CONDITION_IDS:
        if not condition_complete(rows, condition, prompt_ids, layer_ids, output):
            reasons.append(f"incomplete:{condition}")
        for row in rows:
            if row.get("condition") == condition and row.get("failure_stage") in {"preflight", "encode", "decode", "decode_invariant", "protected_invariant", "model_load"}:
                reasons.append(f"infrastructure:{condition}:{row['failure_stage']}")
    resource_metadata = validate_resource_monitor(output, manifest, rows)
    _validate_bound_resource(manifest.get("resource_monitor"), resource_metadata)
    manifest_path = output / "manifest.json"
    gate = {"pass": not reasons, "run_identity_hash": manifest["run_identity_hash"], "scientific_identity_hash": manifest["scientific_identity_hash"], "raw_rows_sha256": sha256_file(output / "results" / "raw_rows.jsonl"), "raw_rows_bytes": (output / "results" / "raw_rows.jsonl").stat().st_size, "manifest_sha256": sha256_file(manifest_path), "prompt_ids": list(manifest["prompt_ids"]), "conditions": list(cross_codec.CONDITION_IDS), "scope": manifest["inference_scope"], "codec_identities": manifest["codec_preflight"], "resource_monitor": manifest.get("resource_monitor", resource_metadata), "reasons": sorted(set(reasons))}
    gate["gate_hash"] = _hash(gate)
    atomic_json(output / "pilot_gate.json", gate)
    return gate


def validate_pilot_gate(path: str | Path, *, expected_profile: str = "pilot") -> dict[str, object]:
    """gate、raw rows、pilot manifestをrehashして検証する。"""
    gate_path = Path(path)
    if gate_path.name == "manifest.json":
        manifest_path = gate_path
        gate_path = gate_path.parent / "pilot_gate.json"
    else:
        manifest_path = gate_path.parent / "manifest.json"
    gate = json.loads(gate_path.read_text(encoding="utf-8"))
    recorded = gate.pop("gate_hash", None)
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if recorded != _hash(gate) or not gate.get("pass") or manifest.get("status") != "completed" or manifest.get("profile") != expected_profile or manifest.get("mode") != "real":
        raise ValueError("pilot gate is missing, failed, or tampered")
    if manifest.get("profile") != "pilot" or manifest.get("steps") != 50 or manifest.get("cfg") != 6.0 or manifest.get("requested_device") != "mps" or len(manifest.get("prompt_ids", [])) != 8 or manifest.get("conditions") != list(cross_codec.CONDITION_IDS):
        raise ValueError("pilot gate scope mismatch")
    raw_path = manifest_path.parent / "results" / "raw_rows.jsonl"
    if gate.get("raw_rows_sha256") != sha256_file(raw_path) or gate.get("raw_rows_bytes") != raw_path.stat().st_size or gate.get("manifest_sha256") != sha256_file(manifest_path):
        raise ValueError("pilot artifact hash mismatch")
    if gate.get("conditions") != list(cross_codec.CONDITION_IDS):
        raise ValueError("pilot condition registry mismatch")
    rows = load_rows_strict(raw_path, str(manifest["run_identity_hash"]))
    resource_metadata = validate_resource_monitor(manifest_path.parent, manifest, rows)
    if gate.get("resource_monitor") != manifest.get("resource_monitor"):
        raise ValueError("pilot resource monitor provenance mismatch")
    _validate_bound_resource(manifest.get("resource_monitor"), resource_metadata)
    if gate.get("prompt_ids") != manifest.get("prompt_ids") or gate.get("scope") != manifest.get("inference_scope") or gate.get("scientific_identity_hash") != manifest.get("scientific_identity_hash") or scientific_identity_hash(manifest["run_identity"]) != manifest.get("scientific_identity_hash"):
        raise ValueError("pilot gate scope mismatch")
    for condition in cross_codec.CONDITION_IDS:
        if not condition_complete(rows, condition, {str(item) for item in manifest["prompt_ids"]}, {str(item) for item in manifest["layer_ids"]}, manifest_path.parent):
            raise ValueError("pilot gate artifact or condition completion mismatch")
    return {"gate": gate, "manifest": manifest}


def run(args: argparse.Namespace, *, preflight_factory: Callable[[], Sequence[cross_codec.CodecPreflight]] | None = None, payload_factory: Callable[[Image.Image, cross_codec.CodecSpec], tuple[bytes, Image.Image]] | None = None, runtime_factory: Callable[[argparse.Namespace, str], tuple[Any, Any]] | None = None, candidate_factory: Callable[[str, np.ndarray], np.ndarray] | None = None) -> Path:
    """cross-codec experimentを実行する。"""
    injected = any(item is not None for item in (preflight_factory, payload_factory, runtime_factory, candidate_factory))
    _validate(args, injected)
    output = Path(args.output)
    source_path, prompt_path = Path(args.image), Path(args.prompts_manifest)
    preflight_snapshot = _early_resource_preflight(output)
    if not source_path.is_file() or not prompt_path.is_file():
        raise FileNotFoundError("image or prompt manifest")
    statuses = tuple(preflight_factory() if preflight_factory else cross_codec.preflight_codecs())
    names = tuple(item.name for item in statuses)
    if names != tuple(cross_codec.CODEC_SPECS) or len(set(names)) != len(names) or not all(item.available for item in statuses):
        raise RuntimeError("codec preflight unavailable or noncanonical; fail closed")
    verified_preflight = MappingProxyType({item.name: item for item in statuses})
    raw_prompts = json.loads(prompt_path.read_text(encoding="utf-8"))
    if not isinstance(raw_prompts, list):
        raise TypeError("prompt manifest must be a list")
    prompts = select_prompts(raw_prompts, args.profile)
    if len(prompts) != args.prompts:
        raise ValueError(f"{args.profile} requires exactly {args.prompts} selected prompts")
    resolved = args.device if injected else resolve_device(args.device)
    if resolved != "mps":
        raise ValueError("resolved device must be mps")
    source_image = np.asarray(Image.open(source_path).convert("RGB"), dtype=np.uint8).copy()
    source_values, layers = _source_values(source_image, json.loads(Path(args.manifest).read_text(encoding="utf-8")))
    identity = _safe({"source_sha256": sha256_file(source_path), "source_bytes": source_path.stat().st_size, "model_sha256": sha256_file(args.model) if Path(args.model).exists() else None, "config_sha256": sha256_file(args.config), "prompt_sha256": sha256_file(prompt_path), "upstream_revision": UPSTREAM_REVISION, "code_hashes": {"runner": sha256_file(__file__), "cross_codec": sha256_file(cross_codec.__file__), "jpeg_repair": sha256_file(jpeg_repair.__file__)}, "runtime_git": _git(Path(__file__).resolve().parent.parent), "runtime_platform": {"system": platform.system(), "release": platform.release(), "machine": platform.machine(), "python": platform.python_version()}, "package_versions": {}, "vae": {"requested": args.vae, "revision": VAE_REVISION}, "clip": {"requested": args.clip, "revision": CLIP_REVISION}, "requested_device": args.device, "resolved_device": resolved, "mode": "fixture" if args.fixture else "fake" if args.fake_runtime else "real", "steps": args.steps, "cfg": args.cfg, "profile": args.profile, "prompt_ids": [str(item["prompt_id"]) for item in prompts], "prompt_selection_hash": _hash(prompts), "condition_registry_hash": cross_codec.semantic_hash(cross_codec.CONDITION_REGISTRY), "codec_preflight": [_safe(item) for item in statuses]})
    identity["scientific_identity_hash"] = scientific_identity_hash(identity)
    run_hash = _hash(identity)
    if args.profile == "full":
        gate_data = validate_pilot_gate(args.pilot_manifest)
        if injected:
            raise ValueError("full profile does not accept injected fixture runtime")
        pilot_manifest = gate_data["manifest"]
        comparisons = {"source_sha256": identity["source_sha256"], "model_sha256": identity["model_sha256"], "config_sha256": identity["config_sha256"], "prompt_sha256": identity["prompt_sha256"], "condition_registry_hash": identity["condition_registry_hash"], "codec_preflight": identity["codec_preflight"], "steps": args.steps, "cfg": args.cfg, "requested_device": args.device}
        if identity.get("mode") != "real" or identity.get("scientific_identity_hash") != pilot_manifest.get("scientific_identity_hash") or scientific_identity_hash(pilot_manifest["run_identity"]) != pilot_manifest.get("scientific_identity_hash"):
            raise ValueError("pilot/full scientific identity mismatch")
        for key, value in comparisons.items():
            if key in pilot_manifest and pilot_manifest[key] != value:
                raise ValueError(f"pilot/full identity mismatch: {key}")
    existing = None
    manifest_path = output / "manifest.json"
    if manifest_path.exists():
        existing = json.loads(manifest_path.read_text(encoding="utf-8"))
        if existing.get("run_identity") != identity or existing.get("run_identity_hash") != run_hash:
            raise ValueError("resume identity mismatch")
        if existing.get("status") == "completed":
            raise ValueError("completed run cannot resume")
        if existing.get("status") not in {"running", "resource_guard_failed"}:
            raise ValueError("invalid resumable manifest state")
    elif output.exists() and any(output.iterdir()):
        raise ValueError("nonempty output without manifest is rejected")
    output.mkdir(parents=True, exist_ok=True)
    lock = output / "run.lock"
    _acquire_lock(lock, run_hash, args.recover_lock)
    try:
        manifest = existing or {"status": "running", "run_identity": identity, "run_identity_hash": run_hash, **identity, "source_bytes": source_path.stat().st_size, "inference_scope": {"profile": args.profile, "steps": args.steps, "cfg": args.cfg, "requested_device": args.device, "resolved_device": resolved}, "prompt_ids": [str(item["prompt_id"]) for item in prompts], "conditions": list(cross_codec.CONDITION_IDS), "layer_ids": list(layers), "started_at": datetime.now(UTC).isoformat()}
        atomic_json(manifest_path, manifest)
        raw_rows_path = output / "results" / "raw_rows.jsonl"
        _save_row(output / "results" / "resource_monitor.jsonl", {**preflight_snapshot, "row_type": "resource_guard", "stage": "preflight", "run_identity_hash": run_hash})
        rows = load_rows_strict(raw_rows_path, run_hash)
        runtime_loader, runtime = runtime_factory(args, resolved) if runtime_factory else build_official_runtime(args, resolved)
        payloads: dict[str, tuple[bytes, Image.Image]] = {}
        payload_failures: dict[str, dict[str, object]] = {}
        for codec_name in ("webp_lossless", *cross_codec.LOSSY_CODEC_NAMES):
            spec = cross_codec.CODEC_SPECS[codec_name]
            try:
                encoded = payload_factory(Image.fromarray(source_image), spec) if payload_factory else (cross_codec.encode_rgb(Image.fromarray(source_image), spec, verified_preflight=verified_preflight), None)
            except Exception as exc:  # noqa: BLE001
                payload_failures[codec_name] = {"class": type(exc).__name__, "reason": str(exc), "first_error": f"{type(exc).__name__}: {exc}", "stage": "encode", "codec": codec_name}
                continue
            try:
                decoded = encoded[1] if encoded[1] is not None else cross_codec.decode_rgb(encoded[0], spec, verified_preflight=verified_preflight)
                if decoded.size != source_image.shape[1::-1]:
                    raise ValueError("decoded dimensions mismatch")
                payloads[codec_name] = (encoded[0], decoded)
            except Exception as exc:  # noqa: BLE001
                payload_failures[codec_name] = {"class": type(exc).__name__, "reason": str(exc), "first_error": f"{type(exc).__name__}: {exc}", "stage": "decode", "codec": codec_name}
        baseline_path = output / "weight_images" / "png_baseline.png"
        if not baseline_path.exists():
            _write(baseline_path, source_path.read_bytes())
        for condition in cross_codec.CONDITION_IDS:
            prompt_ids = {str(item["prompt_id"]) for item in prompts}
            if condition_complete(rows, condition, prompt_ids, set(layers), output):
                continue
            condition_rows = [row for row in rows if row.get("condition") == condition]
            old_weight = next((row for row in condition_rows if row.get("row_type") == "weight"), None)
            spec = cross_codec.CONDITION_REGISTRY[condition]
            candidate = source_values.copy()
            payload_bytes = source_path.read_bytes()
            protected = None
            repaired_count = 0
            blocked = False
            failure_stage = None
            failure_reason = None
            if old_weight:
                if not _verify_artifact_tree(output, old_weight.get("payload"), condition + ".payload") or not _verify_artifact_tree(output, old_weight.get("protected"), condition + ".protected") or not _verify_artifact_tree(output, old_weight.get("materialized"), condition + ".materialized"):
                    raise ValueError("artifact tamper on resume")
            else:
                codec_failure = payload_failures.get(spec.codec)
                if codec_failure is not None and condition != "png_baseline":
                    _record_codec_failure(raw_rows_path, condition, spec, prompts, layers, run_hash, codec_failure)
                    rows = load_rows_strict(raw_rows_path, run_hash)
                    continue
                if condition == "png_baseline":
                    payload_bytes = source_path.read_bytes()
                elif condition == "webp_lossless":
                    payload_bytes, decoded_image = payloads["webp_lossless"]
                    if not np.array_equal(np.asarray(decoded_image.convert("RGB")), source_image):
                        failure = {"class": "BitExactnessError", "reason": "WebP lossless decoded pixels differ from official RGB source", "first_error": "BitExactnessError: WebP lossless decoded pixels differ from official RGB source", "stage": "decode_invariant", "codec": "webp_lossless"}
                        _record_codec_failure(raw_rows_path, condition, spec, prompts, layers, run_hash, failure)
                        rows = load_rows_strict(raw_rows_path, run_hash)
                        continue
                    pixels = np.asarray(decoded_image.convert("RGB"), dtype=np.uint8).reshape(-1, 3)[: source_values.size]
                    candidate = ((pixels[:, 0].astype(np.uint16) << 8) | pixels[:, 1].astype(np.uint16)).view(np.float16)
                else:
                    payload_bytes, decoded_image = payloads[spec.codec]
                    pixels = np.asarray(decoded_image.convert("RGB"), dtype=np.uint8).reshape(-1, 3)[: source_values.size]
                    candidate = ((pixels[:, 0].astype(np.uint16) << 8) | pixels[:, 1].astype(np.uint16)).view(np.float16)
                    if candidate_factory:
                        candidate_condition = condition.replace("_repair_zero", "_raw")
                        candidate = np.asarray(candidate_factory(candidate_condition, candidate), dtype=np.float16)
                    if spec.protected:
                        try:
                            candidate, protected, protected_meta = _protected(source_values, source_image.shape[:2], cross_codec.CODEC_SPECS[spec.codec], payload_factory, output, verified_preflight)
                        except Exception as exc:  # noqa: BLE001
                            failure = {"class": type(exc).__name__, "reason": str(exc), "first_error": f"{type(exc).__name__}: {exc}", "stage": "protected_invariant", "codec": spec.codec}
                            _record_codec_failure(raw_rows_path, condition, spec, prompts, layers, run_hash, failure)
                            rows = load_rows_strict(raw_rows_path, run_hash)
                            continue
                        payload_bytes = output.joinpath(protected["high"]["path"]).read_bytes() + output.joinpath(protected["low"]["path"]).read_bytes()
                        protected["metadata"] = protected_meta
                    elif spec.repair:
                        repaired = repair_nonfinite(candidate, "zero", layers)
                        repaired_count = repaired.repaired_value_count
                        candidate = repaired.values
                    if spec.mode == "raw" and not np.isfinite(candidate).all():
                        blocked, failure_stage, failure_reason = True, "weight_gate", "raw candidate contains nonfinite values"
                    elif not np.isfinite(candidate).all():
                        blocked, failure_stage, failure_reason = True, "weight_gate", "candidate contains nonfinite values"
                payload_path = output / "payloads" / f"{condition}.bin"
                _write(payload_path, payload_bytes)
                payload_info = _artifact(payload_path, output)
                materialized = None
                if condition == "png_baseline":
                    materialized = _artifact(baseline_path, output)
                elif not blocked:
                    materialized_path = output / "weight_images" / f"{condition}.png"
                    _write(materialized_path, materialize_fp16_png(candidate, width=source_image.shape[1]))
                    materialized = _artifact(materialized_path, output)
                metric = _metrics(source_values, candidate, layers)
                png_bytes = source_path.stat().st_size
                weight = {"row_type": "weight", "stage": "weight", "condition": condition, "prompt_id": None, "run_identity_hash": run_hash, "codec": spec.codec, "mode": spec.mode, "attempted": not blocked, "status": "weight_gate_blocked" if blocked else "completed", "blocked": blocked, "failure_stage": failure_stage, "failure_class": None, "failure_reason": failure_reason, "first_failure": failure_reason, "repaired_count": repaired_count, "payload": protected or payload_info, "protected": protected, "materialized": materialized, "payload_bytes": len(payload_bytes), "payload_sha256": hashlib.sha256(payload_bytes).hexdigest(), "payload_over_png": len(payload_bytes) / png_bytes if png_bytes else None, "png_over_payload": png_bytes / len(payload_bytes) if payload_bytes else None, "size_saving_percent": (1 - len(payload_bytes) / png_bytes) * 100 if png_bytes else None, **metric}
                _save_row(raw_rows_path, weight)
                for layer, (start, end) in layers.items():
                    _save_row(raw_rows_path, {"row_type": "layer", "stage": "layer", "condition": condition, "layer": layer, "prompt_id": None, "run_identity_hash": run_hash, **_metrics(source_values[start:end], candidate[start:end], {layer: (0, end - start)}), "repaired_count": repaired_count})
                rows = load_rows_strict(raw_rows_path, run_hash)
                old_weight = weight
            if old_weight and old_weight.get("blocked"):
                for prompt in prompts:
                    if any(row.get("row_type") == "generation" and row.get("condition") == condition and row.get("prompt_id") == str(prompt["prompt_id"]) for row in rows):
                        continue
                    row = _generation_row(condition, prompt, run_hash, blocked=True, failure_stage=old_weight.get("failure_stage") or "weight_gate", first_error=old_weight.get("failure_reason"))
                    _save_row(raw_rows_path, row)
                rows = load_rows_strict(raw_rows_path, run_hash)
                continue
            materialized = old_weight.get("materialized") if old_weight else None
            model = None
            try:
                model = runtime_loader(str(output / materialized["path"])) if runtime_loader and materialized else None
            except Exception as exc:  # noqa: BLE001
                for prompt in prompts:
                    if any(row.get("row_type") == "generation" and row.get("condition") == condition and row.get("prompt_id") == str(prompt["prompt_id"]) for row in rows):
                        continue
                    _save_row(raw_rows_path, _generation_row(condition, prompt, run_hash, blocked=True, failure_stage="model_load", failure_class=type(exc).__name__, first_error=f"{type(exc).__name__}: {exc}"))
                rows = load_rows_strict(raw_rows_path, run_hash)
                continue
            consecutive = 0
            for prompt in prompts:
                if any(row.get("row_type") == "generation" and row.get("condition") == condition and row.get("prompt_id") == str(prompt["prompt_id"]) for row in rows):
                    continue
                resource_path = output / "results" / "resource_monitor.jsonl"
                existing_resource = [
                    item for item in load_rows_strict(resource_path, run_hash)
                    if item.get("row_type") == "resource" and item.get("stage") == "resource" and item.get("condition") == condition and str(item.get("prompt_id")) == str(prompt["prompt_id"])
                ]
                if len(existing_resource) > 1:
                    raise ValueError("conflicting resource monitor pair on resume")
                snapshot = dict(existing_resource[0]) if existing_resource else sample_resource(output, condition, str(prompt["prompt_id"]))
                violation = _resource_violation(snapshot)
                snapshot["violation_reason"] = violation
                _save_row(resource_path, {**snapshot, "row_type": "resource", "stage": "resource", "run_identity_hash": run_hash})
                consecutive = consecutive + 1 if violation else 0
                if consecutive >= 3:
                    manifest.update({"status": "resource_guard_failed", "resource_retry_count": int(manifest.get("resource_retry_count", 0)) + 1})
                    atomic_json(manifest_path, manifest)
                    raise ResourceGuardError(violation or "resource guard")
                started = time.perf_counter()
                row = _generation_row(condition, prompt, run_hash)
                try:
                    image, score = runtime.generate(str(prompt["prompt"]), int(prompt["initial_latent_seed"])) if hasattr(runtime, "generate") else generate_with_runtime(runtime, model, str(prompt["prompt"]), int(prompt["initial_latent_seed"]), args.steps, args.cfg, resolved)
                    if score is None or not math.isfinite(float(score)):
                        raise FloatingPointError("CLIP score is missing or nonfinite")
                    sample = _save_sample(image, output / "samples" / condition / f"{prompt['prompt_id']}.png", base=output)
                    row.update({"attempted": True, "attempt_count": 1, "completed": True, "success": True, "clip_score": float(score), "elapsed_seconds": time.perf_counter() - started, "sample_path": sample["path"], "sample_hash": sample["sha256"], "sample_bytes": sample["bytes"]})
                except Exception as exc:  # noqa: BLE001
                    row.update({"attempted": True, "attempt_count": 1, "completed": True, "failure_stage": "generation", "failure_class": type(exc).__name__, "first_error": f"{type(exc).__name__}: {exc}", "elapsed_seconds": time.perf_counter() - started})
                _save_row(raw_rows_path, row)
                rows = load_rows_strict(raw_rows_path, run_hash)
            del model
        rows = load_rows_strict(raw_rows_path, run_hash)
        pipeline_failure = any(row.get("row_type") == "weight" and row.get("status") == "codec_pipeline_failed" for row in rows)
        complete = all(condition_complete(rows, condition, {str(item["prompt_id"]) for item in prompts}, set(layers), output) for condition in cross_codec.CONDITION_IDS)
        if not complete and not pipeline_failure:
            raise ValueError("completion verification failed")
        resource_metadata = validate_resource_monitor(output, manifest, rows)
        manifest.update({"status": "completed_with_pipeline_failures" if pipeline_failure else "completed", "completed_at": datetime.now(UTC).isoformat(), "raw_rows_sha256": sha256_file(raw_rows_path), "raw_rows_bytes": raw_rows_path.stat().st_size, "resource_monitor": {**resource_metadata, "binding_method": "run_completion"}})
        atomic_json(manifest_path, manifest)
        if args.profile == "pilot":
            _gate(output, manifest, rows)
        return output
    finally:
        if lock.exists():
            lock.unlink()


def bind_completed_resource_monitor(output: str | Path, source_image: Path) -> Path:
    """既存completed runへresource monitor provenanceだけを後付けする。"""
    root = Path(output).resolve()
    manifest_path = root / "manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if manifest.get("status") != "completed" or manifest.get("profile") not in {"pilot", "full"}:
        raise ValueError("resource migration requires a completed pilot/full run")
    raw_path = root / "results" / "raw_rows.jsonl"
    rows = load_rows_strict(raw_path, str(manifest.get("run_identity_hash")))
    prompt_ids = {str(item) for item in manifest.get("prompt_ids", [])}
    layer_ids = {str(item) for item in manifest.get("layer_ids", [])}
    for condition in cross_codec.CONDITION_IDS:
        if not condition_complete(rows, condition, prompt_ids, layer_ids, root):
            raise ValueError(f"completed row contract failed: {condition}")
    source_path = source_image
    if not source_path.is_file():
        raise ValueError("source image is missing")
    if sha256_file(source_path) != manifest.get("source_sha256"):
        raise ValueError("source image hash mismatch")
    source_bytes = source_path.stat().st_size
    if manifest.get("source_bytes") is not None and manifest.get("source_bytes") != source_bytes:
        raise ValueError("source image bytes mismatch")
    resource = validate_resource_monitor(root, manifest, rows)
    binding_code_hash = sha256_file(__file__)
    existing = manifest.get("resource_monitor")
    if existing is not None:
        if not isinstance(existing, Mapping) or existing.get("binding_method") not in {"post_run_verified", "run_completion"}:
            raise ValueError("resource monitor is already bound with a mismatch")
        if any(existing.get(key) != value for key, value in resource.items()) or existing.get("bytes") != resource.get("bytes"):
            raise ValueError("resource monitor is already bound with a mismatch")
        if manifest.get("source_bytes") != source_bytes:
            raise ValueError("resource monitor is already bound with a mismatch")
        return root
    bound = {**resource, "binding_method": "post_run_verified", "bound_at": datetime.now(UTC).isoformat(), "binding_code_hash": binding_code_hash}
    manifest["source_bytes"] = source_bytes
    manifest["resource_monitor"] = bound
    atomic_json(manifest_path, manifest)
    if manifest.get("profile") == "pilot":
        _gate(root, manifest, rows)
    return root


def main() -> None:
    """CLI entrypoint。"""
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    args = build_parser().parse_args()
    if args.bind_resource_monitor:
        bind_completed_resource_monitor(args.output, Path(args.image))
        return
    run(args)


if __name__ == "__main__":
    main()