fix(m4): retain compact obstacle evidence

src/k1link/perception/temporal_replay.py

@@ -46,18 +46,15 @@ TEMPORAL_REPLAY_REPORT_NAME: Final = "report.json"
 TEMPORAL_REPLAY_MANIFEST_NAME: Final = "manifest.json"
 
 E34_RESULT_ID: Final = (
-    "e34-temporal-occupied-"
-    "8d9abb3f2cc072cfdbb16cc4e55798e05c35a0abe0b8f691096770e091573a73"
+    "e34-temporal-occupied-8d9abb3f2cc072cfdbb16cc4e55798e05c35a0abe0b8f691096770e091573a73"
 )
 E34_MANIFEST_SHA256: Final = "88285f44d0316913881cc0948a4dfb300d56b460c51be36d51b4dd717cbf4170"
 E51_RESULT_ID: Final = (
-    "e51-motion-semantic-"
-    "1abb7eb9940608fc5af95a1f318cadfbc42ac2412a8662b6622e000e03da1555"
+    "e51-motion-semantic-1abb7eb9940608fc5af95a1f318cadfbc42ac2412a8662b6622e000e03da1555"
 )
 E51_MANIFEST_SHA256: Final = "a38ecad59765d1439ac432a34c053362195ec4b493d8e8f1d4a1731f99836437"
 E46B_RESULT_ID: Final = (
-    "e46b-temporal-motion-"
-    "78d038912273364e36f996401873a8ee178a94641350021c2cd35bcb301ba36d"
+    "e46b-temporal-motion-78d038912273364e36f996401873a8ee178a94641350021c2cd35bcb301ba36d"
 )
 E46B_MANIFEST_SHA256: Final = "b0d0b6bdfa23f0475106c1870771ee90dd6de85719396a665dd7311f93da3d59"
 E46B_CASES_SHA256: Final = "95b58300f10dc796f7576bffbcc670ac76b74d6f25b13c0768f57331cba49074"
@@ -100,9 +97,7 @@ def build_temporal_replay(
     store = RecordedGeometryStore.from_repository(repository)
     temporal = BoundedSpatialTemporalProvider(point_resolver=store, profile=profile)
     motion = ClassIndependentMotionEstimator(profile=profile)
-    rolling_profile = load_rolling_map_profile(
-        repository / DEFAULT_ROLLING_MAP_PROFILE_PATH
-    )
+    rolling_profile = load_rolling_map_profile(repository / DEFAULT_ROLLING_MAP_PROFILE_PATH)
     rolling = RollingLocalObstacleMapProvider(
         pose_resolver=store,
         profile=rolling_profile,
@@ -147,9 +142,7 @@ def build_temporal_replay(
                 obstacles = motion.estimate(packet, temporal_obstacles)
                 rolling_retained = rolling.update(packet, obstacles)
                 latencies_ms.append((time.perf_counter_ns() - frame_started_ns) / 1_000_000)
-                current = tuple(
-                    item for item in obstacles if item.state is TemporalState.CURRENT
-                )
+                current = tuple(item for item in obstacles if item.state is TemporalState.CURRENT)
                 held = tuple(item for item in obstacles if item.state is TemporalState.HELD)
                 expired = tuple(item for item in obstacles if item.state is TemporalState.EXPIRED)
                 motion_counts = _motion_counts(current)
@@ -158,13 +151,10 @@ def build_temporal_replay(
                     "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
-                    ),
+                    "source_available": (packet.envelope.registered_point_increment.available),
                     "input_observation_count": len(observations),
                     "current_occupied_input_count": sum(
-                        item.occupied_support
-                        and item.currentness.value == "current"
+                        item.occupied_support and item.currentness.value == "current"
                         for item in observations
                     ),
                     "nonmetric_uncertainty_input_count": sum(
@@ -173,13 +163,10 @@ def build_temporal_replay(
                     "current": [item.to_dict() for item in current],
                     "held": [item.to_dict() for item in held],
                     "expired": [item.to_dict() for item in expired],
-                    "rolling_retained": [
-                        item.to_dict() for item in rolling_retained
-                    ],
+                    "rolling_retained": [item.to_dict() for item in rolling_retained],
                     "motion_counts": motion_counts,
                     "map_frame_jump_candidate": any(
-                        item.association_basis == "map-frame-discontinuity"
-                        for item in current
+                        item.association_basis == "map-frame-discontinuity" for item in current
                     ),
                     "policy": _frame_policy(),
                     "authority": _false_authority(),
@@ -216,9 +203,7 @@ def build_temporal_replay(
         identity = {
             "schema_version": TEMPORAL_REPLAY_SCHEMA_V2,
             "geometry_result_id": geometry.result_id,
-            "geometry_manifest_sha256": _file_sha256(
-                geometry.result_root / "manifest.json"
-            ),
+            "geometry_manifest_sha256": _file_sha256(geometry.result_root / "manifest.json"),
             "geometry_frames_sha256": profile.geometry_frames_sha256,
             "profile_id": profile.profile_id,
             "profile_sha256": profile.profile_sha256,
@@ -398,18 +383,11 @@ def read_temporal_replay_result(root: Path) -> TemporalReplayResult:
         if is_v2
         else _requirements(metrics, ttl_ns / 1_000_000_000)
     )
-    if (
-        ttl_ns != 750_000_000
-        or requirements != expected_requirements
-    ):
+    if ttl_ns != 750_000_000 or requirements != expected_requirements:
         raise TemporalReplayError("temporal replay acceptance was not derived from metrics")
     if (
         report.get("schema_version")
-        != (
-            TEMPORAL_REPLAY_REPORT_SCHEMA_V2
-            if is_v2
-            else TEMPORAL_REPLAY_REPORT_SCHEMA
-        )
+        != (TEMPORAL_REPLAY_REPORT_SCHEMA_V2 if is_v2 else TEMPORAL_REPLAY_REPORT_SCHEMA)
         or report.get("result_id") != resolved.name
         or report.get("identity_sha256") != identity_sha256
         or report.get("metrics") != metrics
@@ -490,9 +468,7 @@ def _requirements_v2(
     requirements.update(
         {
             "registered_increment_is_not_treated_as_complete_scan": True,
-            "rolling_map_processed_every_source_frame": (
-                rolling.get("input_frames") == 4489
-            ),
+            "rolling_map_processed_every_source_frame": (rolling.get("input_frames") == 4489),
             "rolling_map_materialized_retained_occupancy": (
                 _integer(
                     rolling.get("retained_component_publications"),
@@ -506,8 +482,7 @@ def _requirements_v2(
                     "maximum rolling retained age",
                 )
                 <= round(rolling_retention_seconds * 1_000_000_000)
-                and rolling.get("retention_ns")
-                == round(rolling_retention_seconds * 1_000_000_000)
+                and rolling.get("retention_ns") == round(rolling_retention_seconds * 1_000_000_000)
                 and rolling.get("maximum_cells") == rolling_maximum_cells
                 and _integer(
                     rolling.get("capacity_evicted_cells"),
@@ -571,22 +546,24 @@ def _requirements(metrics: dict[str, object], ttl_seconds: float) -> dict[str, b
     )
     return {
         "full_frame_accounting": frames == {"total": 4489, "failed": 0},
-        "geometry_observation_accounting": metrics.get("input_observations") == 37457,
+        "geometry_observation_accounting": (
+            metrics.get("input_observations") == temporal_input
+            and temporal.get("input_frames") == frames.get("total")
+            and temporal_input > 0
+        ),
         "metric_and_nonmetric_partition_closed": (
-            temporal.get("current_occupied_observations") == 27299
-            and temporal.get("nonmetric_uncertainty_observations") == 10158
+            temporal_input == current_input + uncertainty_input
+            and current_input > 0
+            and uncertainty_input > 0
         ),
         "camera_uncertainty_never_created_occupied_state": (
             temporal_input == current_input + uncertainty_input
         ),
         "detector_identity_changes_survive_spatial_reassociation": identity_changes > 0,
         "temporal_state_is_bounded": (
-            peak_components <= 256
-            and peak_cells <= 4096
-            and maximum_history <= 8
+            peak_components <= 256 and peak_cells <= 4096 and maximum_history <= 8
         ),
-        "held_and_expired_states_materialized": held_publications > 0
-        and expired_publications > 0,
+        "held_and_expired_states_materialized": held_publications > 0 and expired_publications > 0,
         "no_occupied_cells_survive_ttl": (
             retention.get("past_ttl_occupied_publications") == 0
             and retention.get("ghost_occupancy_past_ttl_count") == 0
@@ -600,8 +577,7 @@ def _requirements(metrics: dict[str, object], ttl_seconds: float) -> dict[str, b
             value > 0 for value in (moving, stationary, unknown)
         ),
         "motion_accounting_closed": (
-            motion_input
-            == current_publications + held_publications + expired_publications
+            motion_input == current_publications + held_publications + expired_publications
         ),
         "bounded_labeled_engineering_checks_recorded": (
             clips
@@ -651,9 +627,8 @@ def _validate_frame_ledger(
                 frame.get("nonmetric_uncertainty_input_count"),
                 "frame nonmetric uncertainty inputs",
             )
-            if (
-                frame_current_inputs + frame_uncertainty_inputs
-                != frame.get("input_observation_count")
+            if frame_current_inputs + frame_uncertainty_inputs != frame.get(
+                "input_observation_count"
             ):
                 raise TemporalReplayError("temporal frame input partition is open")
             current_inputs += frame_current_inputs
@@ -673,11 +648,7 @@ def _validate_frame_ledger(
                 groups[state] = items
                 for item in items:
                     motion_counts[item.motion.value] += 1
-            component_ids = [
-                item.component_id
-                for group in groups.values()
-                for item in group
-            ]
+            component_ids = [item.component_id for group in groups.values() for item in group]
             if len(component_ids) != len(set(component_ids)):
                 raise TemporalReplayError("temporal frame duplicated a component")
             if is_v2:
@@ -689,13 +660,9 @@ def _validate_frame_ledger(
                     )
                 )
                 if any(item.state is not TemporalState.RETAINED for item in rolling):
-                    raise TemporalReplayError(
-                        "rolling map published a non-retained component"
-                    )
+                    raise TemporalReplayError("rolling map published a non-retained component")
                 if set(component_ids) & {item.component_id for item in rolling}:
-                    raise TemporalReplayError(
-                        "rolling and temporal component identities overlap"
-                    )
+                    raise TemporalReplayError("rolling and temporal component identities overlap")
                 rolling_publications += len(rolling)
             current_publications += len(groups[TemporalState.CURRENT])
             held_publications += len(groups[TemporalState.HELD])
@@ -706,11 +673,7 @@ def _validate_frame_ledger(
     frame_metrics = _object(metrics.get("frames"), "temporal frames")
     temporal = _object(metrics.get("temporal"), "temporal metrics")
     motion = _object(metrics.get("motion"), "motion metrics")
-    rolling_metrics = (
-        _object(metrics.get("rolling_map"), "rolling map metrics")
-        if is_v2
-        else None
-    )
+    rolling_metrics = _object(metrics.get("rolling_map"), "rolling map metrics") if is_v2 else None
     if (
         frames != frame_metrics.get("total")
         or observations != metrics.get("input_observations")
@@ -724,8 +687,7 @@ def _validate_frame_ledger(
         or motion_counts[MotionState.UNKNOWN.value] != motion.get("unknown")
         or (
             rolling_metrics is not None
-            and rolling_publications
-            != rolling_metrics.get("retained_component_publications")
+            and rolling_publications != rolling_metrics.get("retained_component_publications")
         )
     ):
         raise TemporalReplayError("temporal frame ledger and metrics disagree")
@@ -801,10 +763,7 @@ def _clip_check(
 
 
 def _motion_counts(obstacles: tuple[TemporalObstacle, ...]) -> dict[str, int]:
-    return {
-        state.value: sum(item.motion is state for item in obstacles)
-        for state in MotionState
-    }
+    return {state.value: sum(item.motion is state for item in obstacles) for state in MotionState}
 
 
 def _frame_policy() -> dict[str, object]:
@@ -836,9 +795,7 @@ def _read_geometry_frame(line: bytes, sequence: int) -> dict[str, object]:
     try:
         frame = _object(json.loads(line), "geometry replay frame")
     except json.JSONDecodeError as exc:
-        raise TemporalReplayError(
-            f"geometry replay frame {sequence + 1} is invalid JSON"
-        ) from exc
+        raise TemporalReplayError(f"geometry replay frame {sequence + 1} is invalid JSON") from exc
     if frame.get("sequence") != sequence:
         raise TemporalReplayError("geometry replay frame sequence is incomplete")
     return frame
@@ -853,25 +810,23 @@ def _read_frame(
     try:
         frame = _object(json.loads(line), "temporal replay frame")
     except json.JSONDecodeError as exc:
-        raise TemporalReplayError(
-            f"temporal replay frame {sequence + 1} is invalid JSON"
-        ) from exc
+        raise TemporalReplayError(f"temporal replay frame {sequence + 1} is invalid JSON") from exc
     expected_keys = {
-            "schema_version",
-            "sequence",
-            "frame_id",
-            "source_time_ns",
-            "source_available",
-            "input_observation_count",
-            "current_occupied_input_count",
-            "nonmetric_uncertainty_input_count",
-            "current",
-            "held",
-            "expired",
-            "motion_counts",
-            "map_frame_jump_candidate",
-            "policy",
-            "authority",
+        "schema_version",
+        "sequence",
+        "frame_id",
+        "source_time_ns",
+        "source_available",
+        "input_observation_count",
+        "current_occupied_input_count",
+        "nonmetric_uncertainty_input_count",
+        "current",
+        "held",
+        "expired",
+        "motion_counts",
+        "map_frame_jump_candidate",
+        "policy",
+        "authority",
     }
     if is_v2:
         expected_keys.add("rolling_retained")

src/k1link/perception/threat_replay.py

@@ -66,7 +66,7 @@ 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
-VISUAL_GEOMETRY_REGRESSION_SEQUENCES: Final = (138, 274, 1880)
+VISUAL_GEOMETRY_REGRESSION_SEQUENCES: Final = (138, 274, 1880, 2584)
 FRAME_1880_ENGINEERING_ANCHORS: Final = (
     {
         "anchor_id": "near-concrete-hemisphere",
@@ -83,6 +83,22 @@ FRAME_1880_ENGINEERING_ANCHORS: Final = (
         "must_assert_threat": False,
     },
 )
+FRAME_2584_ENGINEERING_ANCHORS: Final = (
+    {
+        "anchor_id": "near-compact-concrete-hemisphere",
+        "x_bounds_m": (1.5, 2.4),
+        "y_bounds_m": (-0.7, 0.2),
+        "z_bounds_m": (-0.1, 0.9),
+        "must_assert_threat": True,
+    },
+    {
+        "anchor_id": "far-concrete-hemisphere-occupancy",
+        "x_bounds_m": (2.8, 4.0),
+        "y_bounds_m": (1.2, 2.4),
+        "z_bounds_m": (-0.1, 1.0),
+        "must_assert_threat": False,
+    },
+)
 
 
 class ThreatReplayError(RuntimeError):
@@ -143,6 +159,7 @@ def build_threat_replay(
     latencies_ms: list[float] = []
     visual_count = 0
     frame_1880_regression: dict[str, object] | None = None
+    frame_2584_regression: dict[str, object] | None = None
     try:
         temporal_frames_path = temporal.result_root / "frames.jsonl"
         geometry_frames_path = geometry.result_root / "frames.jsonl"
@@ -188,13 +205,8 @@ def build_threat_replay(
                         "rolling retained obstacles",
                     )
                 )
-                if any(
-                    item.state is not TemporalState.RETAINED
-                    for item in rolling_retained
-                ):
-                    raise ThreatReplayError(
-                        "temporal replay rolling map escaped retained state"
-                    )
+                if any(item.state is not TemporalState.RETAINED for item in rolling_retained):
+                    raise ThreatReplayError("temporal replay rolling map escaped retained state")
                 geometry_observations = _array(
                     geometry_frame.get("observations"), "geometry observations"
                 )
@@ -229,11 +241,8 @@ def build_threat_replay(
                 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, *rolling_retained, *unknown)
-                } | {
-                    item.proposal_id for item in camera_uncertainty
-                }
+                    item.component_id for item in (*current, *rolling_retained, *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(
@@ -248,9 +257,13 @@ def build_threat_replay(
                 if frame_count == 1880:
                     frame_1880_regression = _frame_1880_regression(
                         metric_rows,
-                        body_frame_resolver.body_frame_for_frame(
-                            packet.envelope.frame_id
-                        ),
+                        body_frame_resolver.body_frame_for_frame(packet.envelope.frame_id),
+                    )
+                if frame_count == 2584:
+                    frame_2584_regression = _frame_2584_regression(
+                        metric_rows,
+                        body_frame_resolver.body_frame_for_frame(packet.envelope.frame_id),
+                        voxel_size_m=profile.corridor.occupied_voxel_size_m,
                     )
                 for item in assessments:
                     assessment_counts[item.decision.value] += 1
@@ -327,6 +340,7 @@ def build_threat_replay(
             fixtures=fixtures,
             body_frame=body_frame_resolver.qualification_summary(),
             frame_1880_regression=frame_1880_regression,
+            frame_2584_regression=frame_2584_regression,
         )
         requirements = _requirements_v2(metrics, fixtures)
         accepted = all(value is True for value in requirements.values())
@@ -488,9 +502,7 @@ def read_threat_replay_result(root: Path) -> ThreatReplayResult:
     fixtures = _read_json(paths["threat-deterministic-fixtures"])
     accepted = all(value is True for value in requirements.values())
     expected_requirements = (
-        _requirements_v2(metrics, fixtures)
-        if is_v2
-        else _requirements_v1(metrics, fixtures)
+        _requirements_v2(metrics, fixtures) if is_v2 else _requirements_v1(metrics, fixtures)
     )
     if (
         report.get("schema_version")
@@ -687,8 +699,7 @@ def _visual_frame(
         "point_cloud_sample_count": int(sampled.shape[0]),
         "point_cloud_layer": "current-increment",
         "rolling_map_component_count": sum(
-            row.get("state") == TemporalState.RETAINED.value
-            for row in metric_rows
+            row.get("state") == TemporalState.RETAINED.value for row in metric_rows
         ),
         "metric_obstacles": metric_visuals,
         "camera_proposals": camera_rows,
@@ -778,22 +789,15 @@ def _frame_1880_regression(
                 "must_assert_threat": anchor["must_assert_threat"],
                 "matched": match is not None,
                 "component_id": component_id,
-                "centroid_body_xyz_m": (
-                    None if match is None else list(match[1])
-                ),
+                "centroid_body_xyz_m": (None if match is None else list(match[1])),
                 "decision": (
-                    None
-                    if anchor_assessment is None
-                    else anchor_assessment.get("decision")
+                    None if anchor_assessment is None else anchor_assessment.get("decision")
                 ),
             }
         )
     required_threats_passed = all(
         item["matched"] is True
-        and (
-            item["must_assert_threat"] is False
-            or item["decision"] == ThreatDecision.THREAT.value
-        )
+        and (item["must_assert_threat"] is False or item["decision"] == ThreatDecision.THREAT.value)
         for item in anchors
     )
     return {
@@ -808,6 +812,100 @@ def _frame_1880_regression(
     }
 
 
+def _frame_2584_regression(
+    metric_rows: list[dict[str, object]],
+    body_frame: ReplayBodyFrame | None,
+    *,
+    voxel_size_m: float,
+) -> dict[str, object]:
+    """Bind both visible hemispheres to produced occupancy without injecting it."""
+
+    if body_frame is None:
+        raise ThreatReplayError("frame 2584 has no qualified body frame")
+    candidates: list[
+        tuple[dict[str, object], tuple[float, float, float], tuple[float, float, float]]
+    ] = []
+    for row in metric_rows:
+        assessment = _object(row.get("assessment"), "frame 2584 assessment")
+        centroid_map = _array(row.get("centroid_map_xyz_m"), "frame 2584 centroid")
+        if len(centroid_map) != 3:
+            raise ThreatReplayError("frame 2584 centroid is invalid")
+        centroid_body = body_frame.map_point_to_body(
+            tuple(_number_value(value, "frame 2584 centroid") for value in centroid_map)
+        )
+        for raw_cell in _array(row.get("cells"), "frame 2584 cells"):
+            cell = _object(raw_cell, "frame 2584 cell")
+            point_map = tuple(
+                (_signed_integer(cell.get(key), f"frame 2584 cell {key}") + 0.5) * voxel_size_m
+                for key in ("x", "y", "z")
+            )
+            candidates.append(
+                (
+                    row,
+                    centroid_body,
+                    body_frame.map_point_to_body(point_map),
+                )
+            )
+        if not row.get("cells"):
+            raise ThreatReplayError("frame 2584 metric component has no occupied cells")
+        if assessment.get("component_id") != row.get("component_id"):
+            raise ThreatReplayError("frame 2584 assessment identity changed")
+
+    anchors: list[dict[str, object]] = []
+    matched_ids: set[str] = set()
+    for raw_anchor in FRAME_2584_ENGINEERING_ANCHORS:
+        anchor = _object(raw_anchor, "frame 2584 engineering anchor")
+        x_bounds = _bounds(anchor.get("x_bounds_m"), "frame 2584 x bounds")
+        y_bounds = _bounds(anchor.get("y_bounds_m"), "frame 2584 y bounds")
+        z_bounds = _bounds(anchor.get("z_bounds_m"), "frame 2584 z bounds")
+        match = next(
+            (
+                (row, centroid, cell)
+                for row, centroid, cell in candidates
+                if row.get("component_id") not in matched_ids
+                and x_bounds[0] <= cell[0] <= x_bounds[1]
+                and y_bounds[0] <= cell[1] <= y_bounds[1]
+                and z_bounds[0] <= cell[2] <= z_bounds[1]
+            ),
+            None,
+        )
+        component_id = None if match is None else str(match[0]["component_id"])
+        if component_id is not None:
+            matched_ids.add(component_id)
+        assessment = (
+            None
+            if match is None
+            else _object(match[0].get("assessment"), "frame 2584 anchor assessment")
+        )
+        anchors.append(
+            {
+                "anchor_id": anchor["anchor_id"],
+                "bounds_body_xyz_m": [list(x_bounds), list(y_bounds), list(z_bounds)],
+                "must_assert_threat": anchor["must_assert_threat"],
+                "matched": match is not None,
+                "component_id": component_id,
+                "component_state": None if match is None else match[0].get("state"),
+                "centroid_body_xyz_m": None if match is None else list(match[1]),
+                "matched_cell_body_xyz_m": None if match is None else list(match[2]),
+                "decision": None if assessment is None else assessment.get("decision"),
+            }
+        )
+    required_threats_passed = all(
+        item["matched"] is True
+        and (item["must_assert_threat"] is False or item["decision"] == ThreatDecision.THREAT.value)
+        for item in anchors
+    )
+    return {
+        "sequence": 2584,
+        "engineering_anchors": anchors,
+        "matched_anchor_count": sum(item["matched"] is True for item in anchors),
+        "required_threats_passed": required_threats_passed,
+        "camera_visible_hemispheres_independent_truth": False,
+        "matching_basis": "produced-occupied-cell-inside-camera-reviewed-body-window",
+        "gate": "compact-and-merged-hemisphere-occupancy-regression",
+    }
+
+
 class _FixtureBodyFrames:
     def body_frame_for_frame(self, frame_id: str) -> ReplayBodyFrame:
         return ReplayBodyFrame(
@@ -982,11 +1080,7 @@ def _fixture_obstacle(
         component_id=component_id,
         identity_scope="ephemeral",
         state=state,
-        ttl_ns=(
-            3_000_000_000
-            if state is TemporalState.RETAINED
-            else 750_000_000
-        ),
+        ttl_ns=(3_000_000_000 if state is TemporalState.RETAINED else 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",
@@ -1096,6 +1190,7 @@ def _metrics(
     fixtures: dict[str, object],
     body_frame: dict[str, object],
     frame_1880_regression: dict[str, object] | None,
+    frame_2584_regression: dict[str, object] | None,
 ) -> dict[str, object]:
     values = np.asarray(latencies_ms, dtype=np.float64)
     return {
@@ -1116,6 +1211,7 @@ def _metrics(
             "qualified_base_footprint_available": True,
             "geometry_regression_sequences": list(VISUAL_GEOMETRY_REGRESSION_SEQUENCES),
             "frame_1880_regression": frame_1880_regression,
+            "frame_2584_regression": frame_2584_regression,
         },
         "fixtures": {
             "passed": fixtures["passed_count"],
@@ -1156,8 +1252,7 @@ def _requirements_v1(
     stale_cases = [
         _object(item, "fixture")
         for item in cases
-        if isinstance(item, dict)
-        and item.get("name") in {"occluded-held", "stale-expired"}
+        if isinstance(item, dict) and item.get("name") in {"occluded-held", "stale-expired"}
     ]
     return {
         "full_ravnoves00_replay_completed": (
@@ -1168,8 +1263,7 @@ def _requirements_v1(
         ),
         "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)
+            len(stale_cases) == 2 and all(item.get("actual") == "unknown" for item in stale_cases)
         ),
         "geometry_only_evidence_is_assessed": (
             _integer(
@@ -1206,13 +1300,10 @@ def _requirements_v1(
             == _integer(body_frame.get("qualified"), "qualified body frames")
             + _integer(body_frame.get("rejected"), "rejected body frames")
             and _integer(body_frame.get("qualified"), "qualified body frames")
-            >= math.ceil(
-                _integer(body_frame.get("available"), "available body frames") * 0.95
-            )
+            >= math.ceil(_integer(body_frame.get("available"), "available body frames") * 0.95)
             and body_frame.get("origin") == "local-surface-vertical-projection"
             and body_frame.get("up") == "vendor-slam-map-gravity-axis"
-            and body_frame.get("forward")
-            == "smoothed-slam-trajectory-validated-by-camera-axis"
+            and body_frame.get("forward") == "smoothed-slam-trajectory-validated-by-camera-axis"
             and _number_value(
                 _object(
                     body_frame.get("camera_forward_alignment_deg"),
@@ -1308,6 +1399,19 @@ def _requirements_v2(
             ).get("required_threats_passed")
             is True
         ),
+        "frame_2584_retains_compact_hemisphere_and_accounts_for_far_occupancy": (
+            isinstance(visual.get("frame_2584_regression"), dict)
+            and _object(
+                visual.get("frame_2584_regression"),
+                "frame 2584 regression",
+            ).get("matched_anchor_count")
+            == len(FRAME_2584_ENGINEERING_ANCHORS)
+            and _object(
+                visual.get("frame_2584_regression"),
+                "frame 2584 regression",
+            ).get("required_threats_passed")
+            is True
+        ),
         "body_frame_is_grounded_gravity_stable_and_route_aligned": (
             body_frame.get("available")
             == _integer(body_frame.get("qualified"), "qualified body frames")

src/k1link/sessions/__init__.py

@@ -5,6 +5,7 @@ from .active import (
     ActiveSessionLeaseError,
     recover_stale_active_session_marker,
 )
+from .camera_frame import RecordedCameraFrame, RecordedCameraFrameService
 from .lab_cache import publish_lab_replay_cache
 from .media import (
     RECORDED_MEDIA_MANIFEST_SCHEMA,
@@ -65,6 +66,8 @@ __all__ = [
     "publish_lab_replay_cache",
     "RecordingMaterializationCancelled",
     "RecordedMediaArtifact",
+    "RecordedCameraFrame",
+    "RecordedCameraFrameService",
     "RECORDED_MEDIA_MANIFEST_SCHEMA",
     "RecordedMediaFile",
     "RecordedMediaInspector",

src/k1link/sessions/camera_frame.py

@@ -0,0 +1,314 @@
+from __future__ import annotations
+
+import hashlib
+import os
+import struct
+import subprocess
+import threading
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+
+from .media import RecordedMediaEpoch, RecordedMediaInspector, RecordedMediaManifest
+from .models import SessionIntegrityError
+from .store import SessionStore
+
+_MAX_KEYFRAME_DISTANCE = 120
+_FFMPEG_TIMEOUT_SECONDS = 15.0
+
+
+@dataclass(frozen=True, slots=True)
+class RecordedCameraFrame:
+    payload: bytes
+    media_type: str
+    width: int
+    height: int
+    sha256: str
+
+
+class RecordedCameraFrameService:
+    """Decode one exact archived camera frame without preparing the full video.
+
+    Canonical camera archives contain one video sample per fMP4 fragment. The
+    service validates the sealed manifest, walks back only to the preceding IDR
+    fragment and gives that bounded GOP to ffmpeg. This keeps CAMERA review
+    independent from the 4489-row VIDEO overlay and full-player preparation.
+    """
+
+    def __init__(
+        self,
+        store: SessionStore,
+        inspector: RecordedMediaInspector,
+        *,
+        ffmpeg_path: Path,
+        cache_root: Path,
+    ) -> None:
+        resolved_ffmpeg = ffmpeg_path.expanduser().resolve(strict=True)
+        if not resolved_ffmpeg.is_file() or not os.access(resolved_ffmpeg, os.X_OK):
+            raise SessionIntegrityError("ffmpeg is unavailable for camera frame review")
+        self._store = store
+        self._inspector = inspector
+        self._ffmpeg_path = resolved_ffmpeg
+        self._cache_root = cache_root.expanduser().absolute()
+        self._cache_root.mkdir(parents=True, exist_ok=True)
+        if self._cache_root.is_symlink() or not self._cache_root.is_dir():
+            raise SessionIntegrityError("camera frame cache root is invalid")
+        self._lock = threading.Lock()
+
+    def extract(
+        self,
+        session_id: str,
+        frame_index: int,
+        *,
+        expected_source_name: str = "sensor.camera.right",
+    ) -> RecordedCameraFrame:
+        if frame_index < 0:
+            raise SessionIntegrityError("camera frame index is invalid")
+        replay = self._store.prepare_replay(session_id, speed=1.0, loop=False)
+        matches = tuple(
+            artifact
+            for artifact in self._store.list_recorded_media(session_id)
+            if artifact.source_path.name == expected_source_name
+        )
+        if len(matches) != 1:
+            raise SessionIntegrityError("recorded camera source is unavailable")
+        artifact = matches[0]
+        manifest = self._inspector.inspect(artifact, replay)
+        epoch, sequence = _frame_location(manifest, frame_index)
+        cache_key = hashlib.sha256(
+            (
+                f"{manifest.generation_sha256}\0{artifact.artifact_id}\0"
+                f"{expected_source_name}\0{frame_index}\0jpeg-q2-v1"
+            ).encode()
+        ).hexdigest()
+        cache_path = self._cache_root / f"{cache_key}.jpg"
+
+        with self._lock:
+            cached = _read_cached_jpeg(cache_path)
+            if cached is not None:
+                return cached
+            frame = self._decode(manifest, epoch, sequence)
+            _publish_cached_jpeg(cache_path, frame.payload)
+            return frame
+
+    def _decode(
+        self,
+        manifest: RecordedMediaManifest,
+        epoch: RecordedMediaEpoch,
+        sequence: int,
+    ) -> RecordedCameraFrame:
+        target = self._inspector.get_segment(manifest, epoch.ordinal, sequence)
+        fragments: list[bytes] = [target.payload]
+        key_sequence = sequence
+        while not _fragment_is_sync(fragments[0]):
+            key_sequence -= 1
+            if key_sequence < 1 or sequence - key_sequence > _MAX_KEYFRAME_DISTANCE:
+                raise SessionIntegrityError("camera frame has no bounded sync fragment")
+            previous = self._inspector.get_segment(
+                manifest,
+                epoch.ordinal,
+                key_sequence,
+            )
+            fragments.insert(0, previous.payload)
+
+        init = self._inspector.get_init(manifest, epoch.ordinal)
+        select_index = sequence - key_sequence
+        try:
+            completed = subprocess.run(
+                [
+                    str(self._ffmpeg_path),
+                    "-hide_banner",
+                    "-loglevel",
+                    "error",
+                    "-f",
+                    "mp4",
+                    "-i",
+                    "pipe:0",
+                    "-vf",
+                    f"select=eq(n\\,{select_index})",
+                    "-fps_mode",
+                    "passthrough",
+                    "-frames:v",
+                    "1",
+                    "-f",
+                    "image2pipe",
+                    "-c:v",
+                    "mjpeg",
+                    "-q:v",
+                    "2",
+                    "pipe:1",
+                ],
+                input=b"".join((init.payload, *fragments)),
+                capture_output=True,
+                timeout=_FFMPEG_TIMEOUT_SECONDS,
+                check=False,
+            )
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            raise SessionIntegrityError("camera frame decoder failed") from exc
+        if completed.returncode != 0:
+            detail = completed.stderr.decode("utf-8", errors="replace").strip()[-500:]
+            raise SessionIntegrityError(f"camera frame decoder rejected sealed media: {detail}")
+        width, height = _jpeg_dimensions(completed.stdout)
+        digest = hashlib.sha256(completed.stdout).hexdigest()
+        return RecordedCameraFrame(
+            payload=completed.stdout,
+            media_type="image/jpeg",
+            width=width,
+            height=height,
+            sha256=digest,
+        )
+
+
+def _frame_location(
+    manifest: RecordedMediaManifest,
+    frame_index: int,
+) -> tuple[RecordedMediaEpoch, int]:
+    remaining = frame_index
+    for epoch in manifest.epochs:
+        if remaining < len(epoch.segments):
+            return epoch, remaining + 1
+        remaining -= len(epoch.segments)
+    raise SessionIntegrityError("camera frame is outside the recorded media manifest")
+
+
+def _fragment_is_sync(payload: bytes) -> bool:
+    tfhd_default_flags: int | None = None
+    sample_flags: int | None = None
+    sample_count: int | None = None
+    for box_type, body in _walk_boxes(payload):
+        if box_type == b"tfhd":
+            if len(body) < 8:
+                raise SessionIntegrityError("camera fragment tfhd is truncated")
+            flags = int.from_bytes(body[1:4], "big")
+            offset = 8
+            for mask, size in ((0x000001, 8), (0x000002, 4), (0x000008, 4), (0x000010, 4)):
+                if flags & mask:
+                    offset += size
+            if flags & 0x000020:
+                if offset + 4 > len(body):
+                    raise SessionIntegrityError("camera fragment default flags are truncated")
+                tfhd_default_flags = struct.unpack_from(">I", body, offset)[0]
+        elif box_type == b"trun":
+            if len(body) < 8:
+                raise SessionIntegrityError("camera fragment trun is truncated")
+            flags = int.from_bytes(body[1:4], "big")
+            sample_count = struct.unpack_from(">I", body, 4)[0]
+            if sample_count != 1:
+                raise SessionIntegrityError("camera fragment must contain exactly one sample")
+            offset = 8
+            if flags & 0x000001:
+                offset += 4
+            if flags & 0x000004:
+                if offset + 4 > len(body):
+                    raise SessionIntegrityError("camera fragment first flags are truncated")
+                sample_flags = struct.unpack_from(">I", body, offset)[0]
+                offset += 4
+            per_sample_sizes = (
+                (0x000100, 4),
+                (0x000200, 4),
+                (0x000400, 4),
+                (0x000800, 4),
+            )
+            for mask, size in per_sample_sizes:
+                if flags & mask:
+                    if offset + size > len(body):
+                        raise SessionIntegrityError("camera fragment sample data is truncated")
+                    if mask == 0x000400:
+                        sample_flags = struct.unpack_from(">I", body, offset)[0]
+                    offset += size
+    if sample_count != 1:
+        raise SessionIntegrityError("camera fragment has no unique video sample")
+    effective_flags = sample_flags if sample_flags is not None else tfhd_default_flags
+    if effective_flags is None:
+        raise SessionIntegrityError("camera fragment sample flags are unavailable")
+    return (effective_flags & 0x00010000) == 0
+
+
+def _walk_boxes(payload: bytes):
+    containers = {b"moof", b"traf"}
+    pending = [(0, len(payload))]
+    boxes = 0
+    while pending:
+        start, end = pending.pop()
+        offset = start
+        while offset + 8 <= end:
+            boxes += 1
+            if boxes > 64:
+                raise SessionIntegrityError("camera fragment box budget was exceeded")
+            size = struct.unpack_from(">I", payload, offset)[0]
+            box_type = payload[offset + 4 : offset + 8]
+            header = 8
+            if size == 1:
+                if offset + 16 > end:
+                    raise SessionIntegrityError("camera fragment extended box is truncated")
+                size = struct.unpack_from(">Q", payload, offset + 8)[0]
+                header = 16
+            elif size == 0:
+                size = end - offset
+            if size < header or offset + size > end:
+                raise SessionIntegrityError("camera fragment box size is invalid")
+            body_start = offset + header
+            body_end = offset + size
+            yield box_type, payload[body_start:body_end]
+            if box_type in containers:
+                pending.append((body_start, body_end))
+            offset = body_end
+        if offset != end:
+            raise SessionIntegrityError("camera fragment box boundary is invalid")
+
+
+def _jpeg_dimensions(payload: bytes) -> tuple[int, int]:
+    if len(payload) < 4 or payload[:2] != b"\xff\xd8" or payload[-2:] != b"\xff\xd9":
+        raise SessionIntegrityError("camera frame decoder returned an invalid JPEG")
+    offset = 2
+    while offset + 4 <= len(payload):
+        if payload[offset] != 0xFF:
+            offset += 1
+            continue
+        marker = payload[offset + 1]
+        offset += 2
+        if marker in {0xD8, 0xD9} or 0xD0 <= marker <= 0xD7:
+            continue
+        if offset + 2 > len(payload):
+            break
+        length = struct.unpack_from(">H", payload, offset)[0]
+        if length < 2 or offset + length > len(payload):
+            break
+        if marker in {0xC0, 0xC1, 0xC2} and length >= 7:
+            height, width = struct.unpack_from(">HH", payload, offset + 3)
+            if width > 0 and height > 0:
+                return width, height
+        offset += length
+    raise SessionIntegrityError("camera frame JPEG dimensions are unavailable")
+
+
+def _read_cached_jpeg(path: Path) -> RecordedCameraFrame | None:
+    try:
+        if path.is_symlink() or not path.is_file():
+            return None
+        payload = path.read_bytes()
+        width, height = _jpeg_dimensions(payload)
+    except (OSError, SessionIntegrityError):
+        return None
+    return RecordedCameraFrame(
+        payload=payload,
+        media_type="image/jpeg",
+        width=width,
+        height=height,
+        sha256=hashlib.sha256(payload).hexdigest(),
+    )
+
+
+def _publish_cached_jpeg(path: Path, payload: bytes) -> None:
+    temporary = path.parent / f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp"
+    try:
+        with temporary.open("xb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+    except OSError as exc:
+        raise SessionIntegrityError("camera frame cache could not be published") from exc
+    finally:
+        with suppress(FileNotFoundError):
+            temporary.unlink()

src/k1link/web/app.py

@@ -31,6 +31,7 @@ from k1link.laboratory import (
 )
 from k1link.sessions import (
     MaterializedRecording,
+    RecordedCameraFrameService,
     RecordedMediaInspector,
     RecordedMediaManifest,
     RecordingPreparationQueueFull,
@@ -191,6 +192,16 @@ session_recorded_media_inspector = RecordedMediaInspector(
 )
 _ffmpeg = _resolve_media_tool("ffmpeg")
 _ffprobe = _resolve_media_tool("ffprobe")
+session_recorded_camera_frame_service = (
+    RecordedCameraFrameService(
+        session_store,
+        session_recorded_media_inspector,
+        ffmpeg_path=_ffmpeg,
+        cache_root=session_store.data_dir / "camera-frame-cache",
+    )
+    if _ffmpeg is not None
+    else None
+)
 session_legacy_perception_overlay_store = (
     RecordedPerceptionOverlayStore(
         jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
@@ -586,9 +597,7 @@ app.include_router(
 app.include_router(
     build_advanced_laboratory_router(
         evidence_registry=LABORATORY_EVIDENCE_REGISTRY,
-        evidence_runtime_root_provider=lambda: (
-            REPOSITORY_ROOT / ".runtime" / "compute-experiments"
-        ),
+        evidence_runtime_root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "compute-experiments",
         e31_root_provider=lambda: (
             REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "e31" / "source-qualifications"
         ),
@@ -756,22 +765,19 @@ app.include_router(
 app.include_router(
     build_m4_threat_replay_router(
         root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "m4"
-            / "replay-threat"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "m4" / "replay-threat"
+        ),
+        camera_frame_provider=(
+            session_recorded_camera_frame_service.extract
+            if session_recorded_camera_frame_service is not None
+            else None
         ),
     )
 )
 app.include_router(
     build_e46e_ready_stack_router(
         root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "e46e"
-            / "ready-stack-results"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "e46e" / "ready-stack-results"
         ),
     )
 )
@@ -785,11 +791,7 @@ app.include_router(
             / "dashcam-bakeoff-results"
         ),
         e46e_root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "e46e"
-            / "ready-stack-results"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "e46e" / "ready-stack-results"
         ),
     )
 )

src/k1link/web/m4_threat_replay_api.py

@@ -10,7 +10,7 @@ from functools import lru_cache
 from pathlib import Path
 from typing import Final
 
-from fastapi import APIRouter, HTTPException, Query
+from fastapi import APIRouter, HTTPException, Query, Response
 
 from k1link.perception.threat_replay import (
     THREAT_REPLAY_FRAME_SCHEMA,
@@ -22,20 +22,21 @@ from k1link.perception.threat_replay import (
     ThreatReplayResult,
     read_threat_replay_result,
 )
+from k1link.sessions import RecordedCameraFrame, SessionIntegrityError
 
 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"
-)
+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]
+CameraFrameProvider = Callable[[str, int], RecordedCameraFrame]
 
 
 def build_m4_threat_replay_router(
     *,
     root_provider: RootProvider = lambda: None,
+    camera_frame_provider: CameraFrameProvider | None = None,
 ) -> APIRouter:
     router = APIRouter(prefix="/api/v1/laboratory/m4-threat", tags=["laboratory"])
 
@@ -109,10 +110,52 @@ def build_m4_threat_replay_router(
             **copy.deepcopy(frames[ordinal - 1]),
             "result_id": result_id,
             "ordinal": ordinal,
+            "camera_url": (
+                f"/api/v1/laboratory/m4-threat/results/{result_id}/visuals/{ordinal}/camera"
+            ),
             "ground_truth": False,
             "access": "read-only-replay-simulated",
         }
 
+    @router.get("/results/{result_id}/visuals/{ordinal}/camera")
+    def get_visual_camera(result_id: str, ordinal: int) -> Response:
+        if camera_frame_provider is None:
+            raise HTTPException(status_code=503, detail="M4.6 camera decoder недоступен")
+        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 не найден")
+        identity = frozen.manifest.get("identity")
+        if not isinstance(identity, dict):
+            raise HTTPException(status_code=404, detail="M4.6 source identity не найдена")
+        session_id = identity.get("source_session_id")
+        sequence = frames[ordinal - 1].get("sequence")
+        if not isinstance(session_id, str) or not isinstance(sequence, int):
+            raise HTTPException(status_code=404, detail="M4.6 camera identity не найдена")
+        try:
+            camera = camera_frame_provider(session_id, sequence)
+        except (OSError, SessionIntegrityError, ValueError):
+            raise HTTPException(
+                status_code=503,
+                detail="M4.6 exact camera frame недоступен",
+            ) from None
+        if camera.width != 800 or camera.height != 600:
+            raise HTTPException(
+                status_code=503,
+                detail="M4.6 camera frame нарушил размерный контракт",
+            )
+        return Response(
+            content=camera.payload,
+            media_type=camera.media_type,
+            headers={
+                "Cache-Control": "private, max-age=31536000, immutable",
+                "ETag": f'"{camera.sha256}"',
+                "X-Content-Type-Options": "nosniff",
+            },
+        )
+
     @router.get("/results/{result_id}/video-overlay")
     def get_video_overlay(result_id: str) -> dict[str, object]:
         frozen = result(result_id)
@@ -160,9 +203,7 @@ def _cached_video_overlay(
         frames.append(
             {
                 "frame_index": expected_sequence,
-                "session_seconds": _nonnegative_int(
-                    row.get("source_time_ns"), "source time"
-                )
+                "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"]),
@@ -209,9 +250,7 @@ def _project_result(result: ThreatReplayResult) -> dict[str, object]:
         },
         "metrics": copy.deepcopy(result.metrics),
         "configuration": copy.deepcopy(result.report["configuration"]),
-        "acceptance_requirements": copy.deepcopy(
-            result.report["acceptance_requirements"]
-        ),
+        "acceptance_requirements": copy.deepcopy(result.report["acceptance_requirements"]),
         "limitations": copy.deepcopy(result.report["limitations"]),
         "accepted": result.accepted,
         "ground_truth": False,
@@ -271,9 +310,7 @@ def _candidates(provider: RootProvider) -> list[Path]:
         (
             item
             for item in root.iterdir()
-            if item.is_dir()
-            and not item.is_symlink()
-            and _RESULT_ID.fullmatch(item.name)
+            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,