From e21332b9e14708c0124cdf62f07420876ae3a7a3 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 20:36:21 +0300 Subject: [PATCH] feat(perception): measure M48S load envelope --- config/perception/m48s-load-envelope-v1.json | 89 +++++++++++++++++++ .../run_m48s_reference_graph_shadow_worker.py | 75 ++++++++++++++-- scripts/Invoke-M48SReferenceGraphShadow.ps1 | 18 ++++ .../m48s_reference_graph_runtime.py | 6 ++ src/k1link/perception/recorded_source.py | 35 +++++++- tests/test_perception_graph.py | 44 +++++++++ 6 files changed, 257 insertions(+), 10 deletions(-) create mode 100644 config/perception/m48s-load-envelope-v1.json diff --git a/config/perception/m48s-load-envelope-v1.json b/config/perception/m48s-load-envelope-v1.json new file mode 100644 index 0000000..77cd6ae --- /dev/null +++ b/config/perception/m48s-load-envelope-v1.json @@ -0,0 +1,89 @@ +{ + "schema_version": "missioncore.m48s-load-envelope-profile/v1", + "profile_id": "m48s-rf-detr-reference-graph-load-envelope/v1", + "decision_question": "Can the unchanged single-pass RF-DETR reference graph sustain the recorded production rate and 20 percent reserve on Worker 006, and where does its bounded latest-wins limit begin?", + "hypothesis": "The hardened single-pass graph sustains 10 FPS and 12 FPS without unbounded queues, failed frames, a second inference pass, or more than the predeclared delivery loss; 15 FPS is measured only to locate the limit.", + "source": { + "source_id": "RAVNOVES00", + "source_profile_id": "m4-ravnoves00-recorded-realtime/v1", + "frame_count": 4489, + "graph_id": "reference-perception-graph/v2", + "detector_provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", + "sensor_timestamps_preserved": true + }, + "controlled_change": { + "only_variable": "wall_clock_source_rate_hz", + "unchanged": [ + "source frames and source timestamps", + "RF-DETR TensorRT engine and threshold", + "geometry, temporal, motion, rolling-map and threat providers", + "latest-wins queue capacities", + "single inference pass per admitted detector frame", + "detector prewarm and cyclic-GC hot-loop policy" + ] + }, + "scenarios": [ + { + "id": "production-10fps", + "run_id": "m48s-load-envelope-v1-production-10fps-a1", + "load_purpose": "production-rate", + "source_rate_hz": 10.0, + "thresholds": { + "minimum_delivery_ratio": 0.999, + "minimum_effective_world_state_fps": 9.5, + "maximum_world_state_completion_p95_ms": 125.0 + }, + "required_for_reserve_decision": true + }, + { + "id": "reserve-12fps", + "run_id": "m48s-load-envelope-v1-reserve-12fps-a1", + "load_purpose": "reserve-gate", + "source_rate_hz": 12.0, + "thresholds": { + "minimum_delivery_ratio": 0.995, + "minimum_effective_world_state_fps": 11.4, + "maximum_world_state_completion_p95_ms": 150.0 + }, + "required_for_reserve_decision": true + }, + { + "id": "limit-15fps", + "run_id": "m48s-load-envelope-v1-limit-15fps-a1", + "load_purpose": "limit-discovery", + "source_rate_hz": 15.0, + "thresholds": { + "minimum_delivery_ratio": 0.95, + "minimum_effective_world_state_fps": 14.25, + "maximum_world_state_completion_p95_ms": 175.0 + }, + "required_for_reserve_decision": false + } + ], + "common_integrity_gates": [ + "closed terminal accounting", + "zero failed, stale, rejected or unavailable frames", + "all queue high-watermarks at or below capacity two", + "complete single-pass pipeline timing", + "detector prewarm before source admission", + "cyclic GC disabled only during the hot loop and restored after", + "all command, actuation, navigation and safety authority remains false" + ], + "visual_evidence": { + "binding": "reuse exact M4.8S full camera plus LiDAR plus 3D/PLAN timeline", + "reason": "Only wall-clock pacing changes; source frames, sensor timestamps and graph semantics remain immutable.", + "independent_ground_truth": false + }, + "exit_decision": { + "reserve_accepted_when": "Both production-10fps and reserve-12fps pass every integrity and operating target gate.", + "optimization_scope_when_rejected": "Optimize only the measured bottleneck stage; do not add another detector or inference pass.", + "limit_discovery_interpretation": "The 15 FPS result is valid evidence when integrity gates pass even if its operating target gate fails." + }, + "authority": { + "candidate_accepted": false, + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false, + "production_accepted": false + } +} diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py index 16e8c98..3275323 100644 --- a/experiments/perception/run_m48s_reference_graph_shadow_worker.py +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -52,7 +52,7 @@ from k1link.perception.reference_graph_runtime import ReferenceGraphRuntimePaths from k1link.perception.rolling_map import RollingLocalObstacleMapProvider from k1link.perception.temporal import BoundedSpatialTemporalProvider -SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v3" +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v4" FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v0" GC_POLICY_SCHEMA: Final = "missioncore.cyclic-gc-hot-loop-policy/v0" @@ -63,6 +63,7 @@ AUTHORITY: Final = { "actuation_allowed": False, "navigation_or_safety_accepted": False, } +LOAD_PURPOSES: Final = ("production-rate", "reserve-gate", "limit-discovery") class GpuTelemetry: @@ -359,6 +360,13 @@ def main() -> int: parser.add_argument("--triton-origin", default="http://127.0.0.1:8000") parser.add_argument("--loops", type=int, default=1) parser.add_argument("--maximum-frames", type=int) + parser.add_argument("--source-rate-hz", type=float) + parser.add_argument("--minimum-delivery-ratio", type=float, default=0.0) + parser.add_argument("--minimum-effective-world-state-fps", type=float, default=9.5) + parser.add_argument("--maximum-world-state-completion-p95-ms", type=float, default=175.0) + parser.add_argument("--load-purpose", choices=LOAD_PURPOSES, default="production-rate") + parser.add_argument("--runtime-artifact-sha256", required=True) + parser.add_argument("--runner-sha256", required=True) parser.add_argument("--telemetry-interval-seconds", type=float, default=1.0) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--progress", type=Path, required=True) @@ -370,6 +378,21 @@ def main() -> int: raise RuntimeError("maximum frame count must be positive") if arguments.telemetry_interval_seconds <= 0: raise RuntimeError("telemetry interval must be positive") + if arguments.source_rate_hz is not None and ( + not np.isfinite(arguments.source_rate_hz) or arguments.source_rate_hz <= 0 + ): + raise RuntimeError("source rate must be positive and finite") + if not 0.0 <= arguments.minimum_delivery_ratio <= 1.0: + raise RuntimeError("minimum delivery ratio must be between zero and one") + if arguments.minimum_effective_world_state_fps <= 0: + raise RuntimeError("minimum effective world-state FPS must be positive") + if arguments.maximum_world_state_completion_p95_ms <= 0: + raise RuntimeError("maximum completion p95 must be positive") + for digest in (arguments.runtime_artifact_sha256, arguments.runner_sha256): + if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest): + raise RuntimeError("runtime and runner SHA-256 values must be lowercase hex") + if _sha256(Path(__file__)) != arguments.runner_sha256: + raise RuntimeError("runner SHA-256 changed") output = arguments.output.absolute() progress = arguments.progress.absolute() frame_ledger = arguments.frame_ledger.absolute() @@ -426,6 +449,7 @@ def main() -> int: decode_timing_observer=timing_store.observe_decode, detector_timing_observer=timing_store.observe_detector, maximum_frames=arguments.maximum_frames, + source_rate_hz=arguments.source_rate_hz, ) as runtime: for stage_id, attribute in ( ("geometry", "geometry"), @@ -546,7 +570,9 @@ def main() -> int: identity = _identity_metrics(all_deliveries) semantic = _semantic_metrics(all_deliveries, advisories) world_state_fps = delivered / processing_wall_seconds - checks = { + delivery_ratio = delivered / admitted if admitted else 0.0 + completion_distribution = _distribution(completion_ages_ms) + integrity_checks = { "loop_count_completed": len(loop_documents) == arguments.loops, "graph_stopped_cleanly": all( loop["state"] == GraphState.STOPPED.value for loop in loop_documents @@ -561,10 +587,6 @@ def main() -> int: TerminalOutcomeType.UNAVAILABLE, ) ), - "minimum_world_state_fps": world_state_fps >= 9.5, - "maximum_world_state_completion_p95_ms": ( - _distribution(completion_ages_ms)["p95"] <= 175.0 - ), "bounded_latest_wins_queues": all(value <= 2 for value in queue_high_watermarks.values()), "temporal_identity_reuse_observed": cast( int, @@ -588,6 +610,19 @@ def main() -> int: ), "authority_remains_false": all(value is False for value in AUTHORITY.values()), } + operating_target_checks = { + "minimum_delivery_ratio": delivery_ratio >= arguments.minimum_delivery_ratio, + "minimum_world_state_fps": ( + world_state_fps >= arguments.minimum_effective_world_state_fps + ), + "maximum_world_state_completion_p95_ms": ( + completion_distribution["p95"] + <= arguments.maximum_world_state_completion_p95_ms + ), + } + checks = {**integrity_checks, **operating_target_checks} + evidence_integrity_gate_passed = all(integrity_checks.values()) + operating_target_gate_passed = all(operating_target_checks.values()) integrated_runtime_gate_passed = all(checks.values()) document = { "schema_version": SCHEMA_VERSION, @@ -595,20 +630,28 @@ def main() -> int: "source_id": "RAVNOVES00", "loops": arguments.loops, "maximum_frames_per_loop": arguments.maximum_frames, + "requested_rate_hz": arguments.source_rate_hz, + "pacing_contract": "wall-clock-scaled-source-timestamps-immutable/v1", }, "identity": { "worker_id": "worker-006", "graph_id": "reference-perception-graph/v2", "detector_provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", "inputs": _input_digests(paths, arguments.detector_profile), + "runtime_artifact_sha256": arguments.runtime_artifact_sha256, + "runner_sha256": arguments.runner_sha256, }, "execution": { "run_mode": GraphRunMode.SOURCE_PACED_LATEST_WINS.value, + "load_purpose": arguments.load_purpose, + "requested_source_rate_hz": arguments.source_rate_hz, + "source_timestamps_preserved": True, "wall_seconds": round(wall_seconds, 6), "source_processing_wall_seconds": round(processing_wall_seconds, 6), "admitted_frames": admitted, "delivered_world_states": delivered, "effective_world_state_fps": round(world_state_fps, 6), + "delivery_ratio": round(delivery_ratio, 9), "terminal_outcomes": dict(sorted(accounting.items())), "queue_high_watermarks": queue_high_watermarks, "loops": loop_documents, @@ -624,7 +667,7 @@ def main() -> int: }, }, "metrics": { - "world_state_completion_age_ms": _distribution(completion_ages_ms), + "world_state_completion_age_ms": completion_distribution, "local_obstacle_map_output_age_ms": _distribution(map_output_ages_ms), "identity_continuity": identity, "semantic_advisory": semantic, @@ -635,6 +678,19 @@ def main() -> int: "process_peak_rss_after_mib": round(rss_after_kib / 1024.0, 6), }, "checks": checks, + "integrity_checks": integrity_checks, + "operating_target_checks": operating_target_checks, + "predeclared_thresholds": { + "minimum_delivery_ratio": arguments.minimum_delivery_ratio, + "minimum_effective_world_state_fps": ( + arguments.minimum_effective_world_state_fps + ), + "maximum_world_state_completion_p95_ms": ( + arguments.maximum_world_state_completion_p95_ms + ), + }, + "evidence_integrity_gate_passed": evidence_integrity_gate_passed, + "operating_target_gate_passed": operating_target_gate_passed, "integrated_runtime_gate_passed": integrated_runtime_gate_passed, "independent_track_identity_quality_evaluated": False, "independent_risk_policy_quality_evaluated": False, @@ -647,7 +703,10 @@ def main() -> int: output.write_bytes(_canonical_json(document) + b"\n") print(output) print(json.dumps(checks, indent=2, sort_keys=True)) - return 0 if integrated_runtime_gate_passed else 2 + discovery_completed = ( + arguments.load_purpose == "limit-discovery" and evidence_integrity_gate_passed + ) + return 0 if integrated_runtime_gate_passed or discovery_completed else 2 def _record_completion_age( diff --git a/scripts/Invoke-M48SReferenceGraphShadow.ps1 b/scripts/Invoke-M48SReferenceGraphShadow.ps1 index 7e95501..8a53225 100644 --- a/scripts/Invoke-M48SReferenceGraphShadow.ps1 +++ b/scripts/Invoke-M48SReferenceGraphShadow.ps1 @@ -12,6 +12,16 @@ param( [int]$Loops = 1, [ValidateRange(1, 4489)] [int]$MaximumFrames = 120, + [ValidateRange(1.0, 120.0)] + [double]$SourceRateHz = 10.0, + [ValidateRange(0.0, 1.0)] + [double]$MinimumDeliveryRatio = 0.999, + [ValidateRange(0.1, 120.0)] + [double]$MinimumEffectiveWorldStateFps = 9.5, + [ValidateRange(1.0, 10000.0)] + [double]$MaximumWorldStateCompletionP95Ms = 125.0, + [ValidateSet("production-rate", "reserve-gate", "limit-discovery")] + [string]$LoadPurpose = "production-rate", [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m48s-reference-graph-shadow" ) @@ -103,6 +113,7 @@ $runner = Get-Item -LiteralPath ( if ($runner.PSIsContainer -or ($runner.Attributes -band [IO.FileAttributes]::ReparsePoint)) { throw "M48S graph runner must be a regular file" } +$runnerSha256 = Get-Sha256 $runner.FullName $experimentRoot = Resolve-DDirectory ( "D:\NDC_MISSIONCORE\runtime\experiments\m48t-fixed-detector-20260825T095425Z" @@ -253,6 +264,13 @@ try { "--valid-fov-mask", "/source/mask.png", "--triton-origin", "http://127.0.0.1:8000", "--loops", ([string]$Loops), + "--source-rate-hz", ([string]::Format([Globalization.CultureInfo]::InvariantCulture, "{0:R}", $SourceRateHz)), + "--minimum-delivery-ratio", ([string]::Format([Globalization.CultureInfo]::InvariantCulture, "{0:R}", $MinimumDeliveryRatio)), + "--minimum-effective-world-state-fps", ([string]::Format([Globalization.CultureInfo]::InvariantCulture, "{0:R}", $MinimumEffectiveWorldStateFps)), + "--maximum-world-state-completion-p95-ms", ([string]::Format([Globalization.CultureInfo]::InvariantCulture, "{0:R}", $MaximumWorldStateCompletionP95Ms)), + "--load-purpose", $LoadPurpose, + "--runtime-artifact-sha256", $ExpectedWheelSha256, + "--runner-sha256", $runnerSha256, "--output", "/output/result.json", "--progress", "/output/progress.jsonl", "--frame-ledger", "/output/frames.jsonl" diff --git a/src/k1link/perception/m48s_reference_graph_runtime.py b/src/k1link/perception/m48s_reference_graph_runtime.py index 877256e..3b551ae 100644 --- a/src/k1link/perception/m48s_reference_graph_runtime.py +++ b/src/k1link/perception/m48s_reference_graph_runtime.py @@ -92,6 +92,7 @@ def build_m48s_reference_graph_runtime( decode_timing_observer: DecodeTimingObserver | None = None, detector_timing_observer: DetectorTimingObserver | None = None, maximum_frames: int | None = None, + source_rate_hz: float | None = None, ) -> M48sReferenceGraphRuntime: """Instantiate the complete graph with only its detector pin replaced.""" @@ -125,6 +126,11 @@ def build_m48s_reference_graph_runtime( if run_mode is GraphRunMode.SOURCE_PACED_LATEST_WINS else ReplayPacing.UNCAPPED ), + target_rate_hz=( + source_rate_hz + if run_mode is GraphRunMode.SOURCE_PACED_LATEST_WINS + else None + ), ), decoder=PyAvRecordedImageDecoder(paths.video), timing_observer=decode_timing_observer, diff --git a/src/k1link/perception/recorded_source.py b/src/k1link/perception/recorded_source.py index 36703c9..aef37b9 100644 --- a/src/k1link/perception/recorded_source.py +++ b/src/k1link/perception/recorded_source.py @@ -99,7 +99,11 @@ DecodeTimingObserver = Callable[[DecodedFrameTiming], None] class RecordedRavnoves00Source: - """Emit the admitted synchronized source timeline at 1.0x or uncapped speed.""" + """Emit the admitted timeline at its recorded or an explicit replay rate. + + A target rate changes wall-clock pacing only. Immutable sensor timestamps stay + untouched so accelerated load measurements cannot silently change scene motion. + """ provider_id: str = RECORDED_SOURCE_PROVIDER_ID @@ -111,16 +115,24 @@ class RecordedRavnoves00Source: pacing: ReplayPacing = ReplayPacing.UNCAPPED, expected_frame_count: int = DEFAULT_FRAME_COUNT, expected_source_pack_sha256: str | None = RECORDED_SOURCE_PACK_SHA256, + target_rate_hz: float | None = None, clock_ns: Callable[[], int] = time.monotonic_ns, wait: WaitFunction | None = None, ) -> None: if expected_frame_count < 1: raise RecordedSourceError("expected frame count must be positive") + if target_rate_hz is not None and ( + not np.isfinite(target_rate_hz) or target_rate_hz <= 0 + ): + raise RecordedSourceError("target replay rate must be positive and finite") + if target_rate_hz is not None and pacing is not ReplayPacing.ONE_X: + raise RecordedSourceError("target replay rate requires paced replay") self.camera_index_path = camera_index_path.resolve() self.source_pack_path = source_pack_path.resolve() self.pacing = pacing self.expected_frame_count = expected_frame_count self.expected_source_pack_sha256 = expected_source_pack_sha256 + self.target_rate_hz = target_rate_hz self._clock_ns = clock_ns self._wait = wait or _event_wait @@ -130,6 +142,7 @@ class RecordedRavnoves00Source: repository_root: Path, *, pacing: ReplayPacing = ReplayPacing.UNCAPPED, + target_rate_hz: float | None = None, ) -> RecordedRavnoves00Source: root = repository_root.resolve() camera_index = ( @@ -148,6 +161,7 @@ class RecordedRavnoves00Source: camera_index_path=camera_index, source_pack_path=source_pack, pacing=pacing, + target_rate_hz=target_rate_hz, ) def packets(self, stop_event: Event) -> Iterator[SourcePacket]: @@ -163,6 +177,7 @@ class RecordedRavnoves00Source: started_ns = int(self._clock_ns()) source_origin_ns = _source_time_ns(timeline_rows[0]) + pacing_scale = _pacing_scale(timeline_rows, self.target_rate_hz) for frame_index, (camera, timeline) in enumerate( zip(camera_rows, timeline_rows, strict=True) ): @@ -170,7 +185,8 @@ class RecordedRavnoves00Source: return packet = _packet(frame_index, camera, timeline) if self.pacing is ReplayPacing.ONE_X: - target_ns = started_ns + packet.envelope.timestamps.source_ns - source_origin_ns + source_elapsed_ns = packet.envelope.timestamps.source_ns - source_origin_ns + target_ns = started_ns + round(source_elapsed_ns * pacing_scale) if not self._pace_until(stop_event, target_ns): return yield packet @@ -349,6 +365,21 @@ def _source_time_ns(document: _SourceTimelineRow) -> int: return round(value * 1_000_000_000) +def _pacing_scale( + timeline: tuple[_SourceTimelineRow, ...], + target_rate_hz: float | None, +) -> float: + if target_rate_hz is None: + return 1.0 + if len(timeline) < 2: + raise RecordedSourceError("target-rate replay requires at least two frames") + duration_seconds = timeline[-1].session_seconds - timeline[0].session_seconds + if not np.isfinite(duration_seconds) or duration_seconds <= 0: + raise RecordedSourceError("source timeline duration is invalid") + recorded_rate_hz = (len(timeline) - 1) / duration_seconds + return recorded_rate_hz / target_rate_hz + + def _integer(document: dict[str, object], key: str) -> int: value = document.get(key) if not isinstance(value, int) or isinstance(value, bool) or value < 0: diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py index cf84e27..f49d1ae 100644 --- a/tests/test_perception_graph.py +++ b/tests/test_perception_graph.py @@ -835,6 +835,50 @@ def test_recorded_source_reuses_one_timeline_for_1x_and_uncapped(tmp_path: Path) assert one_x[1].registered_point_increment_payload is None +def test_recorded_source_target_rate_changes_only_wall_clock_pacing(tmp_path: Path) -> None: + camera_path, timeline_path = _write_recorded_fixture(tmp_path) + now = [1_000_000_000] + waits: list[float] = [] + + def wait(stop_event: Event, seconds: float) -> bool: + waits.append(seconds) + now[0] += round(seconds * 1_000_000_000) + return stop_event.is_set() + + source = RecordedRavnoves00Source( + camera_index_path=camera_path, + source_pack_path=timeline_path, + pacing=ReplayPacing.ONE_X, + target_rate_hz=20.0, + expected_frame_count=2, + expected_source_pack_sha256=None, + clock_ns=lambda: now[0], + wait=wait, + ) + + packets = list(source.packets(Event())) + + assert waits == pytest.approx([0.05]) + assert ( + packets[1].envelope.timestamps.source_ns + - packets[0].envelope.timestamps.source_ns + == 100_000_000 + ) + + +def test_recorded_source_rejects_target_rate_for_uncapped_replay(tmp_path: Path) -> None: + camera_path, timeline_path = _write_recorded_fixture(tmp_path) + with pytest.raises(RecordedSourceError, match="requires paced replay"): + RecordedRavnoves00Source( + camera_index_path=camera_path, + source_pack_path=timeline_path, + pacing=ReplayPacing.UNCAPPED, + target_rate_hz=12.0, + expected_frame_count=2, + expected_source_pack_sha256=None, + ) + + def test_recorded_source_rejects_timeline_mismatch(tmp_path: Path) -> None: camera_path, timeline_path = _write_recorded_fixture(tmp_path, mismatched=True) source = RecordedRavnoves00Source(