Recover governed runtime provenance for legacy YOLO models

2026-08-23 23:22:02 +02:00
parent 300fbba5c9
commit be2e092b33
8 changed files with 588 additions and 0 deletions
@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+from hashlib import sha256
+import importlib.util
+import json
+from pathlib import Path
+from uuid import uuid4
+
+from app.services.runtime_model_provenance_service import RuntimeModelProvenanceService
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / "scripts" / "migrate_runtime_model_provenance.py"
+SPEC = importlib.util.spec_from_file_location("migrate_runtime_model_provenance", SCRIPT)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+def _sha(value: bytes) -> str:
+    return sha256(value).hexdigest()
+
+
+def _args(tmp_path: Path):
+    model = tmp_path / "active.pt"
+    checkpoint = tmp_path / "best.pt"
+    base_model = tmp_path / "base.pt"
+    training_args = tmp_path / "args.yaml"
+    training_results = tmp_path / "results.csv"
+    dataset_summary = tmp_path / "dataset-summary.json"
+    dataset_yaml = tmp_path / "dataset.yaml"
+    training_summary = tmp_path / "training-summary.json"
+
+    model.write_bytes(b"exact promoted model bytes")
+    checkpoint.write_bytes(model.read_bytes())
+    base_model.write_bytes(b"exact base model bytes")
+    training_args.write_text("epochs: 30\nseed: 0\n", encoding="utf-8")
+    training_results.write_text("epoch,metric\n1,0.1\n", encoding="utf-8")
+    dataset_yaml.write_text("names:\n  0: building\n", encoding="utf-8")
+    dataset_summary.write_text(
+        json.dumps(
+            {
+                "status": "ok",
+                "class_names": ["building"],
+                "tile_count": 198,
+                "train_tile_count": 180,
+                "val_tile_count": 18,
+                "label_count": 58_820,
+            }
+        ),
+        encoding="utf-8",
+    )
+    training_summary.write_text(
+        json.dumps(
+            {
+                "status": "ok",
+                "trained_model_sha256": _sha(model.read_bytes()),
+                "base_model_sha256": _sha(base_model.read_bytes()),
+                "dataset_summary_sha256": _sha(dataset_summary.read_bytes()),
+                "dataset_yaml_sha256": _sha(dataset_yaml.read_bytes()),
+            }
+        ),
+        encoding="utf-8",
+    )
+    return module.parse_args(
+        [
+            "--model-path",
+            str(model),
+            "--checkpoint-path",
+            str(checkpoint),
+            "--base-model-path",
+            str(base_model),
+            "--training-summary-path",
+            str(training_summary),
+            "--training-args-path",
+            str(training_args),
+            "--training-results-path",
+            str(training_results),
+            "--dataset-summary-path",
+            str(dataset_summary),
+            "--dataset-yaml-path",
+            str(dataset_yaml),
+            "--source-version",
+            "sprint174-smallbld-minpx3-img640-ft30",
+            "--framework-version",
+            "8.4.93",
+        ]
+    )
+
+
+def test_recovered_evidence_requires_byte_identical_checkpoint_and_recorded_hashes(tmp_path: Path) -> None:
+    args = _args(tmp_path)
+
+    evidence = module.inspect_evidence(args)
+
+    assert evidence["checksums"]["model"] == evidence["checksums"]["checkpoint"]
+    assert evidence["checksums"]["training_summary"] == _sha(
+        Path(args.training_summary_path).read_bytes()
+    )
+    assert evidence["class_mapping"] == {"0": "building"}
+
+
+def test_recovered_evidence_rejects_changed_checkpoint(tmp_path: Path) -> None:
+    args = _args(tmp_path)
+    Path(args.checkpoint_path).write_bytes(b"other checkpoint")
+
+    exit_code, payload = module.migrate(args)
+
+    assert exit_code == 2
+    assert payload["status"] == "evidence_invalid"
+    assert "checkpoint/model SHA-256 mismatch" in payload["message"]
+
+
+def test_generated_sidecar_passes_exact_runtime_contract(tmp_path: Path) -> None:
+    args = _args(tmp_path)
+    evidence = module.inspect_evidence(args)
+    payload = module._manifest_payload(
+        args=args,
+        evidence=evidence,
+        source_registry_id=str(uuid4()),
+        source_snapshot_id=str(uuid4()),
+        imported_at="2026-08-23T21:00:00+00:00",
+    )
+    manifest_path = RuntimeModelProvenanceService.manifest_path_for_model(args.model_path)
+    assert module._write_manifest_atomically(manifest_path, payload) is True
+
+    validated = RuntimeModelProvenanceService.validate_for_runtime(
+        model_path=args.model_path,
+        model_id="yolo-configured",
+        task_type="object_detection",
+        expected_model_version="sprint174-smallbld-minpx3-img640-ft30",
+        allowed_frameworks=("ultralytics/pytorch",),
+    )
+
+    assert validated.model_sha256 == evidence["checksums"]["model"]
+    assert validated.runtime_manifest_sha256 == payload["metadata"]["runtime_manifest_sha256"]
+    assert module._write_manifest_atomically(manifest_path, payload) is False