fix(perception): isolate vegetation qualification IO

experiments/perception/worker/Invoke-M49TgsIntegratedGraphShadow.ps1

@@ -205,6 +205,10 @@ foreach ($entry in $source.GetEnumerator()) {
 if ((Get-Sha256 $source.SourcePack) -cne [string]$releaseDocument.source_pack_sha256) {
     throw "RAVNOVES00 source pack digest changed"
 }
+$videoSha256 = [string]$releaseDocument.video_sha256
+if ((Get-Sha256 $source.Video) -cne $videoSha256) {
+    throw "RAVNOVES00 video digest changed"
+}
 
 $vegetation = $null
 if ($VegetationLoadGate) {
@@ -405,6 +409,8 @@ try {
             --checkpoint /models/candidate.pth `
             --dataset-root /data/goose `
             --video /source/right.mp4 `
+            --video-sha256 $videoSha256 `
+            --runtime-video-cache /tmp/vegetation-right.mp4 `
             --source-rate-hz $rate `
             --minimum-effective-fps 11.209069 `
             --maximum-completion-p95-ms 125.0 `

experiments/perception/worker/lab_v1_vegetation_goose/run_vegetation_integrated_load.py

@@ -7,6 +7,7 @@ import argparse
 import json
 import math
 import platform
+import shutil
 import statistics
 import time
 from pathlib import Path
@@ -53,6 +54,8 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--checkpoint", type=Path, required=True)
     parser.add_argument("--dataset-root", type=Path, required=True)
     parser.add_argument("--video", type=Path, required=True)
+    parser.add_argument("--video-sha256", required=True)
+    parser.add_argument("--runtime-video-cache", type=Path, required=True)
     parser.add_argument("--source-rate-hz", type=float, required=True)
     parser.add_argument("--minimum-effective-fps", type=float, required=True)
     parser.add_argument("--maximum-completion-p95-ms", type=float, required=True)
@@ -110,6 +113,28 @@ def validate_sha256(value: str, label: str) -> None:
         raise IntegratedLoadError(f"{label} SHA-256 is invalid")
 
 
+def buffer_compressed_video(source: Path, target: Path, expected_sha256: str) -> dict[str, Any]:
+    if source.is_symlink() or not source.is_file():
+        raise IntegratedLoadError("RAVNOVES video is unavailable")
+    if target.exists() or target.is_symlink():
+        raise IntegratedLoadError("RAVNOVES runtime video cache already exists")
+    if not target.parent.is_dir():
+        raise IntegratedLoadError("RAVNOVES runtime video cache parent is unavailable")
+    started = time.monotonic_ns()
+    shutil.copyfile(source, target)
+    copied_bytes = target.stat().st_size
+    if copied_bytes != source.stat().st_size:
+        raise IntegratedLoadError("RAVNOVES runtime video cache size changed")
+    copied_sha256 = sha256(target)
+    if copied_sha256 != expected_sha256:
+        raise IntegratedLoadError("RAVNOVES runtime video cache digest changed")
+    return {
+        "bytes": copied_bytes,
+        "sha256": copied_sha256,
+        "seconds": round((time.monotonic_ns() - started) / 1_000_000_000.0, 6),
+    }
+
+
 def run() -> int:
     args = parse_args()
     if not torch.cuda.is_available():
@@ -123,6 +148,7 @@ def run() -> int:
     ):
         raise IntegratedLoadError("integrated-load thresholds must be positive and finite")
     validate_sha256(args.release_sha256, "release")
+    validate_sha256(args.video_sha256, "video")
     if args.output.exists() or args.frame_ledger.exists():
         raise IntegratedLoadError("integrated-load output already exists")
 
@@ -143,7 +169,10 @@ def run() -> int:
         config["ravnoves"]["expected_width"],
         config["ravnoves"]["expected_height"],
     )
-    warmup_capture = open_video(args.video)
+    compressed_video_buffer = buffer_compressed_video(
+        args.video, args.runtime_video_cache, args.video_sha256
+    )
+    warmup_capture = open_video(args.runtime_video_cache)
     warmup_source = decode_source(warmup_capture, expected_size)
     warmup_capture.release()
 
@@ -152,7 +181,7 @@ def run() -> int:
     warmup_tensor, _ = preprocess(warmup_source)
     warmup_latencies_ms = [infer(model, warmup_tensor)[1] for _ in range(3)]
     torch.cuda.reset_peak_memory_stats()
-    source_capture = open_video(args.video)
+    source_capture = open_video(args.runtime_video_cache)
     wait_for_shared_start(
         args.shared_start_ready_file,
         args.shared_start_file,
@@ -245,7 +274,9 @@ def run() -> int:
             "capacity_drop_count": 0,
             "deadline_miss_count": late_deadline_count,
             "source_decode": {
-                "mode": "bounded-sequential-h264/v1",
+                "mode": "bounded-compressed-scene-buffer/v1",
+                "compressed_scene_prefetch": True,
+                "compressed_scene_buffer": compressed_video_buffer,
                 "full_route_rgb_prefetch": False,
                 "candidate_local_decoder": True,
                 "runtime_target": "shared-source-frame",