diff --git a/config/laboratories/m48r3-static-occupancy-shadow.json b/config/laboratories/m48r3-static-occupancy-shadow.json new file mode 100644 index 0000000..2902a4a --- /dev/null +++ b/config/laboratories/m48r3-static-occupancy-shadow.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "m48r3-static-occupancy-shadow", + "evidence": { + "runtime_relative_root": "m48/static-occupancy-shadow-results", + "result_id_prefix": "m48r3-static-occupancy-shadow", + "document_name": "manifest.json", + "schema_version": "missioncore.m48r3-static-occupancy-shadow-result/v1" + } +} diff --git a/experiments/perception/seal_m48r3_static_occupancy_shadow.py b/experiments/perception/seal_m48r3_static_occupancy_shadow.py new file mode 100644 index 0000000..03d3fb1 --- /dev/null +++ b/experiments/perception/seal_m48r3_static_occupancy_shadow.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Seal a complete M4.8R3 Worker shadow against the accepted native baseline.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from k1link.laboratory.m48r3_static_occupancy_shadow import ( + M48R3StaticOccupancyShadowError, + build_m48r3_static_occupancy_shadow, +) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--repository-root", type=Path, required=True) + parser.add_argument("--profile", type=Path, required=True) + parser.add_argument("--baseline-result", type=Path, required=True) + parser.add_argument("--baseline-frames", type=Path, required=True) + parser.add_argument("--candidate-result", type=Path, required=True) + parser.add_argument("--candidate-frames", type=Path, required=True) + parser.add_argument("--m48r2-result-root", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + arguments = parser.parse_args() + try: + result = build_m48r3_static_occupancy_shadow( + repository_root=arguments.repository_root, + profile_path=arguments.profile, + baseline_result_path=arguments.baseline_result, + baseline_frames_path=arguments.baseline_frames, + candidate_result_path=arguments.candidate_result, + candidate_frames_path=arguments.candidate_frames, + m48r2_result_root=arguments.m48r2_result_root, + output_root=arguments.output_root, + ) + except (M48R3StaticOccupancyShadowError, OSError, ValueError) as exc: + parser.error(str(exc)) + print( + json.dumps( + { + "result_id": result.result_id, + "result_root": str(result.result_root), + "accepted": result.manifest["accepted"], + "gates": result.report["gates"], + }, + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/laboratory/m48r3_static_occupancy_shadow.py b/src/k1link/laboratory/m48r3_static_occupancy_shadow.py new file mode 100644 index 0000000..bb77c20 --- /dev/null +++ b/src/k1link/laboratory/m48r3_static_occupancy_shadow.py @@ -0,0 +1,845 @@ +"""Seal the full M4.8R3 occupied-only Worker shadow and its bounded diff.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import shutil +import uuid +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final + +import numpy as np + +from k1link.laboratory.m48_static_occupancy_qualification import ( + M48StaticOccupancyQualificationError, + read_m48_static_occupancy_qualification, +) +from k1link.perception.geometry import RecordedGeometryStore +from k1link.perception.geometry_math import project_map_points_kb4 +from k1link.perception.m48_low_step_occupancy import M48_LOW_STEP_PROFILE_SCHEMA + +M48R3_SHADOW_RESULT_SCHEMA: Final = ( + "missioncore.m48r3-static-occupancy-shadow-result/v1" +) +M48R3_SHADOW_REPORT_SCHEMA: Final = ( + "missioncore.m48r3-static-occupancy-shadow-report/v1" +) +M48R3_SHADOW_CASE_SCHEMA: Final = ( + "missioncore.m48r3-static-occupancy-shadow-case/v1" +) +M48R3_SHADOW_DIFF_SCHEMA: Final = ( + "missioncore.m48r3-static-occupancy-frame-diff/v1" +) +M48R3_SHADOW_PREFIX: Final = "m48r3-static-occupancy-shadow-" +FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" +WORKER_RESULT_SCHEMA: Final = "missioncore.m48s-reference-graph-shadow-load/v5" +EXPECTED_FRAMES: Final = 4_489 +VOXEL_SIZE_M: Final = 0.45 +_AUTHORITY: Final = { + "mode": "replay-simulated", + "physical_live": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + +Cell = tuple[int, int, int] + + +class M48R3StaticOccupancyShadowError(RuntimeError): + """The M4.8R3 Worker evidence or immutable result is invalid.""" + + +@dataclass(frozen=True, slots=True) +class M48R3StaticOccupancyShadowResult: + result_id: str + result_root: Path + manifest: dict[str, Any] + report: dict[str, Any] + cases: tuple[dict[str, Any], ...] + + +@dataclass(frozen=True, slots=True) +class LedgerComparison: + frame_count: int + baseline_cell_total: int + candidate_cell_total: int + added_cell_total: int + lost_cell_total: int + baseline_component_total: int + candidate_component_total: int + false_free_count: int + maximum_added_cells_per_frame: int + maximum_candidate_components_per_frame: int + mean_cell_growth_fraction: float + mean_component_growth_fraction: float + diff_rows: tuple[dict[str, Any], ...] + selected_candidate_rows: dict[int, dict[str, Any]] + + +def build_m48r3_static_occupancy_shadow( + *, + repository_root: Path, + profile_path: Path, + baseline_result_path: Path, + baseline_frames_path: Path, + candidate_result_path: Path, + candidate_frames_path: Path, + m48r2_result_root: Path, + output_root: Path, +) -> M48R3StaticOccupancyShadowResult: + """Compare full ledgers and publish one append-only M4.8R3 result.""" + + repository = repository_root.resolve(strict=True) + profile_bytes = profile_path.resolve(strict=True).read_bytes() + profile = _object(json.loads(profile_bytes), "M4.8R3 profile") + if profile.get("schema_version") != M48_LOW_STEP_PROFILE_SCHEMA: + raise M48R3StaticOccupancyShadowError("M4.8R3 profile changed") + profile_sha256 = hashlib.sha256(profile_bytes).hexdigest() + acceptance = _object(profile.get("acceptance"), "M4.8R3 acceptance") + source = _object(profile.get("source"), "M4.8R3 source") + if source.get("frame_count") != EXPECTED_FRAMES: + raise M48R3StaticOccupancyShadowError("M4.8R3 frame contract changed") + + baseline = _read_worker_result(baseline_result_path) + candidate = _read_worker_result(candidate_result_path) + _validate_worker_binding( + baseline, + baseline_frames_path, + expected_frames=EXPECTED_FRAMES, + additive_profile_sha256=None, + ) + _validate_worker_binding( + candidate, + candidate_frames_path, + expected_frames=EXPECTED_FRAMES, + additive_profile_sha256=profile_sha256, + ) + try: + m48r2 = read_m48_static_occupancy_qualification(m48r2_result_root) + except M48StaticOccupancyQualificationError as exc: + raise M48R3StaticOccupancyShadowError("M4.8R2 binding changed") from exc + if ( + m48r2.result_id != source.get("m48r2_result_id") + or _file_sha256(m48r2.result_root / "cases.jsonl") + != source.get("m48r2_cases_sha256") + ): + raise M48R3StaticOccupancyShadowError("M4.8R2 source changed") + + selected_sequences = {int(row["sequence"]) - 1 for row in m48r2.cases} + comparison = compare_m48r3_frame_ledgers( + baseline_frames_path, + candidate_frames_path, + selected_sequences=selected_sequences, + expected_frames=EXPECTED_FRAMES, + ) + store = RecordedGeometryStore.from_repository(repository) + cases = _evaluate_cases( + m48r2.cases, + comparison.selected_candidate_rows, + store, + ) + separation = _evaluate_separation(profile, cases) + baseline_metrics = _worker_metrics(baseline) + candidate_metrics = _worker_metrics(candidate) + provider = _provider_metrics(candidate) + critical = [row for row in cases if row["distance_band"] == "critical-near"] + near_recall = _rate(critical, "matched") + canonical_recall = ( + 1.0 + if comparison.lost_cell_total == 0 + and m48r2.report["metrics"]["canonical_engineering_recall"] == 1.0 + else 0.0 + ) + capacity_drop_count = ( + int(provider["geometry_failed_frames"]) + + int(provider["temporal_failed_frames"]) + + int(provider["rolling_capacity_evicted_cells"]) + ) + fps_regression = max( + 0.0, + (baseline_metrics["fps"] - candidate_metrics["fps"]) + / baseline_metrics["fps"], + ) + world_p95_delta = ( + candidate_metrics["world_p95_ms"] - baseline_metrics["world_p95_ms"] + ) + gates = { + "complete_frame_accounting": candidate_metrics["delivered_frames"] + == EXPECTED_FRAMES, + "minimum_effective_world_state_fps": candidate_metrics["fps"] + >= float(acceptance["minimum_effective_world_state_fps"]), + "maximum_world_state_completion_p95_ms": candidate_metrics[ + "world_p95_ms" + ] + <= float(acceptance["maximum_world_state_completion_p95_ms"]), + "maximum_geometry_stage_p95_ms": candidate_metrics["geometry_p95_ms"] + <= float(acceptance["maximum_geometry_stage_p95_ms"]), + "maximum_geometry_stage_p99_ms": candidate_metrics["geometry_p99_ms"] + <= float(acceptance["maximum_geometry_stage_p99_ms"]), + "maximum_fps_regression_fraction": fps_regression + <= float(acceptance["maximum_fps_regression_fraction_vs_native_baseline"]), + "maximum_world_state_p95_delta_ms": world_p95_delta + <= float(acceptance["maximum_world_state_p95_delta_ms_vs_native_baseline"]), + "maximum_component_mean_growth_fraction": comparison.mean_component_growth_fraction + <= float(acceptance["maximum_additive_component_mean_growth_fraction"]), + "maximum_cell_mean_growth_fraction": comparison.mean_cell_growth_fraction + <= float(acceptance["maximum_additive_cell_mean_growth_fraction"]), + "zero_capacity_drops": capacity_drop_count + <= int(acceptance["maximum_capacity_drop_count"]), + "zero_baseline_cell_loss": comparison.lost_cell_total == 0, + "critical_near_recall": near_recall + >= float(acceptance["minimum_critical_near_recall"]), + "canonical_engineering_recall": canonical_recall + >= float(acceptance["minimum_canonical_engineering_recall"]), + "zero_false_free": comparison.false_free_count + <= int(acceptance["maximum_false_free_count"]), + "separation_expectations": all(row["passed"] for row in separation), + } + accepted = all(gates.values()) + producer_sha256 = _file_sha256(Path(__file__).resolve()) + candidate_frame_sha256 = _file_sha256(candidate_frames_path) + baseline_frame_sha256 = _file_sha256(baseline_frames_path) + created_at = _worker_completed_at(candidate) + identity = { + "schema_version": M48R3_SHADOW_RESULT_SCHEMA, + "created_at_utc": created_at, + "profile_id": profile["profile_id"], + "profile_sha256": profile_sha256, + "producer_sha256": producer_sha256, + "baseline_result_sha256": _file_sha256(baseline_result_path), + "baseline_frames_sha256": baseline_frame_sha256, + "candidate_result_sha256": _file_sha256(candidate_result_path), + "candidate_frames_sha256": candidate_frame_sha256, + "m48r2_result_id": m48r2.result_id, + "authority": dict(_AUTHORITY), + } + result_id = M48R3_SHADOW_PREFIX + _canonical_sha256(identity) + metrics = { + "frames": { + "expected": EXPECTED_FRAMES, + "baseline_delivered": baseline_metrics["delivered_frames"], + "candidate_delivered": candidate_metrics["delivered_frames"], + }, + "performance": { + "baseline": baseline_metrics, + "candidate": candidate_metrics, + "fps_regression_fraction": round(fps_regression, 9), + "world_state_p95_delta_ms": round(world_p95_delta, 6), + }, + "occupancy": { + "baseline_cell_total": comparison.baseline_cell_total, + "candidate_cell_total": comparison.candidate_cell_total, + "added_cell_total": comparison.added_cell_total, + "lost_cell_total": comparison.lost_cell_total, + "baseline_component_total": comparison.baseline_component_total, + "candidate_component_total": comparison.candidate_component_total, + "mean_cell_growth_fraction": round( + comparison.mean_cell_growth_fraction, 9 + ), + "mean_component_growth_fraction": round( + comparison.mean_component_growth_fraction, 9 + ), + "maximum_added_cells_per_frame": comparison.maximum_added_cells_per_frame, + "maximum_candidate_components_per_frame": ( + comparison.maximum_candidate_components_per_frame + ), + }, + "provider": provider, + "assisted_anchors": { + "count": len(cases), + "critical_near_count": len(critical), + "critical_near_recall": near_recall, + "matched_count": sum(bool(row["matched"]) for row in cases), + "canonical_engineering_recall": canonical_recall, + "separation": separation, + }, + "capacity_drop_count": capacity_drop_count, + "false_free_count": comparison.false_free_count, + } + report = { + "schema_version": M48R3_SHADOW_REPORT_SCHEMA, + "result_id": result_id, + "accepted": accepted, + "profile": { + "id": profile["profile_id"], + "sha256": profile_sha256, + "componentization": profile["componentization"], + "acceptance": acceptance, + }, + "metrics": metrics, + "gates": gates, + "decision": { + "state": "accepted-bounded-worker-shadow" + if accepted + else "rejected-bounded-worker-shadow", + "candidate_accepted": accepted, + "production_accepted": False, + "next_action": "product-cutover-decision" if accepted else "reduce-load-or-coverage", + }, + "limitations": [ + "Operator-assisted rectangles are development anchors, not independent truth.", + "Separated occupied components do not assert passability for an unknown chassis.", + ( + "No free-space, navigation, command, actuation or collision-safety " + "authority is granted." + ), + ], + "authority": dict(_AUTHORITY), + } + destination = output_root.resolve(strict=False) / result_id + _publish( + destination, + identity=identity, + report=report, + cases=cases, + diff_rows=comparison.diff_rows, + candidate_result_path=candidate_result_path, + candidate_frames_path=candidate_frames_path, + ) + return read_m48r3_static_occupancy_shadow(destination) + + +def compare_m48r3_frame_ledgers( + baseline_frames_path: Path, + candidate_frames_path: Path, + *, + selected_sequences: set[int], + expected_frames: int, +) -> LedgerComparison: + """Stream two aligned ledgers and retain only a bounded cell-diff.""" + + diffs: list[dict[str, Any]] = [] + selected: dict[int, dict[str, Any]] = {} + baseline_cells_total = 0 + candidate_cells_total = 0 + baseline_components_total = 0 + candidate_components_total = 0 + added_total = 0 + lost_total = 0 + false_free = 0 + maximum_added = 0 + maximum_components = 0 + count = 0 + with baseline_frames_path.open("r", encoding="utf-8") as baseline_stream, ( + candidate_frames_path.open("r", encoding="utf-8") + ) as candidate_stream: + for baseline_line, candidate_line in zip( + baseline_stream, + candidate_stream, + strict=True, + ): + baseline = _frame_row(json.loads(baseline_line)) + candidate = _frame_row(json.loads(candidate_line)) + baseline_sequence = _sequence(baseline) + candidate_sequence = _sequence(candidate) + if baseline_sequence != candidate_sequence or baseline_sequence != count: + raise M48R3StaticOccupancyShadowError("frame ledgers are not aligned") + baseline_map = _obstacle_map(baseline) + candidate_map = _obstacle_map(candidate) + baseline_components = _active_components(baseline_map) + candidate_components = _active_components(candidate_map) + baseline_cells = _cell_union(baseline_components) + candidate_cells = _cell_union(candidate_components) + added = candidate_cells - baseline_cells + lost = baseline_cells - candidate_cells + provenance: dict[str, str] = {} + for component_id, cells in candidate_components: + if not cells.intersection(added): + continue + provenance[component_id] = ( + "mixed" if cells.intersection(baseline_cells) else "additive-low-step" + ) + diffs.append( + { + "schema_version": M48R3_SHADOW_DIFF_SCHEMA, + "sequence": baseline_sequence, + "baseline_cell_count": len(baseline_cells), + "candidate_cell_count": len(candidate_cells), + "added_cells": [list(cell) for cell in sorted(added)], + "lost_cell_count": len(lost), + "component_provenance": provenance, + } + ) + if baseline_sequence in selected_sequences: + selected[baseline_sequence] = candidate + baseline_cells_total += len(baseline_cells) + candidate_cells_total += len(candidate_cells) + baseline_components_total += len(baseline_components) + candidate_components_total += len(candidate_components) + added_total += len(added) + lost_total += len(lost) + false_free += candidate_map.get("free_space_claimed") is True + maximum_added = max(maximum_added, len(added)) + maximum_components = max(maximum_components, len(candidate_components)) + count += 1 + if count != expected_frames or set(selected) != selected_sequences: + raise M48R3StaticOccupancyShadowError("frame ledgers are incomplete") + return LedgerComparison( + frame_count=count, + baseline_cell_total=baseline_cells_total, + candidate_cell_total=candidate_cells_total, + added_cell_total=added_total, + lost_cell_total=lost_total, + baseline_component_total=baseline_components_total, + candidate_component_total=candidate_components_total, + false_free_count=false_free, + maximum_added_cells_per_frame=maximum_added, + maximum_candidate_components_per_frame=maximum_components, + mean_cell_growth_fraction=_growth(candidate_cells_total, baseline_cells_total), + mean_component_growth_fraction=_growth( + candidate_components_total, + baseline_components_total, + ), + diff_rows=tuple(diffs), + selected_candidate_rows=selected, + ) + + +def read_m48r3_static_occupancy_shadow( + result_root: Path, +) -> M48R3StaticOccupancyShadowResult: + root = result_root.resolve(strict=True) + if result_root.is_symlink() or not root.name.startswith(M48R3_SHADOW_PREFIX): + raise M48R3StaticOccupancyShadowError("M4.8R3 result root is invalid") + manifest = _read_json(root / "manifest.json", maximum=4 * 1024 * 1024) + report = _read_json(root / "report.json", maximum=8 * 1024 * 1024) + cases = tuple(_read_jsonl(root / "cases.jsonl", maximum=8 * 1024 * 1024)) + if ( + manifest.get("schema_version") != M48R3_SHADOW_RESULT_SCHEMA + or manifest.get("result_id") != root.name + or report.get("schema_version") != M48R3_SHADOW_REPORT_SCHEMA + or report.get("result_id") != root.name + or manifest.get("authority") != _AUTHORITY + ): + raise M48R3StaticOccupancyShadowError("M4.8R3 result identity changed") + identity = _object(manifest.get("identity"), "M4.8R3 identity") + if ( + root.name != M48R3_SHADOW_PREFIX + _canonical_sha256(identity) + or manifest.get("identity_sha256") != _canonical_sha256(identity) + ): + raise M48R3StaticOccupancyShadowError("M4.8R3 digest changed") + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, list): + raise M48R3StaticOccupancyShadowError("M4.8R3 artifacts changed") + for artifact in artifacts: + item = _object(artifact, "M4.8R3 artifact") + path = (root / str(item.get("path"))).resolve(strict=True) + if path.parent != root or path.is_symlink() or _file_sha256(path) != item.get("sha256"): + raise M48R3StaticOccupancyShadowError("M4.8R3 artifact changed") + return M48R3StaticOccupancyShadowResult(root.name, root, manifest, report, cases) + + +def _evaluate_cases( + source_cases: Iterable[dict[str, Any]], + candidate_rows: dict[int, dict[str, Any]], + store: RecordedGeometryStore, +) -> tuple[dict[str, Any], ...]: + result: list[dict[str, Any]] = [] + for source in source_cases: + display_frame = int(source["sequence"]) + sequence = display_frame - 1 + frame = store.frame_for_index(sequence) + if frame is None: + raise M48R3StaticOccupancyShadowError("anchor frame is unavailable") + extent = source.get("extent_xyxy") + if not isinstance(extent, list) or len(extent) != 4: + raise M48R3StaticOccupancyShadowError("anchor extent changed") + bbox = ( + float(extent[0]) * frame.projection.width, + float(extent[1]) * frame.projection.height, + float(extent[2]) * frame.projection.width, + float(extent[3]) * frame.projection.height, + ) + matches = _component_matches(candidate_rows[sequence], frame, bbox) + result.append( + { + "schema_version": M48R3_SHADOW_CASE_SCHEMA, + "anchor_id": source["anchor_id"], + "display_frame": display_frame, + "source_sequence": sequence, + "extent_xyxy": extent, + "distance_band": source["distance_band"], + "component_count": len(matches), + "components": matches, + "matched": bool(matches), + "authority": "operator-assisted-development-anchor-not-truth", + } + ) + return tuple(result) + + +def _evaluate_separation( + profile: dict[str, Any], + cases: tuple[dict[str, Any], ...], +) -> list[dict[str, Any]]: + by_anchor = {str(row["anchor_id"]): row for row in cases} + result: list[dict[str, Any]] = [] + for value in profile.get("separation_expectations", []): + item = _object(value, "separation expectation") + anchor_id = str(item["anchor_id"]) + case = by_anchor.get(anchor_id) + if case is None or case["display_frame"] != item["sequence"]: + raise M48R3StaticOccupancyShadowError("separation anchor changed") + expected = int(item["expected_minimum_components"]) + observed = int(case["component_count"]) + result.append( + { + "anchor_id": anchor_id, + "display_frame": case["display_frame"], + "source_sequence": case["source_sequence"], + "expected_minimum_components": expected, + "observed_components": observed, + "passed": observed >= expected, + "interpretation": item["interpretation"], + } + ) + return result + + +def _component_matches( + graph_row: dict[str, Any], + frame: Any, + bbox: tuple[float, float, float, float], +) -> list[dict[str, Any]]: + projected_rows: list[dict[str, Any]] = [] + for component_id, cells in _active_components(_obstacle_map(graph_row)): + if not cells: + continue + points = np.asarray( + [ + ( + (cell[0] + 0.5) * VOXEL_SIZE_M, + (cell[1] + 0.5) * VOXEL_SIZE_M, + (cell[2] + 0.5) * VOXEL_SIZE_M, + ) + for cell in cells + ], + dtype=np.float64, + ) + projected = project_map_points_kb4( + points, + position_map_xyz=frame.sensor_position_map, + orientation_map_from_lidar_xyzw=frame.sensor_orientation_xyzw, + profile=frame.projection, + ) + inside = ( + (projected.pixels_xy[:, 0] >= bbox[0]) + & (projected.pixels_xy[:, 0] <= bbox[2]) + & (projected.pixels_xy[:, 1] >= bbox[1]) + & (projected.pixels_xy[:, 1] <= bbox[3]) + ) + if not np.any(inside): + continue + depths = projected.depths_m[inside] + projected_rows.append( + { + "component_id": component_id, + "projected_cell_count": int(depths.size), + "nearest_depth_m": round(float(np.min(depths)), 6), + } + ) + return sorted(projected_rows, key=lambda row: (row["nearest_depth_m"], row["component_id"])) + + +def _worker_metrics(result: dict[str, Any]) -> dict[str, Any]: + execution = _object(result.get("execution"), "worker execution") + timing = _object( + _object(result.get("metrics"), "worker metrics").get("pipeline_timing"), + "pipeline timing", + ) + geometry = _object( + _object(timing.get("provider_ms"), "provider timing").get("geometry"), + "geometry timing", + ) + world = _object( + _object(result.get("metrics"), "worker metrics").get( + "world_state_completion_age_ms" + ), + "world timing", + ) + return { + "admitted_frames": int(execution["admitted_frames"]), + "delivered_frames": int(execution["delivered_world_states"]), + "fps": float(execution["effective_world_state_fps"]), + "world_p95_ms": float(world["p95"]), + "world_p99_ms": float(world["p99"]), + "geometry_p95_ms": float(geometry["p95"]), + "geometry_p99_ms": float(geometry["p99"]), + } + + +def _provider_metrics(result: dict[str, Any]) -> dict[str, Any]: + loops = _object(result["execution"]["loops"][0], "worker loop") + providers = _object(loops.get("providers"), "providers") + geometry = _object(providers.get("geometry"), "geometry provider") + temporal = _object(providers.get("temporal"), "temporal provider") + rolling = _object(providers.get("rolling"), "rolling provider") + additive_duration_ns = int(geometry["additive_core_duration_ns"]) + completed = int(geometry["completed_frames"]) + return { + "additive_observation_count": int(geometry["additive_observation_count"]), + "additive_voxel_count": int(geometry["additive_voxel_count"]), + "frames_with_additions": int(geometry["frames_with_additions"]), + "peak_additive_observations_per_frame": int( + geometry["peak_additive_observations_per_frame"] + ), + "peak_candidate_points_per_frame": int(geometry["peak_candidate_points_per_frame"]), + "peak_voxels_per_component": int(geometry["peak_voxels_per_component"]), + "additive_mean_ms_per_frame": round( + additive_duration_ns / max(1, completed) / 1_000_000, + 6, + ), + "geometry_failed_frames": int(geometry["failed_frames"]), + "temporal_failed_frames": int(temporal["failed_frames"]), + "rolling_capacity_evicted_cells": int(rolling["capacity_evicted_cells"]), + "peak_temporal_components": int(temporal["peak_active_components"]), + "peak_rolling_cells": int(rolling["peak_active_cells"]), + } + + +def _validate_worker_binding( + result: dict[str, Any], + frames_path: Path, + *, + expected_frames: int, + additive_profile_sha256: str | None, +) -> None: + execution = _object(result.get("execution"), "worker execution") + evidence = _object(execution.get("frame_evidence"), "frame evidence") + identity = _object(result.get("identity"), "worker identity") + inputs = _object(identity.get("inputs"), "worker inputs") + if ( + result.get("schema_version") != WORKER_RESULT_SCHEMA + or result.get("completed") is not True + or execution.get("admitted_frames") != expected_frames + or evidence.get("schema_version") != FRAME_EVIDENCE_SCHEMA + or evidence.get("sha256") != _file_sha256(frames_path) + or evidence.get("row_count") != execution.get("delivered_world_states") + or identity.get("worker_id") != "worker-006" + or result.get("authority") + != { + "actuation_allowed": False, + "candidate_accepted": False, + "commands_enabled": False, + "ground_truth": False, + "navigation_or_safety_accepted": False, + } + ): + raise M48R3StaticOccupancyShadowError("worker result binding changed") + if additive_profile_sha256 is None: + if "additive_low_step_profile" in inputs: + raise M48R3StaticOccupancyShadowError("baseline is not native-only") + elif inputs.get("additive_low_step_profile") != additive_profile_sha256: + raise M48R3StaticOccupancyShadowError("candidate profile binding changed") + + +def _read_worker_result(path: Path) -> dict[str, Any]: + return _read_json(path, maximum=128 * 1024 * 1024) + + +def _frame_row(value: object) -> dict[str, Any]: + row = _object(value, "frame row") + if row.get("schema_version") != FRAME_EVIDENCE_SCHEMA: + raise M48R3StaticOccupancyShadowError("frame schema changed") + return row + + +def _sequence(row: dict[str, Any]) -> int: + envelope = _object(row.get("source_envelope"), "source envelope") + value = envelope.get("sequence") + if not isinstance(value, int) or isinstance(value, bool): + raise M48R3StaticOccupancyShadowError("frame sequence changed") + return value + + +def _obstacle_map(row: dict[str, Any]) -> dict[str, Any]: + delivery = _object(row.get("delivery"), "delivery") + return _object(delivery.get("obstacle_map"), "obstacle map") + + +def _active_components(obstacle_map: dict[str, Any]) -> list[tuple[str, set[Cell]]]: + result: list[tuple[str, set[Cell]]] = [] + for collection in (obstacle_map.get("occupied"), obstacle_map.get("unknown")): + if not isinstance(collection, list): + raise M48R3StaticOccupancyShadowError("obstacle collection changed") + for value in collection: + item = _object(value, "obstacle") + component_id = item.get("component_id") + cells = item.get("cells") + if not isinstance(component_id, str) or not isinstance(cells, list): + raise M48R3StaticOccupancyShadowError("obstacle component changed") + parsed: set[Cell] = set() + for cell in cells: + row = _object(cell, "obstacle cell") + x, y, z = row.get("x"), row.get("y"), row.get("z") + if any( + not isinstance(item, int) or isinstance(item, bool) + for item in (x, y, z) + ): + raise M48R3StaticOccupancyShadowError("obstacle cell changed") + assert isinstance(x, int) and isinstance(y, int) and isinstance(z, int) + parsed.add((x, y, z)) + if parsed: + result.append((component_id, parsed)) + return result + + +def _cell_union(components: Iterable[tuple[str, set[Cell]]]) -> set[Cell]: + result: set[Cell] = set() + for _, cells in components: + result.update(cells) + return result + + +def _growth(candidate: int, baseline: int) -> float: + if baseline <= 0: + return math.inf if candidate > 0 else 0.0 + return max(0.0, (candidate - baseline) / baseline) + + +def _rate(rows: list[dict[str, Any]], key: str) -> float: + return sum(bool(row[key]) for row in rows) / len(rows) if rows else 0.0 + + +def _worker_completed_at(result: dict[str, Any]) -> str: + value = result.get("completed_utc_ns") + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise M48R3StaticOccupancyShadowError("worker completion time changed") + return datetime.fromtimestamp(value / 1_000_000_000, UTC).isoformat().replace( + "+00:00", "Z" + ) + + +def _publish( + destination: Path, + *, + identity: dict[str, Any], + report: dict[str, Any], + cases: tuple[dict[str, Any], ...], + diff_rows: tuple[dict[str, Any], ...], + candidate_result_path: Path, + candidate_frames_path: Path, +) -> None: + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = destination.parent / f".{destination.name}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700) + try: + _write_json(staging / "report.json", report) + _write_jsonl(staging / "cases.jsonl", cases) + _write_jsonl(staging / "frame-diff.jsonl", diff_rows) + shutil.copyfile(candidate_result_path, staging / "worker-result.json") + shutil.copyfile(candidate_frames_path, staging / "frames.jsonl") + artifacts = [ + _artifact(staging / name, role) + for name, role in ( + ("report.json", "m48r3-shadow-report"), + ("cases.jsonl", "operator-assisted-anchor-projection"), + ("frame-diff.jsonl", "baseline-versus-candidate-occupied-diff"), + ("worker-result.json", "worker-load-result"), + ("frames.jsonl", "candidate-frame-ledger"), + ) + ] + manifest = { + "schema_version": M48R3_SHADOW_RESULT_SCHEMA, + "result_id": destination.name, + "identity_sha256": _canonical_sha256(identity), + "identity": identity, + "created_at_utc": identity["created_at_utc"], + "accepted": report["accepted"], + "ground_truth": False, + "authority": dict(_AUTHORITY), + "artifacts": artifacts, + } + _write_json(staging / "manifest.json", manifest) + if destination.exists(): + raise M48R3StaticOccupancyShadowError("immutable result already exists") + os.replace(staging, destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + + +def _artifact(path: Path, role: str) -> dict[str, Any]: + return { + "path": path.name, + "role": role, + "byte_length": path.stat().st_size, + "sha256": _file_sha256(path), + "media_type": "application/x-ndjson" if path.suffix == ".jsonl" else "application/json", + } + + +def _read_json(path: Path, *, maximum: int) -> dict[str, Any]: + if path.is_symlink() or not path.is_file() or path.stat().st_size > maximum: + raise M48R3StaticOccupancyShadowError(f"{path.name} is unavailable") + return _object(json.loads(path.read_text("utf-8")), path.name) + + +def _read_jsonl(path: Path, *, maximum: int) -> list[dict[str, Any]]: + if path.is_symlink() or not path.is_file() or path.stat().st_size > maximum: + raise M48R3StaticOccupancyShadowError(f"{path.name} is unavailable") + return [ + _object(json.loads(line), path.name) + for line in path.read_text("utf-8").splitlines() + if line.strip() + ] + + +def _write_json(path: Path, value: object) -> None: + path.write_text( + json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n", + "utf-8", + ) + + +def _write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None: + with path.open("w", encoding="utf-8") as stream: + for row in rows: + stream.write( + json.dumps( + row, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + ) + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _canonical_sha256(value: object) -> str: + return hashlib.sha256( + json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + ).hexdigest() + + +def _object(value: object, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise M48R3StaticOccupancyShadowError(f"{label} is invalid") + return value + + +__all__ = [ + "LedgerComparison", + "M48R3StaticOccupancyShadowError", + "M48R3StaticOccupancyShadowResult", + "build_m48r3_static_occupancy_shadow", + "compare_m48r3_frame_ledgers", + "read_m48r3_static_occupancy_shadow", +] diff --git a/src/k1link/perception/m48s_replay_timeline.py b/src/k1link/perception/m48s_replay_timeline.py index 2840a77..cbc28ae 100644 --- a/src/k1link/perception/m48s_replay_timeline.py +++ b/src/k1link/perception/m48s_replay_timeline.py @@ -70,6 +70,7 @@ class M48sReplayTimeline: frames_name: str = "reference-graph-replay-frames.jsonl", worker_result_name: str = "reference-graph-replay-worker-result.json", frame_evidence_schema: str = FRAME_EVIDENCE_SCHEMA, + frame_diff_name: str | None = None, camera_endpoint_root: str = ( "/api/v1/laboratory/m48s/fixed-class-detector" ), @@ -80,6 +81,10 @@ class M48sReplayTimeline: if ( Path(frames_name).name != frames_name or Path(worker_result_name).name != worker_result_name + or ( + frame_diff_name is not None + and Path(frame_diff_name).name != frame_diff_name + ) or frame_evidence_schema not in { FRAME_EVIDENCE_SCHEMA, M48R3_FRAME_EVIDENCE_SCHEMA, @@ -92,11 +97,23 @@ class M48sReplayTimeline: self.camera_endpoint_root = camera_endpoint_root self.frames_path = (self.result_root / frames_name).resolve(strict=True) self.worker_path = (self.result_root / worker_result_name).resolve(strict=True) + self.frame_diff_path = ( + None + if frame_diff_name is None + else (self.result_root / frame_diff_name).resolve(strict=True) + ) if ( self.frames_path.parent != self.result_root or self.worker_path.parent != self.result_root or self.frames_path.is_symlink() or self.worker_path.is_symlink() + or ( + self.frame_diff_path is not None + and ( + self.frame_diff_path.parent != self.result_root + or self.frame_diff_path.is_symlink() + ) + ) ): raise M48sReplayTimelineError("M4.8S replay artifacts are invalid") self.profile = load_replay_threat_profile( @@ -123,6 +140,11 @@ class M48sReplayTimeline: self.outcomes, frame_evidence_schema=self.frame_evidence_schema, ) + self.frame_diff_offsets = ( + {} + if self.frame_diff_path is None + else _index_frame_diff(self.frame_diff_path) + ) self._cache_lock = Lock() self._chunk_json_cache: OrderedDict[tuple[int, int], bytes] = OrderedDict() self._camera_point_json_cache: OrderedDict[int, bytes] = OrderedDict() @@ -156,6 +178,11 @@ class M48sReplayTimeline: "camera_point_window_seconds": CAMERA_ACCUMULATION_WINDOW_SECONDS, "camera_point_sample_limit": CAMERA_ACCUMULATION_POINT_LIMIT, "world_state_delivery": "source-paced-latest-wins", + "occupancy_provenance_delivery": ( + "baseline-versus-additive-component-diff" + if self.frame_diff_path is not None + else None + ), "world_state_frame_count": len(self.index.offsets_by_sequence), "superseded_frame_count": sum( value == "superseded" for value in self.outcomes.values() @@ -399,6 +426,12 @@ class M48sReplayTimeline: body_frame, occupied_voxel_size_m=self.profile.corridor.occupied_voxel_size_m, ) + provenance = self._component_provenance(sequence) + for visual in metric_visuals: + visual["occupancy_source"] = provenance.get( + str(visual["component_id"]), + "baseline", + ) associated = set(_strings(row.get("associated_proposal_ids"), "associated ids")) for proposal in _objects(row.get("detector_proposals"), "detector proposals"): proposal_id = _text(proposal.get("proposal_id"), "proposal id") @@ -480,6 +513,27 @@ class M48sReplayTimeline: raise M48sReplayTimelineError("M4.8S frame row is invalid") return value + def _component_provenance(self, sequence: int) -> dict[str, str]: + if self.frame_diff_path is None: + return {} + offset = self.frame_diff_offsets.get(sequence) + if offset is None: + raise M48sReplayTimelineError("M4.8R3 frame diff is incomplete") + with self.frame_diff_path.open("rb") as stream: + stream.seek(offset) + line = stream.readline() + value = _object(json.loads(line), "frame diff") + if value.get("sequence") != sequence: + raise M48sReplayTimelineError("M4.8R3 frame diff binding changed") + provenance = value.get("component_provenance") + if not isinstance(provenance, dict) or any( + not isinstance(key, str) + or item not in {"mixed", "additive-low-step"} + for key, item in provenance.items() + ): + raise M48sReplayTimelineError("M4.8R3 component provenance changed") + return provenance + def _index_ledger( path: Path, @@ -517,6 +571,33 @@ def _index_ledger( return _LedgerIndex(offsets) +def _index_frame_diff(path: Path) -> dict[int, int]: + offsets: dict[int, int] = {} + with path.open("rb") as stream: + while True: + offset = stream.tell() + line = stream.readline() + if not line: + break + try: + value = json.loads(line) + except json.JSONDecodeError: + raise M48sReplayTimelineError("M4.8R3 frame diff is invalid") from None + if not isinstance(value, dict): + raise M48sReplayTimelineError("M4.8R3 frame diff is invalid") + sequence = value.get("sequence") + if ( + not isinstance(sequence, int) + or isinstance(sequence, bool) + or sequence != len(offsets) + ): + raise M48sReplayTimelineError("M4.8R3 frame diff sequence changed") + offsets[sequence] = offset + if len(offsets) != EXPECTED_FRAME_COUNT: + raise M48sReplayTimelineError("M4.8R3 frame diff is incomplete") + return offsets + + def _ledger_source_envelope( line: bytes, *, diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 9a5c22b..a0e87c6 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -126,6 +126,9 @@ 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.m48_object_quality_api import build_m48_object_quality_router +from k1link.web.m48r3_static_occupancy_api import ( + build_m48r3_static_occupancy_router, +) from k1link.web.m48s_fixed_class_detector_lab_api import ( build_m48s_fixed_class_detector_lab_router, ) @@ -961,6 +964,23 @@ app.include_router( ), ) ) +app.include_router( + build_m48r3_static_occupancy_router( + root_provider=lambda: ( + REPOSITORY_ROOT + / ".runtime" + / "compute-experiments" + / "m48" + / "static-occupancy-shadow-results" + ), + repository_root_provider=lambda: REPOSITORY_ROOT, + camera_frame_provider=( + session_recorded_camera_frame_service.extract + if session_recorded_camera_frame_service is not None + else None + ), + ) +) app.include_router( build_m48s_fixed_class_detector_lab_router( root_provider=lambda: ( diff --git a/src/k1link/web/m48r3_static_occupancy_api.py b/src/k1link/web/m48r3_static_occupancy_api.py new file mode 100644 index 0000000..f84c090 --- /dev/null +++ b/src/k1link/web/m48r3_static_occupancy_api.py @@ -0,0 +1,290 @@ +"""Read-only API for sealed M4.8R3 static-occupancy Worker shadows.""" + +from __future__ import annotations + +import copy +import re +from collections.abc import Callable +from functools import lru_cache +from pathlib import Path +from typing import Final + +from fastapi import APIRouter, HTTPException, Query, Response + +from k1link.laboratory.m48r3_static_occupancy_shadow import ( + M48R3_SHADOW_PREFIX, + M48R3StaticOccupancyShadowError, + M48R3StaticOccupancyShadowResult, + read_m48r3_static_occupancy_shadow, +) +from k1link.perception.m48s_replay_timeline import ( + M48R3_FRAME_EVIDENCE_SCHEMA, + M48sReplayTimeline, + M48sReplayTimelineError, +) +from k1link.perception.threat_timeline import RECORDED_SPATIAL_MAX_CHUNK_FRAMES +from k1link.sessions import RecordedCameraFrame, SessionIntegrityError + +RootProvider = Callable[[], Path | None] +CameraFrameProvider = Callable[[str, int], RecordedCameraFrame] + +RESULT_ID: Final = re.compile(rf"^{re.escape(M48R3_SHADOW_PREFIX)}[a-f0-9]{{64}}$") +RESULT_VIEW_SCHEMA: Final = "missioncore.m48r3-static-occupancy-shadow-view/v1" +RESULT_CATALOG_SCHEMA: Final = "missioncore.m48r3-static-occupancy-shadow-catalog/v1" +CASE_CATALOG_SCHEMA: Final = "missioncore.m48r3-static-occupancy-shadow-cases/v1" +ENDPOINT_ROOT: Final = "/api/v1/laboratory/m48r3/static-occupancy" + + +def build_m48r3_static_occupancy_router( + *, + root_provider: RootProvider = lambda: None, + repository_root_provider: RootProvider = lambda: None, + camera_frame_provider: CameraFrameProvider | None = None, +) -> APIRouter: + router = APIRouter(prefix=ENDPOINT_ROOT, tags=["laboratory"]) + + def result(result_id: str) -> M48R3StaticOccupancyShadowResult: + candidate = _resolve_candidate(root_provider, result_id) + try: + return _read_result_cached(str(candidate), _result_signature(candidate)) + except (M48R3StaticOccupancyShadowError, OSError, ValueError): + raise HTTPException(status_code=404, detail="M4.8R3 result not found") from None + + def timeline(result_id: str) -> M48sReplayTimeline: + candidate = _resolve_candidate(root_provider, result_id) + repository = _configured_root(repository_root_provider) + if repository is None: + raise HTTPException(status_code=503, detail="M4.8R3 timeline source unavailable") + result(result_id) + try: + return _read_timeline_cached( + str(repository), + str(candidate), + result_id, + _timeline_signature(candidate), + ) + except (M48sReplayTimelineError, OSError, ValueError): + raise HTTPException( + status_code=503, + detail="M4.8R3 bounded timeline failed verification", + ) from None + + @router.get("/results") + def list_results(limit: int = Query(default=1, ge=1, le=10)) -> dict[str, object]: + root = _configured_root(root_provider) + if root is None: + return _empty_catalog(configured=False) + items: list[dict[str, object]] = [] + invalid_total = 0 + for candidate in sorted(root.iterdir()): + if not candidate.is_dir() or RESULT_ID.fullmatch(candidate.name) is None: + continue + try: + sealed = _read_result_cached(str(candidate), _result_signature(candidate)) + items.append(_project_result(sealed)) + except (M48R3StaticOccupancyShadowError, OSError, ValueError): + invalid_total += 1 + items.sort( + key=lambda item: (str(item["created_at_utc"]), str(item["result_id"])), + reverse=True, + ) + return { + "schema_version": RESULT_CATALOG_SCHEMA, + "configured": True, + "items": items[:limit], + "candidate_total": len(items) + invalid_total, + "invalid_total": invalid_total, + "access": "read-only", + } + + @router.get("/{result_id}") + def get_result(result_id: str) -> dict[str, object]: + return _project_result(result(result_id)) + + @router.get("/{result_id}/cases") + def get_cases(result_id: str) -> dict[str, object]: + sealed = result(result_id) + return { + "schema_version": CASE_CATALOG_SCHEMA, + "result_id": result_id, + "cases": copy.deepcopy(sealed.cases), + "case_count": len(sealed.cases), + "ground_truth": False, + "authority": "replay-simulated", + "access": "read-only", + } + + @router.get("/{result_id}/timeline") + def get_timeline(result_id: str) -> dict[str, object]: + return copy.deepcopy(timeline(result_id).metadata()) + + @router.get("/{result_id}/timeline/chunk") + def get_timeline_chunk( + result_id: str, + start: int = Query(default=0, ge=0), + count: int = Query(default=12, ge=1, le=RECORDED_SPATIAL_MAX_CHUNK_FRAMES), + ) -> Response: + try: + content = timeline(result_id).chunk_json( + start_sequence=start, + frame_count=count, + ) + except M48sReplayTimelineError: + raise HTTPException(status_code=404, detail="M4.8R3 chunk not found") from None + return _immutable_json(content) + + @router.get("/{result_id}/timeline/frames/{sequence}/camera-points") + def get_camera_points(result_id: str, sequence: int) -> Response: + try: + content = timeline(result_id).camera_point_overlay_json(sequence=sequence) + except M48sReplayTimelineError: + raise HTTPException( + status_code=404, + detail="M4.8R3 camera points not found", + ) from None + return _immutable_json(content) + + @router.get("/{result_id}/timeline/frames/{sequence}/camera") + def get_camera(result_id: str, sequence: int) -> Response: + if camera_frame_provider is None: + raise HTTPException(status_code=503, detail="M4.8R3 camera decoder unavailable") + projected = timeline(result_id) + if not 0 <= sequence < len(projected.source_times_ns): + raise HTTPException(status_code=404, detail="M4.8R3 frame not found") + try: + camera = camera_frame_provider(projected.profile.session_id, sequence) + except (OSError, SessionIntegrityError, ValueError): + raise HTTPException(status_code=503, detail="M4.8R3 camera unavailable") from None + if camera.width != 800 or camera.height != 600: + raise HTTPException(status_code=503, detail="M4.8R3 camera size changed") + 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", + }, + ) + + return router + + +@lru_cache(maxsize=2) +def _read_result_cached( + result_root: str, + signature: tuple[int, ...], +) -> M48R3StaticOccupancyShadowResult: + del signature + return read_m48r3_static_occupancy_shadow(Path(result_root)) + + +@lru_cache(maxsize=2) +def _read_timeline_cached( + repository_root: str, + result_root: str, + result_id: str, + signature: tuple[int, ...], +) -> M48sReplayTimeline: + del signature + return M48sReplayTimeline( + repository_root=Path(repository_root), + result_root=Path(result_root), + result_id=result_id, + frames_name="frames.jsonl", + worker_result_name="worker-result.json", + frame_evidence_schema=M48R3_FRAME_EVIDENCE_SCHEMA, + frame_diff_name="frame-diff.jsonl", + camera_endpoint_root=ENDPOINT_ROOT, + ) + + +def _project_result(result: M48R3StaticOccupancyShadowResult) -> dict[str, object]: + return { + "schema_version": RESULT_VIEW_SCHEMA, + "result_id": result.result_id, + "created_at_utc": result.manifest["created_at_utc"], + "accepted": result.manifest["accepted"], + "profile": copy.deepcopy(result.report["profile"]), + "metrics": copy.deepcopy(result.report["metrics"]), + "gates": copy.deepcopy(result.report["gates"]), + "decision": copy.deepcopy(result.report["decision"]), + "limitations": copy.deepcopy(result.report["limitations"]), + "ground_truth": False, + "authority": copy.deepcopy(result.report["authority"]), + "access": "read-only", + } + + +def _resolve_candidate(provider: RootProvider, result_id: str) -> Path: + if RESULT_ID.fullmatch(result_id) is None: + raise HTTPException(status_code=404, detail="M4.8R3 result not found") + root = _configured_root(provider) + if root is None: + raise HTTPException(status_code=404, detail="M4.8R3 result not found") + candidate = root / result_id + if candidate.is_symlink() or not candidate.is_dir(): + raise HTTPException(status_code=404, detail="M4.8R3 result not found") + resolved = candidate.resolve(strict=True) + if resolved.parent != root: + raise HTTPException(status_code=404, detail="M4.8R3 result not found") + return resolved + + +def _configured_root(provider: RootProvider) -> Path | None: + value = provider() + if value is None: + return None + if value.is_symlink() or not value.is_dir(): + return None + return value.resolve(strict=True) + + +def _result_signature(candidate: Path) -> tuple[int, ...]: + return _signature( + candidate, + ("manifest.json", "report.json", "cases.jsonl", "worker-result.json", "frames.jsonl"), + ) + + +def _timeline_signature(candidate: Path) -> tuple[int, ...]: + return _signature( + candidate, + ("worker-result.json", "frames.jsonl", "frame-diff.jsonl"), + ) + + +def _signature(candidate: Path, names: tuple[str, ...]) -> tuple[int, ...]: + result: list[int] = [] + for name in names: + path = candidate / name + if path.is_symlink() or not path.is_file(): + raise ValueError("M4.8R3 artifact unavailable") + stat = path.stat() + result.extend((stat.st_size, stat.st_mtime_ns)) + return tuple(result) + + +def _immutable_json(content: bytes) -> Response: + return Response( + content=content, + media_type="application/json", + headers={ + "Cache-Control": "private, max-age=31536000, immutable", + "X-Content-Type-Options": "nosniff", + }, + ) + + +def _empty_catalog(*, configured: bool) -> dict[str, object]: + return { + "schema_version": RESULT_CATALOG_SCHEMA, + "configured": configured, + "items": [], + "candidate_total": 0, + "invalid_total": 0, + "access": "read-only", + } + + +__all__ = ["build_m48r3_static_occupancy_router"] diff --git a/tests/test_m48r3_static_occupancy_api.py b/tests/test_m48r3_static_occupancy_api.py new file mode 100644 index 0000000..736d072 --- /dev/null +++ b/tests/test_m48r3_static_occupancy_api.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from k1link.laboratory.m48r3_static_occupancy_shadow import ( + M48R3StaticOccupancyShadowResult, +) +from k1link.web import m48r3_static_occupancy_api as api + + +def test_m48r3_api_projects_result_cases_and_provenance_timeline( + tmp_path: Path, + monkeypatch, +) -> None: + result_id = "m48r3-static-occupancy-shadow-" + "0" * 64 + root = tmp_path / "results" + result_root = root / result_id + result_root.mkdir(parents=True) + for name in ( + "manifest.json", + "report.json", + "cases.jsonl", + "worker-result.json", + "frames.jsonl", + "frame-diff.jsonl", + ): + (result_root / name).write_text("{}\n", encoding="utf-8") + sealed = M48R3StaticOccupancyShadowResult( + result_id=result_id, + result_root=result_root, + manifest={"created_at_utc": "2026-08-26T00:00:00Z", "accepted": True}, + report={ + "profile": {"id": "profile"}, + "metrics": {"frames": {"candidate_delivered": 4489}}, + "gates": {"complete_frame_accounting": True}, + "decision": {"state": "accepted-bounded-worker-shadow"}, + "limitations": ["replay only"], + "authority": {"mode": "replay-simulated"}, + }, + cases=({"anchor_id": "anchor", "source_sequence": 1855},), + ) + + class Timeline: + source_times_ns = tuple(range(4489)) + profile = SimpleNamespace(session_id="20260720T065719Z_viewer_live") + + def metadata(self) -> dict[str, object]: + return { + "schema_version": "missioncore.recorded-spatial-evidence-timeline/v1", + "result_id": result_id, + "occupancy_provenance_delivery": ( + "baseline-versus-additive-component-diff" + ), + } + + def chunk_json(self, *, start_sequence: int, frame_count: int) -> bytes: + return json.dumps( + {"start_sequence": start_sequence, "frame_count": frame_count} + ).encode() + + def camera_point_overlay_json(self, *, sequence: int) -> bytes: + return json.dumps({"sequence": sequence}).encode() + + monkeypatch.setattr(api, "_read_result_cached", lambda *_args: sealed) + monkeypatch.setattr(api, "_read_timeline_cached", lambda *_args: Timeline()) + app = FastAPI() + app.include_router( + api.build_m48r3_static_occupancy_router( + root_provider=lambda: root, + repository_root_provider=lambda: tmp_path, + ) + ) + client = TestClient(app) + + catalog = client.get("/api/v1/laboratory/m48r3/static-occupancy/results") + assert catalog.status_code == 200 + assert catalog.json()["items"][0]["result_id"] == result_id + result = client.get(f"/api/v1/laboratory/m48r3/static-occupancy/{result_id}") + assert result.status_code == 200 + assert result.json()["accepted"] is True + cases = client.get(f"/api/v1/laboratory/m48r3/static-occupancy/{result_id}/cases") + assert cases.status_code == 200 + assert cases.json()["cases"][0]["source_sequence"] == 1855 + timeline = client.get( + f"/api/v1/laboratory/m48r3/static-occupancy/{result_id}/timeline" + ) + assert timeline.status_code == 200 + assert timeline.json()["occupancy_provenance_delivery"] == ( + "baseline-versus-additive-component-diff" + ) + chunk = client.get( + f"/api/v1/laboratory/m48r3/static-occupancy/{result_id}/timeline/chunk", + params={"start": 1855, "count": 1}, + ) + assert chunk.status_code == 200 + assert chunk.json() == {"start_sequence": 1855, "frame_count": 1} + assert ( + client.get("/api/v1/laboratory/m48r3/static-occupancy/not-a-result").status_code + == 404 + ) diff --git a/tests/test_m48r3_static_occupancy_shadow.py b/tests/test_m48r3_static_occupancy_shadow.py new file mode 100644 index 0000000..547371c --- /dev/null +++ b/tests/test_m48r3_static_occupancy_shadow.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from k1link.laboratory.m48r3_static_occupancy_shadow import ( + FRAME_EVIDENCE_SCHEMA, + compare_m48r3_frame_ledgers, +) + + +def _component(component_id: str, cells: list[tuple[int, int, int]]) -> dict[str, object]: + return { + "component_id": component_id, + "cells": [{"x": x, "y": y, "z": z} for x, y, z in cells], + } + + +def _row( + sequence: int, + occupied: list[dict[str, object]], + *, + unknown: list[dict[str, object]] | None = None, +) -> dict[str, object]: + return { + "schema_version": FRAME_EVIDENCE_SCHEMA, + "source_envelope": {"sequence": sequence}, + "delivery": { + "obstacle_map": { + "occupied": occupied, + "unknown": unknown or [], + "free_space_claimed": False, + } + }, + } + + +def _write(path: Path, rows: list[dict[str, object]]) -> None: + path.write_text( + "".join( + json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n" + for row in rows + ), + "utf-8", + ) + + +def test_streaming_diff_preserves_gaps_and_marks_additive_components(tmp_path: Path) -> None: + baseline = tmp_path / "baseline.jsonl" + candidate = tmp_path / "candidate.jsonl" + _write( + baseline, + [ + _row(0, [_component("base-0", [(0, 0, 0)])]), + _row(1, [_component("base-1", [(10, 0, 0)])]), + ], + ) + _write( + candidate, + [ + _row(0, [_component("mixed-0", [(0, 0, 0), (1, 0, 0)])]), + _row( + 1, + [_component("base-1", [(10, 0, 0)])], + unknown=[_component("step-1", [(20, 0, 0)])], + ), + ], + ) + + result = compare_m48r3_frame_ledgers( + baseline, + candidate, + selected_sequences={1}, + expected_frames=2, + ) + + assert result.frame_count == 2 + assert result.baseline_cell_total == 2 + assert result.candidate_cell_total == 4 + assert result.added_cell_total == 2 + assert result.lost_cell_total == 0 + assert result.mean_cell_growth_fraction == 1.0 + assert result.mean_component_growth_fraction == 0.5 + assert result.diff_rows[0]["component_provenance"] == {"mixed-0": "mixed"} + assert result.diff_rows[1]["component_provenance"] == { + "step-1": "additive-low-step" + } + assert result.diff_rows[0]["added_cells"] == [[1, 0, 0]] + assert set(result.selected_candidate_rows) == {1} +