diff --git a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md index 8d9b310..29dea6e 100644 --- a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md +++ b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md @@ -2,7 +2,7 @@ Date: 2026-08-05 -Status: in progress; M4.0–M4.2 implemented, M4.3 runtime gate open +Status: in progress; M4.0–M4.2 implemented, M4.3 execution seam ready/runtime gate open Audit base: `1b3e0b3` on `feat/simulation-polygon-s1` @@ -369,6 +369,8 @@ Deliverables: - keep the E46J model, score, NMS, valid-FOV and full-frame profile frozen for the first reference run; - account for zero-proposal frames, pathological boxes and provider failures; +- seal a strict digest-bound frame ledger and capacity receipt tied to the exact + worker, container, image, artifact, source and Triton identities; - publish provider latency and GPU metrics through the common telemetry path. Exit: @@ -699,15 +701,34 @@ wrapper: - all 4,489 accepted immutable E46J frame documents and 15,499 detections map to the new proposal contract with exact frame accounting. +The fresh execution path is also product code rather than another experiment +runner: + +- `k1link.perception.detector_replay` performs one sequential provider request + per admitted envelope and seals completed, zero-proposal and failed outcomes; +- contracts/accounting, atomic publication, fail-closed validation and the Worker + CLI are separate modules; no laboratory or web module is imported; +- every result carries exact runtime/container/image/artifact/code identities, + actual model/config/mask digests, read-only source and same-host Triton topology + assertions, per-frame canonical proposals, end-to-end/core FPS, latency, + rejection and failure accounting; +- Triton requests pin `yolox_s` version `1` explicitly; the M4 CLI rejects a + non-loopback tensor endpoint and credentials embedded in its origin; +- the baseline validator now freezes calibration, detector parameters, + non-goals and the complete E15 rollback identity instead of merely retaining + those fields in JSON; +- `require_m4_detector_replay_acceptance` refuses short smoke runs, another + worker/node, failed frames, less than 4,489 frames or less than 10.004 FPS. + This increment does **not** claim a new 4,489-frame Triton execution. The existing E46J 47.840 FPS result remains the baseline evidence. A fresh provider execution requires a digest-bound shadow package; ad-hoc executable staging on Worker 006 is prohibited by the deployment canon. Therefore M4.3 runtime/capacity exit and its final checker remain open, and M4.4 does not start yet. -Validation at this increment: 76 focused-and-related tests and the complete -Python suite (`1221 passed, 1 skipped`). Scoped Ruff and strict mypy pass for the -new compute primitive and complete `src/k1link/perception` package. +Validation after adding the execution seam: 42 focused-and-related tests and the +complete Python suite (`1230 passed, 1 skipped`). Scoped Ruff and strict mypy pass +for the complete `src/k1link/perception` package and the frozen YOLOX primitive. ## Implementation order diff --git a/src/k1link/compute/yolox_object_detector.py b/src/k1link/compute/yolox_object_detector.py index 8545da5..5b338b8 100644 --- a/src/k1link/compute/yolox_object_detector.py +++ b/src/k1link/compute/yolox_object_detector.py @@ -143,11 +143,21 @@ class TritonHttpInferenceBackend: def __init__(self, endpoint: str, *, timeout_seconds: float = 60.0) -> None: parsed = urllib.parse.urlsplit(endpoint) - if parsed.scheme != "http" or not parsed.hostname or parsed.query or parsed.fragment: + if ( + parsed.scheme != "http" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): raise YoloxDetectorError("Triton endpoint must be an explicit HTTP origin") if not math.isfinite(timeout_seconds) or timeout_seconds <= 0: raise YoloxDetectorError("Triton timeout must be positive") - self.path = f"{parsed.path.rstrip('/')}/v2/models/{YOLOX_MODEL_ID}/infer" + self.path = ( + f"{parsed.path.rstrip('/')}/v2/models/{YOLOX_MODEL_ID}" + f"/versions/{YOLOX_MODEL_VERSION}/infer" + ) self.connection = http.client.HTTPConnection( parsed.hostname, parsed.port or 80, diff --git a/src/k1link/perception/__init__.py b/src/k1link/perception/__init__.py index 16681ec..f7fdca5 100644 --- a/src/k1link/perception/__init__.py +++ b/src/k1link/perception/__init__.py @@ -58,6 +58,21 @@ from .detector import ( FrozenYoloxDetectorProvider, proposals_from_detections, ) +from .detector_replay import run_detector_replay +from .detector_replay_result import ( + DETECTOR_REPLAY_FRAME_SCHEMA, + DETECTOR_REPLAY_RESULT_SCHEMA, + DETECTOR_RUNTIME_IDENTITY_SCHEMA, + M4_DETECTOR_REPLAY_GATE, + DetectorReplayFrame, + DetectorReplayGate, + DetectorReplayMetrics, + DetectorReplayResult, + DetectorReplayResultError, + DetectorRuntimeIdentity, + read_detector_replay_result, + require_m4_detector_replay_acceptance, +) from .graph import ( GRAPH_RESULT_SCHEMA, REFERENCE_GRAPH_ID, @@ -148,6 +163,19 @@ __all__ = [ "DetectorProviderSnapshot", "FrozenYoloxDetectorProvider", "proposals_from_detections", + "DETECTOR_REPLAY_FRAME_SCHEMA", + "DETECTOR_REPLAY_RESULT_SCHEMA", + "DETECTOR_RUNTIME_IDENTITY_SCHEMA", + "M4_DETECTOR_REPLAY_GATE", + "DetectorReplayFrame", + "DetectorReplayGate", + "DetectorReplayMetrics", + "DetectorReplayResult", + "DetectorReplayResultError", + "DetectorRuntimeIdentity", + "read_detector_replay_result", + "require_m4_detector_replay_acceptance", + "run_detector_replay", "REFERENCE_GRAPH_CONFIG_SCHEMA", "DetectorProvider", "GeometryAssociationProvider", diff --git a/src/k1link/perception/baseline.py b/src/k1link/perception/baseline.py index 52e6225..281e4d2 100644 --- a/src/k1link/perception/baseline.py +++ b/src/k1link/perception/baseline.py @@ -30,6 +30,9 @@ BASELINE_SOURCE_PACK_ID: Final = ( BASELINE_SOURCE_PACK_SHA256: Final = ( "0685d24219d8236caf8b7f1685e93f6d6b59e7fd015a768d88a92bbe8b154944" ) +BASELINE_PREPROCESS_PROFILE_SHA256: Final = ( + "19c17dbc23f2b1c539eb6b8214e69fa487ea3fec8db9d88768dffc714895fb88" +) _SHA256 = re.compile(r"^[a-f0-9]{64}$") _EXPECTED_EVIDENCE_ROLES: Final = { @@ -72,6 +75,60 @@ _SOURCE_KEYS: Final = { "source_pack_id", "source_pack_artifact_sha256", } +_EXPECTED_CALIBRATION: Final[dict[str, object]] = { + "slot": "camera_1", + "model": "KB4", + "sha256": "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9", + "valid_fov_result_id": ( + "valid-fov-mask-b4dd8ddf2b87c1d520ee8a0868c4fea062d7c14d1bae73ccabd3abe1f3acbac2" + ), + "valid_fov_mask_sha256": ( + "a40cee06b7c6f69b6a09a11563dcfd237f3de833b1ccd31459e66692e528ba63" + ), +} +_EXPECTED_DETECTOR: Final[dict[str, object]] = { + "provider_id": "triton-yolox-s-raw-kb4/v1", + "model_id": "yolox_s", + "model_version": 1, + "model_sha256": "c5c2d13e59ae883e6af3b45daea64af4833a4951c92d116ec270d9ddbe998063", + "config_sha256": "5795c737a7935a655961b069e8404d336d891f9762fb6dffb93956a076479604", + "preprocess_profile_sha256": BASELINE_PREPROCESS_PROFILE_SHA256, + "minimum_score": 0.5, + "nms_iou_threshold": 0.45, + "runtime": "NVIDIA Triton 2.70.0 ONNX Runtime GPU backend", +} +_EXPECTED_NON_GOALS: Final = ( + "semantic-class-quality", + "persistent-reidentification", + "physical-live-k1", + "physical-threat-authority", + "navigation-or-command-authority", + "second-source-transfer", + "second-worker-bootstrap", + "ros2-nav2-px4-gazebo-integration", + "deepstream-migration", +) +_EXPECTED_ROLLBACK: Final[dict[str, object]] = { + "worker_id": "worker-006", + "worker_node": "DESKTOP-OPJ8J04", + "container_name": "ndc-mission-core-perception-worker", + "container_image": ( + "nvcr.io/nvidia/tritonserver:26.06-py3@" + "sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" + ), + "worker_package_id": ( + "e15-worker-package-dbf55ccb75664b778a2e0f5d34284af10ec0971945bbbee67b27bc264b765b51" + ), + "runner_sha256": "86e9b25c80908a520ed483541b708cde1d1c74605e094053fcb62b610ecefb20", + "orchestrator_sha256": ( + "82de50ae83debe26fc5463799b1e1e15217a8db8cc090671ba2993200b8ed95c" + ), + "entrypoint": "python3 /runner/run_e15_shadow_inference.py serve", + "observed_container_id": ( + "db2024d05098a6beb6b73bbf43f02c88ede586a4664b2c91182f436a42b3e3ff" + ), + "observed_at_utc": "2026-08-05T10:30:00Z", +} class BaselineContractError(ValueError): @@ -129,25 +186,37 @@ def load_m4_baseline(path: Path) -> BaselineProfile: raise BaselineContractError("M4 source pack artifact identity changed") if source.get("camera_stream_sha256") != BASELINE_CAMERA_STREAM_SHA256: raise BaselineContractError("M4 camera stream identity changed") - modalities = _string_array(source.get("modalities"), "source modalities") - if set(modalities) != {"image", "registered-point-increment", "pose"}: - raise BaselineContractError("baseline source must bind image, points and pose") - if _integer(source.get("frame_count"), "source frame count") != 4489: + if source.get("frame_count") != 4489: raise BaselineContractError("baseline source frame count changed") + if source.get("duration_seconds") != 448.723: + raise BaselineContractError("baseline source duration changed") + if source.get("frame_rate") != 10.003944527024467: + raise BaselineContractError("baseline source frame rate changed") + modalities = _string_array(source.get("modalities"), "source modalities") + if modalities != ("image", "registered-point-increment", "pose"): + raise BaselineContractError("baseline source must bind image, points and pose") + + calibration = _object(document.get("calibration"), "calibration") + if calibration != _EXPECTED_CALIBRATION: + raise BaselineContractError("baseline calibration identity changed") + detector = _object(document.get("detector"), "detector") + if detector != _EXPECTED_DETECTOR: + raise BaselineContractError("baseline detector identity changed") authority = _object(document.get("authority"), "authority") - if authority.get("mode") != "replay-simulated": - raise BaselineContractError("M4 authority must remain replay-simulated") - for key in ( - "ground_truth", - "physical_live", - "physical_collision_accepted", - "commands_enabled", - "actuation_allowed", - "navigation_or_safety_accepted", - ): - if authority.get(key) is not False: - raise BaselineContractError(f"baseline authority {key} must remain false") + if authority != { + "mode": "replay-simulated", + "ground_truth": False, + "physical_live": False, + "physical_collision_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, + }: + raise BaselineContractError("M4 authority must remain replay-simulated and false") + + if _string_array(document.get("non_goals"), "non-goals") != _EXPECTED_NON_GOALS: + raise BaselineContractError("baseline non-goals changed") evidence_items = document.get("evidence") if not isinstance(evidence_items, list): @@ -161,15 +230,8 @@ def load_m4_baseline(path: Path) -> BaselineProfile: raise BaselineContractError("baseline evidence paths must be unique") rollback = _object(document.get("rollback"), "rollback") - if rollback.get("worker_id") != "worker-006": - raise BaselineContractError("rollback worker identity changed") - if rollback.get("worker_node") != "DESKTOP-OPJ8J04": - raise BaselineContractError("rollback worker node changed") - entrypoint = rollback.get("entrypoint") - if not isinstance(entrypoint, str) or "run_e15_shadow_inference.py serve" not in entrypoint: - raise BaselineContractError("rollback E15 process identity is missing") - _digest(rollback.get("runner_sha256"), "rollback runner digest") - _digest(rollback.get("orchestrator_sha256"), "rollback orchestrator digest") + if rollback != _EXPECTED_ROLLBACK: + raise BaselineContractError("rollback E15 identity changed") return BaselineProfile(path=path, document=document, evidence=evidence) diff --git a/src/k1link/perception/detector_replay.py b/src/k1link/perception/detector_replay.py new file mode 100644 index 0000000..8e866ff --- /dev/null +++ b/src/k1link/perception/detector_replay.py @@ -0,0 +1,113 @@ +"""Sequential product runner for the frozen detector replay/capacity gate.""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from pathlib import Path +from threading import Event + +from .baseline import BASELINE_SESSION_ID, BASELINE_SOURCE_ID +from .detector import FrozenYoloxDetectorProvider +from .detector_replay_result import ( + M4_DETECTOR_REPLAY_GATE, + DetectorReplayFrame, + DetectorReplayGate, + DetectorReplayResult, + DetectorReplayResultError, + DetectorRuntimeIdentity, + build_detector_replay_metrics, + seal_detector_replay_result, +) +from .providers import SourceProvider +from .recorded_source import RECORDED_SOURCE_PROVIDER_ID + + +def run_detector_replay( + *, + source: SourceProvider, + provider: FrozenYoloxDetectorProvider, + runtime: DetectorRuntimeIdentity, + output_root: Path, + gate: DetectorReplayGate = M4_DETECTOR_REPLAY_GATE, + stop_event: Event | None = None, + clock_ns: Callable[[], int] = time.perf_counter_ns, + created_at_utc: str | None = None, +) -> DetectorReplayResult: + """Execute one request per admitted frame and seal even a bounded failed run.""" + + if source.provider_id != RECORDED_SOURCE_PROVIDER_ID: + raise DetectorReplayResultError("detector replay source provider is not admitted") + stop = stop_event or Event() + frames: list[DetectorReplayFrame] = [] + source_failure_code: str | None = None + started_ns = int(clock_ns()) + packets = source.packets(stop) + try: + for packet in packets: + if len(frames) == gate.expected_frames: + source_failure_code = "source-frame-count-exceeded" + break + if ( + packet.envelope.sequence != len(frames) + or packet.envelope.source_id != BASELINE_SOURCE_ID + or packet.envelope.session_id != BASELINE_SESSION_ID + ): + source_failure_code = "source-envelope-sequence-or-identity-mismatch" + break + frame_started_ns = int(clock_ns()) + try: + proposals = provider.detect(packet) + except Exception as exc: + frames.append( + DetectorReplayFrame( + sequence=packet.envelope.sequence, + envelope=packet.envelope, + outcome="failed", + proposals=(), + duration_ns=max(0, int(clock_ns()) - frame_started_ns), + failure_code=type(exc).__name__, + ) + ) + break + frames.append( + DetectorReplayFrame( + sequence=packet.envelope.sequence, + envelope=packet.envelope, + outcome="completed", + proposals=proposals, + duration_ns=max(0, int(clock_ns()) - frame_started_ns), + ) + ) + except Exception as exc: + source_failure_code = type(exc).__name__ + finally: + close = getattr(packets, "close", None) + if callable(close): + try: + close() + except Exception as exc: + if source_failure_code is None: + source_failure_code = type(exc).__name__ + if stop.is_set() and source_failure_code is None: + source_failure_code = "execution-cancelled" + if len(frames) != gate.expected_frames and source_failure_code is None: + source_failure_code = "source-frame-count-mismatch" + ended_ns = int(clock_ns()) + metrics = build_detector_replay_metrics( + tuple(frames), + provider.snapshot(), + run_duration_ns=max(1, ended_ns - started_ns), + ) + return seal_detector_replay_result( + output_root=output_root, + frames=tuple(frames), + metrics=metrics, + runtime=runtime, + gate=gate, + source_failure_code=source_failure_code, + created_at_utc=created_at_utc, + ) + + +__all__ = ["run_detector_replay"] diff --git a/src/k1link/perception/detector_replay_cli.py b/src/k1link/perception/detector_replay_cli.py new file mode 100644 index 0000000..9a328b0 --- /dev/null +++ b/src/k1link/perception/detector_replay_cli.py @@ -0,0 +1,111 @@ +"""Worker entry point for the canonical frozen detector replay gate.""" + +from __future__ import annotations + +import argparse +import ipaddress +import json +import urllib.parse +from pathlib import Path +from threading import Event + +from k1link.compute.yolox_object_detector import ( + TritonHttpInferenceBackend, + load_valid_fov_mask, +) + +from .baseline import load_m4_baseline, verify_m4_baseline +from .detector import FrozenYoloxDetectorProvider +from .detector_replay import run_detector_replay +from .detector_replay_result import ( + DetectorReplayResultError, + DetectorRuntimeIdentity, + require_m4_detector_replay_acceptance, +) +from .recorded_source import ( + DecodedRecordedSource, + PyAvRecordedImageDecoder, + RecordedRavnoves00Source, + ReplayPacing, +) + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run the product-neutral M4 detector replay/capacity gate." + ) + parser.add_argument("--repository-root", type=Path, required=True) + parser.add_argument("--video", type=Path, required=True) + parser.add_argument("--valid-fov-mask", type=Path, required=True) + parser.add_argument("--triton-origin", required=True, type=_loopback_triton_origin) + parser.add_argument("--runtime-identity", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + return parser.parse_args() + + +def _loopback_triton_origin(value: str) -> str: + parsed = urllib.parse.urlsplit(value) + hostname = parsed.hostname + try: + loopback = hostname == "localhost" or ( + hostname is not None and ipaddress.ip_address(hostname).is_loopback + ) + except ValueError: + loopback = False + if not loopback: + raise argparse.ArgumentTypeError("M4 Triton origin must use worker-local loopback") + return value + + +def main() -> int: + args = _arguments() + repository_root = args.repository_root.resolve(strict=True) + baseline = load_m4_baseline( + repository_root / "config/perception/m4-recorded-realtime-baseline-v1.json" + ) + verify_m4_baseline(repository_root, baseline) + runtime_value = json.loads(args.runtime_identity.resolve(strict=True).read_text("utf-8")) + runtime = DetectorRuntimeIdentity.from_dict(runtime_value) + source = DecodedRecordedSource( + source=RecordedRavnoves00Source.from_repository( + repository_root, + pacing=ReplayPacing.UNCAPPED, + ), + decoder=PyAvRecordedImageDecoder(args.video), + ) + backend = TritonHttpInferenceBackend(args.triton_origin) + try: + result = run_detector_replay( + source=source, + provider=FrozenYoloxDetectorProvider( + mask=load_valid_fov_mask(args.valid_fov_mask), + backend=backend, + ), + runtime=runtime, + output_root=args.output_root, + stop_event=Event(), + ) + finally: + backend.close() + try: + require_m4_detector_replay_acceptance(result) + m4_accepted = True + except DetectorReplayResultError: + m4_accepted = False + print( + json.dumps( + { + "result_id": result.result_id, + "result_root": str(result.result_root), + "accepted": m4_accepted, + "metrics": result.metrics.to_dict(), + }, + sort_keys=True, + separators=(",", ":"), + ) + ) + return 0 if m4_accepted else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/perception/detector_replay_contracts.py b/src/k1link/perception/detector_replay_contracts.py new file mode 100644 index 0000000..c098bba --- /dev/null +++ b/src/k1link/perception/detector_replay_contracts.py @@ -0,0 +1,493 @@ +"""Strict contracts and accounting for detector replay evidence.""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +from k1link.compute.yolox_object_detector import ( + YOLOX_CONFIG_SHA256, + YOLOX_MODEL_SHA256, + YOLOX_VALID_FOV_SHA256, +) + +from .contracts import ObjectProposal2D, SourceEnvelope +from .detector import ( + FROZEN_YOLOX_MODEL_ID, + FROZEN_YOLOX_PREPROCESS_ID, + FROZEN_YOLOX_PROVIDER_ID, + DetectorProviderSnapshot, +) + +DETECTOR_REPLAY_RESULT_SCHEMA: Final = "missioncore.perception-detector-replay-result/v1" +DETECTOR_REPLAY_RECEIPT_SCHEMA: Final = "missioncore.perception-detector-replay-receipt/v1" +DETECTOR_REPLAY_FRAME_SCHEMA: Final = "missioncore.perception-detector-replay-frame/v1" +DETECTOR_RUNTIME_IDENTITY_SCHEMA: Final = "missioncore.perception-runtime-identity/v1" +DETECTOR_REPLAY_RESULT_PREFIX: Final = "m4-detector-replay-" +DETECTOR_REPLAY_MANIFEST_NAME: Final = "manifest.json" +DETECTOR_REPLAY_RECEIPT_NAME: Final = "receipt.json" +DETECTOR_REPLAY_FRAMES_NAME: Final = "frames.jsonl" + +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_GIT_REVISION = re.compile(r"^[a-f0-9]{40}$") +_RESULT_ID = re.compile(r"^m4-detector-replay-[a-f0-9]{64}$") +_DOCKER_IMAGE_ID = re.compile(r"^sha256:[a-f0-9]{64}$") +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/-]{0,255}$") + + +class DetectorReplayResultError(ValueError): + """Detector replay evidence is incomplete, mutable or internally inconsistent.""" + + +@dataclass(frozen=True, slots=True) +class DetectorReplayGate: + expected_frames: int + minimum_end_to_end_fps: float + + def __post_init__(self) -> None: + if self.expected_frames < 1: + raise DetectorReplayResultError("detector replay expected frames must be positive") + if ( + not math.isfinite(self.minimum_end_to_end_fps) + or self.minimum_end_to_end_fps <= 0.0 + ): + raise DetectorReplayResultError("detector replay minimum FPS must be positive") + + def to_dict(self) -> dict[str, object]: + return { + "expected_frames": self.expected_frames, + "minimum_end_to_end_fps": self.minimum_end_to_end_fps, + } + + @classmethod + def from_dict(cls, value: object) -> DetectorReplayGate: + document = _object(value, "detector replay gate") + _exact_keys(document, {"expected_frames", "minimum_end_to_end_fps"}, "gate") + return cls( + expected_frames=_integer(document.get("expected_frames"), "expected frames"), + minimum_end_to_end_fps=_number( + document.get("minimum_end_to_end_fps"), "minimum end-to-end FPS" + ), + ) + + +M4_DETECTOR_REPLAY_GATE: Final = DetectorReplayGate( + expected_frames=4489, + minimum_end_to_end_fps=10.004, +) + + +@dataclass(frozen=True, slots=True) +class DetectorRuntimeIdentity: + worker_id: str + worker_node: str + worker_container_id: str + worker_image_id: str + triton_container_id: str + triton_image_id: str + triton_model_sha256: str + triton_model_config_sha256: str + valid_fov_mask_sha256: str + artifact_sha256: str + code_revision: str + source_mount_read_only: bool + model_service_reused: bool + public_worker_port_added: bool + same_host_tensor_transport: bool + + def __post_init__(self) -> None: + _identifier(self.worker_id, "worker id") + _identifier(self.worker_node, "worker node") + for value, label in ( + (self.worker_container_id, "worker container id"), + (self.triton_container_id, "Triton container id"), + ): + if _SHA256.fullmatch(value) is None: + raise DetectorReplayResultError(f"{label} must be a full digest") + for value, label in ( + (self.worker_image_id, "worker image id"), + (self.triton_image_id, "Triton image id"), + ): + if _DOCKER_IMAGE_ID.fullmatch(value) is None: + raise DetectorReplayResultError(f"{label} must be a full image digest") + if self.triton_model_sha256 != YOLOX_MODEL_SHA256: + raise DetectorReplayResultError("runtime Triton model digest changed") + if self.triton_model_config_sha256 != YOLOX_CONFIG_SHA256: + raise DetectorReplayResultError("runtime Triton model config digest changed") + if self.valid_fov_mask_sha256 != YOLOX_VALID_FOV_SHA256: + raise DetectorReplayResultError("runtime valid-FOV mask digest changed") + if _SHA256.fullmatch(self.artifact_sha256) is None: + raise DetectorReplayResultError("worker artifact must be digest-bound") + if _GIT_REVISION.fullmatch(self.code_revision) is None: + raise DetectorReplayResultError("code revision must be a full Git revision") + if not self.source_mount_read_only: + raise DetectorReplayResultError("recorded source mount must be read-only") + if not self.model_service_reused: + raise DetectorReplayResultError("the admitted Triton service must be reused") + if self.public_worker_port_added: + raise DetectorReplayResultError("detector replay must not add a public worker port") + if not self.same_host_tensor_transport: + raise DetectorReplayResultError("full detector tensors must remain on the worker host") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": DETECTOR_RUNTIME_IDENTITY_SCHEMA, + "worker_id": self.worker_id, + "worker_node": self.worker_node, + "worker_container_id": self.worker_container_id, + "worker_image_id": self.worker_image_id, + "triton_container_id": self.triton_container_id, + "triton_image_id": self.triton_image_id, + "triton_model_sha256": self.triton_model_sha256, + "triton_model_config_sha256": self.triton_model_config_sha256, + "valid_fov_mask_sha256": self.valid_fov_mask_sha256, + "artifact_sha256": self.artifact_sha256, + "code_revision": self.code_revision, + "source_mount_read_only": self.source_mount_read_only, + "model_service_reused": self.model_service_reused, + "public_worker_port_added": self.public_worker_port_added, + "same_host_tensor_transport": self.same_host_tensor_transport, + } + + @classmethod + def from_dict(cls, value: object) -> DetectorRuntimeIdentity: + document = _object(value, "detector runtime identity") + fields = { + "worker_id", + "worker_node", + "worker_container_id", + "worker_image_id", + "triton_container_id", + "triton_image_id", + "triton_model_sha256", + "triton_model_config_sha256", + "valid_fov_mask_sha256", + "artifact_sha256", + "code_revision", + "source_mount_read_only", + "model_service_reused", + "public_worker_port_added", + "same_host_tensor_transport", + } + _exact_keys(document, fields | {"schema_version"}, "runtime identity") + if document.get("schema_version") != DETECTOR_RUNTIME_IDENTITY_SCHEMA: + raise DetectorReplayResultError("detector runtime identity schema changed") + return cls( + worker_id=_string(document.get("worker_id"), "worker id"), + worker_node=_string(document.get("worker_node"), "worker node"), + worker_container_id=_string( + document.get("worker_container_id"), "worker container id" + ), + worker_image_id=_string(document.get("worker_image_id"), "worker image id"), + triton_container_id=_string( + document.get("triton_container_id"), "Triton container id" + ), + triton_image_id=_string(document.get("triton_image_id"), "Triton image id"), + triton_model_sha256=_string( + document.get("triton_model_sha256"), "Triton model digest" + ), + triton_model_config_sha256=_string( + document.get("triton_model_config_sha256"), "Triton model config digest" + ), + valid_fov_mask_sha256=_string( + document.get("valid_fov_mask_sha256"), "valid-FOV mask digest" + ), + artifact_sha256=_string(document.get("artifact_sha256"), "artifact digest"), + code_revision=_string(document.get("code_revision"), "code revision"), + source_mount_read_only=_boolean( + document.get("source_mount_read_only"), "source mount read-only" + ), + model_service_reused=_boolean( + document.get("model_service_reused"), "model service reused" + ), + public_worker_port_added=_boolean( + document.get("public_worker_port_added"), "public worker port added" + ), + same_host_tensor_transport=_boolean( + document.get("same_host_tensor_transport"), "same-host tensor transport" + ), + ) + + +@dataclass(frozen=True, slots=True) +class DetectorReplayFrame: + sequence: int + envelope: SourceEnvelope + outcome: str + proposals: tuple[ObjectProposal2D, ...] + duration_ns: int + failure_code: str | None = None + + def __post_init__(self) -> None: + if self.sequence < 0 or self.envelope.sequence != self.sequence: + raise DetectorReplayResultError("detector replay frame sequence is invalid") + if self.outcome not in {"completed", "failed"}: + raise DetectorReplayResultError("detector replay frame outcome is invalid") + if self.duration_ns < 0: + raise DetectorReplayResultError("detector replay frame duration is invalid") + if self.outcome == "completed" and self.failure_code is not None: + raise DetectorReplayResultError("completed detector frame cannot carry a failure") + if self.outcome == "failed" and ( + self.proposals or self.failure_code is None or not self.failure_code + ): + raise DetectorReplayResultError("failed detector frame must carry one failure code") + proposal_ids: set[str] = set() + for proposal in self.proposals: + if ( + proposal.source_id != self.envelope.source_id + or proposal.frame_id != self.envelope.frame_id + or proposal.provider_id != FROZEN_YOLOX_PROVIDER_ID + or proposal.model_id != FROZEN_YOLOX_MODEL_ID + or proposal.preprocess_id != FROZEN_YOLOX_PREPROCESS_ID + or proposal.provider_tracklet is not None + or proposal.proposal_id in proposal_ids + ): + raise DetectorReplayResultError("detector replay proposal ownership changed") + proposal_ids.add(proposal.proposal_id) + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": DETECTOR_REPLAY_FRAME_SCHEMA, + "sequence": self.sequence, + "source_envelope": self.envelope.to_dict(), + "outcome": self.outcome, + "detector_request_count": 1, + "class_routing_used": False, + "duration_ns": self.duration_ns, + "failure_code": self.failure_code, + "proposals": [proposal.to_dict() for proposal in self.proposals], + } + + @classmethod + def from_dict(cls, value: object) -> DetectorReplayFrame: + document = _object(value, "detector replay frame") + _exact_keys( + document, + { + "schema_version", + "sequence", + "source_envelope", + "outcome", + "detector_request_count", + "class_routing_used", + "duration_ns", + "failure_code", + "proposals", + }, + "detector replay frame", + ) + if ( + document.get("schema_version") != DETECTOR_REPLAY_FRAME_SCHEMA + or document.get("detector_request_count") != 1 + or document.get("class_routing_used") is not False + ): + raise DetectorReplayResultError("detector replay frame contract changed") + proposals_value = document.get("proposals") + if not isinstance(proposals_value, list): + raise DetectorReplayResultError("detector replay proposals must be an array") + failure = document.get("failure_code") + if failure is not None and not isinstance(failure, str): + raise DetectorReplayResultError("detector replay failure code is invalid") + return cls( + sequence=_integer(document.get("sequence"), "frame sequence"), + envelope=SourceEnvelope.from_dict(document.get("source_envelope")), + outcome=_string(document.get("outcome"), "frame outcome"), + proposals=tuple(ObjectProposal2D.from_dict(item) for item in proposals_value), + duration_ns=_integer(document.get("duration_ns"), "frame duration"), + failure_code=failure, + ) + + +@dataclass(frozen=True, slots=True) +class DetectorReplayMetrics: + frame_count: int + completed_frame_count: int + failed_frame_count: int + proposal_count: int + zero_proposal_frame_count: int + semantic_hint_count: int + provider_tracklet_count: int + run_duration_ns: int + provider_core_duration_ns: int + end_to_end_fps: float + provider_core_fps: float + frame_latency_p50_ms: float + frame_latency_p95_ms: float + frame_latency_max_ms: float + rejected: tuple[tuple[str, int], ...] + + def to_dict(self) -> dict[str, object]: + return { + "frame_count": self.frame_count, + "completed_frame_count": self.completed_frame_count, + "failed_frame_count": self.failed_frame_count, + "proposal_count": self.proposal_count, + "zero_proposal_frame_count": self.zero_proposal_frame_count, + "semantic_hint_count": self.semantic_hint_count, + "provider_tracklet_count": self.provider_tracklet_count, + "run_duration_ns": self.run_duration_ns, + "provider_core_duration_ns": self.provider_core_duration_ns, + "end_to_end_fps": self.end_to_end_fps, + "provider_core_fps": self.provider_core_fps, + "frame_latency_p50_ms": self.frame_latency_p50_ms, + "frame_latency_p95_ms": self.frame_latency_p95_ms, + "frame_latency_max_ms": self.frame_latency_max_ms, + "rejected": [{"reason": key, "count": count} for key, count in self.rejected], + } + + +@dataclass(frozen=True, slots=True) +class DetectorReplayResult: + result_id: str + result_root: Path + accepted: bool + metrics: DetectorReplayMetrics + runtime: DetectorRuntimeIdentity + gate: DetectorReplayGate + frames: tuple[DetectorReplayFrame, ...] + manifest: dict[str, object] + receipt: dict[str, object] + + +def build_detector_replay_metrics( + frames: tuple[DetectorReplayFrame, ...], + snapshot: DetectorProviderSnapshot, + *, + run_duration_ns: int, +) -> DetectorReplayMetrics: + if run_duration_ns <= 0: + raise DetectorReplayResultError("detector replay duration must be positive") + completed = sum(frame.outcome == "completed" for frame in frames) + failed = len(frames) - completed + proposals = tuple(proposal for frame in frames for proposal in frame.proposals) + latencies_ms = sorted(frame.duration_ns / 1_000_000 for frame in frames) + metrics = DetectorReplayMetrics( + frame_count=len(frames), + completed_frame_count=completed, + failed_frame_count=failed, + proposal_count=len(proposals), + zero_proposal_frame_count=sum( + frame.outcome == "completed" and not frame.proposals for frame in frames + ), + semantic_hint_count=sum( + proposal.semantic_hint is not None for proposal in proposals + ), + provider_tracklet_count=sum( + proposal.provider_tracklet is not None for proposal in proposals + ), + run_duration_ns=run_duration_ns, + provider_core_duration_ns=snapshot.core_duration_ns, + end_to_end_fps=round(completed * 1_000_000_000 / run_duration_ns, 6), + provider_core_fps=( + round(completed * 1_000_000_000 / snapshot.core_duration_ns, 6) + if snapshot.core_duration_ns > 0 + else 0.0 + ), + frame_latency_p50_ms=round(_percentile(latencies_ms, 0.5), 6), + frame_latency_p95_ms=round(_percentile(latencies_ms, 0.95), 6), + frame_latency_max_ms=round(max(latencies_ms, default=0.0), 6), + rejected=snapshot.rejected, + ) + if ( + snapshot.input_frames != len(frames) + or snapshot.completed_frames != completed + or snapshot.failed_frames != failed + or snapshot.proposal_count != len(proposals) + or snapshot.zero_proposal_frames != metrics.zero_proposal_frame_count + ): + raise DetectorReplayResultError("provider and replay accounting disagree") + return metrics + + +def detector_replay_accepted( + metrics: DetectorReplayMetrics, + gate: DetectorReplayGate, + source_failure_code: str | None, +) -> bool: + return ( + source_failure_code is None + and metrics.frame_count == gate.expected_frames + and metrics.completed_frame_count == gate.expected_frames + and metrics.failed_frame_count == 0 + and metrics.provider_tracklet_count == 0 + and metrics.end_to_end_fps >= gate.minimum_end_to_end_fps + ) + + +def _percentile(values: list[float], fraction: float) -> float: + if not values: + return 0.0 + index = (len(values) - 1) * fraction + lower = math.floor(index) + upper = math.ceil(index) + if lower == upper: + return values[lower] + ratio = index - lower + return values[lower] * (1.0 - ratio) + values[upper] * ratio + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise DetectorReplayResultError(f"{label} must be an object") + return value + + +def _exact_keys(document: dict[str, object], expected: set[str], label: str) -> None: + if set(document) != expected: + raise DetectorReplayResultError(f"{label} fields are incompatible") + + +def _string(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise DetectorReplayResultError(f"{label} must be a nonempty string") + return value + + +def _identifier(value: str, label: str) -> None: + if _IDENTIFIER.fullmatch(value) is None: + raise DetectorReplayResultError(f"{label} is invalid") + + +def _integer(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise DetectorReplayResultError(f"{label} must be a nonnegative integer") + return value + + +def _number(value: object, label: str) -> float: + if not isinstance(value, int | float) or isinstance(value, bool): + raise DetectorReplayResultError(f"{label} must be numeric") + converted = float(value) + if not math.isfinite(converted): + raise DetectorReplayResultError(f"{label} must be finite") + return converted + + +def _boolean(value: object, label: str) -> bool: + if not isinstance(value, bool): + raise DetectorReplayResultError(f"{label} must be boolean") + return value + + +__all__ = [ + "DETECTOR_REPLAY_FRAME_SCHEMA", + "DETECTOR_REPLAY_FRAMES_NAME", + "DETECTOR_REPLAY_MANIFEST_NAME", + "DETECTOR_REPLAY_RECEIPT_NAME", + "DETECTOR_REPLAY_RECEIPT_SCHEMA", + "DETECTOR_REPLAY_RESULT_PREFIX", + "DETECTOR_REPLAY_RESULT_SCHEMA", + "DETECTOR_RUNTIME_IDENTITY_SCHEMA", + "M4_DETECTOR_REPLAY_GATE", + "DetectorReplayFrame", + "DetectorReplayGate", + "DetectorReplayMetrics", + "DetectorReplayResult", + "DetectorReplayResultError", + "DetectorRuntimeIdentity", + "build_detector_replay_metrics", + "detector_replay_accepted", +] diff --git a/src/k1link/perception/detector_replay_result.py b/src/k1link/perception/detector_replay_result.py new file mode 100644 index 0000000..fa99072 --- /dev/null +++ b/src/k1link/perception/detector_replay_result.py @@ -0,0 +1,220 @@ +"""Atomic publication API for product detector replay evidence.""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import uuid +from datetime import UTC, datetime +from pathlib import Path + +from k1link.compute.yolox_object_detector import ( + FROZEN_YOLOX_CONFIG, + YOLOX_CONFIG_SHA256, + YOLOX_MODEL_SHA256, + YOLOX_MODEL_VERSION, + YOLOX_VALID_FOV_SHA256, +) + +from .baseline import ( + BASELINE_CAMERA_STREAM_SHA256, + BASELINE_PREPROCESS_PROFILE_SHA256, + BASELINE_PROFILE_ID, + BASELINE_RECORDED_JOB_ID, + BASELINE_SESSION_ID, + BASELINE_SOURCE_ID, +) +from .contracts import FalseAuthority +from .detector import ( + FROZEN_YOLOX_MODEL_ID, + FROZEN_YOLOX_PREPROCESS_ID, + FROZEN_YOLOX_PROVIDER_ID, +) +from .detector_replay_contracts import ( + DETECTOR_REPLAY_FRAME_SCHEMA, + DETECTOR_REPLAY_FRAMES_NAME, + DETECTOR_REPLAY_MANIFEST_NAME, + DETECTOR_REPLAY_RECEIPT_NAME, + DETECTOR_REPLAY_RECEIPT_SCHEMA, + DETECTOR_REPLAY_RESULT_PREFIX, + DETECTOR_REPLAY_RESULT_SCHEMA, + DETECTOR_RUNTIME_IDENTITY_SCHEMA, + M4_DETECTOR_REPLAY_GATE, + DetectorReplayFrame, + DetectorReplayGate, + DetectorReplayMetrics, + DetectorReplayResult, + DetectorReplayResultError, + DetectorRuntimeIdentity, + build_detector_replay_metrics, + detector_replay_accepted, +) +from .detector_replay_validation import ( + read_detector_replay_result as _read_detector_replay_result, +) +from .detector_replay_validation import validate_frame_accounting +from .recorded_source import RECORDED_SOURCE_PROVIDER_ID + + +def seal_detector_replay_result( + *, + output_root: Path, + frames: tuple[DetectorReplayFrame, ...], + metrics: DetectorReplayMetrics, + runtime: DetectorRuntimeIdentity, + gate: DetectorReplayGate, + source_failure_code: str | None, + created_at_utc: str | None = None, +) -> DetectorReplayResult: + validate_frame_accounting(frames, metrics) + root = output_root.expanduser().absolute() + root.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = root / f".detector-replay.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + try: + frames_path = staging / DETECTOR_REPLAY_FRAMES_NAME + with frames_path.open("wb") as handle: + for frame in frames: + handle.write(_canonical_json(frame.to_dict()) + b"\n") + frames_sha256 = _file_sha256(frames_path) + accepted = detector_replay_accepted(metrics, gate, source_failure_code) + identity = { + "schema_version": DETECTOR_REPLAY_RESULT_SCHEMA, + "baseline_profile_id": BASELINE_PROFILE_ID, + "source": { + "provider_id": RECORDED_SOURCE_PROVIDER_ID, + "source_id": BASELINE_SOURCE_ID, + "session_id": BASELINE_SESSION_ID, + "camera_artifact_id": BASELINE_RECORDED_JOB_ID, + "camera_stream_sha256": BASELINE_CAMERA_STREAM_SHA256, + }, + "detector": { + "provider_id": FROZEN_YOLOX_PROVIDER_ID, + "model_id": FROZEN_YOLOX_MODEL_ID, + "model_version": YOLOX_MODEL_VERSION, + "model_sha256": YOLOX_MODEL_SHA256, + "model_config_sha256": YOLOX_CONFIG_SHA256, + "valid_fov_mask_sha256": YOLOX_VALID_FOV_SHA256, + "preprocess_id": FROZEN_YOLOX_PREPROCESS_ID, + "preprocess_profile_sha256": BASELINE_PREPROCESS_PROFILE_SHA256, + "minimum_score": FROZEN_YOLOX_CONFIG.minimum_score, + "nms_iou_threshold": FROZEN_YOLOX_CONFIG.nms_iou_threshold, + "target_class_ids": list(FROZEN_YOLOX_CONFIG.target_class_ids), + "single_inference_per_frame": True, + "class_routing_used": False, + "provider_tracklets_used": False, + }, + "runtime": runtime.to_dict(), + "gate": gate.to_dict(), + "metrics": metrics.to_dict(), + "source_failure_code": source_failure_code, + "frames_sha256": frames_sha256, + "accepted": accepted, + "authority": FalseAuthority().to_dict(), + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"{DETECTOR_REPLAY_RESULT_PREFIX}{identity_sha256}" + created = created_at_utc or datetime.now(UTC).isoformat(timespec="milliseconds").replace( + "+00:00", "Z" + ) + receipt = { + "schema_version": DETECTOR_REPLAY_RECEIPT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "created_at_utc": created, + "accepted": accepted, + "source_failure_code": source_failure_code, + "gate": gate.to_dict(), + "metrics": metrics.to_dict(), + "authority": FalseAuthority().to_dict(), + } + receipt_path = staging / DETECTOR_REPLAY_RECEIPT_NAME + _write_json(receipt_path, receipt) + manifest = { + "schema_version": DETECTOR_REPLAY_RESULT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": created, + "accepted": accepted, + "artifacts": [ + _artifact(receipt_path, "detector-replay-receipt"), + _artifact(frames_path, "detector-replay-frames"), + ], + } + _write_json(staging / DETECTOR_REPLAY_MANIFEST_NAME, manifest) + destination = root / result_id + if destination.exists(): + shutil.rmtree(staging) + return read_detector_replay_result(destination) + os.replace(staging, destination) + return read_detector_replay_result(destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + + +def read_detector_replay_result(root: Path) -> DetectorReplayResult: + return _read_detector_replay_result(root) + + +def require_m4_detector_replay_acceptance(result: DetectorReplayResult) -> None: + """Fail unless a result closes the exact M4.3 Worker 006 detector gate.""" + + if ( + not result.accepted + or result.gate != M4_DETECTOR_REPLAY_GATE + or result.runtime.worker_id != "worker-006" + or result.runtime.worker_node != "DESKTOP-OPJ8J04" + ): + raise DetectorReplayResultError("result does not close the M4.3 Worker 006 gate") + + +def _artifact(path: Path, role: str) -> dict[str, object]: + return { + "role": role, + "path": path.name, + "bytes": path.stat().st_size, + "sha256": _file_sha256(path), + } + + +def _write_json(path: Path, value: object) -> None: + path.write_bytes(_canonical_json(value) + b"\n") + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +__all__ = [ + "DETECTOR_REPLAY_FRAME_SCHEMA", + "DETECTOR_REPLAY_RESULT_SCHEMA", + "DETECTOR_RUNTIME_IDENTITY_SCHEMA", + "M4_DETECTOR_REPLAY_GATE", + "DetectorReplayFrame", + "DetectorReplayGate", + "DetectorReplayMetrics", + "DetectorReplayResult", + "DetectorReplayResultError", + "DetectorRuntimeIdentity", + "build_detector_replay_metrics", + "read_detector_replay_result", + "require_m4_detector_replay_acceptance", + "seal_detector_replay_result", +] diff --git a/src/k1link/perception/detector_replay_validation.py b/src/k1link/perception/detector_replay_validation.py new file mode 100644 index 0000000..957cca7 --- /dev/null +++ b/src/k1link/perception/detector_replay_validation.py @@ -0,0 +1,402 @@ +"""Fail-closed reader and consistency checks for detector replay evidence.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from k1link.compute.yolox_object_detector import ( + FROZEN_YOLOX_CONFIG, + YOLOX_CONFIG_SHA256, + YOLOX_MODEL_SHA256, + YOLOX_MODEL_VERSION, + YOLOX_VALID_FOV_SHA256, +) + +from .baseline import ( + BASELINE_CAMERA_STREAM_SHA256, + BASELINE_PREPROCESS_PROFILE_SHA256, + BASELINE_PROFILE_ID, + BASELINE_RECORDED_JOB_ID, + BASELINE_SESSION_ID, + BASELINE_SOURCE_ID, +) +from .contracts import FalseAuthority +from .detector import ( + FROZEN_YOLOX_MODEL_ID, + FROZEN_YOLOX_PREPROCESS_ID, + FROZEN_YOLOX_PROVIDER_ID, +) +from .detector_replay_contracts import ( + _RESULT_ID, + _SHA256, + DETECTOR_REPLAY_FRAMES_NAME, + DETECTOR_REPLAY_MANIFEST_NAME, + DETECTOR_REPLAY_RECEIPT_NAME, + DETECTOR_REPLAY_RECEIPT_SCHEMA, + DETECTOR_REPLAY_RESULT_PREFIX, + DETECTOR_REPLAY_RESULT_SCHEMA, + DetectorReplayFrame, + DetectorReplayGate, + DetectorReplayMetrics, + DetectorReplayResult, + DetectorReplayResultError, + DetectorRuntimeIdentity, + _exact_keys, + _integer, + _number, + _object, + _percentile, + _string, + detector_replay_accepted, +) +from .recorded_source import RECORDED_SOURCE_PROVIDER_ID + + +def read_detector_replay_result(root: Path) -> DetectorReplayResult: + resolved = root.resolve(strict=True) + if resolved.is_symlink() or _RESULT_ID.fullmatch(resolved.name) is None: + raise DetectorReplayResultError("detector replay result root is invalid") + manifest = _read_json(resolved / DETECTOR_REPLAY_MANIFEST_NAME) + _exact_keys( + manifest, + { + "schema_version", + "result_id", + "identity_sha256", + "identity", + "created_at_utc", + "accepted", + "artifacts", + }, + "detector replay manifest", + ) + identity = _object(manifest.get("identity"), "detector replay identity") + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + if ( + manifest.get("schema_version") != DETECTOR_REPLAY_RESULT_SCHEMA + or manifest.get("result_id") != resolved.name + or manifest.get("identity_sha256") != identity_sha256 + or resolved.name != f"{DETECTOR_REPLAY_RESULT_PREFIX}{identity_sha256}" + ): + raise DetectorReplayResultError("detector replay identity changed") + _validate_identity(identity) + runtime = DetectorRuntimeIdentity.from_dict(identity.get("runtime")) + gate = DetectorReplayGate.from_dict(identity.get("gate")) + artifacts_value = manifest.get("artifacts") + if not isinstance(artifacts_value, list) or len(artifacts_value) != 2: + raise DetectorReplayResultError("detector replay artifact inventory changed") + artifacts = { + _string(_object(item, "artifact").get("role"), "artifact role"): item + for item in artifacts_value + } + if set(artifacts) != {"detector-replay-receipt", "detector-replay-frames"}: + raise DetectorReplayResultError("detector replay artifact roles changed") + receipt_path = _validated_artifact( + resolved, artifacts["detector-replay-receipt"], DETECTOR_REPLAY_RECEIPT_NAME + ) + frames_path = _validated_artifact( + resolved, artifacts["detector-replay-frames"], DETECTOR_REPLAY_FRAMES_NAME + ) + if _file_sha256(frames_path) != identity.get("frames_sha256"): + raise DetectorReplayResultError("detector replay frame digest changed") + receipt = _read_json(receipt_path) + _validate_receipt(receipt, manifest, identity) + frames = tuple( + DetectorReplayFrame.from_dict(value) + for value in _read_jsonl(frames_path, "detector replay frames") + ) + metrics = _metrics_from_dict(identity.get("metrics")) + validate_frame_accounting(frames, metrics) + source_failure = identity.get("source_failure_code") + if source_failure is not None and not isinstance(source_failure, str): + raise DetectorReplayResultError("detector replay source failure code changed") + accepted = detector_replay_accepted(metrics, gate, source_failure) + if manifest.get("accepted") is not accepted or identity.get("accepted") is not accepted: + raise DetectorReplayResultError("detector replay acceptance changed") + return DetectorReplayResult( + result_id=resolved.name, + result_root=resolved, + accepted=accepted, + metrics=metrics, + runtime=runtime, + gate=gate, + frames=frames, + manifest=manifest, + receipt=receipt, + ) + + +def validate_frame_accounting( + frames: tuple[DetectorReplayFrame, ...], metrics: DetectorReplayMetrics +) -> None: + if metrics.run_duration_ns <= 0: + raise DetectorReplayResultError("detector replay duration must be positive") + if any(frame.sequence != index for index, frame in enumerate(frames)): + raise DetectorReplayResultError("detector replay frame sequence is incomplete") + if any( + frame.envelope.source_id != BASELINE_SOURCE_ID + or frame.envelope.session_id != BASELINE_SESSION_ID + for frame in frames + ): + raise DetectorReplayResultError("detector replay source ownership changed") + completed = sum(frame.outcome == "completed" for frame in frames) + failed = len(frames) - completed + proposals = tuple(proposal for frame in frames for proposal in frame.proposals) + latencies_ms = sorted(frame.duration_ns / 1_000_000 for frame in frames) + expected_end_to_end_fps = round( + completed * 1_000_000_000 / metrics.run_duration_ns, 6 + ) + expected_provider_fps = ( + round(completed * 1_000_000_000 / metrics.provider_core_duration_ns, 6) + if metrics.provider_core_duration_ns > 0 + else 0.0 + ) + if ( + metrics.frame_count != len(frames) + or metrics.completed_frame_count != completed + or metrics.failed_frame_count != failed + or metrics.proposal_count != len(proposals) + or metrics.zero_proposal_frame_count + != sum(frame.outcome == "completed" and not frame.proposals for frame in frames) + or metrics.semantic_hint_count + != sum(proposal.semantic_hint is not None for proposal in proposals) + or metrics.provider_tracklet_count + != sum(proposal.provider_tracklet is not None for proposal in proposals) + or metrics.end_to_end_fps != expected_end_to_end_fps + or metrics.provider_core_fps != expected_provider_fps + or metrics.frame_latency_p50_ms != round(_percentile(latencies_ms, 0.5), 6) + or metrics.frame_latency_p95_ms != round(_percentile(latencies_ms, 0.95), 6) + or metrics.frame_latency_max_ms != round(max(latencies_ms, default=0.0), 6) + ): + raise DetectorReplayResultError("detector replay metrics and frames disagree") + + +def _validate_identity(identity: dict[str, object]) -> None: + _exact_keys( + identity, + { + "schema_version", + "baseline_profile_id", + "source", + "detector", + "runtime", + "gate", + "metrics", + "source_failure_code", + "frames_sha256", + "accepted", + "authority", + }, + "detector replay identity", + ) + source = _object(identity.get("source"), "detector replay source") + detector = _object(identity.get("detector"), "detector replay provider") + if ( + identity.get("schema_version") != DETECTOR_REPLAY_RESULT_SCHEMA + or identity.get("baseline_profile_id") != BASELINE_PROFILE_ID + or source + != { + "provider_id": RECORDED_SOURCE_PROVIDER_ID, + "source_id": BASELINE_SOURCE_ID, + "session_id": BASELINE_SESSION_ID, + "camera_artifact_id": BASELINE_RECORDED_JOB_ID, + "camera_stream_sha256": BASELINE_CAMERA_STREAM_SHA256, + } + or detector + != { + "provider_id": FROZEN_YOLOX_PROVIDER_ID, + "model_id": FROZEN_YOLOX_MODEL_ID, + "model_version": YOLOX_MODEL_VERSION, + "model_sha256": YOLOX_MODEL_SHA256, + "model_config_sha256": YOLOX_CONFIG_SHA256, + "valid_fov_mask_sha256": YOLOX_VALID_FOV_SHA256, + "preprocess_id": FROZEN_YOLOX_PREPROCESS_ID, + "preprocess_profile_sha256": BASELINE_PREPROCESS_PROFILE_SHA256, + "minimum_score": FROZEN_YOLOX_CONFIG.minimum_score, + "nms_iou_threshold": FROZEN_YOLOX_CONFIG.nms_iou_threshold, + "target_class_ids": list(FROZEN_YOLOX_CONFIG.target_class_ids), + "single_inference_per_frame": True, + "class_routing_used": False, + "provider_tracklets_used": False, + } + or identity.get("authority") != FalseAuthority().to_dict() + or _SHA256.fullmatch(_string(identity.get("frames_sha256"), "frame digest")) is None + ): + raise DetectorReplayResultError("detector replay frozen identity changed") + + +def _validate_receipt( + receipt: dict[str, object], + manifest: dict[str, object], + identity: dict[str, object], +) -> None: + _exact_keys( + receipt, + { + "schema_version", + "result_id", + "identity_sha256", + "created_at_utc", + "accepted", + "source_failure_code", + "gate", + "metrics", + "authority", + }, + "detector replay receipt", + ) + if ( + receipt.get("schema_version") != DETECTOR_REPLAY_RECEIPT_SCHEMA + or receipt.get("result_id") != manifest.get("result_id") + or receipt.get("identity_sha256") != manifest.get("identity_sha256") + or receipt.get("created_at_utc") != manifest.get("created_at_utc") + or receipt.get("accepted") != manifest.get("accepted") + or receipt.get("source_failure_code") != identity.get("source_failure_code") + or receipt.get("gate") != identity.get("gate") + or receipt.get("metrics") != identity.get("metrics") + or receipt.get("authority") != identity.get("authority") + ): + raise DetectorReplayResultError("detector replay receipt changed") + + +def _metrics_from_dict(value: object) -> DetectorReplayMetrics: + document = _object(value, "detector replay metrics") + fields = { + "frame_count", + "completed_frame_count", + "failed_frame_count", + "proposal_count", + "zero_proposal_frame_count", + "semantic_hint_count", + "provider_tracklet_count", + "run_duration_ns", + "provider_core_duration_ns", + "end_to_end_fps", + "provider_core_fps", + "frame_latency_p50_ms", + "frame_latency_p95_ms", + "frame_latency_max_ms", + "rejected", + } + _exact_keys(document, fields, "detector replay metrics") + rejected_value = document.get("rejected") + if not isinstance(rejected_value, list): + raise DetectorReplayResultError("detector rejected accounting is invalid") + rejected: list[tuple[str, int]] = [] + for item in rejected_value: + row = _object(item, "detector rejection") + _exact_keys(row, {"reason", "count"}, "detector rejection") + rejected.append( + ( + _string(row.get("reason"), "rejection reason"), + _integer(row.get("count"), "rejection count"), + ) + ) + if rejected != sorted(rejected) or len({reason for reason, _ in rejected}) != len(rejected): + raise DetectorReplayResultError("detector rejection accounting is not canonical") + metrics = DetectorReplayMetrics( + frame_count=_integer(document.get("frame_count"), "frame count"), + completed_frame_count=_integer( + document.get("completed_frame_count"), "completed frame count" + ), + failed_frame_count=_integer(document.get("failed_frame_count"), "failed frame count"), + proposal_count=_integer(document.get("proposal_count"), "proposal count"), + zero_proposal_frame_count=_integer( + document.get("zero_proposal_frame_count"), "zero-proposal frame count" + ), + semantic_hint_count=_integer( + document.get("semantic_hint_count"), "semantic hint count" + ), + provider_tracklet_count=_integer( + document.get("provider_tracklet_count"), "provider tracklet count" + ), + run_duration_ns=_integer(document.get("run_duration_ns"), "run duration"), + provider_core_duration_ns=_integer( + document.get("provider_core_duration_ns"), "provider core duration" + ), + end_to_end_fps=_number(document.get("end_to_end_fps"), "end-to-end FPS"), + provider_core_fps=_number(document.get("provider_core_fps"), "provider core FPS"), + frame_latency_p50_ms=_number( + document.get("frame_latency_p50_ms"), "frame p50 latency" + ), + frame_latency_p95_ms=_number( + document.get("frame_latency_p95_ms"), "frame p95 latency" + ), + frame_latency_max_ms=_number( + document.get("frame_latency_max_ms"), "frame maximum latency" + ), + rejected=tuple(rejected), + ) + if metrics.run_duration_ns <= 0 or any( + value < 0.0 + for value in ( + metrics.end_to_end_fps, + metrics.provider_core_fps, + metrics.frame_latency_p50_ms, + metrics.frame_latency_p95_ms, + metrics.frame_latency_max_ms, + ) + ): + raise DetectorReplayResultError("detector replay runtime metrics are invalid") + return metrics + + +def _validated_artifact(root: Path, value: object, expected_name: str) -> Path: + artifact = _object(value, "detector replay artifact") + _exact_keys(artifact, {"role", "path", "bytes", "sha256"}, "artifact") + if artifact.get("path") != expected_name: + raise DetectorReplayResultError("detector replay artifact path changed") + path = (root / expected_name).resolve(strict=True) + if path.parent != root or path.is_symlink(): + raise DetectorReplayResultError("detector replay artifact escapes its result") + if ( + artifact.get("bytes") != path.stat().st_size + or artifact.get("sha256") != _file_sha256(path) + ): + raise DetectorReplayResultError("detector replay artifact changed") + return path + + +def _read_json(path: Path) -> dict[str, object]: + try: + value = json.loads(path.read_text("utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DetectorReplayResultError(f"cannot read detector replay JSON: {path.name}") from exc + return _object(value, path.name) + + +def _read_jsonl(path: Path, label: str) -> tuple[dict[str, object], ...]: + documents: list[dict[str, object]] = [] + try: + lines = path.read_text("utf-8").splitlines() + except OSError as exc: + raise DetectorReplayResultError(f"cannot read {label}") from exc + for line_number, line in enumerate(lines, start=1): + try: + value = json.loads(line) + except json.JSONDecodeError as exc: + raise DetectorReplayResultError(f"{label} line {line_number} is invalid") from exc + documents.append(_object(value, f"{label} line {line_number}")) + return tuple(documents) + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +__all__ = ["read_detector_replay_result", "validate_frame_accounting"] diff --git a/tests/test_detector_replay.py b/tests/test_detector_replay.py new file mode 100644 index 0000000..9521eb0 --- /dev/null +++ b/tests/test_detector_replay.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import argparse +import json +import math +from collections.abc import Iterator +from pathlib import Path +from threading import Event + +import numpy as np +import pytest +from numpy.typing import NDArray + +from k1link.compute.yolox_object_detector import ( + YOLOX_CONFIG_SHA256, + YOLOX_MODEL_SHA256, + YOLOX_VALID_FOV_SHA256, +) +from k1link.perception.contracts import ( + ClockBasis, + ModalityOutcome, + ModalityStatus, + SourceEnvelope, + TimestampBundle, +) +from k1link.perception.detector import FrozenYoloxDetectorProvider +from k1link.perception.detector_replay import run_detector_replay +from k1link.perception.detector_replay_cli import _loopback_triton_origin +from k1link.perception.detector_replay_result import ( + DetectorReplayGate, + DetectorReplayResultError, + DetectorRuntimeIdentity, + read_detector_replay_result, + require_m4_detector_replay_acceptance, +) +from k1link.perception.providers import SourcePacket + + +class _Source: + provider_id = "ravnoves00-recorded-source/v1" + + def __init__(self, packets: tuple[SourcePacket, ...]) -> None: + self._packets = packets + self.closed = False + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + try: + for packet in self._packets: + if stop_event.is_set(): + return + yield packet + finally: + self.closed = True + + +class _Resizer: + def resize( + self, + image: NDArray[np.uint8], + width: int, + height: int, + ) -> NDArray[np.uint8]: + assert image.shape == (600, 800, 3) + return np.zeros((height, width, 3), dtype=np.uint8) + + +class _Backend: + def infer(self, tensor: NDArray[np.float32]) -> NDArray[np.float32]: + assert tensor.shape == (1, 3, 640, 640) + output = np.zeros((1, 8400, 85), dtype=np.float32) + output[0, 0, :4] = [40.0, 30.0, math.log(10.0), math.log(10.0)] + output[0, 0, 4] = 0.9 + output[0, 0, 5] = 0.9 + return output + + +def _status() -> ModalityStatus: + return ModalityStatus(True, ModalityOutcome.AVAILABLE, "test-available") + + +def _packet(sequence: int, image: object) -> SourcePacket: + return SourcePacket( + envelope=SourceEnvelope( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id=f"frame-{sequence:06d}", + sequence=sequence, + timestamps=TimestampBundle( + utc_ns=1_000 + sequence, + monotonic_ns=2_000 + sequence, + source_ns=3_000 + sequence, + clock_basis=ClockBasis.RECORDED_HOST, + ), + source_age_ns=0, + binding_reason="test-recorded-source", + calibration_id="camera-1-kb4-test", + representation_id="registered-map-increment-v1", + image=_status(), + registered_point_increment=_status(), + pose=_status(), + ), + image_payload=image, + registered_point_increment_payload=("points", sequence), + pose_payload=("pose", sequence), + ) + + +def _runtime( + *, + source_mount_read_only: bool = True, + public_worker_port_added: bool = False, + same_host_tensor_transport: bool = True, +) -> DetectorRuntimeIdentity: + return DetectorRuntimeIdentity( + worker_id="worker-006", + worker_node="DESKTOP-OPJ8J04", + worker_container_id="1" * 64, + worker_image_id=f"sha256:{'2' * 64}", + triton_container_id="3" * 64, + triton_image_id=f"sha256:{'4' * 64}", + triton_model_sha256=YOLOX_MODEL_SHA256, + triton_model_config_sha256=YOLOX_CONFIG_SHA256, + valid_fov_mask_sha256=YOLOX_VALID_FOV_SHA256, + artifact_sha256="5" * 64, + code_revision="6" * 40, + source_mount_read_only=source_mount_read_only, + model_service_reused=True, + public_worker_port_added=public_worker_port_added, + same_host_tensor_transport=same_host_tensor_transport, + ) + + +def _provider(clock_values: tuple[int, ...]) -> FrozenYoloxDetectorProvider: + return FrozenYoloxDetectorProvider( + mask=np.ones((600, 800), dtype=np.bool_), + backend=_Backend(), + resizer=_Resizer(), + clock_ns=iter(clock_values).__next__, + ) + + +def test_replay_seals_and_reopens_exact_class_agnostic_capacity_receipt( + tmp_path: Path, +) -> None: + image = np.zeros((600, 800, 3), dtype=np.uint8) + result = run_detector_replay( + source=_Source((_packet(0, image), _packet(1, image))), + provider=_provider((100, 1_000_100, 2_000_100, 3_000_100)), + runtime=_runtime(), + output_root=tmp_path, + gate=DetectorReplayGate(expected_frames=2, minimum_end_to_end_fps=10.004), + clock_ns=iter((0, 10, 20, 30, 40, 100_000_000)).__next__, + created_at_utc="2026-08-05T12:00:00.000Z", + ) + + assert result.accepted is True + assert result.metrics.frame_count == 2 + assert result.metrics.completed_frame_count == 2 + assert result.metrics.proposal_count == 2 + assert result.metrics.semantic_hint_count == 2 + assert result.metrics.provider_tracklet_count == 0 + assert result.metrics.end_to_end_fps == 20.0 + assert result.metrics.provider_core_fps == 1000.0 + assert all(frame.to_dict()["class_routing_used"] is False for frame in result.frames) + assert all(frame.proposals[0].provider_tracklet is None for frame in result.frames) + + reopened = read_detector_replay_result(result.result_root) + assert reopened.result_id == result.result_id + assert reopened.receipt["accepted"] is True + assert reopened.runtime.worker_id == "worker-006" + with pytest.raises(DetectorReplayResultError, match="does not close"): + require_m4_detector_replay_acceptance(reopened) + + +def test_replay_seals_failed_frame_without_fabricating_missing_proposals(tmp_path: Path) -> None: + image = np.zeros((600, 800, 3), dtype=np.uint8) + source = _Source((_packet(0, image), _packet(1, "opaque-image"))) + result = run_detector_replay( + source=source, + provider=_provider((100, 200, 300, 400)), + runtime=_runtime(), + output_root=tmp_path, + gate=DetectorReplayGate(expected_frames=2, minimum_end_to_end_fps=1.0), + clock_ns=iter((0, 10, 20, 30, 40, 1_000_000_000)).__next__, + created_at_utc="2026-08-05T12:01:00.000Z", + ) + + assert result.accepted is False + assert result.metrics.completed_frame_count == 1 + assert result.metrics.failed_frame_count == 1 + assert result.frames[1].outcome == "failed" + assert result.frames[1].failure_code == "DetectorProviderError" + assert result.frames[1].proposals == () + assert source.closed is True + + +def test_replay_result_detects_artifact_tampering(tmp_path: Path) -> None: + image = np.zeros((600, 800, 3), dtype=np.uint8) + result = run_detector_replay( + source=_Source((_packet(0, image),)), + provider=_provider((100, 200)), + runtime=_runtime(), + output_root=tmp_path, + gate=DetectorReplayGate(expected_frames=1, minimum_end_to_end_fps=1.0), + clock_ns=iter((0, 10, 20, 1_000_000)).__next__, + created_at_utc="2026-08-05T12:02:00.000Z", + ) + frames_path = result.result_root / "frames.jsonl" + row = json.loads(frames_path.read_text("utf-8")) + row["class_routing_used"] = True + frames_path.write_text(json.dumps(row) + "\n", "utf-8") + + with pytest.raises(DetectorReplayResultError, match="artifact changed"): + read_detector_replay_result(result.result_root) + + +def test_runtime_identity_rejects_noncanonical_worker_topology() -> None: + with pytest.raises(DetectorReplayResultError, match="read-only"): + _runtime(source_mount_read_only=False) + with pytest.raises(DetectorReplayResultError, match="public worker port"): + _runtime(public_worker_port_added=True) + with pytest.raises(DetectorReplayResultError, match="worker host"): + _runtime(same_host_tensor_transport=False) + + +def test_m4_cli_rejects_remote_triton_tensor_transport() -> None: + assert _loopback_triton_origin("http://127.0.0.1:8000") == "http://127.0.0.1:8000" + with pytest.raises(argparse.ArgumentTypeError, match="worker-local loopback"): + _loopback_triton_origin("http://192.168.68.52:8000") diff --git a/tests/test_perception_architecture.py b/tests/test_perception_architecture.py index 555d10e..5512357 100644 --- a/tests/test_perception_architecture.py +++ b/tests/test_perception_architecture.py @@ -49,6 +49,29 @@ def test_m4_baseline_cannot_silently_select_another_source(tmp_path: Path) -> No load_m4_baseline(path) +@pytest.mark.parametrize( + ("section", "key", "value", "message"), + ( + ("calibration", "valid_fov_mask_sha256", "0" * 64, "calibration"), + ("detector", "minimum_score", 0.51, "detector"), + ("rollback", "worker_node", "worker-007", "rollback"), + ), +) +def test_m4_baseline_cannot_silently_tune_frozen_execution_identity( + tmp_path: Path, + section: str, + key: str, + value: object, + message: str, +) -> None: + document = json.loads(BASELINE_PATH.read_text("utf-8")) + document[section][key] = value + path = tmp_path / "baseline.json" + path.write_text(json.dumps(document), "utf-8") + with pytest.raises(BaselineContractError, match=message): + load_m4_baseline(path) + + def test_reuse_inventory_separates_primitives_from_historical_wrappers() -> None: document = validate_reuse_inventory(REUSE_PATH) assert document["rules"]["bulk_legacy_migration_required"] is False diff --git a/tests/test_yolox_detector_provider.py b/tests/test_yolox_detector_provider.py index 5ae5871..d492d5b 100644 --- a/tests/test_yolox_detector_provider.py +++ b/tests/test_yolox_detector_provider.py @@ -11,6 +11,7 @@ from numpy.typing import NDArray from k1link.compute.yolox_object_detector import ( FrozenYoloxConfig, + TritonHttpInferenceBackend, YoloxDetection, YoloxDetectorError, postprocess_yolox, @@ -173,6 +174,16 @@ def test_frozen_profile_rejects_in_place_threshold_tuning() -> None: FrozenYoloxConfig(minimum_score=0.51) +def test_triton_transport_pins_the_frozen_model_version() -> None: + backend = TritonHttpInferenceBackend("http://127.0.0.1:8000") + try: + assert backend.path == "/v2/models/yolox_s/versions/1/infer" + finally: + backend.close() + with pytest.raises(YoloxDetectorError, match="explicit HTTP origin"): + TritonHttpInferenceBackend("http://user:secret@127.0.0.1:8000") + + def test_all_4489_accepted_e46j_frames_map_to_product_contract_without_class_routing() -> None: assert E46J_FRAMES.is_file() image = np.zeros((600, 800, 3), dtype=np.uint8)