SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(perception): qualify conservative static occupancy

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-26 11:51:56 +03:00
parent 932c5216dc
commit 3c81c39a1c
14 changed files with 1286 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/laboratories/m48-static-occupancy-qualification.json
+10
View File
@@ -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"
}
}
config/laboratory-execution.json
+20
View File
@@ -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",
config/perception/m48-static-occupancy-qualification-v1.json
+56
View File
@@ -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
}
}
scripts/run_m48_static_occupancy_qualification.py
+81
View File
@@ -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())
src/k1link/laboratory/execution.py
+24
View File
@@ -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:
src/k1link/laboratory/m48_static_occupancy_qualification.py
+754
View File
@@ -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",
]
src/k1link/perception/geometry.py
+22
View File
@@ -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."""
src/k1link/web/app.py
+7
View File
@@ -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
src/k1link/web/m48_object_quality_api.py
+86
View File
@@ -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,
*,
tests/test_geometry_association_provider.py
+9
View File
@@ -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:
tests/test_laboratory_evidence_registry.py
+2 -1
View File
@@ -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",
}
tests/test_laboratory_execution.py
+4
View File
@@ -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"
tests/test_m48_object_quality_api.py
+105
View File
@@ -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
tests/test_m48_static_occupancy_qualification.py
+106
View File
@@ -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)