fix(perception): bound vegetation load decoding

config/perception/lab-v1-vegetation-integrated-shadow-v1.json

@@ -50,9 +50,12 @@
     "reference_graph_parameters_unchanged": true,
     "tgs_parameters_unchanged": true,
     "ddrnet_parameters_unchanged": true,
+    "vegetation_source_buffer_bounded": true,
+    "vegetation_full_route_rgb_prefetch_allowed": false,
     "ppliteseg_concurrent_run_allowed": false,
     "camera_semantics_can_clear_rigid_geometry": false,
     "canonical_triton_mutation_allowed": false,
+    "runtime_shared_source_frame_target": true,
     "gauss_or_playcanvas_in_scope": false
   },
   "authority": {

experiments/perception/worker/Invoke-M49TgsIntegratedGraphShadow.ps1

@@ -288,24 +288,6 @@ foreach ($name in $containers) {
 
 $started = [DateTimeOffset]::UtcNow
 try {
-    if ($VegetationLoadGate) {
-        $vegetationFrames = Join-Path $runOutput "vegetation\input-frames"
-        $null = New-Item -ItemType Directory -Path $vegetationFrames
-        & ffmpeg -hide_banner -loglevel error -i $source.Video -map 0:v:0 -fps_mode passthrough (
-            Join-Path $vegetationFrames "frame-%06d.png"
-        )
-        Assert-LastExitCode "RAVNOVES full-video frame extraction"
-        $extractedFrames = @(
-            Get-ChildItem -LiteralPath $vegetationFrames -File -Filter "frame-*.png" |
-                Sort-Object Name
-        )
-        if (
-            $extractedFrames.Count -ne 4489 -or
-            $extractedFrames[0].Name -cne "frame-000001.png" -or
-            $extractedFrames[-1].Name -cne "frame-004489.png"
-        ) { throw "RAVNOVES full-video frame sequence changed" }
-    }
-
     & docker run --rm --name $prepareName --network none --cpus 8 --memory 16g `
         --entrypoint python3 `
         --volume ((Convert-ToDockerPath $source.SourcePack) + ":/source/lidar-pack.npz:ro") `
@@ -414,6 +396,7 @@ try {
             --volume ($dockerRun + ":/shared:rw") `
             --volume ($dockerVegetationDataset + ":/data/goose:ro") `
             --volume ($dockerVegetationCheckpoint + ":/models/candidate.pth:ro") `
+            --volume ((Convert-ToDockerPath $source.Video) + ":/source/right.mp4:ro") `
             $VegetationImageTag run --no-capture-output --name goose python `
             /release/run_vegetation_integrated_load.py `
             --config /release/lab-v1-goose-vegetation-benchmark-v1.json `
@@ -421,7 +404,7 @@ try {
             --provider-map /release/lab-v1-vegetation-provider-label-map-v1.json `
             --checkpoint /models/candidate.pth `
             --dataset-root /data/goose `
-            --frames-root /shared/vegetation/input-frames `
+            --video /source/right.mp4 `
             --source-rate-hz $rate `
             --minimum-effective-fps 11.209069 `
             --maximum-completion-p95-ms 125.0 `
@@ -582,10 +565,6 @@ try {
         Assert-LastExitCode "M49 integrated vegetation evidence gate"
     }
 } finally {
-    $vegetationFrames = Join-Path $runOutput "vegetation\input-frames"
-    if (Test-Path -LiteralPath $vegetationFrames -PathType Container) {
-        Remove-Item -LiteralPath $vegetationFrames -Recurse -Force
-    }
     foreach ($name in $containers) { Remove-ExactContainer $name }
     $canonicalAfter = Get-Container "ndc-mission-core-triton"
     if (

experiments/perception/worker/lab_v1_vegetation_goose/run_vegetation_integrated_load.py

@@ -12,6 +12,7 @@ import time
 from pathlib import Path
 from typing import Any
 
+import cv2
 import torch
 from PIL import Image
 from run_goose_vegetation_benchmark import (
@@ -51,7 +52,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--provider-map", type=Path, required=True)
     parser.add_argument("--checkpoint", type=Path, required=True)
     parser.add_argument("--dataset-root", type=Path, required=True)
-    parser.add_argument("--frames-root", type=Path, required=True)
+    parser.add_argument("--video", 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)
@@ -86,14 +87,30 @@ def distribution(values: list[float]) -> dict[str, float]:
     }
 
 
-def exact_frames(root: Path) -> list[Path]:
-    if root.is_symlink() or not root.is_dir():
-        raise IntegratedLoadError("RAVNOVES frame root is unavailable")
-    frames = sorted(root.glob("frame-*.png"))
-    expected = [f"frame-{sequence + 1:06d}.png" for sequence in range(FRAME_COUNT)]
-    if len(frames) != FRAME_COUNT or [frame.name for frame in frames] != expected:
-        raise IntegratedLoadError("RAVNOVES full-video frame sequence changed")
-    return frames
+def open_video(path: Path, expected_size: tuple[int, int]) -> cv2.VideoCapture:
+    if path.is_symlink() or not path.is_file():
+        raise IntegratedLoadError("RAVNOVES video is unavailable")
+    capture = cv2.VideoCapture(str(path))
+    if not capture.isOpened():
+        raise IntegratedLoadError("RAVNOVES video decoder did not open")
+    metadata = (
+        round(capture.get(cv2.CAP_PROP_FRAME_COUNT)),
+        round(capture.get(cv2.CAP_PROP_FRAME_WIDTH)),
+        round(capture.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+    )
+    if metadata != (FRAME_COUNT, expected_size[0], expected_size[1]):
+        capture.release()
+        raise IntegratedLoadError("RAVNOVES video metadata changed")
+    return capture
+
+
+def decode_source(capture: cv2.VideoCapture, expected_size: tuple[int, int]) -> Image.Image:
+    available, bgr = capture.read()
+    if not available or bgr is None:
+        raise IntegratedLoadError("RAVNOVES video ended before the frozen frame count")
+    if (bgr.shape[1], bgr.shape[0]) != expected_size:
+        raise IntegratedLoadError("RAVNOVES decoded frame dimensions changed")
+    return Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB), mode="RGB")
 
 
 def validate_sha256(value: str, label: str) -> None:
@@ -130,14 +147,20 @@ def run() -> int:
         raise IntegratedLoadError("DDRNet checkpoint digest changed")
     mapping_path = args.dataset_root / config["dataset"]["mapping_relative_path"]
     load_mapping(mapping_path, config["dataset"]["mapping_sha256"])
-    frames = exact_frames(args.frames_root)
+    expected_size = (
+        config["ravnoves"]["expected_width"],
+        config["ravnoves"]["expected_height"],
+    )
+    warmup_capture = open_video(args.video, expected_size)
+    warmup_source = decode_source(warmup_capture, expected_size)
+    warmup_capture.release()
 
     torch.cuda.empty_cache()
     model, model_name, architecture_failures = load_model("ddrnet", args.checkpoint)
-    with Image.open(frames[0]) as image:
-        warmup_tensor, _ = preprocess(image.convert("RGB"))
+    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, expected_size)
     wait_for_shared_start(
         args.shared_start_ready_file,
         args.shared_start_file,
@@ -153,19 +176,13 @@ def run() -> int:
     late_deadline_count = 0
     args.frame_ledger.parent.mkdir(parents=True, exist_ok=True)
     with args.frame_ledger.open("x", encoding="utf-8") as ledger:
-        for sequence, frame in enumerate(frames):
+        for sequence in range(FRAME_COUNT):
             scheduled_ns = start_ns + round(sequence * interval_ns)
             remaining_ns = scheduled_ns - time.monotonic_ns()
             if remaining_ns > 0:
                 time.sleep(remaining_ns / 1_000_000_000.0)
             admitted_ns = time.monotonic_ns()
-            with Image.open(frame) as image:
-                source = image.convert("RGB")
-            if source.size != (
-                config["ravnoves"]["expected_width"],
-                config["ravnoves"]["expected_height"],
-            ):
-                raise IntegratedLoadError("RAVNOVES video frame dimensions changed")
+            source = decode_source(source_capture, expected_size)
             tensor, _ = preprocess(source)
             _, inference_ms = infer(model, tensor)
             completed_ns = time.monotonic_ns()
@@ -181,7 +198,7 @@ def run() -> int:
             row = {
                 "schema_version": FRAME_SCHEMA,
                 "sequence": sequence,
-                "frame_name": frame.name,
+                "frame_name": f"frame-{sequence + 1:06d}",
                 "scheduled_monotonic_ns": scheduled_ns,
                 "admitted_monotonic_ns": admitted_ns,
                 "completed_monotonic_ns": completed_ns,
@@ -192,6 +209,10 @@ def run() -> int:
             ledger.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n")
             if sequence % 64 == 0:
                 ledger.flush()
+    extra_available, _ = source_capture.read()
+    source_capture.release()
+    if extra_available:
+        raise IntegratedLoadError("RAVNOVES video contains frames beyond the frozen timeline")
 
     completed_ns = time.monotonic_ns()
     wall_seconds = (completed_ns - start_ns) / 1_000_000_000.0
@@ -231,6 +252,12 @@ def run() -> int:
             "frame_count": FRAME_COUNT,
             "capacity_drop_count": 0,
             "deadline_miss_count": late_deadline_count,
+            "source_decode": {
+                "mode": "bounded-sequential-h264/v1",
+                "full_route_rgb_prefetch": False,
+                "candidate_local_decoder": True,
+                "runtime_target": "shared-source-frame",
+            },
             "frame_ledger": {
                 "path": args.frame_ledger.name,
                 "rows": FRAME_COUNT,

tests/test_lab_v1_vegetation_integrated_shadow.py

@@ -236,6 +236,9 @@ def test_worker_gate_reuses_shared_barrier_and_keeps_canonical_triton_unchanged(
     wrapper = POWERSHELL_PATH.read_text(encoding="utf-8")
     assert '"source-paced-integrated-shadow/v1"' in runner
     assert "wait_for_shared_start(" in runner
+    assert '"bounded-sequential-h264/v1"' in runner
+    assert '"full_route_rgb_prefetch": False' in runner
+    assert "decode_source(source_capture, expected_size)" in runner
     assert '"camera_semantics_can_clear_rigid_geometry": False' in runner
     assert "$VegetationLoadGate" in wrapper
     assert '"vegetation"' in wrapper