From 3c81c39a1c2c1979a0f38bfcf526196637efc0ad Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 26 Aug 2026 11:51:56 +0300 Subject: [PATCH] feat(perception): qualify conservative static occupancy --- .../m48-static-occupancy-qualification.json | 10 + config/laboratory-execution.json | 20 + ...m48-static-occupancy-qualification-v1.json | 56 ++ .../run_m48_static_occupancy_qualification.py | 81 ++ src/k1link/laboratory/execution.py | 24 + .../m48_static_occupancy_qualification.py | 754 ++++++++++++++++++ src/k1link/perception/geometry.py | 22 + src/k1link/web/app.py | 7 + src/k1link/web/m48_object_quality_api.py | 86 ++ tests/test_geometry_association_provider.py | 9 + tests/test_laboratory_evidence_registry.py | 3 +- tests/test_laboratory_execution.py | 4 + tests/test_m48_object_quality_api.py | 105 +++ ...test_m48_static_occupancy_qualification.py | 106 +++ 14 files changed, 1286 insertions(+), 1 deletion(-) create mode 100644 config/laboratories/m48-static-occupancy-qualification.json create mode 100644 config/perception/m48-static-occupancy-qualification-v1.json create mode 100644 scripts/run_m48_static_occupancy_qualification.py create mode 100644 src/k1link/laboratory/m48_static_occupancy_qualification.py create mode 100644 tests/test_m48_static_occupancy_qualification.py diff --git a/config/laboratories/m48-static-occupancy-qualification.json b/config/laboratories/m48-static-occupancy-qualification.json new file mode 100644 index 0000000..5bcad99 --- /dev/null +++ b/config/laboratories/m48-static-occupancy-qualification.json @@ -0,0 +1,10 @@ +{ + "schema_version": "missioncore.laboratory-evidence-definition/v1", + "work_id": "m48-static-occupancy-qualification", + "evidence": { + "runtime_relative_root": "m48/static-occupancy-qualification-results", + "result_id_prefix": "m48-static-occupancy-qualification", + "document_name": "manifest.json", + "schema_version": "missioncore.m48-static-occupancy-qualification-result/v1" + } +} diff --git a/config/laboratory-execution.json b/config/laboratory-execution.json index e4ec08c..b929cc3 100644 --- a/config/laboratory-execution.json +++ b/config/laboratory-execution.json @@ -1,6 +1,26 @@ { "schema_version": "missioncore.laboratory-execution-registry/v1", "definitions": [ + { + "work_id": "m48-static-occupancy-qualification", + "lifecycle": "canonical", + "isolation": "core-adapter", + "adapter_id": "canonical.m48-static-occupancy-qualification/v1", + "input_roles": [ + "repository_root", + "profile_path", + "m47_lab_root", + "graph_result_root", + "small_static_result_root" + ], + "contracts": { + "source": "missioncore.m48-static-occupancy-source-set/v1", + "provider": "missioncore.m48-additive-low-step-occupancy/v1", + "graph": "missioncore.m48-static-occupancy-case/v1", + "run": "missioncore.laboratory-run/v1", + "evidence": "missioncore.m48-static-occupancy-qualification-result/v1" + } + }, { "work_id": "m48-small-static-passage-regression", "lifecycle": "canonical", diff --git a/config/perception/m48-static-occupancy-qualification-v1.json b/config/perception/m48-static-occupancy-qualification-v1.json new file mode 100644 index 0000000..5c37db1 --- /dev/null +++ b/config/perception/m48-static-occupancy-qualification-v1.json @@ -0,0 +1,56 @@ +{ + "schema_version": "missioncore.m48-static-occupancy-qualification-profile/v1", + "profile_id": "m48-conservative-static-occupancy/v1", + "pipeline_id": "m4-current-rolling-plus-step-static-occupancy/v1", + "experiment_id": "m48-static-occupancy-qualification/v1", + "human_lab_id": "M4.8", + "run_label": "M4.8R2", + "source": { + "source_id": "RAVNOVES00", + "source_session_id": "20260720T065719Z_viewer_live", + "m47_lab_result_id": "m47-reference-graph-lab-49678f0a7c628c7e991af0964fa57d005baa027d2d1eea19f38bbfe27ed39ce5", + "m47_graph_result_id": "m47-reference-graph-5f6a851cd655c7cf07c3025dacadbc188018b0afa97eda3a08802266e12da87d", + "m47_graph_frames_sha256": "d2cd53f8cff555410959600a79eb9a101aad4a8009ba87bdd0f3c5f122948071", + "small_static_result_id": "m48-small-static-passage-regression-3e3a2001f87fd3adcb736de52a65e42515044d83faa3515f892705b40915c084", + "small_static_anchors_sha256": "6a317aa75dde8204d5574a56166bebc9347e8797934dbc82f076a0abfdfaa95a" + }, + "selection": { + "motion": "static", + "requires_avoidance_or_clearance": true, + "canonical_engineering_sequences": [1880, 2584], + "independent_truth": false + }, + "distance_bands_m": { + "critical_near": [0.0, 8.0], + "approach": [8.0, 12.0] + }, + "candidate": { + "point_sources": ["local-surface-occupied", "local-surface-step-candidate"], + "minimum_points": 2, + "minimum_voxels": 1, + "voxel_size_m": 0.35, + "depth_cluster_minimum_gap_m": 0.65, + "depth_cluster_gap_fraction": 0.08, + "spatial_cluster_radius_m": 0.75 + }, + "acceptance": { + "minimum_critical_near_candidate_recall": 1.0, + "minimum_approach_candidate_recall": 0.95, + "minimum_canonical_engineering_recall": 1.0, + "maximum_false_free_count": 0 + }, + "policy": { + "absence_of_points_means_free": false, + "absence_of_camera_detection_means_free": false, + "step_candidate_can_only_add_occupied_or_unknown": true, + "semantic_class_used": false, + "planner_authoritative_free_space_claimed": false + }, + "authority": { + "mode": "replay-simulated", + "physical_live": false, + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false + } +} diff --git a/scripts/run_m48_static_occupancy_qualification.py b/scripts/run_m48_static_occupancy_qualification.py new file mode 100644 index 0000000..76c1096 --- /dev/null +++ b/scripts/run_m48_static_occupancy_qualification.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +"""Publish one canonical append-only M4.8R2 static-occupancy qualification.""" + +from __future__ import annotations + +import argparse +import json +import socket +from pathlib import Path + +from k1link.compute.pipeline_telemetry import JsonlPipelineTelemetrySink +from k1link.laboratory import ( + LaboratoryEvidenceRegistry, + LaboratoryExecutionRegistry, + LaboratoryRunner, + LaboratoryRunRequest, +) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser() + parser.add_argument("--profile", type=Path, required=True) + parser.add_argument("--m47-lab-root", type=Path, required=True) + parser.add_argument("--graph-result-root", type=Path, required=True) + parser.add_argument("--small-static-result-root", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument("--receipt-root", type=Path, required=True) + parser.add_argument("--telemetry-path", type=Path, required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--request-id", required=True) + return parser + + +def main() -> int: + args = _parser().parse_args() + repository_root = Path(__file__).resolve().parents[1] + evidence = LaboratoryEvidenceRegistry.from_directory( + repository_root / "config" / "laboratories" + ) + execution = LaboratoryExecutionRegistry.from_file( + repository_root / "config" / "laboratory-execution.json", + evidence, + ) + runner = LaboratoryRunner( + registry=execution, + evidence_registry=evidence, + sink=JsonlPipelineTelemetrySink(args.telemetry_path), + ) + result = runner.run( + LaboratoryRunRequest( + work_id="m48-static-occupancy-qualification", + run_id=args.run_id, + request_id=args.request_id, + contour_id="mission-core-laboratory", + agent_id="local-control-plane", + node_id=socket.gethostname(), + source_id="RAVNOVES00", + source_package_id=args.m47_lab_root.name, + method_id="m48-conservative-static-occupancy/v1", + inputs={ + "repository_root": repository_root, + "profile_path": args.profile, + "m47_lab_root": args.m47_lab_root, + "graph_result_root": args.graph_result_root, + "small_static_result_root": args.small_static_result_root, + }, + output_root=args.output_root, + receipt_root=args.receipt_root, + ) + ) + print(json.dumps({ + "result_id": result.result_id, + "result_root": str(result.result_root), + "receipt_id": result.receipt_id, + "receipt_root": str(result.receipt_root), + }, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/laboratory/execution.py b/src/k1link/laboratory/execution.py index 8a4c4d8..6ae8a24 100644 --- a/src/k1link/laboratory/execution.py +++ b/src/k1link/laboratory/execution.py @@ -312,6 +312,9 @@ class LaboratoryRunner: def canonical_laboratory_adapters() -> dict[str, LaboratoryAdapter]: return { + "canonical.m48-static-occupancy-qualification/v1": ( + _run_m48_static_occupancy_qualification + ), "canonical.m48-small-static-passage-regression/v1": ( _run_m48_small_static_passage_regression ), @@ -326,6 +329,27 @@ def canonical_laboratory_adapters() -> dict[str, LaboratoryAdapter]: } +def _run_m48_static_occupancy_qualification( + request: LaboratoryRunRequest, +) -> LaboratoryAdapterResult: + from k1link.laboratory.m48_static_occupancy_qualification import ( + build_m48_static_occupancy_qualification, + ) + + result = build_m48_static_occupancy_qualification( + repository_root=request.inputs["repository_root"], + profile_path=request.inputs["profile_path"], + m47_lab_root=request.inputs["m47_lab_root"], + graph_result_root=request.inputs["graph_result_root"], + small_static_result_root=request.inputs["small_static_result_root"], + output_root=request.output_root, + ) + return LaboratoryAdapterResult( + result_root=result.result_root, + result_id=result.result_id, + ) + + def _run_m48s_fixed_class_detector( request: LaboratoryRunRequest, ) -> LaboratoryAdapterResult: diff --git a/src/k1link/laboratory/m48_static_occupancy_qualification.py b/src/k1link/laboratory/m48_static_occupancy_qualification.py new file mode 100644 index 0000000..1ab98be --- /dev/null +++ b/src/k1link/laboratory/m48_static_occupancy_qualification.py @@ -0,0 +1,754 @@ +"""Immutable M4.8R2 qualification of conservative static LiDAR occupancy. + +The experiment does not run a detector and does not mutate the accepted M4.7 +graph. It measures the accepted current/rolling occupied output on the frozen +operator-assisted small-static anchors, then evaluates one additive CPU-only +candidate already present in the local-surface artifact: low step candidates. +Neither missing evidence nor a camera miss is ever converted to free space. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import uuid +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final + +import numpy as np + +from k1link.laboratory.m47_reference_graph import ( + M47ReferenceGraphLabError, + read_m47_reference_graph_lab, +) +from k1link.laboratory.m48_small_static_regression import ( + M48SmallStaticRegressionError, + read_m48_small_static_passage_regression, +) +from k1link.perception.geometry import RecordedGeometryStore +from k1link.perception.geometry_math import ( + POINT_OCCUPIED, + project_map_points_kb4, + semantic_geometry_support, +) + +M48_STATIC_OCCUPANCY_PROFILE_SCHEMA: Final = ( + "missioncore.m48-static-occupancy-qualification-profile/v1" +) +M48_STATIC_OCCUPANCY_RESULT_SCHEMA: Final = ( + "missioncore.m48-static-occupancy-qualification-result/v1" +) +M48_STATIC_OCCUPANCY_REPORT_SCHEMA: Final = ( + "missioncore.m48-static-occupancy-qualification-report/v1" +) +M48_STATIC_OCCUPANCY_CASE_SCHEMA: Final = "missioncore.m48-static-occupancy-case/v1" +M48_STATIC_OCCUPANCY_CANONICAL_SCHEMA: Final = ( + "missioncore.m48-static-occupancy-canonical-anchor/v1" +) +M48_STATIC_OCCUPANCY_PREFIX: Final = "m48-static-occupancy-qualification-" + +_METHOD_SCHEMA: Final = "missioncore.laboratory-method/v1" +_AUTHORITY: Final = { + "mode": "replay-simulated", + "physical_live": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + + +class M48StaticOccupancyQualificationError(RuntimeError): + """The static-occupancy source, method, or immutable result is invalid.""" + + +@dataclass(frozen=True, slots=True) +class M48StaticOccupancyQualificationResult: + result_id: str + result_root: Path + manifest: dict[str, Any] + report: dict[str, Any] + cases: tuple[dict[str, Any], ...] + canonical_anchors: tuple[dict[str, Any], ...] + + +def build_m48_static_occupancy_qualification( + *, + repository_root: Path, + profile_path: Path, + m47_lab_root: Path, + graph_result_root: Path, + small_static_result_root: Path, + output_root: Path, + run_created_at_utc: str | None = None, +) -> M48StaticOccupancyQualificationResult: + """Publish one deterministic, append-only M4.8R2 qualification result.""" + + repository = repository_root.resolve(strict=True) + profile_bytes, profile = _read_profile(profile_path) + source = _object(profile["source"], "M4.8R2 source") + try: + m47 = read_m47_reference_graph_lab(m47_lab_root) + small_static = read_m48_small_static_passage_regression(small_static_result_root) + except (M47ReferenceGraphLabError, M48SmallStaticRegressionError) as exc: + raise M48StaticOccupancyQualificationError( + "accepted M4.7/M4.8R1 evidence is invalid" + ) from exc + if ( + m47.result_id != source.get("m47_lab_result_id") + or small_static.result_id != source.get("small_static_result_id") + or m47.manifest.get("accepted") is not True + or m47.manifest.get("ground_truth") is not False + ): + raise M48StaticOccupancyQualificationError("M4.8R2 source identity changed") + + anchors_path = small_static.result_root / "anchors.jsonl" + if _file_sha256(anchors_path) != source.get("small_static_anchors_sha256"): + raise M48StaticOccupancyQualificationError("M4.8R2 anchor ledger changed") + + graph_root = graph_result_root.resolve(strict=True) + graph_manifest = _read_json(graph_root / "manifest.json", maximum=2 * 1024 * 1024) + frames_path = graph_root / "frames.jsonl" + graph_files = _object(graph_manifest.get("files"), "M4.8R2 graph files") + graph_frames = _object(graph_files.get("frames.jsonl"), "M4.8R2 graph frame artifact") + if ( + graph_root.name != source.get("m47_graph_result_id") + or graph_manifest.get("result_id") != graph_root.name + or graph_manifest.get("accepted") is not True + or graph_frames.get("sha256") != source.get("m47_graph_frames_sha256") + or _file_sha256(frames_path) != source.get("m47_graph_frames_sha256") + ): + raise M48StaticOccupancyQualificationError("M4.8R2 graph binding changed") + + selection = _object(profile["selection"], "M4.8R2 selection") + anchors = tuple( + row + for row in small_static.anchors + if row.get("motion") == selection.get("motion") + and row.get("requires_avoidance_or_clearance") + is selection.get("requires_avoidance_or_clearance") + ) + if not anchors: + raise M48StaticOccupancyQualificationError("M4.8R2 selected no static anchors") + frame_rows = _selected_graph_frames(frames_path, {int(row["sequence"]) for row in anchors}) + store = RecordedGeometryStore.from_repository(repository) + candidate = _object(profile["candidate"], "M4.8R2 candidate") + candidate_association = replace( + store.profile.association, + semantic_minimum_occupied_points=int(candidate["minimum_points"]), + semantic_minimum_occupied_voxels=int(candidate["minimum_voxels"]), + semantic_voxel_size_m=float(candidate["voxel_size_m"]), + depth_cluster_minimum_gap_m=float(candidate["depth_cluster_minimum_gap_m"]), + depth_cluster_gap_fraction=float(candidate["depth_cluster_gap_fraction"]), + spatial_cluster_radius_m=float(candidate["spatial_cluster_radius_m"]), + ) + + cases: list[dict[str, Any]] = [] + for anchor in anchors: + sequence = int(anchor["sequence"]) + graph_row = frame_rows[sequence] + frame = store.frame_for_index(sequence) + if frame is None or not frame.surface_valid: + raise M48StaticOccupancyQualificationError( + "selected M4.8R2 anchor lacks qualified current LiDAR" + ) + projected = project_map_points_kb4( + frame.points_map, + position_map_xyz=frame.sensor_position_map, + orientation_map_from_lidar_xyzw=frame.sensor_orientation_xyzw, + profile=frame.projection, + ) + bbox = _pixel_bbox(anchor["extent_xyxy"], frame.projection.width, frame.projection.height) + baseline = semantic_geometry_support( + bbox, + projected=projected, + frame_points_map=frame.points_map, + point_class=frame.point_class, + profile=store.profile.association, + ) + step_candidates = store.point_step_candidates_for_frame(sequence) + if step_candidates is None: + raise M48StaticOccupancyQualificationError( + "selected M4.8R2 anchor lacks low-step evidence" + ) + union_classes = np.array(frame.point_class, copy=True) + union_classes[step_candidates > 0] = POINT_OCCUPIED + additive = semantic_geometry_support( + bbox, + projected=projected, + frame_points_map=frame.points_map, + point_class=union_classes, + profile=candidate_association, + ) + graph_matches = _graph_component_matches( + graph_row=graph_row, + frame=frame, + bbox=bbox, + voxel_size_m=0.45, + ) + raw_depths = _depths_in_bbox(projected.pixels_xy, projected.depths_m, bbox) + distance = _support_distance( + additive.occupied_depths_m, baseline.occupied_depths_m, raw_depths + ) + band = _distance_band(distance, _object(profile["distance_bands_m"], "distance bands")) + accepted_graph = bool(graph_matches) + baseline_qualified = bool(baseline.qualified or accepted_graph) + candidate_qualified = bool(additive.qualified or accepted_graph) + false_free = graph_row["obstacle_map"].get("free_space_claimed") is True + cases.append( + { + "schema_version": M48_STATIC_OCCUPANCY_CASE_SCHEMA, + "anchor_id": anchor["anchor_id"], + "clip_id": anchor["clip_id"], + "sequence": sequence, + "extent_xyxy": anchor["extent_xyxy"], + "distance_m": distance, + "distance_band": band, + "accepted_graph": { + "matched": accepted_graph, + "component_count": len(graph_matches), + "components": graph_matches, + "free_space_claimed": false_free, + }, + "current_local_surface": _support_projection(baseline), + "additive_step_candidate": _support_projection(additive), + "baseline_qualified": baseline_qualified, + "candidate_qualified": candidate_qualified, + "outcome": ( + "candidate-qualified" + if candidate_qualified + else "unresolved-unknown-never-free" + ), + "authority": "operator-assisted-development-anchor-not-truth", + } + ) + cases.sort(key=lambda row: (int(row["sequence"]), str(row["anchor_id"]))) + + visual_report = _read_json( + m47.result_root / "visual-report.json", + maximum=2 * 1024 * 1024, + ) + canonical = _canonical_anchors(visual_report, selection) + metrics = _metrics(cases, canonical) + acceptance = _object(profile["acceptance"], "M4.8R2 acceptance") + gates = { + "critical_near_candidate_recall": ( + metrics["critical_near_candidate_recall"] + >= float(acceptance["minimum_critical_near_candidate_recall"]) + ), + "approach_candidate_recall": ( + metrics["approach_candidate_recall"] + >= float(acceptance["minimum_approach_candidate_recall"]) + ), + "canonical_engineering_recall": ( + metrics["canonical_engineering_recall"] + >= float(acceptance["minimum_canonical_engineering_recall"]) + ), + "zero_false_free": metrics["false_free_count"] + <= int(acceptance["maximum_false_free_count"]), + "independent_truth_available": False, + } + near_ready = bool( + gates["critical_near_candidate_recall"] + and gates["canonical_engineering_recall"] + and gates["zero_false_free"] + ) + accepted = bool(near_ready and gates["approach_candidate_recall"]) + created_at = _utc_timestamp(run_created_at_utc or datetime.now(UTC).isoformat()) + profile_sha256 = hashlib.sha256(profile_bytes).hexdigest() + producer_sha256 = _file_sha256(Path(__file__).resolve()) + identity = { + "schema_version": M48_STATIC_OCCUPANCY_RESULT_SCHEMA, + "human_lab_id": profile["human_lab_id"], + "run_label": profile["run_label"], + "run_created_at_utc": created_at, + "pipeline_id": profile["pipeline_id"], + "experiment_id": profile["experiment_id"], + "profile_id": profile["profile_id"], + "profile_sha256": profile_sha256, + "producer_sha256": producer_sha256, + "source": source, + "selection": { + "operator_static_anchor_count": len(cases), + "canonical_engineering_anchor_count": len(canonical), + }, + "authority": dict(_AUTHORITY), + } + result_id = M48_STATIC_OCCUPANCY_PREFIX + _canonical_sha256(identity) + method = { + "schema_version": _METHOD_SCHEMA, + "completeness": "complete", + "execution_class": "deterministic", + "pipeline_id": profile["pipeline_id"], + "components": [ + { + "kind": "source", + "name": "accepted M4.7 current/rolling obstacle graph", + "version": source["m47_graph_result_id"], + "role": "immutable baseline occupied/unknown and threat decisions", + "identity_sha256": source["m47_graph_frames_sha256"], + }, + { + "kind": "source", + "name": "M4.8R1 operator-assisted static anchors", + "version": source["small_static_result_id"], + "role": "candidate-visible diagnostic anchors; not independent truth", + "identity_sha256": source["small_static_anchors_sha256"], + }, + { + "kind": "algorithm", + "name": "additive low-step static occupancy candidate", + "version": profile["profile_id"], + "role": "CPU-only occupied-or-unknown evidence; never clearing", + "identity_sha256": producer_sha256, + }, + ], + } + report = { + "schema_version": M48_STATIC_OCCUPANCY_REPORT_SCHEMA, + "result_id": result_id, + "source": source, + "configuration": { + key: profile[key] + for key in ( + "profile_id", + "pipeline_id", + "experiment_id", + "human_lab_id", + "run_label", + "distance_bands_m", + "candidate", + "acceptance", + "policy", + ) + }, + "method": method, + "metrics": metrics, + "gates": gates, + "decision": { + "state": "accepted-bounded-static-occupancy-qualification" + if accepted + else "partial-static-occupancy-qualification", + "critical_near_candidate_ready_for_shadow": near_ready, + "production_accepted": False, + "summary": ( + "Accepted graph covers " + f"{metrics['baseline_qualified_count']}/{len(cases)} static assisted " + "anchors; the additive step candidate covers " + f"{metrics['candidate_qualified_count']}/{len(cases)}." + ), + "next_action": ( + "Integrate the additive step evidence as an occupied-only Worker shadow, " + "then measure full replay FPS, occupancy growth and the unresolved 8-12 m " + "case." + ), + }, + "limitations": [ + ( + "Operator-assisted anchors are candidate-visible development evidence, " + "not independent truth." + ), + "Projected camera rectangles do not define physical 3D colliders or chassis clearance.", + ( + "The step candidate may add conservative false occupancy and therefore " + "requires a full replay load/volume shadow before cutover." + ), + ( + "No ray clearing, planner-authoritative free space, physical navigation, " + "command, actuation or collision-safety authority is granted." + ), + ], + "authority": dict(_AUTHORITY), + } + destination = output_root.resolve(strict=False) / result_id + _publish_result(destination, identity, created_at, accepted, report, tuple(cases), canonical) + return read_m48_static_occupancy_qualification(destination) + + +def read_m48_static_occupancy_qualification( + result_root: Path, +) -> M48StaticOccupancyQualificationResult: + if result_root.is_symlink(): + raise M48StaticOccupancyQualificationError("M4.8R2 result root is invalid") + root = result_root.resolve(strict=True) + if root.name.startswith(M48_STATIC_OCCUPANCY_PREFIX) is False: + raise M48StaticOccupancyQualificationError("M4.8R2 result root is invalid") + manifest = _read_json(root / "manifest.json", maximum=2 * 1024 * 1024) + report = _read_json(root / "report.json", maximum=4 * 1024 * 1024) + cases = tuple(_read_jsonl(root / "cases.jsonl")) + canonical = tuple(_read_jsonl(root / "canonical-anchors.jsonl")) + if ( + manifest.get("schema_version") != M48_STATIC_OCCUPANCY_RESULT_SCHEMA + or manifest.get("result_id") != root.name + or report.get("schema_version") != M48_STATIC_OCCUPANCY_REPORT_SCHEMA + or report.get("result_id") != root.name + or manifest.get("ground_truth") is not False + or manifest.get("authority") != _AUTHORITY + ): + raise M48StaticOccupancyQualificationError("M4.8R2 result identity changed") + identity = _object(manifest.get("identity"), "M4.8R2 identity") + identity_sha256 = _canonical_sha256(identity) + if root.name != M48_STATIC_OCCUPANCY_PREFIX + identity_sha256: + raise M48StaticOccupancyQualificationError("M4.8R2 result identity changed") + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, list): + raise M48StaticOccupancyQualificationError("M4.8R2 artifact proof changed") + artifact_paths = [ + str(_object(item, "M4.8R2 artifact").get("path")) for item in artifacts + ] + if sorted(artifact_paths) != [ + "canonical-anchors.jsonl", + "cases.jsonl", + "report.json", + ]: + raise M48StaticOccupancyQualificationError("M4.8R2 artifact proof changed") + for artifact in artifacts: + item = _object(artifact, "M4.8R2 artifact") + path = root / str(item.get("path")) + if path.parent != root or _file_sha256(path) != item.get("sha256"): + raise M48StaticOccupancyQualificationError("M4.8R2 artifact proof changed") + if manifest.get("identity_sha256") != identity_sha256: + raise M48StaticOccupancyQualificationError("M4.8R2 identity digest changed") + if not cases or any( + row.get("schema_version") != M48_STATIC_OCCUPANCY_CASE_SCHEMA for row in cases + ): + raise M48StaticOccupancyQualificationError("M4.8R2 case ledger changed") + if any( + row.get("schema_version") != M48_STATIC_OCCUPANCY_CANONICAL_SCHEMA + for row in canonical + ): + raise M48StaticOccupancyQualificationError("M4.8R2 canonical ledger changed") + return M48StaticOccupancyQualificationResult( + root.name, root, manifest, report, cases, canonical + ) + + +def _support_projection(value: Any) -> dict[str, object]: + depths = value.occupied_depths_m + return { + "qualified": bool(value.qualified), + "projected_point_count": int(value.projected_points_in_region), + "occupied_point_count": int(value.occupied_points_in_region), + "clustered_occupied_point_count": int(value.occupied_source_indices.size), + "nearest_depth_m": None if not depths.size else round(float(np.min(depths)), 6), + "median_depth_m": None if not depths.size else round(float(np.median(depths)), 6), + } + + +def _graph_component_matches( + *, + graph_row: dict[str, Any], + frame: Any, + bbox: tuple[float, float, float, float], + voxel_size_m: float, +) -> list[dict[str, object]]: + obstacle_map = _object(graph_row.get("obstacle_map"), "M4.8R2 obstacle map") + threats = { + str(row.get("component_id")): row + for row in graph_row.get("threats", []) + if isinstance(row, dict) + } + matches: list[dict[str, object]] = [] + for obstacle in obstacle_map.get("occupied", []): + item = _object(obstacle, "M4.8R2 occupied component") + cells = item.get("cells") + if not isinstance(cells, list) or not cells: + continue + points = np.asarray( + [ + [ + (int(cell["x"]) + 0.5) * voxel_size_m, + (int(cell["y"]) + 0.5) * voxel_size_m, + (int(cell["z"]) + 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, + ) + depths = _depths_in_bbox(projected.pixels_xy, projected.depths_m, bbox) + if not depths.size: + continue + component_id = str(item.get("component_id")) + assessment = _object(threats.get(component_id), "M4.8R2 threat assessment") + matches.append( + { + "component_id": component_id, + "state": item.get("state"), + "decision": assessment.get("decision"), + "projected_cell_count": int(depths.size), + "nearest_depth_m": round(float(np.min(depths)), 6), + } + ) + matches.sort(key=lambda row: (float(row["nearest_depth_m"]), str(row["component_id"]))) + return matches + + +def _canonical_anchors( + report: dict[str, Any], selection: dict[str, Any] +) -> tuple[dict[str, Any], ...]: + metrics = _object(report.get("metrics"), "M4.7 visual metrics") + visual = _object(metrics.get("visual_evidence"), "M4.7 visual evidence") + rows: list[dict[str, Any]] = [] + for sequence in selection.get("canonical_engineering_sequences", []): + regression = _object( + visual.get(f"frame_{sequence}_regression"), + "canonical regression", + ) + for anchor in regression.get("engineering_anchors", []): + item = _object(anchor, "canonical anchor") + rows.append( + { + "schema_version": M48_STATIC_OCCUPANCY_CANONICAL_SCHEMA, + "sequence": int(sequence), + "anchor_id": item.get("anchor_id"), + "matched": item.get("matched") is True, + "decision": item.get("decision"), + "must_assert_threat": item.get("must_assert_threat") is True, + "authority": "camera-reviewed-engineering-anchor-not-truth", + } + ) + return tuple(rows) + + +def _metrics( + cases: list[dict[str, Any]], canonical: tuple[dict[str, Any], ...] +) -> dict[str, object]: + def band_rows(name: str) -> list[dict[str, Any]]: + return [row for row in cases if row["distance_band"] == name] + + def rate(rows: list[dict[str, Any]], key: str) -> float: + if not rows: + return 0.0 + return sum(bool(row[key]) for row in rows) / len(rows) + + near = band_rows("critical-near") + approach = band_rows("approach") + return { + "operator_static_anchor_count": len(cases), + "baseline_qualified_count": sum(bool(row["baseline_qualified"]) for row in cases), + "candidate_qualified_count": sum(bool(row["candidate_qualified"]) for row in cases), + "unresolved_unknown_count": sum(not bool(row["candidate_qualified"]) for row in cases), + "critical_near_anchor_count": len(near), + "critical_near_baseline_recall": rate(near, "baseline_qualified"), + "critical_near_candidate_recall": rate(near, "candidate_qualified"), + "approach_anchor_count": len(approach), + "approach_baseline_recall": rate(approach, "baseline_qualified"), + "approach_candidate_recall": rate(approach, "candidate_qualified"), + "canonical_engineering_anchor_count": len(canonical), + "canonical_engineering_recall": sum(bool(row["matched"]) for row in canonical) + / len(canonical) + if canonical + else 0.0, + "false_free_count": sum(bool(row["accepted_graph"]["free_space_claimed"]) for row in cases), + "independent_truth": False, + } + + +def _selected_graph_frames(path: Path, sequences: set[int]) -> dict[int, dict[str, Any]]: + rows: dict[int, dict[str, Any]] = {} + with path.open("r", encoding="utf-8") as stream: + for line in stream: + row = _object(json.loads(line), "M4.8R2 graph frame") + sequence = row.get("sequence") + if isinstance(sequence, int) and sequence in sequences: + rows[sequence] = row + if len(rows) == len(sequences): + break + if set(rows) != sequences: + raise M48StaticOccupancyQualificationError("M4.8R2 graph frames are incomplete") + return rows + + +def _pixel_bbox(value: object, width: int, height: int) -> tuple[float, float, float, float]: + extent = value if isinstance(value, list) else None + if extent is None or len(extent) != 4: + raise M48StaticOccupancyQualificationError("M4.8R2 anchor extent is invalid") + return ( + float(extent[0]) * width, + float(extent[1]) * height, + float(extent[2]) * width, + float(extent[3]) * height, + ) + + +def _depths_in_bbox( + pixels: np.ndarray, depths: np.ndarray, bbox: tuple[float, float, float, float] +) -> np.ndarray: + if not pixels.size: + return np.empty(0, dtype=np.float64) + inside = ( + (pixels[:, 0] >= bbox[0]) + & (pixels[:, 0] <= bbox[2]) + & (pixels[:, 1] >= bbox[1]) + & (pixels[:, 1] <= bbox[3]) + ) + return depths[inside] + + +def _support_distance(*values: np.ndarray) -> float: + for value in values: + if value.size: + return round(float(np.median(value)), 6) + raise M48StaticOccupancyQualificationError("M4.8R2 anchor has no projected LiDAR depth") + + +def _distance_band(distance: float, bands: dict[str, Any]) -> str: + for key, label in (("critical_near", "critical-near"), ("approach", "approach")): + bounds = bands.get(key) + if ( + isinstance(bounds, list) + and len(bounds) == 2 + and float(bounds[0]) <= distance < float(bounds[1]) + ): + return label + return "outside-qualified-bands" + + +def _read_profile(path: Path) -> tuple[bytes, dict[str, Any]]: + encoded = path.resolve(strict=True).read_bytes() + profile = _object(json.loads(encoded), "M4.8R2 profile") + if ( + profile.get("schema_version") != M48_STATIC_OCCUPANCY_PROFILE_SCHEMA + or profile.get("authority") != _AUTHORITY + ): + raise M48StaticOccupancyQualificationError("M4.8R2 profile is invalid") + return encoded, profile + + +def _publish_result( + destination: Path, + identity: dict[str, Any], + created_at: str, + accepted: bool, + report: dict[str, Any], + cases: tuple[dict[str, Any], ...], + canonical: tuple[dict[str, Any], ...], +) -> 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 / "canonical-anchors.jsonl", canonical) + artifacts = [ + _artifact(staging / name, role) + for name, role in ( + ("cases.jsonl", "operator-static-anchor-comparisons"), + ("canonical-anchors.jsonl", "accepted-canonical-engineering-anchors"), + ("report.json", "m48-static-occupancy-report"), + ) + ] + manifest = { + "schema_version": M48_STATIC_OCCUPANCY_RESULT_SCHEMA, + "result_id": destination.name, + "identity_sha256": _canonical_sha256(identity), + "identity": identity, + "created_at_utc": created_at, + "accepted": accepted, + "ground_truth": False, + "authority": dict(_AUTHORITY), + "artifacts": artifacts, + } + _write_json(staging / "manifest.json", manifest) + if destination.exists(): + existing = { + path.name: _file_sha256(path) for path in destination.iterdir() if path.is_file() + } + proposed = { + path.name: _file_sha256(path) for path in staging.iterdir() if path.is_file() + } + if existing != proposed: + raise M48StaticOccupancyQualificationError("immutable M4.8R2 identity collided") + shutil.rmtree(staging) + return + os.replace(staging, destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + + +def _artifact(path: Path, role: str) -> dict[str, object]: + 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 M48StaticOccupancyQualificationError(f"{path.name} is unavailable") + return _object(json.loads(path.read_text("utf-8")), path.name) + + +def _read_jsonl(path: Path) -> list[dict[str, Any]]: + if path.is_symlink() or not path.is_file() or path.stat().st_size > 8 * 1024 * 1024: + raise M48StaticOccupancyQualificationError(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", encoding="utf-8" + ) + + +def _write_jsonl(path: Path, rows: tuple[dict[str, Any], ...]) -> None: + path.write_text( + "".join( + json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" + for row in rows + ), + encoding="utf-8", + ) + + +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 M48StaticOccupancyQualificationError(f"{label} is invalid") + return value + + +def _utc_timestamp(value: str) -> str: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + raise M48StaticOccupancyQualificationError("M4.8R2 creation time is invalid") + return parsed.astimezone(UTC).isoformat().replace("+00:00", "Z") + + +__all__ = [ + "M48StaticOccupancyQualificationError", + "M48StaticOccupancyQualificationResult", + "build_m48_static_occupancy_qualification", + "read_m48_static_occupancy_qualification", +] diff --git a/src/k1link/perception/geometry.py b/src/k1link/perception/geometry.py index dba7b73..690bdf2 100644 --- a/src/k1link/perception/geometry.py +++ b/src/k1link/perception/geometry.py @@ -344,6 +344,28 @@ class RecordedGeometryStore: points.setflags(write=False) return points + def point_step_candidates_for_frame(self, frame_index: int) -> UInt8Array | None: + """Expose the sealed low-step diagnostic in the source point index space. + + The array is evidence only: a non-zero value may add conservative + occupied/unknown support, but it never clears a cell or claims free + space. Unavailable and surface-invalid frames remain unavailable. + """ + + frame = self.frame_for_index(frame_index) + if frame is None or not frame.surface_valid: + return None + offsets = self._source["cloud_offsets"] + start, end = int(offsets[frame_index]), int(offsets[frame_index + 1]) + values = np.asarray( + self._surface["point_step_candidate"][start:end], + dtype=np.uint8, + ) + if values.shape != (frame.source_point_count,): + raise GeometryProviderError("local-surface step evidence changed") + values.setflags(write=False) + return values + @property def maximum_current_point_count(self) -> int: """Return the immutable source-pack upper bound for one recorded increment.""" diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 824d748..9a5c22b 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -934,6 +934,13 @@ app.include_router( / "m48" / "small-static-passage-regression-results" ), + static_occupancy_result_root_provider=lambda: ( + REPOSITORY_ROOT + / ".runtime" + / "compute-experiments" + / "m48" + / "static-occupancy-qualification-results" + ), camera_frame_provider=( session_recorded_camera_frame_service.extract if session_recorded_camera_frame_service is not None diff --git a/src/k1link/web/m48_object_quality_api.py b/src/k1link/web/m48_object_quality_api.py index 75253c7..a72c633 100644 --- a/src/k1link/web/m48_object_quality_api.py +++ b/src/k1link/web/m48_object_quality_api.py @@ -48,6 +48,11 @@ from k1link.laboratory.m48_small_static_regression import ( M48SmallStaticRegressionResult, read_m48_small_static_passage_regression, ) +from k1link.laboratory.m48_static_occupancy_qualification import ( + M48StaticOccupancyQualificationError, + M48StaticOccupancyQualificationResult, + read_m48_static_occupancy_qualification, +) from k1link.sessions import RecordedCameraPlaybackSource _PACK_ID = re.compile(r"^m48-object-quality-pack-[a-f0-9]{64}$") @@ -57,6 +62,9 @@ _SMALL_STATIC_RESULT_ID = re.compile( r"^m48-small-static-passage-regression-[a-f0-9]{64}$" ) _SMALL_STATIC_ANCHOR_ID = re.compile(r"^anchor-[a-f0-9]{24}$") +_STATIC_OCCUPANCY_RESULT_ID = re.compile( + r"^m48-static-occupancy-qualification-[a-f0-9]{64}$" +) _M47_LAB_RESULT_ID = re.compile(r"^m47-reference-graph-lab-[a-f0-9]{64}$") _FAILURE_ID = re.compile(r"^m48-failure-[a-f0-9]{64}$") _REVIEW_SESSION_ID = re.compile(r"^m48-review-session-[a-f0-9]{64}$") @@ -93,6 +101,12 @@ _SMALL_STATIC_CASE_CATALOG_SCHEMA: Final = ( _SMALL_STATIC_CASE_VIEW_SCHEMA: Final = ( "missioncore.m48-small-static-passage-regression-case-view/v1" ) +_STATIC_OCCUPANCY_RESULT_VIEW_SCHEMA: Final = ( + "missioncore.m48-static-occupancy-qualification-result-view/v1" +) +_STATIC_OCCUPANCY_CASE_CATALOG_SCHEMA: Final = ( + "missioncore.m48-static-occupancy-case-catalog/v1" +) _FAILURE_ATLAS_VIEW_SCHEMA: Final = "missioncore.m48-object-quality-failure-atlas-view/v1" _FAILURE_CASE_VIEW_SCHEMA: Final = "missioncore.m48-object-quality-failure-case-view/v1" _REVIEW_CAPABILITY_HEADER: Final = "X-M48-Review-Capability" @@ -438,6 +452,7 @@ def build_m48_object_quality_router( truth_root_provider: RootProvider = lambda: None, result_root_provider: RootProvider = lambda: None, small_static_result_root_provider: RootProvider = lambda: None, + static_occupancy_result_root_provider: RootProvider = lambda: None, camera_frame_provider: CameraFrameProvider | None = None, camera_playback_provider: CameraPlaybackProvider | None = None, spatial_evidence_provider: SpatialEvidenceProvider | None = None, @@ -1593,6 +1608,56 @@ def build_m48_object_quality_router( "access": "assisted-development-regression-case-read-only", } + @router.get("/regressions/static-occupancy/{result_id}") + def get_static_occupancy_qualification( + result_id: Annotated[str, ApiPath(pattern=_STATIC_OCCUPANCY_RESULT_ID.pattern)], + ) -> dict[str, object]: + result = _resolve_static_occupancy_result( + static_occupancy_result_root_provider, + result_id, + ) + source = _object(result.report.get("source"), "M4.8R2 source") + configuration = _object( + result.report.get("configuration"), + "M4.8R2 configuration", + ) + return { + "schema_version": _STATIC_OCCUPANCY_RESULT_VIEW_SCHEMA, + "result_id": result.result_id, + "created_at_utc": result.manifest.get("created_at_utc"), + "reference_graph_lab_result_id": source.get("m47_lab_result_id"), + "small_static_result_id": source.get("small_static_result_id"), + "run_label": configuration.get("run_label"), + "pipeline_id": configuration.get("pipeline_id"), + "experiment_id": configuration.get("experiment_id"), + "accepted": result.manifest.get("accepted"), + "metrics": copy.deepcopy(result.report.get("metrics")), + "gates": copy.deepcopy(result.report.get("gates")), + "decision": copy.deepcopy(result.report.get("decision")), + "ground_truth": False, + "independent_truth": False, + "authority": dict(_AUTHORITY), + "access": "static-occupancy-qualification-read-only", + } + + @router.get("/regressions/static-occupancy/{result_id}/cases") + def get_static_occupancy_qualification_cases( + result_id: Annotated[str, ApiPath(pattern=_STATIC_OCCUPANCY_RESULT_ID.pattern)], + ) -> dict[str, object]: + result = _resolve_static_occupancy_result( + static_occupancy_result_root_provider, + result_id, + ) + return { + "schema_version": _STATIC_OCCUPANCY_CASE_CATALOG_SCHEMA, + "result_id": result.result_id, + "cases": copy.deepcopy(result.cases), + "case_count": len(result.cases), + "ground_truth": False, + "authority": dict(_AUTHORITY), + "access": "static-occupancy-qualification-read-only", + } + return router @@ -1691,6 +1756,27 @@ def _resolve_small_static_result( ) from None +def _resolve_static_occupancy_result( + provider: RootProvider, + result_id: str, +) -> M48StaticOccupancyQualificationResult: + if _STATIC_OCCUPANCY_RESULT_ID.fullmatch(result_id) is None: + raise HTTPException(status_code=404, detail="M4.8R2 result was not found") + root = _configured_root(provider) + if root is None: + raise HTTPException(status_code=404, detail="M4.8R2 result was not found") + path = root / result_id + if path.is_symlink() or not path.is_dir(): + raise HTTPException(status_code=404, detail="M4.8R2 result was not found") + try: + resolved = path.resolve(strict=True) + if resolved.parent != root: + raise OSError("M4.8R2 result escaped root") + return read_m48_static_occupancy_qualification(resolved) + except (M48StaticOccupancyQualificationError, OSError, TypeError, ValueError): + raise HTTPException(status_code=404, detail="M4.8R2 result was not found") from None + + def _result_sources( result: M48ObjectQualityResult, *, diff --git a/tests/test_geometry_association_provider.py b/tests/test_geometry_association_provider.py index 8249e4c..f58dd7f 100644 --- a/tests/test_geometry_association_provider.py +++ b/tests/test_geometry_association_provider.py @@ -152,6 +152,15 @@ def test_profile_is_strict_digest_bound_and_store_accepts_exact_evidence() -> No assert store.profile.local_surface_sha256 == ( "f57eb2485b6cef47f2a97a2d9ff1aa9fd9265fe1eb69cd5852d12f39e13b8bc6" ) + step_candidates = store.point_step_candidates_for_frame(0) + assert step_candidates is not None + assert step_candidates.shape == (2389,) + assert step_candidates.dtype == np.uint8 + assert step_candidates.flags.writeable is False + with pytest.raises(ValueError): + step_candidates[0] = 0 + with pytest.raises(GeometryProviderError, match="frame index"): + store.point_step_candidates_for_frame(True) def test_provider_arbitrates_points_and_publishes_classless_geometry_only() -> None: diff --git a/tests/test_laboratory_evidence_registry.py b/tests/test_laboratory_evidence_registry.py index 308d67a..d836972 100644 --- a/tests/test_laboratory_evidence_registry.py +++ b/tests/test_laboratory_evidence_registry.py @@ -127,7 +127,7 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None: repository_root / "config" / "laboratories" ) - assert len(registry.definitions) == 38 + assert len(registry.definitions) == 39 assert {item.work_id for item in registry.definitions} >= { "e31-source-binding", "e46j-raw-fisheye-realtime", @@ -141,6 +141,7 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None: "m47-reference-graph-shadow", "m48-object-centric-quality", "m48-small-static-passage-regression", + "m48-static-occupancy-qualification", "m48s-fixed-class-detector", "m48t-risk-quality-temporal", } diff --git a/tests/test_laboratory_execution.py b/tests/test_laboratory_execution.py index 10d4647..0708858 100644 --- a/tests/test_laboratory_execution.py +++ b/tests/test_laboratory_execution.py @@ -92,6 +92,7 @@ def test_repository_registry_classifies_every_evidence_definition() -> None: assert {row.work_id for row in execution.definitions} == { "m48-small-static-passage-regression", + "m48-static-occupancy-qualification", "m48-object-centric-quality", "m4-replay-threat", "e33-worker-shadow", @@ -108,6 +109,9 @@ def test_repository_registry_classifies_every_evidence_definition() -> None: assert by_work_id["m48-object-centric-quality"].evidence_contract == ( "missioncore.m48-object-centric-quality-result/v1" ) + assert by_work_id["m48-static-occupancy-qualification"].evidence_contract == ( + "missioncore.m48-static-occupancy-qualification-result/v1" + ) assert by_work_id["e47-semantic-slam-shadow"].lifecycle == "experimental" assert by_work_id["e47-semantic-slam-shadow"].isolation == "bounded-adapter" assert by_work_id["m48s-fixed-class-detector"].lifecycle == "experimental" diff --git a/tests/test_m48_object_quality_api.py b/tests/test_m48_object_quality_api.py index a5a2d70..d9c7238 100644 --- a/tests/test_m48_object_quality_api.py +++ b/tests/test_m48_object_quality_api.py @@ -217,6 +217,82 @@ def _fixture( lambda _: regression_result, ) + static_occupancy_result_id = f"m48-static-occupancy-qualification-{'e' * 64}" + static_occupancy_root = tmp_path / "static-occupancy" / static_occupancy_result_id + static_occupancy_root.mkdir(parents=True) + static_occupancy_result = SimpleNamespace( + result_id=static_occupancy_result_id, + result_root=static_occupancy_root, + manifest={ + "created_at_utc": "2026-08-26T12:00:00Z", + "accepted": False, + }, + report={ + "source": { + "m47_lab_result_id": f"m47-reference-graph-lab-{'c' * 64}", + "small_static_result_id": regression_result_id, + }, + "configuration": { + "run_label": "M4.8R2", + "pipeline_id": "m4-current-rolling-plus-step-static-occupancy/v1", + "experiment_id": "m48-static-occupancy-qualification/v1", + }, + "metrics": { + "operator_static_anchor_count": 1, + "baseline_qualified_count": 0, + "candidate_qualified_count": 1, + "unresolved_unknown_count": 0, + "critical_near_anchor_count": 1, + "critical_near_baseline_recall": 0.0, + "critical_near_candidate_recall": 1.0, + "approach_anchor_count": 0, + "approach_baseline_recall": 0.0, + "approach_candidate_recall": 0.0, + "canonical_engineering_anchor_count": 4, + "canonical_engineering_recall": 1.0, + "false_free_count": 0, + }, + "gates": { + "critical_near_candidate_recall": True, + "approach_candidate_recall": False, + "canonical_engineering_recall": True, + "zero_false_free": True, + "independent_truth_available": False, + }, + "decision": { + "state": "partial-static-occupancy-qualification", + "critical_near_candidate_ready_for_shadow": True, + "production_accepted": False, + "summary": "fixture", + "next_action": "worker shadow", + }, + }, + cases=( + { + "anchor_id": regression_anchor_id, + "clip_id": "neutral-clip-01", + "sequence": 1, + "extent_xyxy": [0.2, 0.2, 0.3, 0.4], + "distance_m": 3.0, + "distance_band": "critical-near", + "accepted_graph": { + "matched": False, + "component_count": 0, + "components": [], + "free_space_claimed": False, + }, + "baseline_qualified": False, + "candidate_qualified": True, + "outcome": "candidate-qualified", + }, + ), + ) + monkeypatch.setattr( + api, + "read_m48_static_occupancy_qualification", + lambda _: static_occupancy_result, + ) + observations: dict[str, list[dict[str, Any]]] = { "reviews": [], "adjudications": [], @@ -444,6 +520,7 @@ def _fixture( truth_root_provider=lambda: tmp_path / "truth", result_root_provider=lambda: tmp_path / "results", small_static_result_root_provider=lambda: tmp_path / "small-static", + static_occupancy_result_root_provider=lambda: tmp_path / "static-occupancy", camera_frame_provider=camera, camera_playback_provider=camera_playback, # type: ignore[arg-type] spatial_evidence_provider=spatial_frame if spatial else None, @@ -1128,3 +1205,31 @@ def test_m48_small_static_regression_is_separate_read_only_assisted_evidence( assert body["ground_truth"] is False assert body["camera_url"].endswith("/frames/1/camera") assert body["spatial_url"].endswith("/frames/1/spatial") + + +def test_m48_static_occupancy_qualification_is_read_only_bounded_evidence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, _, _ = _fixture(tmp_path, monkeypatch, spatial=True) + result_id = f"m48-static-occupancy-qualification-{'e' * 64}" + + summary = client.get( + f"/api/v1/laboratory/m48/regressions/static-occupancy/{result_id}" + ) + assert summary.status_code == 200 + assert summary.headers["cache-control"] == "no-store" + assert summary.json()["run_label"] == "M4.8R2" + assert summary.json()["accepted"] is False + assert summary.json()["ground_truth"] is False + assert summary.json()["independent_truth"] is False + assert summary.json()["metrics"]["critical_near_candidate_recall"] == 1.0 + assert summary.json()["decision"]["production_accepted"] is False + + catalog = client.get( + f"/api/v1/laboratory/m48/regressions/static-occupancy/{result_id}/cases" + ) + assert catalog.status_code == 200 + assert catalog.json()["case_count"] == 1 + assert catalog.json()["cases"][0]["outcome"] == "candidate-qualified" + assert catalog.json()["cases"][0]["accepted_graph"]["free_space_claimed"] is False diff --git a/tests/test_m48_static_occupancy_qualification.py b/tests/test_m48_static_occupancy_qualification.py new file mode 100644 index 0000000..d240319 --- /dev/null +++ b/tests/test_m48_static_occupancy_qualification.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +import k1link.laboratory.m48_static_occupancy_qualification as qualification +from k1link.laboratory.evidence_registry import LaboratoryEvidenceRegistry +from k1link.laboratory.evidence_report import verify_laboratory_evidence_result + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +AUTHORITY = { + "mode": "replay-simulated", + "physical_live": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + + +def _sealed_result(tmp_path: Path) -> qualification.M48StaticOccupancyQualificationResult: + identity = { + "schema_version": qualification.M48_STATIC_OCCUPANCY_RESULT_SCHEMA, + "human_lab_id": "M4.8R2", + "run_label": "fixture", + "run_created_at_utc": "2026-08-26T12:00:00Z", + "authority": AUTHORITY, + } + result_id = qualification.M48_STATIC_OCCUPANCY_PREFIX + qualification._canonical_sha256( + identity + ) + report = { + "schema_version": qualification.M48_STATIC_OCCUPANCY_REPORT_SCHEMA, + "result_id": result_id, + "metrics": { + "operator_static_anchor_count": 1, + "candidate_qualified_count": 1, + }, + "decision": {"production_accepted": False}, + "authority": AUTHORITY, + } + cases = ( + { + "schema_version": qualification.M48_STATIC_OCCUPANCY_CASE_SCHEMA, + "anchor_id": "anchor-" + "a" * 24, + "sequence": 10, + "baseline_qualified": False, + "candidate_qualified": True, + "outcome": "candidate-qualified", + }, + ) + canonical = ( + { + "schema_version": qualification.M48_STATIC_OCCUPANCY_CANONICAL_SCHEMA, + "anchor_id": "hemisphere-01", + "sequence": 1880, + "matched": True, + }, + ) + destination = tmp_path / "results" / result_id + qualification._publish_result( + destination, + identity, + "2026-08-26T12:00:00Z", + False, + report, + cases, + canonical, + ) + return qualification.read_m48_static_occupancy_qualification(destination) + + +def test_seals_bounded_static_occupancy_evidence_without_production_authority( + tmp_path: Path, +) -> None: + result = _sealed_result(tmp_path) + + assert result.manifest["accepted"] is False + assert result.manifest["ground_truth"] is False + assert result.manifest["authority"] == AUTHORITY + assert result.report["decision"]["production_accepted"] is False + assert result.cases[0]["outcome"] == "candidate-qualified" + + registry = LaboratoryEvidenceRegistry.from_directory( + REPOSITORY_ROOT / "config/laboratories" + ) + definition = next( + row + for row in registry.definitions + if row.work_id == "m48-static-occupancy-qualification" + ) + proof = verify_laboratory_evidence_result(definition, result.result_root) + assert proof["result_id"] == result.result_id + assert proof["artifact_count"] == 3 + + +def test_reader_rejects_changed_static_occupancy_case_ledger(tmp_path: Path) -> None: + result = _sealed_result(tmp_path) + cases_path = result.result_root / "cases.jsonl" + cases_path.write_bytes(cases_path.read_bytes() + b"{}\n") + + with pytest.raises( + qualification.M48StaticOccupancyQualificationError, + match="artifact proof", + ): + qualification.read_m48_static_occupancy_qualification(result.result_root)