From f41428f907e16c5e2c98445ad87a62bf88e53a4c Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Mon, 31 Aug 2026 18:59:47 +0300 Subject: [PATCH] feat(lab): add sealed local component runners --- .../portable_lab_v1_local_runners.py | 1331 +++++++++++++++++ ...servatory_portable_lab_v1_local_runners.py | 531 +++++++ 2 files changed, 1862 insertions(+) create mode 100644 src/k1link/observatory/portable_lab_v1_local_runners.py create mode 100644 tests/test_observatory_portable_lab_v1_local_runners.py diff --git a/src/k1link/observatory/portable_lab_v1_local_runners.py b/src/k1link/observatory/portable_lab_v1_local_runners.py new file mode 100644 index 0000000..187711d --- /dev/null +++ b/src/k1link/observatory/portable_lab_v1_local_runners.py @@ -0,0 +1,1331 @@ +"""Worker-local LAB V1 component runners backed by the Docker Engine API. + +The portable LAB V1 bridge deliberately exposes EoMT and DDRNet as Python +ports. This module provides the concrete Worker-side implementation of those +ports without accepting an image, command, filesystem path, or runtime option +from a queued job. All executable details are sealed at installation time. + +The persistent Worker talks only to its local Docker Engine Unix socket. Each +component runs in a short-lived sibling container with no network, a read-only +root filesystem, all Linux capabilities dropped, and one explicit GPU +request. Dynamic source and result paths are mapped through one pre-installed +controller-to-engine work-root binding; model and runtime assets use exact +Worker-host locators and immutable identities. + +Nothing here opens a K1 connection or grants command, navigation, safety, or +actuation authority. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import secrets +import shutil +import stat +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from pathlib import Path, PurePosixPath, PureWindowsPath +from typing import Final, Literal, Protocol, cast + +import httpx + +from k1link.compute.jobs import validate_camera_compute_job +from k1link.observatory.portable_lab_v1_executor import ( + DDRNET_RESULT_SCHEMA, + EOMT_RESULT_SCHEMA, + PortableLabV1MaterializedSource, + PortableLabV1OrchestrationPlan, +) +from k1link.observatory.portable_lab_v1_worker import PortableLabV1WorkerError +from k1link.observatory.portable_result_contract import ( + OBSERVATION_ONLY_AUTHORITY, + canonical_json, +) +from k1link.observatory.portable_run_definitions import canonical_sha256 + +PORTABLE_LAB_V1_COMPONENT_REQUEST_SCHEMA: Final = ( + "missioncore.observatory-portable-lab-v1-component-request/v1" +) +PORTABLE_LAB_V1_RUNNER_INSTALLATION_SCHEMA: Final = ( + "missioncore.observatory-portable-lab-v1-runner-installation/v1" +) +PORTABLE_LAB_V1_COMPONENT_INSTALLATION_SCHEMA: Final = ( + "missioncore.observatory-portable-lab-v1-component-installation/v1" +) + +_SETUP_ID: Final = "lab-v1-eomt-ddrnet-portable-v1" +_DEFINITION_ID: Final = "lab-v1-eomt-ddrnet-portable" +_DEFINITION_VERSION: Final = 2 +_EOMT_IMAGE_SHA256: Final = "58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" +_DDRNET_IMAGE_SHA256: Final = "591cb382c099eeb05e7ec16e2371e0b2da54d2bb5c49ec0f4ac88dbf72b0f0cd" +_EOMT_PIPELINE: Final = "recorded-semantic-eomt-fisheye-mask/v1" +_DDRNET_CHECKPOINT_SHA256: Final = ( + "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6" +) +_GOOSE_MAPPING_SHA256: Final = "88ae319ba5a3877dd3ae0773f693a6a5fdc283934140de9dfaff029108aefd7f" +_MAX_RESULT_DOCUMENT_BYTES: Final = 8 * 1024 * 1024 +_MAX_ENGINE_ERROR_BYTES: Final = 64 * 1024 +_DOCKER_API_VERSION: Final = "v1.47" +_DOCKER_SOCKET: Final = Path("/var/run/docker.sock") +_SAFE_TOKEN = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,127}$") +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_CONTAINER_ID = re.compile(r"^[a-f0-9]{64}$") +_WINDOWS_D_PATH = re.compile(r"^[dD]:\\[^\x00-\x1f<>\"|?*]*$") +_WINDOWS_RESERVED_NAME = re.compile(r"(?i)^(?:con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?$") + +type PortableLabV1Component = Literal["eomt", "ddrnet"] +type PortableLabV1AssetKind = Literal["file", "tree"] +type PortableLabV1AssetVerification = Literal["sha256", "identity-sha256"] + +_EXPECTED_ASSET_IDENTITIES: Final[dict[PortableLabV1Component, dict[str, str]]] = { + "eomt": { + "eomt-config-json": ("7f4aa94fa4e43c0dbd79a5420edb511120aef62bd82bfbcbcece79948286a650"), + "eomt-dependency-set": ("4eb1f8d33236806e74f9e5bb96b7dce2ac37623dc39b2184be2aa8d7d00e983e"), + "eomt-evaluation-helper": ( + "25baf30c0df564734e08f38ace88cc4bc147cacf240c761622279511e361daa4" + ), + "eomt-model-weights": ("c265da9a74f58f5c3f4826d23ca4ca78beac0b106cca5842beca61580de5b782"), + "eomt-preprocessor-config": ( + "97e2fbf7f0bdba2cfc90251c5133bae9c27ddc9c4410509f40670be2332854e7" + ), + "eomt-profile": ("ea583966bc3409f5cf563cbf4fad05e366907e67187082eb692aff53d9f5d875"), + "eomt-profile-runtime": ( + "01881862d4eaa218955f776a948124bf19c34be2b5ec282115daeacb15c53ae6" + ), + "eomt-recorded-runtime": ( + "4dcc4fc8bdf33702651a199be69d0dd4fadb243d2e65aee1c3d1ae7a58fdf675" + ), + "eomt-runner": ("651e8e06c3912dffb036b7fd08f2c0623f7563d8306cc7aee05db562798518f4"), + "k1-valid-fov-identity": ( + "b4dd8ddf2b87c1d520ee8a0868c4fea062d7c14d1bae73ccabd3abe1f3acbac2" + ), + "k1-valid-fov-mask": ("a40cee06b7c6f69b6a09a11563dcfd237f3de833b1ccd31459e66692e528ba63"), + }, + "ddrnet": { + "ddrnet-checkpoint": _DDRNET_CHECKPOINT_SHA256, + "ddrnet-goose-mapping": _GOOSE_MAPPING_SHA256, + "ddrnet-goose-runner": ("b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1"), + "vegetation-policy": ("b75c4ac841d7b4bcc57f7a9c8417ca2317d8ecfa499e72a9af8a8591a2ec0d35"), + "vegetation-provider-map": ( + "f2b69046b6a740fd9532d2d88e7fabae7c20fb662f783c9502adc9026406f352" + ), + }, +} +_EXPECTED_ASSET_LENGTHS: Final[dict[str, int]] = { + "ddrnet-checkpoint": 259_419_077, + "ddrnet-goose-runner": 32_877, + "eomt-config-json": 1_575, + "eomt-evaluation-helper": 28_915, + "eomt-model-weights": 1_276_175_488, + "eomt-preprocessor-config": 666, + "eomt-profile": 3_805, + "eomt-profile-runtime": 45_789, + "eomt-recorded-runtime": 34_899, + "eomt-runner": 30_720, + "vegetation-policy": 3_022, + "vegetation-provider-map": 2_756, +} + + +class PortableLabV1LocalRunnerError(PortableLabV1WorkerError): + """An installed local component or its Docker execution changed.""" + + +@dataclass(frozen=True, slots=True) +class PortableLabV1HostAsset: + """One exact asset exposed read-only from the Worker host.""" + + asset_id: str + host_path: str + container_path: str + kind: PortableLabV1AssetKind + verification: PortableLabV1AssetVerification + identity_sha256: str + byte_length: int | None = None + + def __post_init__(self) -> None: + if self.kind not in ("file", "tree"): + raise PortableLabV1LocalRunnerError("asset kind is invalid") + if self.verification not in ("sha256", "identity-sha256"): + raise PortableLabV1LocalRunnerError("asset verification is invalid") + _safe_token(self.asset_id, "asset id") + _host_path(self.host_path, "asset host path") + target = _container_path(self.container_path, "asset container path") + if target.parts[:4] != ("/", "opt", "nodedc", "assets"): + raise PortableLabV1LocalRunnerError("asset target is outside /opt/nodedc/assets") + _digest(self.identity_sha256, "asset identity sha256") + if self.kind == "file" and self.verification != "sha256": + raise PortableLabV1LocalRunnerError( + "file assets require byte-level sha256 verification" + ) + if self.kind == "tree" and self.verification != "identity-sha256": + raise PortableLabV1LocalRunnerError( + "tree assets require an identity-sha256 manifest fence" + ) + if self.byte_length is not None and ( + isinstance(self.byte_length, bool) or self.byte_length < 0 + ): + raise PortableLabV1LocalRunnerError("asset byte length is invalid") + + def identity_document(self) -> dict[str, object]: + return { + "asset_id": self.asset_id, + "host_path": self.host_path, + "container_path": self.container_path, + "kind": self.kind, + "verification": self.verification, + "identity_sha256": self.identity_sha256, + "byte_length": self.byte_length, + } + + def request_document(self) -> dict[str, object]: + return { + "asset_id": self.asset_id, + "path": self.container_path, + "kind": self.kind, + "verification": self.verification, + "identity_sha256": self.identity_sha256, + "byte_length": self.byte_length, + } + + +@dataclass(frozen=True, slots=True) +class PortableLabV1ComponentInstallation: + """Immutable executable identity for one installed model component.""" + + component: PortableLabV1Component + image_sha256: str + entrypoint: tuple[str, ...] + command: tuple[str, ...] + assets: tuple[PortableLabV1HostAsset, ...] + timeout_seconds: float + memory_bytes: int + nano_cpus: int + installation_sha256: str + + def __post_init__(self) -> None: + if self.component not in ("eomt", "ddrnet"): + raise PortableLabV1LocalRunnerError("component identity is invalid") + _digest(self.image_sha256, "component image sha256") + expected_image = _EOMT_IMAGE_SHA256 if self.component == "eomt" else _DDRNET_IMAGE_SHA256 + if self.image_sha256 != expected_image: + raise PortableLabV1LocalRunnerError(f"{self.component} image identity changed") + _argv(self.entrypoint, "component entrypoint") + _argv(self.command, "component command") + if not (1.0 <= self.timeout_seconds <= 24 * 60 * 60): + raise PortableLabV1LocalRunnerError("component timeout is invalid") + if not (1024**3 <= self.memory_bytes <= 128 * 1024**3): + raise PortableLabV1LocalRunnerError("component memory limit is invalid") + if not (100_000_000 <= self.nano_cpus <= 64_000_000_000): + raise PortableLabV1LocalRunnerError("component CPU limit is invalid") + asset_ids = tuple(asset.asset_id for asset in self.assets) + if asset_ids != tuple(sorted(asset_ids)) or len(asset_ids) != len(set(asset_ids)): + raise PortableLabV1LocalRunnerError("component assets are not canonical") + targets = tuple(asset.container_path for asset in self.assets) + if len(targets) != len(set(targets)): + raise PortableLabV1LocalRunnerError("component asset targets overlap") + expected = _EXPECTED_ASSET_IDENTITIES[self.component] + if set(asset_ids) != set(expected): + raise PortableLabV1LocalRunnerError(f"{self.component} installed asset set changed") + if any(asset.identity_sha256 != expected[asset.asset_id] for asset in self.assets): + raise PortableLabV1LocalRunnerError( + f"{self.component} installed asset identity changed" + ) + if any( + asset.asset_id in _EXPECTED_ASSET_LENGTHS + and asset.byte_length != _EXPECTED_ASSET_LENGTHS[asset.asset_id] + for asset in self.assets + ): + raise PortableLabV1LocalRunnerError( + f"{self.component} installed asset byte length changed" + ) + _digest(self.installation_sha256, "component installation sha256") + if canonical_sha256(self.identity_document()) != self.installation_sha256: + raise PortableLabV1LocalRunnerError(f"{self.component} installation receipt changed") + + @classmethod + def seal( + cls, + *, + component: PortableLabV1Component, + image_sha256: str, + entrypoint: tuple[str, ...], + command: tuple[str, ...], + assets: tuple[PortableLabV1HostAsset, ...], + timeout_seconds: float, + memory_bytes: int, + nano_cpus: int, + ) -> PortableLabV1ComponentInstallation: + identity = _component_installation_identity( + component=component, + image_sha256=image_sha256, + entrypoint=entrypoint, + command=command, + assets=assets, + timeout_seconds=timeout_seconds, + memory_bytes=memory_bytes, + nano_cpus=nano_cpus, + ) + return cls( + component=component, + image_sha256=image_sha256, + entrypoint=entrypoint, + command=command, + assets=assets, + timeout_seconds=timeout_seconds, + memory_bytes=memory_bytes, + nano_cpus=nano_cpus, + installation_sha256=canonical_sha256(identity), + ) + + def identity_document(self) -> dict[str, object]: + return _component_installation_identity( + component=self.component, + image_sha256=self.image_sha256, + entrypoint=self.entrypoint, + command=self.command, + assets=self.assets, + timeout_seconds=self.timeout_seconds, + memory_bytes=self.memory_bytes, + nano_cpus=self.nano_cpus, + ) + + +@dataclass(frozen=True, slots=True) +class PortableLabV1WorkRootBinding: + """Exact mapping from the persistent container to the Docker host D: root.""" + + controller_root: Path + engine_host_root: str + + def __post_init__(self) -> None: + root = _real_directory(self.controller_root, "controller work root") + _host_path(self.engine_host_root, "engine host work root") + object.__setattr__(self, "controller_root", root) + object.__setattr__(self, "engine_host_root", self.engine_host_root.rstrip("\\")) + + def engine_path(self, path: Path, *, label: str) -> str: + resolved = _real_path(path, label) + try: + relative = resolved.relative_to(self.controller_root) + except ValueError as exc: + raise PortableLabV1LocalRunnerError( + f"{label} is outside the installed work root" + ) from exc + if not relative.parts: + return self.engine_host_root + for part in relative.parts: + _windows_relative_part(part, label) + root = PureWindowsPath(self.engine_host_root) + candidate = root.joinpath(*relative.parts) + try: + candidate.relative_to(root) + except ValueError as exc: + raise PortableLabV1LocalRunnerError( + f"{label} escapes the installed engine host root" + ) from exc + return str(candidate) + + def identity_document(self) -> dict[str, object]: + return { + "controller_root": str(self.controller_root), + "engine_host_root": self.engine_host_root, + } + + +@dataclass(frozen=True, slots=True) +class PortableLabV1RunnerInstallation: + """Install-time binding shared by both sequential LAB V1 runners.""" + + definition_sha256: str + release_candidate_sha256: str + work_root: PortableLabV1WorkRootBinding + eomt: PortableLabV1ComponentInstallation + ddrnet: PortableLabV1ComponentInstallation + receipt_sha256: str + + def __post_init__(self) -> None: + _digest(self.definition_sha256, "definition sha256") + _digest(self.release_candidate_sha256, "release candidate sha256") + if self.eomt.component != "eomt" or self.ddrnet.component != "ddrnet": + raise PortableLabV1LocalRunnerError("component installations are misbound") + _digest(self.receipt_sha256, "runner installation receipt sha256") + if canonical_sha256(self.identity_document()) != self.receipt_sha256: + raise PortableLabV1LocalRunnerError("runner installation receipt changed") + + @classmethod + def seal( + cls, + *, + definition_sha256: str, + release_candidate_sha256: str, + work_root: PortableLabV1WorkRootBinding, + eomt: PortableLabV1ComponentInstallation, + ddrnet: PortableLabV1ComponentInstallation, + ) -> PortableLabV1RunnerInstallation: + identity = _runner_installation_identity( + definition_sha256=definition_sha256, + release_candidate_sha256=release_candidate_sha256, + work_root=work_root, + eomt=eomt, + ddrnet=ddrnet, + ) + return cls( + definition_sha256=definition_sha256, + release_candidate_sha256=release_candidate_sha256, + work_root=work_root, + eomt=eomt, + ddrnet=ddrnet, + receipt_sha256=canonical_sha256(identity), + ) + + def identity_document(self) -> dict[str, object]: + return _runner_installation_identity( + definition_sha256=self.definition_sha256, + release_candidate_sha256=self.release_candidate_sha256, + work_root=self.work_root, + eomt=self.eomt, + ddrnet=self.ddrnet, + ) + + def verify_unchanged(self) -> None: + if ( + canonical_sha256(self.eomt.identity_document()) != self.eomt.installation_sha256 + or canonical_sha256(self.ddrnet.identity_document()) != self.ddrnet.installation_sha256 + or canonical_sha256(self.identity_document()) != self.receipt_sha256 + ): + raise PortableLabV1LocalRunnerError("runner installation changed") + + +@dataclass(frozen=True, slots=True) +class PortableLabV1DockerMount: + host_path: str + container_path: str + read_only: bool + + def __post_init__(self) -> None: + if not isinstance(self.read_only, bool): + raise PortableLabV1LocalRunnerError("Docker mount mode is invalid") + _host_path(self.host_path, "Docker mount host path") + _container_path(self.container_path, "Docker mount target") + + def engine_document(self) -> dict[str, object]: + return { + "Type": "bind", + "Source": self.host_path, + "Target": self.container_path, + "ReadOnly": self.read_only, + "BindOptions": {"Propagation": "rprivate"}, + } + + +@dataclass(frozen=True, slots=True) +class PortableLabV1DockerLaunch: + component: PortableLabV1Component + image_sha256: str + entrypoint: tuple[str, ...] + command: tuple[str, ...] + mounts: tuple[PortableLabV1DockerMount, ...] + labels: Mapping[str, str] + timeout_seconds: float + memory_bytes: int + nano_cpus: int + name_token: str + local_output_root: Path + + def __post_init__(self) -> None: + if self.component not in ("eomt", "ddrnet"): + raise PortableLabV1LocalRunnerError("launch component is invalid") + _digest(self.image_sha256, "launch image sha256") + expected_image = _EOMT_IMAGE_SHA256 if self.component == "eomt" else _DDRNET_IMAGE_SHA256 + if self.image_sha256 != expected_image: + raise PortableLabV1LocalRunnerError("launch image identity changed") + _argv(self.entrypoint, "launch entrypoint") + _argv(self.command, "launch command") + _safe_token(self.name_token, "launch name token") + if not (1.0 <= self.timeout_seconds <= 24 * 60 * 60): + raise PortableLabV1LocalRunnerError("launch timeout is invalid") + if not (1024**3 <= self.memory_bytes <= 128 * 1024**3): + raise PortableLabV1LocalRunnerError("launch memory limit is invalid") + if not (100_000_000 <= self.nano_cpus <= 64_000_000_000): + raise PortableLabV1LocalRunnerError("launch CPU limit is invalid") + targets = tuple(mount.container_path for mount in self.mounts) + if targets != tuple(sorted(targets)) or len(targets) != len(set(targets)): + raise PortableLabV1LocalRunnerError("Docker mounts are not canonical") + writable = tuple(mount for mount in self.mounts if not mount.read_only) + if len(writable) != 1 or writable[0].container_path != "/run/nodedc/output": + raise PortableLabV1LocalRunnerError("Docker writable mount surface changed") + if any( + not ( + mount.container_path.startswith("/run/nodedc/") + or mount.container_path.startswith("/opt/nodedc/assets/") + ) + for mount in self.mounts + ): + raise PortableLabV1LocalRunnerError("Docker mount target surface changed") + output = _real_directory(self.local_output_root, "launch output root") + required_labels = { + "com.nodedc.authority", + "com.nodedc.component", + "com.nodedc.definition-sha256", + "com.nodedc.managed-by", + "com.nodedc.request-sha256", + "com.nodedc.product", + "com.nodedc.release-candidate-sha256", + "com.nodedc.stack", + } + if set(self.labels) != required_labels: + raise PortableLabV1LocalRunnerError("Docker launch label set changed") + if ( + self.labels.get("com.nodedc.authority") != "observation-only" + or self.labels.get("com.nodedc.component") != self.component + or self.labels.get("com.nodedc.managed-by") != "mission-core-worker" + or self.labels.get("com.nodedc.product") != "mission-core" + or self.labels.get("com.nodedc.stack") != "observatory" + ): + raise PortableLabV1LocalRunnerError("Docker launch labels changed") + for key in ( + "com.nodedc.definition-sha256", + "com.nodedc.request-sha256", + "com.nodedc.release-candidate-sha256", + ): + _digest(self.labels[key], f"Docker launch label {key}") + object.__setattr__(self, "local_output_root", output) + + +class PortableLabV1ContainerLauncher(Protocol): + def __call__(self, launch: PortableLabV1DockerLaunch) -> None: ... + + +@dataclass(frozen=True, slots=True) +class DockerEnginePortableLabV1Launcher: + """Run one sealed component through the local Docker Engine API.""" + + socket_path: Path = _DOCKER_SOCKET + api_version: str = _DOCKER_API_VERSION + transport_factory: Callable[[], httpx.BaseTransport] | None = None + + def __post_init__(self) -> None: + if not self.socket_path.is_absolute(): + raise PortableLabV1LocalRunnerError("Docker socket path is not absolute") + if not re.fullmatch(r"v[0-9]+\.[0-9]+", self.api_version): + raise PortableLabV1LocalRunnerError("Docker API version is invalid") + + def __call__(self, launch: PortableLabV1DockerLaunch) -> None: + if self.transport_factory is None: + _require_local_socket(self.socket_path) + transport: httpx.BaseTransport = httpx.HTTPTransport(uds=str(self.socket_path)) + else: + transport = self.transport_factory() + container_id: str | None = None + primary_error: BaseException | None = None + try: + timeout = httpx.Timeout(launch.timeout_seconds, connect=5.0) + with httpx.Client( + base_url="http://docker", + transport=transport, + timeout=timeout, + ) as client: + self._verify_image(client, launch.image_sha256) + container_id, creation_warnings = self._create(client, launch) + if creation_warnings: + raise PortableLabV1LocalRunnerError( + "Docker container creation returned warnings" + ) + self._empty( + client, + "POST", + self._api_path(f"/containers/{container_id}/start"), + {204}, + ) + status_code = self._wait(client, container_id) + if status_code != 0: + log_identity, log_bytes = self._log_identity(client, container_id) + raise PortableLabV1LocalRunnerError( + f"{launch.component} container exited with status {status_code}; " + f"logs={log_identity}:{log_bytes}" + ) + except (httpx.HTTPError, OSError, ValueError) as exc: + primary_error = exc + raise PortableLabV1LocalRunnerError( + f"local Docker Engine {launch.component} execution failed" + ) from exc + except BaseException as exc: + primary_error = exc + raise + finally: + if container_id is not None: + cleanup_error = self._cleanup(container_id) + if cleanup_error is not None: + raise PortableLabV1LocalRunnerError( + f"{launch.component} container cleanup failed after retry" + ) from (primary_error or cleanup_error) + + def _verify_image(self, client: httpx.Client, image_sha256: str) -> None: + response = self._response( + client, + "GET", + self._api_path(f"/images/sha256:{image_sha256}/json"), + {200}, + ) + document = _response_object(response, "Docker image inspection") + if document.get("Id") != f"sha256:{image_sha256}": + raise PortableLabV1LocalRunnerError("Docker image identity changed") + + def _create( + self, + client: httpx.Client, + launch: PortableLabV1DockerLaunch, + ) -> tuple[str, bool]: + name = f"ndc-observatory-lab-v1-{launch.component}-{launch.name_token}" + response = self._response( + client, + "POST", + self._api_path(f"/containers/create?name={name}"), + {201}, + json_body=_container_create_document(launch), + ) + document = _response_object(response, "Docker container creation") + container_id = document.get("Id") + warnings = document.get("Warnings") + if not isinstance(container_id, str) or not _CONTAINER_ID.fullmatch(container_id): + raise PortableLabV1LocalRunnerError("Docker container id is invalid") + return container_id, warnings not in (None, []) + + def _cleanup(self, container_id: str) -> BaseException | None: + last_error: BaseException | None = None + for _attempt in range(2): + try: + cleanup_transport = ( + httpx.HTTPTransport(uds=str(self.socket_path)) + if self.transport_factory is None + else self.transport_factory() + ) + with httpx.Client( + base_url="http://docker", + transport=cleanup_transport, + timeout=httpx.Timeout(10.0, connect=5.0), + ) as cleanup_client: + self._empty( + cleanup_client, + "DELETE", + self._api_path(f"/containers/{container_id}?force=1&v=1"), + {204, 404}, + ) + return None + except ( + httpx.HTTPError, + OSError, + ValueError, + PortableLabV1LocalRunnerError, + ) as exc: + last_error = exc + return last_error + + def _wait(self, client: httpx.Client, container_id: str) -> int: + response = self._response( + client, + "POST", + self._api_path(f"/containers/{container_id}/wait?condition=not-running"), + {200}, + ) + document = _response_object(response, "Docker container wait") + status_code = document.get("StatusCode") + error = document.get("Error") + if isinstance(status_code, bool) or not isinstance(status_code, int): + raise PortableLabV1LocalRunnerError("Docker exit status is invalid") + if error not in (None, {"Message": ""}): + raise PortableLabV1LocalRunnerError("Docker wait returned an engine error") + return status_code + + def _log_identity(self, client: httpx.Client, container_id: str) -> tuple[str, int]: + path = self._api_path(f"/containers/{container_id}/logs?stdout=1&stderr=1&tail=200") + payload = bytearray() + with client.stream("GET", path) as response: + if response.status_code != 200: + raise PortableLabV1LocalRunnerError( + f"Docker Engine rejected GET {path.split('?')[0]} " + f"with status {response.status_code}" + ) + blocks = ( + (response.content,) + if response.is_stream_consumed + else response.iter_raw(chunk_size=16 * 1024) + ) + for block in blocks: + remaining = _MAX_ENGINE_ERROR_BYTES - len(payload) + if remaining <= 0: + break + payload.extend(block[:remaining]) + if len(block) > remaining: + break + return hashlib.sha256(payload).hexdigest(), len(payload) + + def _api_path(self, path: str) -> str: + return f"/{self.api_version}{path}" + + @staticmethod + def _empty( + client: httpx.Client, + method: str, + path: str, + statuses: set[int], + ) -> None: + DockerEnginePortableLabV1Launcher._response( + client, + method, + path, + statuses, + ) + + @staticmethod + def _response( + client: httpx.Client, + method: str, + path: str, + statuses: set[int], + *, + json_body: Mapping[str, object] | None = None, + ) -> httpx.Response: + with client.stream(method, path, json=json_body) as streamed: + if streamed.status_code not in statuses: + raise PortableLabV1LocalRunnerError( + f"Docker Engine rejected {method} {path.split('?')[0]} " + f"with status {streamed.status_code}" + ) + declared_length = streamed.headers.get("content-length") + if declared_length is not None: + try: + length = int(declared_length) + except ValueError as exc: + raise PortableLabV1LocalRunnerError( + "Docker Engine response length is invalid" + ) from exc + if length < 0 or length > _MAX_ENGINE_ERROR_BYTES: + raise PortableLabV1LocalRunnerError("Docker Engine response is too large") + payload = bytearray() + if streamed.is_stream_consumed: + # Mock and other in-memory transports may supply a response whose + # body is already materialized. The production HTTPTransport path + # remains streamed and therefore never allocates an unbounded body. + content = streamed.content + if len(content) > _MAX_ENGINE_ERROR_BYTES: + raise PortableLabV1LocalRunnerError("Docker Engine response is too large") + payload.extend(content) + else: + for block in streamed.iter_raw(chunk_size=16 * 1024): + if len(payload) + len(block) > _MAX_ENGINE_ERROR_BYTES: + raise PortableLabV1LocalRunnerError("Docker Engine response is too large") + payload.extend(block) + return httpx.Response( + status_code=streamed.status_code, + headers=streamed.headers, + content=bytes(payload), + request=streamed.request, + ) + + +@dataclass(frozen=True, slots=True) +class InstalledPortableLabV1EomtRunner: + installation: PortableLabV1RunnerInstallation + launcher: PortableLabV1ContainerLauncher + token_factory: Callable[[], str] = lambda: secrets.token_hex(12) + + def __call__( + self, + *, + source: PortableLabV1MaterializedSource, + plan: PortableLabV1OrchestrationPlan, + output_root: Path, + ) -> None: + _verify_call(self.installation, source=source, plan=plan) + component = self.installation.eomt + token = self.token_factory() + staging, invocation, request_path = _prepare_invocation( + self.installation.work_root, + output_root=output_root, + token=token, + component="eomt", + ) + published = False + try: + request = _component_request( + component="eomt", + source=source, + plan=plan, + assets=component.assets, + effective_config_sha256=None, + ) + _write_request(request_path, request) + launch = _launch( + installation=self.installation, + component=component, + source=source, + staging=staging, + request_path=request_path, + token=token, + ) + self.launcher(launch) + _validate_eomt_result(staging, plan) + os.rename(staging, output_root) + published = True + finally: + shutil.rmtree(invocation, ignore_errors=True) + if not published: + shutil.rmtree(staging, ignore_errors=True) + + +@dataclass(frozen=True, slots=True) +class InstalledPortableLabV1DdrnetRunner: + installation: PortableLabV1RunnerInstallation + launcher: PortableLabV1ContainerLauncher + token_factory: Callable[[], str] = lambda: secrets.token_hex(12) + + def __call__( + self, + *, + source: PortableLabV1MaterializedSource, + plan: PortableLabV1OrchestrationPlan, + effective_config_path: Path, + eomt_result_root: Path, + output_root: Path, + ) -> None: + _verify_call(self.installation, source=source, plan=plan) + effective = _real_file(effective_config_path, "effective DDRNet config") + if ( + effective.read_bytes() != canonical_json(plan.effective_ddrnet_config) + or _sha256_file(effective) != plan.effective_ddrnet_config_sha256 + ): + raise PortableLabV1LocalRunnerError("effective DDRNet config changed") + eomt_root = _real_directory(eomt_result_root, "EoMT result root") + self.installation.work_root.engine_path(eomt_root, label="EoMT result root") + _validate_eomt_result(eomt_root, plan) + component = self.installation.ddrnet + token = self.token_factory() + staging, invocation, request_path = _prepare_invocation( + self.installation.work_root, + output_root=output_root, + token=token, + component="ddrnet", + ) + published = False + try: + request = _component_request( + component="ddrnet", + source=source, + plan=plan, + assets=component.assets, + effective_config_sha256=plan.effective_ddrnet_config_sha256, + ) + _write_request(request_path, request) + extra_mounts = ( + _mapped_mount( + self.installation.work_root, + effective, + "/run/nodedc/effective-ddrnet-config.json", + read_only=True, + label="effective DDRNet config", + ), + _mapped_mount( + self.installation.work_root, + eomt_root, + "/run/nodedc/eomt-result", + read_only=True, + label="EoMT result root", + ), + ) + launch = _launch( + installation=self.installation, + component=component, + source=source, + staging=staging, + request_path=request_path, + token=token, + extra_mounts=extra_mounts, + ) + self.launcher(launch) + _validate_ddrnet_result(staging, plan) + os.rename(staging, output_root) + published = True + finally: + shutil.rmtree(invocation, ignore_errors=True) + if not published: + shutil.rmtree(staging, ignore_errors=True) + + +@dataclass(frozen=True, slots=True) +class PortableLabV1InstalledRunners: + eomt: InstalledPortableLabV1EomtRunner + ddrnet: InstalledPortableLabV1DdrnetRunner + + +def compose_portable_lab_v1_installed_runners( + *, + installation: PortableLabV1RunnerInstallation, + launcher: PortableLabV1ContainerLauncher, +) -> PortableLabV1InstalledRunners: + """Bind one verified installation to the two existing runner protocols.""" + + installation.verify_unchanged() + return PortableLabV1InstalledRunners( + eomt=InstalledPortableLabV1EomtRunner( + installation=installation, + launcher=launcher, + ), + ddrnet=InstalledPortableLabV1DdrnetRunner( + installation=installation, + launcher=launcher, + ), + ) + + +def _component_installation_identity( + *, + component: PortableLabV1Component, + image_sha256: str, + entrypoint: tuple[str, ...], + command: tuple[str, ...], + assets: tuple[PortableLabV1HostAsset, ...], + timeout_seconds: float, + memory_bytes: int, + nano_cpus: int, +) -> dict[str, object]: + return { + "schema_version": PORTABLE_LAB_V1_COMPONENT_INSTALLATION_SCHEMA, + "component": component, + "image_sha256": image_sha256, + "entrypoint": list(entrypoint), + "command": list(command), + "assets": [asset.identity_document() for asset in assets], + "timeout_seconds": timeout_seconds, + "memory_bytes": memory_bytes, + "nano_cpus": nano_cpus, + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + + +def _runner_installation_identity( + *, + definition_sha256: str, + release_candidate_sha256: str, + work_root: PortableLabV1WorkRootBinding, + eomt: PortableLabV1ComponentInstallation, + ddrnet: PortableLabV1ComponentInstallation, +) -> dict[str, object]: + return { + "schema_version": PORTABLE_LAB_V1_RUNNER_INSTALLATION_SCHEMA, + "setup_id": _SETUP_ID, + "definition_id": _DEFINITION_ID, + "definition_version": _DEFINITION_VERSION, + "definition_sha256": definition_sha256, + "release_candidate_sha256": release_candidate_sha256, + "work_root": work_root.identity_document(), + "components": { + "eomt": eomt.installation_sha256, + "ddrnet": ddrnet.installation_sha256, + }, + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + + +def _verify_call( + installation: PortableLabV1RunnerInstallation, + *, + source: PortableLabV1MaterializedSource, + plan: PortableLabV1OrchestrationPlan, +) -> None: + installation.verify_unchanged() + plan.require_executable() + if ( + plan.setup_id != _SETUP_ID + or plan.definition_id != _DEFINITION_ID + or plan.definition_version != _DEFINITION_VERSION + or plan.definition_sha256 != installation.definition_sha256 + or plan.release_candidate_sha256 != installation.release_candidate_sha256 + or plan.source_input != source.descriptor + ): + raise PortableLabV1LocalRunnerError("runner call differs from its installation") + installation.work_root.engine_path(source.root, label="materialized source root") + installation.work_root.engine_path(source.camera_job_root, label="camera job root") + validate_camera_compute_job(source.camera_job_root) + + +def _prepare_invocation( + binding: PortableLabV1WorkRootBinding, + *, + output_root: Path, + token: str, + component: PortableLabV1Component, +) -> tuple[Path, Path, Path]: + _safe_token(token, "runner token") + parent = _real_directory(output_root.parent, "component output parent") + binding.engine_path(parent, label="component output parent") + if output_root.exists() or output_root.is_symlink(): + raise PortableLabV1LocalRunnerError("component output root already exists") + staging = parent / f".lab-v1-{component}-result-{token}" + invocation = parent / f".lab-v1-{component}-invocation-{token}" + if staging.exists() or invocation.exists(): + raise PortableLabV1LocalRunnerError("component invocation token collided") + try: + staging.mkdir(mode=0o700) + invocation.mkdir(mode=0o700) + binding.engine_path(staging, label="component staging root") + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + shutil.rmtree(invocation, ignore_errors=True) + raise + return staging, invocation, invocation / "request.json" + + +def _component_request( + *, + component: PortableLabV1Component, + source: PortableLabV1MaterializedSource, + plan: PortableLabV1OrchestrationPlan, + assets: tuple[PortableLabV1HostAsset, ...], + effective_config_sha256: str | None, +) -> dict[str, object]: + return { + "schema_version": PORTABLE_LAB_V1_COMPONENT_REQUEST_SCHEMA, + "component": component, + "plan_sha256": plan.plan_sha256, + "definition_sha256": plan.definition_sha256, + "release_candidate_sha256": plan.release_candidate_sha256, + "source": source.descriptor.as_dict(), + "paths": { + "camera_job_root": "/run/nodedc/source", + "request": "/run/nodedc/request.json", + "output_root": "/run/nodedc/output", + "effective_ddrnet_config": ( + "/run/nodedc/effective-ddrnet-config.json" if component == "ddrnet" else None + ), + "eomt_result_root": ("/run/nodedc/eomt-result" if component == "ddrnet" else None), + "decoded_frames_root": ( + "/run/nodedc/output/source-frames" + if component == "eomt" + else "/run/nodedc/eomt-result/source-frames" + ), + }, + "effective_ddrnet_config_sha256": effective_config_sha256, + "assets": [asset.request_document() for asset in assets], + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + + +def _write_request(path: Path, document: Mapping[str, object]) -> None: + path.write_bytes(canonical_json(document)) + os.chmod(path, 0o400) + + +def _launch( + *, + installation: PortableLabV1RunnerInstallation, + component: PortableLabV1ComponentInstallation, + source: PortableLabV1MaterializedSource, + staging: Path, + request_path: Path, + token: str, + extra_mounts: tuple[PortableLabV1DockerMount, ...] = (), +) -> PortableLabV1DockerLaunch: + binding = installation.work_root + mounts = [ + _mapped_mount( + binding, + staging, + "/run/nodedc/output", + read_only=False, + label="component staging root", + ), + _mapped_mount( + binding, + request_path, + "/run/nodedc/request.json", + read_only=True, + label="component request", + ), + _mapped_mount( + binding, + source.camera_job_root, + "/run/nodedc/source", + read_only=True, + label="camera job root", + ), + *extra_mounts, + ] + mounts.extend( + PortableLabV1DockerMount( + host_path=asset.host_path, + container_path=asset.container_path, + read_only=True, + ) + for asset in component.assets + ) + labels = { + "com.nodedc.authority": "observation-only", + "com.nodedc.component": component.component, + "com.nodedc.definition-sha256": installation.definition_sha256, + "com.nodedc.managed-by": "mission-core-worker", + "com.nodedc.request-sha256": _read_request_digest(request_path), + "com.nodedc.product": "mission-core", + "com.nodedc.release-candidate-sha256": (installation.release_candidate_sha256), + "com.nodedc.stack": "observatory", + } + return PortableLabV1DockerLaunch( + component=component.component, + image_sha256=component.image_sha256, + entrypoint=component.entrypoint, + command=(*component.command, "--request", "/run/nodedc/request.json"), + mounts=tuple(sorted(mounts, key=lambda item: item.container_path)), + labels=labels, + timeout_seconds=component.timeout_seconds, + memory_bytes=component.memory_bytes, + nano_cpus=component.nano_cpus, + name_token=token, + local_output_root=staging, + ) + + +def _mapped_mount( + binding: PortableLabV1WorkRootBinding, + path: Path, + container_path: str, + *, + read_only: bool, + label: str, +) -> PortableLabV1DockerMount: + return PortableLabV1DockerMount( + host_path=binding.engine_path(path, label=label), + container_path=container_path, + read_only=read_only, + ) + + +def _container_create_document(launch: PortableLabV1DockerLaunch) -> dict[str, object]: + return { + "Image": f"sha256:{launch.image_sha256}", + "Entrypoint": list(launch.entrypoint), + "Cmd": list(launch.command), + "Env": [ + "HF_HUB_OFFLINE=1", + "TRANSFORMERS_OFFLINE=1", + "PYTHONDONTWRITEBYTECODE=1", + ], + "Labels": dict(sorted(launch.labels.items())), + "NetworkDisabled": True, + "OpenStdin": False, + "StdinOnce": False, + "Tty": False, + "AttachStdout": True, + "AttachStderr": True, + "HostConfig": { + "AutoRemove": False, + "CapDrop": ["ALL"], + "DeviceRequests": [ + { + "Driver": "nvidia", + "Count": 1, + "Capabilities": [["gpu"]], + } + ], + "Init": True, + "IpcMode": "private", + "Memory": launch.memory_bytes, + "MemorySwap": launch.memory_bytes, + "Mounts": [mount.engine_document() for mount in launch.mounts], + "NanoCpus": launch.nano_cpus, + "NetworkMode": "none", + "PidsLimit": 512, + "Privileged": False, + "ReadonlyRootfs": True, + "SecurityOpt": ["no-new-privileges:true"], + "ShmSize": 4 * 1024**3, + "Tmpfs": {"/tmp": "rw,noexec,nosuid,nodev,size=2147483648,mode=1777"}, + }, + } + + +def _validate_eomt_result(root: Path, plan: PortableLabV1OrchestrationPlan) -> None: + document = _result_document(root, "EoMT") + identity = _object(document.get("identity"), "EoMT identity") + configuration = _object(identity.get("configuration"), "EoMT configuration") + if ( + document.get("schema_version") != EOMT_RESULT_SCHEMA + or document.get("job_id") != plan.source_input.camera_job_id + or document.get("input_sha256") != plan.source_input.camera_input_sha256 + or document.get("session_id") != plan.source_input.source_session_id + or document.get("source_id") != plan.source_input.camera_source_id + or document.get("codec_epoch") != plan.source_input.codec_epoch + or document.get("frames_processed") != plan.source_input.frame_count + or document.get("ground_truth") is not False + or configuration.get("pipeline") != _EOMT_PIPELINE + ): + raise PortableLabV1LocalRunnerError("EoMT component result changed") + frames_root = _real_directory(root / "source-frames", "decoded source frames") + expected_names = tuple( + f"frame-{sequence:06d}.png" for sequence in range(1, plan.source_input.frame_count + 1) + ) + try: + observed_names = tuple(sorted(path.name for path in frames_root.iterdir())) + except OSError as exc: + raise PortableLabV1LocalRunnerError("decoded source frames are unavailable") from exc + if observed_names != expected_names: + raise PortableLabV1LocalRunnerError("decoded source frame set changed") + for name in expected_names: + _real_file(frames_root / name, "decoded source frame") + + +def _validate_ddrnet_result(root: Path, plan: PortableLabV1OrchestrationPlan) -> None: + document = _result_document(root, "DDRNet") + candidate = _object(document.get("candidate"), "DDRNet candidate") + source = _object(document.get("source"), "DDRNet source") + provenance = _object(document.get("provenance"), "DDRNet provenance") + effective_source = _object( + plan.effective_ddrnet_config.get("ravnoves"), + "effective DDRNet source", + ) + if ( + document.get("schema_version") != DDRNET_RESULT_SCHEMA + or document.get("mode") != "ravnoves-video" + or candidate.get("candidate_id") != "goose-ddrnet-class-512" + or candidate.get("candidate_key") != "ddrnet" + or candidate.get("checkpoint_sha256") != _DDRNET_CHECKPOINT_SHA256 + or source.get("source_id") != effective_source.get("source_id") + or source.get("input_count") != plan.source_input.frame_count + or source.get("mapping_sha256") != _GOOSE_MAPPING_SHA256 + or provenance.get("config_sha256") != plan.effective_ddrnet_config_sha256 + ): + raise PortableLabV1LocalRunnerError("DDRNet component result changed") + + +def _result_document(root: Path, label: str) -> dict[str, object]: + resolved = _real_directory(root, f"{label} result root") + path = _real_file(resolved / "result.json", f"{label} result document") + if path.parent != resolved or path.stat().st_size > _MAX_RESULT_DOCUMENT_BYTES: + raise PortableLabV1LocalRunnerError(f"{label} result document is invalid") + try: + value: object = json.loads(path.read_text(encoding="utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise PortableLabV1LocalRunnerError(f"{label} result document is not valid JSON") from exc + return _object(value, f"{label} result document") + + +def _response_object(response: httpx.Response, label: str) -> dict[str, object]: + if len(response.content) > _MAX_ENGINE_ERROR_BYTES: + raise PortableLabV1LocalRunnerError(f"{label} response is too large") + try: + value: object = response.json() + except json.JSONDecodeError as exc: + raise PortableLabV1LocalRunnerError(f"{label} response is invalid") from exc + return _object(value, label) + + +def _read_request_digest(path: Path) -> str: + return _sha256_file(_real_file(path, "component request")) + + +def _real_path(path: Path, label: str) -> Path: + try: + if not path.is_absolute() or path.is_symlink(): + raise PortableLabV1LocalRunnerError(f"{label} is a symbolic link") + resolved = path.resolve(strict=True) + if path.absolute() != resolved: + raise PortableLabV1LocalRunnerError(f"{label} contains a symbolic link or traversal") + return resolved + except OSError as exc: + raise PortableLabV1LocalRunnerError(f"{label} is unavailable") from exc + + +def _real_directory(path: Path, label: str) -> Path: + resolved = _real_path(path, label) + if not resolved.is_dir(): + raise PortableLabV1LocalRunnerError(f"{label} is not a directory") + return resolved + + +def _real_file(path: Path, label: str) -> Path: + resolved = _real_path(path, label) + if not resolved.is_file(): + raise PortableLabV1LocalRunnerError(f"{label} is not a regular file") + return resolved + + +def _host_path(value: str, label: str) -> None: + if not isinstance(value, str) or not _WINDOWS_D_PATH.fullmatch(value) or "/" in value: + raise PortableLabV1LocalRunnerError(f"{label} is not an absolute D: path") + parts = value[3:].split("\\") if len(value) > 3 else [] + if any(part in ("", ".", "..") or part.endswith((" ", ".")) for part in parts): + raise PortableLabV1LocalRunnerError(f"{label} is not canonical") + if value.casefold() != "d:\\nDC_missioncore".casefold() and not value.casefold().startswith( + "d:\\ndc_missioncore\\" + ): + raise PortableLabV1LocalRunnerError(f"{label} is outside D:\\NDC_MISSIONCORE") + + +def _windows_relative_part(value: str, label: str) -> None: + if ( + not value + or value in (".", "..") + or value.endswith((" ", ".")) + or any(character in value for character in '\\/:*?"<>|') + or any(ord(character) < 32 for character in value) + or _WINDOWS_RESERVED_NAME.fullmatch(value) is not None + ): + raise PortableLabV1LocalRunnerError(f"{label} contains a Windows-unsafe path component") + + +def _container_path(value: str, label: str) -> PurePosixPath: + path = PurePosixPath(value) + if ( + not value.startswith("/") + or "//" in value + or path == PurePosixPath("/") + or any(part in (".", "..") for part in path.parts) + ): + raise PortableLabV1LocalRunnerError(f"{label} is invalid") + return path + + +def _argv(values: tuple[str, ...], label: str) -> None: + if not values or len(values) > 64: + raise PortableLabV1LocalRunnerError(f"{label} is invalid") + if any( + not isinstance(value, str) or not value or len(value) > 4096 or "\x00" in value + for value in values + ): + raise PortableLabV1LocalRunnerError(f"{label} is invalid") + + +def _safe_token(value: str, label: str) -> None: + if not isinstance(value, str) or not _SAFE_TOKEN.fullmatch(value): + raise PortableLabV1LocalRunnerError(f"{label} is invalid") + + +def _digest(value: str, label: str) -> None: + if not isinstance(value, str) or not _SHA256.fullmatch(value): + raise PortableLabV1LocalRunnerError(f"{label} is invalid") + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise PortableLabV1LocalRunnerError(f"{label} is not an object") + return cast(dict[str, object], value) + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _require_local_socket(path: Path) -> None: + try: + mode = path.stat().st_mode + except OSError as exc: + raise PortableLabV1LocalRunnerError("local Docker socket is unavailable") from exc + if not stat.S_ISSOCK(mode): + raise PortableLabV1LocalRunnerError("local Docker socket is not a socket") diff --git a/tests/test_observatory_portable_lab_v1_local_runners.py b/tests/test_observatory_portable_lab_v1_local_runners.py new file mode 100644 index 0000000..e0ae168 --- /dev/null +++ b/tests/test_observatory_portable_lab_v1_local_runners.py @@ -0,0 +1,531 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +import httpx +import pytest + +from k1link.observatory import portable_lab_v1_local_runners as local_runners +from k1link.observatory.portable_lab_v1_executor import ( + PortableLabV1MaterializedSource, + PortableLabV1OrchestrationPlan, + PortableLabV1SourceInput, +) +from k1link.observatory.portable_result_contract import canonical_json +from k1link.observatory.portable_run_definitions import canonical_sha256 + +EOMT_IMAGE_SHA256 = "58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" +DDRNET_IMAGE_SHA256 = "591cb382c099eeb05e7ec16e2371e0b2da54d2bb5c49ec0f4ac88dbf72b0f0cd" +DEFINITION_SHA256 = "a" * 64 +RELEASE_SHA256 = "b" * 64 + + +@dataclass(frozen=True) +class _Plan: + setup_id: str + definition_id: str + definition_version: int + definition_sha256: str + release_candidate_sha256: str + source_input: PortableLabV1SourceInput + effective_ddrnet_config: dict[str, object] + effective_ddrnet_config_sha256: str + plan_sha256: str + + def require_executable(self) -> None: + return None + + +def _assets( + component: local_runners.PortableLabV1Component, +) -> tuple[local_runners.PortableLabV1HostAsset, ...]: + identities = local_runners._EXPECTED_ASSET_IDENTITIES[component] + tree_assets = {"eomt-dependency-set", "k1-valid-fov-identity"} + return tuple( + local_runners.PortableLabV1HostAsset( + asset_id=asset_id, + host_path=f"D:\\NDC_MISSIONCORE\\assets\\{asset_id}", + container_path=f"/opt/nodedc/assets/{asset_id}", + kind="tree" if asset_id in tree_assets else "file", + verification=("identity-sha256" if asset_id in tree_assets else "sha256"), + identity_sha256=identity, + byte_length=local_runners._EXPECTED_ASSET_LENGTHS.get(asset_id), + ) + for asset_id, identity in sorted(identities.items()) + ) + + +def _component( + component: local_runners.PortableLabV1Component, +) -> local_runners.PortableLabV1ComponentInstallation: + return local_runners.PortableLabV1ComponentInstallation.seal( + component=component, + image_sha256=(EOMT_IMAGE_SHA256 if component == "eomt" else DDRNET_IMAGE_SHA256), + entrypoint=("/opt/nodedc/bin/portable-lab-v1-agent",), + command=(f"run-{component}",), + assets=_assets(component), + timeout_seconds=3600.0, + memory_bytes=16 * 1024**3, + nano_cpus=4_000_000_000, + ) + + +def _installation(tmp_path: Path) -> local_runners.PortableLabV1RunnerInstallation: + return local_runners.PortableLabV1RunnerInstallation.seal( + definition_sha256=DEFINITION_SHA256, + release_candidate_sha256=RELEASE_SHA256, + work_root=local_runners.PortableLabV1WorkRootBinding( + controller_root=tmp_path, + engine_host_root="D:\\NDC_MISSIONCORE\\runtime\\observatory-worker", + ), + eomt=_component("eomt"), + ddrnet=_component("ddrnet"), + ) + + +def _source(tmp_path: Path) -> PortableLabV1MaterializedSource: + root = tmp_path / "source" + camera = root / "camera-job" + camera.mkdir(parents=True) + descriptor = PortableLabV1SourceInput( + observatory_job_id=f"observatory-run-{'1' * 32}", + observatory_request_sha256="2" * 64, + observatory_identity_sha256="3" * 64, + source_session_id="20260831T083000Z_viewer_live", + source_catalog_sha256="4" * 64, + source_bundle_sha256="5" * 64, + source_capability_manifest_sha256="6" * 64, + source_adapter_sha256="7" * 64, + camera_job_id=f"recorded-camera-{'8' * 24}", + camera_input_sha256="9" * 64, + camera_source_id="sensor.camera.right", + codec_epoch=1, + input_byte_length=64, + frame_count=2, + timeline_start_seconds=1.0, + timeline_end_seconds=3.0, + camera_generation_sha256="c" * 64, + calibration_sha256="d" * 64, + ) + return PortableLabV1MaterializedSource( + root=root, + camera_job_root=camera, + descriptor=descriptor, + ) + + +def _plan(source: PortableLabV1MaterializedSource) -> PortableLabV1OrchestrationPlan: + effective: dict[str, object] = { + "schema_version": "missioncore.lab-v1-goose-vegetation-benchmark/v1", + "ravnoves": { + "source_id": ( + "portable-k1/20260831T083000Z_viewer_live/" + f"sensor.camera.right@{source.descriptor.camera_input_sha256}" + ), + "source_sha256": source.descriptor.camera_input_sha256, + }, + } + plan = _Plan( + setup_id="lab-v1-eomt-ddrnet-portable-v1", + definition_id="lab-v1-eomt-ddrnet-portable", + definition_version=2, + definition_sha256=DEFINITION_SHA256, + release_candidate_sha256=RELEASE_SHA256, + source_input=source.descriptor, + effective_ddrnet_config=effective, + effective_ddrnet_config_sha256=canonical_sha256(effective), + plan_sha256="e" * 64, + ) + return cast(PortableLabV1OrchestrationPlan, plan) + + +def _docker_launch(tmp_path: Path) -> local_runners.PortableLabV1DockerLaunch: + output = tmp_path / "output" + output.mkdir() + return local_runners.PortableLabV1DockerLaunch( + component="eomt", + image_sha256=EOMT_IMAGE_SHA256, + entrypoint=("/opt/nodedc/bin/portable-lab-v1-agent",), + command=("run-eomt", "--request", "/run/nodedc/request.json"), + mounts=( + local_runners.PortableLabV1DockerMount( + host_path="D:\\NDC_MISSIONCORE\\runtime\\output", + container_path="/run/nodedc/output", + read_only=False, + ), + local_runners.PortableLabV1DockerMount( + host_path="D:\\NDC_MISSIONCORE\\runtime\\request.json", + container_path="/run/nodedc/request.json", + read_only=True, + ), + ), + labels={ + "com.nodedc.authority": "observation-only", + "com.nodedc.component": "eomt", + "com.nodedc.definition-sha256": DEFINITION_SHA256, + "com.nodedc.managed-by": "mission-core-worker", + "com.nodedc.request-sha256": "c" * 64, + "com.nodedc.product": "mission-core", + "com.nodedc.release-candidate-sha256": RELEASE_SHA256, + "com.nodedc.stack": "observatory", + }, + timeout_seconds=60.0, + memory_bytes=8 * 1024**3, + nano_cpus=2_000_000_000, + name_token="0123456789abcdef", + local_output_root=output, + ) + + +def test_docker_engine_launcher_uses_exact_hardened_sibling_contract( + tmp_path: Path, +) -> None: + requests: list[httpx.Request] = [] + create_document: dict[str, object] = {} + container_id = "f" * 64 + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path.startswith("/v1.47/images/"): + return httpx.Response(200, json={"Id": f"sha256:{EOMT_IMAGE_SHA256}"}) + if request.url.path == "/v1.47/containers/create": + create_document.update(json.loads(request.content)) + return httpx.Response(201, json={"Id": container_id, "Warnings": None}) + if request.url.path.endswith("/start"): + return httpx.Response(204) + if request.url.path.endswith("/wait"): + return httpx.Response(200, json={"StatusCode": 0, "Error": None}) + if request.method == "DELETE": + return httpx.Response(204) + raise AssertionError(f"unexpected request: {request.method} {request.url}") + + launcher = local_runners.DockerEnginePortableLabV1Launcher( + transport_factory=lambda: httpx.MockTransport(handler) + ) + launcher(_docker_launch(tmp_path)) + + assert [request.method for request in requests] == [ + "GET", + "POST", + "POST", + "POST", + "DELETE", + ] + assert create_document["Image"] == f"sha256:{EOMT_IMAGE_SHA256}" + assert create_document["NetworkDisabled"] is True + host = cast(dict[str, object], create_document["HostConfig"]) + assert host["NetworkMode"] == "none" + assert host["ReadonlyRootfs"] is True + assert host["CapDrop"] == ["ALL"] + assert host["SecurityOpt"] == ["no-new-privileges:true"] + assert host["Privileged"] is False + assert host["DeviceRequests"] == [{"Driver": "nvidia", "Count": 1, "Capabilities": [["gpu"]]}] + mounts = cast(list[dict[str, object]], host["Mounts"]) + assert sum(not cast(bool, mount["ReadOnly"]) for mount in mounts) == 1 + assert next(mount for mount in mounts if not mount["ReadOnly"])["Target"] == ( + "/run/nodedc/output" + ) + + +def test_docker_engine_launcher_fails_closed_and_cleans_nonzero_container( + tmp_path: Path, +) -> None: + requests: list[httpx.Request] = [] + container_id = "a" * 64 + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path.startswith("/v1.47/images/"): + return httpx.Response(200, json={"Id": f"sha256:{EOMT_IMAGE_SHA256}"}) + if request.url.path == "/v1.47/containers/create": + return httpx.Response(201, json={"Id": container_id, "Warnings": None}) + if request.url.path.endswith("/start"): + return httpx.Response(204) + if request.url.path.endswith("/wait"): + return httpx.Response(200, json={"StatusCode": 17, "Error": None}) + if request.url.path.endswith("/logs"): + return httpx.Response(200, content=b"sensitive worker error") + if request.method == "DELETE": + return httpx.Response(204) + raise AssertionError(f"unexpected request: {request.method} {request.url}") + + launcher = local_runners.DockerEnginePortableLabV1Launcher( + transport_factory=lambda: httpx.MockTransport(handler) + ) + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match=r"exited with status 17; logs=[a-f0-9]{64}:22", + ): + launcher(_docker_launch(tmp_path)) + assert requests[-1].method == "DELETE" + assert "sensitive worker error" not in str(requests) + + +def test_work_root_binding_rejects_escape_and_non_d_host_root(tmp_path: Path) -> None: + binding = local_runners.PortableLabV1WorkRootBinding( + controller_root=tmp_path, + engine_host_root="D:\\NDC_MISSIONCORE\\runtime\\observatory-worker", + ) + admitted = tmp_path / "jobs" / "job-001" + admitted.mkdir(parents=True) + assert binding.engine_path(admitted, label="job") == ( + "D:\\NDC_MISSIONCORE\\runtime\\observatory-worker\\jobs\\job-001" + ) + outside = tmp_path.parent / "outside" + outside.mkdir(exist_ok=True) + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="outside the installed work root", + ): + binding.engine_path(outside, label="outside") + windows_escape = tmp_path / r"foo\..\outside" + windows_escape.mkdir() + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="Windows-unsafe path component", + ): + binding.engine_path(windows_escape, label="Windows escape") + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="absolute D: path", + ): + local_runners.PortableLabV1WorkRootBinding( + controller_root=tmp_path, + engine_host_root="C:\\temp", + ) + + +def test_docker_engine_launcher_bounds_control_plane_responses(tmp_path: Path) -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response( + 200, + headers={"content-length": str(local_runners._MAX_ENGINE_ERROR_BYTES + 1)}, + ) + + launcher = local_runners.DockerEnginePortableLabV1Launcher( + transport_factory=lambda: httpx.MockTransport(handler) + ) + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="Docker Engine response is too large", + ): + launcher(_docker_launch(tmp_path)) + + assert [request.method for request in requests] == ["GET"] + + +def test_docker_engine_launcher_retries_cleanup_after_create_warning( + tmp_path: Path, +) -> None: + requests: list[httpx.Request] = [] + container_id = "b" * 64 + cleanup_attempts = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal cleanup_attempts + requests.append(request) + if request.url.path.startswith("/v1.47/images/"): + return httpx.Response(200, json={"Id": f"sha256:{EOMT_IMAGE_SHA256}"}) + if request.url.path == "/v1.47/containers/create": + return httpx.Response( + 201, + json={"Id": container_id, "Warnings": ["engine warning"]}, + ) + if request.method == "DELETE": + cleanup_attempts += 1 + return httpx.Response(500 if cleanup_attempts == 1 else 204) + raise AssertionError(f"unexpected request: {request.method} {request.url}") + + launcher = local_runners.DockerEnginePortableLabV1Launcher( + transport_factory=lambda: httpx.MockTransport(handler) + ) + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="Docker container creation returned warnings", + ): + launcher(_docker_launch(tmp_path)) + + assert cleanup_attempts == 2 + assert [request.method for request in requests] == ["GET", "POST", "DELETE", "DELETE"] + + +def test_installed_runners_bind_plan_config_and_component_outputs( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + local_runners, + "validate_camera_compute_job", + lambda _path: object(), + ) + installation = _installation(tmp_path) + source = _source(tmp_path) + plan = _plan(source) + launches: list[local_runners.PortableLabV1DockerLaunch] = [] + + def launcher(launch: local_runners.PortableLabV1DockerLaunch) -> None: + launches.append(launch) + if launch.component == "eomt": + frames_root = launch.local_output_root / "source-frames" + frames_root.mkdir() + for sequence in range(1, plan.source_input.frame_count + 1): + (frames_root / f"frame-{sequence:06d}.png").write_bytes( + f"frame-{sequence}".encode() + ) + document: dict[str, object] = { + "schema_version": "missioncore.recorded-perception-result/v2", + "job_id": plan.source_input.camera_job_id, + "input_sha256": plan.source_input.camera_input_sha256, + "session_id": plan.source_input.source_session_id, + "source_id": plan.source_input.camera_source_id, + "codec_epoch": plan.source_input.codec_epoch, + "frames_processed": plan.source_input.frame_count, + "ground_truth": False, + "identity": { + "configuration": {"pipeline": "recorded-semantic-eomt-fisheye-mask/v1"} + }, + } + else: + effective_source = cast( + dict[str, object], + plan.effective_ddrnet_config["ravnoves"], + ) + document = { + "schema_version": "missioncore.lab-v1-goose-vegetation-run/v1", + "mode": "ravnoves-video", + "candidate": { + "candidate_id": "goose-ddrnet-class-512", + "candidate_key": "ddrnet", + "checkpoint_sha256": ( + "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6" + ), + }, + "source": { + "source_id": effective_source["source_id"], + "input_count": plan.source_input.frame_count, + "mapping_sha256": ( + "88ae319ba5a3877dd3ae0773f693a6a5fdc283934140de9dfaff029108aefd7f" + ), + }, + "provenance": {"config_sha256": plan.effective_ddrnet_config_sha256}, + } + (launch.local_output_root / "result.json").write_bytes(canonical_json(document)) + + installed = local_runners.compose_portable_lab_v1_installed_runners( + installation=installation, + launcher=launcher, + ) + eomt_root = tmp_path / "eomt-result" + installed.eomt(source=source, plan=plan, output_root=eomt_root) + effective_path = tmp_path / "effective-ddrnet-config.json" + effective_path.write_bytes(canonical_json(plan.effective_ddrnet_config)) + ddrnet_root = tmp_path / "ddrnet-result" + installed.ddrnet( + source=source, + plan=plan, + effective_config_path=effective_path, + eomt_result_root=eomt_root, + output_root=ddrnet_root, + ) + + assert [launch.component for launch in launches] == ["eomt", "ddrnet"] + assert (eomt_root / "result.json").is_file() + assert (ddrnet_root / "result.json").is_file() + assert all( + launch.command[-2:] == ("--request", "/run/nodedc/request.json") for launch in launches + ) + assert all(sum(not mount.read_only for mount in launch.mounts) == 1 for launch in launches) + assert not tuple(tmp_path.glob(".lab-v1-*-invocation-*")) + + +def test_ddrnet_runner_rejects_mutated_effective_config_before_launch( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + local_runners, + "validate_camera_compute_job", + lambda _path: object(), + ) + installation = _installation(tmp_path) + source = _source(tmp_path) + plan = _plan(source) + eomt_root = tmp_path / "eomt-result" + eomt_root.mkdir() + frames_root = eomt_root / "source-frames" + frames_root.mkdir() + for sequence in range(1, plan.source_input.frame_count + 1): + (frames_root / f"frame-{sequence:06d}.png").write_bytes(f"frame-{sequence}".encode()) + (eomt_root / "result.json").write_bytes( + canonical_json( + { + "schema_version": "missioncore.recorded-perception-result/v2", + "job_id": plan.source_input.camera_job_id, + "input_sha256": plan.source_input.camera_input_sha256, + "session_id": plan.source_input.source_session_id, + "source_id": plan.source_input.camera_source_id, + "codec_epoch": plan.source_input.codec_epoch, + "frames_processed": plan.source_input.frame_count, + "ground_truth": False, + "identity": { + "configuration": {"pipeline": "recorded-semantic-eomt-fisheye-mask/v1"} + }, + } + ) + ) + effective_path = tmp_path / "effective-ddrnet-config.json" + effective_path.write_text("{}", encoding="utf-8") + called = False + + def launcher(_launch: local_runners.PortableLabV1DockerLaunch) -> None: + nonlocal called + called = True + + runner = local_runners.InstalledPortableLabV1DdrnetRunner( + installation=installation, + launcher=cast(local_runners.PortableLabV1ContainerLauncher, launcher), + ) + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="effective DDRNet config changed", + ): + runner( + source=source, + plan=plan, + effective_config_path=effective_path, + eomt_result_root=eomt_root, + output_root=tmp_path / "ddrnet-result", + ) + assert called is False + + +def test_component_installation_rejects_wrong_worker006_asset_identity() -> None: + assets = list(_assets("ddrnet")) + assets[0] = local_runners.PortableLabV1HostAsset( + asset_id=assets[0].asset_id, + host_path=assets[0].host_path, + container_path=assets[0].container_path, + kind=assets[0].kind, + verification=assets[0].verification, + identity_sha256="0" * 64, + ) + with pytest.raises( + local_runners.PortableLabV1LocalRunnerError, + match="installed asset identity changed", + ): + local_runners.PortableLabV1ComponentInstallation.seal( + component="ddrnet", + image_sha256=DDRNET_IMAGE_SHA256, + entrypoint=("/opt/nodedc/bin/portable-lab-v1-agent",), + command=("run-ddrnet",), + assets=tuple(assets), + timeout_seconds=3600.0, + memory_bytes=16 * 1024**3, + nano_cpus=4_000_000_000, + )