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

feat(lab): seal native risk review evidence

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-26 10:15:49 +03:00
parent c05db1fbe0
commit f9a76fed0e
7 changed files with 975 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/laboratories/m48t-risk-quality-temporal.json
+17 -7
View File
@@ -1,10 +1,20 @@
{
"schema_version": "missioncore.laboratory-evidence-definition/v1",
"schema_version": "missioncore.laboratory-evidence-definition/v2",
"work_id": "m48t-risk-quality-temporal",
"evidence": {
"runtime_relative_root": "m48t-risk-quality/lab-results",
"result_id_prefix": "m48t-risk-quality-temporal-lab",
"document_name": "manifest.json",
"schema_version": "missioncore.m48t-risk-quality-temporal-lab/v1"
}
"evidence_lifecycle": [
{
"phase": "legacy-quality",
"runtime_relative_root": "m48t-risk-quality/lab-results",
"result_id_prefix": "m48t-risk-quality-temporal-lab",
"document_name": "manifest.json",
"schema_version": "missioncore.m48t-risk-quality-temporal-lab/v1"
},
{
"phase": "result",
"runtime_relative_root": "m48t-risk-quality/native-lab-results",
"result_id_prefix": "m48q-native-risk-quality-lab",
"document_name": "manifest.json",
"schema_version": "missioncore.m48q-native-risk-quality-lab/v1"
}
]
}
experiments/perception/seal_m48q_native_risk_quality_lab.py
+39
View File
@@ -0,0 +1,39 @@
#!/usr/bin/env python3
"""Publish the immutable M4.8Q native raw-fisheye review LAB result."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from k1link.laboratory.m48q_native_risk_quality_lab import (
build_m48q_native_risk_quality_lab,
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repository-root", type=Path, required=True)
parser.add_argument("--output-root", type=Path, required=True)
arguments = parser.parse_args()
result = build_m48q_native_risk_quality_lab(
repository_root=arguments.repository_root,
output_root=arguments.output_root,
)
print(
json.dumps(
{
"result_id": result.result_id,
"result_root": str(result.result_root),
"status": result.manifest["status"],
},
ensure_ascii=False,
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
src/k1link/laboratory/m48q_native_risk_quality_lab.py
+471
View File
@@ -0,0 +1,471 @@
"""Seal native raw-fisheye M4.8Q risk review evidence as a terminal LAB phase."""
from __future__ import annotations
import hashlib
import json
import re
import shutil
import tempfile
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Final
from k1link.perception.fixed_class_detector_tournament import (
canonical_json,
false_authority,
sha256_path,
)
LAB_SCHEMA: Final = "missioncore.m48q-native-risk-quality-lab/v1"
REPORT_SCHEMA: Final = "missioncore.m48q-native-risk-quality-report/v1"
CATALOG_SCHEMA: Final = "missioncore.m48q-native-risk-review-catalog/v1"
CASE_SCHEMA: Final = "missioncore.m48q-native-risk-review-case/v1"
RESULT_PREFIX: Final = "m48q-native-risk-quality-lab-"
PROFILE_RELATIVE_PATH: Final = Path("config/perception/m48q-native-risk-case-mining-v1.json")
DETECTOR_PROFILE_RELATIVE_PATH: Final = Path(
"config/perception/rf-detr-large-native-kb4-risk-shadow-v0.json"
)
WORKER_RELATIVE_ROOT: Final = Path(
".runtime/worker-results/m48q-native-risk-case-mining/ravnoves00-native-raw-review-v3"
)
EXPECTED_PROFILE_SHA256: Final = "c455055e63578d505edc80b66d67e795916a3d0297cc18bfee5a6af7b032ceda"
EXPECTED_DETECTOR_PROFILE_SHA256: Final = (
"dbf4da5dbad6c3c22b1280b46ffcad81719bd183c81c263a4859847d829019b6"
)
EXPECTED_WORKER_RESULT_SHA256: Final = (
"a8f98d6d68abd6fd627caa05aaa3f048d69279a989b07fc2552b508af909a623"
)
EXPECTED_CASES_SHA256: Final = "9cea2a4a4558d761bc76b0d70474673ff07210473f53bd32cf178e8e0c8e179c"
EXPECTED_REPORT_IDENTITY_SHA256: Final = (
"b744775c0e08d18d728f04021cd2acf9b1ed6fcacf1520352ee04e06a6d5d324"
)
CASE_ID: Final = re.compile(r"^[0-9]{6}$")
class M48QNativeRiskQualityLabError(RuntimeError):
"""Raised when native M4.8Q evidence cannot be sealed honestly."""
@dataclass(frozen=True, slots=True)
class M48QNativeRiskQualityLabResult:
result_root: Path
result_id: str
manifest: dict[str, Any]
def build_m48q_native_risk_quality_lab(
*,
repository_root: Path,
output_root: Path,
) -> M48QNativeRiskQualityLabResult:
repository = repository_root.expanduser().resolve(strict=True)
profile_path = repository / PROFILE_RELATIVE_PATH
detector_profile_path = repository / DETECTOR_PROFILE_RELATIVE_PATH
worker_root = repository / WORKER_RELATIVE_ROOT
worker_result_path = worker_root / "result.json"
cases_path = worker_root / "cases.jsonl"
images_root = worker_root / "cases"
for path in (
profile_path,
detector_profile_path,
worker_result_path,
cases_path,
):
if path.is_symlink() or not path.is_file():
raise M48QNativeRiskQualityLabError(f"required M4.8Q evidence is missing: {path.name}")
if images_root.is_symlink() or not images_root.is_dir():
raise M48QNativeRiskQualityLabError("M4.8Q image evidence is missing")
expected_hashes = {
profile_path: EXPECTED_PROFILE_SHA256,
detector_profile_path: EXPECTED_DETECTOR_PROFILE_SHA256,
worker_result_path: EXPECTED_WORKER_RESULT_SHA256,
cases_path: EXPECTED_CASES_SHA256,
}
if any(sha256_path(path) != expected for path, expected in expected_hashes.items()):
raise M48QNativeRiskQualityLabError("M4.8Q source evidence identity changed")
profile = _read_object(profile_path)
detector_profile = _read_object(detector_profile_path)
worker_result = _read_object(worker_result_path)
cases = _read_cases(cases_path)
_validate_inputs(
profile=profile,
detector_profile=detector_profile,
worker_result=worker_result,
cases=cases,
images_root=images_root,
)
method = _method(profile, detector_profile, worker_result)
identity = {
"schema_version": LAB_SCHEMA,
"profile": {
"profile_id": profile["profile_id"],
"sha256": EXPECTED_PROFILE_SHA256,
},
"detector_profile": {
"profile_id": detector_profile["profile_id"],
"sha256": EXPECTED_DETECTOR_PROFILE_SHA256,
},
"source": {
"source_id": "RAVNOVES00",
"video_sha256": profile["source"]["video_sha256"],
"graph_result_sha256": worker_result["source"]["graph_result_sha256"],
"graph_frames_sha256": worker_result["source"]["graph_frames_sha256"],
"comparison_result_sha256": worker_result["source"]["comparison_result_sha256"],
"comparison_frames_sha256": worker_result["source"]["comparison_frames_sha256"],
"worker_result_sha256": EXPECTED_WORKER_RESULT_SHA256,
"cases_sha256": EXPECTED_CASES_SHA256,
},
"method": method,
"authority": false_authority(),
}
identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest()
result_id = RESULT_PREFIX + identity_sha256
root = output_root.expanduser().absolute()
root.mkdir(mode=0o700, parents=True, exist_ok=True)
destination = root / result_id
if destination.exists():
manifest = _read_object(destination / "manifest.json")
if (
manifest.get("schema_version") != LAB_SCHEMA
or manifest.get("identity_sha256") != identity_sha256
or manifest.get("result_id") != result_id
):
raise M48QNativeRiskQualityLabError("existing M4.8Q LAB identity conflicts")
return M48QNativeRiskQualityLabResult(destination, result_id, manifest)
created_at_utc = datetime.now(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z")
temporary = Path(tempfile.mkdtemp(prefix=".m48q-native-risk-lab-", dir=root))
try:
(temporary / "cases").mkdir(mode=0o700)
shutil.copyfile(profile_path, temporary / "profile.json")
shutil.copyfile(detector_profile_path, temporary / "detector-profile.json")
shutil.copyfile(worker_result_path, temporary / "worker-result.json")
shutil.copyfile(cases_path, temporary / "cases.jsonl")
catalog_cases: list[dict[str, object]] = []
for case in cases:
image = case["image"]
source_image = worker_root / image["path"]
destination_image = temporary / image["path"]
shutil.copyfile(source_image, destination_image)
catalog_cases.append(
{
"case_id": case["case_id"],
"sequence": case["sequence"],
"frame_id": case["frame_id"],
"evidence_time_ns": case["evidence_time_ns"],
"path": image["path"],
"media_type": "image/jpeg",
"width": 800,
"height": 600,
"byte_length": destination_image.stat().st_size,
"sha256": sha256_path(destination_image),
"geometric_resampling": False,
"selection_buckets": case["selection_buckets"],
"comparison": case["comparison"],
"proposals": case["proposals"],
}
)
catalog = {
"schema_version": CATALOG_SCHEMA,
"result_id": result_id,
"case_count": len(catalog_cases),
"source_raster": {"width": 800, "height": 600},
"overlay": {
"client_rendered": True,
"toggleable": True,
"box_coordinates": "source-pixel-xyxy",
"class_names": True,
"scores": True,
},
"ground_truth": False,
"cases": catalog_cases,
}
catalog_path = temporary / "catalog.json"
catalog_path.write_bytes(canonical_json(catalog) + b"\n")
graph_qualification = detector_profile["qualification"][
"full_ravnoves00_integrated_reference_graph"
]
decision = {
"review_ready": True,
"quality_evaluated": False,
"ground_truth": False,
"candidate_accepted": False,
"production_accepted": False,
"next_action": "operator-adjudication-in-existing-m48-review-instrument",
}
report = {
"schema_version": REPORT_SCHEMA,
"result_id": result_id,
"source": {
**profile["source"],
"graph_result_sha256": worker_result["source"]["graph_result_sha256"],
"graph_frames_sha256": worker_result["source"]["graph_frames_sha256"],
"independent_route_truth_available": False,
},
"configuration": {
"candidate": profile["candidate"],
"selection": profile["selection"],
},
"method": method,
"execution": {
"worker": worker_result["worker"],
"full_graph_frames": graph_qualification["frame_count"],
"requested_source_rate_hz": graph_qualification["requested_source_rate_hz"],
"effective_world_state_fps": graph_qualification["effective_world_state_fps"],
"world_state_completion_p95_ms": graph_qualification[
"world_state_completion_p95_ms"
],
"detector_total_p95_ms": graph_qualification["detector_total_p95_ms"],
"gpu_utilization_p95_percent": graph_qualification["gpu_utilization_p95_percent"],
"gpu_memory_used_maximum_mib": graph_qualification["gpu_memory_used_maximum_mib"],
"additional_inference_passes": 0,
},
"metrics": {
"selection": worker_result["selection"],
"runtime": {
"delivery_ratio": graph_qualification["delivery_ratio"],
"integrated_runtime_gate_passed": graph_qualification[
"integrated_runtime_gate_passed"
],
"operating_target_gate_passed": graph_qualification[
"operating_target_gate_passed"
],
},
"native_tensor_parity": detector_profile["qualification"][
"native_pytorch_tensorrt_parity"
],
"native_vs_legacy_704": detector_profile["qualification"][
"full_ravnoves00_native_vs_legacy_704"
],
},
"acceptance": {
"review_ready": True,
"integrated_runtime_gate_passed": True,
"independent_quality_evaluated": False,
"semantic_candidate_accepted": False,
},
"decision": decision,
"limitations": worker_result["limitations"],
"authority": false_authority(),
"visual_evidence": {
"kind": "native-raw-fisheye-risk-case-review",
"case_count": 24,
"source_raster": "800x600",
"geometric_resampling": False,
"boxes_baked_into_images": False,
"ground_truth": False,
},
}
report_path = temporary / "report.json"
report_path.write_bytes(canonical_json(report) + b"\n")
artifacts = _artifact_manifest(temporary)
manifest = {
"schema_version": LAB_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": created_at_utc,
"status": "complete-review-ready-quality-not-adjudicated",
"completed": True,
"bounded_question_accepted": True,
"ground_truth": False,
"method": method,
"metrics": report["metrics"],
"decision": decision,
"limitations": worker_result["limitations"],
"authority": false_authority(),
"artifacts": artifacts,
}
(temporary / "manifest.json").write_bytes(canonical_json(manifest) + b"\n")
temporary.replace(destination)
except BaseException:
shutil.rmtree(temporary, ignore_errors=True)
raise
return M48QNativeRiskQualityLabResult(destination, result_id, manifest)
def _validate_inputs(
*,
profile: dict[str, Any],
detector_profile: dict[str, Any],
worker_result: dict[str, Any],
cases: list[dict[str, Any]],
images_root: Path,
) -> None:
if (
profile.get("schema_version") != "missioncore.m48q-native-risk-case-mining-profile/v1"
or profile.get("source", {}).get("geometric_resampling") is not False
or profile.get("source", {}).get("raster_width") != 800
or profile.get("source", {}).get("raster_height") != 600
or profile.get("authority") != false_authority()
or detector_profile.get("schema_version")
!= "missioncore.rf-detr-native-risk-shadow-profile/v0"
or detector_profile.get("preprocessing", {}).get("resize") is not False
or detector_profile.get("preprocessing", {}).get("geometric_resampling") is not False
or detector_profile.get("status", {}).get("integrated_world_state_gate_passed") is not True
or worker_result.get("schema_version")
!= "missioncore.m48q-native-risk-case-mining-result/v1"
or worker_result.get("status") != "complete-review-ready-quality-not-adjudicated"
or worker_result.get("completed") is not True
or worker_result.get("report_identity_sha256") != EXPECTED_REPORT_IDENTITY_SHA256
or worker_result.get("artifacts", {}).get("cases", {}).get("sha256")
!= EXPECTED_CASES_SHA256
or worker_result.get("selection", {}).get("case_count") != 24
or worker_result.get("decision", {}).get("quality_evaluated") is not False
or worker_result.get("authority") != false_authority()
or len(cases) != 24
or len({case.get("case_id") for case in cases}) != 24
):
raise M48QNativeRiskQualityLabError("M4.8Q evidence contract changed")
for case in cases:
image = case.get("image")
proposals = case.get("proposals")
if (
case.get("schema_version") != CASE_SCHEMA
or not isinstance(case.get("case_id"), str)
or CASE_ID.fullmatch(case["case_id"]) is None
or case.get("case_id") != f"{case.get('sequence'):06d}"
or case.get("frame_id") != f"frame-{case.get('sequence'):06d}"
or case.get("ground_truth") is not False
or case.get("quality_evaluated") is not False
or case.get("authority") != false_authority()
or not isinstance(image, dict)
or image.get("path") != f"cases/frame-{case['case_id']}.jpg"
or image.get("width") != 800
or image.get("height") != 600
or image.get("geometric_resampling") is not False
or not isinstance(proposals, list)
or not proposals
):
raise M48QNativeRiskQualityLabError("M4.8Q review case contract changed")
image_path = images_root.parent / image["path"]
if (
image_path.is_symlink()
or not image_path.is_file()
or image_path.stat().st_size != image.get("byte_length")
or sha256_path(image_path) != image.get("sha256")
):
raise M48QNativeRiskQualityLabError("M4.8Q review image proof changed")
def _method(
profile: dict[str, Any],
detector_profile: dict[str, Any],
worker_result: dict[str, Any],
) -> dict[str, object]:
return {
"schema_version": "missioncore.laboratory-method/v1",
"completeness": "complete",
"execution_class": "hybrid",
"pipeline_id": "m48q-native-raw-fisheye-risk-case-review/v1",
"components": [
{
"kind": "source",
"name": "RAVNOVES00 raw KB4 video",
"version": "800x600 immutable recording",
"role": "unrectified review raster",
"identity_sha256": profile["source"]["video_sha256"],
},
{
"kind": "model",
"name": "RF-DETR-L native KB4 TensorRT",
"version": profile["candidate"]["model_id"],
"role": "fixed-class behavior-risk shadow proposals",
"identity_sha256": profile["candidate"]["engine_sha256"],
},
{
"kind": "runtime",
"name": "M4.7 native reference graph",
"version": "full RAVNOVES00 at recorded 12 FPS",
"role": "proposal and realtime evidence source",
"identity_sha256": worker_result["source"]["graph_result_sha256"],
},
{
"kind": "algorithm",
"name": "bounded diagnostic case miner",
"version": profile["profile_id"],
"role": "deterministic risk-family and edge-case sampling",
"identity_sha256": worker_result["identity"]["runner_sha256"],
},
{
"kind": "tool",
"name": "M4.8 existing image-case review instrument",
"version": "client-rendered source-pixel overlay/v1",
"role": "operator review without baked boxes",
"identity_sha256": EXPECTED_CASES_SHA256,
},
],
}
def _artifact_manifest(root: Path) -> list[dict[str, object]]:
artifacts: list[dict[str, object]] = []
for path in sorted(item for item in root.rglob("*") if item.is_file()):
relative = path.relative_to(root).as_posix()
if relative == "manifest.json":
continue
media_type = "application/json"
schema_version: str | None = None
role = "supporting-evidence"
if relative == "report.json":
role = "laboratory-report"
schema_version = REPORT_SCHEMA
elif relative == "catalog.json":
role = "visual-evidence-catalog"
schema_version = CATALOG_SCHEMA
elif relative == "profile.json":
role = "predeclared-native-risk-review-profile"
schema_version = "missioncore.m48q-native-risk-case-mining-profile/v1"
elif relative == "detector-profile.json":
role = "qualified-native-detector-profile"
schema_version = "missioncore.rf-detr-native-risk-shadow-profile/v0"
elif relative == "worker-result.json":
role = "upstream-worker-case-mining-result"
schema_version = "missioncore.m48q-native-risk-case-mining-result/v1"
elif relative == "cases.jsonl":
media_type = "application/x-ndjson"
role = "native-risk-review-case-ledger"
schema_version = CASE_SCHEMA
elif relative.endswith(".jpg"):
media_type = "image/jpeg"
role = "visual-evidence-native-raw-fisheye-frame"
artifacts.append(
{
"role": role,
"path": relative,
"byte_length": path.stat().st_size,
"sha256": sha256_path(path),
"media_type": media_type,
"schema_version": schema_version,
}
)
return artifacts
def _read_object(path: Path) -> dict[str, Any]:
try:
value: object = json.loads(path.read_text("utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise M48QNativeRiskQualityLabError(f"invalid JSON evidence: {path.name}") from exc
if not isinstance(value, dict):
raise M48QNativeRiskQualityLabError(f"JSON evidence must be an object: {path.name}")
return value
def _read_cases(path: Path) -> list[dict[str, Any]]:
cases: list[dict[str, Any]] = []
try:
with path.open("r", encoding="utf-8") as stream:
for line in stream:
value: object = json.loads(line)
if not isinstance(value, dict):
raise M48QNativeRiskQualityLabError("M4.8Q case ledger row is not an object")
cases.append(value)
except (OSError, json.JSONDecodeError) as exc:
raise M48QNativeRiskQualityLabError("M4.8Q case ledger cannot be read") from exc
return cases
src/k1link/web/app.py
+8 -1
View File
@@ -153,8 +153,8 @@ from k1link.web.runtime_readiness import (
build_runtime_readiness,
)
from k1link.web.session_api import build_session_router
from k1link.web.simulation_world_provider_api import build_simulation_world_provider_router
from k1link.web.simulation_projects_api import build_simulation_projects_router
from k1link.web.simulation_world_provider_api import build_simulation_world_provider_router
from k1link.web.system_telemetry_api import build_system_telemetry_router
from k1link.web.viewer_diagnostics_api import build_viewer_diagnostics_router
@@ -980,6 +980,13 @@ app.include_router(
/ "m48t-risk-quality"
/ "lab-results"
),
native_root_provider=lambda: (
REPOSITORY_ROOT
/ ".runtime"
/ "compute-experiments"
/ "m48t-risk-quality"
/ "native-lab-results"
),
)
)
app.include_router(
src/k1link/web/m48t_risk_quality_lab_api.py
+290 -66
View File
@@ -1,4 +1,4 @@
"""Read-only API for the sealed M4.8T quality and temporal LAB."""
"""Read-only API for the M4.8T legacy quality and M4.8Q native review lifecycle."""
from __future__ import annotations
@@ -8,7 +8,7 @@ import json
import re
from collections.abc import Callable
from pathlib import Path, PurePosixPath
from typing import Any, Final
from typing import Any, Final, Literal
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import FileResponse
@@ -18,30 +18,61 @@ from k1link.laboratory.evidence_report import (
LaboratoryEvidenceReportError,
verify_laboratory_evidence_result,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
CATALOG_SCHEMA as NATIVE_CATALOG_SCHEMA,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
LAB_SCHEMA as NATIVE_LAB_SCHEMA,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
REPORT_SCHEMA as NATIVE_REPORT_SCHEMA,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
RESULT_PREFIX as NATIVE_RESULT_PREFIX,
)
from k1link.laboratory.m48t_risk_quality_lab import (
CATALOG_SCHEMA,
LAB_SCHEMA,
REPORT_SCHEMA,
RESULT_PREFIX,
CATALOG_SCHEMA as LEGACY_CATALOG_SCHEMA,
)
from k1link.laboratory.m48t_risk_quality_lab import (
LAB_SCHEMA as LEGACY_LAB_SCHEMA,
)
from k1link.laboratory.m48t_risk_quality_lab import (
REPORT_SCHEMA as LEGACY_REPORT_SCHEMA,
)
from k1link.laboratory.m48t_risk_quality_lab import (
RESULT_PREFIX as LEGACY_RESULT_PREFIX,
)
from k1link.perception.fixed_class_detector_tournament import false_authority
RootProvider = Callable[[], Path | None]
RESULT_ID: Final = re.compile(rf"^{re.escape(RESULT_PREFIX)}[a-f0-9]{{64}}$")
CASE_ID: Final = re.compile(r"^[0-9]{12}$")
VIEW_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-view/v1"
CATALOG_VIEW_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-catalog/v1"
_DEFINITION: Final = LaboratoryEvidenceDefinition(
Variant = Literal["legacy", "native"]
LEGACY_RESULT_ID: Final = re.compile(rf"^{re.escape(LEGACY_RESULT_PREFIX)}[a-f0-9]{{64}}$")
NATIVE_RESULT_ID: Final = re.compile(rf"^{re.escape(NATIVE_RESULT_PREFIX)}[a-f0-9]{{64}}$")
LEGACY_CASE_ID: Final = re.compile(r"^[0-9]{12}$")
NATIVE_CASE_ID: Final = re.compile(r"^[0-9]{6}$")
LEGACY_VIEW_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-view/v1"
NATIVE_VIEW_SCHEMA: Final = "missioncore.m48q-native-risk-quality-view/v1"
CATALOG_VIEW_SCHEMA: Final = "missioncore.m48t-risk-quality-lifecycle-catalog/v1"
_LEGACY_DEFINITION: Final = LaboratoryEvidenceDefinition(
work_id="m48t-risk-quality-temporal",
runtime_relative_root=PurePosixPath("m48t-risk-quality/lab-results"),
result_id_prefix="m48t-risk-quality-temporal-lab",
document_name="manifest.json",
result_schema_version=LAB_SCHEMA,
result_schema_version=LEGACY_LAB_SCHEMA,
)
_NATIVE_DEFINITION: Final = LaboratoryEvidenceDefinition(
work_id="m48t-risk-quality-temporal",
runtime_relative_root=PurePosixPath("m48t-risk-quality/native-lab-results"),
result_id_prefix="m48q-native-risk-quality-lab",
document_name="manifest.json",
result_schema_version=NATIVE_LAB_SCHEMA,
)
def build_m48t_risk_quality_lab_router(
*, root_provider: RootProvider = lambda: None
*,
root_provider: RootProvider = lambda: None,
native_root_provider: RootProvider = lambda: None,
) -> APIRouter:
router = APIRouter(
prefix="/api/v1/laboratory/m48t/risk-quality",
@@ -50,15 +81,23 @@ def build_m48t_risk_quality_lab_router(
@router.get("/results")
def list_results(limit: int = Query(default=1, ge=1, le=10)) -> dict[str, object]:
root = _configured_root(root_provider)
if root is None:
return _empty_catalog(False)
candidates = _candidates(root)
configured = False
candidates: list[tuple[Path, Variant]] = []
providers: tuple[tuple[RootProvider, Variant, re.Pattern[str]], ...] = (
(root_provider, "legacy", LEGACY_RESULT_ID),
(native_root_provider, "native", NATIVE_RESULT_ID),
)
for provider, variant, pattern in providers:
root = _configured_root(provider)
if root is None:
continue
configured = True
candidates.extend((item, variant) for item in _candidates(root, pattern))
items: list[dict[str, object]] = []
invalid_total = 0
for candidate in candidates:
for candidate, variant in candidates:
try:
items.append(_project_result(candidate))
items.append(_project_result(candidate, variant))
except RuntimeError:
invalid_total += 1
items.sort(
@@ -67,7 +106,7 @@ def build_m48t_risk_quality_lab_router(
)
return {
"schema_version": CATALOG_VIEW_SCHEMA,
"configured": True,
"configured": configured,
"items": items[:limit],
"candidate_total": len(candidates),
"invalid_total": invalid_total,
@@ -77,18 +116,37 @@ def build_m48t_risk_quality_lab_router(
@router.get("/results/{result_id}")
def get_result(result_id: str) -> dict[str, object]:
try:
return _project_result(_resolve_candidate(root_provider, result_id))
candidate, variant = _resolve_candidate(
root_provider,
native_root_provider,
result_id,
)
return _project_result(candidate, variant)
except RuntimeError:
raise HTTPException(status_code=404, detail="M4.8T result not found") from None
raise HTTPException(
status_code=404,
detail="M4.8T/M4.8Q result not found",
) from None
@router.get("/results/{result_id}/review/{case_id}.jpg")
def get_review_image(result_id: str, case_id: str) -> FileResponse:
if CASE_ID.fullmatch(case_id) is None:
raise HTTPException(status_code=404, detail="M4.8T review case not found")
try:
loaded = _load_result(_resolve_candidate(root_provider, result_id))
candidate, variant = _resolve_candidate(
root_provider,
native_root_provider,
result_id,
)
invalid_case = (variant == "legacy" and LEGACY_CASE_ID.fullmatch(case_id) is None) or (
variant == "native" and NATIVE_CASE_ID.fullmatch(case_id) is None
)
if invalid_case:
raise RuntimeError("case identity is invalid")
loaded = _load_result(candidate, variant)
except RuntimeError:
raise HTTPException(status_code=404, detail="M4.8T result not found") from None
raise HTTPException(
status_code=404,
detail="M4.8T/M4.8Q review case not found",
) from None
descriptor = next(
(
item
@@ -98,8 +156,10 @@ def build_m48t_risk_quality_lab_router(
None,
)
if not isinstance(descriptor, dict):
raise HTTPException(status_code=404, detail="M4.8T review case not found")
candidate = loaded["root"]
raise HTTPException(
status_code=404,
detail="M4.8T/M4.8Q review case not found",
)
path = (candidate / str(descriptor["path"])).resolve()
if (
not path.is_relative_to(candidate)
@@ -108,7 +168,10 @@ def build_m48t_risk_quality_lab_router(
or descriptor.get("byte_length") != path.stat().st_size
or descriptor.get("sha256") != _sha256(path)
):
raise HTTPException(status_code=404, detail="M4.8T review case not found")
raise HTTPException(
status_code=404,
detail="M4.8T/M4.8Q review case not found",
)
return FileResponse(
path,
media_type="image/jpeg",
@@ -122,8 +185,12 @@ def build_m48t_risk_quality_lab_router(
return router
def _project_result(candidate: Path) -> dict[str, object]:
loaded = _load_result(candidate)
def _project_result(candidate: Path, variant: Variant) -> dict[str, object]:
loaded = _load_result(candidate, variant)
return _project_native(loaded) if variant == "native" else _project_legacy(loaded)
def _project_legacy(loaded: dict[str, Any]) -> dict[str, object]:
manifest = loaded["manifest"]
report = loaded["report"]
catalog = loaded["catalog"]
@@ -143,7 +210,8 @@ def _project_result(candidate: Path) -> dict[str, object]:
for item in catalog["cases"]
]
return {
"schema_version": VIEW_SCHEMA,
"schema_version": LEGACY_VIEW_SCHEMA,
"variant": "legacy-coco-quality",
"result_id": result_id,
"created_at_utc": manifest["created_at_utc"],
"status": manifest["status"],
@@ -165,26 +233,97 @@ def _project_result(candidate: Path) -> dict[str, object]:
}
def _load_result(candidate: Path) -> dict[str, Any]:
def _project_native(loaded: dict[str, Any]) -> dict[str, object]:
manifest = loaded["manifest"]
report = loaded["report"]
catalog = loaded["catalog"]
result_id = str(manifest["result_id"])
review_cases = [
{
"case_id": item["case_id"],
"sequence": item["sequence"],
"frame_id": item["frame_id"],
"evidence_time_ns": item["evidence_time_ns"],
"image_url": (
f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}"
f"/review/{item['case_id']}.jpg"
),
"media_type": item["media_type"],
"width": item["width"],
"height": item["height"],
"byte_length": item["byte_length"],
"sha256": item["sha256"],
"geometric_resampling": item["geometric_resampling"],
"selection_buckets": copy.deepcopy(item["selection_buckets"]),
"comparison": copy.deepcopy(item["comparison"]),
"proposals": copy.deepcopy(item["proposals"]),
}
for item in catalog["cases"]
]
return {
"schema_version": NATIVE_VIEW_SCHEMA,
"variant": "native-risk-review",
"result_id": result_id,
"created_at_utc": manifest["created_at_utc"],
"status": manifest["status"],
"source": copy.deepcopy(report["source"]),
"configuration": copy.deepcopy(report["configuration"]),
"method": copy.deepcopy(report["method"]),
"execution": copy.deepcopy(report["execution"]),
"metrics": copy.deepcopy(report["metrics"]),
"acceptance": copy.deepcopy(report["acceptance"]),
"decision": copy.deepcopy(report["decision"]),
"limitations": copy.deepcopy(report["limitations"]),
"review": {
"source_raster": copy.deepcopy(catalog["source_raster"]),
"overlay": copy.deepcopy(catalog["overlay"]),
"cases": review_cases,
},
"ground_truth": False,
"authority": copy.deepcopy(report["authority"]),
"access": "read-only",
}
def _load_result(candidate: Path, variant: Variant) -> dict[str, Any]:
pattern = NATIVE_RESULT_ID if variant == "native" else LEGACY_RESULT_ID
definition = _NATIVE_DEFINITION if variant == "native" else _LEGACY_DEFINITION
if (
not candidate.is_dir()
or candidate.is_symlink()
or RESULT_ID.fullmatch(candidate.name) is None
or pattern.fullmatch(candidate.name) is None
):
raise RuntimeError("M4.8T result candidate is invalid")
raise RuntimeError("M4.8 result candidate is invalid")
try:
verify_laboratory_evidence_result(_DEFINITION, candidate)
verify_laboratory_evidence_result(definition, candidate)
manifest = _read_object(candidate / "manifest.json")
report = _read_object(candidate / "report.json")
catalog = _read_object(candidate / "catalog.json")
except (LaboratoryEvidenceReportError, OSError, ValueError) as exc:
raise RuntimeError("M4.8T result integrity failed") from exc
raise RuntimeError("M4.8 result integrity failed") from exc
if variant == "native":
_validate_native(candidate, manifest, report, catalog)
else:
_validate_legacy(candidate, manifest, report, catalog)
return {
"root": candidate,
"manifest": manifest,
"report": report,
"catalog": catalog,
}
def _validate_legacy(
candidate: Path,
manifest: dict[str, Any],
report: dict[str, Any],
catalog: dict[str, Any],
) -> None:
identity = manifest.get("identity")
if (
manifest.get("schema_version") != LAB_SCHEMA
manifest.get("schema_version") != LEGACY_LAB_SCHEMA
or manifest.get("result_id") != candidate.name
or manifest.get("status")
!= "complete-quality-gate-failed-temporal-invariant-passed"
or manifest.get("status") != "complete-quality-gate-failed-temporal-invariant-passed"
or manifest.get("completed") is not True
or manifest.get("bounded_question_accepted") is not False
or manifest.get("ground_truth") is not False
@@ -194,47 +333,143 @@ def _load_result(candidate: Path) -> dict[str, Any]:
or not candidate.name.endswith(str(manifest.get("identity_sha256")))
or identity.get("authority") != false_authority()
or manifest.get("authority") != false_authority()
or report.get("schema_version") != REPORT_SCHEMA
or report.get("schema_version") != LEGACY_REPORT_SCHEMA
or report.get("result_id") != candidate.name
or report.get("authority") != false_authority()
or report.get("decision", {}).get("quality_accepted") is not False
or report.get("decision", {}).get("temporal_invariant_passed") is not True
or catalog.get("schema_version") != CATALOG_SCHEMA
or catalog.get("schema_version") != LEGACY_CATALOG_SCHEMA
or catalog.get("result_id") != candidate.name
or catalog.get("case_count") != 16
or not isinstance(catalog.get("cases"), list)
or len(catalog["cases"]) != 16
or any(not _valid_case(item) for item in catalog["cases"])
or any(not _valid_legacy_case(item) for item in catalog["cases"])
):
raise RuntimeError("M4.8T result contract changed")
return {"root": candidate, "manifest": manifest, "report": report, "catalog": catalog}
def _valid_case(value: object) -> bool:
def _validate_native(
candidate: Path,
manifest: dict[str, Any],
report: dict[str, Any],
catalog: dict[str, Any],
) -> None:
identity = manifest.get("identity")
cases = catalog.get("cases")
if (
manifest.get("schema_version") != NATIVE_LAB_SCHEMA
or manifest.get("result_id") != candidate.name
or manifest.get("status") != "complete-review-ready-quality-not-adjudicated"
or manifest.get("completed") is not True
or manifest.get("bounded_question_accepted") is not True
or manifest.get("ground_truth") is not False
or not isinstance(manifest.get("created_at_utc"), str)
or not isinstance(identity, dict)
or manifest.get("identity_sha256") != _canonical_sha256(identity)
or not candidate.name.endswith(str(manifest.get("identity_sha256")))
or identity.get("authority") != false_authority()
or manifest.get("authority") != false_authority()
or report.get("schema_version") != NATIVE_REPORT_SCHEMA
or report.get("result_id") != candidate.name
or report.get("authority") != false_authority()
or report.get("decision", {}).get("review_ready") is not True
or report.get("decision", {}).get("quality_evaluated") is not False
or report.get("decision", {}).get("candidate_accepted") is not False
or catalog.get("schema_version") != NATIVE_CATALOG_SCHEMA
or catalog.get("result_id") != candidate.name
or catalog.get("case_count") != 24
or catalog.get("ground_truth") is not False
or not isinstance(cases, list)
or len(cases) != 24
or any(not _valid_native_case(item) for item in cases)
):
raise RuntimeError("M4.8Q result contract changed")
def _valid_legacy_case(value: object) -> bool:
return (
isinstance(value, dict)
and isinstance(value.get("case_id"), str)
and CASE_ID.fullmatch(value["case_id"]) is not None
and LEGACY_CASE_ID.fullmatch(value["case_id"]) is not None
and value.get("image_id") == int(value["case_id"])
and value.get("path") == f"review/review-{value['case_id']}.jpg"
and value.get("media_type") == "image/jpeg"
and isinstance(value.get("byte_length"), int)
and _valid_file_proof(value)
)
def _valid_native_case(value: object) -> bool:
if not isinstance(value, dict) or not isinstance(value.get("case_id"), str):
return False
case_id = value["case_id"]
proposals = value.get("proposals")
return (
NATIVE_CASE_ID.fullmatch(case_id) is not None
and value.get("sequence") == int(case_id)
and value.get("frame_id") == f"frame-{case_id}"
and isinstance(value.get("evidence_time_ns"), int)
and value.get("path") == f"cases/frame-{case_id}.jpg"
and value.get("media_type") == "image/jpeg"
and value.get("width") == 800
and value.get("height") == 600
and value.get("geometric_resampling") is False
and isinstance(value.get("selection_buckets"), list)
and isinstance(value.get("comparison"), dict)
and isinstance(proposals, list)
and len(proposals) > 0
and all(_valid_native_proposal(item) for item in proposals)
and _valid_file_proof(value)
)
def _valid_native_proposal(value: object) -> bool:
if not isinstance(value, dict):
return False
box = value.get("box_xyxy")
return (
isinstance(value.get("proposal_id"), str)
and isinstance(value.get("class_name"), str)
and isinstance(value.get("risk_family"), str)
and isinstance(value.get("score"), (int, float))
and not isinstance(value.get("score"), bool)
and 0.25 <= value["score"] <= 1
and isinstance(box, list)
and len(box) == 4
and all(isinstance(item, (int, float)) and not isinstance(item, bool) for item in box)
and 0 <= box[0] < box[2] <= 800
and 0 <= box[1] < box[3] <= 600
)
def _valid_file_proof(value: dict[str, Any]) -> bool:
return (
isinstance(value.get("byte_length"), int)
and value["byte_length"] > 0
and isinstance(value.get("sha256"), str)
and re.fullmatch(r"[a-f0-9]{64}", value["sha256"]) is not None
)
def _resolve_candidate(provider: RootProvider, result_id: str) -> Path:
if RESULT_ID.fullmatch(result_id) is None:
raise HTTPException(status_code=404, detail="M4.8T result not found")
def _resolve_candidate(
legacy_provider: RootProvider,
native_provider: RootProvider,
result_id: str,
) -> tuple[Path, Variant]:
provider: RootProvider
variant: Variant
if NATIVE_RESULT_ID.fullmatch(result_id) is not None:
provider, variant = native_provider, "native"
elif LEGACY_RESULT_ID.fullmatch(result_id) is not None:
provider, variant = legacy_provider, "legacy"
else:
raise HTTPException(status_code=404, detail="M4.8 result not found")
root = _configured_root(provider)
if root is None:
raise HTTPException(status_code=404, detail="M4.8T result not found")
raise HTTPException(status_code=404, detail="M4.8 result not found")
candidate = (root / result_id).resolve()
if candidate.parent != root or candidate.is_symlink() or not candidate.is_dir():
raise HTTPException(status_code=404, detail="M4.8T result not found")
return candidate
raise HTTPException(status_code=404, detail="M4.8 result not found")
return candidate, variant
def _configured_root(provider: RootProvider) -> Path | None:
@@ -251,29 +486,18 @@ def _configured_root(provider: RootProvider) -> Path | None:
return root if root.is_dir() else None
def _candidates(root: Path) -> list[Path]:
def _candidates(root: Path, pattern: re.Pattern[str]) -> list[Path]:
return sorted(
(
item
for item in root.iterdir()
if item.is_dir() and not item.is_symlink() and RESULT_ID.fullmatch(item.name)
if item.is_dir() and not item.is_symlink() and pattern.fullmatch(item.name)
),
key=lambda item: item.stat().st_mtime_ns,
reverse=True,
)
def _empty_catalog(configured: bool) -> dict[str, object]:
return {
"schema_version": CATALOG_VIEW_SCHEMA,
"configured": configured,
"items": [],
"candidate_total": 0,
"invalid_total": 0,
"access": "read-only",
}
def _read_object(path: Path) -> dict[str, Any]:
value = json.loads(path.read_text("utf-8"))
if not isinstance(value, dict):
tests/test_laboratory_evidence_registry.py
+12 -2
View File
@@ -145,11 +145,21 @@ def test_product_registry_declares_every_advanced_evidence_source() -> None:
"m48t-risk-quality-temporal",
}
m48 = next(
item for item in registry.definitions
if item.work_id == "m48-object-centric-quality"
item for item in registry.definitions if item.work_id == "m48-object-centric-quality"
)
assert [variant.phase for variant in m48.evidence_variants] == ["review", "result"]
assert [variant.result_id_prefix for variant in m48.evidence_variants] == [
"m48-object-quality-pack",
"m48-object-quality-result",
]
m48t = next(
item for item in registry.definitions if item.work_id == "m48t-risk-quality-temporal"
)
assert [variant.phase for variant in m48t.evidence_variants] == [
"legacy-quality",
"result",
]
assert [variant.result_id_prefix for variant in m48t.evidence_variants] == [
"m48t-risk-quality-temporal-lab",
"m48q-native-risk-quality-lab",
]
tests/test_m48t_risk_quality_lab_api.py
+138
View File
@@ -6,6 +6,18 @@ from pathlib import Path
from fastapi import FastAPI
from fastapi.testclient import TestClient
from k1link.laboratory.m48q_native_risk_quality_lab import (
CATALOG_SCHEMA as NATIVE_CATALOG_SCHEMA,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
LAB_SCHEMA as NATIVE_LAB_SCHEMA,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
REPORT_SCHEMA as NATIVE_REPORT_SCHEMA,
)
from k1link.laboratory.m48q_native_risk_quality_lab import (
RESULT_PREFIX as NATIVE_RESULT_PREFIX,
)
from k1link.laboratory.m48t_risk_quality_lab import (
CATALOG_SCHEMA,
LAB_SCHEMA,
@@ -150,3 +162,129 @@ def test_m48t_lab_api_fails_closed_after_visual_tamper(tmp_path: Path) -> None:
catalog = client.get("/api/v1/laboratory/m48t/risk-quality/results")
assert catalog.json()["items"] == []
assert catalog.json()["invalid_total"] == 1
def _native_fixture(tmp_path: Path) -> tuple[TestClient, Path, str]:
root = tmp_path / "native-results"
root.mkdir()
identity = {"schema_version": NATIVE_LAB_SCHEMA, "authority": false_authority()}
identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest()
result_id = NATIVE_RESULT_PREFIX + identity_sha256
result_root = root / result_id
cases_root = result_root / "cases"
cases_root.mkdir(parents=True)
cases = []
image_paths = []
for index in range(24):
case_id = f"{index * 20 + 12:06d}"
image_path = cases_root / f"frame-{case_id}.jpg"
image_path.write_bytes(b"native-jpeg" + bytes([index]))
image_paths.append(image_path)
cases.append(
{
"case_id": case_id,
"sequence": int(case_id),
"frame_id": f"frame-{case_id}",
"evidence_time_ns": 35_000_000_000 + index * 1_000_000,
"path": f"cases/{image_path.name}",
"media_type": "image/jpeg",
"width": 800,
"height": 600,
"byte_length": image_path.stat().st_size,
"sha256": hashlib.sha256(image_path.read_bytes()).hexdigest(),
"geometric_resampling": False,
"selection_buckets": ["person"],
"comparison": {
"native_detection_count": 1,
"legacy_704_detection_count": 1,
"matched_detection_count_iou_at_least_0_5": 1,
},
"proposals": [
{
"proposal_id": f"proposal-{index}-0",
"class_name": "person",
"risk_family": "person",
"score": 0.75,
"box_xyxy": [10.0, 20.0, 100.0, 200.0],
}
],
}
)
catalog = {
"schema_version": NATIVE_CATALOG_SCHEMA,
"result_id": result_id,
"case_count": 24,
"source_raster": {"width": 800, "height": 600},
"overlay": {"client_rendered": True, "toggleable": True},
"ground_truth": False,
"cases": cases,
}
report = {
"schema_version": NATIVE_REPORT_SCHEMA,
"result_id": result_id,
"source": {},
"configuration": {},
"method": {},
"execution": {},
"metrics": {},
"acceptance": {},
"decision": {
"review_ready": True,
"quality_evaluated": False,
"candidate_accepted": False,
},
"limitations": [],
"authority": false_authority(),
}
catalog_path = result_root / "catalog.json"
report_path = result_root / "report.json"
_write_json(catalog_path, catalog)
_write_json(report_path, report)
artifacts = [
_descriptor(catalog_path, result_root, "visual-evidence-catalog", "application/json"),
_descriptor(report_path, result_root, "laboratory-report", "application/json"),
*[
_descriptor(path, result_root, "visual-evidence-native-raw-fisheye-frame", "image/jpeg")
for path in image_paths
],
]
manifest = {
"schema_version": NATIVE_LAB_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": "2026-08-26T09:45:15.574709Z",
"status": "complete-review-ready-quality-not-adjudicated",
"completed": True,
"bounded_question_accepted": True,
"ground_truth": False,
"authority": false_authority(),
"artifacts": artifacts,
}
_write_json(result_root / "manifest.json", manifest)
app = FastAPI()
app.include_router(build_m48t_risk_quality_lab_router(native_root_provider=lambda: root))
return TestClient(app), result_root, result_id
def test_m48q_lab_api_projects_native_raw_cases_without_quality_promotion(
tmp_path: Path,
) -> None:
client, result_root, result_id = _native_fixture(tmp_path)
catalog = client.get("/api/v1/laboratory/m48t/risk-quality/results")
assert catalog.status_code == 200
assert catalog.json()["items"][0]["variant"] == "native-risk-review"
result = client.get(f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}")
assert result.status_code == 200
assert result.json()["ground_truth"] is False
assert result.json()["decision"]["quality_evaluated"] is False
assert len(result.json()["review"]["cases"]) == 24
assert result.json()["review"]["cases"][0]["geometric_resampling"] is False
case_id = "000012"
image = client.get(
f"/api/v1/laboratory/m48t/risk-quality/results/{result_id}/review/{case_id}.jpg"
)
assert image.status_code == 200
assert image.content == (result_root / f"cases/frame-{case_id}.jpg").read_bytes()