From 05590867b724e81e5dfca88d56fa2c8370cf6ce9 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 16:44:26 +0300 Subject: [PATCH] feat(perception): seal M4.8S reference graph replay --- ...48s-rf-detr-reference-graph-shadow-v0.json | 95 +++ .../run_m48s_reference_graph_shadow_worker.py | 581 ++++++++++++++++++ .../seal_m48s_reference_graph_replay.py | 251 ++++++++ .../seal_m48s_reference_graph_shadow.py | 251 ++++++++ scripts/Invoke-M48SReferenceGraphShadow.ps1 | 287 +++++++++ src/k1link/perception/graph.py | 34 + src/k1link/perception/m48s_advisory.py | 198 ++++++ .../m48s_reference_graph_runtime.py | 252 ++++++++ src/k1link/perception/m48s_replay_timeline.py | 426 +++++++++++++ tests/test_m48s_reference_graph_shadow.py | 78 +++ ...test_m48s_reference_graph_shadow_result.py | 49 ++ tests/test_perception_graph.py | 77 ++- 12 files changed, 2578 insertions(+), 1 deletion(-) create mode 100644 config/perception/m48s-rf-detr-reference-graph-shadow-v0.json create mode 100644 experiments/perception/run_m48s_reference_graph_shadow_worker.py create mode 100644 experiments/perception/seal_m48s_reference_graph_replay.py create mode 100644 experiments/perception/seal_m48s_reference_graph_shadow.py create mode 100644 scripts/Invoke-M48SReferenceGraphShadow.ps1 create mode 100644 src/k1link/perception/m48s_advisory.py create mode 100644 src/k1link/perception/m48s_reference_graph_runtime.py create mode 100644 src/k1link/perception/m48s_replay_timeline.py create mode 100644 tests/test_m48s_reference_graph_shadow.py create mode 100644 tests/test_m48s_reference_graph_shadow_result.py diff --git a/config/perception/m48s-rf-detr-reference-graph-shadow-v0.json b/config/perception/m48s-rf-detr-reference-graph-shadow-v0.json new file mode 100644 index 0000000..eacf4c1 --- /dev/null +++ b/config/perception/m48s-rf-detr-reference-graph-shadow-v0.json @@ -0,0 +1,95 @@ +{ + "schema_version": "missioncore.reference-perception-graph-config/v2", + "graph_id": "reference-perception-graph/v2", + "source_profile_id": "m4-ravnoves00-recorded-realtime/v1", + "providers": [ + { + "role": "source", + "provider_id": "ravnoves00-recorded-source/v1", + "version": "1.0.0", + "revision": "m4-ravnoves00-recorded-realtime/v1", + "sha256": "ea10359339e6cce31b5780a2710299771cab7cc0c1c2a2b56a1621f786b31fa8" + }, + { + "role": "detector", + "provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", + "version": "0.1.0", + "revision": "rf-detr-large-coco-704-trt11-fp16-risk-shadow/v0", + "sha256": "0c307fd2d19cedd2c9267b6be2effdce82161a719a742f315fcd7a76f7061b08" + }, + { + "role": "geometry", + "provider_id": "ravnoves00-geometry-association/v1", + "version": "1.0.0", + "revision": "m4-ravnoves00-e29-e32-geometry/v1", + "sha256": "cc666c9389a5e221957faddec89584709b66918d14abaf646f1832e001421999" + }, + { + "role": "temporal", + "provider_id": "bounded-spatial-temporal-layer/v1", + "version": "1.0.0", + "revision": "m4-bounded-temporal-motion/v1", + "sha256": "7130eaee24a95c7d888bf7598010e03e129e1c3ac5b34bcd8401015ff4244b39" + }, + { + "role": "motion", + "provider_id": "class-independent-motion-estimator/v1", + "version": "1.0.0", + "revision": "m4-bounded-temporal-motion/v1", + "sha256": "7130eaee24a95c7d888bf7598010e03e129e1c3ac5b34bcd8401015ff4244b39" + }, + { + "role": "rolling", + "provider_id": "rolling-local-obstacle-map/v1", + "version": "1.0.0", + "revision": "ravnoves00-rolling-local-obstacle-map/v1", + "sha256": "f7e3315eaf6ffaf3aee1e04913933812092cf82bbcc9984c1a6fa2d9250e6784" + }, + { + "role": "threat", + "provider_id": "dual-evidence-replay-threat/v3", + "version": "3.0.0", + "revision": "m4-ravnoves00-virtual-corridor/v3", + "sha256": "8c3a5aa837da1f028f5998fb504a1381f9b2b68de6420a32160410b6dc0887c7" + } + ], + "queues": [ + { + "stage_id": "detector", + "capacity": 2, + "deadline_ns": 1000000000, + "terminal_timeout_ns": 90000000000 + }, + { + "stage_id": "geometry", + "capacity": 2, + "deadline_ns": 1500000000, + "terminal_timeout_ns": 90000000000 + }, + { + "stage_id": "temporal", + "capacity": 2, + "deadline_ns": 1750000000, + "terminal_timeout_ns": 90000000000 + }, + { + "stage_id": "rolling", + "capacity": 2, + "deadline_ns": 2000000000, + "terminal_timeout_ns": 90000000000 + }, + { + "stage_id": "threat", + "capacity": 2, + "deadline_ns": 2250000000, + "terminal_timeout_ns": 90000000000 + } + ], + "authority": { + "mode": "replay-simulated", + "physical_live": false, + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false + } +} diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py new file mode 100644 index 0000000..58875db --- /dev/null +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -0,0 +1,581 @@ +#!/usr/bin/env python3 +"""Run RF-DETR through the complete source-paced reference graph on Worker 006.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import resource +import subprocess +import threading +import time +from collections import Counter, defaultdict +from collections.abc import Mapping +from dataclasses import asdict, fields +from functools import partial +from pathlib import Path +from typing import Any, Final, TextIO, cast + +import numpy as np + +from k1link.perception.contracts import MotionState +from k1link.perception.contracts import ObjectProposal2D +from k1link.perception.detector import ( + DetectorProviderSnapshot, + RfDetrShadowDetectorProvider, +) +from k1link.perception.geometry import Ravnoves00GeometryAssociationProvider +from k1link.perception.graph_contracts import ( + DeliveredFrame, + GraphRunMode, + GraphRunResultV2, + GraphState, + TerminalOutcomeType, +) +from k1link.perception.m48s_advisory import ( + AdvisoryFamily, + M48sSemanticAdvisory, + advisory_policy_matrix, + project_m48s_advisories, +) +from k1link.perception.m48s_reference_graph_runtime import ( + build_m48s_reference_graph_runtime, +) +from k1link.perception.motion import ClassIndependentMotionEstimator +from k1link.perception.object_understanding import AdvisoryResponse +from k1link.perception.reference_graph_runtime import ReferenceGraphRuntimePaths +from k1link.perception.providers import SourcePacket +from k1link.perception.rolling_map import RollingLocalObstacleMapProvider +from k1link.perception.temporal import BoundedSpatialTemporalProvider + +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v0" +FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v0" +AUTHORITY: Final = { + "ground_truth": False, + "candidate_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + + +class GpuTelemetry: + def __init__(self, interval_seconds: float) -> None: + self.interval_seconds = interval_seconds + self.samples: list[dict[str, float]] = [] + self._stop = threading.Event() + self._thread = threading.Thread(target=self._run, daemon=True) + + def __enter__(self) -> GpuTelemetry: + self._thread.start() + return self + + def __exit__(self, *_args: object) -> None: + self._stop.set() + self._thread.join(timeout=10.0) + + def _run(self) -> None: + while not self._stop.is_set(): + try: + completed = subprocess.run( + [ + "nvidia-smi", + "--query-gpu=utilization.gpu,memory.used,power.draw,temperature.gpu", + "--format=csv,noheader,nounits", + ], + check=True, + capture_output=True, + text=True, + timeout=10.0, + ) + values = [float(value.strip()) for value in completed.stdout.split(",")] + if len(values) == 4: + self.samples.append( + { + "gpu_utilization_percent": values[0], + "gpu_memory_used_mib": values[1], + "gpu_power_w": values[2], + "gpu_temperature_c": values[3], + } + ) + except (OSError, ValueError, subprocess.SubprocessError): + pass + self._stop.wait(self.interval_seconds) + + +def main() -> int: + parser = argparse.ArgumentParser() + for name in ( + "graph-config", + "baseline-profile", + "detector-profile", + "geometry-profile", + "temporal-motion-profile", + "rolling-map-profile", + "threat-profile", + "camera-index", + "source-pack", + "local-surface", + "video", + "valid-fov-mask", + ): + parser.add_argument(f"--{name}", type=Path, required=True) + 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("--telemetry-interval-seconds", type=float, default=1.0) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--progress", type=Path, required=True) + parser.add_argument("--frame-ledger", type=Path, required=True) + arguments = parser.parse_args() + if arguments.loops < 1: + raise RuntimeError("loop count must be positive") + if arguments.maximum_frames is not None and arguments.maximum_frames < 1: + raise RuntimeError("maximum frame count must be positive") + if arguments.telemetry_interval_seconds <= 0: + raise RuntimeError("telemetry interval must be positive") + output = arguments.output.absolute() + progress = arguments.progress.absolute() + frame_ledger = arguments.frame_ledger.absolute() + if output.exists() or progress.exists() or frame_ledger.exists(): + raise RuntimeError("result, progress, or frame ledger artifact already exists") + output.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + paths = ReferenceGraphRuntimePaths( + graph_config=arguments.graph_config, + baseline_profile=arguments.baseline_profile, + geometry_profile=arguments.geometry_profile, + temporal_motion_profile=arguments.temporal_motion_profile, + rolling_map_profile=arguments.rolling_map_profile, + threat_profile=arguments.threat_profile, + camera_index=arguments.camera_index, + source_pack=arguments.source_pack, + local_surface=arguments.local_surface, + video=arguments.video, + valid_fov_mask=arguments.valid_fov_mask, + ) + started_ns = time.monotonic_ns() + started_utc_ns = time.time_ns() + rss_before_kib = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + loop_documents: list[dict[str, object]] = [] + completion_ages_ms: list[float] = [] + map_output_ages_ms: list[float] = [] + all_deliveries: list[DeliveredFrame] = [] + with ( + progress.open("x", encoding="utf-8") as progress_stream, + frame_ledger.open("x", encoding="utf-8") as frame_ledger_stream, + GpuTelemetry(arguments.telemetry_interval_seconds) as gpu, + ): + for loop_index in range(arguments.loops): + loop_completion_ages_ns: list[int] = [] + setup_started_ns = time.monotonic_ns() + + with build_m48s_reference_graph_runtime( + paths=paths, + detector_profile=arguments.detector_profile, + triton_origin=arguments.triton_origin, + run_mode=GraphRunMode.SOURCE_PACED_LATEST_WINS, + delivery_observer=partial( + _record_completion_age, + loop_completion_ages_ns, + ), + delivery_evidence_observer=partial( + _record_delivery_evidence, + frame_ledger_stream, + loop_index, + ), + maximum_frames=arguments.maximum_frames, + ) as runtime: + loop_started_ns = time.monotonic_ns() + result = runtime.graph.run() + loop_completed_ns = time.monotonic_ns() + detector_snapshot = cast( + RfDetrShadowDetectorProvider, + runtime.graph.detector, + ).snapshot() + provider_snapshots = { + "geometry": asdict( + cast( + Ravnoves00GeometryAssociationProvider, + runtime.graph.geometry, + ).snapshot() + ), + "temporal": asdict( + cast( + BoundedSpatialTemporalProvider, + runtime.graph.temporal, + ).snapshot() + ), + "motion": asdict( + cast( + ClassIndependentMotionEstimator, + runtime.graph.motion, + ).snapshot() + ), + "rolling": asdict( + cast( + RollingLocalObstacleMapProvider, + runtime.graph.rolling, + ).snapshot() + ), + } + if not isinstance(result, GraphRunResultV2): + raise RuntimeError("RF-DETR shadow did not return a V2 graph result") + loop_wall_seconds = (loop_completed_ns - loop_started_ns) / 1_000_000_000.0 + setup_seconds = (loop_started_ns - setup_started_ns) / 1_000_000_000.0 + if len(loop_completion_ages_ns) != len(result.deliveries): + raise RuntimeError("final delivery timing accounting did not close") + loop_document = _loop_document( + loop_index=loop_index, + result=result, + detector_snapshot=detector_snapshot, + provider_snapshots=provider_snapshots, + completion_ages_ns=loop_completion_ages_ns, + wall_seconds=loop_wall_seconds, + setup_seconds=setup_seconds, + ) + loop_documents.append(loop_document) + completion_ages_ms.extend(value / 1_000_000.0 for value in loop_completion_ages_ns) + map_output_ages_ms.extend( + delivery.obstacle_map.output_age_ns / 1_000_000.0 + for delivery in result.deliveries + ) + all_deliveries.extend(result.deliveries) + frame_ledger_stream.flush() + progress_row = { + "loop": loop_index + 1, + "loops": arguments.loops, + "elapsed_seconds": round( + (time.monotonic_ns() - started_ns) / 1_000_000_000.0, + 3, + ), + "admitted": result.admitted_count, + "delivered": len(result.deliveries), + "world_state_completion_p95_ms": _distribution( + [value / 1_000_000.0 for value in loop_completion_ages_ns] + )["p95"], + } + progress_stream.write(json.dumps(progress_row, separators=(",", ":")) + "\n") + progress_stream.flush() + print(json.dumps(progress_row, sort_keys=True), flush=True) + + completed_ns = time.monotonic_ns() + wall_seconds = (completed_ns - started_ns) / 1_000_000_000.0 + rss_after_kib = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + accounting: Counter[str] = Counter() + for loop in loop_documents: + accounting.update(cast(Mapping[str, int], loop["terminal_outcomes"])) + admitted = sum(cast(int, loop["admitted_count"]) for loop in loop_documents) + delivered = len(all_deliveries) + processing_wall_seconds = sum( + cast(float, loop["wall_seconds"]) for loop in loop_documents + ) + queue_high_watermarks = { + stage: max( + cast(dict[str, int], loop["queue_high_watermarks"])[stage] + for loop in loop_documents + ) + for stage in ("detector", "geometry", "temporal", "rolling", "threat") + } + advisories = tuple( + advisory + for delivery in all_deliveries + for advisory in project_m48s_advisories(delivery) + ) + identity = _identity_metrics(all_deliveries) + semantic = _semantic_metrics(all_deliveries, advisories) + world_state_fps = delivered / processing_wall_seconds + checks = { + "loop_count_completed": len(loop_documents) == arguments.loops, + "graph_stopped_cleanly": all( + loop["state"] == GraphState.STOPPED.value for loop in loop_documents + ), + "closed_terminal_accounting": admitted == sum(accounting.values()), + "zero_failed_stale_rejected_or_unavailable": all( + accounting[outcome.value] == 0 + for outcome in ( + TerminalOutcomeType.FAILED, + TerminalOutcomeType.STALE, + TerminalOutcomeType.REJECTED, + 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, + identity["multi_frame_component_count"], + ) + > 0, + "conservative_unknown_motion_advisory": _unknown_motion_is_conservative(advisories), + "distinct_class_family_policy": len(set(advisory_policy_matrix().values())) + == len(AdvisoryFamily), + "authority_remains_false": all(value is False for value in AUTHORITY.values()), + } + integrated_runtime_gate_passed = all(checks.values()) + document = { + "schema_version": SCHEMA_VERSION, + "source": { + "source_id": "RAVNOVES00", + "loops": arguments.loops, + "maximum_frames_per_loop": arguments.maximum_frames, + }, + "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), + }, + "execution": { + "run_mode": GraphRunMode.SOURCE_PACED_LATEST_WINS.value, + "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), + "terminal_outcomes": dict(sorted(accounting.items())), + "queue_high_watermarks": queue_high_watermarks, + "loops": loop_documents, + "frame_evidence": { + "schema_version": FRAME_EVIDENCE_SCHEMA, + "path": frame_ledger.name, + "row_count": delivered, + "sha256": _sha256(frame_ledger), + }, + }, + "metrics": { + "world_state_completion_age_ms": _distribution(completion_ages_ms), + "local_obstacle_map_output_age_ms": _distribution(map_output_ages_ms), + "identity_continuity": identity, + "semantic_advisory": semantic, + "gpu": _telemetry_summary(gpu.samples), + "process_peak_rss_before_mib": round(rss_before_kib / 1024.0, 6), + "process_peak_rss_after_mib": round(rss_after_kib / 1024.0, 6), + }, + "checks": checks, + "integrated_runtime_gate_passed": integrated_runtime_gate_passed, + "independent_track_identity_quality_evaluated": False, + "independent_risk_policy_quality_evaluated": False, + "production_accepted": False, + "started_utc_ns": started_utc_ns, + "completed_utc_ns": time.time_ns(), + "completed": True, + "authority": AUTHORITY, + } + 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 + + +def _record_completion_age( + samples: list[int], + _delivery: DeliveredFrame, + completion_age_ns: int, +) -> None: + samples.append(completion_age_ns) + + +def _record_delivery_evidence( + stream: TextIO, + loop_index: int, + delivery: DeliveredFrame, + packet: SourcePacket, + proposals: tuple[ObjectProposal2D, ...], + associated_proposal_ids: frozenset[str], + completion_age_ns: int, +) -> None: + if delivery.sequence != packet.envelope.sequence: + raise RuntimeError("delivery evidence sequence binding changed") + row = { + "schema_version": FRAME_EVIDENCE_SCHEMA, + "loop_index": loop_index, + "source_envelope": packet.envelope.to_dict(), + "completion_age_ns": completion_age_ns, + "local_obstacle_map_output_age_ns": delivery.obstacle_map.output_age_ns, + "delivery": delivery.canonical_dict(), + "detector_proposals": [proposal.to_dict() for proposal in proposals], + "associated_proposal_ids": sorted(associated_proposal_ids), + "semantic_advisories": [ + advisory.to_dict() for advisory in project_m48s_advisories(delivery) + ], + "authority": AUTHORITY, + } + stream.write(_canonical_json(row).decode("utf-8") + "\n") + + +def _loop_document( + *, + loop_index: int, + result: GraphRunResultV2, + detector_snapshot: DetectorProviderSnapshot, + provider_snapshots: dict[str, dict[str, object]], + completion_ages_ns: list[int], + wall_seconds: float, + setup_seconds: float, +) -> dict[str, object]: + outcomes = Counter(item.outcome.value for item in result.terminal_outcomes) + outcome_stages = Counter( + f"{item.outcome.value}:{item.stage_id}:{item.reason}" + for item in result.terminal_outcomes + ) + return { + "loop_index": loop_index, + "state": result.state.value, + "wall_seconds": round(wall_seconds, 6), + "setup_seconds": round(setup_seconds, 6), + "admitted_count": result.admitted_count, + "delivered_count": len(result.deliveries), + "effective_world_state_fps": round(len(result.deliveries) / wall_seconds, 6), + "terminal_outcomes": dict(sorted(outcomes.items())), + "terminal_outcome_stages": dict(sorted(outcome_stages.items())), + "terminal_outcome_details": [item.to_dict() for item in result.terminal_outcomes], + "queue_high_watermarks": dict(result.queue_high_watermarks), + "canonical_payload_sha256": result.canonical_payload_sha256, + "world_state_completion_age_ms": _distribution( + [value / 1_000_000.0 for value in completion_ages_ns] + ), + "detector": asdict(detector_snapshot), + "providers": provider_snapshots, + } + + +def _identity_metrics(deliveries: list[DeliveredFrame]) -> dict[str, object]: + appearances: dict[str, list[int]] = defaultdict(list) + duplicate_components = 0 + for delivery in deliveries: + components = ( + *delivery.obstacle_map.occupied, + *delivery.obstacle_map.unknown, + ) + ids = [item.component_id for item in components] + duplicate_components += len(ids) - len(set(ids)) + for component_id in ids: + appearances[component_id].append(delivery.sequence) + lengths = [len(values) for values in appearances.values()] + return { + "unique_component_count": len(appearances), + "multi_frame_component_count": sum(value > 1 for value in lengths), + "maximum_component_publications": max(lengths, default=0), + "mean_component_publications": round(float(np.mean(lengths)), 6) if lengths else 0.0, + "duplicate_component_ids_within_frame": duplicate_components, + "identity_scope": "ephemeral", + "independent_truth_available": False, + } + + +def _semantic_metrics( + deliveries: list[DeliveredFrame], + advisories: tuple[M48sSemanticAdvisory, ...], +) -> dict[str, object]: + hints: Counter[str] = Counter() + motions: Counter[str] = Counter() + for delivery in deliveries: + obstacles = ( + *delivery.obstacle_map.occupied, + *delivery.obstacle_map.unknown, + ) + for obstacle in obstacles: + hints[obstacle.semantic_hint or "geometry-only"] += 1 + motions[obstacle.motion.value] += 1 + for proposal in delivery.obstacle_map.camera_uncertainty: + hints[proposal.semantic_hint or "unclassified-camera"] += 1 + motions[MotionState.UNKNOWN.value] += 1 + families = Counter(item.family.value for item in advisories) + responses = Counter( + response.value + for item in advisories + for response in item.responses + ) + return { + "semantic_hint_counts": dict(sorted(hints.items())), + "motion_counts": dict(sorted(motions.items())), + "advisory_family_counts": dict(sorted(families.items())), + "advisory_response_counts": dict(sorted(responses.items())), + "policy_matrix": { + family.value: [response.value for response in policy] + for family, policy in advisory_policy_matrix().items() + }, + "additional_inference_passes": 0, + "authority": AUTHORITY, + } + + +def _unknown_motion_is_conservative( + advisories: tuple[M48sSemanticAdvisory, ...], +) -> bool: + for item in advisories: + if item.motion is not MotionState.UNKNOWN: + continue + responses = set(item.responses) + if not responses.intersection( + { + AdvisoryResponse.REDUCE_SPEED, + AdvisoryResponse.YIELD, + AdvisoryResponse.STOP, + AdvisoryResponse.ROUTE_AROUND, + } + ): + return False + return True + + +def _distribution(values: list[float]) -> dict[str, float]: + if not values: + return {"mean": 0.0, "p50": 0.0, "p95": 0.0, "p99": 0.0, "maximum": 0.0} + array = np.asarray(values, dtype=np.float64) + return { + "mean": round(float(array.mean()), 6), + "p50": round(float(np.percentile(array, 50)), 6), + "p95": round(float(np.percentile(array, 95)), 6), + "p99": round(float(np.percentile(array, 99)), 6), + "maximum": round(float(array.max()), 6), + } + + +def _telemetry_summary(samples: list[dict[str, float]]) -> dict[str, Any]: + result: dict[str, Any] = {"sample_count": len(samples)} + for key in ( + "gpu_utilization_percent", + "gpu_memory_used_mib", + "gpu_power_w", + "gpu_temperature_c", + ): + result[key] = _distribution([sample[key] for sample in samples]) + return result + + +def _input_digests( + paths: ReferenceGraphRuntimePaths, + detector_profile: Path, +) -> dict[str, str]: + values = { + field.name: _sha256(getattr(paths, field.name)) + for field in fields(ReferenceGraphRuntimePaths) + } + values["detector_profile"] = _sha256(detector_profile) + return dict(sorted(values.items())) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.resolve(strict=True).open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/seal_m48s_reference_graph_replay.py b/experiments/perception/seal_m48s_reference_graph_replay.py new file mode 100644 index 0000000..9e659b4 --- /dev/null +++ b/experiments/perception/seal_m48s_reference_graph_replay.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Seal exact per-frame evidence from the full M4.8S reference-graph replay.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import shutil +import tempfile +from pathlib import Path +from typing import Any, Final + +from k1link.perception.fixed_class_detector_tournament import canonical_json, sha256_path + +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-replay/v0" +RAW_SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v0" +FRAME_SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-frame-evidence/v0" +RESULT_PREFIX: Final = "m48s-reference-graph-replay-" +AUTHORITY: Final = { + "ground_truth": False, + "candidate_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + + +def main() -> int: + repository = Path(__file__).resolve().parents[2] + parser = argparse.ArgumentParser() + parser.add_argument("--raw-result", type=Path, required=True) + parser.add_argument("--progress", type=Path, required=True) + parser.add_argument("--frames", type=Path, required=True) + parser.add_argument("--wheel-sha256", required=True) + parser.add_argument("--worker-release-id", required=True) + parser.add_argument( + "--output-root", + type=Path, + default=( + repository + / ".runtime/compute-experiments/m48s-semantic-shadow/" + "reference-graph-replay-results" + ), + ) + arguments = parser.parse_args() + paths = { + "raw_result": arguments.raw_result.resolve(strict=True), + "progress": arguments.progress.resolve(strict=True), + "frames": arguments.frames.resolve(strict=True), + } + raw = _load_object(paths["raw_result"]) + frame_summary = _validate(raw, paths["frames"]) + if ( + len(arguments.wheel_sha256) != 64 + or any(value not in "0123456789abcdef" for value in arguments.wheel_sha256) + ): + raise RuntimeError("worker wheel digest is invalid") + if not arguments.worker_release_id.startswith("mission-core-m48s-reference-graph-replay-"): + raise RuntimeError("worker replay release identity is invalid") + + evidence = { + "files": { + name: {"sha256": sha256_path(path), "size_bytes": path.stat().st_size} + for name, path in sorted(paths.items()) + }, + "worker": { + "worker_id": "worker-006", + "release_id": arguments.worker_release_id, + "wheel_sha256": arguments.wheel_sha256, + "historical_triton_action": "none", + "durable_worker_action": "none", + }, + "execution": raw["execution"], + "metrics": raw["metrics"], + "frame_summary": frame_summary, + "checks": raw["checks"], + } + identity = { + "schema_version": SCHEMA_VERSION, + "source_session_id": "20260720T065719Z_viewer_live", + "source_id": "RAVNOVES00", + "graph_id": "reference-perception-graph/v2", + "detector_provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", + "evidence": evidence, + "completed": True, + "accepted": True, + "production_accepted": False, + "authority": AUTHORITY, + } + result_id = RESULT_PREFIX + hashlib.sha256(canonical_json(identity)).hexdigest() + output_root = arguments.output_root.expanduser().absolute() + output_root.mkdir(mode=0o700, parents=True, exist_ok=True) + destination = output_root / result_id + if destination.exists(): + existing = _load_object(destination / "manifest.json") + if existing.get("result_id") != result_id or existing.get("identity") != identity: + raise RuntimeError("immutable replay result exists with different identity") + print(result_id) + return 0 + + temporary = Path(tempfile.mkdtemp(prefix=".m48s-reference-replay-", dir=output_root)) + try: + shutil.copyfile(paths["raw_result"], temporary / "worker-result.json") + shutil.copyfile(paths["progress"], temporary / "worker-progress.jsonl") + shutil.copyfile(paths["frames"], temporary / "frames.jsonl") + report = { + "schema_version": SCHEMA_VERSION, + "result_id": result_id, + "completed": True, + "accepted": True, + "production_accepted": False, + "frame_summary": frame_summary, + "limitations": [ + "This is source-paced replay evidence, not physical-live authority.", + "Latest-wins superseded source frames retain exact camera/LiDAR source evidence but have no invented world state.", + "LiDAR camera points are generated later from the pinned factory KB4 calibration and are not semantic labels.", + ], + "authority": AUTHORITY, + } + (temporary / "report.json").write_bytes(canonical_json(report) + b"\n") + artifacts = { + path.name: {"sha256": sha256_path(path), "size_bytes": path.stat().st_size} + for path in sorted(temporary.iterdir()) + if path.is_file() + } + manifest = { + "schema_version": SCHEMA_VERSION, + "result_id": result_id, + "identity": identity, + "artifacts": artifacts, + } + (temporary / "manifest.json").write_bytes(canonical_json(manifest) + b"\n") + temporary.rename(destination) + except Exception: + shutil.rmtree(temporary, ignore_errors=True) + raise + print(result_id) + print(json.dumps(frame_summary, indent=2, sort_keys=True)) + return 0 + + +def _validate(raw: dict[str, Any], frames_path: Path) -> dict[str, object]: + if ( + raw.get("schema_version") != RAW_SCHEMA_VERSION + or raw.get("completed") is not True + or raw.get("integrated_runtime_gate_passed") is not True + or raw.get("production_accepted") is not False + or raw.get("authority") != AUTHORITY + ): + raise RuntimeError("raw graph replay is not accepted shadow evidence") + checks = raw.get("checks") + execution = raw.get("execution") + if ( + not isinstance(checks, dict) + or not checks + or not all(value is True for value in checks.values()) + or not isinstance(execution, dict) + or execution.get("admitted_frames") != 4489 + or execution.get("run_mode") != "source-paced-latest-wins" + or not isinstance(execution.get("loops"), list) + or len(execution["loops"]) != 1 + ): + raise RuntimeError("raw graph replay execution contract failed") + ledger = execution.get("frame_evidence") + if ( + not isinstance(ledger, dict) + or ledger.get("schema_version") != FRAME_SCHEMA_VERSION + or ledger.get("sha256") != sha256_path(frames_path) + or ledger.get("path") != frames_path.name + ): + raise RuntimeError("frame ledger binding failed") + loop = execution["loops"][0] + if not isinstance(loop, dict) or not isinstance(loop.get("terminal_outcome_details"), list): + raise RuntimeError("terminal outcome details are unavailable") + outcomes = loop["terminal_outcome_details"] + if len(outcomes) != 4489: + raise RuntimeError("terminal outcome details do not close source accounting") + outcome_by_sequence: dict[int, str] = {} + superseded_sequences: list[int] = [] + delivered_sequences: list[int] = [] + for item in outcomes: + if not isinstance(item, dict) or item.get("schema_version") != "missioncore.perception-terminal-outcome/v1": + raise RuntimeError("terminal outcome row is invalid") + sequence = item.get("sequence") + outcome = item.get("outcome") + if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence in outcome_by_sequence: + raise RuntimeError("terminal sequence identity is invalid") + if outcome not in {"delivered", "superseded"}: + raise RuntimeError("unexpected terminal outcome in accepted replay") + outcome_by_sequence[sequence] = outcome + (delivered_sequences if outcome == "delivered" else superseded_sequences).append(sequence) + if sorted(outcome_by_sequence) != list(range(4489)): + raise RuntimeError("terminal sequence set is incomplete") + + ledger_sequences: list[int] = [] + source_times_ns: list[int] = [] + with frames_path.open("r", encoding="utf-8") as stream: + for line_number, line in enumerate(stream, 1): + try: + row = json.loads(line) + except json.JSONDecodeError as exc: + raise RuntimeError(f"invalid frame ledger JSON at line {line_number}") from exc + if ( + not isinstance(row, dict) + or row.get("schema_version") != FRAME_SCHEMA_VERSION + or row.get("loop_index") != 0 + or row.get("authority") != AUTHORITY + ): + raise RuntimeError(f"invalid frame ledger contract at line {line_number}") + envelope = row.get("source_envelope") + delivery = row.get("delivery") + if not isinstance(envelope, dict) or not isinstance(delivery, dict): + raise RuntimeError("frame ledger source or delivery is missing") + sequence = envelope.get("sequence") + timestamps = envelope.get("timestamps") + if ( + not isinstance(sequence, int) + or isinstance(sequence, bool) + or delivery.get("sequence") != sequence + or outcome_by_sequence.get(sequence) != "delivered" + or not isinstance(timestamps, dict) + or not isinstance(timestamps.get("source_ns"), int) + ): + raise RuntimeError(f"frame ledger binding failed at line {line_number}") + ledger_sequences.append(sequence) + source_times_ns.append(timestamps["source_ns"]) + if ledger_sequences != sorted(delivered_sequences) or len(ledger_sequences) != ledger.get("row_count"): + raise RuntimeError("frame ledger delivered sequence set does not close") + if any(right <= left for left, right in zip(source_times_ns, source_times_ns[1:])): + raise RuntimeError("frame ledger source clock is not monotonic") + return { + "source_frame_count": 4489, + "world_state_frame_count": len(ledger_sequences), + "superseded_frame_count": len(superseded_sequences), + "superseded_sequences": sorted(superseded_sequences), + "first_source_time_ns": source_times_ns[0], + "last_source_time_ns": source_times_ns[-1], + "frame_schema_version": FRAME_SCHEMA_VERSION, + } + + +def _load_object(path: Path) -> dict[str, Any]: + document = json.loads(path.read_text("utf-8")) + if not isinstance(document, dict): + raise RuntimeError(f"JSON document must be an object: {path}") + return document + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/seal_m48s_reference_graph_shadow.py b/experiments/perception/seal_m48s_reference_graph_shadow.py new file mode 100644 index 0000000..d692d38 --- /dev/null +++ b/experiments/perception/seal_m48s_reference_graph_shadow.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Seal the complete RF-DETR reference-graph runtime gate.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import shutil +from pathlib import Path +from typing import Any, Final + +from k1link.perception.fixed_class_detector_tournament import ( + canonical_json, + false_authority, + sha256_path, +) + +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-gate/v0" +RAW_SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v0" +RESULT_PREFIX: Final = "m48s-reference-graph-shadow-gate-" +EXPECTED_GRAPH_CONFIG_SHA256: Final = ( + "e607916c0d2db5a1078bc194fa0e5e1bac1ca336de8daad29359ed0d2791b6cd" +) +EXPECTED_DETECTOR_PROFILE_SHA256: Final = ( + "0c307fd2d19cedd2c9267b6be2effdce82161a719a742f315fcd7a76f7061b08" +) + + +def main() -> int: + repository = Path(__file__).resolve().parents[2] + parser = argparse.ArgumentParser() + parser.add_argument( + "--raw-result", + type=Path, + default=( + repository + / ".runtime/m48s-reference-graph-shadow/full-1x-a6ee52c9/result.json" + ), + ) + parser.add_argument( + "--progress", + type=Path, + default=( + repository + / ".runtime/m48s-reference-graph-shadow/full-1x-a6ee52c9/progress.jsonl" + ), + ) + parser.add_argument( + "--graph-config", + type=Path, + default=( + repository + / "config/perception/m48s-rf-detr-reference-graph-shadow-v0.json" + ), + ) + parser.add_argument( + "--detector-profile", + type=Path, + default=repository / "config/perception/rf-detr-large-risk-shadow-v0.json", + ) + parser.add_argument( + "--wheel-sha256", + default="a6ee52c9113bd298f899352b323f9e2b8175d15ca33f342cb8a9b896c41d64ca", + ) + parser.add_argument( + "--worker-release-id", + default="mission-core-m48s-reference-graph-shadow-a6ee52c9", + ) + parser.add_argument( + "--output-root", + type=Path, + default=( + repository + / ".runtime/compute-experiments/m48s-semantic-shadow/" + "reference-graph-shadow-results" + ), + ) + arguments = parser.parse_args() + paths = { + "raw_result": arguments.raw_result.resolve(strict=True), + "progress": arguments.progress.resolve(strict=True), + "graph_config": arguments.graph_config.resolve(strict=True), + "detector_profile": arguments.detector_profile.resolve(strict=True), + } + raw = _load_object(paths["raw_result"]) + _validate( + raw, + graph_config_sha256=sha256_path(paths["graph_config"]), + detector_profile_sha256=sha256_path(paths["detector_profile"]), + wheel_sha256=arguments.wheel_sha256, + worker_release_id=arguments.worker_release_id, + ) + execution = raw["execution"] + metrics = raw["metrics"] + evidence = { + "files": { + name: {"sha256": sha256_path(path), "size_bytes": path.stat().st_size} + for name, path in sorted(paths.items()) + }, + "worker": { + "worker_id": "worker-006", + "release_id": arguments.worker_release_id, + "wheel_sha256": arguments.wheel_sha256, + "historical_triton_action": "none", + "durable_worker_action": "none", + }, + "execution": execution, + "world_state_completion_age_ms": metrics["world_state_completion_age_ms"], + "local_obstacle_map_output_age_ms": metrics[ + "local_obstacle_map_output_age_ms" + ], + "identity_continuity": metrics["identity_continuity"], + "semantic_advisory": metrics["semantic_advisory"], + "gpu": metrics["gpu"], + "checks": raw["checks"], + } + decision = { + "complete_reference_graph_shadow_passed": True, + "source_paced_runtime_gate_accepted": True, + "ready_for_visual_lab": True, + "independent_track_identity_quality_evaluated": False, + "independent_risk_policy_quality_evaluated": False, + "detector_replacement_authorized": False, + "production_accepted": False, + "next_gate": ( + "publish the complete world-state replay in Mission Core LAB and retain " + "independent object-centric review as a separate acceptance" + ), + } + identity = { + "schema_version": SCHEMA_VERSION, + "graph_id": "reference-perception-graph/v2", + "detector_provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", + "evidence": evidence, + "decision": decision, + "completed": True, + "accepted": True, + "production_accepted": False, + "authority": false_authority(), + } + result_id = RESULT_PREFIX + hashlib.sha256(canonical_json(identity)).hexdigest() + destination = arguments.output_root.absolute() / result_id + if destination.exists(): + existing = _load_object(destination / "manifest.json") + if existing.get("result_id") != result_id or existing.get("identity") != identity: + raise RuntimeError("immutable M48S graph result already exists with different bytes") + print(result_id) + return 0 + destination.mkdir(mode=0o700, parents=True) + shutil.copyfile(paths["raw_result"], destination / "worker-result.json") + shutil.copyfile(paths["progress"], destination / "worker-progress.jsonl") + report = { + "schema_version": SCHEMA_VERSION, + "result_id": result_id, + "completed": True, + "accepted": True, + "production_accepted": False, + "evidence": evidence, + "decision": decision, + "authority": false_authority(), + } + (destination / "report.json").write_bytes(canonical_json(report) + b"\n") + manifest = { + "schema_version": SCHEMA_VERSION, + "result_id": result_id, + "identity": identity, + "artifacts": { + "worker-result.json": sha256_path(destination / "worker-result.json"), + "worker-progress.jsonl": sha256_path(destination / "worker-progress.jsonl"), + "report.json": sha256_path(destination / "report.json"), + }, + } + (destination / "manifest.json").write_bytes(canonical_json(manifest) + b"\n") + print(result_id) + print(json.dumps(decision, indent=2, sort_keys=True)) + return 0 + + +def _validate( + raw: dict[str, Any], + *, + graph_config_sha256: str, + detector_profile_sha256: str, + wheel_sha256: str, + worker_release_id: str, +) -> None: + if ( + raw.get("schema_version") != RAW_SCHEMA_VERSION + or raw.get("completed") is not True + or raw.get("integrated_runtime_gate_passed") is not True + or raw.get("production_accepted") is not False + or raw.get("authority") != false_authority() + ): + raise RuntimeError("M48S complete graph result is not accepted shadow evidence") + if graph_config_sha256 != EXPECTED_GRAPH_CONFIG_SHA256: + raise RuntimeError("M48S complete graph config identity changed") + if detector_profile_sha256 != EXPECTED_DETECTOR_PROFILE_SHA256: + raise RuntimeError("M48S detector profile identity changed") + if len(wheel_sha256) != 64 or any(value not in "0123456789abcdef" for value in wheel_sha256): + raise RuntimeError("M48S worker wheel digest is invalid") + if worker_release_id != "mission-core-m48s-reference-graph-shadow-a6ee52c9": + raise RuntimeError("M48S worker release identity changed") + checks = raw.get("checks") + execution = raw.get("execution") + metrics = raw.get("metrics") + if ( + not isinstance(checks, dict) + or not checks + or not all(value is True for value in checks.values()) + or not isinstance(execution, dict) + or not isinstance(metrics, dict) + ): + raise RuntimeError("M48S complete graph checks are incomplete") + outcomes = execution.get("terminal_outcomes") + queues = execution.get("queue_high_watermarks") + completion = metrics.get("world_state_completion_age_ms") + identity = metrics.get("identity_continuity") + semantic = metrics.get("semantic_advisory") + if ( + execution.get("admitted_frames") != 4489 + or not isinstance(outcomes, dict) + or sum(outcomes.values()) != 4489 + or outcomes.get("delivered") != execution.get("delivered_world_states") + or any(outcomes.get(key, 0) != 0 for key in ("failed", "stale", "rejected", "unavailable")) + or float(execution.get("effective_world_state_fps", 0.0)) < 9.5 + or not isinstance(queues, dict) + or set(queues) != {"detector", "geometry", "temporal", "rolling", "threat"} + or any(not isinstance(value, int) or value > 2 for value in queues.values()) + or not isinstance(completion, dict) + or float(completion.get("p95", 1_000.0)) > 175.0 + or not isinstance(identity, dict) + or identity.get("duplicate_component_ids_within_frame") != 0 + or int(identity.get("multi_frame_component_count", 0)) < 1 + or identity.get("independent_truth_available") is not False + or not isinstance(semantic, dict) + or semantic.get("additional_inference_passes") != 0 + or semantic.get("authority") != false_authority() + ): + raise RuntimeError("M48S complete graph runtime contract failed") + + +def _load_object(path: Path) -> dict[str, Any]: + document = json.loads(path.read_text("utf-8")) + if not isinstance(document, dict): + raise RuntimeError(f"JSON document must be an object: {path}") + return document + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/Invoke-M48SReferenceGraphShadow.ps1 b/scripts/Invoke-M48SReferenceGraphShadow.ps1 new file mode 100644 index 0000000..7e95501 --- /dev/null +++ b/scripts/Invoke-M48SReferenceGraphShadow.ps1 @@ -0,0 +1,287 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$ReleaseRoot, + [Parameter(Mandatory = $true)] + [ValidatePattern("^[a-f0-9]{64}$")] + [string]$ExpectedWheelSha256, + [Parameter(Mandatory = $true)] + [ValidatePattern("^[A-Za-z0-9._-]{1,96}$")] + [string]$RunId, + [ValidateRange(1, 10)] + [int]$Loops = 1, + [ValidateRange(1, 4489)] + [int]$MaximumFrames = 120, + [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m48s-reference-graph-shadow" +) + +$ErrorActionPreference = "Stop" +$ProgressPreference = "SilentlyContinue" + +function Assert-LastExitCode([string]$Operation) { + if ($LASTEXITCODE -ne 0) { + throw "$Operation failed with exit code $LASTEXITCODE" + } +} + +function Get-Sha256([string]$Path) { + return (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant() +} + +function Assert-File([string]$Path, [string]$ExpectedSha256, [string]$Label) { + $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force + if ($item.PSIsContainer -or ($item.Attributes -band [IO.FileAttributes]::ReparsePoint)) { + throw "$Label must be a regular file" + } + if ((Get-Sha256 $item.FullName) -cne $ExpectedSha256) { + throw "$Label SHA-256 changed" + } + return $item.FullName +} + +function Resolve-DDirectory([string]$Path, [string]$Label, [bool]$Create) { + if ($Create -and -not (Test-Path -LiteralPath $Path)) { + $null = New-Item -ItemType Directory -Path $Path + } + $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force + if ( + -not $item.PSIsContainer -or + ($item.Attributes -band [IO.FileAttributes]::ReparsePoint) -or + [IO.Path]::GetPathRoot($item.FullName).TrimEnd("\") -ine "D:" + ) { + throw "$Label must be a real D: directory" + } + return $item.FullName +} + +function Convert-ToDockerPath([string]$Path) { + return $Path.Replace("\", "/") +} + +function Get-Container([string]$Name) { + $rows = @((& docker inspect $Name) | ConvertFrom-Json) + Assert-LastExitCode "Docker inspection for $Name" + if ($rows.Count -ne 1) { + throw "Container identity for $Name is not unique" + } + return $rows[0] +} + +if ($env:COMPUTERNAME -cne "DESKTOP-OPJ8J04") { + throw "M48S graph shadow is pinned to DESKTOP-OPJ8J04" +} + +$release = Resolve-DDirectory $ReleaseRoot "M48S release root" $false +$output = Resolve-DDirectory $OutputRoot "M48S output root" $true +$runOutput = Join-Path $output $RunId +if (Test-Path -LiteralPath $runOutput) { + throw "M48S run output already exists" +} +$null = New-Item -ItemType Directory -Path $runOutput +$runOutput = Resolve-DDirectory $runOutput "M48S run output" $false +$wheel = Assert-File ( + Join-Path $release "nodedc_mission_core-0.1.0-py3-none-any.whl" +) $ExpectedWheelSha256 "M48S wheel" + +$expectedConfigs = [ordered]@{ + "m48s-rf-detr-reference-graph-shadow-v0.json" = "e607916c0d2db5a1078bc194fa0e5e1bac1ca336de8daad29359ed0d2791b6cd" + "m4-recorded-realtime-baseline-v1.json" = "ea10359339e6cce31b5780a2710299771cab7cc0c1c2a2b56a1621f786b31fa8" + "rf-detr-large-risk-shadow-v0.json" = "0c307fd2d19cedd2c9267b6be2effdce82161a719a742f315fcd7a76f7061b08" + "m4-geometry-association-v1.json" = "cc666c9389a5e221957faddec89584709b66918d14abaf646f1832e001421999" + "m4-temporal-motion-v1.json" = "7130eaee24a95c7d888bf7598010e03e129e1c3ac5b34bcd8401015ff4244b39" + "m4-rolling-local-map-v1.json" = "f7e3315eaf6ffaf3aee1e04913933812092cf82bbcc9984c1a6fa2d9250e6784" + "m4-replay-threat-v3.json" = "8c3a5aa837da1f028f5998fb504a1381f9b2b68de6420a32160410b6dc0887c7" +} +foreach ($entry in $expectedConfigs.GetEnumerator()) { + $null = Assert-File (Join-Path $release $entry.Key) $entry.Value ( + "M48S config {0}" -f $entry.Key + ) +} +$runner = Get-Item -LiteralPath ( + Join-Path $release "run_m48s_reference_graph_shadow_worker.py" +) +if ($runner.PSIsContainer -or ($runner.Attributes -band [IO.FileAttributes]::ReparsePoint)) { + throw "M48S graph runner must be a regular file" +} + +$experimentRoot = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\experiments\m48t-fixed-detector-20260825T095425Z" +) "M48S RF-DETR experiment root" $false +$modelRoot = Resolve-DDirectory ( + (Join-Path $experimentRoot "triton-models") +) "M48S RF-DETR Triton model root" $false +$null = Assert-File ( + (Join-Path $modelRoot "rf_detr_large\1\model.plan") +) "986399ce706b7380472cf5e473232249fed6e628971d8007f6609e83128d46b8" ( + "RF-DETR TensorRT engine" +) +$null = Assert-File ( + (Join-Path $modelRoot "rf_detr_large\config.pbtxt") +) "80947cad235e5b000f11aa869a33af0e8c727f07e04046691468df1e171479b6" ( + "RF-DETR Triton config" +) + +$image = "nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" +& docker image inspect $image *> $null +Assert-LastExitCode "Pinned M48S image inspection" +$historicalTriton = Get-Container "ndc-mission-core-triton" +if (-not $historicalTriton.State.Running -or $historicalTriton.State.Health.Status -cne "healthy") { + throw "Historical Triton must remain healthy during M48S shadow" +} +$historicalTritonId = [string]$historicalTriton.Id +$tritonName = "ndc-mission-core-m48s-rf-detr-triton-shadow" +$graphName = "ndc-mission-core-m48s-reference-graph-shadow" +foreach ($name in @($tritonName, $graphName)) { + if (& docker ps -a --format "{{.Names}}" --filter "name=^/$name$") { + throw "M48S candidate container $name already exists" + } +} + +$source = [ordered]@{ + CameraIndex = "D:\NDC_MISSIONCORE\runtime\jobs\recorded-camera-602ac89026ed12978619801d\input\camera\sensor.camera.right\epoch-1\index.jsonl" + SourcePack = "D:\NDC_MISSIONCORE\runtime\derived\e10-lidar-pack-576c994a6c814e2592dd6240ace3902a5db94843312c759a73ba0c9166157d2b\lidar-pack.npz" + LocalSurface = "D:\NDC_MISSIONCORE\runtime\derived\k1-local-surface-23762244c8bdb97de26fb721ac957d7a00bc9a63571ac4cfa4be19c4effc7d55\local-surface.npz" + Video = "D:\NDC_MISSIONCORE\runtime\experiments\e46e\inputs\right-cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8.mp4" + Mask = "D:\NDC_MISSIONCORE\runtime\inputs\e2\valid-fov-mask-b4dd8ddf2b87c1d520ee8a0868c4fea062d7c14d1bae73ccabd3abe1f3acbac2\mask.png" +} +foreach ($entry in $source.GetEnumerator()) { + if (-not (Test-Path -LiteralPath $entry.Value -PathType Leaf)) { + throw "M48S source $($entry.Key) is missing" + } +} + +$media = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-e15-media-pyav180-lz445-v1" +) "PyAV dependency" $false +$opencv = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-e3-opencv413092-v1\packages" +) "OpenCV dependency" $false +$pillow = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-p0-env-v1" +) "Pillow dependency" $false + +try { + & docker create ` + --name $tritonName ` + --read-only ` + --security-opt "no-new-privileges:true" ` + --cap-drop ALL ` + --pids-limit 512 ` + --shm-size 1g ` + --gpus all ` + --tmpfs "/tmp:rw,noexec,nosuid,size=2g" ` + --health-cmd "curl --fail --silent http://127.0.0.1:8000/v2/health/ready" ` + --health-interval 5s ` + --health-timeout 3s ` + --health-start-period 20s ` + --health-retries 24 ` + -v ("{0}:/models:ro" -f (Convert-ToDockerPath $modelRoot)) ` + $image ` + tritonserver ` + --model-repository=/models ` + --model-control-mode=explicit ` + --load-model=rf_detr_large ` + --disable-auto-complete-config ` + --strict-readiness=true ` + --exit-on-error=true ` + --allow-http=true ` + --allow-grpc=false ` + --allow-metrics=false *> $null + Assert-LastExitCode "M48S Triton creation" + & docker start $tritonName *> $null + Assert-LastExitCode "M48S Triton start" + $ready = $false + foreach ($attempt in 1..60) { + Start-Sleep -Seconds 2 + $candidate = Get-Container $tritonName + if (-not $candidate.State.Running) { + throw "M48S Triton stopped during startup" + } + if ($candidate.State.Health.Status -ceq "healthy") { + $ready = $true + break + } + } + if (-not $ready) { + throw "M48S Triton did not become healthy" + } + if (@((Get-Container $tritonName).HostConfig.PortBindings.PSObject.Properties).Count -ne 0) { + throw "M48S Triton published a host port" + } + + $dockerRelease = Convert-ToDockerPath $release + $dockerOutput = Convert-ToDockerPath $runOutput + $maximumArguments = @() + if ($MaximumFrames -gt 0) { + $maximumArguments = @("--maximum-frames", ([string]$MaximumFrames)) + } + $arguments = @( + "run", "--name", $graphName, + "--network", ("container:{0}" -f $tritonName), + "--read-only", + "--security-opt", "no-new-privileges:true", + "--cap-drop", "ALL", + "--pids-limit", "256", + "--gpus", "all", + "--tmpfs", "/tmp:rw,noexec,nosuid,size=2g", + "-e", "PYTHONDONTWRITEBYTECODE=1", + "-e", "PYTHONPATH=/release/nodedc_mission_core-0.1.0-py3-none-any.whl:/opt/media:/opt/opencv:/opt/pillow", + "-v", ("{0}:/release:ro" -f $dockerRelease), + "-v", ("{0}:/output:rw" -f $dockerOutput), + "-v", ("{0}:/opt/media:ro" -f (Convert-ToDockerPath $media)), + "-v", ("{0}:/opt/opencv:ro" -f (Convert-ToDockerPath $opencv)), + "-v", ("{0}:/opt/pillow:ro" -f (Convert-ToDockerPath $pillow)), + "-v", ("{0}:/source/camera-index.jsonl:ro" -f (Convert-ToDockerPath $source.CameraIndex)), + "-v", ("{0}:/source/source-pack.npz:ro" -f (Convert-ToDockerPath $source.SourcePack)), + "-v", ("{0}:/source/local-surface.npz:ro" -f (Convert-ToDockerPath $source.LocalSurface)), + "-v", ("{0}:/source/right.mp4:ro" -f (Convert-ToDockerPath $source.Video)), + "-v", ("{0}:/source/mask.png:ro" -f (Convert-ToDockerPath $source.Mask)), + "--entrypoint", "python3", + $image, + "/release/run_m48s_reference_graph_shadow_worker.py", + "--graph-config", "/release/m48s-rf-detr-reference-graph-shadow-v0.json", + "--baseline-profile", "/release/m4-recorded-realtime-baseline-v1.json", + "--detector-profile", "/release/rf-detr-large-risk-shadow-v0.json", + "--geometry-profile", "/release/m4-geometry-association-v1.json", + "--temporal-motion-profile", "/release/m4-temporal-motion-v1.json", + "--rolling-map-profile", "/release/m4-rolling-local-map-v1.json", + "--threat-profile", "/release/m4-replay-threat-v3.json", + "--camera-index", "/source/camera-index.jsonl", + "--source-pack", "/source/source-pack.npz", + "--local-surface", "/source/local-surface.npz", + "--video", "/source/right.mp4", + "--valid-fov-mask", "/source/mask.png", + "--triton-origin", "http://127.0.0.1:8000", + "--loops", ([string]$Loops), + "--output", "/output/result.json", + "--progress", "/output/progress.jsonl", + "--frame-ledger", "/output/frames.jsonl" + ) + $maximumArguments + & docker @arguments + Assert-LastExitCode "M48S complete reference graph shadow" + if (-not (Test-Path -LiteralPath (Join-Path $runOutput "result.json") -PathType Leaf)) { + throw "M48S graph result was not written" + } + if (-not (Test-Path -LiteralPath (Join-Path $runOutput "frames.jsonl") -PathType Leaf)) { + throw "M48S frame evidence ledger was not written" + } +} finally { + foreach ($name in @($graphName, $tritonName)) { + if (& docker ps -a --format "{{.Names}}" --filter "name=^/$name$") { + & docker rm -f $name *> $null + } + } + $historicalAfter = Get-Container "ndc-mission-core-triton" + if ( + $historicalAfter.Id -cne $historicalTritonId -or + -not $historicalAfter.State.Running -or + $historicalAfter.State.Health.Status -cne "healthy" + ) { + throw "Historical Triton changed during M48S shadow" + } +} + +Write-Output ("M48S_RESULT={0}" -f (Join-Path $runOutput "result.json")) +Write-Output "HISTORICAL_TRITON_ACTION=none" +Write-Output "DURABLE_WORKER_ACTION=none" +Write-Output "PRODUCTION_ACCEPTED=false" diff --git a/src/k1link/perception/graph.py b/src/k1link/perception/graph.py index c2401e3..e172c4c 100644 --- a/src/k1link/perception/graph.py +++ b/src/k1link/perception/graph.py @@ -110,6 +110,10 @@ _RollingItem = _Temporal | _StopSignal _ThreatItem = _Temporal | _Rolled | _StopSignal _QueueItem = SourcePacket | _Detected | _Associated | _Temporal | _Rolled | _StopSignal _QueueItemT = TypeVar("_QueueItemT", bound=_QueueItem) +DeliveryEvidenceObserver = Callable[ + [DeliveredFrame, SourcePacket, tuple[ObjectProposal2D, ...], frozenset[str], int], + None, +] class ReferencePerceptionGraphV1: @@ -127,6 +131,8 @@ class ReferencePerceptionGraphV1: threat: ThreatProvider, telemetry_identity: PipelineTelemetryIdentity | None = None, telemetry_sink: PipelineTelemetrySink | None = None, + delivery_observer: Callable[[DeliveredFrame, int], None] | None = None, + delivery_evidence_observer: DeliveryEvidenceObserver | None = None, clock_ns: Callable[[], int] = time.monotonic_ns, ) -> None: if config.graph_id != REFERENCE_GRAPH_ID: @@ -145,6 +151,8 @@ class ReferencePerceptionGraphV1: run_mode=GraphRunMode.SOURCE_PACED_LATEST_WINS, telemetry_identity=telemetry_identity, telemetry_sink=telemetry_sink, + delivery_observer=delivery_observer, + delivery_evidence_observer=delivery_evidence_observer, clock_ns=clock_ns, ) @@ -162,6 +170,8 @@ class ReferencePerceptionGraphV1: run_mode: GraphRunMode, telemetry_identity: PipelineTelemetryIdentity | None, telemetry_sink: PipelineTelemetrySink | None, + delivery_observer: Callable[[DeliveredFrame, int], None] | None, + delivery_evidence_observer: DeliveryEvidenceObserver | None, clock_ns: Callable[[], int], ) -> None: if (telemetry_identity is None) is not (telemetry_sink is None): @@ -177,6 +187,8 @@ class ReferencePerceptionGraphV1: self.run_mode = run_mode self.telemetry_identity = telemetry_identity self.telemetry_sink = telemetry_sink + self.delivery_observer = delivery_observer + self.delivery_evidence_observer = delivery_evidence_observer self._clock_ns = clock_ns self._state = GraphState.CREATED self._state_lock = Lock() @@ -510,6 +522,23 @@ class ReferencePerceptionGraphV1: obstacle_map=obstacle_map, threats=threats, ) + completed_ns = self._now() + with self._result_lock: + admitted_at_ns = self._admitted_at_ns[item.packet.envelope.sequence] + completion_age_ns = item.packet.envelope.source_age_ns + max( + 0, + completed_ns - admitted_at_ns, + ) + if self.delivery_observer is not None: + self.delivery_observer(delivery, completion_age_ns) + if self.delivery_evidence_observer is not None: + self.delivery_evidence_observer( + delivery, + item.packet, + item.proposals, + item.associated_proposal_ids, + completion_age_ns, + ) with self._result_lock: self._deliveries.append(delivery) self._terminal( @@ -845,6 +874,8 @@ class ReferencePerceptionGraphV2(ReferencePerceptionGraphV1): run_mode: GraphRunMode, telemetry_identity: PipelineTelemetryIdentity | None = None, telemetry_sink: PipelineTelemetrySink | None = None, + delivery_observer: Callable[[DeliveredFrame, int], None] | None = None, + delivery_evidence_observer: DeliveryEvidenceObserver | None = None, clock_ns: Callable[[], int] = time.monotonic_ns, ) -> None: if config.graph_id != REFERENCE_GRAPH_ID_V2: @@ -865,6 +896,8 @@ class ReferencePerceptionGraphV2(ReferencePerceptionGraphV1): run_mode=run_mode, telemetry_identity=telemetry_identity, telemetry_sink=telemetry_sink, + delivery_observer=delivery_observer, + delivery_evidence_observer=delivery_evidence_observer, clock_ns=clock_ns, ) @@ -903,6 +936,7 @@ __all__ = [ "REFERENCE_GRAPH_ID_V2", "TERMINAL_OUTCOME_SCHEMA", "DeliveredFrame", + "DeliveryEvidenceObserver", "GraphExecutionError", "GraphRunResult", "GraphRunResultV2", diff --git a/src/k1link/perception/m48s_advisory.py b/src/k1link/perception/m48s_advisory.py new file mode 100644 index 0000000..27d35e4 --- /dev/null +++ b/src/k1link/perception/m48s_advisory.py @@ -0,0 +1,198 @@ +"""Class-aware advisory projection over one delivered reference-graph frame.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import StrEnum +from typing import Final + +from .contracts import FalseAuthority, MotionState, ThreatDecision +from .graph_contracts import DeliveredFrame +from .object_understanding import AdvisoryResponse + +M48S_ADVISORY_SCHEMA: Final = "missioncore.m48s-semantic-advisory/v0" +M48S_ADVISORY_POLICY_ID: Final = "m48s-behavior-relevant-object-advisory/v0" +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") + + +class AdvisoryFamily(StrEnum): + GENERIC_OBSTACLE = "generic-obstacle" + PERSON = "person" + ANIMAL = "animal" + LIGHT_ROAD_USER = "light-road-user" + VEHICLE = "vehicle" + + +class M48sAdvisoryError(ValueError): + """A semantic advisory escaped its bounded shadow-only policy.""" + + +@dataclass(frozen=True, slots=True) +class M48sSemanticAdvisory: + component_id: str + semantic_hint: str | None + family: AdvisoryFamily + motion: MotionState + threat_decision: ThreatDecision + responses: tuple[AdvisoryResponse, ...] + reason_codes: tuple[str, ...] + policy_id: str = M48S_ADVISORY_POLICY_ID + authority: FalseAuthority = FalseAuthority() + + def __post_init__(self) -> None: + if _IDENTIFIER.fullmatch(self.component_id) is None: + raise M48sAdvisoryError("advisory component id is invalid") + if self.semantic_hint is not None and _IDENTIFIER.fullmatch(self.semantic_hint) is None: + raise M48sAdvisoryError("advisory semantic hint is invalid") + if self.policy_id != M48S_ADVISORY_POLICY_ID: + raise M48sAdvisoryError("advisory policy identity changed") + if not self.responses or len(set(self.responses)) != len(self.responses): + raise M48sAdvisoryError("advisory responses must be unique and nonempty") + if not self.reason_codes or len(set(self.reason_codes)) != len(self.reason_codes): + raise M48sAdvisoryError("advisory reasons must be unique and nonempty") + if self.threat_decision is ThreatDecision.THREAT and ( + AdvisoryResponse.STOP not in self.responses + ): + raise M48sAdvisoryError("replay threat must retain a stop advisory") + if self.family is AdvisoryFamily.GENERIC_OBSTACLE and ( + AdvisoryResponse.ROUTE_AROUND not in self.responses + and AdvisoryResponse.STOP not in self.responses + ): + raise M48sAdvisoryError("generic obstacles must remain route-around advisories") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": M48S_ADVISORY_SCHEMA, + "policy_id": self.policy_id, + "component_id": self.component_id, + "semantic_hint": self.semantic_hint, + "family": self.family.value, + "motion": self.motion.value, + "threat_decision": self.threat_decision.value, + "responses": [item.value for item in self.responses], + "reason_codes": list(self.reason_codes), + "authority": self.authority.to_dict(), + } + + +def project_m48s_advisories( + delivery: DeliveredFrame, +) -> tuple[M48sSemanticAdvisory, ...]: + """Project class-specific caution without changing occupancy or graph threats.""" + + threat_by_component = {item.component_id: item.decision for item in delivery.threats} + advisories = [ + _project( + component_id=obstacle.component_id, + semantic_hint=obstacle.semantic_hint, + motion=obstacle.motion, + threat_decision=threat_by_component[obstacle.component_id], + ) + for obstacle in (*delivery.obstacle_map.occupied, *delivery.obstacle_map.unknown) + ] + advisories.extend( + _project( + component_id=proposal.proposal_id, + semantic_hint=proposal.semantic_hint, + motion=MotionState.UNKNOWN, + threat_decision=threat_by_component[proposal.proposal_id], + ) + for proposal in delivery.obstacle_map.camera_uncertainty + ) + return tuple(advisories) + + +def advisory_policy_matrix() -> dict[AdvisoryFamily, tuple[AdvisoryResponse, ...]]: + """Expose the fixed class-family policy for executable contract checks.""" + + return { + AdvisoryFamily.GENERIC_OBSTACLE: (AdvisoryResponse.ROUTE_AROUND,), + AdvisoryFamily.PERSON: ( + AdvisoryResponse.YIELD, + AdvisoryResponse.REDUCE_SPEED, + ), + AdvisoryFamily.ANIMAL: ( + AdvisoryResponse.REDUCE_SPEED, + AdvisoryResponse.STOP, + ), + AdvisoryFamily.LIGHT_ROAD_USER: ( + AdvisoryResponse.YIELD, + AdvisoryResponse.REDUCE_SPEED, + AdvisoryResponse.MONITOR, + ), + AdvisoryFamily.VEHICLE: ( + AdvisoryResponse.MONITOR, + AdvisoryResponse.YIELD, + ), + } + + +def _project( + *, + component_id: str, + semantic_hint: str | None, + motion: MotionState, + threat_decision: ThreatDecision, +) -> M48sSemanticAdvisory: + family = _family(semantic_hint) + responses = list(advisory_policy_matrix()[family]) + reasons = [f"family-{family.value}"] + if motion is MotionState.UNKNOWN: + reasons.append("unknown-motion-conservative") + elif motion is MotionState.MOVING: + reasons.append("observed-moving") + if AdvisoryResponse.REDUCE_SPEED not in responses: + responses.append(AdvisoryResponse.REDUCE_SPEED) + else: + reasons.append("observed-stationary-not-permanent") + if threat_decision is ThreatDecision.THREAT: + reasons.append("reference-graph-replay-threat") + responses.insert(0, AdvisoryResponse.STOP) + elif threat_decision is ThreatDecision.UNKNOWN: + reasons.append("reference-graph-threat-unknown") + else: + reasons.append("reference-graph-threat-clear-at-observation") + return M48sSemanticAdvisory( + component_id=component_id, + semantic_hint=semantic_hint, + family=family, + motion=motion, + threat_decision=threat_decision, + responses=tuple(dict.fromkeys(responses)), + reason_codes=tuple(reasons), + ) + + +def _family(semantic_hint: str | None) -> AdvisoryFamily: + if semantic_hint == "person": + return AdvisoryFamily.PERSON + if semantic_hint in { + "bird", + "cat", + "dog", + "horse", + "sheep", + "cow", + "elephant", + "bear", + "zebra", + "giraffe", + }: + return AdvisoryFamily.ANIMAL + if semantic_hint in {"bicycle", "motorcycle", "skateboard"}: + return AdvisoryFamily.LIGHT_ROAD_USER + if semantic_hint in {"car", "bus", "truck"}: + return AdvisoryFamily.VEHICLE + return AdvisoryFamily.GENERIC_OBSTACLE + + +__all__ = [ + "M48S_ADVISORY_POLICY_ID", + "M48S_ADVISORY_SCHEMA", + "AdvisoryFamily", + "M48sAdvisoryError", + "M48sSemanticAdvisory", + "advisory_policy_matrix", + "project_m48s_advisories", +] diff --git a/src/k1link/perception/m48s_reference_graph_runtime.py b/src/k1link/perception/m48s_reference_graph_runtime.py new file mode 100644 index 0000000..adbcff9 --- /dev/null +++ b/src/k1link/perception/m48s_reference_graph_runtime.py @@ -0,0 +1,252 @@ +"""Shadow assembly for RF-DETR inside the source-neutral reference graph.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Callable, Iterator +from dataclasses import dataclass +from pathlib import Path +from threading import Event + +from .baseline import load_m4_baseline +from .detector import RF_DETR_SHADOW_PROVIDER_ID, RfDetrShadowDetectorProvider +from .geometry import ( + Ravnoves00GeometryAssociationProvider, + RecordedGeometryStore, + load_geometry_profile, +) +from .graph import DeliveryEvidenceObserver, ReferencePerceptionGraphV2 +from .graph_contracts import DeliveredFrame, GraphRunMode +from .motion import ClassIndependentMotionEstimator +from .providers import ( + ProviderRole, + ReferencePerceptionGraphConfigV2, + SourcePacket, + SourceProvider, +) +from .recorded_source import ( + DecodedRecordedSource, + PyAvRecordedImageDecoder, + RecordedRavnoves00Source, + ReplayPacing, +) +from .reference_graph_runtime import ReferenceGraphRuntimePaths +from .rf_detr_object_detector import ( + RF_DETR_ENGINE_SHA256, + RF_DETR_MODEL_ID, + RF_DETR_MODEL_VERSION, + TritonRfDetrHttpInferenceBackend, +) +from .rolling_map import RollingLocalObstacleMapProvider, load_rolling_map_profile +from .temporal import BoundedSpatialTemporalProvider, load_temporal_motion_profile +from .threat import ( + DualEvidenceReplayThreatProvider, + RecordedReplayBodyFrameResolver, + load_replay_threat_profile, +) +from .yolox_object_detector import load_valid_fov_mask + + +class M48sReferenceGraphRuntimeError(RuntimeError): + """The RF-DETR graph shadow cannot be assembled from its pinned inputs.""" + + +@dataclass(slots=True) +class M48sReferenceGraphRuntime: + """Own one RF-DETR shadow graph and its persistent inference transport.""" + + graph: ReferencePerceptionGraphV2 + inference_backend: TritonRfDetrHttpInferenceBackend + + def close(self) -> None: + self.inference_backend.close() + + def __enter__(self) -> M48sReferenceGraphRuntime: + return self + + def __exit__(self, *args: object) -> None: + self.close() + + +def build_m48s_reference_graph_runtime( + *, + paths: ReferenceGraphRuntimePaths, + detector_profile: Path, + triton_origin: str, + run_mode: GraphRunMode, + delivery_observer: Callable[[DeliveredFrame, int], None] | None = None, + delivery_evidence_observer: DeliveryEvidenceObserver | None = None, + maximum_frames: int | None = None, +) -> M48sReferenceGraphRuntime: + """Instantiate the complete graph with only its detector pin replaced.""" + + config = _load_graph_config(paths.graph_config) + pinned_files = { + ProviderRole.SOURCE: paths.baseline_profile, + ProviderRole.DETECTOR: detector_profile, + ProviderRole.GEOMETRY: paths.geometry_profile, + ProviderRole.TEMPORAL: paths.temporal_motion_profile, + ProviderRole.MOTION: paths.temporal_motion_profile, + ProviderRole.ROLLING: paths.rolling_map_profile, + ProviderRole.THREAT: paths.threat_profile, + } + _validate_provider_digests(config, pinned_files) + _validate_detector_profile(detector_profile) + + load_m4_baseline(paths.baseline_profile) + geometry_profile = load_geometry_profile(paths.geometry_profile) + temporal_motion_profile = load_temporal_motion_profile(paths.temporal_motion_profile) + rolling_map_profile = load_rolling_map_profile(paths.rolling_map_profile) + threat_profile = load_replay_threat_profile(paths.threat_profile) + + if maximum_frames is not None and maximum_frames < 1: + raise M48sReferenceGraphRuntimeError("maximum frame count must be positive") + source: SourceProvider = DecodedRecordedSource( + source=RecordedRavnoves00Source( + camera_index_path=paths.camera_index, + source_pack_path=paths.source_pack, + pacing=( + ReplayPacing.ONE_X + if run_mode is GraphRunMode.SOURCE_PACED_LATEST_WINS + else ReplayPacing.UNCAPPED + ), + ), + decoder=PyAvRecordedImageDecoder(paths.video), + ) + if maximum_frames is not None: + source = _LimitedSource(source, maximum_frames) + backend = TritonRfDetrHttpInferenceBackend(triton_origin) + try: + store = RecordedGeometryStore( + source_pack_path=paths.source_pack, + local_surface_path=paths.local_surface, + profile=geometry_profile, + ) + body_frame_resolver = RecordedReplayBodyFrameResolver( + store, + profile=threat_profile.body_frame, + ) + graph = ReferencePerceptionGraphV2( + config=config, + source=source, + detector=RfDetrShadowDetectorProvider( + mask=load_valid_fov_mask(paths.valid_fov_mask), + backend=backend, + ), + geometry=Ravnoves00GeometryAssociationProvider(store=store), + temporal=BoundedSpatialTemporalProvider( + point_resolver=store, + profile=temporal_motion_profile, + ), + motion=ClassIndependentMotionEstimator(profile=temporal_motion_profile), + rolling=RollingLocalObstacleMapProvider( + pose_resolver=store, + profile=rolling_map_profile, + ), + threat=DualEvidenceReplayThreatProvider( + body_frame_resolver=body_frame_resolver, + profile=threat_profile, + ), + run_mode=run_mode, + delivery_observer=delivery_observer, + delivery_evidence_observer=delivery_evidence_observer, + ) + except Exception: + backend.close() + raise + return M48sReferenceGraphRuntime(graph=graph, inference_backend=backend) + + +class _LimitedSource: + """Bound a pilot without changing source or graph provider identity.""" + + def __init__(self, source: SourceProvider, maximum_frames: int) -> None: + self.source = source + self.maximum_frames = maximum_frames + self.provider_id = source.provider_id + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + packets = self.source.packets(stop_event) + for _index in range(self.maximum_frames): + try: + packet = next(packets) + except StopIteration: + return + yield packet + + +def _load_graph_config(path: Path) -> ReferencePerceptionGraphConfigV2: + try: + document = json.loads(path.resolve(strict=True).read_text("utf-8")) + return ReferencePerceptionGraphConfigV2.from_dict(document) + except (OSError, json.JSONDecodeError, ValueError) as exc: + raise M48sReferenceGraphRuntimeError("RF-DETR graph config is invalid") from exc + + +def _validate_provider_digests( + config: ReferencePerceptionGraphConfigV2, + pinned_files: dict[ProviderRole, Path], +) -> None: + pins = {pin.role: pin for pin in config.providers} + if set(pins) != set(pinned_files): + raise M48sReferenceGraphRuntimeError("RF-DETR graph provider pins are incomplete") + for role, path in pinned_files.items(): + if _sha256_file(path) != pins[role].sha256: + raise M48sReferenceGraphRuntimeError( + f"{role.value} provider profile digest changed" + ) + + +def _validate_detector_profile(path: Path) -> None: + try: + document = json.loads(path.resolve(strict=True).read_text("utf-8")) + model = document["model"] + status = document["status"] + authority = document["authority"] + except (OSError, KeyError, TypeError, json.JSONDecodeError) as exc: + raise M48sReferenceGraphRuntimeError("RF-DETR profile is incomplete") from exc + if ( + document.get("schema_version") + != "missioncore.rf-detr-risk-shadow-profile/v0" + or document.get("provider_id") != RF_DETR_SHADOW_PROVIDER_ID + or model.get("model_id") != RF_DETR_MODEL_ID + or model.get("model_version") != RF_DETR_MODEL_VERSION + or model.get("worker_006_rtx4090_tensorrt_11_engine_sha256") + != RF_DETR_ENGINE_SHA256 + or status.get("detector_load_gate_passed") is not True + or status.get("production_accepted") is not False + or any( + authority.get(key) is not False + for key in ( + "candidate_accepted", + "commands_enabled", + "actuation_allowed", + "navigation_or_safety_accepted", + ) + ) + ): + raise M48sReferenceGraphRuntimeError("RF-DETR shadow profile identity changed") + + +def _sha256_file(path: Path) -> str: + try: + resolved = path.resolve(strict=True) + except OSError as exc: + raise M48sReferenceGraphRuntimeError("pinned RF-DETR graph input is missing") from exc + if resolved.is_symlink() or not resolved.is_file(): + raise M48sReferenceGraphRuntimeError( + "pinned RF-DETR graph input must be a regular file" + ) + digest = hashlib.sha256() + with resolved.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +__all__ = [ + "M48sReferenceGraphRuntime", + "M48sReferenceGraphRuntimeError", + "build_m48s_reference_graph_runtime", +] diff --git a/src/k1link/perception/m48s_replay_timeline.py b/src/k1link/perception/m48s_replay_timeline.py new file mode 100644 index 0000000..9da7e77 --- /dev/null +++ b/src/k1link/perception/m48s_replay_timeline.py @@ -0,0 +1,426 @@ +"""Bounded camera, LiDAR, and world-state projection of sealed M4.8S replay evidence.""" + +from __future__ import annotations + +import copy +import json +import math +import statistics +from dataclasses import dataclass +from itertools import pairwise +from pathlib import Path +from threading import RLock +from typing import Final + +import numpy as np + +from .geometry import RecordedGeometryStore +from .geometry_math import project_map_points_kb4 +from .recorded_source import RECORDED_REPRESENTATION_ID +from .spatial_evidence import project_metric_obstacles_to_body, sample_points_in_body_frame +from .threat import ( + DEFAULT_REPLAY_THREAT_PROFILE_PATH, + RecordedReplayBodyFrameResolver, + load_replay_threat_profile, +) +from .threat_timeline import ( + RECORDED_LOCAL_SURFACE_POINT_LIMIT, + RECORDED_LOCAL_SURFACE_RADIUS_M, + RECORDED_LOCAL_SURFACE_VOXEL_SIZE_M, + RECORDED_LOCAL_SURFACE_WINDOW_SECONDS, + RECORDED_SPATIAL_CHUNK_SCHEMA, + RECORDED_SPATIAL_FRAME_SCHEMA, + RECORDED_SPATIAL_MAX_CHUNK_FRAMES, + RECORDED_SPATIAL_POINT_LIMIT, + RECORDED_SPATIAL_TIMELINE_SCHEMA, +) + +FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v0" +EXPECTED_FRAME_COUNT: Final = 4_489 + + +class M48sReplayTimelineError(RuntimeError): + """The sealed M4.8S evidence cannot produce an exact bounded timeline.""" + + +@dataclass(frozen=True, slots=True) +class _LedgerIndex: + offsets_by_sequence: dict[int, int] + + +class M48sReplayTimeline: + """Read source-indexed chunks while preserving latest-wins world-state gaps.""" + + def __init__(self, *, repository_root: Path, result_root: Path, result_id: str) -> None: + self.repository_root = repository_root.resolve(strict=True) + self.result_root = result_root.resolve(strict=True) + self.result_id = result_id + self.frames_path = ( + self.result_root / "reference-graph-replay-frames.jsonl" + ).resolve(strict=True) + self.worker_path = ( + self.result_root / "reference-graph-replay-worker-result.json" + ).resolve(strict=True) + if ( + self.frames_path.parent != self.result_root + or self.worker_path.parent != self.result_root + or self.frames_path.is_symlink() + or self.worker_path.is_symlink() + ): + raise M48sReplayTimelineError("M4.8S replay artifacts are invalid") + self.profile = load_replay_threat_profile( + self.repository_root / DEFAULT_REPLAY_THREAT_PROFILE_PATH + ) + self.store = RecordedGeometryStore.from_repository(self.repository_root) + if self.store.profile.frame_count != EXPECTED_FRAME_COUNT: + raise M48sReplayTimelineError("M4.8S source frame count changed") + self.body_frames = RecordedReplayBodyFrameResolver( + self.store, + profile=self.profile.body_frame, + ) + self.source_times_ns = tuple( + self.store.temporal_binding_for_index(sequence).source_time_ns + for sequence in range(EXPECTED_FRAME_COUNT) + ) + if any(right <= left for left, right in pairwise(self.source_times_ns)): + raise M48sReplayTimelineError("M4.8S source clock is not monotonic") + worker = _object(json.loads(self.worker_path.read_text("utf-8")), "worker result") + self.outcomes = _terminal_outcomes(worker) + self.index = _index_ledger(self.frames_path, self.source_times_ns, self.outcomes) + self._lock = RLock() + + def metadata(self) -> dict[str, object]: + intervals = [ + (current - previous) / 1_000_000_000 + for previous, current in pairwise(self.source_times_ns) + ] + nominal_interval = statistics.median(intervals) + return { + "schema_version": RECORDED_SPATIAL_TIMELINE_SCHEMA, + "result_id": self.result_id, + "recorded_source": { + "session_id": self.profile.session_id, + "source_id": self.profile.source_id, + "representation_id": RECORDED_REPRESENTATION_ID, + "synchronization": "host-arrival-best-effort", + }, + "frame_count": EXPECTED_FRAME_COUNT, + "frame_times_ns": list(self.source_times_ns), + "timeline_start_seconds": self.source_times_ns[0] / 1_000_000_000, + "timeline_end_seconds": self.source_times_ns[-1] / 1_000_000_000, + "nominal_frame_interval_seconds": nominal_interval, + "nominal_rate_hz": 1 / nominal_interval, + "max_chunk_frames": RECORDED_SPATIAL_MAX_CHUNK_FRAMES, + "point_sample_limit": RECORDED_SPATIAL_POINT_LIMIT, + "maximum_source_points_per_frame": self.store.maximum_current_point_count, + "point_delivery": "exact-current-increment", + "camera_point_delivery": "factory-kb4-projected-current-increment", + "camera_point_sample_limit": RECORDED_SPATIAL_POINT_LIMIT, + "world_state_delivery": "source-paced-latest-wins", + "world_state_frame_count": len(self.index.offsets_by_sequence), + "superseded_frame_count": sum(value == "superseded" for value in self.outcomes.values()), + "local_surface_visualization": { + "derivation": "bounded-registered-increment-accumulation", + "window_seconds": RECORDED_LOCAL_SURFACE_WINDOW_SECONDS, + "voxel_size_m": RECORDED_LOCAL_SURFACE_VOXEL_SIZE_M, + "radius_m": RECORDED_LOCAL_SURFACE_RADIUS_M, + "point_limit": RECORDED_LOCAL_SURFACE_POINT_LIMIT, + "authority": "visual-derived", + }, + "image_width": 800, + "image_height": 600, + "rig": { + "length_m": self.profile.rig.body_length_m, + "width_m": self.profile.rig.body_width_m, + "nominal_sensor_height_m": self.profile.rig.nominal_sensor_height_m, + }, + "corridor": { + "forward_length_m": self.profile.corridor.forward_length_m, + "rear_margin_m": self.profile.corridor.rear_margin_m, + "occupied_voxel_size_m": self.profile.corridor.occupied_voxel_size_m, + "half_width_m": ( + self.profile.rig.body_width_m / 2 + + self.profile.corridor.lateral_clearance_m + ), + "prediction_horizon_seconds": self.profile.corridor.prediction_horizon_seconds, + }, + "ground_truth": False, + "authority": "replay-simulated", + "access": "read-only-bounded-recorded-replay", + } + + def chunk(self, *, start_sequence: int, frame_count: int) -> dict[str, object]: + if not 0 <= start_sequence < EXPECTED_FRAME_COUNT: + raise M48sReplayTimelineError("M4.8S timeline start is invalid") + if not 1 <= frame_count <= RECORDED_SPATIAL_MAX_CHUNK_FRAMES: + raise M48sReplayTimelineError("M4.8S timeline chunk size is invalid") + stop = min(EXPECTED_FRAME_COUNT, start_sequence + frame_count) + with self._lock: + frames = [self._project_frame(sequence) for sequence in range(start_sequence, stop)] + return { + "schema_version": RECORDED_SPATIAL_CHUNK_SCHEMA, + "result_id": self.result_id, + "start_sequence": start_sequence, + "frame_count": len(frames), + "next_sequence": stop if stop < EXPECTED_FRAME_COUNT else None, + "frames": frames, + "ground_truth": False, + "authority": "replay-simulated", + "access": "read-only-bounded-recorded-replay", + } + + def _project_frame(self, sequence: int) -> dict[str, object]: + terminal_outcome = self.outcomes[sequence] + row = self._row(sequence) + frame_id = ( + _text(_object(row["source_envelope"], "source envelope").get("frame_id"), "frame id") + if row is not None + else f"frame-{sequence:06d}" + ) + binding = self.store.temporal_binding_for_index(sequence) + frame = self.store.frame_for_index(sequence) + body_frame = self.body_frames.body_frame_for_frame(frame_id) + point_cloud: list[list[float]] = [] + point_source_count = 0 + projected_points: list[list[float]] = [] + projected_source_count = 0 + projected_front_count = 0 + projected_total = 0 + if frame is not None: + projected = project_map_points_kb4( + frame.points_map, + position_map_xyz=frame.sensor_position_map, + orientation_map_from_lidar_xyzw=frame.sensor_orientation_xyzw, + profile=frame.projection, + ) + projected_source_count = projected.source_point_count + projected_front_count = projected.camera_front_point_count + projected_total = projected.projected_point_count + stride = max(1, math.ceil(projected_total / RECORDED_SPATIAL_POINT_LIMIT)) + indices = np.arange(0, projected_total, stride, dtype=np.int64)[ + :RECORDED_SPATIAL_POINT_LIMIT + ] + if indices.size: + xy = projected.pixels_xy[indices] + depth = projected.depths_m[indices, None] + projected_points = np.round(np.concatenate((xy, depth), axis=1), 4).tolist() + if body_frame is not None: + point_cloud, point_source_count = sample_points_in_body_frame( + frame.points_map, + body_frame, + point_limit=RECORDED_SPATIAL_POINT_LIMIT, + ) + + metric_visuals: list[dict[str, object]] = [] + camera_proposals: list[dict[str, object]] = [] + assessments: list[dict[str, object]] = [] + if row is not None: + envelope = _object(row.get("source_envelope"), "source envelope") + timestamps = _object(envelope.get("timestamps"), "source timestamps") + if ( + envelope.get("sequence") != sequence + or timestamps.get("source_ns") != binding.source_time_ns + or terminal_outcome != "delivered" + ): + raise M48sReplayTimelineError("M4.8S delivered frame binding changed") + delivery = _object(row.get("delivery"), "delivery") + obstacle_map = _object(delivery.get("obstacle_map"), "obstacle map") + assessments = _objects(delivery.get("threats"), "threats") + assessment_by_component = { + _text(item.get("component_id"), "assessment component"): item + for item in assessments + } + metric_rows: list[dict[str, object]] = [] + for obstacle in ( + *_objects(obstacle_map.get("occupied"), "occupied obstacles"), + *_objects(obstacle_map.get("unknown"), "unknown obstacles"), + ): + component_id = _text(obstacle.get("component_id"), "component id") + centroid = obstacle.get("last_centroid_xyz_m") + if centroid is None: + continue + metric_rows.append( + { + "component_id": component_id, + "state": obstacle.get("state"), + "motion": obstacle.get("motion"), + "centroid_map_xyz_m": centroid, + "cells": obstacle.get("cells"), + "assessment": assessment_by_component[component_id], + } + ) + if body_frame is not None: + metric_visuals = project_metric_obstacles_to_body( + metric_rows, + body_frame, + occupied_voxel_size_m=self.profile.corridor.occupied_voxel_size_m, + ) + associated = set(_strings(row.get("associated_proposal_ids"), "associated ids")) + for proposal in _objects(row.get("detector_proposals"), "detector proposals"): + proposal_id = _text(proposal.get("proposal_id"), "proposal id") + region = _object(proposal.get("region"), "proposal region") + assessment = assessment_by_component.get(proposal_id) + camera_proposals.append( + { + "proposal_id": proposal_id, + "bbox_xyxy": [ + region.get("x_min"), + region.get("y_min"), + region.get("x_max"), + region.get("y_max"), + ], + "objectness": proposal.get("objectness"), + "semantic_hint": proposal.get("semantic_hint"), + "occupied_support": proposal_id in associated, + "range_m": None, + "threat_decision": None if assessment is None else assessment.get("decision"), + "threat_reason_codes": [] + if assessment is None + else assessment.get("reason_codes"), + } + ) + return { + "schema_version": RECORDED_SPATIAL_FRAME_SCHEMA, + "sequence": sequence, + "frame_id": frame_id, + "source_time_ns": binding.source_time_ns, + "session_seconds": binding.source_time_ns / 1_000_000_000, + "source_available": binding.source_available, + "spatial_available": body_frame is not None and frame is not None, + "world_state_available": row is not None, + "terminal_outcome": terminal_outcome, + "body_frame": None + if body_frame is None + else { + "origin_map_xyz_m": list(body_frame.origin_map_xyz_m), + "basis_map_from_body": [list(value) for value in body_frame.basis_map_from_body], + }, + "point_cloud_body_xyz_m": point_cloud, + "point_cloud_source_count": point_source_count, + "point_cloud_sample_count": len(point_cloud), + "point_cloud_layer": "current-increment", + "camera_projected_points_xyd": projected_points, + "camera_projected_source_count": projected_source_count, + "camera_projected_front_count": projected_front_count, + "camera_projected_point_count": projected_total, + "camera_projected_sample_count": len(projected_points), + "camera_projection": "factory-kb4-exact", + "rolling_map_component_count": sum( + item.get("state") == "retained" for item in metric_visuals + ), + "metric_obstacles": metric_visuals, + "camera_proposals": camera_proposals, + "decision_counts": _decision_counts(assessments), + "camera_url": ( + "/api/v1/laboratory/m48s/fixed-class-detector/" + f"{self.result_id}/timeline/frames/{sequence}/camera" + ), + "ground_truth": False, + "authority": "replay-simulated", + } + + def _row(self, sequence: int) -> dict[str, object] | None: + offset = self.index.offsets_by_sequence.get(sequence) + if offset is None: + return None + with self.frames_path.open("rb") as stream: + stream.seek(offset) + line = stream.readline() + value = json.loads(line) + if not isinstance(value, dict) or value.get("schema_version") != FRAME_EVIDENCE_SCHEMA: + raise M48sReplayTimelineError("M4.8S frame row is invalid") + return value + + +def _index_ledger( + path: Path, + source_times_ns: tuple[int, ...], + outcomes: dict[int, str], +) -> _LedgerIndex: + offsets: dict[int, int] = {} + with path.open("rb") as stream: + while True: + offset = stream.tell() + line = stream.readline() + if not line: + break + row = json.loads(line) + if not isinstance(row, dict) or row.get("schema_version") != FRAME_EVIDENCE_SCHEMA: + raise M48sReplayTimelineError("M4.8S ledger schema changed") + envelope = _object(row.get("source_envelope"), "source envelope") + timestamps = _object(envelope.get("timestamps"), "source timestamps") + sequence = envelope.get("sequence") + if ( + not isinstance(sequence, int) + or isinstance(sequence, bool) + or not 0 <= sequence < EXPECTED_FRAME_COUNT + or sequence in offsets + or outcomes.get(sequence) != "delivered" + or timestamps.get("source_ns") != source_times_ns[sequence] + ): + raise M48sReplayTimelineError("M4.8S ledger source binding changed") + offsets[sequence] = offset + delivered = {sequence for sequence, outcome in outcomes.items() if outcome == "delivered"} + if set(offsets) != delivered: + raise M48sReplayTimelineError("M4.8S ledger does not match delivered outcomes") + return _LedgerIndex(offsets) + + +def _terminal_outcomes(worker: dict[str, object]) -> dict[int, str]: + execution = _object(worker.get("execution"), "execution") + loops = execution.get("loops") + if not isinstance(loops, list) or len(loops) != 1: + raise M48sReplayTimelineError("M4.8S worker loop identity changed") + details = _object(loops[0], "worker loop").get("terminal_outcome_details") + result: dict[int, str] = {} + for item in _objects(details, "terminal outcomes"): + sequence = item.get("sequence") + outcome = item.get("outcome") + if ( + not isinstance(sequence, int) + or isinstance(sequence, bool) + or outcome not in {"delivered", "superseded"} + or sequence in result + ): + raise M48sReplayTimelineError("M4.8S terminal outcome is invalid") + result[sequence] = outcome + if sorted(result) != list(range(EXPECTED_FRAME_COUNT)): + raise M48sReplayTimelineError("M4.8S terminal outcomes are incomplete") + return result + + +def _decision_counts(assessments: list[dict[str, object]]) -> dict[str, int]: + result = {"threat": 0, "not-threat": 0, "unknown": 0} + for assessment in assessments: + decision = assessment.get("decision") + if not isinstance(decision, str) or decision not in result: + raise M48sReplayTimelineError("M4.8S threat decision is invalid") + result[decision] += 1 + return result + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict): + raise M48sReplayTimelineError(f"M4.8S {label} is invalid") + return value + + +def _objects(value: object, label: str) -> list[dict[str, object]]: + if not isinstance(value, list) or any(not isinstance(item, dict) for item in value): + raise M48sReplayTimelineError(f"M4.8S {label} are invalid") + return copy.deepcopy(value) + + +def _strings(value: object, label: str) -> list[str]: + if not isinstance(value, list) or any(not isinstance(item, str) for item in value): + raise M48sReplayTimelineError(f"M4.8S {label} are invalid") + return value + + +def _text(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise M48sReplayTimelineError(f"M4.8S {label} is invalid") + return value + + +__all__ = ["M48sReplayTimeline", "M48sReplayTimelineError"] diff --git a/tests/test_m48s_reference_graph_shadow.py b/tests/test_m48s_reference_graph_shadow.py new file mode 100644 index 0000000..761d056 --- /dev/null +++ b/tests/test_m48s_reference_graph_shadow.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from k1link.perception.detector import RF_DETR_SHADOW_PROVIDER_ID +from k1link.perception.m48s_advisory import ( + AdvisoryFamily, + advisory_policy_matrix, +) +from k1link.perception.object_understanding import AdvisoryResponse +from k1link.perception.providers import ProviderRole, ReferencePerceptionGraphConfigV2 + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +GRAPH_CONFIG = ( + REPOSITORY_ROOT + / "config/perception/m48s-rf-detr-reference-graph-shadow-v0.json" +) + + +def test_m48s_reference_graph_replaces_only_the_detector_pin() -> None: + shadow_document = json.loads(GRAPH_CONFIG.read_text("utf-8")) + canonical_document = json.loads( + (REPOSITORY_ROOT / "config/perception/m4-reference-graph-v2.json").read_text( + "utf-8" + ) + ) + shadow = ReferencePerceptionGraphConfigV2.from_dict(shadow_document) + canonical = ReferencePerceptionGraphConfigV2.from_dict(canonical_document) + shadow_pins = {item.role: item for item in shadow.providers} + canonical_pins = {item.role: item for item in canonical.providers} + + assert shadow.graph_id == canonical.graph_id == "reference-perception-graph/v2" + assert shadow.source_profile_id == canonical.source_profile_id + assert shadow.queues == canonical.queues + assert shadow.authority == canonical.authority + assert shadow_pins[ProviderRole.DETECTOR].provider_id == RF_DETR_SHADOW_PROVIDER_ID + assert shadow_pins[ProviderRole.DETECTOR] != canonical_pins[ProviderRole.DETECTOR] + assert all( + shadow_pins[role] == canonical_pins[role] + for role in ProviderRole + if role is not ProviderRole.DETECTOR + ) + + +def test_m48s_reference_graph_pins_every_profile_digest() -> None: + config = ReferencePerceptionGraphConfigV2.from_dict( + json.loads(GRAPH_CONFIG.read_text("utf-8")) + ) + paths = { + ProviderRole.SOURCE: "m4-recorded-realtime-baseline-v1.json", + ProviderRole.DETECTOR: "rf-detr-large-risk-shadow-v0.json", + ProviderRole.GEOMETRY: "m4-geometry-association-v1.json", + ProviderRole.TEMPORAL: "m4-temporal-motion-v1.json", + ProviderRole.MOTION: "m4-temporal-motion-v1.json", + ProviderRole.ROLLING: "m4-rolling-local-map-v1.json", + ProviderRole.THREAT: "m4-replay-threat-v3.json", + } + pins = {item.role: item for item in config.providers} + + for role, name in paths.items(): + payload = (REPOSITORY_ROOT / "config/perception" / name).read_bytes() + assert pins[role].sha256 == hashlib.sha256(payload).hexdigest() + + +def test_m48s_advisory_policy_is_bounded_distinct_and_commandless() -> None: + matrix = advisory_policy_matrix() + + assert set(matrix) == set(AdvisoryFamily) + assert matrix[AdvisoryFamily.GENERIC_OBSTACLE] == ( + AdvisoryResponse.ROUTE_AROUND, + ) + assert AdvisoryResponse.YIELD in matrix[AdvisoryFamily.PERSON] + assert AdvisoryResponse.STOP in matrix[AdvisoryFamily.ANIMAL] + assert AdvisoryResponse.REDUCE_SPEED in matrix[AdvisoryFamily.LIGHT_ROAD_USER] + assert AdvisoryResponse.MONITOR in matrix[AdvisoryFamily.VEHICLE] + assert len(set(matrix.values())) == len(matrix) diff --git a/tests/test_m48s_reference_graph_shadow_result.py b/tests/test_m48s_reference_graph_shadow_result.py new file mode 100644 index 0000000..8d064fe --- /dev/null +++ b/tests/test_m48s_reference_graph_shadow_result.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RESULT_ID = ( + "m48s-reference-graph-shadow-gate-" + "e8da7a521768daba0ead1a6e4803871ce3a85f91a7d8ee36c5719ac10433e791" +) +RESULT_ROOT = ( + REPOSITORY_ROOT + / ".runtime/compute-experiments/m48s-semantic-shadow/" + "reference-graph-shadow-results" + / RESULT_ID +) + + +def test_m48s_complete_reference_graph_shadow_is_accepted_without_production_authority() -> None: + manifest = json.loads((RESULT_ROOT / "manifest.json").read_text("utf-8")) + identity = manifest["identity"] + evidence = identity["evidence"] + execution = evidence["execution"] + completion = evidence["world_state_completion_age_ms"] + + assert manifest["result_id"] == RESULT_ID + assert identity["completed"] is True + assert identity["accepted"] is True + assert identity["production_accepted"] is False + assert identity["decision"]["source_paced_runtime_gate_accepted"] is True + assert identity["decision"]["detector_replacement_authorized"] is False + assert execution["admitted_frames"] == 4489 + assert execution["delivered_world_states"] == 4481 + assert execution["terminal_outcomes"] == {"delivered": 4481, "superseded": 8} + assert execution["effective_world_state_fps"] >= 9.5 + assert completion["p95"] <= 175.0 + assert all(evidence["checks"].values()) + assert evidence["semantic_advisory"]["additional_inference_passes"] == 0 + assert evidence["identity_continuity"]["duplicate_component_ids_within_frame"] == 0 + assert identity["authority"] == { + "actuation_allowed": False, + "candidate_accepted": False, + "commands_enabled": False, + "ground_truth": False, + "navigation_or_safety_accepted": False, + } + for name, expected_sha256 in manifest["artifacts"].items(): + assert hashlib.sha256((RESULT_ROOT / name).read_bytes()).hexdigest() == expected_sha256 diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py index 48521ee..7848d7b 100644 --- a/tests/test_perception_graph.py +++ b/tests/test_perception_graph.py @@ -4,7 +4,7 @@ import json import subprocess import sys import threading -from collections.abc import Iterator +from collections.abc import Callable, Iterator from dataclasses import replace from pathlib import Path from queue import Queue @@ -39,6 +39,7 @@ from k1link.perception.contracts import ( TimestampBundle, ) from k1link.perception.graph import ( + DeliveredFrame, GraphExecutionError, GraphRunMode, GraphRunResult, @@ -460,6 +461,8 @@ def _graph_v2( capacity: int = 8, run_mode: GraphRunMode = GraphRunMode.LOSSLESS_REPLAY, terminal_timeout_ns: int = 500_000_000, + delivery_observer: Callable[[DeliveredFrame, int], None] | None = None, + delivery_evidence_observer: Callable[..., None] | None = None, ) -> ReferencePerceptionGraphV2: return ReferencePerceptionGraphV2( config=_config_v2(capacity, terminal_timeout_ns), @@ -471,6 +474,8 @@ def _graph_v2( rolling=_Rolling(), threat=_Threat(), run_mode=run_mode, + delivery_observer=delivery_observer, + delivery_evidence_observer=delivery_evidence_observer, clock_ns=lambda: 10_000, ) @@ -527,6 +532,50 @@ def test_reference_graph_v2_publishes_current_and_retained_occupancy() -> None: } +def test_reference_graph_v2_observes_final_delivery_completion_age() -> None: + observed: list[tuple[int, int, int]] = [] + + def observe(delivery: DeliveredFrame, completion_age_ns: int) -> None: + observed.append( + ( + delivery.sequence, + delivery.obstacle_map.output_age_ns, + completion_age_ns, + ) + ) + + result = _graph_v2( + _Source((_packet(0, source_age_ns=123),)), + delivery_observer=observe, + ).run() + + assert result.state is GraphState.STOPPED + assert observed == [(0, 123, 123)] + + +def test_reference_graph_v2_observes_exact_delivery_evidence_inputs() -> None: + observed: list[tuple[int, int, tuple[str, ...], frozenset[str], int]] = [] + + def observe(delivery, packet, proposals, associated_ids, completion_age_ns) -> None: + observed.append( + ( + delivery.sequence, + packet.envelope.sequence, + tuple(proposal.proposal_id for proposal in proposals), + associated_ids, + completion_age_ns, + ) + ) + + result = _graph_v2( + _Source((_packet(0, source_age_ns=123),)), + delivery_evidence_observer=observe, + ).run() + + assert result.state is GraphState.STOPPED + assert observed == [(0, 0, ("proposal-0",), frozenset({"proposal-0"}), 123)] + + def test_reference_graph_v2_lossless_mode_applies_bounded_backpressure() -> None: release = Event() @@ -784,6 +833,32 @@ def test_decoded_recorded_source_attaches_images_without_detector_logic(tmp_path assert int(packets[1].image_payload[0, 0, 0]) == 7 +def test_decoded_recorded_source_primes_decoder_before_source_clock() -> None: + events: list[str] = [] + + class Source(RecordedRavnoves00Source): + def __init__(self) -> None: + pass + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + events.append("source-clock-started") + if not stop_event.is_set(): + yield _packet(0) + + class Decoder: + def frames(self, stop_event: Event) -> Iterator[np.ndarray]: + events.append("decoder-primed") + if not stop_event.is_set(): + yield np.zeros((600, 800, 3), dtype=np.uint8) + + packets = list( + DecodedRecordedSource(source=Source(), decoder=Decoder()).packets(Event()) + ) + + assert len(packets) == 1 + assert events == ["decoder-primed", "source-clock-started"] + + def test_camera_only_path_never_invents_metric_occupancy_or_free_space() -> None: result = _graph(_Source((_packet(0, lidar=False),))).run() delivery = result.deliveries[0]