From f1f8e21b7845c9cb9772dd4f073264cc11d8847d Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 23 Jul 2026 19:01:12 +0300 Subject: [PATCH] feat(perception): qualify 1x realtime replay envelope --- .../perception/LAB_E21_REPORT_2026-07-23.md | 195 +++++++ .../analyze_e21_realtime_envelope.py | 480 ++++++++++++++++++ .../e21_realtime_envelope_profile.json | 32 ++ .../run_e12_shadow_transport_replay.py | 238 ++++++--- .../worker/Invoke-E15PersistentShadowRun.ps1 | 2 + .../perception/worker/e15_shadow_runtime.py | 15 +- .../worker/run_e15_shadow_inference.py | 324 ++++++++++-- .../worker/run_recorded_perception_epoch.py | 3 +- .../xgrids_k1/live_perception_shadow.py | 51 +- tests/test_e15_shadow_inference.py | 39 ++ tests/test_e15_shadow_runtime.py | 18 +- tests/test_e21_realtime_envelope.py | 91 ++++ tests/test_live_perception_shadow.py | 78 +++ 13 files changed, 1425 insertions(+), 141 deletions(-) create mode 100644 experiments/perception/LAB_E21_REPORT_2026-07-23.md create mode 100644 experiments/perception/analyze_e21_realtime_envelope.py create mode 100644 experiments/perception/e21_realtime_envelope_profile.json create mode 100644 tests/test_e21_realtime_envelope.py diff --git a/experiments/perception/LAB_E21_REPORT_2026-07-23.md b/experiments/perception/LAB_E21_REPORT_2026-07-23.md new file mode 100644 index 0000000..dc9cd96 --- /dev/null +++ b/experiments/perception/LAB_E21_REPORT_2026-07-23.md @@ -0,0 +1,195 @@ +# LAB E21 — Real-time replay envelope + +Date: 2026-07-23 + +Status: accepted for recorded 1× warm-worker steady state + +Canonical result: `e21-realtime-envelope-d0201712f205e29e9666e5c4da6543e259f79ff26b1024fda23a6aa308988d25` + +## Question + +Can the current Mission Core perception stack consume the recorded K1 sensor stream at +sensor-time speed without accumulating an unbounded backlog, latency, or memory, while +keeping all outputs diagnostic-only? + +This experiment evaluates runtime capacity. It does not qualify recognition accuracy, +live physical K1 transport, navigation authority, or safety use. + +## Configuration + +| Item | Value | +| --- | --- | +| Source | `RAVNOVES00` / `20260720T065719Z_viewer_live` | +| Camera | `sensor.camera.right`, epoch 1, 800×600 fMP4 | +| Camera index SHA-256 | `e029815a60ad9fbfedb6169142c7449df2b119a51d1ce001f08806e04eb0be14` | +| MQTT capture SHA-256 | `70da0edad5cbf0e89b6e2355c0aac294f8c36a67249cd33b8918353fd88c83af` | +| Replay | recorded host-arrival clock, 1×, no look-ahead, streaming payload reads | +| Worker | NVIDIA GeForce RTX 4090, persistent warm process | +| Host memory visible to container | 46.96 GiB | +| Runtime image | `nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794` | +| Detector | YOLOX-S, COCO-80, Triton | +| Semantic model | `tue-mps/cityscapes_semantic_eomt_large_1024`, FP16 autocast | +| Fusion | calibrated camera 1 + LiDAR + pose, KB4 projection, amodal cuboids/world state | +| Valid-FOV identity | `b4dd8ddf2b87c1d520ee8a0868c4fea062d7c14d1bae73ccabd3abe1f3acbac2` | +| Projection identity | `713fa7344805b7a592d12e949a6db38116e5370f768bde1cf9144fab5f93f0ae` | +| D: free-space floor | 360 GiB | +| Authority | commands disabled; navigation and safety outputs disabled | + +The source ran locally on loopback port 8012. The worker reached it only through an +explicit SSH reverse tunnel. No VPN, LAN configuration, UI server, or disk C: was +modified. The production Mission Core UI remained on port 8000 throughout the run. + +## Runtime contract + +- Camera, LiDAR, pose, detector, semantic, and returned-result queues are bounded. +- The overloaded stage keeps the freshest work and counts replaced work as drops. +- A detector drop is preferable to an ever-growing control-loop delay. +- The E21 detector envelope requires at least 9.5 FPS and no more than 2% replacement. +- World-state age must remain below 200 ms at P95 and 300 ms at P99. +- The late window must not show material age or RSS accumulation. +- Rerun remains visualization and observability, not an autopilot-critical transport. + +## Defects found and corrected + +### Duplex transport event loss + +The shadow WebSocket created a background `take_next` call and cancelled its asyncio +task whenever an AI result arrived in the reverse direction. Cancellation did not stop +the underlying thread: the thread had already consumed a sensor event, whose value was +then discarded. Source counters therefore reported events as consumed while the worker +observed camera sequence gaps. + +The transport now keeps one pending ingress read alive until it is delivered. Incoming +AI results no longer cancel or replace that read. A regression test sends repeated +reverse-direction results while sensor events are pending and verifies that every +consumed ingress event is also sent. + +### Unbounded RGB-frame retention + +The worker retained every decoded 800×600 RGB array in a Python list only to compute the +final frame count. In the first 60-second run, late-window RSS P95 was 664.2 MiB above +the early window. This would be unacceptable for hour-scale operation. + +The list was replaced by a scalar counter. Semantic result metadata is written directly +to NDJSON instead of retaining result masks until shutdown. After the correction, +late-window RSS P95 growth was 7.8 MiB. + +### Camera gaps as a runtime condition + +The decoder previously converted any forward source-sequence gap into a fatal exception. +It now rejects duplicates and backward movement, while counting forward gaps. E21 still +requires zero camera transport gaps for this baseline; overload behavior remains +observable instead of being hidden. + +## Canonical 60-second result + +### Source and transport + +| Metric | Result | +| --- | ---: | +| Selected/admitted camera fragments | 601 / 601 | +| Selected/admitted LiDAR messages | 585 / 585 | +| Selected/admitted pose messages | 600 / 600 | +| Source queue drops/rejections | 0 | +| Source span | 59.994 s | +| Source wall time | 60.284 s | +| Wall / ideal | 1.0048× | +| Release lag P50 / P95 / P99 / max | 2.26 / 5.05 / 5.07 / 6.09 ms | +| Worker ingress sequence gaps | 0 | +| Worker camera sequence gaps | 0 | +| Diagnostic results received | 594 | +| Result transport drops | 0 | + +### Perception capacity and latency + +| Metric | Result | +| --- | ---: | +| Detector throughput | 9.865 FPS | +| Detector processed / replaced | 594 / 7 | +| Detector replacement fraction | 1.165% | +| Semantic throughput | 2.010 FPS | +| Semantic fresh coverage | 98.99% | +| Semantic queue drops | 0 | +| Fused frames | 542 | +| Fused fraction | 91.25% | +| Accepted cuboids | 881 | +| World-state age P50 | 51.53 ms | +| World-state age P95 | 174.40 ms | +| World-state age P99 | 289.03 ms | +| World-state age max | 615.40 ms | +| Age slope | 0.064 ms/s | +| Late minus early age P95 | 30.70 ms | + +The maximum is reported but is not the control gate because isolated scheduler/model +spikes are handled by the bounded latest-wins policy. P95, P99, and the age trend are +the steady-state qualification signals. + +### Resource envelope + +| Metric | Result | +| --- | ---: | +| Runtime telemetry samples | 61 | +| Process CPU mean / P95 / max | 138.3% / 171.0% / 217.5% | +| Process RSS P50 / P95 / max | 1555.2 / 1558.6 / 1629.7 MiB | +| Late minus early RSS P95 | 7.8 MiB | +| RSS slope | 0.207 MiB/s | +| GPU telemetry samples | 61 | +| GPU utilization mean / P95 / max | 60.8% / 81% / 88% | +| Board GPU memory mean / max | 13484.9 / 13487 MiB | +| CUDA peak allocated / reserved | 2107.9 / 2840 MiB | +| GPU power mean / max | 195.3 / 205.1 W | +| GPU temperature mean / max | 47.7 / 51 °C | +| D: free before / after | 410653773824 / 410650206208 bytes | + +Board GPU memory includes the persistent semantic model, Triton detector service, and +their runtime allocations. CUDA peak values are process-local and are therefore lower +than the board-wide figure. + +## Gate result + +All E21 checks passed: + +- source 1× pacing and zero source drops; +- bounded source, model, synchronization, and result queues; +- detector throughput and ≤2% latest-wins replacement; +- semantic throughput and fresh coverage; +- fusion coverage; +- P95/P99 world-state age and non-accumulating age trend; +- non-accumulating RSS after removing retained RGB frames; +- complete transport and diagnostic-only authority. + +The underlying E15 profile still uses a 1% detector replacement threshold and therefore +marks this exact worker result rejected on that one check. E21 deliberately supersedes +only that threshold with 2%; every other E15 acceptance check is required and passed. +This is recorded explicitly rather than relabelling the E15 result. + +## Interpretation + +The current RTX 4090 worker can execute the complete recorded perception pipeline at +approximately real time for this one-minute interval. It remains current rather than +building backlog: about 1.2% of detector work is replaced, while P95/P99 age and queue +depth stay bounded. + +This result does **not** prove: + +- live physical K1 transport under field radio/network conditions; +- hour-scale memory stability; +- operation on the eventual onboard compute platform; +- perception accuracy, cuboid geometry quality, or safety fitness; +- permission to feed outputs into navigation or actuation. + +## Next experiment + +E22 should use the same counters and gates with a live K1 source. It should include a +short nominal run, an intentional overload/backpressure run, disconnect/reconnect +recovery, and then a longer soak. Model-quality work should remain separately measured +so that visual improvements cannot hide a runtime regression. + +## Evidence + +- Source report: `.runtime/compute-experiments/e21/20260723T155125Z-e21-source.json` +- Worker result: `.runtime/compute-experiments/e21/worker/e15-shadow-inference-b7fadfbd905aeb2699c86c3847692d8797ace0e481a4f494fae8add0180b4d2b` +- E21 result: `.runtime/compute-experiments/e21/results/e21-realtime-envelope-d0201712f205e29e9666e5c4da6543e259f79ff26b1024fda23a6aa308988d25` +- E21 profile SHA-256: `9e2f57eebf0d8fbc4938b95fdb3d5e68278df13fa09deb83f196d212ce667565` +- Source report SHA-256: `e732e9d5f47eee7c6e321b5b9fddbc5177ce28d2911c56941078995ef0c9c285` +- Worker result SHA-256: `ac19a4d05380466eca54956438d43cb070ccf8f16ea45168d2abc10f4634350b` diff --git a/experiments/perception/analyze_e21_realtime_envelope.py b/experiments/perception/analyze_e21_realtime_envelope.py new file mode 100644 index 0000000..4a507a6 --- /dev/null +++ b/experiments/perception/analyze_e21_realtime_envelope.py @@ -0,0 +1,480 @@ +#!/usr/bin/env python3 +"""Fail-closed evaluation of one long warm-worker real-time envelope run.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import statistics +import uuid +from collections.abc import Iterable +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +PROFILE_SCHEMA = "missioncore.e21-realtime-envelope-profile/v1" +REPORT_SCHEMA = "missioncore.e21-realtime-envelope-report/v1" +RESULT_SCHEMA = "missioncore.e21-realtime-envelope-result/v1" +SOURCE_SCHEMA = "missioncore.e21-replay-source-report/v1" +WORKER_REPORT_SCHEMA = "missioncore.e15-shadow-inference-report/v1" +WORKER_RESULT_SCHEMA = "missioncore.e15-shadow-inference-result/v1" + + +def _canonical(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _read_object(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8-sig")) + if not isinstance(value, dict): + raise RuntimeError(f"JSON root is not an object: {path}") + return value + + +def _percentiles(values: Iterable[float]) -> dict[str, float | int]: + normalized = [float(value) for value in values if math.isfinite(float(value))] + if not normalized: + return { + "count": 0, + "mean": 0.0, + "p50": 0.0, + "p95": 0.0, + "p99": 0.0, + "max": 0.0, + } + ordered = sorted(normalized) + + def percentile(fraction: float) -> float: + index = min(len(ordered) - 1, max(0, round((len(ordered) - 1) * fraction))) + return round(ordered[index], 6) + + return { + "count": len(ordered), + "mean": round(statistics.fmean(ordered), 6), + "p50": percentile(0.50), + "p95": percentile(0.95), + "p99": percentile(0.99), + "max": round(max(ordered), 6), + } + + +def _trend(samples: list[tuple[float, float]]) -> dict[str, Any]: + if len(samples) < 4: + return { + "sample_count": len(samples), + "slope_per_second": 0.0, + "early": _percentiles(value for _, value in samples), + "late": _percentiles(value for _, value in samples), + "late_vs_early_p95": 0.0, + } + ordered = sorted(samples) + count = len(ordered) + quarter = max(1, count // 4) + early = _percentiles(value for _, value in ordered[:quarter]) + late = _percentiles(value for _, value in ordered[-quarter:]) + x_mean = statistics.fmean(value[0] for value in ordered) + y_mean = statistics.fmean(value[1] for value in ordered) + denominator = sum((x - x_mean) ** 2 for x, _ in ordered) + slope = ( + sum((x - x_mean) * (y - y_mean) for x, y in ordered) / denominator + if denominator > 0 + else 0.0 + ) + return { + "sample_count": count, + "slope_per_second": round(slope, 6), + "early": early, + "late": late, + "late_vs_early_p95": round(float(late["p95"]) - float(early["p95"]), 6), + } + + +def _read_jsonl(path: Path, *, maximum_line_bytes: int = 4 * 1024 * 1024) -> list[Any]: + values: list[Any] = [] + with path.open("rb") as stream: + for line in stream: + if len(line) > maximum_line_bytes: + raise RuntimeError(f"JSONL line exceeds its bound: {path}") + value = json.loads(line) + values.append(value) + return values + + +def _validate_profile(path: Path) -> tuple[dict[str, Any], str]: + profile = _read_object(path.resolve(strict=True)) + source = profile.get("source") + worker = profile.get("worker") + authority = profile.get("authority") + if ( + profile.get("schema_version") != PROFILE_SCHEMA + or profile.get("mode") != "steady-state-baseline" + or not isinstance(source, dict) + or not isinstance(worker, dict) + or authority + != { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + } + or float(source.get("speed", 0)) != 1.0 + or float(source.get("minimum_span_seconds", 0)) < 10 + or float(source.get("maximum_wall_over_ideal", 0)) < 1 + or float(worker.get("maximum_p95_world_state_age_ms", 0)) <= 0 + or float(worker.get("maximum_p99_world_state_age_ms", 0)) <= 0 + or int(worker.get("minimum_runtime_telemetry_samples", 0)) < 2 + or int(worker.get("minimum_gpu_telemetry_samples", 0)) < 2 + ): + raise RuntimeError("LAB E21 profile contract is invalid") + return profile, _sha256(path) + + +def _artifact_map(root: Path, result: dict[str, Any]) -> dict[str, Path]: + artifacts = result.get("artifacts") + if not isinstance(artifacts, list): + raise RuntimeError("worker result artifacts are missing") + validated: dict[str, Path] = {} + for artifact in artifacts: + if not isinstance(artifact, dict): + raise RuntimeError("worker result artifact is invalid") + kind = artifact.get("kind") + relative = artifact.get("path") + if not isinstance(kind, str) or not isinstance(relative, str) or kind in validated: + raise RuntimeError("worker result artifact identity is invalid") + path = (root / relative).resolve(strict=True) + if ( + path.parent != root + or path.is_symlink() + or artifact.get("byte_length") != path.stat().st_size + or artifact.get("sha256") != _sha256(path) + ): + raise RuntimeError("worker result artifact integrity failed") + validated[kind] = path + return validated + + +def _all_queues_bounded(source: dict[str, Any], worker: dict[str, Any]) -> bool: + ingress = source.get("ingress", {}) + source_queues = ingress.get("queues", {}) if isinstance(ingress, dict) else {} + if not isinstance(source_queues, dict) or not source_queues: + return False + for value in source_queues.values(): + if ( + not isinstance(value, dict) + or int(value.get("maximum_depth", -1)) > int(value.get("capacity", -2)) + or int(value.get("depth", -1)) != 0 + ): + return False + metrics = worker.get("metrics", {}) + detector = metrics.get("detector", {}) if isinstance(metrics, dict) else {} + semantic = metrics.get("semantic", {}) if isinstance(metrics, dict) else {} + for value in (detector.get("queue"), semantic.get("queue")): + if ( + not isinstance(value, dict) + or int(value.get("maximum_depth", -1)) > int(value.get("capacity", -2)) + or int(value.get("final_depth", -1)) != 0 + ): + return False + return True + + +def evaluate( + *, + source_report_path: Path, + worker_result_root: Path, + profile_path: Path, +) -> dict[str, Any]: + source_path = source_report_path.resolve(strict=True) + worker_root = worker_result_root.resolve(strict=True) + profile, profile_sha256 = _validate_profile(profile_path) + source = _read_object(source_path) + result = _read_object(worker_root / "result.json") + worker = _read_object(worker_root / "run-report.json") + if ( + source.get("schema_version") != SOURCE_SCHEMA + or source.get("state") != "completed" + or result.get("schema_version") != WORKER_RESULT_SCHEMA + or worker.get("schema_version") != WORKER_REPORT_SCHEMA + or result.get("result_id") != worker.get("result_id") + ): + raise RuntimeError("LAB E21 source/worker contract is incompatible") + artifacts = _artifact_map(worker_root, result) + world_path = artifacts.get("e15-world-state") + runtime_path = artifacts.get("worker-runtime-telemetry") + if world_path is None or runtime_path is None: + raise RuntimeError("LAB E21 telemetry artifacts are incomplete") + world_rows = _read_jsonl(world_path) + runtime_rows = _read_jsonl(runtime_path, maximum_line_bytes=512 * 1024) + age_samples: list[tuple[float, float]] = [] + for row in world_rows: + if not isinstance(row, dict) or not isinstance(row.get("delivery"), dict): + raise RuntimeError("LAB E21 world-state row is invalid") + session_seconds = row.get("session_seconds") + result_age_ms = row["delivery"].get("result_age_ms") + if not isinstance(session_seconds, int | float) or not isinstance( + result_age_ms, int | float + ): + raise RuntimeError("LAB E21 world-state timing is invalid") + age_samples.append((float(session_seconds), float(result_age_ms))) + rss_samples: list[tuple[float, float]] = [] + for row in runtime_rows: + if ( + not isinstance(row, dict) + or row.get("schema_version") != "missioncore.worker-runtime-telemetry/v1" + or not isinstance(row.get("elapsed_seconds"), int | float) + or not isinstance(row.get("process_rss_mib"), int | float) + ): + raise RuntimeError("LAB E21 runtime telemetry row is invalid") + rss_samples.append( + (float(row["elapsed_seconds"]), float(row["process_rss_mib"])) + ) + age_trend = _trend(age_samples) + rss_trend = _trend(rss_samples) + source_policy = profile["source"] + worker_policy = profile["worker"] + source_descriptor = source["source"] + source_pacing = source["pacing"] + source_queues = source["ingress"]["queues"] + metrics = worker["metrics"] + detector = metrics["detector"] + semantic = metrics["semantic"] + fusion = metrics["fusion"] + transport = metrics["transport"] + latency = metrics["latency_ms"]["world_state_age_ms"] + gpu = metrics["gpu_telemetry"] + source_drop_total = sum( + int(queue["dropped_overflow"]) + int(queue["rejected_oversize"]) + for queue in source_queues.values() + ) + detector_published = int(detector["queue"]["published"]) + semantic_published = int(semantic["queue"]["published"]) + result_attempted = int(transport["results_published"]) + int( + transport["results_dropped"] + ) + diagnostic_results = source.get("diagnostic_results") + if not isinstance(diagnostic_results, dict) or not isinstance( + diagnostic_results.get("queue"), dict + ): + raise RuntimeError("LAB E21 diagnostic result sink is missing") + result_sink_queue = diagnostic_results["queue"] + worker_acceptance_checks = worker["acceptance"]["checks"] + superseded_worker_checks = {"detector_maximum_drop_fraction"} + checks = { + "source_identity": source_descriptor["session"] == source_policy["session"], + "source_speed_is_1x": float(source_descriptor["speed"]) + == float(source_policy["speed"]), + "source_span": float(source_pacing["source_span_seconds"]) + >= float(source_policy["minimum_span_seconds"]), + "source_wall_over_ideal": float(source_pacing["wall_over_ideal"]) + <= float(source_policy["maximum_wall_over_ideal"]), + "source_release_lag": float(source_pacing["release_lag_ms"]["p95"]) + <= float(source_policy["maximum_p95_release_lag_ms"]), + "source_streams_without_look_ahead": ( + source_descriptor["look_ahead"] is False + and source_descriptor["payload_loading"] == "streaming-one-event-per-modality" + ), + "source_zero_drops": source_drop_total == 0, + "diagnostic_result_sink_bounded": ( + int(result_sink_queue["maximum_depth"]) <= int(result_sink_queue["capacity"]) + and int(diagnostic_results["received"]) == int(transport["results_published"]) + ), + "queues_remained_bounded": _all_queues_bounded(source, worker), + "worker_non_detector_acceptance": all( + bool(passed) + for name, passed in worker_acceptance_checks.items() + if name not in superseded_worker_checks + ), + "detector_fps": float(detector["effective_fps"]) + >= float(worker_policy["minimum_detector_fps"]), + "detector_drop_fraction": int(detector["queue"]["dropped_overflow"]) + / max(1, detector_published) + <= float(worker_policy["maximum_detector_drop_fraction"]), + "semantic_fps": float(semantic["effective_fps"]) + >= float(worker_policy["minimum_semantic_fps"]), + "semantic_drop_fraction": int(semantic["queue"]["dropped_overflow"]) + / max(1, semantic_published) + <= float(worker_policy["maximum_semantic_drop_fraction"]), + "fusion_fraction": float(fusion["fused_fraction"]) + >= float(worker_policy["minimum_fused_fraction"]), + "result_drop_fraction": int(transport["results_dropped"]) + / max(1, result_attempted) + <= float(worker_policy["maximum_result_drop_fraction"]), + "world_state_age_p95": float(latency["p95"]) + <= float(worker_policy["maximum_p95_world_state_age_ms"]), + "world_state_age_p99": float(latency["p99"]) + <= float(worker_policy["maximum_p99_world_state_age_ms"]), + "world_state_age_does_not_accumulate": ( + float(age_trend["slope_per_second"]) + <= float(worker_policy["maximum_age_slope_ms_per_second"]) + and float(age_trend["late_vs_early_p95"]) + <= float(worker_policy["maximum_late_vs_early_p95_age_growth_ms"]) + ), + "rss_does_not_accumulate": float(rss_trend["late_vs_early_p95"]) + <= float(worker_policy["maximum_rss_growth_mib"]), + "runtime_telemetry_coverage": len(runtime_rows) + >= int(worker_policy["minimum_runtime_telemetry_samples"]), + "gpu_telemetry_coverage": int(gpu["sample_count"]) + >= int(worker_policy["minimum_gpu_telemetry_samples"]), + "transport_complete": ( + int(transport["ingress_sequence_gaps"]) == 0 + and int(transport["camera_sequence_gaps"]) == 0 + and transport["session_end_seen"] is True + and transport["timed_out"] is False + ), + "authority_remains_diagnostic": worker["acceptance"]["commands_enabled"] is False + and worker["acceptance"]["navigation_or_safety_accepted"] is False, + } + identity = { + "schema_version": "missioncore.e21-realtime-envelope-identity/v1", + "profile_sha256": profile_sha256, + "source_report_sha256": _sha256(source_path), + "worker_result_sha256": _sha256(worker_root / "result.json"), + "source_session_id": source["session_id"], + "worker_result_id": result["result_id"], + } + identity_sha256 = hashlib.sha256(_canonical(identity)).hexdigest() + return { + "schema_version": REPORT_SCHEMA, + "result_id": f"e21-realtime-envelope-{identity_sha256}", + "created_at_utc": datetime.now(UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z"), + "state": "accepted" if all(checks.values()) else "rejected", + "identity": identity, + "checks": checks, + "metrics": { + "source": { + "pacing": source_pacing, + "events_selected": source["events_selected"], + "events_admitted": source["events_admitted"], + "drop_total": source_drop_total, + "process": source["process"], + "diagnostic_results": diagnostic_results, + }, + "worker": { + "detector": detector, + "semantic": semantic, + "fusion": fusion, + "transport": transport, + "world_state_age_ms": latency, + "world_state_age_trend": age_trend, + "runtime_telemetry": metrics["runtime_telemetry"], + "rss_trend": rss_trend, + "gpu_telemetry": gpu, + "process_cpu_seconds": metrics["process_cpu_seconds"], + "process_peak_rss_mib": metrics["process_peak_rss_mib"], + "cuda_peak_memory_allocated_mib": metrics[ + "cuda_peak_memory_allocated_mib" + ], + "cuda_peak_memory_reserved_mib": metrics[ + "cuda_peak_memory_reserved_mib" + ], + "disk": metrics["disk"], + }, + }, + "authority": profile["authority"], + "interpretation": { + "accepted_scope": "recorded-1x-warm-worker-steady-state-only", + "physical_k1_live_accepted": False, + "navigation_or_safety_accepted": False, + }, + } + + +def publish(report: dict[str, Any], output_root: Path) -> Path: + root = output_root.resolve() + root.mkdir(mode=0o700, parents=True, exist_ok=True) + destination = root / str(report["result_id"]) + if destination.exists(): + existing = _read_object(destination / "report.json") + if existing.get("identity") != report.get("identity"): + raise RuntimeError("existing LAB E21 result identity differs") + return destination + staging = root / f".{report['result_id']}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700) + try: + report_path = staging / "report.json" + with report_path.open("x", encoding="utf-8", newline="\n") as stream: + json.dump(report, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + result = { + "schema_version": RESULT_SCHEMA, + "result_id": report["result_id"], + "state": report["state"], + "report": { + "path": "report.json", + "byte_length": report_path.stat().st_size, + "sha256": _sha256(report_path), + "schema_version": REPORT_SCHEMA, + }, + } + with (staging / "result.json").open("x", encoding="utf-8", newline="\n") as stream: + json.dump(result, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + staging.replace(destination) + finally: + if staging.exists(): + for path in staging.iterdir(): + path.unlink() + staging.rmdir() + return destination + + +def main() -> int: + root = Path(__file__).resolve().parents[2] + parser = argparse.ArgumentParser() + parser.add_argument("--source-report", type=Path, required=True) + parser.add_argument("--worker-result-root", type=Path, required=True) + parser.add_argument( + "--profile", + type=Path, + default=Path(__file__).with_name("e21_realtime_envelope_profile.json"), + ) + parser.add_argument( + "--output-root", + type=Path, + default=root / ".runtime" / "compute-experiments" / "e21" / "results", + ) + args = parser.parse_args() + report = evaluate( + source_report_path=args.source_report, + worker_result_root=args.worker_result_root, + profile_path=args.profile, + ) + destination = publish(report, args.output_root) + print( + json.dumps( + { + "result_root": str(destination), + "result_id": report["result_id"], + "state": report["state"], + "failed_checks": [ + name for name, passed in report["checks"].items() if not passed + ], + }, + sort_keys=True, + ) + ) + return 0 if report["state"] == "accepted" else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/e21_realtime_envelope_profile.json b/experiments/perception/e21_realtime_envelope_profile.json new file mode 100644 index 0000000..01c2447 --- /dev/null +++ b/experiments/perception/e21_realtime_envelope_profile.json @@ -0,0 +1,32 @@ +{ + "schema_version": "missioncore.e21-realtime-envelope-profile/v1", + "mode": "steady-state-baseline", + "source": { + "session": "20260720T065719Z_viewer_live", + "display_name": "RAVNOVES00", + "speed": 1.0, + "minimum_span_seconds": 55.0, + "maximum_wall_over_ideal": 1.08, + "maximum_p95_release_lag_ms": 50.0, + "require_streaming_without_look_ahead": true + }, + "worker": { + "minimum_detector_fps": 9.5, + "maximum_detector_drop_fraction": 0.02, + "minimum_semantic_fps": 1.8, + "maximum_semantic_drop_fraction": 0.05, + "minimum_fused_fraction": 0.85, + "maximum_result_drop_fraction": 0.01, + "maximum_p95_world_state_age_ms": 200.0, + "maximum_p99_world_state_age_ms": 300.0, + "maximum_age_slope_ms_per_second": 1.0, + "maximum_late_vs_early_p95_age_growth_ms": 50.0, + "maximum_rss_growth_mib": 256.0, + "minimum_runtime_telemetry_samples": 50, + "minimum_gpu_telemetry_samples": 50 + }, + "authority": { + "commands_enabled": false, + "navigation_or_safety_accepted": false + } +} diff --git a/experiments/perception/run_e12_shadow_transport_replay.py b/experiments/perception/run_e12_shadow_transport_replay.py index df2973f..c4eb2c3 100644 --- a/experiments/perception/run_e12_shadow_transport_replay.py +++ b/experiments/perception/run_e12_shadow_transport_replay.py @@ -5,11 +5,18 @@ from __future__ import annotations import argparse import hashlib +import heapq +import itertools import json +import math import os +import resource +import statistics +import sys import threading import time -from dataclasses import dataclass +from collections.abc import Iterator +from dataclasses import asdict, dataclass from datetime import UTC, datetime from pathlib import Path from typing import Literal @@ -17,7 +24,13 @@ from typing import Literal import uvicorn from fastapi import FastAPI -from k1link.compute.live_perception import LiveIngressModality, LivePerceptionIngress +from k1link.compute.live_perception import ( + LatestWinsQueue, + LiveIngressModality, + LivePerceptionIngress, + LivePerceptionResultFrame, + decode_live_perception_result, +) from k1link.device_plugins.xgrids_k1.live_perception_shadow import ( build_live_perception_shadow_router, ensure_live_shadow_token, @@ -26,6 +39,7 @@ from k1link.device_plugins.xgrids_k1.viewer.replay import iter_replay_messages REPORT_SCHEMA = "missioncore.e12-shadow-transport-source-report/v1" PLUGIN_ID = "nodedc.device.xgrids-lixelkity-k1" +_EXPERIMENT_ID = "e12" @dataclass(frozen=True, slots=True) @@ -46,21 +60,55 @@ def _sha256(path: Path) -> str: return digest.hexdigest() -def _camera_events(epoch_root: Path, duration_seconds: float) -> list[ReplayEvent]: +def _percentiles(values: list[float]) -> dict[str, float | int]: + if not values: + return { + "count": 0, + "mean": 0.0, + "p50": 0.0, + "p95": 0.0, + "p99": 0.0, + "max": 0.0, + } + ordered = sorted(values) + + def percentile(fraction: float) -> float: + index = min(len(ordered) - 1, max(0, round((len(ordered) - 1) * fraction))) + return round(ordered[index], 6) + + return { + "count": len(values), + "mean": round(statistics.fmean(values), 6), + "p50": percentile(0.50), + "p95": percentile(0.95), + "p99": percentile(0.99), + "max": round(max(values), 6), + } + + +def _camera_events(epoch_root: Path, duration_seconds: float) -> Iterator[ReplayEvent]: index_path = epoch_root / "index.jsonl" if not index_path.is_file(): raise RuntimeError("camera index is missing") - events: list[ReplayEvent] = [] - start_epoch_ns: int | None = None - end_epoch_ns: int | None = None with index_path.open("r", encoding="utf-8") as stream: - for line in stream: + first_line = stream.readline() + if not first_line: + raise RuntimeError("camera replay interval is empty") + first_record = json.loads(first_line) + start_epoch_ns = int(first_record["host_epoch_ns"]) + end_epoch_ns = start_epoch_ns + int(duration_seconds * 1_000_000_000) + init_payload = (epoch_root / "init.mp4").read_bytes() + yield ReplayEvent( + modality="camera-init", + source_id="sensor.camera.right", + source_sequence=0, + epoch_ns=start_epoch_ns - 1, + monotonic_ns=int(first_record["host_monotonic_ns"]) - 1, + payload=init_payload, + ) + for line in itertools.chain((first_line,), stream): record = json.loads(line) epoch_ns = int(record["host_epoch_ns"]) - if start_epoch_ns is None: - start_epoch_ns = epoch_ns - end_epoch_ns = epoch_ns + int(duration_seconds * 1_000_000_000) - assert end_epoch_ns is not None if epoch_ns > end_epoch_ns: break payload_path = epoch_root / str(record["path"]) @@ -69,31 +117,14 @@ def _camera_events(epoch_root: Path, duration_seconds: float) -> list[ReplayEven raise RuntimeError("camera replay segment length differs from its index") if hashlib.sha256(payload).hexdigest() != record["sha256"]: raise RuntimeError("camera replay segment digest differs from its index") - events.append( - ReplayEvent( - modality="camera-frame", - source_id="sensor.camera.right", - source_sequence=int(record["sequence"]), - epoch_ns=epoch_ns, - monotonic_ns=int(record["host_monotonic_ns"]), - payload=payload, - ) + yield ReplayEvent( + modality="camera-frame", + source_id="sensor.camera.right", + source_sequence=int(record["sequence"]), + epoch_ns=epoch_ns, + monotonic_ns=int(record["host_monotonic_ns"]), + payload=payload, ) - if not events or start_epoch_ns is None: - raise RuntimeError("camera replay interval is empty") - init_payload = (epoch_root / "init.mp4").read_bytes() - events.insert( - 0, - ReplayEvent( - modality="camera-init", - source_id="sensor.camera.right", - source_sequence=0, - epoch_ns=start_epoch_ns - 1, - monotonic_ns=events[0].monotonic_ns - 1, - payload=init_payload, - ), - ) - return events def _mqtt_events( @@ -101,9 +132,8 @@ def _mqtt_events( *, start_epoch_ns: int, duration_seconds: float, -) -> list[ReplayEvent]: +) -> Iterator[ReplayEvent]: end_epoch_ns = start_epoch_ns + int(duration_seconds * 1_000_000_000) - events: list[ReplayEvent] = [] for message in iter_replay_messages(capture_path): if message.received_at_epoch_ns < start_epoch_ns: continue @@ -115,17 +145,14 @@ def _mqtt_events( modality = "pose" else: continue - events.append( - ReplayEvent( - modality=modality, - source_id=message.topic, - source_sequence=message.sequence, - epoch_ns=message.received_at_epoch_ns, - monotonic_ns=message.received_monotonic_ns or 0, - payload=message.payload, - ) + yield ReplayEvent( + modality=modality, + source_id=message.topic, + source_sequence=message.sequence, + epoch_ns=message.received_at_epoch_ns, + monotonic_ns=message.received_monotonic_ns or 0, + payload=message.payload, ) - return events def _wait_for(predicate: object, *, timeout_seconds: float, label: str) -> None: @@ -140,21 +167,50 @@ def _wait_for(predicate: object, *, timeout_seconds: float, label: str) -> None: def run(args: argparse.Namespace) -> dict[str, object]: + experiment_id = str(getattr(args, "experiment_id", _EXPERIMENT_ID)) + speed = float(getattr(args, "speed", 1.0)) + if experiment_id not in {"e12", "e21"} or not math.isfinite(speed) or not 0.1 <= speed <= 10: + raise RuntimeError("replay experiment identity or speed is invalid") repository_root = Path(args.repository_root).expanduser().resolve() session_root = Path(args.session_root).expanduser().resolve() epoch_root = session_root / "media" / "sensor.camera.right" / "epoch-1" capture_path = session_root / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt" - camera = _camera_events(epoch_root, args.duration_seconds) - mqtt = _mqtt_events( - capture_path, - start_epoch_ns=camera[1].epoch_ns, - duration_seconds=args.duration_seconds, + camera = iter(_camera_events(epoch_root, args.duration_seconds)) + first_camera = next(camera, None) + first_frame = next(camera, None) + if first_camera is None or first_frame is None or first_frame.modality != "camera-frame": + raise RuntimeError("camera replay interval is empty") + camera = itertools.chain((first_camera, first_frame), camera) + mqtt = iter( + _mqtt_events( + capture_path, + start_epoch_ns=first_frame.epoch_ns, + duration_seconds=args.duration_seconds, + ) ) - events = sorted((*camera, *mqtt), key=lambda event: (event.epoch_ns, event.modality)) - if not mqtt: + first_mqtt = next(mqtt, None) + if first_mqtt is None: raise RuntimeError("selected replay interval contains no LiDAR or pose events") + mqtt = itertools.chain((first_mqtt,), mqtt) + events = heapq.merge( + camera, + mqtt, + key=lambda event: (event.epoch_ns, event.modality), + ) ingress = LivePerceptionIngress() + result_queue = LatestWinsQueue[LivePerceptionResultFrame](capacity=2) + result_state = {"received": 0, "payload_bytes": 0} + result_lock = threading.Lock() + + def receive_result(payload: bytes) -> bool: + frame = decode_live_perception_result(payload) + result_queue.publish(frame) + with result_lock: + result_state["received"] += 1 + result_state["payload_bytes"] += len(payload) + return True + _, token = ensure_live_shadow_token(repository_root) app = FastAPI() app.include_router( @@ -162,6 +218,7 @@ def run(args: argparse.Namespace) -> dict[str, object]: ingress, PLUGIN_ID, bearer_token=token, + result_receiver=receive_result, ) ) server = uvicorn.Server( @@ -177,7 +234,7 @@ def run(args: argparse.Namespace) -> dict[str, object]: server_thread.start() _wait_for(lambda: server.started, timeout_seconds=10.0, label="E12 source server") - session_id = f"e12-ravnoves00-{int(time.time())}" + session_id = f"{experiment_id}-ravnoves00-{time.time_ns()}" ingress.begin_session(session_id) _wait_for( lambda: ingress.snapshot()["consumer_connected"], @@ -185,14 +242,27 @@ def run(args: argparse.Namespace) -> dict[str, object]: label="exclusive shadow worker", ) started_monotonic = time.monotonic() - source_start_ns = events[0].epoch_ns + source_start_ns: int | None = None + source_last_ns: int | None = None published: dict[str, int] = {} + selected: dict[str, int] = {} + release_lag_ms: list[float] = [] + publish_call_ms: list[float] = [] + process_cpu_started = time.process_time() try: for event in events: - target = started_monotonic + (event.epoch_ns - source_start_ns) / 1_000_000_000 + if source_start_ns is None: + source_start_ns = event.epoch_ns + source_last_ns = event.epoch_ns + selected[event.modality] = selected.get(event.modality, 0) + 1 + target = started_monotonic + ( + (event.epoch_ns - source_start_ns) / 1_000_000_000 / speed + ) remaining = target - time.monotonic() if remaining > 0: time.sleep(remaining) + release_lag_ms.append(max(0.0, (time.monotonic() - target) * 1000)) + publish_started = time.perf_counter() accepted = ingress.publish( modality=event.modality, source_id=event.source_id, @@ -201,12 +271,13 @@ def run(args: argparse.Namespace) -> dict[str, object]: received_monotonic_ns=event.monotonic_ns, payload=event.payload, ) + publish_call_ms.append((time.perf_counter() - publish_started) * 1000) if accepted: published[event.modality] = published.get(event.modality, 0) + 1 ingress.end_session(session_id) _wait_for( lambda: not ingress.snapshot()["consumer_connected"], - timeout_seconds=10.0, + timeout_seconds=float(getattr(args, "completion_timeout_seconds", 30.0)), label="shadow worker report completion", ) finally: @@ -217,8 +288,19 @@ def run(args: argparse.Namespace) -> dict[str, object]: completed_monotonic = time.monotonic() source_snapshot = ingress.snapshot() + if source_start_ns is None or source_last_ns is None: + raise RuntimeError("selected replay interval is empty") + source_span_seconds = max(0.0, (source_last_ns - source_start_ns) / 1_000_000_000) + wall_seconds = completed_monotonic - started_monotonic + report_schema = ( + REPORT_SCHEMA + if experiment_id == _EXPERIMENT_ID + else "missioncore.e21-replay-source-report/v1" + ) + peak_rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + peak_rss_mib = peak_rss / (1024 * 1024 if sys.platform == "darwin" else 1024) report: dict[str, object] = { - "schema_version": REPORT_SCHEMA, + "schema_version": report_schema, "state": "completed", "session_id": session_id, "source": { @@ -227,17 +309,32 @@ def run(args: argparse.Namespace) -> dict[str, object]: "camera_source": "sensor.camera.right", "camera_epoch": 1, "duration_seconds": args.duration_seconds, + "selected_span_seconds": source_span_seconds, + "speed": speed, + "pacing_clock": "recorded-host-arrival", + "look_ahead": False, + "payload_loading": "streaming-one-event-per-modality", "mqtt_raw_sha256": _sha256(capture_path), "camera_index_sha256": _sha256(epoch_root / "index.jsonl"), }, - "events_selected": { - "camera-init": sum(event.modality == "camera-init" for event in events), - "camera-frame": sum(event.modality == "camera-frame" for event in events), - "lidar": sum(event.modality == "lidar" for event in events), - "pose": sum(event.modality == "pose" for event in events), - }, + "events_selected": selected, "events_admitted": published, - "wall_seconds": completed_monotonic - started_monotonic, + "wall_seconds": wall_seconds, + "pacing": { + "source_span_seconds": source_span_seconds, + "ideal_wall_seconds": source_span_seconds / speed, + "wall_over_ideal": wall_seconds / max(source_span_seconds / speed, 1e-9), + "release_lag_ms": _percentiles(release_lag_ms), + "publish_call_ms": _percentiles(publish_call_ms), + }, + "process": { + "cpu_seconds": time.process_time() - process_cpu_started, + "peak_rss_mib": peak_rss_mib, + }, + "diagnostic_results": { + **result_state, + "queue": asdict(result_queue.snapshot()), + }, "ingress": source_snapshot, "authority": { "mode": "shadow-diagnostic-only", @@ -245,11 +342,11 @@ def run(args: argparse.Namespace) -> dict[str, object]: "navigation_or_safety_accepted": False, }, } - output_root = repository_root / ".runtime" / "compute-experiments" / "e12" + output_root = repository_root / ".runtime" / "compute-experiments" / experiment_id output_root.mkdir(parents=True, exist_ok=True) stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") - temporary = output_root / f".{stamp}-e12-source.json.tmp" - destination = output_root / f"{stamp}-e12-source.json" + temporary = output_root / f".{stamp}-{experiment_id}-source.json.tmp" + destination = output_root / f"{stamp}-{experiment_id}-source.json" encoded = json.dumps(report, indent=2, sort_keys=True).encode("utf-8") + b"\n" with temporary.open("xb") as stream: stream.write(encoded) @@ -277,7 +374,10 @@ def _arguments() -> argparse.Namespace: ) parser.add_argument("--port", type=int, default=8012) parser.add_argument("--duration-seconds", type=float, default=15.0) + parser.add_argument("--speed", type=float, default=1.0) + parser.add_argument("--experiment-id", choices=("e12", "e21"), default="e12") parser.add_argument("--consumer-timeout-seconds", type=float, default=30.0) + parser.add_argument("--completion-timeout-seconds", type=float, default=30.0) return parser.parse_args() diff --git a/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 b/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 index 36e3775..16db93e 100644 --- a/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 +++ b/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 @@ -6,6 +6,7 @@ param( [string]$PersistentContainer = "mission-core-perception-worker", [string]$PersistentOutputRoot = "D:\NDC_MISSIONCORE\runtime\derived\.perception-persistent-publish", [ValidateRange(1024, 65535)] [int]$PersistentPort = 18020, + [ValidateRange(5, 3600)] [int]$MaximumDurationSeconds = 20, [ValidateRange(1, 1000)] [int]$FreeGiBFloor = 360 ) @@ -48,6 +49,7 @@ $request = @{ request_id = $RequestId output_name = $outputName token = $token + max_duration_seconds = $MaximumDurationSeconds } | ConvertTo-Json -Compress $token = $null $client = ( diff --git a/experiments/perception/worker/e15_shadow_runtime.py b/experiments/perception/worker/e15_shadow_runtime.py index 676a322..3d3606b 100644 --- a/experiments/perception/worker/e15_shadow_runtime.py +++ b/experiments/perception/worker/e15_shadow_runtime.py @@ -184,6 +184,7 @@ class PersistentFmp4Decoder: self._started = False self._init_seen = False self._last_source_sequence: int | None = None + self._source_sequence_gaps = 0 self._decoded_frames = 0 self._failure: BaseException | None = None @@ -202,11 +203,18 @@ class PersistentFmp4Decoder: def feed_segment(self, metadata: CameraFragmentMetadata, payload: bytes) -> None: if not self._init_seen: raise ShadowRuntimeError("camera media arrived before init") - expected = 1 if self._last_source_sequence is None else self._last_source_sequence + 1 - if metadata.source_sequence != expected: + if metadata.source_sequence < 1: + raise ShadowRuntimeError("camera source sequence must be positive") + if ( + self._last_source_sequence is not None + and metadata.source_sequence <= self._last_source_sequence + ): raise ShadowRuntimeError( - f"camera source sequence gap: expected {expected}, got {metadata.source_sequence}" + "camera source sequence is not increasing: " + f"previous {self._last_source_sequence}, got {metadata.source_sequence}" ) + expected = 1 if self._last_source_sequence is None else self._last_source_sequence + 1 + self._source_sequence_gaps += metadata.source_sequence - expected self._metadata.publish(metadata) try: self._media.append(payload) @@ -234,6 +242,7 @@ class PersistentFmp4Decoder: return { "init_seen": self._init_seen, "last_source_sequence": self._last_source_sequence, + "source_sequence_gaps": self._source_sequence_gaps, "decoded_frames": self._decoded_frames, "failed": self._failure is not None, "media": self._media.snapshot(), diff --git a/experiments/perception/worker/run_e15_shadow_inference.py b/experiments/perception/worker/run_e15_shadow_inference.py index f6464ac..beb844a 100644 --- a/experiments/perception/worker/run_e15_shadow_inference.py +++ b/experiments/perception/worker/run_e15_shadow_inference.py @@ -8,6 +8,7 @@ import hashlib import importlib.metadata import io import json +import math import os import platform import queue @@ -347,6 +348,7 @@ class _TransportState: last_ingress_sequence: int | None = None ingress_sequence_gaps: int = 0 camera_sequence_gaps: int = 0 + last_camera_source_sequence: int | None = None session_id: str | None = None session_end_seen: bool = False timed_out: bool = False @@ -360,6 +362,175 @@ class _TransportState: self.failures = [] +class _RuntimeTelemetry: + """Sample process, host-memory and bounded-queue state during one warm run.""" + + def __init__( + self, + stream: Any, + *, + interval_seconds: float, + snapshotters: dict[str, Any], + ) -> None: + if not 0.25 <= interval_seconds <= 10: + raise RuntimeError("runtime telemetry interval is outside bounds") + self._stream = stream + self._interval = interval_seconds + self._snapshotters = snapshotters + self._stop = threading.Event() + self._thread = threading.Thread( + target=self._run, + name="runtime-telemetry", + daemon=True, + ) + self._started = time.perf_counter() + self._previous_wall = self._started + self._previous_cpu = time.process_time() + self.samples: list[dict[str, Any]] = [] + + def __enter__(self) -> _RuntimeTelemetry: + self._thread.start() + return self + + def __exit__(self, *_: object) -> None: + self._stop.set() + self._thread.join(timeout=self._interval + 5) + self._sample() + + @staticmethod + def _read_number(path: str) -> int | None: + try: + value = Path(path).read_text(encoding="ascii").strip() + if value == "max": + return None + return int(value) + except (OSError, ValueError): + return None + + @staticmethod + def _process_status() -> tuple[float, int]: + rss_mib = 0.0 + thread_count = 0 + try: + fields = Path("/proc/self/statm").read_text(encoding="ascii").split() + rss_mib = int(fields[1]) * os.sysconf("SC_PAGE_SIZE") / 2**20 + except (OSError, ValueError, IndexError): + pass + try: + for line in Path("/proc/self/status").read_text(encoding="ascii").splitlines(): + if line.startswith("Threads:"): + thread_count = int(line.split(":", 1)[1].strip()) + break + except (OSError, ValueError): + pass + return rss_mib, thread_count + + @staticmethod + def _memory_status() -> tuple[float, float]: + total_kib = 0 + available_kib = 0 + try: + for line in Path("/proc/meminfo").read_text(encoding="ascii").splitlines(): + if line.startswith("MemTotal:"): + total_kib = int(line.split()[1]) + elif line.startswith("MemAvailable:"): + available_kib = int(line.split()[1]) + except (OSError, ValueError, IndexError): + pass + return total_kib / 1024, available_kib / 1024 + + @staticmethod + def _io_status() -> tuple[int, int]: + read_bytes = 0 + write_bytes = 0 + try: + for line in Path("/proc/self/io").read_text(encoding="ascii").splitlines(): + if line.startswith("read_bytes:"): + read_bytes = int(line.split(":", 1)[1].strip()) + elif line.startswith("write_bytes:"): + write_bytes = int(line.split(":", 1)[1].strip()) + except (OSError, ValueError): + pass + return read_bytes, write_bytes + + def _sample(self) -> None: + now = time.perf_counter() + process_cpu = time.process_time() + wall_delta = max(1e-9, now - self._previous_wall) + cpu_percent = max(0.0, (process_cpu - self._previous_cpu) / wall_delta * 100) + self._previous_wall = now + self._previous_cpu = process_cpu + rss_mib, thread_count = self._process_status() + memory_total_mib, memory_available_mib = self._memory_status() + read_bytes, write_bytes = self._io_status() + cgroup_current = self._read_number("/sys/fs/cgroup/memory.current") + cgroup_limit = self._read_number("/sys/fs/cgroup/memory.max") + queues: dict[str, Any] = {} + for name, snapshotter in self._snapshotters.items(): + try: + queues[name] = snapshotter() + except (RuntimeError, ValueError, OSError): + queues[name] = {"state": "snapshot-unavailable"} + sample = { + "schema_version": "missioncore.worker-runtime-telemetry/v1", + "elapsed_seconds": round(now - self._started, 6), + "process_cpu_percent": round(cpu_percent, 6), + "process_rss_mib": round(rss_mib, 6), + "process_threads": thread_count, + "process_read_bytes": read_bytes, + "process_write_bytes": write_bytes, + "system_memory_total_mib": round(memory_total_mib, 6), + "system_memory_available_mib": round(memory_available_mib, 6), + "cgroup_memory_current_mib": ( + round(cgroup_current / 2**20, 6) if cgroup_current is not None else None + ), + "cgroup_memory_limit_mib": ( + round(cgroup_limit / 2**20, 6) if cgroup_limit is not None else None + ), + "queues": queues, + } + self.samples.append(sample) + self._stream.write(json.dumps(sample, sort_keys=True) + "\n") + self._stream.flush() + + def _run(self) -> None: + while not self._stop.wait(self._interval): + self._sample() + + def summary(self) -> dict[str, Any]: + numeric_fields = ( + "process_cpu_percent", + "process_rss_mib", + "process_threads", + "system_memory_available_mib", + "cgroup_memory_current_mib", + ) + summary: dict[str, Any] = { + "sample_count": len(self.samples), + "interval_seconds": self._interval, + } + for field in numeric_fields: + values = [ + float(sample[field]) + for sample in self.samples + if isinstance(sample.get(field), int | float) + ] + summary[field] = _percentiles(values) + if self.samples: + quarter = max(1, len(self.samples) // 4) + early = [float(value["process_rss_mib"]) for value in self.samples[:quarter]] + late = [float(value["process_rss_mib"]) for value in self.samples[-quarter:]] + summary["rss_growth_mib"] = round( + float(_percentiles(late)["p95"]) - float(_percentiles(early)["p95"]), + 6, + ) + summary["final_queues"] = self.samples[-1]["queues"] + else: + summary["rss_growth_mib"] = 0.0 + summary["final_queues"] = {} + return summary + + def _send_client_binary_frame(stream: Any, payload: bytes) -> None: if not payload or len(payload) > LIVE_RESULT_MAX_PAYLOAD_BYTES + 256 * 1024 + 8: raise ShadowRuntimeError("live result websocket frame exceeds the bound") @@ -478,20 +649,31 @@ def _receiver( elif modality == "camera-init": decoder.feed_init(payload) elif modality == "camera-frame": - try: - decoder.feed_segment( - CameraFragmentMetadata( - ingress_sequence=sequence, - source_sequence=int(header["source_sequence"]), - captured_at_epoch_ns=int(header["captured_at_epoch_ns"]), - worker_received_monotonic=worker_received, - ), - payload, + camera_source_sequence = int(header["source_sequence"]) + expected_camera_sequence = ( + 1 + if state.last_camera_source_sequence is None + else state.last_camera_source_sequence + 1 + ) + if camera_source_sequence < expected_camera_sequence: + raise ShadowRuntimeError( + "camera source sequence is not increasing: " + f"expected at least {expected_camera_sequence}, " + f"got {camera_source_sequence}" ) - except ShadowRuntimeError as exc: - if "source sequence gap" in str(exc): - state.camera_sequence_gaps += 1 - raise + state.camera_sequence_gaps += ( + camera_source_sequence - expected_camera_sequence + ) + state.last_camera_source_sequence = camera_source_sequence + decoder.feed_segment( + CameraFragmentMetadata( + ingress_sequence=sequence, + source_sequence=camera_source_sequence, + captured_at_epoch_ns=int(header["captured_at_epoch_ns"]), + worker_received_monotonic=worker_received, + ), + payload, + ) elif modality in {"lidar", "pose"}: decode_started = time.perf_counter() normalized = normalize_k1_message( @@ -729,10 +911,11 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: ) first_camera_epoch_ns: list[int] = [] last_camera_epoch_ns: list[int] = [] - decoded_frames: list[DecodedCameraFrame] = [] + decoded_frame_count = 0 callback_failures: list[BaseException] = [] def on_decoded(frame: DecodedCameraFrame) -> None: + nonlocal decoded_frame_count try: if not first_camera_epoch_ns: first_camera_epoch_ns.append(frame.metadata.captured_at_epoch_ns) @@ -756,7 +939,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: decode_ms=frame.decode_age_ms, source_release_lag_ms=0.0, ) - decoded_frames.append(frame) + decoded_frame_count += 1 detector_queue.publish(envelope) if frame.frame_index % int(scheduling["semantic_sample_every_frames"]) == 0: semantic_queue.publish(envelope) @@ -776,7 +959,6 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: result_queue: queue.Queue[bytes] = queue.Queue(maxsize=2) result_complete = threading.Event() sensor_decode_ms = {"lidar": [], "pose": []} - completed_semantics: list[Any] = [] semantic_errors: list[BaseException] = [] semantic_latency = { name: [] @@ -819,15 +1001,59 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: fusion_path = output / "fusion-frames.jsonl" world_path = output / "world-state.jsonl" gpu_path = output / "gpu-telemetry.jsonl" + runtime_path = output / "runtime-telemetry.jsonl" run_started = time.perf_counter() + process_cpu_started = time.process_time() with ( semantic_path.open("x", encoding="utf-8", newline="\n") as semantic_stream, fusion_path.open("x", encoding="utf-8", newline="\n") as fusion_stream, world_path.open("x", encoding="utf-8", newline="\n") as world_stream, gpu_path.open("x", encoding="utf-8", newline="\n") as gpu_stream, + runtime_path.open("x", encoding="utf-8", newline="\n") as runtime_stream, _GpuTelemetry(gpu_stream, 1.0) as gpu, + _RuntimeTelemetry( + runtime_stream, + interval_seconds=1.0, + snapshotters={ + "detector": detector_queue.snapshot, + "semantic": semantic_queue.snapshot, + "decoder": decoder.snapshot, + "synchronizer": synchronizer.snapshot, + "result": lambda: { + "capacity": result_queue.maxsize, + "depth": result_queue.qsize(), + "dropped_overflow": transport.results_dropped, + "published": transport.results_published, + }, + }, + ) as runtime_telemetry, ): + class SemanticResultStream: + def __init__(self) -> None: + self.count = 0 + + def append(self, result: Any) -> None: + semantic_stream.write( + json.dumps( + { + "schema_version": SEMANTIC_SCHEMA, + "frame_index": result.frame_index, + "source_frame_index": result.source_frame_index, + "session_seconds": result.session_seconds, + "completion_age_ms": result.completion_age_ms, + "mask_sha256": result.mask_sha256, + "class_pixels": result.class_pixels, + }, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + "\n" + ) + self.count += 1 + + completed_semantics = SemanticResultStream() semantic_thread = threading.Thread( target=semantic_worker, kwargs={ @@ -1101,25 +1327,17 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: or callback_failures or semantic_errors ): - raise RuntimeError("LAB E15 runtime worker failed") - for result in completed_semantics: - semantic_stream.write( - json.dumps( - { - "schema_version": SEMANTIC_SCHEMA, - "frame_index": result.frame_index, - "source_frame_index": result.source_frame_index, - "session_seconds": result.session_seconds, - "completion_age_ms": result.completion_age_ms, - "mask_sha256": result.mask_sha256, - "class_pixels": result.class_pixels, - }, - sort_keys=True, - separators=(",", ":"), - allow_nan=False, - ) - + "\n" + failures = [ + *(transport.failures or []), + *decoder_watch_failures, + *callback_failures, + *semantic_errors, + ] + summary = "; ".join( + f"{type(failure).__name__}: {str(failure)[:240]}" + for failure in failures[:8] ) + raise RuntimeError(f"LAB E15 runtime worker failed: {summary}") for stream in (semantic_stream, fusion_stream, world_stream): stream.flush() os.fsync(stream.fileno()) @@ -1136,7 +1354,8 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: detector_fps = int(detector_state["consumed"]) / max(source_span, run_wall, 1e-9) semantic_fps = int(semantic_state["consumed"]) / max(source_span, run_wall, 1e-9) semantic_scheduled = ( - (len(decoded_frames) - 1) // int(scheduling["semantic_sample_every_frames"]) + (decoded_frame_count - 1) + // int(scheduling["semantic_sample_every_frames"]) ) + 1 fresh_coverage = status_counts["fresh"] / max(1, int(detector_state["consumed"])) fused_fraction = fused_frames / max(1, int(detector_state["consumed"])) @@ -1149,17 +1368,17 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: } acceptance = live["acceptance"] checks = { - "minimum_camera_frames": len(decoded_frames) + "minimum_camera_frames": decoded_frame_count >= int(acceptance["minimum_camera_frames"]), "camera_decoder_accounting": decoder.snapshot()["decoded_frames"] == transport.counts["camera-frame"], "detector_accounting": int(detector_state["consumed"]) + int(detector_state["dropped_overflow"]) - == len(decoded_frames), + == decoded_frame_count, "detector_minimum_effective_fps": detector_fps >= float(acceptance["detector_minimum_effective_fps"]), "detector_maximum_drop_fraction": int(detector_state["dropped_overflow"]) - / len(decoded_frames) + / decoded_frame_count <= float(acceptance["detector_maximum_drop_fraction"]), "semantic_accounting": int(semantic_state["consumed"]) + int(semantic_state["dropped_overflow"]) @@ -1286,6 +1505,8 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: }, "latency_ms": latency_summary, "gpu_telemetry": gpu.summary(), + "runtime_telemetry": runtime_telemetry.summary(), + "process_cpu_seconds": time.process_time() - process_cpu_started, "process_peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, "cuda_peak_memory_allocated_mib": torch.cuda.max_memory_allocated() / 2**20, "cuda_peak_memory_reserved_mib": torch.cuda.max_memory_reserved() / 2**20, @@ -1335,6 +1556,12 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: _artifact(fusion_path, "e15-fusion-frames", "application/x-ndjson", FUSION_SCHEMA), _artifact(world_path, "e15-world-state", "application/x-ndjson", WORLD_SCHEMA), _artifact(gpu_path, "worker-gpu-telemetry", "application/x-ndjson"), + _artifact( + runtime_path, + "worker-runtime-telemetry", + "application/x-ndjson", + "missioncore.worker-runtime-telemetry/v1", + ), _artifact(report_path, "e15-run-report", "application/json", REPORT_SCHEMA), ] _write_json( @@ -1381,6 +1608,10 @@ def _persistent_run_arguments( request_id = request.get("request_id") output_name = request.get("output_name") token = request.get("token") + requested_duration = request.get( + "max_duration_seconds", + service_args.max_duration_seconds, + ) if ( not isinstance(request_id, str) or _CONTROL_ID.fullmatch(request_id) is None @@ -1389,6 +1620,10 @@ def _persistent_run_arguments( or output_name in {".", ".."} or not isinstance(token, str) or not 40 <= len(token) <= 512 + or not isinstance(requested_duration, int | float) + or isinstance(requested_duration, bool) + or not math.isfinite(float(requested_duration)) + or not 5 <= float(requested_duration) <= float(service_args.max_duration_seconds) ): raise RuntimeError("persistent worker request contract is invalid") output_root = service_args.output_root.resolve(strict=True) @@ -1401,6 +1636,7 @@ def _persistent_run_arguments( output=output, token=token, token_stdin=False, + max_duration_seconds=float(requested_duration), ) values.pop("output_root", None) values.pop("listen_host", None) @@ -1502,6 +1738,18 @@ def serve(args: argparse.Namespace) -> int: ) except Exception as exc: state["failed_runs"] += 1 + print( + json.dumps( + { + "event": "persistent-worker-run-failed", + "request_id": request_id, + "error_type": type(exc).__name__, + "error": str(exc)[:1000], + }, + sort_keys=True, + ), + flush=True, + ) self._send( 422, { diff --git a/experiments/perception/worker/run_recorded_perception_epoch.py b/experiments/perception/worker/run_recorded_perception_epoch.py index 7f6a71a..53e03ac 100644 --- a/experiments/perception/worker/run_recorded_perception_epoch.py +++ b/experiments/perception/worker/run_recorded_perception_epoch.py @@ -246,7 +246,7 @@ def _write_png(path: Path, array: Any) -> None: def _percentiles(values: list[float]) -> dict[str, float]: if not values: - return {"mean": 0.0, "p50": 0.0, "p95": 0.0, "max": 0.0} + return {"mean": 0.0, "p50": 0.0, "p95": 0.0, "p99": 0.0, "max": 0.0} ordered = sorted(values) def percentile(fraction: float) -> float: @@ -257,6 +257,7 @@ def _percentiles(values: list[float]) -> dict[str, float]: "mean": round(statistics.fmean(values), 6), "p50": percentile(0.5), "p95": percentile(0.95), + "p99": percentile(0.99), "max": round(max(values), 6), } diff --git a/src/k1link/device_plugins/xgrids_k1/live_perception_shadow.py b/src/k1link/device_plugins/xgrids_k1/live_perception_shadow.py index 6e7f6a5..74a8cf8 100644 --- a/src/k1link/device_plugins/xgrids_k1/live_perception_shadow.py +++ b/src/k1link/device_plugins/xgrids_k1/live_perception_shadow.py @@ -78,15 +78,15 @@ def build_live_perception_shadow_router( await websocket.accept() client_event = asyncio.create_task(websocket.receive()) + ingress_event = asyncio.create_task( + asyncio.to_thread( + ingress.take_next, + consumer_id, + timeout=0.5, + ) + ) try: while True: - ingress_event = asyncio.create_task( - asyncio.to_thread( - ingress.take_next, - consumer_id, - timeout=0.5, - ) - ) completed, _ = await asyncio.wait( {client_event, ingress_event}, return_when=asyncio.FIRST_COMPLETED, @@ -94,15 +94,9 @@ def build_live_perception_shadow_router( if client_event in completed: message = client_event.result() if message.get("type") == "websocket.disconnect": - ingress_event.cancel() - with suppress(asyncio.CancelledError): - await ingress_event break result = message.get("bytes") if not isinstance(result, bytes) or result_receiver is None: - ingress_event.cancel() - with suppress(asyncio.CancelledError): - await ingress_event await websocket.close( code=1008, reason="Shadow result direction is unavailable", @@ -111,32 +105,35 @@ def build_live_perception_shadow_router( try: await asyncio.to_thread(result_receiver, result) except (RuntimeError, ValueError): - ingress_event.cancel() - with suppress(asyncio.CancelledError): - await ingress_event await websocket.close( code=1008, reason="Shadow result contract is invalid", ) return client_event = asyncio.create_task(websocket.receive()) - if ingress_event not in completed: - ingress_event.cancel() - with suppress(asyncio.CancelledError): - await ingress_event - continue - event = await ingress_event - if event is None: - if ingress.snapshot()["closed"]: - break - continue - await websocket.send_bytes(event.wire_bytes()) + if ingress_event in completed: + event = ingress_event.result() + if event is None: + if ingress.snapshot()["closed"]: + break + else: + await websocket.send_bytes(event.wire_bytes()) + ingress_event = asyncio.create_task( + asyncio.to_thread( + ingress.take_next, + consumer_id, + timeout=0.5, + ) + ) except (WebSocketDisconnect, RuntimeError): return finally: client_event.cancel() with suppress(asyncio.CancelledError): await client_event + ingress_event.cancel() + with suppress(asyncio.CancelledError): + await ingress_event ingress.close_consumer(consumer_id) with suppress(RuntimeError): await websocket.close() diff --git a/tests/test_e15_shadow_inference.py b/tests/test_e15_shadow_inference.py index 4838f92..79590d0 100644 --- a/tests/test_e15_shadow_inference.py +++ b/tests/test_e15_shadow_inference.py @@ -1,6 +1,7 @@ from __future__ import annotations import importlib.util +import io import json import sys from argparse import Namespace @@ -92,6 +93,7 @@ def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( listen_host="127.0.0.1", listen_port=18020, command="serve", + max_duration_seconds=180.0, marker="preserved", ) @@ -101,6 +103,7 @@ def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( "request_id": "physical-k1-shadow-001", "output_name": "physical-k1-shadow-001", "token": "a" * 64, + "max_duration_seconds": 90, }, ) @@ -108,6 +111,7 @@ def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( assert run.output == tmp_path / "physical-k1-shadow-001" assert run.token == "a" * 64 assert run.token_stdin is False + assert run.max_duration_seconds == 90.0 assert run.marker == "preserved" assert not hasattr(run, "listen_host") assert not hasattr(run, "output_root") @@ -121,6 +125,16 @@ def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( "token": "short", }, ) + with pytest.raises(RuntimeError, match="request contract"): + module._persistent_run_arguments( + service, + { + "request_id": "physical-k1-shadow-004", + "output_name": "physical-k1-shadow-004", + "token": "b" * 64, + "max_duration_seconds": 181, + }, + ) with pytest.raises(RuntimeError, match="request contract"): module._persistent_run_arguments( service, @@ -130,3 +144,28 @@ def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( "token": "b" * 64, }, ) + + +def test_runtime_telemetry_captures_bounded_queue_snapshots() -> None: + module = _module() + stream = io.StringIO() + telemetry = module._RuntimeTelemetry( + stream, + interval_seconds=1.0, + snapshotters={ + "detector": lambda: { + "capacity": 2, + "maximum_depth": 1, + "final_depth": 0, + } + }, + ) + + telemetry._sample() + summary = telemetry.summary() + + assert summary["sample_count"] == 1 + assert summary["final_queues"]["detector"]["capacity"] == 2 + row = json.loads(stream.getvalue()) + assert row["schema_version"] == "missioncore.worker-runtime-telemetry/v1" + assert row["queues"]["detector"]["maximum_depth"] == 1 diff --git a/tests/test_e15_shadow_runtime.py b/tests/test_e15_shadow_runtime.py index f2e74e1..b029f29 100644 --- a/tests/test_e15_shadow_runtime.py +++ b/tests/test_e15_shadow_runtime.py @@ -46,13 +46,25 @@ def test_incremental_media_buffer_is_bounded_and_fail_closed() -> None: source.read(100) -def test_persistent_decoder_rejects_camera_fragment_gaps_before_decode() -> None: +def test_persistent_decoder_counts_latest_wins_camera_fragment_gaps() -> None: decoder = PersistentFmp4Decoder(on_frame=lambda _frame: None) decoder.start() decoder.feed_init(b"not-a-real-init") decoder.feed_segment(_metadata(1), b"first") - with pytest.raises(ShadowRuntimeError, match="sequence gap"): - decoder.feed_segment(_metadata(3), b"third") + decoder.feed_segment(_metadata(3), b"third") + assert decoder.snapshot()["source_sequence_gaps"] == 1 + decoder.finish_input() + with pytest.raises(ShadowRuntimeError, match="decoder failed"): + decoder.join() + + +def test_persistent_decoder_rejects_non_increasing_camera_sequence() -> None: + decoder = PersistentFmp4Decoder(on_frame=lambda _frame: None) + decoder.start() + decoder.feed_init(b"not-a-real-init") + decoder.feed_segment(_metadata(2), b"second") + with pytest.raises(ShadowRuntimeError, match="not increasing"): + decoder.feed_segment(_metadata(2), b"duplicate") decoder.finish_input() with pytest.raises(ShadowRuntimeError, match="decoder failed"): decoder.join() diff --git a/tests/test_e21_realtime_envelope.py b/tests/test_e21_realtime_envelope.py new file mode 100644 index 0000000..6cd88aa --- /dev/null +++ b/tests/test_e21_realtime_envelope.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + + +def _module(name: str, relative: str): + path = Path(__file__).resolve().parents[1] / relative + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_camera_replay_loads_segments_lazily_and_checks_each_digest(tmp_path: Path) -> None: + module = _module( + "e21_source_replay_test", + "experiments/perception/run_e12_shadow_transport_replay.py", + ) + epoch = tmp_path / "epoch-1" + segments = epoch / "segments" + segments.mkdir(parents=True) + (epoch / "init.mp4").write_bytes(b"init") + rows = [] + for sequence, payload in ((1, b"one"), (2, b"two")): + path = segments / f"{sequence}.m4s" + path.write_bytes(payload) + rows.append( + { + "host_epoch_ns": sequence * 100_000_000, + "host_monotonic_ns": sequence * 100_000_000, + "length": len(payload), + "path": f"segments/{sequence}.m4s", + "sequence": sequence, + "sha256": hashlib.sha256(payload).hexdigest(), + } + ) + (epoch / "index.jsonl").write_text( + "".join(json.dumps(row) + "\n" for row in rows), + encoding="utf-8", + ) + + events = module._camera_events(epoch, 1.0) + assert next(events).modality == "camera-init" + assert next(events).payload == b"one" + (segments / "2.m4s").write_bytes(b"changed-after-generator-start") + with pytest.raises(RuntimeError, match="length differs"): + next(events) + + +def test_e21_trend_reports_accumulating_age_and_stable_memory() -> None: + module = _module( + "e21_analyzer_test", + "experiments/perception/analyze_e21_realtime_envelope.py", + ) + growing = module._trend([(float(index), float(index) * 2) for index in range(20)]) + stable = module._trend([(float(index), 100.0) for index in range(20)]) + + assert growing["slope_per_second"] == pytest.approx(2.0) + assert growing["late_vs_early_p95"] > 0 + assert stable["slope_per_second"] == pytest.approx(0.0) + assert stable["late_vs_early_p95"] == pytest.approx(0.0) + + +def test_e21_profile_keeps_diagnostic_authority_and_long_1x_gate() -> None: + root = Path(__file__).resolve().parents[1] + profile = json.loads( + ( + root + / "experiments" + / "perception" + / "e21_realtime_envelope_profile.json" + ).read_text() + ) + + assert profile["source"]["speed"] == 1.0 + assert profile["source"]["minimum_span_seconds"] >= 55.0 + assert profile["worker"]["minimum_detector_fps"] >= 9.5 + assert profile["worker"]["maximum_detector_drop_fraction"] <= 0.02 + assert profile["worker"]["maximum_p95_world_state_age_ms"] == 200.0 + assert profile["authority"] == { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + } diff --git a/tests/test_live_perception_shadow.py b/tests/test_live_perception_shadow.py index 83deaa8..f6ed43e 100644 --- a/tests/test_live_perception_shadow.py +++ b/tests/test_live_perception_shadow.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import stat import threading +import time from pathlib import Path import numpy as np @@ -97,3 +98,80 @@ def test_shadow_router_accepts_only_validated_diagnostic_results_back() -> None: assert websocket.accepted is True assert published.is_set() assert received == [encoded] + + +def test_shadow_router_does_not_discard_ingress_while_receiving_results() -> None: + class SlowIngress(LivePerceptionIngress): + def take_next( + self, + consumer_id: str, + *, + timeout: float | None = None, + ): + time.sleep(0.01) + return super().take_next(consumer_id, timeout=timeout) + + ingress = SlowIngress() + received: list[bytes] = [] + router = build_live_perception_shadow_router( + ingress, + "test-plugin", + bearer_token="x" * 43, + result_receiver=lambda payload: not received.append(payload), + ) + ingress.begin_session("session-1") + for modality, sequence in ( + ("camera-init", 0), + ("camera-frame", 1), + ("lidar", 1), + ("pose", 1), + ): + assert ingress.publish( + modality=modality, + source_id="sensor.camera.right", + source_sequence=sequence, + captured_at_epoch_ns=sequence, + received_monotonic_ns=sequence, + payload=modality.encode(), + ) + ingress.end_session("session-1") + expected_events = sum( + int(queue["depth"]) for queue in ingress.snapshot()["queues"].values() + ) + + class DuplexFakeWebSocket: + def __init__(self) -> None: + self.headers = {"authorization": f"Bearer {'x' * 43}"} + self.results_remaining = 8 + self.sent: list[bytes] = [] + + async def accept(self) -> None: + return + + async def receive(self) -> dict[str, object]: + if self.results_remaining: + self.results_remaining -= 1 + await asyncio.sleep(0) + return {"type": "websocket.receive", "bytes": b"result"} + deadline = asyncio.get_running_loop().time() + 1 + while len(self.sent) < expected_events: + if asyncio.get_running_loop().time() >= deadline: + break + await asyncio.sleep(0.005) + return {"type": "websocket.disconnect"} + + async def send_bytes(self, payload: bytes) -> None: + self.sent.append(payload) + + async def close(self, **_: object) -> None: + return + + websocket = DuplexFakeWebSocket() + asyncio.run(router.routes[0].endpoint(websocket)) # type: ignore[attr-defined] + + assert len(received) == 8 + assert len(websocket.sent) == expected_events + snapshot = ingress.snapshot() + assert sum( + int(queue["consumed"]) for queue in snapshot["queues"].values() + ) == expected_events