| """Validate a Simam3D run manifest and its exported artifact fingerprints.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import sys |
| from pathlib import Path |
|
|
|
|
| def file_sha256(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def validate_manifest(manifest_path: str | Path) -> tuple[bool, list[str]]: |
| manifest_path = Path(manifest_path) |
| errors: list[str] = [] |
| try: |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) |
| except (OSError, json.JSONDecodeError) as exc: |
| return False, [f"manifest unreadable: {exc}"] |
| if manifest.get("project") != "Simam3D": |
| errors.append("project is not Simam3D") |
| if int(manifest.get("manifest_schema_version", 0)) < 3: |
| errors.append("manifest schema is older than provenance-contract schema 3") |
| if manifest.get("status") == "failed": |
| required = {"run_id", "backend", "runtime", "stage", "error", "input_sha256"} |
| missing = sorted(key for key in required if key not in manifest) |
| if missing: |
| errors.append(f"failure manifest missing fields: {', '.join(missing)}") |
| if not isinstance(manifest.get("error"), dict): |
| errors.append("failure manifest error field is invalid") |
| elif not all(manifest["error"].get(key) for key in ("type", "message")): |
| errors.append("failure manifest error type/message is incomplete") |
| claims = manifest.get("claims") |
| if not isinstance(claims, dict) or claims.get("failure_report_only") is not True: |
| errors.append("failure manifest must declare failure_report_only") |
| return not errors, errors |
| claims = manifest.get("claims") |
| required_claims = { |
| "observed_input_geometry", |
| "generated_novel_views", |
| "metric_scale_reconstruction", |
| "learned_gaussian_optimization", |
| "reveal_content_is_verified", |
| } |
| if not isinstance(claims, dict) or not required_claims.issubset(claims): |
| errors.append("claims provenance contract is missing or incomplete") |
| outputs = manifest.get("outputs") |
| hashes = manifest.get("output_sha256") |
| if not isinstance(outputs, list) or not isinstance(hashes, dict): |
| return False, errors + ["outputs/output_sha256 fields are missing or invalid"] |
| for raw_output in outputs: |
| output = Path(str(raw_output)) |
| if not output.is_absolute() and not output.is_file(): |
| sibling = manifest_path.parent / output.name |
| if sibling.is_file(): |
| output = sibling |
| if not output.is_file(): |
| errors.append(f"missing artifact: {raw_output}") |
| continue |
| expected = hashes.get(output.name) |
| if not expected: |
| errors.append(f"missing hash: {output.name}") |
| continue |
| actual = file_sha256(output) |
| if actual != expected: |
| errors.append(f"hash mismatch: {output.name}") |
| return not errors, errors |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| argv = argv or sys.argv[1:] |
| if len(argv) != 1: |
| print("usage: python validate_manifest.py PATH_TO_RUN_JSON", file=sys.stderr) |
| return 2 |
| valid, errors = validate_manifest(argv[0]) |
| if valid: |
| print("VALID: all Simam3D artifacts match manifest fingerprints") |
| return 0 |
| print("INVALID") |
| for error in errors: |
| print(f"- {error}") |
| return 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
