| """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: |
| 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: |
| 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: |
| 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: |
| 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: |
| 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() |
|
|
