From b0d0bc8d7f388b33aa71bebe1ad587ff965eea26 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 5 Aug 2026 19:18:01 +0300 Subject: [PATCH] feat(perception): add dual evidence replay threat --- config/laboratories/m4-replay-threat.json | 10 + config/laboratory-execution.json | 19 + config/laboratory-value-review.json | 9 +- config/perception/m4-replay-threat-v1.json | 55 + ...E_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md | 62 +- ...40-dual-evidence-replay-threat-boundary.md | 75 + scripts/build_m4_worker_shadow_artifact.py | 2 +- src/k1link/laboratory/execution.py | 17 + src/k1link/perception/geometry.py | 42 + src/k1link/perception/graph_validation.py | 10 +- src/k1link/perception/threat.py | 782 +++++++++++ src/k1link/perception/threat_replay.py | 1215 +++++++++++++++++ src/k1link/perception/threat_replay_cli.py | 43 + src/k1link/web/app.py | 12 + src/k1link/web/m4_threat_replay_api.py | 315 +++++ tests/test_laboratory_evidence_registry.py | 3 +- tests/test_laboratory_execution.py | 1 + .../test_laboratory_value_review_registry.py | 3 +- tests/test_m4_threat_replay_result.py | 113 ++ tests/test_perception_architecture.py | 33 + tests/test_perception_graph.py | 38 +- tests/test_replay_threat_provider.py | 277 ++++ 22 files changed, 3122 insertions(+), 14 deletions(-) create mode 100644 config/laboratories/m4-replay-threat.json create mode 100644 config/perception/m4-replay-threat-v1.json create mode 100644 docs/adr/0040-dual-evidence-replay-threat-boundary.md create mode 100644 src/k1link/perception/threat.py create mode 100644 src/k1link/perception/threat_replay.py create mode 100644 src/k1link/perception/threat_replay_cli.py create mode 100644 src/k1link/web/m4_threat_replay_api.py create mode 100644 tests/test_m4_threat_replay_result.py create mode 100644 tests/test_replay_threat_provider.py diff --git a/config/laboratories/m4-replay-threat.json b/config/laboratories/m4-replay-threat.json new file mode 100644 index 0000000..5dddf3a --- /dev/null +++ b/config/laboratories/m4-replay-threat.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "m4-replay-threat", + "evidence": { + "runtime_relative_root": "m4/replay-threat", + "result_id_prefix": "m4-threat-replay", + "document_name": "manifest.json", + "schema_version": "missioncore.perception-threat-replay-result/v1" + } +} diff --git a/config/laboratory-execution.json b/config/laboratory-execution.json index 4ca162d..e971a1c 100644 --- a/config/laboratory-execution.json +++ b/config/laboratory-execution.json @@ -1,6 +1,25 @@ { "schema_version": "missioncore.laboratory-execution-registry/v1", "definitions": [ + { + "work_id": "m4-replay-threat", + "lifecycle": "canonical", + "isolation": "core-adapter", + "adapter_id": "canonical.m4-replay-threat/v1", + "input_roles": [ + "repository_root", + "temporal_result_root", + "geometry_result_root", + "detector_result_root" + ], + "contracts": { + "source": "missioncore.perception-temporal-replay-result/v1", + "provider": "missioncore.dual-evidence-threat-provider/v1", + "graph": "missioncore.perception-threat-replay-graph/v1", + "run": "missioncore.laboratory-run/v1", + "evidence": "missioncore.perception-threat-replay-result/v1" + } + }, { "work_id": "e33-worker-shadow", "lifecycle": "canonical", diff --git a/config/laboratory-value-review.json b/config/laboratory-value-review.json index a31267f..1160f45 100644 --- a/config/laboratory-value-review.json +++ b/config/laboratory-value-review.json @@ -1,6 +1,6 @@ { "schema_version": "missioncore.laboratory-value-review-registry/v1", - "reviewed_at_utc": "2026-08-05T08:30:00Z", + "reviewed_at_utc": "2026-08-05T15:34:00Z", "entries": [ { "catalog_id": "e28-local-surface", @@ -191,6 +191,13 @@ "lifecycle": "current", "visual_evidence": "available" }, + { + "catalog_id": "m4-replay-threat", + "evidence_id": "m4-threat-replay-7e1613a3ea35638b5ea7a3f7c1c78fe9eba1a3adae540b652dec167f815d45b2", + "signal": "progress", + "lifecycle": "current", + "visual_evidence": "available" + }, { "catalog_id": "l34-right-yolox-truth-island-freeze", "evidence_id": "l34-right-yolox-truth-island-freeze-5175a03144978b25130019da6d37bceb8c6ed6aa3d0d3a4d2df4483e1e27ae76", diff --git a/config/perception/m4-replay-threat-v1.json b/config/perception/m4-replay-threat-v1.json new file mode 100644 index 0000000..fabb745 --- /dev/null +++ b/config/perception/m4-replay-threat-v1.json @@ -0,0 +1,55 @@ +{ + "schema_version": "missioncore.replay-threat-profile/v1", + "profile_id": "m4-ravnoves00-virtual-corridor/v1", + "provider_id": "dual-evidence-replay-threat/v1", + "source": { + "source_id": "RAVNOVES00", + "session_id": "20260720T065719Z_viewer_live", + "temporal_result_id": "m4-temporal-replay-9ed5dcd249ed3bcb81661dd18e2b854a7ffedf3fd2b92b9c994c3c70c34533f2", + "temporal_frames_sha256": "1bf1365bdb3f20214443d3f8b87a0fa88f9848af8ca0456b7ca364d37631c3fc", + "geometry_result_id": "m4-geometry-replay-8daf3109e3cf30b960b4b376032ff3b5ec58ca42a1e5899b841cf29fbcf14ad8", + "geometry_frames_sha256": "b4db5d0ebaba4d6268a1006707dc313c229f3dbdfd73b1d863cad5d853be8ac4", + "detector_result_id": "m4-detector-replay-11f83f2e0b81758ac2a5a5fc54e9d293b501678df5f6ef97b5c6069ba08605c5", + "detector_frames_sha256": "9bf5ae17938cd57c112278781b38d54a7187cf2dabc7bb0332acdb1efad721f5", + "source_pack_id": "e10-lidar-pack-576c994a6c814e2592dd6240ace3902a5db94843312c759a73ba0c9166157d2b", + "source_pack_sha256": "0685d24219d8236caf8b7f1685e93f6d6b59e7fd015a768d88a92bbe8b154944" + }, + "calibration": { + "calibration_id": "camera-1-kb4-05f3ad9b", + "content_identity_sha256": "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9", + "usage": "projection-binding-only" + }, + "virtual_rig": { + "profile_id": "virtual-handheld-body-1000x600/v1", + "body_length_m": 1.0, + "body_width_m": 0.6, + "lidar_reference": "virtual-body-center", + "nominal_sensor_height_m": 1.25, + "physical_mount_claimed": false + }, + "corridor": { + "profile_id": "ravnoves00-forward-corridor-8m/v1", + "forward_length_m": 8.0, + "rear_margin_m": 0.5, + "lateral_clearance_m": 0.2, + "prediction_horizon_seconds": 5.0, + "occupied_voxel_size_m": 0.45, + "minimum_motion_span_seconds": 0.2 + }, + "policy": { + "camera_only_decision": "unknown", + "held_or_stale_decision": "unknown", + "semantic_class_used": false, + "detector_identity_used": false, + "absence_of_points_means_free": false, + "geometry_only_is_eligible": true + }, + "authority": { + "mode": "replay-simulated", + "physical_live": false, + "physical_collision_accepted": false, + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false + } +} diff --git a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md index a05721e..b3606da 100644 --- a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md +++ b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md @@ -2,7 +2,7 @@ Date: 2026-08-05 -Status: in progress; M4.0–M4.5 accepted, M4.6 replay threat is next +Status: in progress; M4.0–M4.6 accepted, M4.7 Worker 006 cutover is next Audit base: `1b3e0b3` on `feat/simulation-polygon-s1` @@ -107,7 +107,7 @@ Those systems remain separate platform workstreams. | Degradation | E35 executes six deterministic full-source variants; maximum recovery is 0.102 s against a 0.25 s gate | Accepted reusable regression primitive | | Recorded pacing | E33 processes all 4,489 frames at 10.006 FPS with depth-two queues, zero replacement/drop/deadline miss and 2.668 ms result-age p95 | Accepted stage runner evidence, not end-to-end perception evidence | | Motion | E51 emits 22,885 bounded motion candidates with complete accounting and no map-frame jump candidate | Candidate implementation exists; moving/static correctness is not accepted | -| Threat/collision | E51/E53 intentionally publish collision unavailable because body and LiDAR-to-body geometry are unbound | Replay simulation is possible with an explicit virtual rig; physical threat acceptance remains deferred | +| Threat/collision | M4.6 evaluates all 4,489 RAVNOVES00 frames with an explicit 1.0 × 0.6 m virtual body, 1.25 m sensor height and 8 m corridor | Accepted only as `replay-simulated`; physical threat/collision acceptance remains deferred | | Realtime worker | Worker 006 has healthy Triton and persistent perception containers plus live Telegraf | Infrastructure exists | | Worker graph | The persistent process still executes `run_e15_shadow_inference.py serve` with E15/E19/E8/E3/E23 profiles | Open architectural blocker: runtime remains LAB-generation-specific | | Model service | Canonical Triton currently exposes pinned `yolox_s` and `pointpillars`; PointPillars was rejected as a K1 product candidate | Reuse `yolox_s`; do not reopen PointPillars | @@ -425,6 +425,10 @@ Exit: ### M4.6 — implement replay-only threat assessment +Status: accepted on 2026-08-05. See the implementation record below and ADR +0040. The virtual dimensions are a replay hypothesis, not a retroactive physical +rig measurement. + Deliverables: - define a versioned virtual rig and corridor profile for RAVNOVES00 replay; @@ -613,7 +617,7 @@ Milestone 4 is complete only when all of the following are true: contract. - [ ] Current, held, stale, unavailable and conflict states are explicit. - [ ] Moving/static/unknown state is measured without semantic-class dependence. -- [ ] Replay-only threat assessment is explicit and cannot claim physical authority. +- [x] Replay-only threat assessment is explicit and cannot claim physical authority. - [ ] Worker 006 runs the canonical graph instead of the E15-specific server. - [ ] Full source-paced replay, deterministic replay, degradation, recovery and soak gates pass. @@ -908,8 +912,56 @@ canonical temporal or motion bytes. All fifteen M4.5 acceptance requirements are true. Synthetic tests cover ID and semantic-hint changes, moving, stationary, held, expiry, camera-only uncertainty -and map-frame discontinuity. M4.5 is closed; M4.6 replay-only threat assessment -is the next implementation phase. +and map-frame discontinuity. This closed M4.5 and supplied the immutable input to +the following M4.6 replay-only threat phase. + +### 2026-08-05 — M4.6 dual-evidence replay threat + +M4.6 is closed by `k1link.perception.threat` and the immutable replay builder in +`k1link.perception.threat_replay`: + +- `DualEvidenceReplayThreatProvider` consumes the canonical `LocalObstacleMap`; + it does not select camera-first or LiDAR-first execution; +- current LiDAR metric components are eligible for corridor assessment even when + they have no semantic class or camera association; +- camera-only observations and held/expired metric evidence publish `unknown`, + never `not-threat`; +- semantic hint and ephemeral detector/component identity do not participate in + corridor intersection, closest approach or TTC; +- the versioned replay profile fixes a virtual `1.0 × 0.6 m` body, nominal + `1.25 m` sensor height, `8 m` forward corridor and `5 s` bounded prediction + horizon; all documents retain `replay-simulated`, physical-collision false and + actuation false authority. + +The accepted immutable result is +`m4-threat-replay-7e1613a3ea35638b5ea7a3f7c1c78fe9eba1a3adae540b652dec167f815d45b2`: + +- `4,489 / 4,489` frames completed, zero failed; +- `27,299` current metric, `37,995` stale/held and `10,158` camera-only evidence + publications were each assessed exactly once; +- decisions: `8,010 threat`, `6,610 not-threat`, `60,832 unknown`; +- `21,958` geometry-only assessments remained in the decision path without a + class requirement; +- deterministic fixtures passed `9 / 9`; all four critical fixtures avoided a + false `not-threat` outcome; +- local uncapped execution measured `132.812 FPS`; provider latency was + `3.932 ms` p50 and `17.567 ms` p95; +- deterministic frame, visual and fixture ledgers are sealed by SHA-256 + `bf690358efb45c323db7172251074b33c3ef7ede6ae99bd8d3da53cfba86b142`, + `fb022c6efd84f27c0916a6c87887443c9b43993ac4b1f9910332433152533dea` + and `e217b61f3e8cf444f2620c0d815c18b2131eaafca29352bf12f78e05db96ee13`. + +The standard LAB catalog exposes the exact result with a common evidence viewer: +full recorded VIDEO, exact CAMERA samples with ranges/unknown boxes, and the same +32 synchronized LiDAR point-cloud samples in interactive 3D and plan view. The +recorded box overlay was extracted from E46C into a reusable component rather +than copied into an M4-specific renderer. Visual availability is evidence for +inspection, not independent ground truth. + +M4.6 does not close moving/static correctness or object-presence correctness; +those remain the independent M4.8 gate. It also does not authorize a physical +mount, live K1, navigation, collision safety or commands. M4.7 is now the next +implementation phase. ## Implementation order diff --git a/docs/adr/0040-dual-evidence-replay-threat-boundary.md b/docs/adr/0040-dual-evidence-replay-threat-boundary.md new file mode 100644 index 0000000..1244229 --- /dev/null +++ b/docs/adr/0040-dual-evidence-replay-threat-boundary.md @@ -0,0 +1,75 @@ +# ADR 0040: Dual-evidence replay threat boundary + +Date: 2026-08-05 +Status: accepted and implemented for M4.6 + +## Context + +Historical camera-first experiments correctly kept camera semantics separate +from LiDAR metric support, but the phrase "camera-first" is not an acceptable +product threat architecture. The RAVNOVES00 camera detector visibly misses some +unclassified occupied structures, while camera proposals without qualified +LiDAR support cannot establish metric clearance. Making either sensor a gate for +the other would discard useful evidence. + +The portable RAVNOVES00 recording also has no admitted measured vehicle body or +qualified LiDAR-to-body mount. A recorded threat experiment therefore needs an +explicit virtual geometry without weakening the physical rig contract in ADR +0035. + +## Decision + +Mission Core threat assessment consumes two independent evidence paths: + +```text +camera proposals ---------------------> camera-only uncertainty + | | + +---- optional association ----+ | + v v +LiDAR occupied geometry ----------> LocalObstacleMap ---> ThreatAssessment +``` + +Neither path is called first: + +- camera publishes image-space object proposals and optional semantics; +- LiDAR publishes metric occupied components, including geometry with no class; +- association enriches evidence but is not an admission gate; +- current metric geometry may produce `threat` or `not-threat` from corridor + geometry and bounded relative motion; +- camera-only, held, expired or otherwise incomplete evidence produces + `unknown`, never a safe decision; +- semantic class, detector ID and persistent identity are excluded from the + threat calculation. + +M4.6 fixes a versioned replay hypothesis: body length `1.0 m`, width `0.6 m`, +nominal sensor height `1.25 m`, forward corridor `8 m`, rear margin `0.5 m`, +lateral clearance `0.2 m` and prediction horizon `5 s`. These values may be used +only with `replay-simulated` authority. They do not populate or qualify +`missioncore.rig-geometry/v1`, and they cannot support physical collision, +navigation, safety or actuation claims. + +## Evidence and presentation + +The accepted replay must publish immutable frame, fixture, report and visual +ledgers. Visual evidence uses the common LAB viewer and reusable renderers: + +- full recorded camera video with synchronized proposal boxes; +- exact camera samples with metric range or explicit missing range; +- synchronized point cloud, occupied cells, virtual body and corridor in 3D and + plan view; +- visible threat/not-threat/unknown and `replay-simulated` authority. + +Visuals are an inspection surface, not ground truth. Independent object-centric +labels remain a separate gate. + +## Consequences + +- Unclassified concrete, vegetation or road furniture can remain visible to the + metric path without inventing a semantic label. +- A camera detection cannot become safe merely because LiDAR support is absent. +- New detectors and LiDAR geometry providers may replace either side behind the + existing contracts without changing the threat provider. +- Physical body/mount qualification and live acceptance remain intentional debt. +- ADR 0035 remains valid for ownership of semantics, metric support and physical + rig qualification; this ADR supersedes camera-first wording for the canonical + product decision graph. diff --git a/scripts/build_m4_worker_shadow_artifact.py b/scripts/build_m4_worker_shadow_artifact.py index 00d4fe6..5ec04d7 100644 --- a/scripts/build_m4_worker_shadow_artifact.py +++ b/scripts/build_m4_worker_shadow_artifact.py @@ -25,7 +25,7 @@ WHEEL_NAME = "nodedc_mission_core-0.1.0-py3-none-any.whl" RUNNER_NAME = RUNNER.name PATCH_ID = re.compile(r"^[A-Za-z0-9._-]{1,96}$") EXPECTED_BASELINE_SHA256 = "ea10359339e6cce31b5780a2710299771cab7cc0c1c2a2b56a1621f786b31fa8" -EXPECTED_WHEEL_SHA256 = "19d8caf9a522747c461fb3ca30aafe54169959d8bd8e671fa6fc8c0ac107875d" +EXPECTED_WHEEL_SHA256 = "94ed4b7e70471d343eafa9728ce1bd496551a6ee2c2d9e3e67fa5c6c2eadfc5b" PAYLOAD_FILES = ( RUNNER_NAME, WHEEL_NAME, diff --git a/src/k1link/laboratory/execution.py b/src/k1link/laboratory/execution.py index ae0417d..e7ecc56 100644 --- a/src/k1link/laboratory/execution.py +++ b/src/k1link/laboratory/execution.py @@ -311,12 +311,29 @@ class LaboratoryRunner: def canonical_laboratory_adapters() -> dict[str, LaboratoryAdapter]: return { + "canonical.m4-replay-threat/v1": _run_m4_replay_threat, "canonical.e33-worker-shadow/v1": _run_e33, "canonical.e35-degradation-recovery/v1": _run_e35, "canonical.e46j-raw-fisheye-realtime/v1": _run_e46j, } +def _run_m4_replay_threat(request: LaboratoryRunRequest) -> LaboratoryAdapterResult: + from k1link.perception.threat_replay import build_threat_replay + + result = build_threat_replay( + repository_root=request.inputs["repository_root"], + temporal_result_root=request.inputs["temporal_result_root"], + geometry_result_root=request.inputs["geometry_result_root"], + detector_result_root=request.inputs["detector_result_root"], + output_root=request.output_root, + ) + return LaboratoryAdapterResult( + result_root=result.result_root, + result_id=result.result_id, + ) + + def _run_e33(request: LaboratoryRunRequest) -> LaboratoryAdapterResult: from k1link.compute.e33_worker_shadow import run_e33_worker_shadow diff --git a/src/k1link/perception/geometry.py b/src/k1link/perception/geometry.py index 0a93c5a..e0bb81f 100644 --- a/src/k1link/perception/geometry.py +++ b/src/k1link/perception/geometry.py @@ -245,6 +245,48 @@ class RecordedGeometryStore: points.setflags(write=False) return points + def pose_values_for_frame( + self, + frame_id: str, + ) -> tuple[tuple[float, float, float], tuple[float, float, float, float]] | None: + """Return one verified replay pose without exposing the source archive.""" + + prefix = "frame-" + if not frame_id.startswith(prefix) or not frame_id[len(prefix) :].isdigit(): + raise GeometryProviderError("replay pose frame identity is invalid") + frame_index = int(frame_id[len(prefix) :]) + if not 0 <= frame_index < self.profile.frame_count: + raise GeometryProviderError("replay pose frame is outside the source profile") + if not bool(self._source["sample_available"][frame_index]): + return None + position = np.asarray( + self._source["pose_positions_map"][frame_index], + dtype=np.float64, + ) + orientation = np.asarray( + self._source["pose_quaternions_map_from_lidar"][frame_index], + dtype=np.float64, + ) + if not np.isfinite(position).all() or not np.isfinite(orientation).all(): + raise GeometryProviderError("available replay pose is not finite") + return ( + (float(position[0]), float(position[1]), float(position[2])), + ( + float(orientation[0]), + float(orientation[1]), + float(orientation[2]), + float(orientation[3]), + ), + ) + + def available_frame_indices(self) -> tuple[int, ...]: + """Expose the immutable availability partition for deterministic sampling.""" + + return tuple( + int(index) + for index in np.flatnonzero(self._source["sample_available"]) + ) + def _validate(self) -> None: source_required = { "frame_indices", diff --git a/src/k1link/perception/graph_validation.py b/src/k1link/perception/graph_validation.py index f376927..34ce01c 100644 --- a/src/k1link/perception/graph_validation.py +++ b/src/k1link/perception/graph_validation.py @@ -73,14 +73,20 @@ def validate_threats( obstacle_map: LocalObstacleMap, threats: tuple[ThreatAssessment, ...], ) -> None: - component_ids = { + evidence_ids = { obstacle.component_id for obstacle in (*obstacle_map.occupied, *obstacle_map.unknown) } + evidence_ids.update(proposal.proposal_id for proposal in obstacle_map.camera_uncertainty) assessment_ids = [threat.assessment_id for threat in threats] + assessed_ids = [threat.component_id for threat in threats] if len(set(assessment_ids)) != len(assessment_ids): raise GraphExecutionError("threat assessment identities are duplicated") - if any(threat.component_id not in component_ids for threat in threats): + if len(set(assessed_ids)) != len(assessed_ids): + raise GraphExecutionError("threat evidence references are duplicated") + if any(threat.component_id not in evidence_ids for threat in threats): raise GraphExecutionError("threat assessment references an unknown component") + if set(assessed_ids) != evidence_ids: + raise GraphExecutionError("threat assessment coverage is incomplete") __all__ = [ diff --git a/src/k1link/perception/threat.py b/src/k1link/perception/threat.py new file mode 100644 index 0000000..e2df007 --- /dev/null +++ b/src/k1link/perception/threat.py @@ -0,0 +1,782 @@ +"""Replay-only virtual-corridor threat assessment for Mission Core M4.6. + +The provider consumes the canonical object map and a source-bound replay pose. +It never reads semantic class or detector identity when calculating geometry, +motion, corridor intersection, closest approach or TTC. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from dataclasses import dataclass +from pathlib import Path +from typing import Final, Protocol + +from .contracts import ( + CorridorIntersection, + LocalObstacleMap, + MotionState, + QualificationState, + TemporalObstacle, + TemporalState, + ThreatAssessment, + ThreatDecision, +) +from .geometry_math import quaternion_xyzw_to_rotation_matrix + +REPLAY_THREAT_PROFILE_SCHEMA: Final = "missioncore.replay-threat-profile/v1" +REPLAY_THREAT_PROVIDER_ID: Final = "dual-evidence-replay-threat/v1" +DEFAULT_REPLAY_THREAT_PROFILE_PATH: Final = ( + "config/perception/m4-replay-threat-v1.json" +) + + +class ReplayThreatError(ValueError): + """A virtual rig, source pose or threat input is ambiguous or unsafe.""" + + +@dataclass(frozen=True, slots=True) +class ReplayPose: + frame_id: str + position_map_xyz_m: tuple[float, float, float] + orientation_map_from_lidar_xyzw: tuple[float, float, float, float] + + def __post_init__(self) -> None: + if not self.frame_id: + raise ReplayThreatError("replay pose frame id is empty") + if ( + len(self.position_map_xyz_m) != 3 + or len(self.orientation_map_from_lidar_xyzw) != 4 + or not all( + math.isfinite(value) + for value in ( + *self.position_map_xyz_m, + *self.orientation_map_from_lidar_xyzw, + ) + ) + ): + raise ReplayThreatError("replay pose is not finite") + norm = math.sqrt(sum(value * value for value in self.orientation_map_from_lidar_xyzw)) + if norm < 1e-9: + raise ReplayThreatError("replay pose orientation has no usable norm") + + def map_point_to_body( + self, + point_map_xyz_m: tuple[float, float, float], + ) -> tuple[float, float, float]: + rotation = quaternion_xyzw_to_rotation_matrix( + self.orientation_map_from_lidar_xyzw + ) + delta = tuple( + point_map_xyz_m[index] - self.position_map_xyz_m[index] + for index in range(3) + ) + values = tuple( + float(sum(delta[row] * rotation[row, column] for row in range(3))) + for column in range(3) + ) + return values[0], values[1], values[2] + + +class ReplayPoseResolver(Protocol): + def pose_for_frame(self, frame_id: str) -> ReplayPose | None: ... + + +class RecordedReplayPoseResolver: + """Adapt the verified geometry store to the source-neutral pose seam.""" + + def __init__(self, store: object) -> None: + method = getattr(store, "pose_values_for_frame", None) + if not callable(method): + raise ReplayThreatError("recorded pose store does not expose verified poses") + self._pose_values_for_frame = method + + def pose_for_frame(self, frame_id: str) -> ReplayPose | None: + values = self._pose_values_for_frame(frame_id) + if values is None: + return None + position, orientation = values + return ReplayPose( + frame_id=frame_id, + position_map_xyz_m=position, + orientation_map_from_lidar_xyzw=orientation, + ) + + +@dataclass(frozen=True, slots=True) +class VirtualRigProfile: + profile_id: str + body_length_m: float + body_width_m: float + lidar_reference: str + nominal_sensor_height_m: float + + +@dataclass(frozen=True, slots=True) +class VirtualCorridorProfile: + profile_id: str + forward_length_m: float + rear_margin_m: float + lateral_clearance_m: float + prediction_horizon_seconds: float + occupied_voxel_size_m: float + minimum_motion_span_seconds: float + + +@dataclass(frozen=True, slots=True) +class ReplayThreatProfile: + profile_id: str + provider_id: str + source_id: str + session_id: str + temporal_result_id: str + temporal_frames_sha256: str + geometry_result_id: str + geometry_frames_sha256: str + detector_result_id: str + detector_frames_sha256: str + source_pack_id: str + source_pack_sha256: str + calibration_id: str + calibration_content_sha256: str + rig: VirtualRigProfile + corridor: VirtualCorridorProfile + profile_sha256: str + + +class DualEvidenceReplayThreatProvider: + """Assess metric and nonmetric evidence without choosing a primary sensor.""" + + provider_id: str = REPLAY_THREAT_PROVIDER_ID + + def __init__( + self, + *, + pose_resolver: ReplayPoseResolver, + profile: ReplayThreatProfile, + ) -> None: + if profile.provider_id != self.provider_id: + raise ReplayThreatError("threat provider identity changed") + self.pose_resolver = pose_resolver + self.profile = profile + + def assess(self, obstacle_map: LocalObstacleMap) -> tuple[ThreatAssessment, ...]: + if ( + obstacle_map.source_id != self.profile.source_id + or obstacle_map.session_id != self.profile.session_id + ): + raise ReplayThreatError("obstacle map escaped the threat profile") + pose = self.pose_resolver.pose_for_frame(obstacle_map.frame_id) + assessments = [ + self._metric_or_stale(obstacle_map.frame_id, obstacle, pose) + for obstacle in (*obstacle_map.occupied, *obstacle_map.unknown) + ] + assessments.extend( + self._camera_only(obstacle_map.frame_id, proposal.proposal_id) + for proposal in obstacle_map.camera_uncertainty + ) + return tuple(assessments) + + def _metric_or_stale( + self, + frame_id: str, + obstacle: TemporalObstacle, + pose: ReplayPose | None, + ) -> ThreatAssessment: + if obstacle.state is not TemporalState.CURRENT: + return self._unknown( + frame_id, + obstacle.component_id, + ("stale-evidence", f"temporal-state-{obstacle.state.value}"), + ) + if pose is None or obstacle.last_centroid_xyz_m is None or not obstacle.cells: + return self._unknown( + frame_id, + obstacle.component_id, + ("current-pose-or-metric-geometry-unavailable",), + ) + + centroid_body = pose.map_point_to_body(obstacle.last_centroid_xyz_m) + cells_body = tuple( + pose.map_point_to_body( + ( + (cell.x + 0.5) * self.profile.corridor.occupied_voxel_size_m, + (cell.y + 0.5) * self.profile.corridor.occupied_voxel_size_m, + (cell.z + 0.5) * self.profile.corridor.occupied_voxel_size_m, + ) + ) + for cell in obstacle.cells + ) + velocity_body = self._relative_velocity_body(obstacle, pose) + corridor_entry = _first_corridor_entry_seconds( + cells_body, + velocity_body, + rig=self.profile.rig, + corridor=self.profile.corridor, + ) + current_intersection = _intersects_corridor_now( + cells_body, + rig=self.profile.rig, + corridor=self.profile.corridor, + ) + motion_complete = ( + obstacle.motion is not MotionState.UNKNOWN and velocity_body is not None + ) + if current_intersection or (motion_complete and corridor_entry is not None): + intersection = CorridorIntersection.INTERSECTS + decision = ThreatDecision.THREAT + reasons = [ + "current-corridor-intersection" + if current_intersection + else "predicted-corridor-intersection", + "metric-lidar-geometry", + ] + elif motion_complete: + intersection = CorridorIntersection.CLEAR + decision = ThreatDecision.NOT_THREAT + reasons = ["predicted-corridor-clear", "metric-lidar-geometry"] + else: + intersection = CorridorIntersection.UNKNOWN + decision = ThreatDecision.UNKNOWN + reasons = ["motion-incomplete", "metric-lidar-geometry"] + + closest = _closest_body_clearance_m( + cells_body, + velocity_body, + rig=self.profile.rig, + horizon_seconds=self.profile.corridor.prediction_horizon_seconds, + ) + ttc = _first_body_entry_seconds( + cells_body, + velocity_body, + rig=self.profile.rig, + voxel_size_m=self.profile.corridor.occupied_voxel_size_m, + horizon_seconds=self.profile.corridor.prediction_horizon_seconds, + ) + relative_speed = _closing_speed_mps(centroid_body, velocity_body) + if velocity_body is None: + reasons.append(f"motion-{obstacle.motion_reason}") + else: + reasons.append(f"motion-{obstacle.motion.value}") + if obstacle.semantic_hint is None: + reasons.append("geometry-only-evidence") + else: + reasons.append("camera-lidar-associated-evidence") + return ThreatAssessment( + assessment_id=_assessment_id(frame_id, obstacle.component_id), + component_id=obstacle.component_id, + rig_profile_id=self.profile.rig.profile_id, + corridor_profile_id=self.profile.corridor.profile_id, + qualification=QualificationState.QUALIFIED, + relative_speed_mps=relative_speed, + closest_approach_m=closest, + ttc_seconds=ttc, + corridor_intersection=intersection, + decision=decision, + reason_codes=tuple(reasons), + ) + + def _relative_velocity_body( + self, + obstacle: TemporalObstacle, + current_pose: ReplayPose, + ) -> tuple[float, float] | None: + if len(obstacle.history) < 2: + return None + first = obstacle.history[0] + last = obstacle.history[-1] + span_seconds = (last.evidence_time_ns - first.evidence_time_ns) / 1_000_000_000 + if span_seconds < self.profile.corridor.minimum_motion_span_seconds: + return None + first_pose = self.pose_resolver.pose_for_frame(first.frame_id) + last_pose = self.pose_resolver.pose_for_frame(last.frame_id) + if first_pose is None or last_pose is None or last.frame_id != current_pose.frame_id: + return None + obstacle_delta = tuple( + last.centroid_xyz_m[index] - first.centroid_xyz_m[index] + for index in range(3) + ) + rig_delta = tuple( + last_pose.position_map_xyz_m[index] - first_pose.position_map_xyz_m[index] + for index in range(3) + ) + relative_map = tuple( + (obstacle_delta[index] - rig_delta[index]) / span_seconds + for index in range(3) + ) + rotation = quaternion_xyzw_to_rotation_matrix( + current_pose.orientation_map_from_lidar_xyzw + ) + body = tuple( + float(sum(relative_map[row] * rotation[row, column] for row in range(3))) + for column in range(3) + ) + return body[0], body[1] + + def _camera_only(self, frame_id: str, proposal_id: str) -> ThreatAssessment: + return self._unknown( + frame_id, + proposal_id, + ("camera-only-no-metric-geometry", "absence-of-lidar-is-not-safe"), + ) + + def _unknown( + self, + frame_id: str, + component_id: str, + reasons: tuple[str, ...], + ) -> ThreatAssessment: + return ThreatAssessment( + assessment_id=_assessment_id(frame_id, component_id), + component_id=component_id, + rig_profile_id=self.profile.rig.profile_id, + corridor_profile_id=self.profile.corridor.profile_id, + qualification=QualificationState.UNQUALIFIED, + relative_speed_mps=None, + closest_approach_m=None, + ttc_seconds=None, + corridor_intersection=CorridorIntersection.UNKNOWN, + decision=ThreatDecision.UNKNOWN, + reason_codes=reasons, + ) + + +def load_replay_threat_profile(path: Path) -> ReplayThreatProfile: + if not path.is_file() or path.is_symlink(): + raise ReplayThreatError("replay threat profile is not a regular file") + raw = path.read_bytes() + try: + document = _object(json.loads(raw), "replay threat profile") + except json.JSONDecodeError as exc: + raise ReplayThreatError("replay threat profile JSON is invalid") from exc + _exact_keys( + document, + { + "schema_version", + "profile_id", + "provider_id", + "source", + "calibration", + "virtual_rig", + "corridor", + "policy", + "authority", + }, + "replay threat profile", + ) + if ( + document["schema_version"] != REPLAY_THREAT_PROFILE_SCHEMA + or document["provider_id"] != REPLAY_THREAT_PROVIDER_ID + ): + raise ReplayThreatError("replay threat profile identity is incompatible") + source = _object(document["source"], "threat source") + calibration = _object(document["calibration"], "threat calibration") + rig = _object(document["virtual_rig"], "virtual rig") + corridor = _object(document["corridor"], "virtual corridor") + policy = _object(document["policy"], "threat policy") + authority = _object(document["authority"], "threat authority") + _exact_keys( + source, + { + "source_id", + "session_id", + "temporal_result_id", + "temporal_frames_sha256", + "geometry_result_id", + "geometry_frames_sha256", + "detector_result_id", + "detector_frames_sha256", + "source_pack_id", + "source_pack_sha256", + }, + "threat source", + ) + _exact_keys( + calibration, + {"calibration_id", "content_identity_sha256", "usage"}, + "threat calibration", + ) + _exact_keys( + rig, + { + "profile_id", + "body_length_m", + "body_width_m", + "lidar_reference", + "nominal_sensor_height_m", + "physical_mount_claimed", + }, + "virtual rig", + ) + _exact_keys( + corridor, + { + "profile_id", + "forward_length_m", + "rear_margin_m", + "lateral_clearance_m", + "prediction_horizon_seconds", + "occupied_voxel_size_m", + "minimum_motion_span_seconds", + }, + "virtual corridor", + ) + _exact_keys( + policy, + { + "camera_only_decision", + "held_or_stale_decision", + "semantic_class_used", + "detector_identity_used", + "absence_of_points_means_free", + "geometry_only_is_eligible", + }, + "threat policy", + ) + _exact_keys( + authority, + { + "mode", + "physical_live", + "physical_collision_accepted", + "commands_enabled", + "actuation_allowed", + "navigation_or_safety_accepted", + }, + "threat authority", + ) + if ( + calibration.get("usage") != "projection-binding-only" + or rig.get("physical_mount_claimed") is not False + or policy + != { + "camera_only_decision": "unknown", + "held_or_stale_decision": "unknown", + "semantic_class_used": False, + "detector_identity_used": False, + "absence_of_points_means_free": False, + "geometry_only_is_eligible": True, + } + or authority + != { + "mode": "replay-simulated", + "physical_live": False, + "physical_collision_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, + } + ): + raise ReplayThreatError("replay threat policy or authority is unsafe") + virtual_rig = VirtualRigProfile( + profile_id=_string(rig, "profile_id"), + body_length_m=_positive_number(rig, "body_length_m"), + body_width_m=_positive_number(rig, "body_width_m"), + lidar_reference=_string(rig, "lidar_reference"), + nominal_sensor_height_m=_positive_number(rig, "nominal_sensor_height_m"), + ) + virtual_corridor = VirtualCorridorProfile( + profile_id=_string(corridor, "profile_id"), + forward_length_m=_positive_number(corridor, "forward_length_m"), + rear_margin_m=_nonnegative_number(corridor, "rear_margin_m"), + lateral_clearance_m=_nonnegative_number(corridor, "lateral_clearance_m"), + prediction_horizon_seconds=_positive_number( + corridor, "prediction_horizon_seconds" + ), + occupied_voxel_size_m=_positive_number(corridor, "occupied_voxel_size_m"), + minimum_motion_span_seconds=_positive_number( + corridor, "minimum_motion_span_seconds" + ), + ) + if virtual_rig.lidar_reference != "virtual-body-center": + raise ReplayThreatError("virtual LiDAR reference is unsupported") + for value in ( + source.get("temporal_frames_sha256"), + source.get("geometry_frames_sha256"), + source.get("detector_frames_sha256"), + source.get("source_pack_sha256"), + calibration.get("content_identity_sha256"), + ): + _sha256(value, "threat evidence digest") + return ReplayThreatProfile( + profile_id=_string(document, "profile_id"), + provider_id=_string(document, "provider_id"), + source_id=_string(source, "source_id"), + session_id=_string(source, "session_id"), + temporal_result_id=_string(source, "temporal_result_id"), + temporal_frames_sha256=_string(source, "temporal_frames_sha256"), + geometry_result_id=_string(source, "geometry_result_id"), + geometry_frames_sha256=_string(source, "geometry_frames_sha256"), + detector_result_id=_string(source, "detector_result_id"), + detector_frames_sha256=_string(source, "detector_frames_sha256"), + source_pack_id=_string(source, "source_pack_id"), + source_pack_sha256=_string(source, "source_pack_sha256"), + calibration_id=_string(calibration, "calibration_id"), + calibration_content_sha256=_string(calibration, "content_identity_sha256"), + rig=virtual_rig, + corridor=virtual_corridor, + profile_sha256=hashlib.sha256(raw).hexdigest(), + ) + + +def _assessment_id(frame_id: str, component_id: str) -> str: + digest = hashlib.sha256(f"{frame_id}\0{component_id}".encode()).hexdigest() + return f"threat-{digest}" + + +def _corridor_bounds( + rig: VirtualRigProfile, + corridor: VirtualCorridorProfile, + *, + expansion_m: float, +) -> tuple[float, float, float, float]: + return ( + -(rig.body_length_m / 2 + corridor.rear_margin_m + expansion_m), + rig.body_length_m / 2 + corridor.forward_length_m + expansion_m, + -(rig.body_width_m / 2 + corridor.lateral_clearance_m + expansion_m), + rig.body_width_m / 2 + corridor.lateral_clearance_m + expansion_m, + ) + + +def _body_bounds( + rig: VirtualRigProfile, + *, + expansion_m: float, +) -> tuple[float, float, float, float]: + return ( + -(rig.body_length_m / 2 + expansion_m), + rig.body_length_m / 2 + expansion_m, + -(rig.body_width_m / 2 + expansion_m), + rig.body_width_m / 2 + expansion_m, + ) + + +def _intersects_corridor_now( + cells_body: tuple[tuple[float, float, float], ...], + *, + rig: VirtualRigProfile, + corridor: VirtualCorridorProfile, +) -> bool: + expansion = corridor.occupied_voxel_size_m * math.sqrt(2) / 2 + bounds = _corridor_bounds(rig, corridor, expansion_m=expansion) + return any(_inside((point[0], point[1]), bounds) for point in cells_body) + + +def _first_corridor_entry_seconds( + cells_body: tuple[tuple[float, float, float], ...], + velocity_body: tuple[float, float] | None, + *, + rig: VirtualRigProfile, + corridor: VirtualCorridorProfile, +) -> float | None: + if velocity_body is None: + return None + expansion = corridor.occupied_voxel_size_m * math.sqrt(2) / 2 + bounds = _corridor_bounds(rig, corridor, expansion_m=expansion) + entries = ( + _ray_box_entry((point[0], point[1]), velocity_body, bounds) + for point in cells_body + ) + valid = [ + entry + for entry in entries + if entry is not None and entry <= corridor.prediction_horizon_seconds + ] + return None if not valid else round(min(valid), 12) + + +def _first_body_entry_seconds( + cells_body: tuple[tuple[float, float, float], ...], + velocity_body: tuple[float, float] | None, + *, + rig: VirtualRigProfile, + voxel_size_m: float, + horizon_seconds: float, +) -> float | None: + if velocity_body is None: + return None + expansion = voxel_size_m * math.sqrt(2) / 2 + bounds = _body_bounds(rig, expansion_m=expansion) + valid = [ + entry + for point in cells_body + if (entry := _ray_box_entry((point[0], point[1]), velocity_body, bounds)) + is not None + and entry <= horizon_seconds + ] + return None if not valid else round(min(valid), 12) + + +def _ray_box_entry( + point: tuple[float, float], + velocity: tuple[float, float], + bounds: tuple[float, float, float, float], +) -> float | None: + t_min = 0.0 + t_max = math.inf + for coordinate, speed, lower, upper in ( + (point[0], velocity[0], bounds[0], bounds[1]), + (point[1], velocity[1], bounds[2], bounds[3]), + ): + if abs(speed) < 1e-12: + if coordinate < lower or coordinate > upper: + return None + continue + first = (lower - coordinate) / speed + second = (upper - coordinate) / speed + near, far = min(first, second), max(first, second) + t_min = max(t_min, near) + t_max = min(t_max, far) + if t_min > t_max: + return None + return max(0.0, t_min) if t_max >= 0.0 else None + + +def _closest_body_clearance_m( + cells_body: tuple[tuple[float, float, float], ...], + velocity_body: tuple[float, float] | None, + *, + rig: VirtualRigProfile, + horizon_seconds: float, +) -> float: + bounds = _body_bounds(rig, expansion_m=0.0) + candidates = {0.0, horizon_seconds} + if velocity_body is not None: + speed_squared = velocity_body[0] ** 2 + velocity_body[1] ** 2 + if speed_squared > 1e-12: + for point in cells_body: + candidates.add( + min( + horizon_seconds, + max( + 0.0, + -( + point[0] * velocity_body[0] + + point[1] * velocity_body[1] + ) + / speed_squared, + ), + ) + ) + for coordinate, speed, lower, upper in ( + (point[0], velocity_body[0], bounds[0], bounds[1]), + (point[1], velocity_body[1], bounds[2], bounds[3]), + ): + if abs(speed) > 1e-12: + candidates.add( + min(horizon_seconds, max(0.0, (lower - coordinate) / speed)) + ) + candidates.add( + min(horizon_seconds, max(0.0, (upper - coordinate) / speed)) + ) + velocity = velocity_body or (0.0, 0.0) + clearance = min( + _point_box_distance( + (point[0] + velocity[0] * time_s, point[1] + velocity[1] * time_s), + bounds, + ) + for point in cells_body + for time_s in candidates + ) + return round(clearance, 12) + + +def _point_box_distance( + point: tuple[float, float], + bounds: tuple[float, float, float, float], +) -> float: + dx = max(bounds[0] - point[0], 0.0, point[0] - bounds[1]) + dy = max(bounds[2] - point[1], 0.0, point[1] - bounds[3]) + return math.hypot(dx, dy) + + +def _closing_speed_mps( + centroid_body: tuple[float, float, float], + velocity_body: tuple[float, float] | None, +) -> float | None: + if velocity_body is None: + return None + distance = math.hypot(centroid_body[0], centroid_body[1]) + if distance < 1e-9: + return round(math.hypot(*velocity_body), 12) + return round( + -( + centroid_body[0] * velocity_body[0] + + centroid_body[1] * velocity_body[1] + ) + / distance, + 12, + ) + + +def _inside( + point: tuple[float, float], + bounds: tuple[float, float, float, float], +) -> bool: + return bounds[0] <= point[0] <= bounds[1] and bounds[2] <= point[1] <= bounds[3] + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise ReplayThreatError(f"{label} must be an object") + return value + + +def _exact_keys(document: dict[str, object], expected: set[str], label: str) -> None: + if set(document) != expected: + raise ReplayThreatError(f"{label} fields are incompatible") + + +def _string(document: dict[str, object], key: str) -> str: + value = document.get(key) + if not isinstance(value, str) or not value: + raise ReplayThreatError(f"{key} must be a nonempty string") + return value + + +def _positive_number(document: dict[str, object], key: str) -> float: + value = _number(document, key) + if value <= 0.0: + raise ReplayThreatError(f"{key} must be positive") + return value + + +def _nonnegative_number(document: dict[str, object], key: str) -> float: + value = _number(document, key) + if value < 0.0: + raise ReplayThreatError(f"{key} must be nonnegative") + return value + + +def _number(document: dict[str, object], key: str) -> float: + value = document.get(key) + if not isinstance(value, int | float) or isinstance(value, bool) or not math.isfinite(value): + raise ReplayThreatError(f"{key} must be a finite number") + return float(value) + + +def _sha256(value: object, label: str) -> None: + if ( + not isinstance(value, str) + or len(value) != 64 + or any(character not in "0123456789abcdef" for character in value) + ): + raise ReplayThreatError(f"{label} is invalid") + + +__all__ = [ + "DEFAULT_REPLAY_THREAT_PROFILE_PATH", + "DualEvidenceReplayThreatProvider", + "REPLAY_THREAT_PROFILE_SCHEMA", + "REPLAY_THREAT_PROVIDER_ID", + "RecordedReplayPoseResolver", + "ReplayPose", + "ReplayPoseResolver", + "ReplayThreatError", + "ReplayThreatProfile", + "VirtualCorridorProfile", + "VirtualRigProfile", + "load_replay_threat_profile", +] diff --git a/src/k1link/perception/threat_replay.py b/src/k1link/perception/threat_replay.py new file mode 100644 index 0000000..2343a09 --- /dev/null +++ b/src/k1link/perception/threat_replay.py @@ -0,0 +1,1215 @@ +"""Immutable full-source M4.6 replay threat and visual evidence.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import shutil +import time +import uuid +from collections import Counter +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from threading import Event +from typing import Final + +import numpy as np + +from .contracts import ( + BoundingRegion2D, + GridCell, + HistorySample, + LocalObstacleMap, + MotionState, + ObjectProposal2D, + SourceAccounting, + TemporalObstacle, + TemporalState, + ThreatAssessment, + ThreatDecision, +) +from .detector_replay_contracts import DetectorReplayResult +from .detector_replay_result import read_detector_replay_result +from .geometry import RecordedGeometryStore +from .geometry_math import quaternion_xyzw_to_rotation_matrix +from .geometry_replay import GeometryReplayResult, read_geometry_replay_result +from .providers import SourcePacket +from .recorded_source import RecordedRavnoves00Source, ReplayPacing +from .temporal_replay import TemporalReplayResult, read_temporal_replay_result +from .threat import ( + DEFAULT_REPLAY_THREAT_PROFILE_PATH, + DualEvidenceReplayThreatProvider, + RecordedReplayPoseResolver, + ReplayPose, + ReplayThreatProfile, + load_replay_threat_profile, +) + +THREAT_REPLAY_SCHEMA: Final = "missioncore.perception-threat-replay-result/v1" +THREAT_REPLAY_FRAME_SCHEMA: Final = "missioncore.perception-threat-replay-frame/v1" +THREAT_REPLAY_VISUAL_SCHEMA: Final = "missioncore.perception-threat-visual-frame/v1" +THREAT_REPLAY_FIXTURE_SCHEMA: Final = "missioncore.perception-threat-fixtures/v1" +THREAT_REPLAY_REPORT_SCHEMA: Final = "missioncore.perception-threat-replay-report/v1" +THREAT_REPLAY_RESULT_PREFIX: Final = "m4-threat-replay-" +THREAT_REPLAY_FRAMES_NAME: Final = "frames.jsonl" +THREAT_REPLAY_VISUALS_NAME: Final = "visual-frames.jsonl" +THREAT_REPLAY_FIXTURES_NAME: Final = "fixtures.json" +THREAT_REPLAY_REPORT_NAME: Final = "report.json" +THREAT_REPLAY_MANIFEST_NAME: Final = "manifest.json" +VISUAL_FRAME_COUNT: Final = 32 +VISUAL_POINT_LIMIT: Final = 4_000 + + +class ThreatReplayError(RuntimeError): + """The M4.6 replay is incomplete, mutable or source-inconsistent.""" + + +@dataclass(frozen=True, slots=True) +class ThreatReplayResult: + result_id: str + result_root: Path + accepted: bool + metrics: dict[str, object] + report: dict[str, object] + manifest: dict[str, object] + + +def build_threat_replay( + *, + repository_root: Path, + temporal_result_root: Path, + geometry_result_root: Path, + detector_result_root: Path, + output_root: Path, +) -> ThreatReplayResult: + repository = repository_root.resolve() + profile = load_replay_threat_profile( + repository / DEFAULT_REPLAY_THREAT_PROFILE_PATH + ) + temporal = read_temporal_replay_result(temporal_result_root) + geometry = read_geometry_replay_result(geometry_result_root) + detector = read_detector_replay_result(detector_result_root) + _validate_upstream(profile, temporal, geometry, detector) + + store = RecordedGeometryStore.from_repository(repository) + pose_resolver = RecordedReplayPoseResolver(store) + provider = DualEvidenceReplayThreatProvider( + pose_resolver=pose_resolver, + profile=profile, + ) + source = RecordedRavnoves00Source.from_repository( + repository, + pacing=ReplayPacing.UNCAPPED, + ) + visual_sequences = _visual_sequences(store.available_frame_indices()) + + root = output_root.expanduser().absolute() + root.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = root / f".threat-replay.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + started_ns = time.perf_counter_ns() + frame_count = 0 + failed_frames = 0 + assessment_counts: Counter[str] = Counter() + evidence_counts: Counter[str] = Counter() + motion_decisions: Counter[str] = Counter() + reason_counts: Counter[str] = Counter() + latencies_ms: list[float] = [] + visual_count = 0 + try: + temporal_frames_path = temporal.result_root / "frames.jsonl" + geometry_frames_path = geometry.result_root / "frames.jsonl" + frames_path = staging / THREAT_REPLAY_FRAMES_NAME + visuals_path = staging / THREAT_REPLAY_VISUALS_NAME + with ( + temporal_frames_path.open("rb") as temporal_stream, + geometry_frames_path.open("rb") as geometry_stream, + frames_path.open("wb") as output, + visuals_path.open("wb") as visual_output, + ): + packets = source.packets(Event()) + if len(detector.frames) != 4489: + raise ThreatReplayError("detector replay frame count changed") + for detector_frame, packet in zip(detector.frames, packets, strict=True): + temporal_frame = _read_json_line( + temporal_stream.readline(), "temporal frame", frame_count + ) + geometry_frame = _read_json_line( + geometry_stream.readline(), "geometry frame", frame_count + ) + _validate_frame_binding( + frame_count, + packet.envelope.frame_id, + detector_frame.sequence, + detector_frame.envelope.frame_id, + temporal_frame, + geometry_frame, + ) + current = tuple( + TemporalObstacle.from_dict(value) + for value in _array(temporal_frame.get("current"), "current obstacles") + ) + unknown = tuple( + TemporalObstacle.from_dict(value) + for key in ("held", "expired") + for value in _array(temporal_frame.get(key), f"{key} obstacles") + ) + geometry_observations = _array( + geometry_frame.get("observations"), "geometry observations" + ) + associated_proposals = { + proposal_id + for raw in geometry_observations + if isinstance(raw, dict) and raw.get("occupied_support") is True + for proposal_id in _string_array( + raw.get("proposal_ids"), "geometry proposal ids" + ) + } + proposals = detector_frame.proposals + camera_uncertainty = tuple( + proposal + for proposal in proposals + if proposal.proposal_id not in associated_proposals + ) + obstacle_map = LocalObstacleMap( + source_id=packet.envelope.source_id, + session_id=packet.envelope.session_id, + frame_id=packet.envelope.frame_id, + graph_id="reference-perception-graph/v1", + generated_monotonic_ns=0, + output_age_ns=0, + occupied=current, + unknown=unknown, + camera_uncertainty=camera_uncertainty, + accounting=SourceAccounting(1, 1, 0, 0), + ) + frame_started_ns = time.perf_counter_ns() + assessments = provider.assess(obstacle_map) + latencies_ms.append( + (time.perf_counter_ns() - frame_started_ns) / 1_000_000 + ) + by_id = {item.component_id: item for item in assessments} + expected_ids = { + item.component_id for item in (*current, *unknown) + } | {item.proposal_id for item in camera_uncertainty} + if set(by_id) != expected_ids: + raise ThreatReplayError("threat assessment coverage is incomplete") + camera_rows = _camera_rows( + proposals, + geometry_observations, + by_id, + ) + metric_rows = [ + _metric_row(item, by_id[item.component_id]) + for item in (*current, *unknown) + ] + for item in assessments: + assessment_counts[item.decision.value] += 1 + reason_counts.update(item.reason_codes) + evidence_counts["current-metric"] += len(current) + evidence_counts["stale-or-held"] += len(unknown) + evidence_counts["camera-only"] += len(camera_uncertainty) + for obstacle in current: + motion_decisions[ + f"{obstacle.motion.value}:{by_id[obstacle.component_id].decision.value}" + ] += 1 + frame_document = { + "schema_version": THREAT_REPLAY_FRAME_SCHEMA, + "sequence": frame_count, + "frame_id": packet.envelope.frame_id, + "source_time_ns": packet.envelope.timestamps.source_ns, + "source_available": ( + packet.envelope.registered_point_increment.available + ), + "pose_available": pose_resolver.pose_for_frame( + packet.envelope.frame_id + ) + is not None, + "metric_obstacles": metric_rows, + "camera_proposals": camera_rows, + "assessments": [item.to_dict() for item in assessments], + "accounting": { + "metric_obstacles": len(metric_rows), + "camera_proposals": len(proposals), + "camera_only": len(camera_uncertainty), + "assessments": len(assessments), + }, + "authority": _false_authority(), + } + output.write(_canonical_json(frame_document) + b"\n") + if frame_count in visual_sequences: + visual_output.write( + _canonical_json( + _visual_frame( + packet=packet, + store=store, + pose=pose_resolver.pose_for_frame(packet.envelope.frame_id), + metric_rows=metric_rows, + camera_rows=camera_rows, + profile=profile, + ) + ) + + b"\n" + ) + visual_count += 1 + frame_count += 1 + if temporal_stream.readline() or geometry_stream.readline(): + raise ThreatReplayError("upstream frame ledger exceeds recorded source") + if frame_count != 4489 or visual_count != VISUAL_FRAME_COUNT: + raise ThreatReplayError("full replay or visual sample coverage is incomplete") + + fixtures = _fixture_document(profile) + fixtures_path = staging / THREAT_REPLAY_FIXTURES_NAME + _write_json(fixtures_path, fixtures) + elapsed_ns = time.perf_counter_ns() - started_ns + metrics = _metrics( + frame_count=frame_count, + failed_frames=failed_frames, + assessment_counts=assessment_counts, + evidence_counts=evidence_counts, + motion_decisions=motion_decisions, + reason_counts=reason_counts, + latencies_ms=latencies_ms, + elapsed_ns=elapsed_ns, + visual_count=visual_count, + fixtures=fixtures, + ) + requirements = _requirements(metrics, fixtures) + accepted = all(value is True for value in requirements.values()) + frames_sha256 = _file_sha256(frames_path) + visuals_sha256 = _file_sha256(visuals_path) + fixtures_sha256 = _file_sha256(fixtures_path) + identity = { + "schema_version": THREAT_REPLAY_SCHEMA, + "profile_id": profile.profile_id, + "profile_sha256": profile.profile_sha256, + "provider_id": provider.provider_id, + "source_id": profile.source_id, + "source_session_id": profile.session_id, + "temporal_result_id": temporal.result_id, + "temporal_frames_sha256": profile.temporal_frames_sha256, + "geometry_result_id": geometry.result_id, + "geometry_frames_sha256": profile.geometry_frames_sha256, + "detector_result_id": detector.result_id, + "detector_frames_sha256": profile.detector_frames_sha256, + "source_pack_id": profile.source_pack_id, + "source_pack_sha256": profile.source_pack_sha256, + "calibration_id": profile.calibration_id, + "calibration_content_sha256": profile.calibration_content_sha256, + "rig_profile_id": profile.rig.profile_id, + "corridor_profile_id": profile.corridor.profile_id, + "producer_sha256": _producer_hashes(repository), + "frames_sha256": frames_sha256, + "visuals_sha256": visuals_sha256, + "fixtures_sha256": fixtures_sha256, + "metrics": metrics, + "acceptance_requirements": requirements, + "accepted": accepted, + "authority": _false_authority(), + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"{THREAT_REPLAY_RESULT_PREFIX}{identity_sha256}" + report = { + "schema_version": THREAT_REPLAY_REPORT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "status": "accepted" if accepted else "rejected", + "metrics": metrics, + "acceptance_requirements": requirements, + "configuration": { + "virtual_body_m": [ + profile.rig.body_length_m, + profile.rig.body_width_m, + ], + "nominal_sensor_height_m": profile.rig.nominal_sensor_height_m, + "forward_corridor_m": profile.corridor.forward_length_m, + "prediction_horizon_seconds": ( + profile.corridor.prediction_horizon_seconds + ), + }, + "limitations": [ + "The body and corridor are replay-simulated, not a measured physical mount.", + "The LiDAR archive is the vendor mapped point increment, not every raw beam.", + "TTC uses bounded constant-relative-velocity replay extrapolation.", + "Camera-only evidence remains unknown and cannot establish metric clearance.", + "M4.8 independent object-centric labels remain the correctness gate.", + ], + "authority": _false_authority(), + } + report_path = staging / THREAT_REPLAY_REPORT_NAME + _write_json(report_path, report) + manifest = { + "schema_version": THREAT_REPLAY_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": datetime.now(UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z"), + "accepted": accepted, + "artifacts": [ + _artifact(frames_path, "threat-replay-frames"), + _artifact(visuals_path, "threat-visual-frames"), + _artifact(fixtures_path, "threat-deterministic-fixtures"), + _artifact(report_path, "threat-replay-report"), + ], + } + _write_json(staging / THREAT_REPLAY_MANIFEST_NAME, manifest) + destination = root / result_id + if destination.exists(): + shutil.rmtree(staging) + return read_threat_replay_result(destination) + os.replace(staging, destination) + return read_threat_replay_result(destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + + +def read_threat_replay_result(root: Path) -> ThreatReplayResult: + resolved = root.resolve(strict=True) + if resolved.is_symlink() or not resolved.name.startswith(THREAT_REPLAY_RESULT_PREFIX): + raise ThreatReplayError("threat replay result root is invalid") + manifest = _read_json(resolved / THREAT_REPLAY_MANIFEST_NAME) + _exact_keys( + manifest, + { + "schema_version", + "result_id", + "identity_sha256", + "identity", + "created_at_utc", + "accepted", + "artifacts", + }, + "threat replay manifest", + ) + identity = _object(manifest.get("identity"), "threat replay identity") + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + if ( + manifest.get("schema_version") != THREAT_REPLAY_SCHEMA + or manifest.get("result_id") != resolved.name + or manifest.get("identity_sha256") != identity_sha256 + or resolved.name != f"{THREAT_REPLAY_RESULT_PREFIX}{identity_sha256}" + ): + raise ThreatReplayError("threat replay identity changed") + artifacts = _array(manifest.get("artifacts"), "threat artifacts") + by_role = { + _object(item, "threat artifact").get("role"): item for item in artifacts + } + expected = { + "threat-replay-frames": (THREAT_REPLAY_FRAMES_NAME, "frames_sha256"), + "threat-visual-frames": (THREAT_REPLAY_VISUALS_NAME, "visuals_sha256"), + "threat-deterministic-fixtures": ( + THREAT_REPLAY_FIXTURES_NAME, + "fixtures_sha256", + ), + "threat-replay-report": (THREAT_REPLAY_REPORT_NAME, None), + } + if set(by_role) != set(expected): + raise ThreatReplayError("threat artifact inventory changed") + paths: dict[str, Path] = {} + for role, (name, identity_key) in expected.items(): + path = _validated_artifact(resolved, by_role[role], name) + paths[role] = path + if identity_key is not None and _file_sha256(path) != identity.get(identity_key): + raise ThreatReplayError("threat artifact identity changed") + report = _read_json(paths["threat-replay-report"]) + metrics = _object(identity.get("metrics"), "threat metrics") + requirements = _object( + identity.get("acceptance_requirements"), "threat requirements" + ) + fixtures = _read_json(paths["threat-deterministic-fixtures"]) + accepted = all(value is True for value in requirements.values()) + if ( + report.get("schema_version") != THREAT_REPLAY_REPORT_SCHEMA + or report.get("result_id") != resolved.name + or report.get("identity_sha256") != identity_sha256 + or report.get("metrics") != metrics + or report.get("acceptance_requirements") != requirements + or report.get("authority") != _false_authority() + or identity.get("authority") != _false_authority() + or manifest.get("accepted") is not accepted + or identity.get("accepted") is not accepted + or requirements != _requirements(metrics, fixtures) + ): + raise ThreatReplayError("threat replay report or acceptance changed") + _validate_ledgers( + paths["threat-replay-frames"], + paths["threat-visual-frames"], + metrics, + ) + return ThreatReplayResult( + result_id=resolved.name, + result_root=resolved, + accepted=accepted, + metrics=metrics, + report=report, + manifest=manifest, + ) + + +def _validate_upstream( + profile: ReplayThreatProfile, + temporal: TemporalReplayResult, + geometry: GeometryReplayResult, + detector: DetectorReplayResult, +) -> None: + if not temporal.accepted or not geometry.accepted or not detector.accepted: + raise ThreatReplayError("an upstream M4 result is not accepted") + detector_identity = _object(detector.manifest.get("identity"), "detector identity") + geometry_identity = _object(geometry.manifest.get("identity"), "geometry identity") + temporal_identity = _object(temporal.manifest.get("identity"), "temporal identity") + if ( + temporal.result_id != profile.temporal_result_id + or temporal_identity.get("frames_sha256") != profile.temporal_frames_sha256 + or geometry.result_id != profile.geometry_result_id + or geometry_identity.get("frames_sha256") != profile.geometry_frames_sha256 + or detector.result_id != profile.detector_result_id + or detector_identity.get("frames_sha256") != profile.detector_frames_sha256 + ): + raise ThreatReplayError("upstream M4 evidence escaped the threat profile") + + +def _validate_frame_binding( + sequence: int, + frame_id: str, + detector_sequence: int, + detector_frame_id: str, + temporal_frame: dict[str, object], + geometry_frame: dict[str, object], +) -> None: + if ( + detector_sequence != sequence + or detector_frame_id != frame_id + or temporal_frame.get("sequence") != sequence + or temporal_frame.get("frame_id") != frame_id + or geometry_frame.get("sequence") != sequence + or geometry_frame.get("frame_id") != frame_id + ): + raise ThreatReplayError("M4 frame ledgers are not source-aligned") + + +def _metric_row( + obstacle: TemporalObstacle, + assessment: ThreatAssessment, +) -> dict[str, object]: + return { + "component_id": obstacle.component_id, + "state": obstacle.state.value, + "motion": obstacle.motion.value, + "motion_reason": obstacle.motion_reason, + "semantic_hint": obstacle.semantic_hint, + "centroid_map_xyz_m": ( + None + if obstacle.last_centroid_xyz_m is None + else list(obstacle.last_centroid_xyz_m) + ), + "cells": [item.to_dict() for item in obstacle.cells], + "history": [item.to_dict() for item in obstacle.history], + "assessment": assessment.to_dict(), + } + + +def _camera_rows( + proposals: tuple[ObjectProposal2D, ...], + observations: list[object], + assessments: dict[str, ThreatAssessment], +) -> list[dict[str, object]]: + support: dict[str, dict[str, object]] = {} + for raw in observations: + if not isinstance(raw, dict): + raise ThreatReplayError("geometry observation is not an object") + for proposal_id in _string_array(raw.get("proposal_ids"), "proposal ids"): + metric = raw.get("metric_geometry") + range_m = metric.get("range_m") if isinstance(metric, dict) else None + support[proposal_id] = { + "occupied_support": raw.get("occupied_support") is True, + "range_m": range_m, + "reason_codes": _string_array(raw.get("reason_codes"), "reason codes"), + } + rows = [] + for proposal in proposals: + geometry = support.get( + proposal.proposal_id, + { + "occupied_support": False, + "range_m": None, + "reason_codes": ["proposal-without-metric-observation"], + }, + ) + assessment = assessments.get(proposal.proposal_id) + rows.append( + { + "proposal_id": proposal.proposal_id, + "bbox_xyxy": list(proposal.region.as_tuple()), + "objectness": proposal.objectness, + "semantic_hint": proposal.semantic_hint, + "occupied_support": geometry["occupied_support"], + "range_m": geometry["range_m"], + "geometry_reason_codes": geometry["reason_codes"], + "threat_decision": ( + None if assessment is None else assessment.decision.value + ), + "threat_reason_codes": ( + [] if assessment is None else list(assessment.reason_codes) + ), + } + ) + return rows + + +def _visual_frame( + *, + packet: SourcePacket, + store: RecordedGeometryStore, + pose: ReplayPose | None, + metric_rows: list[dict[str, object]], + camera_rows: list[dict[str, object]], + profile: ReplayThreatProfile, +) -> dict[str, object]: + if pose is None: + raise ThreatReplayError("visual frame has no source pose") + points = store.current_points(packet) + if points is None: + raise ThreatReplayError("visual frame has no current point cloud") + rotation = quaternion_xyzw_to_rotation_matrix( + pose.orientation_map_from_lidar_xyzw + ) + position = np.asarray(pose.position_map_xyz_m, dtype=np.float64) + points_body = (points - position) @ rotation + stride = max(1, math.ceil(points_body.shape[0] / VISUAL_POINT_LIMIT)) + sampled = points_body[::stride][:VISUAL_POINT_LIMIT] + metric_visuals = [] + for row in metric_rows: + centroid = row.get("centroid_map_xyz_m") + cells = row.get("cells") + if not isinstance(centroid, list) or not isinstance(cells, list): + continue + centroid_body = pose.map_point_to_body( + (float(centroid[0]), float(centroid[1]), float(centroid[2])) + ) + cell_centers = [] + for raw_cell in cells: + cell = _object(raw_cell, "visual occupied cell") + point_map = tuple( + (_signed_integer(cell.get(key), f"cell {key}") + 0.5) + * profile.corridor.occupied_voxel_size_m + for key in ("x", "y", "z") + ) + cell_centers.append( + list( + pose.map_point_to_body( + (point_map[0], point_map[1], point_map[2]) + ) + ) + ) + metric_visuals.append( + { + "component_id": row["component_id"], + "state": row["state"], + "motion": row["motion"], + "centroid_body_xyz_m": list(centroid_body), + "cell_centers_body_xyz_m": cell_centers, + "assessment": row["assessment"], + } + ) + return { + "schema_version": THREAT_REPLAY_VISUAL_SCHEMA, + "sequence": packet.envelope.sequence, + "frame_id": packet.envelope.frame_id, + "source_time_ns": packet.envelope.timestamps.source_ns, + "point_cloud_body_xyz_m": np.round(sampled, 6).tolist(), + "point_cloud_source_count": int(points.shape[0]), + "point_cloud_sample_count": int(sampled.shape[0]), + "metric_obstacles": metric_visuals, + "camera_proposals": camera_rows, + "rig": { + "length_m": profile.rig.body_length_m, + "width_m": profile.rig.body_width_m, + "nominal_sensor_height_m": profile.rig.nominal_sensor_height_m, + }, + "corridor": { + "forward_length_m": profile.corridor.forward_length_m, + "rear_margin_m": profile.corridor.rear_margin_m, + "half_width_m": ( + profile.rig.body_width_m / 2 + + profile.corridor.lateral_clearance_m + ), + "prediction_horizon_seconds": ( + profile.corridor.prediction_horizon_seconds + ), + }, + "authority": _false_authority(), + } + + +class _FixturePoses: + def pose_for_frame(self, frame_id: str) -> ReplayPose: + return ReplayPose( + frame_id=frame_id, + position_map_xyz_m=(0.0, 0.0, 0.0), + orientation_map_from_lidar_xyzw=(0.0, 0.0, 0.0, 1.0), + ) + + +def _fixture_document(profile: ReplayThreatProfile) -> dict[str, object]: + provider = DualEvidenceReplayThreatProvider( + pose_resolver=_FixturePoses(), + profile=profile, + ) + frame_id = "frame-000002" + cases = [ + _fixture_case( + provider, + "static-in-corridor", + _fixture_obstacle( + "fixture-static-in", + GridCell(6, 0, 0), + MotionState.STATIONARY, + ( + ("frame-000000", 0, (2.925, 0.225, 0.225)), + (frame_id, 300_000_000, (2.925, 0.225, 0.225)), + ), + ), + ThreatDecision.THREAT, + critical=True, + ), + _fixture_case( + provider, + "static-outside", + _fixture_obstacle( + "fixture-static-out", + GridCell(6, 7, 0), + MotionState.STATIONARY, + ( + ("frame-000000", 0, (2.925, 3.375, 0.225)), + (frame_id, 300_000_000, (2.925, 3.375, 0.225)), + ), + ), + ThreatDecision.NOT_THREAT, + ), + _fixture_case( + provider, + "crossing", + _fixture_obstacle( + "fixture-crossing", + GridCell(6, 3, 0), + MotionState.MOVING, + ( + ("frame-000000", 0, (2.925, 2.575, 0.225)), + (frame_id, 300_000_000, (2.925, 1.575, 0.225)), + ), + ), + ThreatDecision.THREAT, + critical=True, + ), + _fixture_case( + provider, + "approaching", + _fixture_obstacle( + "fixture-approaching", + GridCell(9, 0, 0), + MotionState.MOVING, + ( + ("frame-000000", 0, (6.275, 0.225, 0.225)), + (frame_id, 300_000_000, (4.275, 0.225, 0.225)), + ), + ), + ThreatDecision.THREAT, + critical=True, + ), + _fixture_case( + provider, + "receding", + _fixture_obstacle( + "fixture-receding", + GridCell(-5, 0, 0), + MotionState.MOVING, + ( + ("frame-000000", 0, (-1.025, 0.225, 0.225)), + (frame_id, 300_000_000, (-2.025, 0.225, 0.225)), + ), + ), + ThreatDecision.NOT_THREAT, + ), + _fixture_case( + provider, + "occluded-held", + _fixture_obstacle( + "fixture-held", + GridCell(6, 0, 0), + MotionState.UNKNOWN, + ((frame_id, 300_000_000, (2.925, 0.225, 0.225)),), + state=TemporalState.HELD, + ), + ThreatDecision.UNKNOWN, + ), + _fixture_case( + provider, + "stale-expired", + _fixture_obstacle( + "fixture-expired", + GridCell(6, 0, 0), + MotionState.UNKNOWN, + ((frame_id, 300_000_000, (2.925, 0.225, 0.225)),), + state=TemporalState.EXPIRED, + ), + ThreatDecision.UNKNOWN, + ), + _fixture_camera_case(provider, frame_id), + _fixture_case( + provider, + "geometry-only", + _fixture_obstacle( + "fixture-geometry-only", + GridCell(4, 0, 0), + MotionState.STATIONARY, + ( + ("frame-000000", 0, (2.025, 0.225, 0.225)), + (frame_id, 300_000_000, (2.025, 0.225, 0.225)), + ), + ), + ThreatDecision.THREAT, + critical=True, + ), + ] + return { + "schema_version": THREAT_REPLAY_FIXTURE_SCHEMA, + "cases": cases, + "critical_case_count": sum(item["critical"] is True for item in cases), + "critical_false_not_threat_count": sum( + item["critical"] is True and item["actual"] == "not-threat" + for item in cases + ), + "passed_count": sum(item["passed"] is True for item in cases), + "total_count": len(cases), + "authority": _false_authority(), + } + + +def _fixture_obstacle( + component_id: str, + cell: GridCell, + motion: MotionState, + history: tuple[tuple[str, int, tuple[float, float, float]], ...], + *, + state: TemporalState = TemporalState.CURRENT, +) -> TemporalObstacle: + samples = tuple(HistorySample(*item) for item in history) + last = samples[-1] + return TemporalObstacle( + component_id=component_id, + identity_scope="ephemeral", + state=state, + ttl_ns=750_000_000, + last_hit_ns=last.evidence_time_ns, + age_ns=0 if state is TemporalState.CURRENT else 100_000_000, + association_basis="deterministic-fixture", + history=samples, + cells=() if state is TemporalState.EXPIRED else (cell,), + coordinate_frame=None if state is TemporalState.EXPIRED else "map", + last_centroid_xyz_m=None if state is TemporalState.EXPIRED else last.centroid_xyz_m, + motion=motion if state is TemporalState.CURRENT else MotionState.UNKNOWN, + motion_confidence=( + 0.0 + if state is not TemporalState.CURRENT or motion is MotionState.UNKNOWN + else 1.0 + ), + motion_reason=( + "stale-support" + if state is not TemporalState.CURRENT + else "bounded-map-history-moving" + if motion is MotionState.MOVING + else "bounded-map-history-stationary" + if motion is MotionState.STATIONARY + else "insufficient-history" + ), + ) + + +def _fixture_case( + provider: DualEvidenceReplayThreatProvider, + name: str, + obstacle: TemporalObstacle, + expected: ThreatDecision, + *, + critical: bool = False, +) -> dict[str, object]: + obstacle_map = LocalObstacleMap( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id="frame-000002", + graph_id="reference-perception-graph/v1", + generated_monotonic_ns=0, + output_age_ns=0, + occupied=(obstacle,) if obstacle.state is TemporalState.CURRENT else (), + unknown=(obstacle,) if obstacle.state is not TemporalState.CURRENT else (), + camera_uncertainty=(), + accounting=SourceAccounting(1, 1, 0, 0), + ) + assessment = provider.assess(obstacle_map)[0] + return { + "name": name, + "expected": expected.value, + "actual": assessment.decision.value, + "critical": critical, + "passed": assessment.decision is expected, + "assessment": assessment.to_dict(), + } + + +def _fixture_camera_case( + provider: DualEvidenceReplayThreatProvider, + frame_id: str, +) -> dict[str, object]: + proposal = ObjectProposal2D( + proposal_id="fixture-camera-only", + source_id="RAVNOVES00", + frame_id=frame_id, + region=BoundingRegion2D(10.0, 10.0, 20.0, 20.0), + objectness=0.9, + provider_id="fixture-detector/v1", + model_id="fixture-model/v1", + preprocess_id="fixture-preprocess/v1", + ) + obstacle_map = LocalObstacleMap( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id=frame_id, + graph_id="reference-perception-graph/v1", + generated_monotonic_ns=0, + output_age_ns=0, + occupied=(), + unknown=(), + camera_uncertainty=(proposal,), + accounting=SourceAccounting(1, 1, 0, 0), + ) + assessment = provider.assess(obstacle_map)[0] + return { + "name": "camera-only", + "expected": "unknown", + "actual": assessment.decision.value, + "critical": False, + "passed": assessment.decision is ThreatDecision.UNKNOWN, + "assessment": assessment.to_dict(), + } + + +def _metrics( + *, + frame_count: int, + failed_frames: int, + assessment_counts: Counter[str], + evidence_counts: Counter[str], + motion_decisions: Counter[str], + reason_counts: Counter[str], + latencies_ms: list[float], + elapsed_ns: int, + visual_count: int, + fixtures: dict[str, object], +) -> dict[str, object]: + values = np.asarray(latencies_ms, dtype=np.float64) + return { + "frames": {"total": frame_count, "failed": failed_frames}, + "evidence": dict(sorted(evidence_counts.items())), + "decisions": dict(sorted(assessment_counts.items())), + "motion_decisions": dict(sorted(motion_decisions.items())), + "reason_counts": dict(sorted(reason_counts.items())), + "visual_evidence": { + "frame_count": visual_count, + "point_limit_per_frame": VISUAL_POINT_LIMIT, + "video_overlay_available": True, + "camera_boxes_available": True, + "point_cloud_available": True, + "metric_distance_available": True, + "virtual_corridor_available": True, + }, + "fixtures": { + "passed": fixtures["passed_count"], + "total": fixtures["total_count"], + "critical": fixtures["critical_case_count"], + "critical_false_not_threat": fixtures[ + "critical_false_not_threat_count" + ], + }, + "runtime": { + "elapsed_ns": elapsed_ns, + "frames_per_second": round(frame_count / (elapsed_ns / 1_000_000_000), 6), + "provider_latency_p50_ms": round(float(np.percentile(values, 50)), 6), + "provider_latency_p95_ms": round(float(np.percentile(values, 95)), 6), + "provider_latency_max_ms": round(float(np.max(values)), 6), + }, + } + + +def _requirements( + metrics: dict[str, object], + fixtures: dict[str, object], +) -> dict[str, bool]: + frames = _object(metrics.get("frames"), "frame metrics") + evidence = _object(metrics.get("evidence"), "evidence metrics") + decisions = _object(metrics.get("decisions"), "decision metrics") + visual = _object(metrics.get("visual_evidence"), "visual metrics") + total_evidence = sum( + _integer(value, "evidence count") for value in evidence.values() + ) + total_decisions = sum( + _integer(value, "decision count") for value in decisions.values() + ) + cases = _array(fixtures.get("cases"), "fixture cases") + camera_case = next( + ( + _object(item, "fixture") + for item in cases + if isinstance(item, dict) and item.get("name") == "camera-only" + ), + {}, + ) + stale_cases = [ + _object(item, "fixture") + for item in cases + if isinstance(item, dict) and item.get("name") in {"occluded-held", "stale-expired"} + ] + return { + "full_ravnoves00_replay_completed": ( + frames.get("total") == 4489 and frames.get("failed") == 0 + ), + "every_metric_or_camera_evidence_received_one_assessment": ( + total_evidence == total_decisions and total_evidence > 0 + ), + "camera_only_is_unknown_never_safe": camera_case.get("actual") == "unknown", + "held_and_stale_are_unknown_never_safe": ( + len(stale_cases) == 2 + and all(item.get("actual") == "unknown" for item in stale_cases) + ), + "geometry_only_evidence_is_assessed": ( + _integer( + _object(metrics.get("reason_counts"), "reason metrics").get( + "geometry-only-evidence", 0 + ), + "geometry-only count", + ) + > 0 + ), + "deterministic_fixture_matrix_passed": ( + fixtures.get("passed_count") == fixtures.get("total_count") == 9 + ), + "zero_critical_fixture_false_not_threat": ( + fixtures.get("critical_false_not_threat_count") == 0 + ), + "visual_video_camera_cloud_distance_and_corridor_are_available": ( + visual.get("frame_count") == VISUAL_FRAME_COUNT + and all( + visual.get(key) is True + for key in ( + "video_overlay_available", + "camera_boxes_available", + "point_cloud_available", + "metric_distance_available", + "virtual_corridor_available", + ) + ) + ), + "physical_collision_and_actuation_authority_remain_false": ( + fixtures.get("authority") == _false_authority() + ), + } + + +def _validate_ledgers( + frames_path: Path, + visuals_path: Path, + metrics: dict[str, object], +) -> None: + frame_count = 0 + assessment_count = 0 + for sequence, frame in enumerate(_read_jsonl(frames_path)): + if ( + frame.get("schema_version") != THREAT_REPLAY_FRAME_SCHEMA + or frame.get("sequence") != sequence + or frame.get("authority") != _false_authority() + ): + raise ThreatReplayError("threat frame ledger changed") + accounting = _object(frame.get("accounting"), "threat frame accounting") + assessments = _array(frame.get("assessments"), "threat assessments") + if accounting.get("assessments") != len(assessments): + raise ThreatReplayError("threat frame accounting changed") + assessment_count += len(assessments) + frame_count += 1 + visuals = list(_read_jsonl(visuals_path)) + if ( + frame_count != _object(metrics.get("frames"), "frames").get("total") + or assessment_count + != sum( + _integer(value, "decision count") + for value in _object(metrics.get("decisions"), "decisions").values() + ) + or len(visuals) != VISUAL_FRAME_COUNT + or any(item.get("schema_version") != THREAT_REPLAY_VISUAL_SCHEMA for item in visuals) + ): + raise ThreatReplayError("threat replay ledger and metrics disagree") + + +def _visual_sequences(available: tuple[int, ...]) -> frozenset[int]: + if len(available) < VISUAL_FRAME_COUNT: + raise ThreatReplayError("not enough available source frames for visual evidence") + selected = { + available[round(index * (len(available) - 1) / (VISUAL_FRAME_COUNT - 1))] + for index in range(VISUAL_FRAME_COUNT) + } + if len(selected) != VISUAL_FRAME_COUNT: + raise ThreatReplayError("visual sample selection is not unique") + return frozenset(selected) + + +def _producer_hashes(repository: Path) -> dict[str, str]: + return { + name: _file_sha256(repository / "src/k1link/perception" / name) + for name in ("threat.py", "threat_replay.py") + } + + +def _false_authority() -> dict[str, object]: + return { + "mode": "replay-simulated", + "ground_truth": False, + "physical_live": False, + "physical_collision_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, + } + + +def _artifact(path: Path, role: str) -> dict[str, object]: + return { + "role": role, + "path": path.name, + "byte_length": path.stat().st_size, + "sha256": _file_sha256(path), + } + + +def _validated_artifact(root: Path, raw: object, expected_name: str) -> Path: + item = _object(raw, "threat artifact") + _exact_keys( + item, + {"role", "path", "byte_length", "sha256"}, + "threat artifact", + ) + if item.get("path") != expected_name: + raise ThreatReplayError("threat artifact path changed") + path = (root / expected_name).resolve(strict=True) + if ( + path.parent != root + or path.is_symlink() + or not path.is_file() + or path.stat().st_size != item.get("byte_length") + or _file_sha256(path) != item.get("sha256") + ): + raise ThreatReplayError("threat artifact content changed") + return path + + +def _read_json_line(raw: bytes, label: str, sequence: int) -> dict[str, object]: + if not raw: + raise ThreatReplayError(f"{label} ended before frame {sequence}") + try: + return _object(json.loads(raw), label) + except json.JSONDecodeError as exc: + raise ThreatReplayError(f"{label} is invalid JSON") from exc + + +def _read_json(path: Path) -> dict[str, object]: + try: + return _object(json.loads(path.read_bytes()), path.name) + except json.JSONDecodeError as exc: + raise ThreatReplayError(f"{path.name} is invalid JSON") from exc + + +def _read_jsonl(path: Path) -> Iterator[dict[str, object]]: + with path.open("rb") as handle: + for line in handle: + if line.strip(): + try: + yield _object(json.loads(line), path.name) + except json.JSONDecodeError as exc: + raise ThreatReplayError(f"{path.name} is invalid JSONL") from exc + + +def _write_json(path: Path, value: object) -> None: + path.write_bytes(_canonical_json(value) + b"\n") + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode() + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise ThreatReplayError(f"{label} must be an object") + return value + + +def _array(value: object, label: str) -> list[object]: + if not isinstance(value, list): + raise ThreatReplayError(f"{label} must be an array") + return value + + +def _string_array(value: object, label: str) -> list[str]: + values = _array(value, label) + if any(not isinstance(item, str) for item in values): + raise ThreatReplayError(f"{label} must contain strings") + return [str(item) for item in values] + + +def _integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ThreatReplayError(f"{label} must be a nonnegative integer") + return value + + +def _signed_integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool): + raise ThreatReplayError(f"{label} must be an integer") + return value + + +def _exact_keys(document: dict[str, object], expected: set[str], label: str) -> None: + if set(document) != expected: + raise ThreatReplayError(f"{label} fields are incompatible") + + +__all__ = [ + "THREAT_REPLAY_FIXTURE_SCHEMA", + "THREAT_REPLAY_FRAME_SCHEMA", + "THREAT_REPLAY_MANIFEST_NAME", + "THREAT_REPLAY_REPORT_SCHEMA", + "THREAT_REPLAY_RESULT_PREFIX", + "THREAT_REPLAY_SCHEMA", + "THREAT_REPLAY_VISUAL_SCHEMA", + "ThreatReplayError", + "ThreatReplayResult", + "build_threat_replay", + "read_threat_replay_result", +] diff --git a/src/k1link/perception/threat_replay_cli.py b/src/k1link/perception/threat_replay_cli.py new file mode 100644 index 0000000..cc1a1b6 --- /dev/null +++ b/src/k1link/perception/threat_replay_cli.py @@ -0,0 +1,43 @@ +"""Command-line entrypoint for the local M4.6 replay threat run.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from .threat_replay import build_threat_replay + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repository-root", type=Path, required=True) + parser.add_argument("--temporal-result-root", type=Path, required=True) + parser.add_argument("--geometry-result-root", type=Path, required=True) + parser.add_argument("--detector-result-root", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + result = build_threat_replay( + repository_root=args.repository_root, + temporal_result_root=args.temporal_result_root, + geometry_result_root=args.geometry_result_root, + detector_result_root=args.detector_result_root, + output_root=args.output_root, + ) + print( + json.dumps( + { + "result_id": result.result_id, + "result_root": str(result.result_root), + "accepted": result.accepted, + "metrics": result.metrics, + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 0 if result.accepted else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index b58f2c5..e5e3f73 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -112,6 +112,7 @@ from k1link.web.laboratory_api import build_laboratory_router from k1link.web.laboratory_report_api import build_laboratory_report_router from k1link.web.lidar_api import build_lidar_router from k1link.web.lidar_local_surface_service import K1LocalSurfaceReadService +from k1link.web.m4_threat_replay_api import build_m4_threat_replay_router from k1link.web.map_api import ( MapGatewayConfiguration, MapGatewayProxy, @@ -752,6 +753,17 @@ app.include_router( ), ) ) +app.include_router( + build_m4_threat_replay_router( + root_provider=lambda: ( + REPOSITORY_ROOT + / ".runtime" + / "compute-experiments" + / "m4" + / "replay-threat" + ), + ) +) app.include_router( build_e46e_ready_stack_router( root_provider=lambda: ( diff --git a/src/k1link/web/m4_threat_replay_api.py b/src/k1link/web/m4_threat_replay_api.py new file mode 100644 index 0000000..f8482fc --- /dev/null +++ b/src/k1link/web/m4_threat_replay_api.py @@ -0,0 +1,315 @@ +"""Read-only LAB projection of the canonical M4.6 replay threat result.""" + +from __future__ import annotations + +import copy +import json +import re +from collections.abc import Callable, Iterator +from functools import lru_cache +from pathlib import Path +from typing import Final + +from fastapi import APIRouter, HTTPException, Query + +from k1link.perception.threat_replay import ( + THREAT_REPLAY_FRAME_SCHEMA, + THREAT_REPLAY_RESULT_PREFIX, + THREAT_REPLAY_VISUAL_SCHEMA, + ThreatReplayError, + ThreatReplayResult, + read_threat_replay_result, +) + +M4_THREAT_CATALOG_SCHEMA: Final = "missioncore.m4-threat-replay-catalog/v1" +M4_THREAT_VIEW_SCHEMA: Final = "missioncore.m4-threat-replay-view/v1" +M4_THREAT_VIDEO_SCHEMA: Final = "missioncore.m4-threat-video-overlay/v1" +M4_THREAT_VISUAL_CATALOG_SCHEMA: Final = ( + "missioncore.m4-threat-visual-catalog/v1" +) +_RESULT_ID = re.compile(rf"^{THREAT_REPLAY_RESULT_PREFIX}[a-f0-9]{{64}}$") +RootProvider = Callable[[], Path | None] + + +def build_m4_threat_replay_router( + *, + root_provider: RootProvider = lambda: None, +) -> APIRouter: + router = APIRouter(prefix="/api/v1/laboratory/m4-threat", tags=["laboratory"]) + + def result(result_id: str) -> ThreatReplayResult: + if _RESULT_ID.fullmatch(result_id) is None: + raise HTTPException(status_code=404, detail="M4.6 result не найден") + root = _configured_root(root_provider) + if root is None: + raise HTTPException(status_code=404, detail="M4.6 result не найден") + path = (root / result_id).resolve() + if path.parent != root or path.is_symlink(): + raise HTTPException(status_code=404, detail="M4.6 result не найден") + try: + return _read_threat_result_cached(str(path), _result_signature(path)) + except (ThreatReplayError, OSError, ValueError): + raise HTTPException(status_code=404, detail="M4.6 result не найден") from None + + @router.get("/results") + def list_results(limit: int = Query(default=1, ge=1, le=10)) -> dict[str, object]: + candidates = _candidates(root_provider) + items: list[dict[str, object]] = [] + invalid_total = 0 + for candidate in candidates: + try: + frozen = result(candidate.name) + if len(items) < limit: + items.append(_project_result(frozen)) + except HTTPException: + invalid_total += 1 + return { + "schema_version": M4_THREAT_CATALOG_SCHEMA, + "configured": _configured_root(root_provider) is not None, + "items": items, + "candidate_total": len(candidates), + "invalid_total": invalid_total, + "access": "read-only-replay-simulated", + } + + @router.get("/results/{result_id}/visuals") + def list_visuals(result_id: str) -> dict[str, object]: + frozen = result(result_id) + frames = _read_jsonl(frozen.result_root / "visual-frames.jsonl") + return { + "schema_version": M4_THREAT_VISUAL_CATALOG_SCHEMA, + "result_id": result_id, + "items": [ + { + "ordinal": index + 1, + "sequence": item["sequence"], + "frame_id": item["frame_id"], + "source_time_ns": item["source_time_ns"], + "metric_obstacle_count": len(_array(item.get("metric_obstacles"))), + "camera_proposal_count": len(_array(item.get("camera_proposals"))), + "point_cloud_sample_count": item["point_cloud_sample_count"], + } + for index, item in enumerate(frames) + ], + "access": "read-only-replay-simulated", + } + + @router.get("/results/{result_id}/visuals/{ordinal}") + def get_visual(result_id: str, ordinal: int) -> dict[str, object]: + frozen = result(result_id) + if not 1 <= ordinal <= 32: + raise HTTPException(status_code=404, detail="M4.6 visual frame не найден") + frames = _read_jsonl(frozen.result_root / "visual-frames.jsonl") + if len(frames) != 32: + raise HTTPException(status_code=404, detail="M4.6 visual frame не найден") + return { + **copy.deepcopy(frames[ordinal - 1]), + "result_id": result_id, + "ordinal": ordinal, + "ground_truth": False, + "access": "read-only-replay-simulated", + } + + @router.get("/results/{result_id}/video-overlay") + def get_video_overlay(result_id: str) -> dict[str, object]: + frozen = result(result_id) + identity = frozen.manifest["identity"] + assert isinstance(identity, dict) + return copy.deepcopy( + _cached_video_overlay( + result_id, + str(frozen.result_root), + str(identity["frames_sha256"]), + str(identity["source_session_id"]), + ) + ) + + return router + + +@lru_cache(maxsize=4) +def _read_threat_result_cached( + root_value: str, + signature: tuple[int, ...], +) -> ThreatReplayResult: + del signature + return read_threat_replay_result(Path(root_value)) + + +@lru_cache(maxsize=4) +def _cached_video_overlay( + result_id: str, + root_value: str, + frames_sha256: str, + source_session_id: str, +) -> dict[str, object]: + root = Path(root_value).resolve(strict=True) + if root.is_symlink() or not root.is_dir() or len(frames_sha256) != 64: + raise ValueError("M4.6 video evidence identity changed") + frames = [] + for expected_sequence, row in enumerate(_iter_jsonl(root / "frames.jsonl")): + if ( + row.get("schema_version") != THREAT_REPLAY_FRAME_SCHEMA + or row.get("sequence") != expected_sequence + ): + raise ValueError("M4.6 video frame order changed") + frames.append( + { + "frame_index": expected_sequence, + "session_seconds": _nonnegative_int( + row.get("source_time_ns"), "source time" + ) + / 1_000_000_000, + "source_available": row["source_available"], + "camera_proposals": copy.deepcopy(row["camera_proposals"]), + "decision_counts": _decision_counts(_array(row.get("assessments"))), + } + ) + if len(frames) != 4489: + raise ValueError("M4.6 video frame coverage changed") + return { + "schema_version": M4_THREAT_VIDEO_SCHEMA, + "result_id": result_id, + "recorded_source": { + "session_id": source_session_id, + "source_id": "sensor.camera.right", + "synchronization": "host-arrival-best-effort", + }, + "image_width": 800, + "image_height": 600, + "timeline_start_seconds": frames[0]["session_seconds"], + "timeline_end_seconds": frames[-1]["session_seconds"], + "frame_count": len(frames), + "frames": frames, + "ground_truth": False, + "authority": "replay-simulated", + "access": "read-only-replay-simulated-video", + } + + +def _project_result(result: ThreatReplayResult) -> dict[str, object]: + identity = result.manifest["identity"] + assert isinstance(identity, dict) + return { + "schema_version": M4_THREAT_VIEW_SCHEMA, + "result_id": result.result_id, + "created_at_utc": result.manifest["created_at_utc"], + "status": result.report["status"], + "profile_id": identity["profile_id"], + "rig_profile_id": identity["rig_profile_id"], + "corridor_profile_id": identity["corridor_profile_id"], + "source_result_ids": { + "detector": identity["detector_result_id"], + "geometry": identity["geometry_result_id"], + "temporal": identity["temporal_result_id"], + }, + "metrics": copy.deepcopy(result.metrics), + "configuration": copy.deepcopy(result.report["configuration"]), + "acceptance_requirements": copy.deepcopy( + result.report["acceptance_requirements"] + ), + "limitations": copy.deepcopy(result.report["limitations"]), + "accepted": result.accepted, + "ground_truth": False, + "authority": "replay-simulated", + "physical_collision_accepted": False, + "actuation_allowed": False, + "access": "read-only-replay-simulated", + } + + +def _decision_counts(raw: list[object]) -> dict[str, int]: + result = {"threat": 0, "not-threat": 0, "unknown": 0} + for item in raw: + assessment = item if isinstance(item, dict) else {} + decision = assessment.get("decision") + if isinstance(decision, str) and decision in result: + result[decision] += 1 + return result + + +def _configured_root(provider: RootProvider) -> Path | None: + value = provider() + if value is None: + return None + candidate = value.expanduser().absolute() + if candidate.is_symlink(): + return None + try: + root = candidate.resolve(strict=True) + except OSError: + return None + return root if root.is_dir() else None + + +def _result_signature(root: Path) -> tuple[int, ...]: + signature: list[int] = [] + for name in ( + "manifest.json", + "report.json", + "fixtures.json", + "frames.jsonl", + "visual-frames.jsonl", + ): + path = root / name + if not path.is_file() or path.is_symlink(): + raise ValueError("M4.6 result artifact is invalid") + stat = path.stat() + signature.extend((stat.st_size, stat.st_mtime_ns)) + return tuple(signature) + + +def _candidates(provider: RootProvider) -> list[Path]: + root = _configured_root(provider) + if root is None: + return [] + return sorted( + ( + item + for item in root.iterdir() + if item.is_dir() + and not item.is_symlink() + and _RESULT_ID.fullmatch(item.name) + ), + key=lambda item: item.stat().st_mtime_ns, + reverse=True, + ) + + +def _read_jsonl(path: Path) -> list[dict[str, object]]: + return list(_iter_jsonl(path)) + + +def _iter_jsonl(path: Path) -> Iterator[dict[str, object]]: + with path.open("r", encoding="utf-8") as handle: + for line in handle: + value = json.loads(line) + if not isinstance(value, dict): + raise ValueError("M4.6 JSONL row is invalid") + if value.get("schema_version") not in { + THREAT_REPLAY_FRAME_SCHEMA, + THREAT_REPLAY_VISUAL_SCHEMA, + }: + raise ValueError("M4.6 JSONL schema is invalid") + yield value + + +def _array(value: object) -> list[object]: + if not isinstance(value, list): + raise ValueError("M4.6 array is invalid") + return value + + +def _nonnegative_int(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError(f"M4.6 {label} is invalid") + return value + + +__all__ = [ + "M4_THREAT_CATALOG_SCHEMA", + "M4_THREAT_VIDEO_SCHEMA", + "M4_THREAT_VIEW_SCHEMA", + "M4_THREAT_VISUAL_CATALOG_SCHEMA", + "build_m4_threat_replay_router", +] diff --git a/tests/test_laboratory_evidence_registry.py b/tests/test_laboratory_evidence_registry.py index 5b1a06d..1692c3c 100644 --- a/tests/test_laboratory_evidence_registry.py +++ b/tests/test_laboratory_evidence_registry.py @@ -127,7 +127,7 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None: repository_root / "config" / "laboratories" ) - assert len(registry.definitions) == 31 + assert len(registry.definitions) == 32 assert {item.work_id for item in registry.definitions} >= { "e31-source-binding", "e46j-raw-fisheye-realtime", @@ -136,4 +136,5 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None: "l32-pointpillars-camera-review", "l33-camera-first-detector-review", "l34f-adjudicated-reference", + "m4-replay-threat", } diff --git a/tests/test_laboratory_execution.py b/tests/test_laboratory_execution.py index e90684f..1e40a52 100644 --- a/tests/test_laboratory_execution.py +++ b/tests/test_laboratory_execution.py @@ -90,6 +90,7 @@ def test_repository_registry_classifies_every_evidence_definition() -> None: evidence, execution = _registries() assert {row.work_id for row in execution.definitions} == { + "m4-replay-threat", "e33-worker-shadow", "e35-degradation-recovery", "e46j-raw-fisheye-realtime", diff --git a/tests/test_laboratory_value_review_registry.py b/tests/test_laboratory_value_review_registry.py index 41b77e4..3f5b271 100644 --- a/tests/test_laboratory_value_review_registry.py +++ b/tests/test_laboratory_value_review_registry.py @@ -80,11 +80,12 @@ def test_product_value_review_registry_covers_reviewed_laboratory_families() -> root / "config" / "laboratory-value-review.json" ) - assert len(registry.entries) == 34 + assert len(registry.entries) == 35 assert {entry.catalog_id for entry in registry.entries} >= { "e28-local-surface", "e46d-temporal-failure-audit", "e46j-raw-fisheye-realtime", "l31-pointpillars-ravnoves", "l34f-adjudicated-reference", + "m4-replay-threat", } diff --git a/tests/test_m4_threat_replay_result.py b/tests/test_m4_threat_replay_result.py new file mode 100644 index 0000000..fd301f9 --- /dev/null +++ b/tests/test_m4_threat_replay_result.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +from pathlib import Path + +from fastapi.routing import APIRoute + +from k1link.perception.threat_replay import read_threat_replay_result +from k1link.web.m4_threat_replay_api import build_m4_threat_replay_router + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RESULT_ID = ( + "m4-threat-replay-" + "7e1613a3ea35638b5ea7a3f7c1c78fe9eba1a3adae540b652dec167f815d45b2" +) +RESULTS_ROOT = REPOSITORY_ROOT / ".runtime/compute-experiments/m4/replay-threat" + + +def _endpoint(path: str): + router = build_m4_threat_replay_router(root_provider=lambda: RESULTS_ROOT) + return next( + route.endpoint + for route in router.routes + if isinstance(route, APIRoute) and route.path == path + ) + + +def test_full_source_threat_result_closes_m4_6_contract() -> None: + result = read_threat_replay_result(RESULTS_ROOT / RESULT_ID) + + assert result.accepted is True + assert result.metrics["frames"] == {"total": 4489, "failed": 0} + assert result.metrics["evidence"] == { + "camera-only": 10158, + "current-metric": 27299, + "stale-or-held": 37995, + } + assert result.metrics["decisions"] == { + "not-threat": 6610, + "threat": 8010, + "unknown": 60832, + } + assert result.metrics["fixtures"] == { + "critical": 4, + "critical_false_not_threat": 0, + "passed": 9, + "total": 9, + } + + +def test_threat_result_is_content_bound_and_visual_evidence_is_complete() -> None: + result = read_threat_replay_result(RESULTS_ROOT / RESULT_ID) + identity = result.manifest["identity"] + assert isinstance(identity, dict) + + assert identity["frames_sha256"] == ( + "bf690358efb45c323db7172251074b33c3ef7ede6ae99bd8d3da53cfba86b142" + ) + assert identity["visuals_sha256"] == ( + "fb022c6efd84f27c0916a6c87887443c9b43993ac4b1f9910332433152533dea" + ) + visual = result.metrics["visual_evidence"] + assert isinstance(visual, dict) + assert visual["frame_count"] == 32 + assert all( + visual[key] is True + for key in ( + "video_overlay_available", + "camera_boxes_available", + "point_cloud_available", + "metric_distance_available", + "virtual_corridor_available", + ) + ) + + +def test_m4_6_lab_api_projects_report_and_exact_visual_frame() -> None: + list_results = _endpoint("/api/v1/laboratory/m4-threat/results") + list_visuals = _endpoint( + "/api/v1/laboratory/m4-threat/results/{result_id}/visuals" + ) + get_visual = _endpoint( + "/api/v1/laboratory/m4-threat/results/{result_id}/visuals/{ordinal}" + ) + + catalog = list_results(limit=1) + assert catalog["items"][0]["result_id"] == RESULT_ID + assert catalog["items"][0]["authority"] == "replay-simulated" + visuals = list_visuals(RESULT_ID) + assert len(visuals["items"]) == 32 + frame = get_visual(RESULT_ID, 1) + assert frame["schema_version"] == "missioncore.perception-threat-visual-frame/v1" + assert frame["point_cloud_sample_count"] > 0 + assert frame["rig"] == { + "length_m": 1.0, + "nominal_sensor_height_m": 1.25, + "width_m": 0.6, + } + + +def test_m4_6_video_overlay_covers_the_exact_recorded_camera_timeline() -> None: + get_overlay = _endpoint( + "/api/v1/laboratory/m4-threat/results/{result_id}/video-overlay" + ) + + overlay = get_overlay(RESULT_ID) + + assert overlay["frame_count"] == 4489 + assert overlay["recorded_source"]["session_id"] == ( + "20260720T065719Z_viewer_live" + ) + assert overlay["frames"][0]["frame_index"] == 0 + assert overlay["frames"][-1]["frame_index"] == 4488 + assert overlay["authority"] == "replay-simulated" diff --git a/tests/test_perception_architecture.py b/tests/test_perception_architecture.py index c3b7632..35e19fc 100644 --- a/tests/test_perception_architecture.py +++ b/tests/test_perception_architecture.py @@ -52,6 +52,20 @@ TEMPORAL_RUNTIME_MODULES = ( "temporal_replay.py", "temporal_replay_cli.py", ) +THREAT_RUNTIME_MODULES = ( + "contracts.py", + "detector_replay_contracts.py", + "detector_replay_result.py", + "geometry.py", + "geometry_math.py", + "geometry_replay.py", + "providers.py", + "recorded_source.py", + "temporal_replay.py", + "threat.py", + "threat_replay.py", + "threat_replay_cli.py", +) def _imports(path: Path) -> set[str]: @@ -224,6 +238,25 @@ def test_temporal_runtime_closure_imports_no_legacy_compute_or_device_package() assert {name: modules for name, modules in violations.items() if modules} == {} +def test_threat_runtime_closure_imports_no_legacy_compute_device_lab_or_web_package() -> None: + violations = { + name: sorted( + module + for module in _imports(PERCEPTION_ROOT / name) + if module.startswith( + ( + "k1link.compute", + "k1link.device_plugins", + "k1link.laboratory", + "k1link.web", + ) + ) + ) + for name in THREAT_RUNTIME_MODULES + } + assert {name: modules for name, modules in violations.items() if modules} == {} + + def test_new_perception_boundary_has_no_experiment_specific_imports() -> None: violations: dict[str, str] = {} for path in PERCEPTION_ROOT.glob("*.py"): diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py index 8a16d0e..9bfb69a 100644 --- a/tests/test_perception_graph.py +++ b/tests/test_perception_graph.py @@ -234,8 +234,7 @@ class _Threat: provider_id = "test-threat/v1" def assess(self, obstacle_map: LocalObstacleMap) -> tuple[ThreatAssessment, ...]: - occupied = obstacle_map.occupied - return tuple( + metric = tuple( ThreatAssessment( assessment_id=f"assessment-{item.component_id}", component_id=item.component_id, @@ -249,8 +248,41 @@ class _Threat: decision=ThreatDecision.NOT_THREAT, reason_codes=("qualified-corridor-clear",), ) - for item in occupied + for item in obstacle_map.occupied ) + unknown = tuple( + ThreatAssessment( + assessment_id=f"assessment-{item.component_id}", + component_id=item.component_id, + rig_profile_id="ravnoves00-virtual-rig/v1", + corridor_profile_id="ravnoves00-virtual-corridor/v1", + qualification=QualificationState.UNQUALIFIED, + relative_speed_mps=None, + closest_approach_m=None, + ttc_seconds=None, + corridor_intersection=CorridorIntersection.UNKNOWN, + decision=ThreatDecision.UNKNOWN, + reason_codes=("incomplete-evidence",), + ) + for item in obstacle_map.unknown + ) + camera = tuple( + ThreatAssessment( + assessment_id=f"assessment-{item.proposal_id}", + component_id=item.proposal_id, + rig_profile_id="ravnoves00-virtual-rig/v1", + corridor_profile_id="ravnoves00-virtual-corridor/v1", + qualification=QualificationState.UNQUALIFIED, + relative_speed_mps=None, + closest_approach_m=None, + ttc_seconds=None, + corridor_intersection=CorridorIntersection.UNKNOWN, + decision=ThreatDecision.UNKNOWN, + reason_codes=("camera-only",), + ) + for item in obstacle_map.camera_uncertainty + ) + return (*metric, *unknown, *camera) def _config( diff --git a/tests/test_replay_threat_provider.py b/tests/test_replay_threat_provider.py new file mode 100644 index 0000000..2dcc7f5 --- /dev/null +++ b/tests/test_replay_threat_provider.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +from pathlib import Path + +from k1link.perception.contracts import ( + BoundingRegion2D, + CorridorIntersection, + GridCell, + HistorySample, + LocalObstacleMap, + MotionState, + ObjectProposal2D, + SourceAccounting, + TemporalObstacle, + TemporalState, + ThreatDecision, +) +from k1link.perception.graph_validation import validate_threats +from k1link.perception.threat import ( + DualEvidenceReplayThreatProvider, + ReplayPose, + load_replay_threat_profile, +) + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m4-replay-threat-v1.json" + + +class _Poses: + def pose_for_frame(self, frame_id: str) -> ReplayPose: + return ReplayPose( + frame_id=frame_id, + position_map_xyz_m=(0.0, 0.0, 0.0), + orientation_map_from_lidar_xyzw=(0.0, 0.0, 0.0, 1.0), + ) + + +def _obstacle( + component_id: str, + cell: GridCell, + *, + motion: MotionState, + history: tuple[tuple[str, int, tuple[float, float, float]], ...], + state: TemporalState = TemporalState.CURRENT, + semantic_hint: str | None = None, +) -> TemporalObstacle: + samples = tuple( + HistorySample(frame_id=frame_id, evidence_time_ns=time_ns, centroid_xyz_m=point) + for frame_id, time_ns, point in history + ) + current = samples[-1] + return TemporalObstacle( + component_id=component_id, + identity_scope="ephemeral", + state=state, + ttl_ns=750_000_000, + last_hit_ns=current.evidence_time_ns, + age_ns=0 if state is TemporalState.CURRENT else 100_000_000, + association_basis="test-spatial-support", + history=samples, + cells=() if state is TemporalState.EXPIRED else (cell,), + coordinate_frame=None if state is TemporalState.EXPIRED else "map", + last_centroid_xyz_m=None if state is TemporalState.EXPIRED else current.centroid_xyz_m, + motion=MotionState.UNKNOWN if state is not TemporalState.CURRENT else motion, + motion_confidence=( + 0.0 + if state is not TemporalState.CURRENT or motion is MotionState.UNKNOWN + else 1.0 + ), + motion_reason=( + "stale-support" + if state is not TemporalState.CURRENT + else "bounded-map-history-moving" + if motion is MotionState.MOVING + else "bounded-map-history-stationary" + if motion is MotionState.STATIONARY + else "insufficient-history" + ), + semantic_hint=semantic_hint, + ) + + +def _proposal(frame_id: str = "frame-000002") -> ObjectProposal2D: + return ObjectProposal2D( + proposal_id="proposal-camera-only", + source_id="RAVNOVES00", + frame_id=frame_id, + region=BoundingRegion2D(10.0, 10.0, 20.0, 20.0), + objectness=0.8, + provider_id="test-detector/v1", + model_id="test-model/v1", + preprocess_id="test-preprocess/v1", + semantic_hint="person", + ) + + +def _map( + *, + occupied: tuple[TemporalObstacle, ...] = (), + unknown: tuple[TemporalObstacle, ...] = (), + camera: tuple[ObjectProposal2D, ...] = (), + frame_id: str = "frame-000002", +) -> LocalObstacleMap: + return LocalObstacleMap( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id=frame_id, + graph_id="reference-perception-graph/v1", + generated_monotonic_ns=0, + output_age_ns=0, + occupied=occupied, + unknown=unknown, + camera_uncertainty=camera, + accounting=SourceAccounting(1, 1, 0, 0), + ) + + +def test_replay_threat_profile_freezes_virtual_authority_and_dual_evidence_policy() -> None: + profile = load_replay_threat_profile(PROFILE_PATH) + + assert profile.rig.body_length_m == 1.0 + assert profile.rig.body_width_m == 0.6 + assert profile.rig.nominal_sensor_height_m == 1.25 + assert profile.corridor.forward_length_m == 8.0 + assert profile.calibration_content_sha256 == ( + "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9" + ) + + +def test_static_crossing_approaching_and_geometry_only_critical_cases_are_never_safe() -> None: + provider = DualEvidenceReplayThreatProvider( + pose_resolver=_Poses(), + profile=load_replay_threat_profile(PROFILE_PATH), + ) + current_frame = "frame-000002" + critical = ( + _obstacle( + "static-in-corridor", + GridCell(6, 0, 0), + motion=MotionState.STATIONARY, + history=( + ("frame-000000", 0, (2.925, 0.225, 0.225)), + (current_frame, 300_000_000, (2.925, 0.225, 0.225)), + ), + ), + _obstacle( + "crossing", + GridCell(6, 3, 0), + motion=MotionState.MOVING, + history=( + ("frame-000000", 0, (2.925, 2.575, 0.225)), + (current_frame, 300_000_000, (2.925, 1.575, 0.225)), + ), + ), + _obstacle( + "approaching", + GridCell(9, 0, 0), + motion=MotionState.MOVING, + history=( + ("frame-000000", 0, (6.275, 0.225, 0.225)), + (current_frame, 300_000_000, (4.275, 0.225, 0.225)), + ), + semantic_hint="car", + ), + _obstacle( + "geometry-only", + GridCell(4, 0, 0), + motion=MotionState.STATIONARY, + history=( + ("frame-000000", 0, (2.025, 0.225, 0.225)), + (current_frame, 300_000_000, (2.025, 0.225, 0.225)), + ), + semantic_hint=None, + ), + ) + + result = provider.assess(_map(occupied=critical)) + + assert {item.decision for item in result} == {ThreatDecision.THREAT} + assert all(item.corridor_intersection is CorridorIntersection.INTERSECTS for item in result) + assert ( + next(item for item in result if item.component_id == "approaching").ttc_seconds + is not None + ) + assert "geometry-only-evidence" in next( + item for item in result if item.component_id == "geometry-only" + ).reason_codes + + +def test_receding_and_static_outside_are_clear_but_incomplete_evidence_is_unknown() -> None: + provider = DualEvidenceReplayThreatProvider( + pose_resolver=_Poses(), + profile=load_replay_threat_profile(PROFILE_PATH), + ) + current_frame = "frame-000002" + clear = ( + _obstacle( + "static-outside", + GridCell(6, 7, 0), + motion=MotionState.STATIONARY, + history=( + ("frame-000000", 0, (2.925, 3.375, 0.225)), + (current_frame, 300_000_000, (2.925, 3.375, 0.225)), + ), + ), + _obstacle( + "receding-behind", + GridCell(-5, 0, 0), + motion=MotionState.MOVING, + history=( + ("frame-000000", 0, (-1.025, 0.225, 0.225)), + (current_frame, 300_000_000, (-2.025, 0.225, 0.225)), + ), + ), + ) + incomplete = _obstacle( + "unknown-motion", + GridCell(6, 7, 0), + motion=MotionState.UNKNOWN, + history=((current_frame, 300_000_000, (2.925, 3.375, 0.225)),), + ) + held = _obstacle( + "occluded-held", + GridCell(6, 0, 0), + motion=MotionState.UNKNOWN, + history=((current_frame, 300_000_000, (2.925, 0.225, 0.225)),), + state=TemporalState.HELD, + ) + obstacle_map = _map( + occupied=(*clear, incomplete), + unknown=(held,), + camera=(_proposal(),), + ) + + result = provider.assess(obstacle_map) + by_id = {item.component_id: item for item in result} + + assert by_id["static-outside"].decision is ThreatDecision.NOT_THREAT + assert by_id["receding-behind"].decision is ThreatDecision.NOT_THREAT + assert by_id["unknown-motion"].decision is ThreatDecision.UNKNOWN + assert by_id["occluded-held"].decision is ThreatDecision.UNKNOWN + assert by_id["proposal-camera-only"].decision is ThreatDecision.UNKNOWN + assert by_id["proposal-camera-only"].closest_approach_m is None + validate_threats(obstacle_map, result) + + +def test_semantic_hint_and_ephemeral_component_name_do_not_change_threat_geometry() -> None: + provider = DualEvidenceReplayThreatProvider( + pose_resolver=_Poses(), + profile=load_replay_threat_profile(PROFILE_PATH), + ) + history = ( + ("frame-000000", 0, (2.925, 0.225, 0.225)), + ("frame-000002", 300_000_000, (2.925, 0.225, 0.225)), + ) + first = _obstacle( + "ephemeral-a", + GridCell(6, 0, 0), + motion=MotionState.STATIONARY, + history=history, + semantic_hint="car", + ) + second = _obstacle( + "ephemeral-b", + GridCell(6, 0, 0), + motion=MotionState.STATIONARY, + history=history, + semantic_hint=None, + ) + + values = provider.assess(_map(occupied=(first, second))) + + assert values[0].decision == values[1].decision + assert values[0].corridor_intersection == values[1].corridor_intersection + assert values[0].relative_speed_mps == values[1].relative_speed_mps + assert values[0].closest_approach_m == values[1].closest_approach_m + assert values[0].ttc_seconds == values[1].ttc_seconds