diff --git a/src/k1link/observatory/portable_lab_v1_worker_service.py b/src/k1link/observatory/portable_lab_v1_worker_service.py new file mode 100644 index 0000000..1b0e333 --- /dev/null +++ b/src/k1link/observatory/portable_lab_v1_worker_service.py @@ -0,0 +1,983 @@ +"""Install-time composition for the sealed portable LAB V1 Worker adapter. + +This module is deliberately narrower than the Worker entrypoint. It loads one +bounded, canonical installation receipt, reconstructs the already reviewed +local EoMT and DDRNet runner identities, and returns one exact builder +registration. It does not read executable instructions from a queued job and +does not register a blocked runtime candidate. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import stat +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Final, cast + +from k1link.observatory.portable_lab_v1_executor import PortableLabV1ReleaseCandidate +from k1link.observatory.portable_lab_v1_local_runners import ( + PORTABLE_LAB_V1_COMPONENT_INSTALLATION_SCHEMA, + PORTABLE_LAB_V1_RUNNER_INSTALLATION_SCHEMA, + DockerEnginePortableLabV1Launcher, + PortableLabV1AssetKind, + PortableLabV1AssetVerification, + PortableLabV1Component, + PortableLabV1ComponentInstallation, + PortableLabV1ContainerLauncher, + PortableLabV1HostAsset, + PortableLabV1LocalRunnerError, + PortableLabV1WorkRootBinding, + compose_portable_lab_v1_installed_runners, +) +from k1link.observatory.portable_lab_v1_local_runners import ( + PortableLabV1RunnerInstallation as PortableLabV1ComponentRunnerInstallation, +) +from k1link.observatory.portable_lab_v1_worker import ( + PortableLabV1DdrnetRunner, + PortableLabV1EomtRunner, + compose_lab_v1_portable_executor_adapter, +) +from k1link.observatory.portable_lab_v1_worker import ( + PortableLabV1RunnerInstallation as PortableLabV1ReleaseInstallation, +) +from k1link.observatory.portable_result_contract import ( + OBSERVATION_ONLY_AUTHORITY, + canonical_json, +) +from k1link.observatory.portable_run_definitions import PortableRunDefinition +from k1link.observatory.portable_worker_runtime import ( + PortableWorkerRuntimeAdmission, + PortableWorkerRuntimeCandidate, +) +from k1link.observatory.worker_agent import ObservatoryWorkerExecutorRegistration +from k1link.observatory.worker_service import ( + ObservatoryWorkerExecutorBuildContext, + ObservatoryWorkerExecutorBuilderRegistration, +) + +PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_SCHEMA: Final = ( + "missioncore.observatory-portable-lab-v1-worker-installation-ready-receipt/v1" +) +PORTABLE_LAB_V1_COMPONENT_IMAGE_BUILD_SEAL_SCHEMA: Final = ( + "missioncore.observatory-portable-lab-v1-component-image-build-seal/v1" +) +PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_ASSET_ID: Final = ( + "lab-v1-worker-installation-receipt" +) +PORTABLE_LAB_V1_SETUP_ID: Final = "lab-v1-eomt-ddrnet-portable-v1" +PORTABLE_LAB_V1_ADAPTER_ID: Final = "lab-v1-eomt-ddrnet-worker006-v2" + +_WORKER_ID: Final = "worker-006" +_DEFINITION_ID: Final = "lab-v1-eomt-ddrnet-portable" +_DEFINITION_VERSION: Final = 2 +_COMPONENT_BASE_IMAGE_SHA256S: Final[dict[PortableLabV1Component, str]] = { + "eomt": "58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794", + "ddrnet": "591cb382c099eeb05e7ec16e2371e0b2da54d2bb5c49ec0f4ac88dbf72b0f0cd", +} +_MAX_RECEIPT_BYTES: Final = 256 * 1024 +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_SOURCE_REVISION = re.compile(r"^[a-f0-9]{40}$") + + +class PortableLabV1WorkerCompositionError(RuntimeError): + """The installed LAB V1 identity cannot safely enter Worker 006.""" + + +@dataclass(frozen=True, slots=True) +class PortableLabV1ComponentImageBuildSeal: + """Canonical offline base/source-to-derived-image promotion identity.""" + + component: PortableLabV1Component + base_image_sha256: str + derived_image_sha256: str + dockerfile_sha256: str + shared_adapter_sha256: str + component_adapter_sha256: str + network: str + seal_sha256: str + + def __post_init__(self) -> None: + if self.component not in ("eomt", "ddrnet"): + raise PortableLabV1WorkerCompositionError( + "LAB V1 component image build seal is misbound" + ) + for value, label in ( + (self.base_image_sha256, "LAB V1 component base image SHA-256"), + (self.derived_image_sha256, "LAB V1 derived component image SHA-256"), + (self.dockerfile_sha256, "LAB V1 component Dockerfile SHA-256"), + (self.shared_adapter_sha256, "LAB V1 shared adapter SHA-256"), + (self.component_adapter_sha256, "LAB V1 component adapter SHA-256"), + (self.seal_sha256, "LAB V1 component image build seal SHA-256"), + ): + _digest(value, label) + if ( + self.base_image_sha256 + != _COMPONENT_BASE_IMAGE_SHA256S[self.component] + or self.network != "none" + or self.seal_sha256 + != hashlib.sha256(canonical_json(self.identity_document())).hexdigest() + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 component image build provenance changed" + ) + + @classmethod + def seal( + cls, + *, + component: PortableLabV1Component, + base_image_sha256: str, + derived_image_sha256: str, + dockerfile_sha256: str, + shared_adapter_sha256: str, + component_adapter_sha256: str, + ) -> PortableLabV1ComponentImageBuildSeal: + identity = _component_image_build_identity( + component=component, + base_image_sha256=base_image_sha256, + derived_image_sha256=derived_image_sha256, + dockerfile_sha256=dockerfile_sha256, + shared_adapter_sha256=shared_adapter_sha256, + component_adapter_sha256=component_adapter_sha256, + ) + return cls( + component=component, + base_image_sha256=base_image_sha256, + derived_image_sha256=derived_image_sha256, + dockerfile_sha256=dockerfile_sha256, + shared_adapter_sha256=shared_adapter_sha256, + component_adapter_sha256=component_adapter_sha256, + network="none", + seal_sha256=hashlib.sha256(canonical_json(identity)).hexdigest(), + ) + + def identity_document(self) -> dict[str, object]: + return _component_image_build_identity( + component=self.component, + base_image_sha256=self.base_image_sha256, + derived_image_sha256=self.derived_image_sha256, + dockerfile_sha256=self.dockerfile_sha256, + shared_adapter_sha256=self.shared_adapter_sha256, + component_adapter_sha256=self.component_adapter_sha256, + ) + + +@dataclass(frozen=True, slots=True) +class PortableLabV1WorkerInstallationReceipt: + """Canonical receipt containing the complete local-runner identity.""" + + path: Path + file_byte_length: int + file_sha256: str + source_revision: str + release_candidate_sha256: str + runner_installation: PortableLabV1ComponentRunnerInstallation + eomt_image_build: PortableLabV1ComponentImageBuildSeal + ddrnet_image_build: PortableLabV1ComponentImageBuildSeal + receipt_sha256: str + + def __post_init__(self) -> None: + if not self.path.is_absolute(): + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt path is not absolute" + ) + if not 1 <= self.file_byte_length <= _MAX_RECEIPT_BYTES: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt size is invalid" + ) + _digest(self.file_sha256, "LAB V1 installation receipt file SHA-256") + if _SOURCE_REVISION.fullmatch(self.source_revision) is None: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt source revision is invalid" + ) + _digest(self.release_candidate_sha256, "LAB V1 release candidate SHA-256") + _digest(self.receipt_sha256, "LAB V1 installation receipt SHA-256") + if self.runner_installation.release_candidate_sha256 != ( + self.release_candidate_sha256 + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 runner receipt belongs to another release candidate" + ) + if ( + self.eomt_image_build.component != "eomt" + or self.ddrnet_image_build.component != "ddrnet" + or self.eomt_image_build.derived_image_sha256 + != self.runner_installation.eomt.image_sha256 + or self.ddrnet_image_build.derived_image_sha256 + != self.runner_installation.ddrnet.image_sha256 + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 component image builds differ from the runner installation" + ) + self.runner_installation.verify_unchanged() + if self.receipt_sha256 != _receipt_identity_sha256( + source_revision=self.source_revision, + release_candidate_sha256=self.release_candidate_sha256, + runner_installation=self.runner_installation, + eomt_image_build=self.eomt_image_build, + ddrnet_image_build=self.ddrnet_image_build, + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt identity changed" + ) + + +@dataclass(frozen=True, slots=True) +class _PortableLabV1WorkerExecutorBuilder: + receipt: PortableLabV1WorkerInstallationReceipt + definition: PortableRunDefinition + candidate: PortableWorkerRuntimeCandidate + admission: PortableWorkerRuntimeAdmission + release_installation: PortableLabV1ReleaseInstallation + eomt_runner: PortableLabV1EomtRunner + ddrnet_runner: PortableLabV1DdrnetRunner + created_at_utc: Callable[[], str] + + def __call__( + self, + context: ObservatoryWorkerExecutorBuildContext, + ) -> ObservatoryWorkerExecutorRegistration: + if ( + context.definition.setup_id != self.definition.setup_id + or context.definition.definition_sha256 + != self.definition.definition_sha256 + or context.candidate.setup_id != self.candidate.setup_id + or context.candidate.candidate_sha256 != self.candidate.candidate_sha256 + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 builder context differs from its sealed installation" + ) + adapter = compose_lab_v1_portable_executor_adapter( + candidate=context.candidate, + definition=context.definition, + admission=self.admission, + source_transport=context.source_transport, + result_transport=context.result_transport, + installation=self.release_installation, + source_output_parent=context.work_root / "lab-v1-portable" / "source-output", + created_at_utc=self.created_at_utc, + eomt_runner=self.eomt_runner, + ddrnet_runner=self.ddrnet_runner, + ) + return ObservatoryWorkerExecutorRegistration( + identity=context.candidate.executor_identity(), + adapter=adapter, + ) + + +def load_portable_lab_v1_worker_installation_receipt( + path: Path, +) -> PortableLabV1WorkerInstallationReceipt: + """Load one canonical regular receipt without following a filesystem link.""" + + receipt_path = path.expanduser().absolute() + if not path.is_absolute(): + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt path is not absolute" + ) + payload = _read_bounded_regular_file(receipt_path) + try: + document = json.loads( + payload.decode("utf-8"), + object_pairs_hook=_unique_object, + parse_constant=_reject_json_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt is not canonical JSON" + ) from exc + row = _object(document, "LAB V1 installation receipt") + if canonical_json(row) != payload: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt is not canonical JSON" + ) + _exact_keys( + row, + { + "schema_version", + "receipt_state", + "worker_id", + "source_revision", + "release_candidate_sha256", + "runner_installation", + "component_installations", + "component_image_build_seals", + "fixture_smoke", + "blockers", + "authority", + "receipt_sha256", + }, + "LAB V1 installation receipt", + ) + blockers = _array(row["blockers"], "LAB V1 installation blockers") + if ( + row["schema_version"] + != PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_SCHEMA + or row["receipt_state"] != "installed-ready" + or row["worker_id"] != _WORKER_ID + or row["fixture_smoke"] != "passed" + or blockers + or row["authority"] != OBSERVATION_ONLY_AUTHORITY + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt is not an accepted ready installation" + ) + try: + components = _object( + row["component_installations"], + "LAB V1 component installations", + ) + _exact_keys( + components, + {"eomt", "ddrnet"}, + "LAB V1 component installations", + ) + eomt = _component_installation( + components["eomt"], + expected_component="eomt", + ) + ddrnet = _component_installation( + components["ddrnet"], + expected_component="ddrnet", + ) + runner = _runner_installation( + row["runner_installation"], + eomt=eomt, + ddrnet=ddrnet, + ) + except PortableLabV1LocalRunnerError as exc: + raise PortableLabV1WorkerCompositionError( + "LAB V1 local runner installation identity changed" + ) from exc + image_builds = _object( + row["component_image_build_seals"], + "LAB V1 component image build seals", + ) + _exact_keys( + image_builds, + {"eomt", "ddrnet"}, + "LAB V1 component image build seals", + ) + eomt_image_build = _component_image_build_seal( + image_builds["eomt"], + expected_component="eomt", + ) + ddrnet_image_build = _component_image_build_seal( + image_builds["ddrnet"], + expected_component="ddrnet", + ) + return PortableLabV1WorkerInstallationReceipt( + path=receipt_path, + file_byte_length=len(payload), + file_sha256=hashlib.sha256(payload).hexdigest(), + source_revision=_string(row["source_revision"], "LAB V1 source revision"), + release_candidate_sha256=_string( + row["release_candidate_sha256"], + "LAB V1 release candidate SHA-256", + ), + runner_installation=runner, + eomt_image_build=eomt_image_build, + ddrnet_image_build=ddrnet_image_build, + receipt_sha256=_string( + row["receipt_sha256"], + "LAB V1 installation receipt SHA-256", + ), + ) + + +def compose_installed_lab_v1_executor_builder( + *, + receipt: PortableLabV1WorkerInstallationReceipt, + definition: PortableRunDefinition, + candidate: PortableWorkerRuntimeCandidate, + admission: PortableWorkerRuntimeAdmission, + release_installation: PortableLabV1ReleaseInstallation, + launcher: PortableLabV1ContainerLauncher | None = None, + created_at_utc: Callable[[], str] | None = None, +) -> ObservatoryWorkerExecutorBuilderRegistration: + """Return the LAB V1 builder only after every installed identity agrees.""" + + installed_receipt = load_portable_lab_v1_worker_installation_receipt( + receipt.path + ) + if installed_receipt != receipt: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt changed after it was loaded" + ) + receipt = installed_receipt + if ( + definition.setup_id != PORTABLE_LAB_V1_SETUP_ID + or candidate.setup_id != PORTABLE_LAB_V1_SETUP_ID + or candidate.adapter_id != PORTABLE_LAB_V1_ADAPTER_ID + or receipt.runner_installation.definition_sha256 + != definition.definition_sha256 + or receipt.release_candidate_sha256 + != release_installation.release.candidate_sha256 + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 definition, runtime, release and installation receipt disagree" + ) + candidate.bind_definition(definition) + if ( + not candidate.ready + or not admission.ready + or admission.candidate_sha256 != candidate.candidate_sha256 + or tuple(item.asset_id for item in admission.assets) + != tuple(item.asset_id for item in candidate.reusable_assets) + or any(item.state != "matched" for item in admission.assets) + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 runtime candidate has no complete local asset admission" + ) + receipt_requirement = next( + ( + requirement + for requirement in candidate.reusable_assets + if requirement.asset_id + == PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_ASSET_ID + ), + None, + ) + if ( + receipt_requirement is None + or receipt_requirement.kind != "local-file" + or receipt_requirement.sha256 != receipt.file_sha256 + or receipt_requirement.byte_length != receipt.file_byte_length + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 runtime candidate does not seal its installation receipt" + ) + release = release_installation.release + release.bind_definition(definition) + _verify_release_covers_runner_installation( + release=release, + receipt=receipt, + ) + seal = release.seal(release_installation.inspection) + executor = candidate.executor + if ( + executor is None + or executor.release_id != seal.release_id + or executor.release_sha256 != seal.release_sha256 + or executor.image_sha256 != seal.executor_image_sha256 + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 runtime executor differs from its admitted release" + ) + receipt.runner_installation.verify_unchanged() + installed = compose_portable_lab_v1_installed_runners( + installation=receipt.runner_installation, + launcher=launcher or DockerEnginePortableLabV1Launcher(), + ) + clock = _utc_now if created_at_utc is None else created_at_utc + if not callable(clock): + raise PortableLabV1WorkerCompositionError("LAB V1 result clock is not callable") + return ObservatoryWorkerExecutorBuilderRegistration( + setup_id=PORTABLE_LAB_V1_SETUP_ID, + builder=_PortableLabV1WorkerExecutorBuilder( + receipt=receipt, + definition=definition, + candidate=candidate, + admission=admission, + release_installation=release_installation, + eomt_runner=installed.eomt, + ddrnet_runner=installed.ddrnet, + created_at_utc=clock, + ), + ) + + +def portable_lab_v1_worker_receipt_document( + *, + source_revision: str, + release_candidate_sha256: str, + runner_installation: PortableLabV1ComponentRunnerInstallation, + eomt_image_build: PortableLabV1ComponentImageBuildSeal, + ddrnet_image_build: PortableLabV1ComponentImageBuildSeal, +) -> dict[str, object]: + """Return the canonical serializable receipt document for an installer.""" + + identity = _receipt_identity_document( + source_revision=source_revision, + release_candidate_sha256=release_candidate_sha256, + runner_installation=runner_installation, + eomt_image_build=eomt_image_build, + ddrnet_image_build=ddrnet_image_build, + ) + return {**identity, "receipt_sha256": hashlib.sha256(canonical_json(identity)).hexdigest()} + + +def _receipt_identity_sha256( + *, + source_revision: str, + release_candidate_sha256: str, + runner_installation: PortableLabV1ComponentRunnerInstallation, + eomt_image_build: PortableLabV1ComponentImageBuildSeal, + ddrnet_image_build: PortableLabV1ComponentImageBuildSeal, +) -> str: + return hashlib.sha256( + canonical_json( + _receipt_identity_document( + source_revision=source_revision, + release_candidate_sha256=release_candidate_sha256, + runner_installation=runner_installation, + eomt_image_build=eomt_image_build, + ddrnet_image_build=ddrnet_image_build, + ) + ) + ).hexdigest() + + +def _receipt_identity_document( + *, + source_revision: str, + release_candidate_sha256: str, + runner_installation: PortableLabV1ComponentRunnerInstallation, + eomt_image_build: PortableLabV1ComponentImageBuildSeal, + ddrnet_image_build: PortableLabV1ComponentImageBuildSeal, +) -> dict[str, object]: + return { + "schema_version": PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_SCHEMA, + "receipt_state": "installed-ready", + "worker_id": _WORKER_ID, + "source_revision": source_revision, + "release_candidate_sha256": release_candidate_sha256, + "runner_installation": { + **runner_installation.identity_document(), + "receipt_sha256": runner_installation.receipt_sha256, + }, + "component_installations": { + "eomt": _component_installation_document(runner_installation.eomt), + "ddrnet": _component_installation_document(runner_installation.ddrnet), + }, + "component_image_build_seals": { + "eomt": _component_image_build_document(eomt_image_build), + "ddrnet": _component_image_build_document(ddrnet_image_build), + }, + "fixture_smoke": "passed", + "blockers": [], + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + + +def _component_installation_document( + installation: PortableLabV1ComponentInstallation, +) -> dict[str, object]: + return { + **installation.identity_document(), + "installation_sha256": installation.installation_sha256, + } + + +def _component_image_build_identity( + *, + component: PortableLabV1Component, + base_image_sha256: str, + derived_image_sha256: str, + dockerfile_sha256: str, + shared_adapter_sha256: str, + component_adapter_sha256: str, +) -> dict[str, object]: + return { + "schema_version": PORTABLE_LAB_V1_COMPONENT_IMAGE_BUILD_SEAL_SCHEMA, + "component": component, + "base_image_sha256": base_image_sha256, + "derived_image_sha256": derived_image_sha256, + "dockerfile_sha256": dockerfile_sha256, + "shared_adapter_sha256": shared_adapter_sha256, + "component_adapter_sha256": component_adapter_sha256, + "network": "none", + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + + +def _component_image_build_document( + seal: PortableLabV1ComponentImageBuildSeal, +) -> dict[str, object]: + return {**seal.identity_document(), "seal_sha256": seal.seal_sha256} + + +def _component_image_build_seal( + value: object, + *, + expected_component: PortableLabV1Component, +) -> PortableLabV1ComponentImageBuildSeal: + row = _object(value, f"LAB V1 {expected_component} component image build seal") + _exact_keys( + row, + { + "schema_version", + "component", + "base_image_sha256", + "derived_image_sha256", + "dockerfile_sha256", + "shared_adapter_sha256", + "component_adapter_sha256", + "network", + "authority", + "seal_sha256", + }, + f"LAB V1 {expected_component} component image build seal", + ) + if ( + row["schema_version"] + != PORTABLE_LAB_V1_COMPONENT_IMAGE_BUILD_SEAL_SCHEMA + or row["component"] != expected_component + or row["authority"] != OBSERVATION_ONLY_AUTHORITY + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 component image build seal contract changed" + ) + return PortableLabV1ComponentImageBuildSeal( + component=expected_component, + base_image_sha256=_string( + row["base_image_sha256"], + "LAB V1 component base image SHA-256", + ), + derived_image_sha256=_string( + row["derived_image_sha256"], + "LAB V1 derived component image SHA-256", + ), + dockerfile_sha256=_string( + row["dockerfile_sha256"], + "LAB V1 component Dockerfile SHA-256", + ), + shared_adapter_sha256=_string( + row["shared_adapter_sha256"], + "LAB V1 shared adapter SHA-256", + ), + component_adapter_sha256=_string( + row["component_adapter_sha256"], + "LAB V1 component adapter SHA-256", + ), + network=_string(row["network"], "LAB V1 component build network"), + seal_sha256=_string( + row["seal_sha256"], + "LAB V1 component image build seal SHA-256", + ), + ) + + +def _verify_release_covers_runner_installation( + *, + release: PortableLabV1ReleaseCandidate, + receipt: PortableLabV1WorkerInstallationReceipt, +) -> None: + release_payload_sha256s = {asset.sha256 for asset in release.assets} + release_image_sha256s = { + asset.sha256 for asset in release.assets if asset.kind == "container-image" + } + components = ( + receipt.runner_installation.eomt, + receipt.runner_installation.ddrnet, + ) + image_builds = (receipt.eomt_image_build, receipt.ddrnet_image_build) + installed_asset_sha256s = { + asset.identity_sha256 + for component in components + for asset in component.assets + } + installed_image_sha256s = { + digest + for build in image_builds + for digest in (build.base_image_sha256, build.derived_image_sha256) + } + installed_build_sha256s = { + digest + for build in image_builds + for digest in ( + build.dockerfile_sha256, + build.shared_adapter_sha256, + build.component_adapter_sha256, + build.seal_sha256, + ) + } + if ( + not installed_asset_sha256s.issubset(release_payload_sha256s) + or not installed_build_sha256s.issubset(release_payload_sha256s) + or not installed_image_sha256s.issubset(release_image_sha256s) + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 release does not cover its installed component identities" + ) + + +def _component_installation( + value: object, + *, + expected_component: PortableLabV1Component, +) -> PortableLabV1ComponentInstallation: + row = _object(value, f"LAB V1 {expected_component} installation") + _exact_keys( + row, + { + "schema_version", + "component", + "image_sha256", + "entrypoint", + "command", + "assets", + "timeout_seconds", + "memory_bytes", + "nano_cpus", + "authority", + "installation_sha256", + }, + f"LAB V1 {expected_component} installation", + ) + if row["component"] != expected_component: + raise PortableLabV1WorkerCompositionError( + "LAB V1 component installation identities are misbound" + ) + if ( + row["schema_version"] != PORTABLE_LAB_V1_COMPONENT_INSTALLATION_SCHEMA + or row["authority"] != OBSERVATION_ONLY_AUTHORITY + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 component installation contract changed" + ) + entrypoint = _string_array(row["entrypoint"], "LAB V1 component entrypoint") + command = _string_array(row["command"], "LAB V1 component command") + assets = tuple( + _host_asset(item) + for item in _array(row["assets"], "LAB V1 component assets") + ) + return PortableLabV1ComponentInstallation( + component=expected_component, + image_sha256=_string(row["image_sha256"], "LAB V1 component image SHA-256"), + entrypoint=entrypoint, + command=command, + assets=assets, + timeout_seconds=_number(row["timeout_seconds"], "LAB V1 component timeout"), + memory_bytes=_integer(row["memory_bytes"], "LAB V1 component memory"), + nano_cpus=_integer(row["nano_cpus"], "LAB V1 component CPU limit"), + installation_sha256=_string( + row["installation_sha256"], + "LAB V1 component installation SHA-256", + ), + ) + + +def _host_asset(value: object) -> PortableLabV1HostAsset: + row = _object(value, "LAB V1 host asset") + _exact_keys( + row, + { + "asset_id", + "host_path", + "container_path", + "kind", + "verification", + "identity_sha256", + "byte_length", + }, + "LAB V1 host asset", + ) + byte_length = row["byte_length"] + if byte_length is not None: + byte_length = _integer(byte_length, "LAB V1 host asset byte length") + return PortableLabV1HostAsset( + asset_id=_string(row["asset_id"], "LAB V1 host asset id"), + host_path=_string(row["host_path"], "LAB V1 host asset path"), + container_path=_string(row["container_path"], "LAB V1 asset target"), + kind=cast(PortableLabV1AssetKind, _string(row["kind"], "LAB V1 asset kind")), + verification=cast( + PortableLabV1AssetVerification, + _string(row["verification"], "LAB V1 asset verification"), + ), + identity_sha256=_string( + row["identity_sha256"], + "LAB V1 asset identity SHA-256", + ), + byte_length=byte_length, + ) + + +def _runner_installation( + value: object, + *, + eomt: PortableLabV1ComponentInstallation, + ddrnet: PortableLabV1ComponentInstallation, +) -> PortableLabV1ComponentRunnerInstallation: + row = _object(value, "LAB V1 runner installation") + _exact_keys( + row, + { + "schema_version", + "setup_id", + "definition_id", + "definition_version", + "definition_sha256", + "release_candidate_sha256", + "work_root", + "components", + "authority", + "receipt_sha256", + }, + "LAB V1 runner installation", + ) + components = _object(row["components"], "LAB V1 runner components") + _exact_keys(components, {"eomt", "ddrnet"}, "LAB V1 runner components") + if ( + row["schema_version"] != PORTABLE_LAB_V1_RUNNER_INSTALLATION_SCHEMA + or row["setup_id"] != PORTABLE_LAB_V1_SETUP_ID + or row["definition_id"] != _DEFINITION_ID + or row["definition_version"] != _DEFINITION_VERSION + or components["eomt"] != eomt.installation_sha256 + or components["ddrnet"] != ddrnet.installation_sha256 + or row["authority"] != OBSERVATION_ONLY_AUTHORITY + ): + raise PortableLabV1WorkerCompositionError( + "LAB V1 runner installation identity changed" + ) + work = _object(row["work_root"], "LAB V1 runner work root") + _exact_keys(work, {"controller_root", "engine_host_root"}, "LAB V1 runner work root") + return PortableLabV1ComponentRunnerInstallation( + definition_sha256=_string( + row["definition_sha256"], + "LAB V1 runner definition SHA-256", + ), + release_candidate_sha256=_string( + row["release_candidate_sha256"], + "LAB V1 runner release candidate SHA-256", + ), + work_root=PortableLabV1WorkRootBinding( + controller_root=Path( + _string(work["controller_root"], "LAB V1 controller work root") + ), + engine_host_root=_string( + work["engine_host_root"], + "LAB V1 engine-host work root", + ), + ), + eomt=eomt, + ddrnet=ddrnet, + receipt_sha256=_string( + row["receipt_sha256"], + "LAB V1 runner installation SHA-256", + ), + ) + + +def _read_bounded_regular_file(path: Path) -> bytes: + descriptor: int | None = None + try: + descriptor = os.open( + path, + os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0), + ) + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt must be a regular file" + ) + if not 1 <= metadata.st_size <= _MAX_RECEIPT_BYTES: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt size is invalid" + ) + with os.fdopen(descriptor, "rb") as stream: + descriptor = None + payload = stream.read(_MAX_RECEIPT_BYTES + 1) + except PortableLabV1WorkerCompositionError: + raise + except OSError as exc: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt is unavailable" + ) from exc + finally: + if descriptor is not None: + os.close(descriptor) + if len(payload) > _MAX_RECEIPT_BYTES: + raise PortableLabV1WorkerCompositionError( + "LAB V1 installation receipt size is invalid" + ) + return payload + + +def _unique_object(pairs: Sequence[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate JSON key") + result[key] = value + return result + + +def _reject_json_constant(value: str) -> object: + raise ValueError(f"invalid JSON constant: {value}") + + +def _exact_keys(row: Mapping[str, object], expected: set[str], label: str) -> None: + if set(row) != expected: + raise PortableLabV1WorkerCompositionError(f"{label} fields are 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 PortableLabV1WorkerCompositionError(f"{label} must be an object") + return cast(dict[str, object], value) + + +def _array(value: object, label: str) -> list[object]: + if not isinstance(value, list): + raise PortableLabV1WorkerCompositionError(f"{label} must be an array") + return value + + +def _string(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise PortableLabV1WorkerCompositionError(f"{label} must be a non-empty string") + return value + + +def _string_array(value: object, label: str) -> tuple[str, ...]: + rows = _array(value, label) + if not rows: + raise PortableLabV1WorkerCompositionError(f"{label} cannot be empty") + return tuple(_string(item, label) for item in rows) + + +def _integer(value: object, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise PortableLabV1WorkerCompositionError(f"{label} must be an integer") + return value + + +def _number(value: object, label: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise PortableLabV1WorkerCompositionError(f"{label} must be numeric") + result = float(value) + if not math.isfinite(result): + raise PortableLabV1WorkerCompositionError(f"{label} must be finite") + return result + + +def _digest(value: str, label: str) -> None: + if _SHA256.fullmatch(value) is None: + raise PortableLabV1WorkerCompositionError(f"{label} is invalid") + + +def _utc_now() -> str: + return datetime.now(tz=UTC).isoformat().replace("+00:00", "Z") + + +__all__ = [ + "PORTABLE_LAB_V1_COMPONENT_IMAGE_BUILD_SEAL_SCHEMA", + "PORTABLE_LAB_V1_ADAPTER_ID", + "PORTABLE_LAB_V1_SETUP_ID", + "PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_ASSET_ID", + "PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_SCHEMA", + "PortableLabV1ComponentImageBuildSeal", + "PortableLabV1WorkerCompositionError", + "PortableLabV1WorkerInstallationReceipt", + "compose_installed_lab_v1_executor_builder", + "load_portable_lab_v1_worker_installation_receipt", + "portable_lab_v1_worker_receipt_document", +] diff --git a/tests/test_observatory_portable_lab_v1_worker_service.py b/tests/test_observatory_portable_lab_v1_worker_service.py new file mode 100644 index 0000000..a64cea0 --- /dev/null +++ b/tests/test_observatory_portable_lab_v1_worker_service.py @@ -0,0 +1,625 @@ +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, replace +from pathlib import Path +from typing import cast + +import pytest + +from k1link.observatory import portable_lab_v1_local_runners as local_runners +from k1link.observatory import portable_lab_v1_worker_service as service_module +from k1link.observatory.portable_lab_v1_executor import ( + PortableLabV1ExecutorSeal, + PortableLabV1ReleaseInspection, +) +from k1link.observatory.portable_lab_v1_worker import ( + PORTABLE_LAB_V1_RUNTIME_PHASES, +) +from k1link.observatory.portable_lab_v1_worker import ( + PortableLabV1RunnerInstallation as PortableLabV1ReleaseInstallation, +) +from k1link.observatory.portable_result_contract import ( + OBSERVATION_ONLY_AUTHORITY, + canonical_json, +) +from k1link.observatory.portable_run_definitions import ( + PortableExecutorAvailability, + PortableRunDefinition, + PortableRunDefinitionRegistry, + canonical_sha256, +) +from k1link.observatory.portable_worker_runtime import ( + PORTABLE_WORKER_RUNTIME_CANDIDATE_SCHEMA, + PortableWorkerAssetRequirement, + PortableWorkerAssetVerification, + PortableWorkerExecutorSeal, + PortableWorkerResultPublisher, + PortableWorkerRuntimeAdmission, + PortableWorkerRuntimeCandidate, + PortableWorkerRuntimePhase, + PortableWorkerSourceMaterializer, +) +from k1link.observatory.worker_service import ObservatoryWorkerExecutorBuildContext + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +DEFINITIONS_PATH = REPOSITORY_ROOT / "config/observatory-portable-run-definitions.json" +RELEASE_ID = "lab-v1-eomt-ddrnet-v1" +RELEASE_SHA256 = "7" * 64 +IMAGE_SHA256 = "8" * 64 +RELEASE_CANDIDATE_SHA256 = "9" * 64 +SOURCE_REVISION = "a" * 40 +BASE_IMAGE_SHA256S = { + "eomt": "58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794", + "ddrnet": "591cb382c099eeb05e7ec16e2371e0b2da54d2bb5c49ec0f4ac88dbf72b0f0cd", +} + + +@dataclass(frozen=True) +class _ReleaseAsset: + kind: str + sha256: str + + +@dataclass(frozen=True) +class _Release: + definition_contract_sha256: str + result_contract_sha256: str + candidate_sha256: str = RELEASE_CANDIDATE_SHA256 + executor_image_sha256: str = IMAGE_SHA256 + assets: tuple[_ReleaseAsset, ...] = () + declared_blockers: tuple[str, ...] = () + + def bind_definition(self, definition: PortableRunDefinition) -> None: + if ( + definition.setup_id != service_module.PORTABLE_LAB_V1_SETUP_ID + or definition.definition_id != "lab-v1-eomt-ddrnet-portable" + or definition.version != 2 + or definition.executable_contract_sha256 + != self.definition_contract_sha256 + or definition.result_contract.contract_sha256 + != self.result_contract_sha256 + ): + raise ValueError("release belongs to another definition") + + def seal( + self, + inspection: PortableLabV1ReleaseInspection, + ) -> PortableLabV1ExecutorSeal: + if ( + inspection.candidate_sha256 != self.candidate_sha256 + or not inspection.ready + or inspection.blockers + ): + raise ValueError("release inspection is not ready") + return PortableLabV1ExecutorSeal( + release_id=RELEASE_ID, + candidate_sha256=self.candidate_sha256, + definition_contract_sha256=self.definition_contract_sha256, + executor_image_sha256=self.executor_image_sha256, + release_sha256=RELEASE_SHA256, + ) + + +@dataclass(frozen=True) +class _ReleaseInstallation: + release: _Release + inspection: PortableLabV1ReleaseInspection + + +class _Launcher: + def __call__(self, launch: local_runners.PortableLabV1DockerLaunch) -> None: + del launch + raise AssertionError("composition must not launch a component") + + +def _ready_definition() -> PortableRunDefinition: + base = PortableRunDefinitionRegistry.from_file(DEFINITIONS_PATH).resolve( + service_module.PORTABLE_LAB_V1_SETUP_ID, + "3692d41cec3949f348a36eb60a501fb2cd483fed1645679b0ec58061a2fc6dc2", + ) + executor = PortableExecutorAvailability( + contour_id=base.executor.contour_id, + state="ready", + release_id=RELEASE_ID, + release_sha256=RELEASE_SHA256, + image_sha256=IMAGE_SHA256, + reason_code=None, + reason=None, + ) + identity = base.identity_document() + identity["executor"] = executor.identity_document() + return replace( + base, + executor=executor, + definition_sha256=canonical_sha256(identity), + ) + + +def _runtime_candidate( + definition: PortableRunDefinition, + receipt: service_module.PortableLabV1WorkerInstallationReceipt, +) -> PortableWorkerRuntimeCandidate: + executor = PortableWorkerExecutorSeal( + release_id=RELEASE_ID, + release_sha256=RELEASE_SHA256, + image_sha256=IMAGE_SHA256, + ) + phases = tuple( + PortableWorkerRuntimePhase(phase_id=phase_id, state="implemented") + for phase_id in PORTABLE_LAB_V1_RUNTIME_PHASES + ) + receipt_requirement = PortableWorkerAssetRequirement( + asset_id=service_module.PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_ASSET_ID, + kind="local-file", + sha256=receipt.file_sha256, + byte_length=receipt.file_byte_length, + component_id=None, + model_release_id=None, + model_artifact_role=None, + ) + identity = { + "schema_version": PORTABLE_WORKER_RUNTIME_CANDIDATE_SCHEMA, + "adapter_id": service_module.PORTABLE_LAB_V1_ADAPTER_ID, + "setup_id": definition.setup_id, + "definition_id": definition.definition_id, + "definition_version": definition.version, + "definition_sha256": definition.definition_sha256, + "source_adapter_sha256": definition.source_adapter.contract_sha256, + "model_manifest_sha256": definition.model_manifest_sha256, + "resource_profile_sha256": definition.resource_profile.profile_sha256, + "result_contract_sha256": definition.result_contract.contract_sha256, + "state": "ready", + "executor": executor.as_dict(), + "reusable_assets": [receipt_requirement.as_dict()], + "phases": [phase.as_dict() for phase in phases], + "blockers": [], + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + return PortableWorkerRuntimeCandidate( + adapter_id=service_module.PORTABLE_LAB_V1_ADAPTER_ID, + setup_id=definition.setup_id, + definition_id=definition.definition_id, + definition_version=definition.version, + definition_sha256=definition.definition_sha256, + source_adapter_sha256=definition.source_adapter.contract_sha256, + model_manifest_sha256=definition.model_manifest_sha256, + resource_profile_sha256=definition.resource_profile.profile_sha256, + result_contract_sha256=definition.result_contract.contract_sha256, + state="ready", + executor=executor, + reusable_assets=(receipt_requirement,), + phases=phases, + blockers=(), + candidate_sha256=canonical_sha256(identity), + ) + + +def _assets( + component: local_runners.PortableLabV1Component, +) -> tuple[local_runners.PortableLabV1HostAsset, ...]: + return tuple( + local_runners.PortableLabV1HostAsset( + asset_id=asset_id, + host_path=f"D:\\NDC_MISSIONCORE\\assets\\{asset_id}", + container_path=container_path, + kind="tree" if component == "eomt" else "file", + verification="identity-sha256" if component == "eomt" else "sha256", + identity_sha256=local_runners._EXPECTED_FIXED_ASSET_IDENTITIES.get( + asset_id, + hashlib.sha256(asset_id.encode()).hexdigest(), + ), + byte_length=local_runners._EXPECTED_ASSET_LENGTHS.get(asset_id), + ) + for asset_id, container_path in sorted( + local_runners._EXPECTED_ASSET_TARGETS[component].items() + ) + ) + + +def _component( + component: local_runners.PortableLabV1Component, +) -> local_runners.PortableLabV1ComponentInstallation: + return local_runners.PortableLabV1ComponentInstallation.seal( + component=component, + image_sha256="b" * 64 if component == "eomt" else "c" * 64, + entrypoint=local_runners._EXPECTED_ENTRYPOINTS[component], + command=local_runners._EXPECTED_COMMANDS[component], + assets=_assets(component), + timeout_seconds=3600.0, + memory_bytes=16 * 1024**3, + nano_cpus=4_000_000_000, + ) + + +def _image_build( + component: local_runners.PortableLabV1Component, + *, + derived_image_sha256: str, +) -> service_module.PortableLabV1ComponentImageBuildSeal: + return service_module.PortableLabV1ComponentImageBuildSeal.seal( + component=component, + base_image_sha256=BASE_IMAGE_SHA256S[component], + derived_image_sha256=derived_image_sha256, + dockerfile_sha256=hashlib.sha256( + f"{component}:dockerfile".encode() + ).hexdigest(), + shared_adapter_sha256=hashlib.sha256(b"shared-adapter").hexdigest(), + component_adapter_sha256=hashlib.sha256( + f"{component}:adapter".encode() + ).hexdigest(), + ) + + +def _runner_installation( + tmp_path: Path, + *, + definition_sha256: str, +) -> local_runners.PortableLabV1RunnerInstallation: + controller_root = tmp_path / "worker006-runtime" + controller_root.mkdir(exist_ok=True) + return local_runners.PortableLabV1RunnerInstallation.seal( + definition_sha256=definition_sha256, + release_candidate_sha256=RELEASE_CANDIDATE_SHA256, + work_root=local_runners.PortableLabV1WorkRootBinding( + controller_root=controller_root, + engine_host_root="D:\\NDC_MISSIONCORE\\runtime", + ), + eomt=_component("eomt"), + ddrnet=_component("ddrnet"), + ) + + +def _write_receipt( + tmp_path: Path, + *, + definition_sha256: str, + path_name: str = "lab-v1-worker-installation-receipt.json", + source_revision: str = SOURCE_REVISION, +) -> Path: + runner = _runner_installation( + tmp_path, + definition_sha256=definition_sha256, + ) + document = service_module.portable_lab_v1_worker_receipt_document( + source_revision=source_revision, + release_candidate_sha256=RELEASE_CANDIDATE_SHA256, + runner_installation=runner, + eomt_image_build=_image_build( + "eomt", + derived_image_sha256=runner.eomt.image_sha256, + ), + ddrnet_image_build=_image_build( + "ddrnet", + derived_image_sha256=runner.ddrnet.image_sha256, + ), + ) + path = tmp_path / path_name + path.write_bytes(canonical_json(document)) + return path + + +def _composition_inputs( + tmp_path: Path, +) -> tuple[ + PortableRunDefinition, + PortableWorkerRuntimeCandidate, + service_module.PortableLabV1WorkerInstallationReceipt, + PortableWorkerRuntimeAdmission, + PortableLabV1ReleaseInstallation, +]: + definition = _ready_definition() + receipt_path = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + ) + receipt = service_module.load_portable_lab_v1_worker_installation_receipt( + receipt_path + ) + candidate = _runtime_candidate(definition, receipt) + admission = PortableWorkerRuntimeAdmission( + candidate_sha256=candidate.candidate_sha256, + ready=True, + blockers=(), + assets=( + PortableWorkerAssetVerification( + asset_id=( + service_module.PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_ASSET_ID + ), + state="matched", + reason=None, + ), + ), + ) + release = _Release( + definition_contract_sha256=definition.executable_contract_sha256, + result_contract_sha256=definition.result_contract.contract_sha256, + assets=tuple( + _ReleaseAsset(kind=kind, sha256=sha256) + for kind, sha256 in sorted( + { + *( + ("container-image", digest) + for build in ( + receipt.eomt_image_build, + receipt.ddrnet_image_build, + ) + for digest in ( + build.base_image_sha256, + build.derived_image_sha256, + ) + ), + *( + ("runtime-artifact", asset.identity_sha256) + for component in ( + receipt.runner_installation.eomt, + receipt.runner_installation.ddrnet, + ) + for asset in component.assets + ), + *( + ("runtime-artifact", digest) + for build in ( + receipt.eomt_image_build, + receipt.ddrnet_image_build, + ) + for digest in ( + build.dockerfile_sha256, + build.shared_adapter_sha256, + build.component_adapter_sha256, + build.seal_sha256, + ) + ), + } + ) + ), + ) + inspection = PortableLabV1ReleaseInspection( + candidate_sha256=release.candidate_sha256, + matched_assets=(), + blockers=(), + ready=True, + ) + installation = cast( + PortableLabV1ReleaseInstallation, + _ReleaseInstallation(release=release, inspection=inspection), + ) + return definition, candidate, receipt, admission, installation + + +def test_receipt_loader_round_trips_full_runner_identity(tmp_path: Path) -> None: + definition = _ready_definition() + path = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + ) + + receipt = service_module.load_portable_lab_v1_worker_installation_receipt(path) + + assert receipt.path == path + assert receipt.release_candidate_sha256 == RELEASE_CANDIDATE_SHA256 + assert receipt.eomt_image_build.derived_image_sha256 == ( + receipt.runner_installation.eomt.image_sha256 + ) + assert receipt.ddrnet_image_build.network == "none" + assert receipt.runner_installation.definition_sha256 == ( + definition.definition_sha256 + ) + assert tuple(asset.asset_id for asset in receipt.runner_installation.eomt.assets) == ( + tuple(sorted(local_runners._EXPECTED_ASSET_TARGETS["eomt"])) + ) + assert receipt.file_sha256 == hashlib.sha256(path.read_bytes()).hexdigest() + + +def test_receipt_loader_rejects_noncanonical_links_and_oversized_files( + tmp_path: Path, +) -> None: + definition = _ready_definition() + path = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + ) + path.write_bytes(path.read_bytes() + b"\n") + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="not canonical JSON", + ): + service_module.load_portable_lab_v1_worker_installation_receipt(path) + + valid = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + path_name="valid.json", + ) + link = tmp_path / "linked.json" + link.symlink_to(valid) + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="unavailable", + ): + service_module.load_portable_lab_v1_worker_installation_receipt(link) + + oversized = tmp_path / "oversized.json" + oversized.write_bytes(b"x" * (256 * 1024 + 1)) + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="size is invalid", + ): + service_module.load_portable_lab_v1_worker_installation_receipt(oversized) + + +def test_receipt_loader_rejects_duplicate_keys_and_identity_tampering( + tmp_path: Path, +) -> None: + duplicate = tmp_path / "duplicate.json" + duplicate.write_bytes(b'{"schema_version":"a","schema_version":"b"}') + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="not canonical JSON", + ): + service_module.load_portable_lab_v1_worker_installation_receipt(duplicate) + + definition = _ready_definition() + path = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + ) + payload = path.read_bytes().replace(SOURCE_REVISION.encode(), ("f" * 40).encode()) + path.write_bytes(payload) + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="identity changed", + ): + service_module.load_portable_lab_v1_worker_installation_receipt(path) + + +def test_compose_returns_exact_lab_v1_builder_registration( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + definition, candidate, receipt, admission, installation = _composition_inputs( + tmp_path + ) + captured: dict[str, object] = {} + adapter = object() + + def _compose_adapter(**kwargs: object) -> object: + captured.update(kwargs) + return adapter + + monkeypatch.setattr( + service_module, + "compose_lab_v1_portable_executor_adapter", + _compose_adapter, + ) + builder = service_module.compose_installed_lab_v1_executor_builder( + receipt=receipt, + definition=definition, + candidate=candidate, + admission=admission, + release_installation=installation, + launcher=_Launcher(), + created_at_utc=lambda: "2026-08-31T12:00:00Z", + ) + + context = ObservatoryWorkerExecutorBuildContext( + definition=definition, + candidate=candidate, + source_transport=cast(PortableWorkerSourceMaterializer, object()), + result_transport=cast(PortableWorkerResultPublisher, object()), + work_root=tmp_path, + ) + registration = builder.builder(context) + + assert builder.setup_id == service_module.PORTABLE_LAB_V1_SETUP_ID + assert registration.identity == candidate.executor_identity() + assert registration.adapter is adapter + assert captured["candidate"] is candidate + assert captured["definition"] is definition + assert captured["admission"] is admission + assert captured["installation"] is installation + assert captured["source_output_parent"] == ( + tmp_path / "lab-v1-portable/source-output" + ) + assert captured["eomt_runner"].__class__.__name__ == ( + "InstalledPortableLabV1EomtRunner" + ) + assert captured["ddrnet_runner"].__class__.__name__ == ( + "InstalledPortableLabV1DdrnetRunner" + ) + + +def test_compose_fails_closed_on_changed_receipt_or_incomplete_admission( + tmp_path: Path, +) -> None: + definition, candidate, receipt, admission, installation = _composition_inputs( + tmp_path + ) + changed_path = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + source_revision="f" * 40, + ) + assert changed_path == receipt.path + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="changed after it was loaded", + ): + service_module.compose_installed_lab_v1_executor_builder( + receipt=receipt, + definition=definition, + candidate=candidate, + admission=admission, + release_installation=installation, + launcher=_Launcher(), + ) + + current = service_module.load_portable_lab_v1_worker_installation_receipt( + changed_path + ) + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="does not seal its installation receipt", + ): + service_module.compose_installed_lab_v1_executor_builder( + receipt=current, + definition=definition, + candidate=candidate, + admission=admission, + release_installation=installation, + launcher=_Launcher(), + ) + + restored_path = _write_receipt( + tmp_path, + definition_sha256=definition.definition_sha256, + ) + restored = service_module.load_portable_lab_v1_worker_installation_receipt( + restored_path + ) + release = cast(_Release, installation.release) + uncovered_installation = cast( + PortableLabV1ReleaseInstallation, + _ReleaseInstallation( + release=replace(release, assets=()), + inspection=installation.inspection, + ), + ) + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="does not cover its installed component identities", + ): + service_module.compose_installed_lab_v1_executor_builder( + receipt=restored, + definition=definition, + candidate=candidate, + admission=admission, + release_installation=uncovered_installation, + launcher=_Launcher(), + ) + + with pytest.raises( + service_module.PortableLabV1WorkerCompositionError, + match="has no complete local asset admission", + ): + service_module.compose_installed_lab_v1_executor_builder( + receipt=restored, + definition=definition, + candidate=candidate, + admission=PortableWorkerRuntimeAdmission( + candidate_sha256=candidate.candidate_sha256, + ready=False, + blockers=("asset-missing",), + assets=( + PortableWorkerAssetVerification( + asset_id=( + service_module.PORTABLE_LAB_V1_WORKER_INSTALLATION_RECEIPT_ASSET_ID + ), + state="missing", + reason="not-bound", + ), + ), + ), + release_installation=installation, + launcher=_Launcher(), + )