perf(perception): split safety and vegetation rates

2026-08-28 12:59:34 +03:00
parent e1c3e7e90a
commit 7a3b0b84d0
7 changed files with 292 additions and 56 deletions
@@ -148,12 +148,12 @@ foreach ($directory in @("bin", "control", "graph", "tgs", "vegetation")) {

 $releaseDocument = Get-Content -LiteralPath (Join-Path $payload "release.json") -Raw | ConvertFrom-Json
 $expectedReleaseSchema = if ($VegetationLoadGate) {
-    "missioncore.lab-v1-vegetation-integrated-worker-release/v1"
+    "missioncore.lab-v1-vegetation-integrated-worker-release/v2"
 } else {
    "missioncore.m49-tgs-integrated-graph-worker-release/v1"
 }
 $expectedTransition = if ($VegetationLoadGate) {
-    "lab-v1-vegetation-m49-integrated-shadow/v1"
+    "lab-v1-vegetation-m49-integrated-multirate-shadow/v2"
 } else {
    "m49-tgs-native-risk-integrated-shadow/v1"
 }
@@ -380,7 +380,7 @@ try {

    & docker create --name $tgsName --network none --cpus 16 --memory 24g `
        --read-only --security-opt "no-new-privileges:true" --cap-drop ALL `
-        --pids-limit 256 --tmpfs "/tmp:rw,noexec,nosuid,size=1g" `
+        --pids-limit 256 --tmpfs "/tmp:rw,noexec,nosuid,size=2g" `
        -e ("M49_SOURCE_RATE_HZ={0}" -f $rate) `
        --entrypoint /bin/bash `
        --volume ($dockerRelease + ":/release:ro") `
@@ -412,8 +412,11 @@ try {
            --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 `
+            --inference-stride 2 `
+            --minimum-effective-timeline-fps 11.209069 `
+            --minimum-effective-inference-fps 5.604534 `
+            --maximum-inference-completion-p95-ms 125.0 `
+            --maximum-evidence-source-age-ms 125.0 `
            --shared-start-ready-file /shared/control/vegetation.ready `
            --shared-start-file /shared/control/start.signal `
            --frame-ledger /shared/vegetation/frames.jsonl `
@@ -559,7 +562,7 @@ try {
            --volume ($dockerRelease + ":/release:ro") `
            --volume ($dockerRun + ":/shared:rw") `
            $ParityImageTag /release/build_vegetation_integrated_graph_evidence.py `
-            --profile /release/lab-v1-vegetation-integrated-shadow-v1.json `
+            --profile /release/lab-v1-vegetation-integrated-multirate-shadow-v2.json `
            --m49-result /shared/m49-result.json `
            --graph-frames /shared/graph/frames.jsonl `
            --tgs-timing /shared/tgs/tgs-full-timing.tsv `
@@ -588,7 +591,7 @@ if (-not (Test-Path -LiteralPath $resultPath -PathType Leaf)) {
 $result = Get-Content -LiteralPath $resultPath -Raw | ConvertFrom-Json
 $summary = [ordered]@{
    schema_version = if ($VegetationLoadGate) {
-        "missioncore.lab-v1-vegetation-integrated-worker-summary/v1"
+        "missioncore.lab-v1-vegetation-integrated-worker-summary/v2"
    } else {
        "missioncore.m49-tgs-integrated-graph-worker-summary/v1"
    }
@@ -28,8 +28,8 @@ from run_goose_vegetation_benchmark import (
    validate_contracts,
 )

-SCHEMA = "missioncore.lab-v1-vegetation-integrated-load/v1"
-FRAME_SCHEMA = "missioncore.lab-v1-vegetation-integrated-frame/v1"
+SCHEMA = "missioncore.lab-v1-vegetation-integrated-load/v2"
+FRAME_SCHEMA = "missioncore.lab-v1-vegetation-integrated-frame/v2"
 FRAME_COUNT = 4_489
 AUTHORITY = {
    "ground_truth": False,
@@ -57,8 +57,11 @@ def parse_args() -> argparse.Namespace:
    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)
+    parser.add_argument("--inference-stride", type=int, required=True)
+    parser.add_argument("--minimum-effective-timeline-fps", type=float, required=True)
+    parser.add_argument("--minimum-effective-inference-fps", type=float, required=True)
+    parser.add_argument("--maximum-inference-completion-p95-ms", type=float, required=True)
+    parser.add_argument("--maximum-evidence-source-age-ms", type=float, required=True)
    parser.add_argument("--shared-start-ready-file", type=Path, required=True)
    parser.add_argument("--shared-start-file", type=Path, required=True)
    parser.add_argument("--shared-start-timeout-seconds", type=float, default=600.0)
@@ -139,12 +142,16 @@ def run() -> int:
    args = parse_args()
    if not torch.cuda.is_available():
        raise IntegratedLoadError("CUDA is required for Worker 006 qualification")
-    if (
-        not math.isfinite(args.source_rate_hz)
-        or args.source_rate_hz <= 0
-        or args.minimum_effective_fps <= 0
-        or args.maximum_completion_p95_ms <= 0
-        or args.shared_start_timeout_seconds <= 0
+    positive_finite_values = (
+        args.source_rate_hz,
+        args.minimum_effective_timeline_fps,
+        args.minimum_effective_inference_fps,
+        args.maximum_inference_completion_p95_ms,
+        args.maximum_evidence_source_age_ms,
+        args.shared_start_timeout_seconds,
+    )
+    if args.inference_stride <= 0 or any(
+        not math.isfinite(value) or value <= 0 for value in positive_finite_values
    ):
        raise IntegratedLoadError("integrated-load thresholds must be positive and finite")
    validate_sha256(args.release_sha256, "release")
@@ -192,9 +199,13 @@ def run() -> int:
    start_ns = time.monotonic_ns()
    started_utc_ns = time.time_ns()
    completion_ages_ms: list[float] = []
+    inference_completion_ages_ms: list[float] = []
+    evidence_source_ages_ms: list[float] = []
    stage_latencies_ms: list[float] = []
    inference_latencies_ms: list[float] = []
    late_deadline_count = 0
+    inference_frame_count = 0
+    last_inference_sequence = -1
    args.frame_ledger.parent.mkdir(parents=True, exist_ok=True)
    with args.frame_ledger.open("x", encoding="utf-8") as ledger:
        for sequence in range(FRAME_COUNT):
@@ -204,14 +215,30 @@ def run() -> int:
                time.sleep(remaining_ns / 1_000_000_000.0)
            admitted_ns = time.monotonic_ns()
            source = decode_source(source_capture, expected_size)
-            tensor, _ = preprocess(source)
-            _, inference_ms = infer(model, tensor)
+            inference_executed = sequence % args.inference_stride == 0
+            inference_ms: float | None = None
+            if inference_executed:
+                tensor, _ = preprocess(source)
+                _, inference_ms = infer(model, tensor)
+                last_inference_sequence = sequence
+                inference_frame_count += 1
+            if last_inference_sequence < 0:
+                raise IntegratedLoadError("semantic evidence is unavailable for the timeline")
            completed_ns = time.monotonic_ns()
            completion_age_ms = (completed_ns - scheduled_ns) / 1_000_000.0
            stage_ms = (completed_ns - admitted_ns) / 1_000_000.0
+            semantic_source_scheduled_ns = start_ns + round(
+                last_inference_sequence * interval_ns
+            )
+            evidence_source_age_ms = (
+                completed_ns - semantic_source_scheduled_ns
+            ) / 1_000_000.0
            completion_ages_ms.append(completion_age_ms)
+            evidence_source_ages_ms.append(evidence_source_age_ms)
            stage_latencies_ms.append(stage_ms)
-            inference_latencies_ms.append(inference_ms)
+            if inference_ms is not None:
+                inference_latencies_ms.append(inference_ms)
+                inference_completion_ages_ms.append(completion_age_ms)
            if sequence + 1 < FRAME_COUNT and completed_ns > start_ns + round(
                (sequence + 1) * interval_ns
            ):
@@ -225,7 +252,10 @@ def run() -> int:
                "completed_monotonic_ns": completed_ns,
                "completion_age_ms": round(completion_age_ms, 6),
                "stage_ms": round(stage_ms, 6),
-                "inference_ms": round(inference_ms, 6),
+                "inference_executed": inference_executed,
+                "inference_ms": round(inference_ms, 6) if inference_ms is not None else None,
+                "semantic_source_sequence": last_inference_sequence,
+                "semantic_evidence_source_age_ms": round(evidence_source_age_ms, 6),
            }
            ledger.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n")
            if sequence % 64 == 0:
@@ -237,13 +267,23 @@ def run() -> int:

    completed_ns = time.monotonic_ns()
    wall_seconds = (completed_ns - start_ns) / 1_000_000_000.0
-    effective_fps = FRAME_COUNT / wall_seconds
+    effective_timeline_fps = FRAME_COUNT / wall_seconds
+    effective_inference_fps = inference_frame_count / wall_seconds
    completion = distribution(completion_ages_ms)
+    inference_completion = distribution(inference_completion_ages_ms)
+    evidence_source_age = distribution(evidence_source_ages_ms)
+    expected_inference_frames = (FRAME_COUNT + args.inference_stride - 1) // args.inference_stride
    checks = {
        "all_frames_accounted": len(completion_ages_ms) == FRAME_COUNT,
-        "minimum_effective_fps": effective_fps >= args.minimum_effective_fps,
-        "maximum_completion_p95_ms": completion["p95"]
-        <= args.maximum_completion_p95_ms,
+        "exact_multirate_schedule": inference_frame_count == expected_inference_frames,
+        "minimum_effective_timeline_fps": effective_timeline_fps
+        >= args.minimum_effective_timeline_fps,
+        "minimum_effective_inference_fps": effective_inference_fps
+        >= args.minimum_effective_inference_fps,
+        "maximum_inference_completion_p95_ms": inference_completion["p95"]
+        <= args.maximum_inference_completion_p95_ms,
+        "maximum_evidence_source_age_ms": evidence_source_age["maximum"]
+        <= args.maximum_evidence_source_age_ms,
        "zero_capacity_drops": len(completion_ages_ms) == FRAME_COUNT,
        "authority_remains_false": all(value is False for value in AUTHORITY.values()),
    }
@@ -266,10 +306,15 @@ def run() -> int:
            "checkpoint_sha256": checkpoint_sha256,
        },
        "execution": {
-            "run_mode": "source-paced-integrated-shadow/v1",
+            "run_mode": "source-paced-multirate-integrated-shadow/v2",
            "started_utc_ns": started_utc_ns,
            "wall_seconds": round(wall_seconds, 6),
-            "effective_fps": round(effective_fps, 6),
+            "effective_fps": round(effective_timeline_fps, 6),
+            "effective_timeline_fps": round(effective_timeline_fps, 6),
+            "effective_inference_fps": round(effective_inference_fps, 6),
+            "inference_stride": args.inference_stride,
+            "inference_frame_count": inference_frame_count,
+            "held_evidence_frame_count": FRAME_COUNT - inference_frame_count,
            "frame_count": FRAME_COUNT,
            "capacity_drop_count": 0,
            "deadline_miss_count": late_deadline_count,
@@ -292,6 +337,8 @@ def run() -> int:
            "prewarm_latency_ms_first": round(warmup_latencies_ms[0], 6),
            "prewarm_latency_ms_last": round(warmup_latencies_ms[-1], 6),
            "completion_age_ms": completion,
+            "inference_completion_age_ms": inference_completion,
+            "semantic_evidence_source_age_ms": evidence_source_age,
            "stage_ms": distribution(stage_latencies_ms),
            "inference_ms": distribution(inference_latencies_ms),
        },
@@ -311,8 +358,12 @@ def run() -> int:
            "runner_sha256": sha256(Path(__file__)),
        },
        "predeclared_thresholds": {
-            "minimum_effective_fps": args.minimum_effective_fps,
-            "maximum_completion_p95_ms": args.maximum_completion_p95_ms,
+            "minimum_effective_timeline_fps": args.minimum_effective_timeline_fps,
+            "minimum_effective_inference_fps": args.minimum_effective_inference_fps,
+            "maximum_inference_completion_p95_ms": (
+                args.maximum_inference_completion_p95_ms
+            ),
+            "maximum_evidence_source_age_ms": args.maximum_evidence_source_age_ms,
            "capacity_drop_count_max": 0,
        },
        "checks": checks,
@@ -15,10 +15,10 @@ from typing import Any

 import numpy as np

-PROFILE_SCHEMA = "missioncore.lab-v1-vegetation-integrated-shadow-profile/v1"
+PROFILE_SCHEMA = "missioncore.lab-v1-vegetation-integrated-shadow-profile/v2"
 M49_SCHEMA = "missioncore.m49-tgs-integrated-graph-shadow-result/v1"
-VEGETATION_SCHEMA = "missioncore.lab-v1-vegetation-integrated-load/v1"
-RESULT_SCHEMA = "missioncore.lab-v1-vegetation-integrated-shadow-result/v1"
+VEGETATION_SCHEMA = "missioncore.lab-v1-vegetation-integrated-load/v2"
+RESULT_SCHEMA = "missioncore.lab-v1-vegetation-integrated-shadow-result/v2"
 FRAME_COUNT = 4_489


@@ -92,22 +92,51 @@ def tgs_completion_ages(path: Path) -> list[float]:
    return values


-def vegetation_completion_ages(path: Path) -> list[float]:
-    values: list[float] = []
+def vegetation_frame_metrics(
+    path: Path, *, inference_stride: int, source_rate_hz: float
+) -> dict[str, object]:
+    completion_ages: list[float] = []
+    evidence_source_ages: list[float] = []
+    inference_count = 0
    with path.open("r", encoding="utf-8") as stream:
        for expected, line in enumerate(stream):
            row = json.loads(line)
-            if row.get("schema_version") != "missioncore.lab-v1-vegetation-integrated-frame/v1":
+            if row.get("schema_version") != "missioncore.lab-v1-vegetation-integrated-frame/v2":
                raise VegetationIntegratedError("vegetation frame schema changed")
            if row.get("sequence") != expected:
                raise VegetationIntegratedError("vegetation frame sequence changed")
            age = row.get("completion_age_ms")
            if not isinstance(age, (int, float)) or not math.isfinite(age) or age < 0:
                raise VegetationIntegratedError("vegetation completion age is invalid")
-            values.append(float(age))
-    if len(values) != FRAME_COUNT:
+            inference_executed = row.get("inference_executed")
+            expected_inference = expected % inference_stride == 0
+            if inference_executed is not expected_inference:
+                raise VegetationIntegratedError("vegetation inference schedule changed")
+            expected_source = expected - (expected % inference_stride)
+            if row.get("semantic_source_sequence") != expected_source:
+                raise VegetationIntegratedError("vegetation evidence source changed")
+            evidence_age = row.get("semantic_evidence_source_age_ms")
+            expected_evidence_age = float(age) + (
+                (expected - expected_source) * 1000.0 / source_rate_hz
+            )
+            if (
+                not isinstance(evidence_age, (int, float))
+                or not math.isfinite(evidence_age)
+                or evidence_age < 0
+                or abs(float(evidence_age) - expected_evidence_age) > 0.001
+            ):
+                raise VegetationIntegratedError("vegetation evidence source age changed")
+            completion_ages.append(float(age))
+            evidence_source_ages.append(float(evidence_age))
+            inference_count += int(expected_inference)
+    if len(completion_ages) != FRAME_COUNT:
        raise VegetationIntegratedError("vegetation frame ledger is incomplete")
-    return values
+    return {
+        "completion_ages": completion_ages,
+        "evidence_source_ages": evidence_source_ages,
+        "inference_count": inference_count,
+        "held_count": FRAME_COUNT - inference_count,
+    }


 _SIZE = re.compile(r"^\s*([0-9.]+)\s*([kmgt]?i?b)\s*$", re.IGNORECASE)
@@ -200,9 +229,17 @@ def build(
    if vegetation.get("schema_version") != VEGETATION_SCHEMA:
        raise VegetationIntegratedError("vegetation load result schema changed")

+    source_rate_hz = float(profile["source"]["requested_source_rate_hz"])
+    inference_stride = int(profile["stages"]["vegetation"]["inference_stride"])
    graph_ages = graph_completion_ages(graph_frames_path)
    tgs_ages = tgs_completion_ages(tgs_timing_path)
-    vegetation_ages = vegetation_completion_ages(vegetation_frames_path)
+    vegetation_frames = vegetation_frame_metrics(
+        vegetation_frames_path,
+        inference_stride=inference_stride,
+        source_rate_hz=source_rate_hz,
+    )
+    vegetation_ages = vegetation_frames["completion_ages"]
+    assert isinstance(vegetation_ages, list)
    combined_ages = [
        max(graph, tgs, semantic)
        for graph, tgs, semantic in zip(
@@ -240,6 +277,14 @@ def build(
            vegetation.get("source", {}).get("requested_source_rate_hz")
            == profile["source"]["requested_source_rate_hz"]
        ),
+        "vegetation_load_gate_passed": vegetation.get("integrated_load_gate_passed") is True,
+        "vegetation_multirate_schedule_frozen": (
+            vegetation_execution.get("inference_stride") == inference_stride
+            and vegetation_execution.get("inference_frame_count")
+            == vegetation_frames["inference_count"]
+            and vegetation_execution.get("held_evidence_frame_count")
+            == vegetation_frames["held_count"]
+        ),
        "exact_three_layer_sequence_join": len(combined_ages) == FRAME_COUNT,
        "all_graph_frames_delivered": (
            m49_accounting.get("graph_admitted") == FRAME_COUNT
@@ -253,12 +298,24 @@ def build(
            m49_performance.get("effective_world_state_fps", 0.0)
        )
        >= float(acceptance["minimum_graph_world_state_fps"]),
-        "minimum_vegetation_fps": float(vegetation_execution.get("effective_fps", 0.0))
-        >= float(acceptance["minimum_vegetation_fps"]),
-        "maximum_vegetation_completion_p95_ms": float(
-            vegetation_timing.get("completion_age_ms", {}).get("p95", math.inf)
+        "minimum_vegetation_timeline_fps": float(
+            vegetation_execution.get("effective_timeline_fps", 0.0)
        )
-        <= float(acceptance["maximum_vegetation_completion_p95_ms"]),
+        >= float(acceptance["minimum_vegetation_timeline_fps"]),
+        "minimum_vegetation_inference_fps": float(
+            vegetation_execution.get("effective_inference_fps", 0.0)
+        )
+        >= float(acceptance["minimum_vegetation_inference_fps"]),
+        "maximum_vegetation_inference_completion_p95_ms": float(
+            vegetation_timing.get("inference_completion_age_ms", {}).get(
+                "p95", math.inf
+            )
+        )
+        <= float(acceptance["maximum_vegetation_inference_completion_p95_ms"]),
+        "maximum_semantic_evidence_source_age_ms": max(
+            vegetation_frames["evidence_source_ages"]
+        )
+        <= float(acceptance["maximum_semantic_evidence_source_age_ms"]),
        "maximum_combined_output_age_p99_ms": combined["p99"]
        <= float(acceptance["maximum_combined_output_age_p99_ms"]),
        "zero_capacity_drops": (
@@ -304,8 +361,19 @@ def build(
        "performance": {
            "graph_tgs": m49_performance,
            "vegetation": {
-                "effective_fps": vegetation_execution.get("effective_fps"),
+                "effective_timeline_fps": vegetation_execution.get(
+                    "effective_timeline_fps"
+                ),
+                "effective_inference_fps": vegetation_execution.get(
+                    "effective_inference_fps"
+                ),
                "completion_age_ms": vegetation_timing.get("completion_age_ms"),
+                "inference_completion_age_ms": vegetation_timing.get(
+                    "inference_completion_age_ms"
+                ),
+                "semantic_evidence_source_age_ms": distribution(
+                    vegetation_frames["evidence_source_ages"]
+                ),
                "stage_ms": vegetation_timing.get("stage_ms"),
                "inference_ms": vegetation_timing.get("inference_ms"),
                "resource": vegetation.get("resource"),
@@ -317,6 +385,8 @@ def build(
            "graph_frames": m49_accounting.get("graph_delivered"),
            "tgs_frames": m49_accounting.get("tgs_timeline_frames"),
            "vegetation_frames": vegetation_execution.get("frame_count"),
+            "vegetation_inference_frames": vegetation_frames["inference_count"],
+            "vegetation_held_evidence_frames": vegetation_frames["held_count"],
            "capacity_drop_count": int(m49_accounting.get("tgs_capacity_drops", 0))
            + int(vegetation_execution.get("capacity_drop_count", 0)),
        },
@@ -5,6 +5,9 @@ readonly BINARY=/shared/bin/run_tgs_full_shadow
 readonly INPUT_ROOT=/shared/tgs/inputs
 readonly OUTPUT_ROOT=/shared/tgs/outputs/causal_rolling_1s
 readonly TIMING_PATH=/shared/tgs/tgs-full-timing.tsv
+readonly RUNTIME_ROOT=/tmp/m49-tgs-runtime
+readonly RUNTIME_OUTPUT_ROOT=${RUNTIME_ROOT}/outputs
+readonly RUNTIME_TIMING_PATH=${RUNTIME_ROOT}/tgs-full-timing.tsv
 readonly READY_FILE=/shared/control/tgs.ready
 readonly START_FILE=/shared/control/start.signal
 readonly SOURCE_RATE_HZ=${M49_SOURCE_RATE_HZ:-12.0}
@@ -15,12 +18,20 @@ test -f "${INPUT_ROOT}/schedule.tsv"
 test ! -e /shared/tgs/outputs
 test ! -e "${TIMING_PATH}"
 test ! -e "${READY_FILE}"
-mkdir -p "${OUTPUT_ROOT}"
-exec /usr/bin/time -v "${BINARY}" \
+test ! -e "${RUNTIME_ROOT}"
+mkdir -p "${RUNTIME_OUTPUT_ROOT}"
+/usr/bin/time -v "${BINARY}" \
  "${INPUT_ROOT}/profiles/causal_rolling_1s" \
  "${INPUT_ROOT}/schedule.tsv" \
-  "${OUTPUT_ROOT}" \
-  "${TIMING_PATH}" \
+  "${RUNTIME_OUTPUT_ROOT}" \
+  "${RUNTIME_TIMING_PATH}" \
  "${SOURCE_RATE_HZ}" \
  "${READY_FILE}" \
  "${START_FILE}"
+test -f "${RUNTIME_TIMING_PATH}"
+mkdir -p "${OUTPUT_ROOT}"
+copy_started=$(date +%s%N)
+cp -R "${RUNTIME_OUTPUT_ROOT}/." "${OUTPUT_ROOT}/"
+cp "${RUNTIME_TIMING_PATH}" "${TIMING_PATH}"
+copy_completed=$(date +%s%N)
+echo "[TGS-FULL] evidence_copy_ms=$(((copy_completed - copy_started) / 1000000))"