feat(perception): mine native fisheye risk cases

config/perception/m48q-native-risk-case-mining-v1.json

@@ -0,0 +1,58 @@
+{
+  "schema_version": "missioncore.m48q-native-risk-case-mining-profile/v1",
+  "profile_id": "m48q-ravnoves00-native-raw-kb4-risk-review/v1",
+  "question": "Which real RAVNOVES00 raw-fisheye frames expose the native RF-DETR risk semantics that require operator review before independent route truth exists?",
+  "source": {
+    "source_id": "RAVNOVES00",
+    "frame_count": 4489,
+    "raster_width": 800,
+    "raster_height": 600,
+    "video_sha256": "cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8",
+    "geometric_resampling": false,
+    "rectification": false,
+    "warp": false
+  },
+  "candidate": {
+    "provider_id": "triton-rf-detr-large-coco-native-kb4-risk-fp16-shadow/v0",
+    "model_id": "rf_detr_large_native_kb4:1",
+    "preprocess_id": "raw-kb4-uint8-fused-mask-rgb-pad8-imagenet-trt/v0",
+    "engine_sha256": "b8a40b3580edff001ec9680de68707242294ff590ab296000fae371f1083f695",
+    "minimum_score": 0.25
+  },
+  "risk_families": {
+    "person": ["person"],
+    "animal": ["bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe"],
+    "light-road-user": ["bicycle", "motorcycle", "skateboard"],
+    "vehicle": ["car", "bus", "truck"]
+  },
+  "selection": {
+    "case_count": 24,
+    "minimum_sequence_separation": 12,
+    "low_confidence_maximum_score": 0.4,
+    "fisheye_edge_margin_fraction": 0.12,
+    "bucket_quotas": {
+      "person": 3,
+      "animal": 3,
+      "light-road-user": 3,
+      "vehicle": 3,
+      "low-confidence": 3,
+      "fisheye-edge": 3,
+      "native-fewer-than-legacy": 3,
+      "native-more-than-legacy": 3
+    }
+  },
+  "scope": {
+    "quality_evaluated": false,
+    "ground_truth": false,
+    "candidate_accepted": false,
+    "production_accepted": false,
+    "purpose": "bounded-native-risk-case-review"
+  },
+  "authority": {
+    "ground_truth": false,
+    "candidate_accepted": false,
+    "commands_enabled": false,
+    "actuation_allowed": false,
+    "navigation_or_safety_accepted": false
+  }
+}

experiments/perception/run_m48q_native_risk_case_mining_worker.py

@@ -0,0 +1,677 @@
+#!/usr/bin/env python3
+"""Mine bounded native raw-fisheye risk cases from immutable Worker 006 ledgers."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import sys
+from collections import Counter
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Final
+
+PROFILE_SCHEMA: Final = "missioncore.m48q-native-risk-case-mining-profile/v1"
+RESULT_SCHEMA: Final = "missioncore.m48q-native-risk-case-mining-result/v1"
+CASE_SCHEMA: Final = "missioncore.m48q-native-risk-review-case/v1"
+GRAPH_FRAME_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1"
+COMPARISON_FRAME_SCHEMA: Final = "missioncore.m48n-native-vs-704-frame/v0"
+FALSE_AUTHORITY: Final = {
+    "ground_truth": False,
+    "candidate_accepted": False,
+    "commands_enabled": False,
+    "actuation_allowed": False,
+    "navigation_or_safety_accepted": False,
+}
+
+
+class M48QCaseMiningError(RuntimeError):
+    """Raised when immutable native review evidence is inconsistent."""
+
+
+@dataclass(frozen=True, slots=True)
+class Proposal:
+    proposal_id: str
+    class_name: str
+    risk_family: str
+    score: float
+    box_xyxy: tuple[float, float, float, float]
+
+
+@dataclass(frozen=True, slots=True)
+class FrameCandidate:
+    sequence: int
+    frame_id: str
+    evidence_time_ns: int
+    proposals: tuple[Proposal, ...]
+    native_count: int
+    legacy_count: int
+    matched_count: int
+    buckets: frozenset[str]
+
+
+def canonical_json(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def sha256_path(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 _read_object(path: Path) -> dict[str, Any]:
+    try:
+        value: object = json.loads(path.read_text("utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise M48QCaseMiningError(f"cannot read JSON object: {path.name}") from exc
+    if not isinstance(value, dict):
+        raise M48QCaseMiningError(f"JSON evidence is not an object: {path.name}")
+    return value
+
+
+def _verify_hash(path: Path, expected: str, label: str) -> str:
+    actual = sha256_path(path)
+    if actual != expected:
+        raise M48QCaseMiningError(f"{label} SHA-256 changed: {actual}")
+    return actual
+
+
+def _risk_family_map(profile: Mapping[str, Any]) -> dict[str, str]:
+    raw = profile.get("risk_families")
+    if not isinstance(raw, dict):
+        raise M48QCaseMiningError("risk family profile is invalid")
+    result: dict[str, str] = {}
+    for family, classes in raw.items():
+        if (
+            family not in {"person", "animal", "light-road-user", "vehicle"}
+            or not isinstance(classes, list)
+            or not classes
+        ):
+            raise M48QCaseMiningError("risk family profile changed")
+        for class_name in classes:
+            if not isinstance(class_name, str) or class_name in result:
+                raise M48QCaseMiningError("risk classes are invalid or duplicated")
+            result[class_name] = family
+    return result
+
+
+def _proposal(
+    value: object,
+    *,
+    sequence: int,
+    profile: Mapping[str, Any],
+    families: Mapping[str, str],
+) -> Proposal:
+    if not isinstance(value, dict):
+        raise M48QCaseMiningError("detector proposal is not an object")
+    candidate = profile["candidate"]
+    expected_frame_id = f"frame-{sequence:06d}"
+    if (
+        value.get("frame_id") != expected_frame_id
+        or value.get("provider_id") != candidate["provider_id"]
+        or value.get("model_id") != candidate["model_id"]
+        or value.get("preprocess_id") != candidate["preprocess_id"]
+    ):
+        raise M48QCaseMiningError("native detector proposal identity changed")
+    proposal_id = value.get("proposal_id")
+    class_name = value.get("semantic_hint")
+    score = value.get("objectness")
+    region = value.get("region")
+    if (
+        not isinstance(proposal_id, str)
+        or not isinstance(class_name, str)
+        or class_name not in families
+        or not isinstance(score, (float, int))
+        or isinstance(score, bool)
+        or not math.isfinite(float(score))
+        or float(score) < candidate["minimum_score"]
+        or not isinstance(region, dict)
+    ):
+        raise M48QCaseMiningError("native detector proposal contract changed")
+    coordinates = tuple(region.get(key) for key in ("x_min", "y_min", "x_max", "y_max"))
+    if not all(
+        isinstance(item, (float, int)) and not isinstance(item, bool) and math.isfinite(float(item))
+        for item in coordinates
+    ):
+        raise M48QCaseMiningError("native detector box is invalid")
+    left, top, right, bottom = (float(item) for item in coordinates)
+    source = profile["source"]
+    if not (
+        0 <= left < right <= source["raster_width"] and 0 <= top < bottom <= source["raster_height"]
+    ):
+        raise M48QCaseMiningError("native detector box escaped the raw source raster")
+    return Proposal(
+        proposal_id=proposal_id,
+        class_name=class_name,
+        risk_family=families[class_name],
+        score=float(score),
+        box_xyxy=(left, top, right, bottom),
+    )
+
+
+def load_comparison_rows(path: Path, expected_count: int) -> dict[int, tuple[int, int, int]]:
+    rows: dict[int, tuple[int, int, int]] = {}
+    try:
+        with path.open("r", encoding="utf-8") as stream:
+            for line_number, line in enumerate(stream, start=1):
+                value = json.loads(line)
+                if not isinstance(value, dict):
+                    raise M48QCaseMiningError("comparison frame is not an object")
+                if value.get("schema_version") not in {None, COMPARISON_FRAME_SCHEMA}:
+                    raise M48QCaseMiningError("comparison frame schema changed")
+                sequence = value.get("frame_index")
+                counts = tuple(
+                    value.get(key)
+                    for key in (
+                        "native_detection_count",
+                        "baseline_detection_count",
+                        "matched_detection_count_iou_at_least_0_5",
+                    )
+                )
+                if (
+                    not isinstance(sequence, int)
+                    or isinstance(sequence, bool)
+                    or sequence in rows
+                    or not all(
+                        isinstance(item, int) and not isinstance(item, bool) and item >= 0
+                        for item in counts
+                    )
+                ):
+                    raise M48QCaseMiningError(f"comparison frame {line_number} is invalid")
+                rows[sequence] = counts
+    except (OSError, json.JSONDecodeError) as exc:
+        raise M48QCaseMiningError("comparison frame ledger cannot be read") from exc
+    if set(rows) != set(range(expected_count)):
+        raise M48QCaseMiningError("comparison frame ledger is incomplete")
+    return rows
+
+
+def _case_buckets(
+    proposals: Sequence[Proposal],
+    *,
+    native_count: int,
+    legacy_count: int,
+    profile: Mapping[str, Any],
+) -> frozenset[str]:
+    selection = profile["selection"]
+    buckets = {item.risk_family for item in proposals}
+    if any(item.score <= selection["low_confidence_maximum_score"] for item in proposals):
+        buckets.add("low-confidence")
+    width = profile["source"]["raster_width"]
+    height = profile["source"]["raster_height"]
+    margin = selection["fisheye_edge_margin_fraction"]
+    if any(
+        item.box_xyxy[0] <= width * margin
+        or item.box_xyxy[2] >= width * (1 - margin)
+        or item.box_xyxy[1] <= height * margin
+        or item.box_xyxy[3] >= height * (1 - margin)
+        for item in proposals
+    ):
+        buckets.add("fisheye-edge")
+    if native_count < legacy_count:
+        buckets.add("native-fewer-than-legacy")
+    elif native_count > legacy_count:
+        buckets.add("native-more-than-legacy")
+    return frozenset(buckets)
+
+
+def load_graph_candidates(
+    path: Path,
+    *,
+    profile: Mapping[str, Any],
+    comparisons: Mapping[int, tuple[int, int, int]],
+) -> list[FrameCandidate]:
+    families = _risk_family_map(profile)
+    expected_count = profile["source"]["frame_count"]
+    candidates: list[FrameCandidate] = []
+    try:
+        with path.open("r", encoding="utf-8") as stream:
+            for sequence, line in enumerate(stream):
+                value = json.loads(line)
+                if not isinstance(value, dict) or value.get("schema_version") != GRAPH_FRAME_SCHEMA:
+                    raise M48QCaseMiningError("graph frame schema changed")
+                source = value.get("source_envelope")
+                raw_proposals = value.get("detector_proposals")
+                if (
+                    not isinstance(source, dict)
+                    or source.get("sequence") != sequence
+                    or source.get("frame_id") != f"frame-{sequence:06d}"
+                    or not isinstance(raw_proposals, list)
+                    or value.get("authority") != FALSE_AUTHORITY
+                ):
+                    raise M48QCaseMiningError("graph frame identity or authority changed")
+                timestamps = source.get("timestamps")
+                evidence_time_ns = (
+                    timestamps.get("source_ns") if isinstance(timestamps, dict) else None
+                )
+                if (
+                    not isinstance(evidence_time_ns, int)
+                    or isinstance(evidence_time_ns, bool)
+                    or evidence_time_ns < 0
+                ):
+                    raise M48QCaseMiningError("graph frame evidence time is invalid")
+                proposals = tuple(
+                    _proposal(item, sequence=sequence, profile=profile, families=families)
+                    for item in raw_proposals
+                )
+                native_count, legacy_count, matched_count = comparisons[sequence]
+                if len(proposals) != native_count:
+                    raise M48QCaseMiningError("native graph/comparison detector count changed")
+                buckets = _case_buckets(
+                    proposals,
+                    native_count=native_count,
+                    legacy_count=legacy_count,
+                    profile=profile,
+                )
+                candidates.append(
+                    FrameCandidate(
+                        sequence=sequence,
+                        frame_id=source["frame_id"],
+                        evidence_time_ns=evidence_time_ns,
+                        proposals=proposals,
+                        native_count=native_count,
+                        legacy_count=legacy_count,
+                        matched_count=matched_count,
+                        buckets=buckets,
+                    )
+                )
+    except (OSError, json.JSONDecodeError) as exc:
+        raise M48QCaseMiningError("graph frame ledger cannot be read") from exc
+    if len(candidates) != expected_count:
+        raise M48QCaseMiningError("graph frame ledger is incomplete")
+    return candidates
+
+
+def _quantile_order(items: Sequence[FrameCandidate]) -> Iterable[FrameCandidate]:
+    if not items:
+        return ()
+    indexes: list[int] = []
+    left = 0
+    right = len(items) - 1
+    while left <= right:
+        middle = (left + right) // 2
+        indexes.append(middle)
+        if middle - left > 0:
+            indexes.append((left + middle - 1) // 2)
+        if right - middle > 0:
+            indexes.append((middle + 1 + right) // 2)
+        left += 1
+        right -= 1
+    seen: set[int] = set()
+    return (items[index] for index in indexes if not (index in seen or seen.add(index)))
+
+
+def select_cases(
+    candidates: Sequence[FrameCandidate],
+    *,
+    bucket_quotas: Mapping[str, int],
+    minimum_sequence_separation: int,
+) -> list[FrameCandidate]:
+    selected: list[FrameCandidate] = []
+    selected_sequences: set[int] = set()
+    for bucket, quota in bucket_quotas.items():
+        eligible = [item for item in candidates if bucket in item.buckets]
+        admitted = 0
+        for item in _quantile_order(eligible):
+            if item.sequence in selected_sequences:
+                continue
+            if any(
+                abs(item.sequence - prior.sequence) < minimum_sequence_separation
+                for prior in selected
+            ):
+                continue
+            selected.append(item)
+            selected_sequences.add(item.sequence)
+            admitted += 1
+            if admitted == quota:
+                break
+        if admitted != quota:
+            raise M48QCaseMiningError(
+                f"selection bucket {bucket} produced {admitted}/{quota} separated cases"
+            )
+    expected = sum(bucket_quotas.values())
+    if len(selected) != expected:
+        raise M48QCaseMiningError("selected case count changed")
+    return sorted(selected, key=lambda item: item.sequence)
+
+
+def render_selected_frames(
+    *,
+    video_path: Path,
+    selected: Sequence[FrameCandidate],
+    cases_root: Path,
+    width: int,
+    height: int,
+) -> dict[int, dict[str, object]]:
+    import av
+    from av.error import FFmpegError
+    from PIL import Image
+
+    wanted = {item.sequence for item in selected}
+    rendered: dict[int, dict[str, object]] = {}
+    cases_root.mkdir(mode=0o700, parents=True, exist_ok=False)
+    try:
+        with av.open(str(video_path), mode="r") as container:
+            streams = list(container.streams.video)
+            if len(streams) != 1:
+                raise M48QCaseMiningError("RAVNOVES00 video stream contract changed")
+            for sequence, frame in enumerate(container.decode(streams[0])):
+                if sequence not in wanted:
+                    continue
+                array = frame.to_ndarray(format="rgb24")
+                if array.shape != (height, width, 3):
+                    raise M48QCaseMiningError("decoded raw-fisheye raster changed")
+                name = f"frame-{sequence:06d}.jpg"
+                path = cases_root / name
+                Image.fromarray(array, mode="RGB").save(
+                    path,
+                    format="JPEG",
+                    quality=94,
+                    subsampling=0,
+                    optimize=False,
+                )
+                rendered[sequence] = {
+                    "path": f"cases/{name}",
+                    "media_type": "image/jpeg",
+                    "width": width,
+                    "height": height,
+                    "byte_length": path.stat().st_size,
+                    "sha256": sha256_path(path),
+                    "geometric_resampling": False,
+                }
+                if len(rendered) == len(wanted):
+                    break
+    except (FFmpegError, OSError) as exc:
+        raise M48QCaseMiningError("RAVNOVES00 raw-fisheye frames cannot be decoded") from exc
+    if set(rendered) != wanted:
+        raise M48QCaseMiningError("not every selected raw-fisheye frame was decoded")
+    return rendered
+
+
+def _case_document(
+    candidate: FrameCandidate,
+    *,
+    image: Mapping[str, object],
+) -> dict[str, object]:
+    return {
+        "schema_version": CASE_SCHEMA,
+        "case_id": f"{candidate.sequence:06d}",
+        "sequence": candidate.sequence,
+        "frame_id": candidate.frame_id,
+        "evidence_time_ns": candidate.evidence_time_ns,
+        "selection_buckets": sorted(candidate.buckets),
+        "comparison": {
+            "native_detection_count": candidate.native_count,
+            "legacy_704_detection_count": candidate.legacy_count,
+            "matched_detection_count_iou_at_least_0_5": candidate.matched_count,
+            "quality_interpretation": "diagnostic-only",
+        },
+        "image": dict(image),
+        "proposals": [
+            {
+                "proposal_id": item.proposal_id,
+                "class_name": item.class_name,
+                "risk_family": item.risk_family,
+                "score": round(item.score, 9),
+                "box_xyxy": [round(value, 6) for value in item.box_xyxy],
+            }
+            for item in candidate.proposals
+        ],
+        "ground_truth": False,
+        "quality_evaluated": False,
+        "authority": dict(FALSE_AUTHORITY),
+    }
+
+
+def run(args: argparse.Namespace) -> dict[str, object]:
+    for path in (
+        args.profile,
+        args.graph_result,
+        args.graph_frames,
+        args.comparison_result,
+        args.comparison_frames,
+        args.video,
+    ):
+        if path.is_symlink() or not path.is_file():
+            raise M48QCaseMiningError(f"required evidence is missing: {path.name}")
+    profile_sha256 = _verify_hash(args.profile, args.expected_profile_sha256, "profile")
+    graph_result_sha256 = _verify_hash(
+        args.graph_result, args.expected_graph_result_sha256, "graph result"
+    )
+    graph_frames_sha256 = _verify_hash(
+        args.graph_frames, args.expected_graph_frames_sha256, "graph frames"
+    )
+    comparison_result_sha256 = _verify_hash(
+        args.comparison_result,
+        args.expected_comparison_result_sha256,
+        "comparison result",
+    )
+    comparison_frames_sha256 = _verify_hash(
+        args.comparison_frames,
+        args.expected_comparison_frames_sha256,
+        "comparison frames",
+    )
+    video_sha256 = _verify_hash(args.video, args.expected_video_sha256, "video")
+    runner_sha256 = _verify_hash(Path(__file__), args.expected_runner_sha256, "runner")
+
+    profile = _read_object(args.profile)
+    graph_result = _read_object(args.graph_result)
+    comparison_result = _read_object(args.comparison_result)
+    source = profile.get("source")
+    candidate = profile.get("candidate")
+    selection = profile.get("selection")
+    scope = profile.get("scope")
+    if (
+        profile.get("schema_version") != PROFILE_SCHEMA
+        or not isinstance(source, dict)
+        or source.get("video_sha256") != video_sha256
+        or source.get("raster_width") != 800
+        or source.get("raster_height") != 600
+        or source.get("frame_count") != 4489
+        or source.get("geometric_resampling") is not False
+        or source.get("rectification") is not False
+        or source.get("warp") is not False
+        or not isinstance(candidate, dict)
+        or candidate.get("engine_sha256")
+        != "b8a40b3580edff001ec9680de68707242294ff590ab296000fae371f1083f695"
+        or not isinstance(selection, dict)
+        or selection.get("case_count") != 24
+        or not isinstance(selection.get("bucket_quotas"), dict)
+        or sum(selection["bucket_quotas"].values()) != 24
+        or scope
+        != {
+            "quality_evaluated": False,
+            "ground_truth": False,
+            "candidate_accepted": False,
+            "production_accepted": False,
+            "purpose": "bounded-native-risk-case-review",
+        }
+        or profile.get("authority") != FALSE_AUTHORITY
+    ):
+        raise M48QCaseMiningError("M4.8Q profile contract changed")
+    if (
+        graph_result.get("schema_version") != "missioncore.m48s-reference-graph-shadow-load/v5"
+        or graph_result.get("completed") is not True
+        or graph_result.get("identity", {}).get("detector_provider_id") != candidate["provider_id"]
+        or graph_result.get("identity", {}).get("inputs", {}).get("video") != video_sha256
+        or graph_result.get("execution", {}).get("frame_evidence", {}).get("sha256")
+        != graph_frames_sha256
+        or graph_result.get("authority") != FALSE_AUTHORITY
+        or graph_result.get("production_accepted") is not False
+    ):
+        raise M48QCaseMiningError("native graph result contract changed")
+    if (
+        comparison_result.get("schema_version") != "missioncore.m48n-native-vs-704-ravnoves00/v0"
+        or comparison_result.get("completed") is not True
+        or comparison_result.get("source", {}).get("video_sha256") != video_sha256
+        or comparison_result.get("source", {}).get("frame_count") != source["frame_count"]
+        or comparison_result.get("execution", {}).get("frames_sha256") != comparison_frames_sha256
+        or comparison_result.get("authority") != FALSE_AUTHORITY
+    ):
+        raise M48QCaseMiningError("native/legacy comparison result contract changed")
+
+    output_root = args.output_root
+    if output_root.exists():
+        raise M48QCaseMiningError("M4.8Q output root already exists")
+    output_root.mkdir(mode=0o700, parents=True)
+    comparisons = load_comparison_rows(args.comparison_frames, source["frame_count"])
+    candidates = load_graph_candidates(
+        args.graph_frames,
+        profile=profile,
+        comparisons=comparisons,
+    )
+    selected = select_cases(
+        candidates,
+        bucket_quotas=selection["bucket_quotas"],
+        minimum_sequence_separation=selection["minimum_sequence_separation"],
+    )
+    rendered = render_selected_frames(
+        video_path=args.video,
+        selected=selected,
+        cases_root=output_root / "cases",
+        width=source["raster_width"],
+        height=source["raster_height"],
+    )
+    cases = [
+        _case_document(
+            item,
+            image=rendered[item.sequence],
+        )
+        for item in selected
+    ]
+    cases_path = output_root / "cases.jsonl"
+    with cases_path.open("xb") as stream:
+        for case in cases:
+            stream.write(canonical_json(case) + b"\n")
+
+    class_counts = Counter(proposal.class_name for item in selected for proposal in item.proposals)
+    selected_bucket_coverage = Counter(bucket for item in selected for bucket in item.buckets)
+    identity = {
+        "schema_version": RESULT_SCHEMA,
+        "profile": {"profile_id": profile["profile_id"], "sha256": profile_sha256},
+        "source": {
+            "source_id": source["source_id"],
+            "video_sha256": video_sha256,
+            "graph_result_sha256": graph_result_sha256,
+            "graph_frames_sha256": graph_frames_sha256,
+            "comparison_result_sha256": comparison_result_sha256,
+            "comparison_frames_sha256": comparison_frames_sha256,
+        },
+        "candidate": dict(candidate),
+        "runner_sha256": runner_sha256,
+        "cases_sha256": sha256_path(cases_path),
+        "authority": dict(FALSE_AUTHORITY),
+    }
+    result: dict[str, object] = {
+        "schema_version": RESULT_SCHEMA,
+        "status": "complete-review-ready-quality-not-adjudicated",
+        "completed": True,
+        "worker": {"worker_id": "worker-006", "node": "DESKTOP-OPJ8J04"},
+        "identity": identity,
+        "report_identity_sha256": hashlib.sha256(canonical_json(identity)).hexdigest(),
+        "source": {
+            **dict(source),
+            "graph_result_sha256": graph_result_sha256,
+            "graph_frames_sha256": graph_frames_sha256,
+            "comparison_result_sha256": comparison_result_sha256,
+            "comparison_frames_sha256": comparison_frames_sha256,
+        },
+        "candidate": dict(candidate),
+        "selection": {
+            "case_count": len(cases),
+            "minimum_sequence_separation": selection["minimum_sequence_separation"],
+            "configured_bucket_quotas": selection["bucket_quotas"],
+            "selected_bucket_coverage": dict(sorted(selected_bucket_coverage.items())),
+            "selected_sequences": [item.sequence for item in selected],
+            "selected_class_counts": dict(sorted(class_counts.items())),
+        },
+        "artifacts": {
+            "cases": {
+                "path": "cases.jsonl",
+                "schema_version": CASE_SCHEMA,
+                "row_count": len(cases),
+                "sha256": sha256_path(cases_path),
+            },
+            "images": {
+                "root": "cases",
+                "count": len(rendered),
+                "geometric_resampling": False,
+            },
+        },
+        "decision": {
+            "quality_evaluated": False,
+            "ground_truth": False,
+            "candidate_accepted": False,
+            "production_accepted": False,
+            "next_action": "operator-adjudication-in-existing-m48-review-instrument",
+        },
+        "limitations": [
+            "The selected RAVNOVES00 cases are diagnostic samples, not independent truth.",
+            (
+                "Native-versus-legacy detection-count differences are not quality "
+                "verdicts because legacy 704 stretches the raw 4:3 raster."
+            ),
+            (
+                "No child/adult, behavior, physical track identity, navigation or "
+                "actuation claim is made."
+            ),
+            (
+                "JPEG review derivatives preserve the 800x600 raster without geometric "
+                "resampling but are not lossless source frames."
+            ),
+        ],
+        "authority": dict(FALSE_AUTHORITY),
+    }
+    (output_root / "result.json").write_bytes(canonical_json(result) + b"\n")
+    return result
+
+
+def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    for name in (
+        "profile",
+        "graph-result",
+        "graph-frames",
+        "comparison-result",
+        "comparison-frames",
+        "video",
+        "output-root",
+    ):
+        parser.add_argument(f"--{name}", type=Path, required=True)
+    for name in (
+        "profile",
+        "graph-result",
+        "graph-frames",
+        "comparison-result",
+        "comparison-frames",
+        "video",
+        "runner",
+    ):
+        parser.add_argument(f"--expected-{name}-sha256", required=True)
+    return parser.parse_args(argv)
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    try:
+        result = run(parse_args(argv))
+    except M48QCaseMiningError as exc:
+        print(f"M4.8Q case mining refused: {exc}", file=sys.stderr)
+        return 2
+    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

experiments/perception/worker/Invoke-M48QNativeRiskCaseMining.ps1

@@ -0,0 +1,207 @@
+[CmdletBinding()]
+param(
+    [Parameter(Mandatory = $true)]
+    [string]$ReleaseRoot,
+    [Parameter(Mandatory = $true)]
+    [ValidatePattern("^[A-Za-z0-9._-]{1,96}$")]
+    [string]$RunId,
+    [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m48q-native-risk-case-mining"
+)
+
+$ErrorActionPreference = "Stop"
+$ProgressPreference = "SilentlyContinue"
+
+function Assert-LastExitCode([string]$Operation) {
+    if ($LASTEXITCODE -ne 0) { throw "$Operation failed with exit code $LASTEXITCODE" }
+}
+
+function Get-Sha256([string]$Path) {
+    return (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant()
+}
+
+function Resolve-DDirectory([string]$Path, [string]$Label, [bool]$Create) {
+    if ($Create -and -not (Test-Path -LiteralPath $Path)) {
+        $null = New-Item -ItemType Directory -Path $Path
+    }
+    $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force
+    if (
+        -not $item.PSIsContainer -or
+        ($item.Attributes -band [IO.FileAttributes]::ReparsePoint) -or
+        [IO.Path]::GetPathRoot($item.FullName).TrimEnd("\") -ine "D:"
+    ) {
+        throw "$Label must be a real D: directory"
+    }
+    return $item.FullName
+}
+
+function Assert-RegularFile([string]$Path, [string]$Label) {
+    $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force
+    if ($item.PSIsContainer -or ($item.Attributes -band [IO.FileAttributes]::ReparsePoint)) {
+        throw "$Label must be a regular file"
+    }
+    return $item.FullName
+}
+
+function Convert-ToDockerPath([string]$Path) { return $Path.Replace("\", "/") }
+
+function Get-Container([string]$Name) {
+    $rows = @((& docker inspect $Name) | ConvertFrom-Json)
+    Assert-LastExitCode "Docker inspection for $Name"
+    if ($rows.Count -ne 1) { throw "Container identity for $Name is not unique" }
+    return $rows[0]
+}
+
+if ($env:COMPUTERNAME -cne "DESKTOP-OPJ8J04") {
+    throw "M4.8Q native risk case mining is pinned to DESKTOP-OPJ8J04"
+}
+
+$release = Resolve-DDirectory $ReleaseRoot "M4.8Q release root" $false
+$output = Resolve-DDirectory $OutputRoot "M4.8Q output root" $true
+$runOutput = Join-Path $output $RunId
+if (Test-Path -LiteralPath $runOutput) { throw "M4.8Q output already exists" }
+
+$runner = Assert-RegularFile (
+    Join-Path $release "run_m48q_native_risk_case_mining_worker.py"
+) "M4.8Q runner"
+$profile = Assert-RegularFile (
+    Join-Path $release "m48q-native-risk-case-mining-v1.json"
+) "M4.8Q profile"
+$runnerSha256 = Get-Sha256 $runner
+$profileSha256 = Get-Sha256 $profile
+if ($runnerSha256 -cne "39bb1d810b167d6e97dcb091505b52d4a812d69419dcb32fe23ea5d14d605d67") {
+    throw "M4.8Q runner SHA-256 changed"
+}
+if ($profileSha256 -cne "c455055e63578d505edc80b66d67e795916a3d0297cc18bfee5a6af7b032ceda") {
+    throw "M4.8Q profile SHA-256 changed"
+}
+
+$graphRoot = Resolve-DDirectory (
+    "D:\NDC_MISSIONCORE\runtime\results\m48n-native-reference-graph-shadow\ravnoves00-full-12fps-v0"
+) "M4.8Q graph evidence root" $false
+$comparisonRoot = Resolve-DDirectory (
+    "D:\NDC_MISSIONCORE\runtime\results\m48n-native-vs-704\ravnoves00-full-v0"
+) "M4.8Q comparison evidence root" $false
+$graphResult = Assert-RegularFile (Join-Path $graphRoot "result.json") "graph result"
+$graphFrames = Assert-RegularFile (Join-Path $graphRoot "frames.jsonl") "graph frames"
+$comparisonResult = Assert-RegularFile (Join-Path $comparisonRoot "result.json") "comparison result"
+$comparisonFrames = Assert-RegularFile (Join-Path $comparisonRoot "frames.jsonl") "comparison frames"
+$video = Assert-RegularFile (
+    "D:\NDC_MISSIONCORE\runtime\experiments\e46e\inputs\right-cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8.mp4"
+) "RAVNOVES00 video"
+
+$graphResultSha256 = Get-Sha256 $graphResult
+$graphFramesSha256 = Get-Sha256 $graphFrames
+$comparisonResultSha256 = Get-Sha256 $comparisonResult
+$comparisonFramesSha256 = Get-Sha256 $comparisonFrames
+$videoSha256 = Get-Sha256 $video
+if ($graphResultSha256 -cne "c5c3a831b1d1c3271161c91fa5c5c40533eb663ca288ed0220334a147aed6e15") {
+    throw "M4.8Q graph result SHA-256 changed"
+}
+if ($graphFramesSha256 -cne "b245af969600670d0975e89cae02b44206e3f1328eff48d5b4639b1b2ab57346") {
+    throw "M4.8Q graph frame evidence SHA-256 changed"
+}
+if ($comparisonResultSha256 -cne "fd91c2f1f477038d5af51656d510aba39d69ad2e1a387f15d98ade19dd3a0c49") {
+    throw "M4.8Q comparison result SHA-256 changed"
+}
+if ($comparisonFramesSha256 -cne "6a35125d4e003edfd8a33b4239bad0910f137fde3a770e9d5baab721cdc145b2") {
+    throw "M4.8Q comparison frame evidence SHA-256 changed"
+}
+if ($videoSha256 -cne "cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8") {
+    throw "RAVNOVES00 video SHA-256 changed"
+}
+
+$media = Resolve-DDirectory (
+    "D:\NDC_MISSIONCORE\runtime\derived\perception-e15-media-pyav180-lz445-v1"
+) "PyAV dependency" $false
+$pillow = Resolve-DDirectory (
+    "D:\NDC_MISSIONCORE\runtime\derived\perception-p0-env-v1"
+) "Pillow dependency" $false
+$image = (
+    "nvcr.io/nvidia/tritonserver:26.06-py3@" +
+    "sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794"
+)
+& docker image inspect $image *> $null
+Assert-LastExitCode "pinned M4.8Q image inspection"
+
+$canonicalTriton = Get-Container "ndc-mission-core-triton"
+if (-not $canonicalTriton.State.Running -or $canonicalTriton.State.Health.Status -cne "healthy") {
+    throw "Canonical Triton must remain healthy during M4.8Q mining"
+}
+$canonicalTritonId = [string]$canonicalTriton.Id
+$runnerName = "ndc-mission-core-m48q-native-risk-case-mining"
+if (& docker ps -a --format "{{.Names}}" --filter "name=^/$runnerName$") {
+    throw "M4.8Q bounded container already exists"
+}
+
+try {
+    & docker run `
+        --name $runnerName `
+        --label "com.nodedc.product=mission-core" `
+        --label "com.nodedc.stack=ndc-mission-core-compute" `
+        --label "com.nodedc.role=bounded-native-risk-case-mining" `
+        --label "com.nodedc.managed-by=codex-bounded-experiment" `
+        --network none `
+        --read-only `
+        --security-opt "no-new-privileges:true" `
+        --cap-drop ALL `
+        --pids-limit 128 `
+        --cpus 4 `
+        --memory 4g `
+        --tmpfs "/tmp:rw,noexec,nosuid,size=512m" `
+        -e "PYTHONDONTWRITEBYTECODE=1" `
+        -e "PYTHONPATH=/opt/media:/opt/pillow" `
+        -v ((Convert-ToDockerPath $release) + ":/release:ro") `
+        -v ((Convert-ToDockerPath $output) + ":/output:rw") `
+        -v ((Convert-ToDockerPath $media) + ":/opt/media:ro") `
+        -v ((Convert-ToDockerPath $pillow) + ":/opt/pillow:ro") `
+        -v ((Convert-ToDockerPath $graphRoot) + ":/evidence/graph:ro") `
+        -v ((Convert-ToDockerPath $comparisonRoot) + ":/evidence/comparison:ro") `
+        -v ((Convert-ToDockerPath $video) + ":/source/right.mp4:ro") `
+        --entrypoint python3 `
+        $image `
+        /release/run_m48q_native_risk_case_mining_worker.py `
+        --profile /release/m48q-native-risk-case-mining-v1.json `
+        --graph-result /evidence/graph/result.json `
+        --graph-frames /evidence/graph/frames.jsonl `
+        --comparison-result /evidence/comparison/result.json `
+        --comparison-frames /evidence/comparison/frames.jsonl `
+        --video /source/right.mp4 `
+        --output-root ("/output/{0}" -f $RunId) `
+        --expected-profile-sha256 $profileSha256 `
+        --expected-graph-result-sha256 $graphResultSha256 `
+        --expected-graph-frames-sha256 $graphFramesSha256 `
+        --expected-comparison-result-sha256 $comparisonResultSha256 `
+        --expected-comparison-frames-sha256 $comparisonFramesSha256 `
+        --expected-video-sha256 $videoSha256 `
+        --expected-runner-sha256 $runnerSha256
+    Assert-LastExitCode "M4.8Q native risk case mining"
+    if (
+        -not (Test-Path -LiteralPath (Join-Path $runOutput "result.json") -PathType Leaf) -or
+        -not (Test-Path -LiteralPath (Join-Path $runOutput "cases.jsonl") -PathType Leaf) -or
+        @(Get-ChildItem -LiteralPath (Join-Path $runOutput "cases") -Filter "*.jpg" -File).Count -ne 24
+    ) {
+        throw "M4.8Q result artifacts are incomplete"
+    }
+} finally {
+    if (& docker ps -a --format "{{.Names}}" --filter "name=^/$runnerName$") {
+        & docker rm -f $runnerName *> $null
+    }
+    $canonicalAfter = Get-Container "ndc-mission-core-triton"
+    if (
+        [string]$canonicalAfter.Id -cne $canonicalTritonId -or
+        -not $canonicalAfter.State.Running -or
+        $canonicalAfter.State.Health.Status -cne "healthy"
+    ) {
+        throw "Canonical Triton changed during M4.8Q mining"
+    }
+}
+
+$result = Get-Content -LiteralPath (Join-Path $runOutput "result.json") -Raw | ConvertFrom-Json
+[pscustomobject]@{
+    run_id = $RunId
+    output_root = $runOutput
+    status = $result.status
+    case_count = $result.selection.case_count
+    report_identity_sha256 = $result.report_identity_sha256
+    canonical_triton = "healthy-and-unchanged"
+} | ConvertTo-Json -Depth 4

tests/test_m48q_native_risk_case_mining.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+RUNNER_PATH = (
+    REPOSITORY_ROOT / "experiments" / "perception" / "run_m48q_native_risk_case_mining_worker.py"
+)
+SPEC = importlib.util.spec_from_file_location("m48q_case_mining", RUNNER_PATH)
+assert SPEC is not None and SPEC.loader is not None
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+
+
+def candidate(sequence: int, *buckets: str):
+    return MODULE.FrameCandidate(
+        sequence=sequence,
+        frame_id=f"frame-{sequence:06d}",
+        evidence_time_ns=sequence * 1_000_000,
+        proposals=(),
+        native_count=0,
+        legacy_count=0,
+        matched_count=0,
+        buckets=frozenset(buckets),
+    )
+
+
+def test_selection_is_deterministic_balanced_and_separated() -> None:
+    candidates = [
+        candidate(sequence, "person" if sequence % 2 == 0 else "vehicle")
+        for sequence in range(0, 400, 5)
+    ]
+    quotas = {"person": 3, "vehicle": 3}
+
+    first = MODULE.select_cases(
+        candidates,
+        bucket_quotas=quotas,
+        minimum_sequence_separation=10,
+    )
+    second = MODULE.select_cases(
+        candidates,
+        bucket_quotas=quotas,
+        minimum_sequence_separation=10,
+    )
+
+    assert [item.sequence for item in first] == [item.sequence for item in second]
+    assert len(first) == 6
+    assert all(
+        abs(left.sequence - right.sequence) >= 10
+        for index, left in enumerate(first)
+        for right in first[index + 1 :]
+    )
+
+
+def test_selection_refuses_missing_bucket_coverage() -> None:
+    with pytest.raises(MODULE.M48QCaseMiningError, match="animal produced 0/1"):
+        MODULE.select_cases(
+            [candidate(0, "person")],
+            bucket_quotas={"animal": 1},
+            minimum_sequence_separation=1,
+        )
+
+
+def test_profile_freezes_raw_raster_and_false_authority() -> None:
+    import json
+
+    profile = json.loads(
+        (
+            REPOSITORY_ROOT / "config" / "perception" / "m48q-native-risk-case-mining-v1.json"
+        ).read_text("utf-8")
+    )
+
+    assert profile["source"] == {
+        "source_id": "RAVNOVES00",
+        "frame_count": 4489,
+        "raster_width": 800,
+        "raster_height": 600,
+        "video_sha256": "cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8",
+        "geometric_resampling": False,
+        "rectification": False,
+        "warp": False,
+    }
+    assert sum(profile["selection"]["bucket_quotas"].values()) == 24
+    assert profile["scope"]["quality_evaluated"] is False
+    assert profile["authority"] == MODULE.FALSE_AUTHORITY