feat(perception): add RF-DETR upstream parity gate

config/perception/m48t-upstream-parity-v1.json

@@ -0,0 +1,50 @@
+{
+  "schema_version": "missioncore.m48t-upstream-parity-profile/v1",
+  "profile_id": "m48t-rf-detr-large-upstream-coco-parity/v1",
+  "model": {
+    "package": "rfdetr",
+    "package_version": "1.9.4",
+    "upstream_revision": "9b009fa928d6218320439803d1da01869a85c072",
+    "checkpoint": "rf-detr-large-2026.pth",
+    "checkpoint_sha256": "0f4e20e19a99c0f8a62b5685f57f6c8b5c371c59081feda6752a0561a79ccf38",
+    "resolution": [704, 704],
+    "maximum_query_class_pairs": 300
+  },
+  "dataset": {
+    "dataset_id": "coco-2017-val",
+    "image_count": 5000,
+    "annotations_sha256": "e8c7f7908f1d7278341fae127d0da654f102f11bd7b21d8aeefa635b8c810b6f",
+    "evaluation": "pycocotools.COCOeval/bbox",
+    "all_categories": true,
+    "official_crowd_ignore": true,
+    "custom_area_filtering": false
+  },
+  "providers": {
+    "pytorch": {
+      "preprocessing": "official-rfdetr-predict-direct-original-to-704",
+      "dtype": "float16",
+      "confidence_prefilter": 0.0
+    },
+    "tensorrt": {
+      "provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0",
+      "engine_sha256": "986399ce706b7380472cf5e473232249fed6e628971d8007f6609e83128d46b8",
+      "preprocessing": "official-rfdetr-direct-original-to-704",
+      "confidence_prefilter": 0.0
+    }
+  },
+  "published_reference": {
+    "coco_ap_50_95": 0.565,
+    "coco_ap_50": 0.751,
+    "maximum_absolute_reproduction_delta": 0.01
+  },
+  "parity_gates": {
+    "maximum_absolute_ap_50_95_delta": 0.005,
+    "maximum_absolute_ap_50_delta": 0.005
+  },
+  "authority": {
+    "candidate_accepted": false,
+    "commands_enabled": false,
+    "actuation_allowed": false,
+    "navigation_or_safety_accepted": false
+  }
+}

experiments/perception/run_m48t_upstream_parity_worker.py

@@ -0,0 +1,618 @@
+#!/usr/bin/env python3
+"""Reproduce official RF-DETR COCO quality and compare the pinned TensorRT engine."""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import gc
+import gzip
+import hashlib
+import importlib.metadata
+import io
+import json
+import math
+import statistics
+import time
+from collections.abc import Iterable, Mapping
+from pathlib import Path
+from typing import Any, Final
+
+import numpy as np
+from PIL import Image
+
+from k1link.perception.rf_detr_object_detector import (
+    COCO_SPARSE_IDS,
+    TritonRfDetrHttpInferenceBackend,
+)
+
+PROFILE_SCHEMA: Final = "missioncore.m48t-upstream-parity-profile/v1"
+REPORT_SCHEMA: Final = "missioncore.m48t-upstream-parity-report/v1"
+PROGRESS_SCHEMA: Final = "missioncore.m48t-upstream-parity-progress/v1"
+FALSE_AUTHORITY: Final = {
+    "candidate_accepted": False,
+    "commands_enabled": False,
+    "actuation_allowed": False,
+    "navigation_or_safety_accepted": False,
+}
+COCO_METRIC_NAMES: Final = (
+    "ap_50_95",
+    "ap_50",
+    "ap_75",
+    "ap_small",
+    "ap_medium",
+    "ap_large",
+    "ar_1",
+    "ar_10",
+    "ar_100",
+    "ar_small",
+    "ar_medium",
+    "ar_large",
+)
+
+
+def parse_arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--profile", type=Path, required=True)
+    parser.add_argument("--annotations", type=Path, required=True)
+    parser.add_argument("--images-root", type=Path, required=True)
+    parser.add_argument("--checkpoint", type=Path, required=True)
+    parser.add_argument("--triton-origin", default="http://127.0.0.1:8000")
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--pytorch-predictions", type=Path, required=True)
+    parser.add_argument("--tensorrt-predictions", type=Path, required=True)
+    parser.add_argument("--progress", type=Path, required=True)
+    parser.add_argument("--runtime-image", required=True)
+    parser.add_argument("--maximum-images", type=int, default=0)
+    return parser.parse_args()
+
+
+def main() -> None:
+    arguments = parse_arguments()
+    profile, profile_sha256 = load_profile(arguments.profile)
+    if arguments.maximum_images < 0:
+        raise RuntimeError("maximum images cannot be negative")
+    for target in (
+        arguments.output,
+        arguments.pytorch_predictions,
+        arguments.tensorrt_predictions,
+        arguments.progress,
+    ):
+        if target.exists():
+            raise RuntimeError(f"output already exists: {target}")
+        target.parent.mkdir(parents=True, exist_ok=True)
+
+    model_profile = _object(profile.get("model"), "model")
+    dataset_profile = _object(profile.get("dataset"), "dataset")
+    annotations_sha256 = sha256_path(arguments.annotations)
+    if annotations_sha256 != _string(dataset_profile, "annotations_sha256"):
+        raise RuntimeError("COCO annotations SHA-256 changed")
+    if sha256_path(arguments.checkpoint) != _string(model_profile, "checkpoint_sha256"):
+        raise RuntimeError("RF-DETR checkpoint SHA-256 changed")
+    if importlib.metadata.version("rfdetr") != _string(model_profile, "package_version"):
+        raise RuntimeError("RF-DETR package version changed")
+
+    coco_document = _object(json.loads(arguments.annotations.read_text("utf-8")), "COCO")
+    images = load_coco_images(coco_document)
+    expected_count = _integer(dataset_profile, "image_count")
+    if len(images) != expected_count:
+        raise RuntimeError("COCO image count changed")
+    if arguments.maximum_images:
+        images = images[: arguments.maximum_images]
+    if not images:
+        raise RuntimeError("COCO parity selection is empty")
+    category_ids_by_name = load_category_ids_by_name(coco_document)
+    image_ids = [_integer(item, "id") for item in images]
+    full_admission_run = len(images) == expected_count
+
+    started_at = time.time_ns()
+    with arguments.progress.open("x", encoding="utf-8") as progress:
+        pytorch_rows, pytorch_timing = run_pytorch_predictions(
+            images=images,
+            images_root=arguments.images_root,
+            checkpoint=arguments.checkpoint,
+            category_ids_by_name=category_ids_by_name,
+            progress=progress,
+        )
+        write_gzip_jsonl(arguments.pytorch_predictions, pytorch_rows)
+        pytorch_metrics, pytorch_summary = evaluate_coco(
+            annotations=arguments.annotations,
+            predictions=pytorch_rows,
+            image_ids=image_ids,
+        )
+        pytorch_count = len(pytorch_rows)
+        del pytorch_rows
+        gc.collect()
+
+        tensorrt_rows, tensorrt_timing = run_tensorrt_predictions(
+            images=images,
+            images_root=arguments.images_root,
+            triton_origin=arguments.triton_origin,
+            progress=progress,
+        )
+        write_gzip_jsonl(arguments.tensorrt_predictions, tensorrt_rows)
+        tensorrt_metrics, tensorrt_summary = evaluate_coco(
+            annotations=arguments.annotations,
+            predictions=tensorrt_rows,
+            image_ids=image_ids,
+        )
+        tensorrt_count = len(tensorrt_rows)
+        del tensorrt_rows
+        gc.collect()
+
+    decision = build_parity_decision(
+        profile=profile,
+        pytorch_metrics=pytorch_metrics,
+        tensorrt_metrics=tensorrt_metrics,
+        full_admission_run=full_admission_run,
+    )
+    completed_at = time.time_ns()
+    report: dict[str, object] = {
+        "schema_version": REPORT_SCHEMA,
+        "profile_id": _string(profile, "profile_id"),
+        "dataset": {
+            "dataset_id": _string(dataset_profile, "dataset_id"),
+            "image_count": len(images),
+            "full_admission_run": full_admission_run,
+            "all_categories": True,
+            "official_crowd_ignore": True,
+            "custom_area_filtering": False,
+            "confidence_prefilter": 0.0,
+        },
+        "providers": {
+            "pytorch": {
+                "prediction_count": pytorch_count,
+                "metrics": pytorch_metrics,
+                "coco_summary": pytorch_summary,
+                "timing": pytorch_timing,
+            },
+            "tensorrt": {
+                "prediction_count": tensorrt_count,
+                "metrics": tensorrt_metrics,
+                "coco_summary": tensorrt_summary,
+                "timing": tensorrt_timing,
+            },
+        },
+        "decision": decision,
+        "execution": {
+            "worker": "DESKTOP-OPJ8J04",
+            "runtime_image": arguments.runtime_image,
+            "started_at_unix_ns": started_at,
+            "completed_at_unix_ns": completed_at,
+            "duration_seconds": (completed_at - started_at) / 1_000_000_000,
+            "packages": package_versions(
+                ("rfdetr", "torch", "torchvision", "numpy", "pycocotools", "tritonclient")
+            ),
+        },
+        "provenance": {
+            "profile_sha256": profile_sha256,
+            "annotations_sha256": annotations_sha256,
+            "checkpoint_sha256": sha256_path(arguments.checkpoint),
+            "pytorch_predictions_sha256": sha256_path(arguments.pytorch_predictions),
+            "tensorrt_predictions_sha256": sha256_path(arguments.tensorrt_predictions),
+        },
+        "authority": FALSE_AUTHORITY,
+    }
+    report["report_identity_sha256"] = hashlib.sha256(canonical_json(report).encode()).hexdigest()
+    arguments.output.write_text(canonical_json(report) + "\n", "utf-8")
+    print(
+        canonical_json(
+            {
+                "result": str(arguments.output),
+                "image_count": len(images),
+                "pytorch_ap_50_95": pytorch_metrics["ap_50_95"],
+                "tensorrt_ap_50_95": tensorrt_metrics["ap_50_95"],
+                "diagnosis": decision["diagnosis"],
+                "passed": decision["passed"],
+            }
+        ),
+        flush=True,
+    )
+
+
+def load_profile(path: Path) -> tuple[dict[str, object], str]:
+    raw = path.read_bytes()
+    profile = _object(json.loads(raw), "profile")
+    if profile.get("schema_version") != PROFILE_SCHEMA:
+        raise RuntimeError("upstream parity profile schema changed")
+    if _object(profile.get("authority"), "authority") != FALSE_AUTHORITY:
+        raise RuntimeError("upstream parity authority changed")
+    return profile, hashlib.sha256(raw).hexdigest()
+
+
+def load_coco_images(document: Mapping[str, object]) -> list[dict[str, object]]:
+    values = document.get("images")
+    if not isinstance(values, list):
+        raise RuntimeError("COCO images are invalid")
+    images = [_object(value, "COCO image") for value in values]
+    for image in images:
+        _integer(image, "id")
+        _positive_integer(image, "width")
+        _positive_integer(image, "height")
+        _string(image, "file_name")
+    return sorted(images, key=lambda item: _integer(item, "id"))
+
+
+def load_category_ids_by_name(document: Mapping[str, object]) -> dict[str, int]:
+    values = document.get("categories")
+    if not isinstance(values, list):
+        raise RuntimeError("COCO categories are invalid")
+    result: dict[str, int] = {}
+    for value in values:
+        category = _object(value, "COCO category")
+        name = _string(category, "name")
+        category_id = _positive_integer(category, "id")
+        if name in result:
+            raise RuntimeError("COCO category names are duplicated")
+        result[name] = category_id
+    if set(result.values()) != set(COCO_SPARSE_IDS):
+        raise RuntimeError("COCO sparse category ids changed")
+    return result
+
+
+def run_pytorch_predictions(
+    *,
+    images: list[dict[str, object]],
+    images_root: Path,
+    checkpoint: Path,
+    category_ids_by_name: Mapping[str, int],
+    progress: Any,
+) -> tuple[list[dict[str, float | int | list[float]]], dict[str, object]]:
+    import torch  # type: ignore[import-not-found]
+    from rfdetr import RFDETRLarge  # type: ignore[import-not-found]
+
+    started_at = time.perf_counter_ns()
+    model = RFDETRLarge(pretrain_weights=str(checkpoint))
+    model.inference(compile=False, dtype=torch.float16, inplace=True)
+    rows: list[dict[str, float | int | list[float]]] = []
+    timings: list[float] = []
+    try:
+        for index, image in enumerate(images, start=1):
+            image_started = time.perf_counter_ns()
+            image_path = (images_root / _string(image, "file_name")).resolve(strict=True)
+            with Image.open(image_path) as opened:
+                prediction = model.predict(
+                    opened.convert("RGB"), threshold=0.0, include_source_image=False
+                )
+            boxes = np.asarray(prediction.xyxy, dtype=np.float32)
+            scores = np.asarray(prediction.confidence, dtype=np.float32)
+            names = np.asarray(prediction.data["class_name"])
+            if not (len(boxes) == len(scores) == len(names)):
+                raise RuntimeError("official RF-DETR prediction columns disagree")
+            for box, score, raw_name in zip(boxes, scores, names, strict=True):
+                name = str(raw_name)
+                category_id = category_ids_by_name.get(name)
+                if category_id is None:
+                    continue
+                row = coco_prediction_row(
+                    image_id=_integer(image, "id"),
+                    category_id=category_id,
+                    score=float(score),
+                    bbox_xyxy=tuple(float(value) for value in box),
+                    image_width=_positive_integer(image, "width"),
+                    image_height=_positive_integer(image, "height"),
+                )
+                if row is not None:
+                    rows.append(row)
+            timings.append((time.perf_counter_ns() - image_started) / 1_000_000)
+            write_progress(progress, "pytorch", index, len(images), len(rows), timings[-1])
+    finally:
+        del model
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+    return rows, timing_report(started_at, timings)
+
+
+def run_tensorrt_predictions(
+    *,
+    images: list[dict[str, object]],
+    images_root: Path,
+    triton_origin: str,
+    progress: Any,
+) -> tuple[list[dict[str, float | int | list[float]]], dict[str, object]]:
+    import torch
+    import torchvision.transforms.functional as vision_functional  # type: ignore[import-not-found]
+
+    means = [0.485, 0.456, 0.406]
+    stds = [0.229, 0.224, 0.225]
+    backend = TritonRfDetrHttpInferenceBackend(triton_origin)
+    rows: list[dict[str, float | int | list[float]]] = []
+    timings: list[float] = []
+    started_at = time.perf_counter_ns()
+    try:
+        for index, image in enumerate(images, start=1):
+            image_started = time.perf_counter_ns()
+            image_path = (images_root / _string(image, "file_name")).resolve(strict=True)
+            with Image.open(image_path) as opened:
+                rgb = opened.convert("RGB")
+                tensor = vision_functional.to_tensor(rgb)
+            tensor = vision_functional.resize(tensor, [704, 704], antialias=False)
+            tensor = vision_functional.normalize(tensor, means, stds)
+            batch = np.ascontiguousarray(tensor.unsqueeze(0).numpy(), dtype=np.float32)
+            output = backend.infer(batch)
+            rows.extend(
+                decode_tensorrt_coco_rows(
+                    image_id=_integer(image, "id"),
+                    image_width=_positive_integer(image, "width"),
+                    image_height=_positive_integer(image, "height"),
+                    boxes=output.boxes,
+                    logits=output.logits,
+                    maximum_detections=300,
+                )
+            )
+            timings.append((time.perf_counter_ns() - image_started) / 1_000_000)
+            write_progress(progress, "tensorrt", index, len(images), len(rows), timings[-1])
+    finally:
+        backend.close()
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+    return rows, timing_report(started_at, timings)
+
+
+def decode_tensorrt_coco_rows(
+    *,
+    image_id: int,
+    image_width: int,
+    image_height: int,
+    boxes: np.ndarray[Any, Any],
+    logits: np.ndarray[Any, Any],
+    maximum_detections: int,
+) -> list[dict[str, float | int | list[float]]]:
+    if boxes.shape != (1, 300, 4) or logits.shape != (1, 300, 91):
+        raise RuntimeError("RF-DETR TensorRT output shape changed")
+    probabilities = 1.0 / (1.0 + np.exp(-np.clip(logits[0].astype(np.float32), -80, 80)))
+    flattened = probabilities.reshape(-1)
+    topk = np.argsort(-flattened, kind="stable")[:maximum_detections]
+    valid_category_ids = set(COCO_SPARSE_IDS)
+    result: list[dict[str, float | int | list[float]]] = []
+    for flat_index in topk:
+        category_id = int(flat_index % logits.shape[2])
+        if category_id not in valid_category_ids:
+            continue
+        query_index = int(flat_index // logits.shape[2])
+        center_x, center_y, width, height = (
+            float(value) for value in boxes[0, query_index].astype(np.float32)
+        )
+        row = coco_prediction_row(
+            image_id=image_id,
+            category_id=category_id,
+            score=float(flattened[flat_index]),
+            bbox_xyxy=(
+                (center_x - width / 2) * image_width,
+                (center_y - height / 2) * image_height,
+                (center_x + width / 2) * image_width,
+                (center_y + height / 2) * image_height,
+            ),
+            image_width=image_width,
+            image_height=image_height,
+        )
+        if row is not None:
+            result.append(row)
+    return result
+
+
+def coco_prediction_row(
+    *,
+    image_id: int,
+    category_id: int,
+    score: float,
+    bbox_xyxy: tuple[float, ...],
+    image_width: int,
+    image_height: int,
+) -> dict[str, float | int | list[float]] | None:
+    if len(bbox_xyxy) != 4 or not all(math.isfinite(value) for value in bbox_xyxy):
+        raise RuntimeError("prediction box is invalid")
+    if not math.isfinite(score) or not 0 <= score <= 1:
+        raise RuntimeError("prediction score is invalid")
+    x1 = min(max(bbox_xyxy[0], 0.0), float(image_width))
+    y1 = min(max(bbox_xyxy[1], 0.0), float(image_height))
+    x2 = min(max(bbox_xyxy[2], 0.0), float(image_width))
+    y2 = min(max(bbox_xyxy[3], 0.0), float(image_height))
+    width = x2 - x1
+    height = y2 - y1
+    if width <= 0 or height <= 0:
+        return None
+    return {
+        "image_id": image_id,
+        "category_id": category_id,
+        "bbox": [x1, y1, width, height],
+        "score": score,
+    }
+
+
+def evaluate_coco(
+    *,
+    annotations: Path,
+    predictions: list[dict[str, float | int | list[float]]],
+    image_ids: list[int],
+) -> tuple[dict[str, float], str]:
+    from pycocotools.coco import COCO  # type: ignore[import-untyped]
+    from pycocotools.cocoeval import COCOeval  # type: ignore[import-untyped]
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        truth = COCO(str(annotations))
+        detections = truth.loadRes(predictions)
+        evaluator = COCOeval(truth, detections, "bbox")
+        evaluator.params.imgIds = image_ids
+        evaluator.evaluate()
+        evaluator.accumulate()
+        evaluator.summarize()
+    metrics = {
+        name: round(float(value), 9)
+        for name, value in zip(COCO_METRIC_NAMES, evaluator.stats, strict=True)
+    }
+    return metrics, output.getvalue()
+
+
+def build_parity_decision(
+    *,
+    profile: Mapping[str, object],
+    pytorch_metrics: Mapping[str, float],
+    tensorrt_metrics: Mapping[str, float],
+    full_admission_run: bool,
+) -> dict[str, object]:
+    reference = _object(profile.get("published_reference"), "published_reference")
+    gates = _object(profile.get("parity_gates"), "parity_gates")
+    reproduction_tolerance = _number(reference, "maximum_absolute_reproduction_delta")
+    pytorch_reference_deltas = {
+        "ap_50_95": abs(pytorch_metrics["ap_50_95"] - _number(reference, "coco_ap_50_95")),
+        "ap_50": abs(pytorch_metrics["ap_50"] - _number(reference, "coco_ap_50")),
+    }
+    provider_deltas = {
+        "ap_50_95": abs(tensorrt_metrics["ap_50_95"] - pytorch_metrics["ap_50_95"]),
+        "ap_50": abs(tensorrt_metrics["ap_50"] - pytorch_metrics["ap_50"]),
+    }
+    reproduction_passed = all(
+        value <= reproduction_tolerance for value in pytorch_reference_deltas.values()
+    )
+    provider_parity_passed = provider_deltas["ap_50_95"] <= _number(
+        gates, "maximum_absolute_ap_50_95_delta"
+    ) and provider_deltas["ap_50"] <= _number(gates, "maximum_absolute_ap_50_delta")
+    if not full_admission_run:
+        diagnosis = "bounded-smoke-only"
+    elif not reproduction_passed:
+        diagnosis = "upstream-reproduction-failed"
+    elif not provider_parity_passed:
+        diagnosis = "tensorrt-deployment-parity-failed"
+    else:
+        diagnosis = "upstream-and-tensorrt-parity-passed"
+    passed = full_admission_run and reproduction_passed and provider_parity_passed
+    return {
+        "passed": passed,
+        "diagnosis": diagnosis,
+        "full_admission_run": full_admission_run,
+        "pytorch_reference_deltas": pytorch_reference_deltas,
+        "provider_deltas": provider_deltas,
+        "upstream_reproduction_passed": reproduction_passed,
+        "provider_parity_passed": provider_parity_passed,
+        "semantic_authority_changed": False,
+    }
+
+
+def write_progress(
+    progress: Any,
+    provider: str,
+    completed: int,
+    total: int,
+    prediction_count: int,
+    last_image_ms: float,
+) -> None:
+    if completed != 1 and completed % 100 != 0 and completed != total:
+        return
+    progress.write(
+        canonical_json(
+            {
+                "schema_version": PROGRESS_SCHEMA,
+                "provider": provider,
+                "completed_images": completed,
+                "total_images": total,
+                "prediction_count": prediction_count,
+                "last_image_ms": last_image_ms,
+            }
+        )
+        + "\n"
+    )
+    progress.flush()
+
+
+def timing_report(started_at: int, timings: list[float]) -> dict[str, object]:
+    if not timings:
+        raise RuntimeError("provider timing is empty")
+    duration_seconds = (time.perf_counter_ns() - started_at) / 1_000_000_000
+    ordered = sorted(timings)
+    return {
+        "duration_seconds": duration_seconds,
+        "effective_images_per_second": len(timings) / duration_seconds,
+        "image_ms": {
+            "mean": statistics.fmean(ordered),
+            "p50": percentile(ordered, 0.5),
+            "p95": percentile(ordered, 0.95),
+            "p99": percentile(ordered, 0.99),
+            "maximum": ordered[-1],
+        },
+    }
+
+
+def percentile(values: list[float], quantile: float) -> float:
+    position = (len(values) - 1) * quantile
+    lower = math.floor(position)
+    upper = math.ceil(position)
+    if lower == upper:
+        return values[lower]
+    return values[lower] + (values[upper] - values[lower]) * (position - lower)
+
+
+def write_gzip_jsonl(path: Path, rows: Iterable[Mapping[str, float | int | list[float]]]) -> None:
+    with (
+        path.open("xb") as raw_handle,
+        gzip.GzipFile(fileobj=raw_handle, mode="wb", compresslevel=6, mtime=0) as compressed,
+        io.TextIOWrapper(compressed, encoding="utf-8") as handle,
+    ):
+        for row in rows:
+            handle.write(canonical_json(row) + "\n")
+
+
+def package_versions(names: Iterable[str]) -> dict[str, str]:
+    result: dict[str, str] = {}
+    for name in names:
+        try:
+            result[name] = importlib.metadata.version(name)
+        except importlib.metadata.PackageNotFoundError:
+            result[name] = "not-installed"
+    return result
+
+
+def sha256_path(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()
+
+
+def canonical_json(value: object) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+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 RuntimeError(f"{label} is not an object")
+    return value
+
+
+def _string(value: Mapping[str, object], key: str) -> str:
+    result = value.get(key)
+    if not isinstance(result, str) or not result:
+        raise RuntimeError(f"{key} is not a non-empty string")
+    return result
+
+
+def _integer(value: Mapping[str, object], key: str) -> int:
+    result = value.get(key)
+    if not isinstance(result, int) or isinstance(result, bool):
+        raise RuntimeError(f"{key} is not an integer")
+    return result
+
+
+def _positive_integer(value: Mapping[str, object], key: str) -> int:
+    result = _integer(value, key)
+    if result <= 0:
+        raise RuntimeError(f"{key} is not positive")
+    return result
+
+
+def _number(value: Mapping[str, object], key: str) -> float:
+    result = value.get(key)
+    if not isinstance(result, (int, float)) or isinstance(result, bool):
+        raise RuntimeError(f"{key} is not numeric")
+    converted = float(result)
+    if not math.isfinite(converted):
+        raise RuntimeError(f"{key} is not finite")
+    return converted
+
+
+if __name__ == "__main__":
+    main()

experiments/perception/worker/Invoke-M48TUpstreamParity.ps1

@@ -0,0 +1,312 @@
+[CmdletBinding()]
+param(
+    [Parameter(Mandatory = $true)]
+    [string]$ReleaseRoot,
+    [Parameter(Mandatory = $true)]
+    [ValidatePattern("^[a-f0-9]{64}$")]
+    [string]$ExpectedWheelSha256,
+    [Parameter(Mandatory = $true)]
+    [ValidatePattern("^[A-Za-z0-9._-]{1,96}$")]
+    [string]$RunId,
+    [ValidateRange(0, 5000)]
+    [int]$MaximumImages = 0,
+    [string]$DatasetRoot = "D:\NDC_MISSIONCORE\datasets\coco-2017-val",
+    [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m48t-upstream-parity"
+)
+
+$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 "M48T upstream parity is pinned to DESKTOP-OPJ8J04"
+}
+
+$release = Resolve-DDirectory $ReleaseRoot "M48T parity release root" $false
+$dataset = Resolve-DDirectory $DatasetRoot "M48T parity dataset root" $false
+$output = Resolve-DDirectory $OutputRoot "M48T parity output root" $true
+$runOutput = Join-Path $output $RunId
+if (Test-Path -LiteralPath $runOutput) { throw "M48T parity output already exists" }
+$null = New-Item -ItemType Directory -Path $runOutput
+$runOutput = Resolve-DDirectory $runOutput "M48T parity run output" $false
+
+$wheel = Assert-RegularFile (Join-Path $release "nodedc_mission_core-0.1.0-py3-none-any.whl") "wheel"
+if ((Get-Sha256 $wheel) -cne $ExpectedWheelSha256) { throw "wheel SHA-256 changed" }
+$profile = Assert-RegularFile (Join-Path $release "m48t-upstream-parity-v1.json") "profile"
+$runner = Assert-RegularFile (Join-Path $release "run_m48t_upstream_parity_worker.py") "runner"
+$annotations = Assert-RegularFile (Join-Path $dataset "instances_val2017.json") "COCO annotations"
+$imagesRoot = Resolve-DDirectory (Join-Path $dataset "val2017") "COCO val2017 images" $false
+if (@(Get-ChildItem -LiteralPath $imagesRoot -File -Filter "*.jpg").Count -ne 5000) {
+    throw "COCO val2017 image count changed"
+}
+
+$experimentRoot = Resolve-DDirectory (
+    "D:\NDC_MISSIONCORE\runtime\experiments\m48t-fixed-detector-20260825T095425Z"
+) "RF-DETR experiment root" $false
+$checkpoint = Assert-RegularFile (Join-Path $experimentRoot "weights\rf-detr-large-2026.pth") "checkpoint"
+if ((Get-Sha256 $checkpoint) -cne "0f4e20e19a99c0f8a62b5685f57f6c8b5c371c59081feda6752a0561a79ccf38") {
+    throw "RF-DETR checkpoint SHA-256 changed"
+}
+$modelRoot = Resolve-DDirectory (Join-Path $experimentRoot "triton-models") "model root" $false
+$engine = Assert-RegularFile (Join-Path $modelRoot "rf_detr_large\1\model.plan") "TensorRT engine"
+if ((Get-Sha256 $engine) -cne "986399ce706b7380472cf5e473232249fed6e628971d8007f6609e83128d46b8") {
+    throw "RF-DETR TensorRT engine SHA-256 changed"
+}
+
+$historical = Get-Container "ndc-mission-core-triton"
+if (-not $historical.State.Running -or $historical.State.Health.Status -cne "healthy") {
+    throw "Historical Triton must remain healthy"
+}
+$historicalId = [string]$historical.Id
+$baseImage = "nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794"
+& docker image inspect $baseImage *> $null
+Assert-LastExitCode "pinned base image inspection"
+$baseImageId = (& docker image inspect $baseImage --format "{{.Id}}").Trim()
+Assert-LastExitCode "pinned base image identity"
+$runtimeVolume = "ndc-mission-core-m48t-upstream-parity-env"
+if (-not (& docker volume ls --quiet --filter "name=^$runtimeVolume$")) {
+    & docker volume create `
+        --label "com.nodedc.product=mission-core" `
+        --label "com.nodedc.stack=ndc-mission-core-compute" `
+        --label "com.nodedc.role=bounded-rf-detr-upstream-parity" `
+        --label "com.nodedc.managed-by=codex-bounded-experiment" `
+        $runtimeVolume *> $null
+    Assert-LastExitCode "M48T upstream parity dependency volume creation"
+}
+$torchReady = $false
+$strictErrorActionPreference = $ErrorActionPreference
+$ErrorActionPreference = "Continue"
+& docker run --rm `
+    --read-only `
+    --security-opt "no-new-privileges:true" `
+    --cap-drop ALL `
+    --tmpfs "/tmp:rw,noexec,nosuid,size=512m" `
+    -e "PYTHONPATH=/opt/parity" `
+    -v ($runtimeVolume + ":/opt/parity:ro") `
+    --entrypoint python3 `
+    $baseImage `
+    -c "import importlib.metadata as m; assert m.version('numpy') == '1.26.4'; assert m.version('torch') == '2.9.1+cu130'; assert m.version('torchvision') == '0.24.1+cu130'" *> $null
+if ($LASTEXITCODE -eq 0) { $torchReady = $true }
+$ErrorActionPreference = $strictErrorActionPreference
+if (-not $torchReady) {
+    & docker run --rm `
+        --name "ndc-mission-core-m48t-upstream-parity-env-torch" `
+        --read-only `
+        --security-opt "no-new-privileges:true" `
+        --cap-drop ALL `
+        --pids-limit 512 `
+        --tmpfs "/tmp:rw,noexec,nosuid,size=8g" `
+        -e "PIP_DISABLE_PIP_VERSION_CHECK=1" `
+        -v ($runtimeVolume + ":/opt/parity:rw") `
+        --entrypoint python3 `
+        $baseImage `
+        -m pip install --no-cache-dir --target /opt/parity `
+        --index-url https://download.pytorch.org/whl/cu130 `
+        "numpy==1.26.4" "torch==2.9.1+cu130" "torchvision==0.24.1+cu130"
+    Assert-LastExitCode "M48T upstream parity PyTorch environment initialization"
+}
+$rfdetrReady = $false
+$ErrorActionPreference = "Continue"
+& docker run --rm `
+    --read-only `
+    --security-opt "no-new-privileges:true" `
+    --cap-drop ALL `
+    --tmpfs "/tmp:rw,noexec,nosuid,size=512m" `
+    -e "PYTHONPATH=/opt/parity" `
+    -v ($runtimeVolume + ":/opt/parity:ro") `
+    --entrypoint python3 `
+    $baseImage `
+    -c "import importlib.metadata as m; import rfdetr; assert m.version('rfdetr') == '1.9.4'; assert m.version('pycocotools') == '2.0.10'; assert m.version('tritonclient') == '2.71.0'" *> $null
+if ($LASTEXITCODE -eq 0) { $rfdetrReady = $true }
+$ErrorActionPreference = $strictErrorActionPreference
+if (-not $rfdetrReady) {
+    & docker run --rm `
+        --name "ndc-mission-core-m48t-upstream-parity-env-dependencies" `
+        --read-only `
+        --security-opt "no-new-privileges:true" `
+        --cap-drop ALL `
+        --pids-limit 512 `
+        --tmpfs "/tmp:rw,noexec,nosuid,size=8g" `
+        -e "PIP_DISABLE_PIP_VERSION_CHECK=1" `
+        -e "PYTHONPATH=/opt/parity" `
+        -v ($runtimeVolume + ":/opt/parity:rw") `
+        --entrypoint python3 `
+        $baseImage `
+        -m pip install --no-cache-dir --upgrade --target /opt/parity `
+        "numpy==1.26.4" "requests" "tqdm" "transformers>=5.1.0,<6.0.0" `
+        "pydantic>=2.0,<3.0" "supervision>=0.29.0,<1.0" "pyDeprecate>=0.9,<0.10" `
+        "pycocotools==2.0.10" "tritonclient[http]==2.71.0"
+    Assert-LastExitCode "M48T upstream parity dependency initialization"
+    & docker run --rm `
+        --name "ndc-mission-core-m48t-upstream-parity-env-rfdetr" `
+        --read-only `
+        --security-opt "no-new-privileges:true" `
+        --cap-drop ALL `
+        --pids-limit 512 `
+        --tmpfs "/tmp:rw,noexec,nosuid,size=2g" `
+        -e "PIP_DISABLE_PIP_VERSION_CHECK=1" `
+        -e "PYTHONPATH=/opt/parity" `
+        -v ($runtimeVolume + ":/opt/parity:rw") `
+        --entrypoint python3 `
+        $baseImage `
+        -m pip install --no-cache-dir --no-deps --upgrade --target /opt/parity "rfdetr==1.9.4"
+    Assert-LastExitCode "M48T upstream parity RF-DETR package initialization"
+}
+& docker run --rm `
+    --read-only `
+    --security-opt "no-new-privileges:true" `
+    --cap-drop ALL `
+    --tmpfs "/tmp:rw,noexec,nosuid,size=512m" `
+    -e "PYTHONPATH=/opt/parity" `
+    -v ($runtimeVolume + ":/opt/parity:ro") `
+    --entrypoint python3 `
+    $baseImage `
+    -c "import importlib.metadata as m; import rfdetr, torch, torchvision; assert m.version('numpy') == '1.26.4'; assert m.version('rfdetr') == '1.9.4'; assert m.version('torch') == '2.9.1+cu130'; assert m.version('torchvision') == '0.24.1+cu130'; assert m.version('pycocotools') == '2.0.10'; assert m.version('tritonclient') == '2.71.0'"
+Assert-LastExitCode "M48T upstream parity dependency volume verification"
+$runtimeIdentity = $baseImageId + "+volume:" + $runtimeVolume
+
+$tritonName = "ndc-mission-core-m48t-upstream-parity-triton"
+$runnerName = "ndc-mission-core-m48t-upstream-parity"
+foreach ($name in @($tritonName, $runnerName)) {
+    if (& docker ps -a --format "{{.Names}}" --filter "name=^/$name$") {
+        throw "M48T parity container $name already exists"
+    }
+}
+
+try {
+    & docker create `
+        --name $tritonName `
+        --read-only `
+        --security-opt "no-new-privileges:true" `
+        --cap-drop ALL `
+        --pids-limit 512 `
+        --shm-size 1g `
+        --gpus all `
+        --tmpfs "/tmp:rw,noexec,nosuid,size=2g" `
+        --health-cmd "curl --fail --silent http://127.0.0.1:8000/v2/health/ready" `
+        --health-interval 5s `
+        --health-timeout 3s `
+        --health-start-period 20s `
+        --health-retries 24 `
+        -v ((Convert-ToDockerPath $modelRoot) + ":/models:ro") `
+        $baseImage `
+        tritonserver `
+        --model-repository=/models `
+        --model-control-mode=explicit `
+        --load-model=rf_detr_large `
+        --disable-auto-complete-config `
+        --strict-readiness=true `
+        --exit-on-error=true `
+        --allow-http=true `
+        --allow-grpc=false `
+        --allow-metrics=false *> $null
+    Assert-LastExitCode "M48T parity Triton creation"
+    & docker start $tritonName *> $null
+    Assert-LastExitCode "M48T parity Triton start"
+    $ready = $false
+    foreach ($attempt in 1..60) {
+        Start-Sleep -Seconds 2
+        $candidate = Get-Container $tritonName
+        if (-not $candidate.State.Running) { throw "M48T parity Triton stopped during startup" }
+        if ($candidate.State.Health.Status -ceq "healthy") { $ready = $true; break }
+    }
+    if (-not $ready) { throw "M48T parity Triton did not become healthy" }
+
+    $maximumArguments = @()
+    if ($MaximumImages -gt 0) { $maximumArguments = @("--maximum-images", ([string]$MaximumImages)) }
+    $arguments = @(
+        "run", "--name", $runnerName,
+        "--network", ("container:{0}" -f $tritonName),
+        "--read-only",
+        "--security-opt", "no-new-privileges:true",
+        "--cap-drop", "ALL",
+        "--pids-limit", "512",
+        "--gpus", "all",
+        "--shm-size", "2g",
+        "--tmpfs", "/tmp:rw,noexec,nosuid,size=8g",
+        "-e", "PYTHONDONTWRITEBYTECODE=1",
+        "-e", "PYTHONPATH=/opt/parity:/release/nodedc_mission_core-0.1.0-py3-none-any.whl",
+        "-v", ($runtimeVolume + ":/opt/parity:ro"),
+        "-v", ((Convert-ToDockerPath $release) + ":/release:ro"),
+        "-v", ((Convert-ToDockerPath $imagesRoot) + ":/dataset/val2017:ro"),
+        "-v", ((Convert-ToDockerPath $annotations) + ":/dataset/instances_val2017.json:ro"),
+        "-v", ((Convert-ToDockerPath $checkpoint) + ":/model/rf-detr-large-2026.pth:ro"),
+        "-v", ((Convert-ToDockerPath $runOutput) + ":/output:rw"),
+        "--entrypoint", "python3",
+        $baseImage,
+        "/release/run_m48t_upstream_parity_worker.py",
+        "--profile", "/release/m48t-upstream-parity-v1.json",
+        "--annotations", "/dataset/instances_val2017.json",
+        "--images-root", "/dataset/val2017",
+        "--checkpoint", "/model/rf-detr-large-2026.pth",
+        "--triton-origin", "http://127.0.0.1:8000",
+        "--output", "/output/result.json",
+        "--pytorch-predictions", "/output/pytorch-predictions.jsonl.gz",
+        "--tensorrt-predictions", "/output/tensorrt-predictions.jsonl.gz",
+        "--progress", "/output/progress.jsonl",
+        "--runtime-image", $runtimeIdentity
+    ) + $maximumArguments
+    & docker @arguments
+    Assert-LastExitCode "M48T upstream parity evaluation"
+    if (-not (Test-Path -LiteralPath (Join-Path $runOutput "result.json") -PathType Leaf)) {
+        throw "M48T upstream parity result was not written"
+    }
+} finally {
+    foreach ($name in @($runnerName, $tritonName)) {
+        if (& docker ps -a --format "{{.Names}}" --filter "name=^/$name$") {
+            & docker rm -f $name *> $null
+        }
+    }
+    $historicalAfter = Get-Container "ndc-mission-core-triton"
+    if (
+        $historicalAfter.Id -cne $historicalId -or
+        -not $historicalAfter.State.Running -or
+        $historicalAfter.State.Health.Status -cne "healthy"
+    ) {
+        throw "Historical Triton changed during M48T upstream parity"
+    }
+}
+
+Write-Output ("M48T_UPSTREAM_PARITY_RESULT={0}" -f (Join-Path $runOutput "result.json"))
+Write-Output ("RUNTIME_IDENTITY={0}" -f $runtimeIdentity)
+Write-Output "HISTORICAL_TRITON_ACTION=none"
+Write-Output "SEMANTIC_AUTHORITY_CHANGED=false"

tests/test_m48t_upstream_parity.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+from types import ModuleType
+
+import numpy as np
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+RUNNER_PATH = REPOSITORY_ROOT / "experiments/perception/run_m48t_upstream_parity_worker.py"
+PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48t-upstream-parity-v1.json"
+
+
+def load_runner() -> ModuleType:
+    specification = importlib.util.spec_from_file_location("m48t_upstream_parity", RUNNER_PATH)
+    assert specification is not None and specification.loader is not None
+    module = importlib.util.module_from_spec(specification)
+    specification.loader.exec_module(module)
+    return module
+
+
+def test_upstream_parity_profile_freezes_official_and_deployed_providers() -> None:
+    profile = json.loads(PROFILE_PATH.read_text("utf-8"))
+
+    assert profile["model"]["package_version"] == "1.9.4"
+    assert profile["model"]["resolution"] == [704, 704]
+    assert profile["dataset"]["image_count"] == 5000
+    assert profile["providers"]["pytorch"]["confidence_prefilter"] == 0.0
+    assert profile["providers"]["tensorrt"]["confidence_prefilter"] == 0.0
+    assert profile["authority"]["navigation_or_safety_accepted"] is False
+
+
+def test_tensorrt_decoder_uses_sparse_coco_ids_and_original_image_geometry() -> None:
+    runner = load_runner()
+    boxes = np.zeros((1, 300, 4), dtype=np.float16)
+    logits = np.full((1, 300, 91), -20, dtype=np.float16)
+    boxes[0, 4] = np.asarray((0.5, 0.5, 0.2, 0.4), dtype=np.float16)
+    logits[0, 4, 1] = np.float16(4.0)  # COCO sparse id 1: person
+    logits[0, 7, 12] = np.float16(5.0)  # sparse gap: must never escape
+
+    rows = runner.decode_tensorrt_coco_rows(
+        image_id=42,
+        image_width=1000,
+        image_height=500,
+        boxes=boxes,
+        logits=logits,
+        maximum_detections=2,
+    )
+
+    assert len(rows) == 1
+    assert rows[0]["image_id"] == 42
+    assert rows[0]["category_id"] == 1
+    assert np.allclose(rows[0]["bbox"], [400.0244, 149.9756, 199.9512, 200.0488], atol=0.1)
+
+
+def test_parity_decision_localizes_upstream_and_deployment_failures() -> None:
+    runner = load_runner()
+    profile = json.loads(PROFILE_PATH.read_text("utf-8"))
+
+    passed = runner.build_parity_decision(
+        profile=profile,
+        pytorch_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
+        tensorrt_metrics={"ap_50_95": 0.563, "ap_50": 0.749},
+        full_admission_run=True,
+    )
+    assert passed["passed"] is True
+    assert passed["diagnosis"] == "upstream-and-tensorrt-parity-passed"
+
+    deployment_failure = runner.build_parity_decision(
+        profile=profile,
+        pytorch_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
+        tensorrt_metrics={"ap_50_95": 0.54, "ap_50": 0.72},
+        full_admission_run=True,
+    )
+    assert deployment_failure["passed"] is False
+    assert deployment_failure["diagnosis"] == "tensorrt-deployment-parity-failed"
+
+    smoke = runner.build_parity_decision(
+        profile=profile,
+        pytorch_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
+        tensorrt_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
+        full_admission_run=False,
+    )
+    assert smoke["passed"] is False
+    assert smoke["diagnosis"] == "bounded-smoke-only"