diff --git a/src/k1link/observatory/m49_worker_service.py b/src/k1link/observatory/m49_worker_service.py index cd01cb5..ed6c460 100644 --- a/src/k1link/observatory/m49_worker_service.py +++ b/src/k1link/observatory/m49_worker_service.py @@ -42,22 +42,21 @@ from k1link.observatory.portable_run_definitions import ( ) from k1link.observatory.portable_worker_runtime import ( PortableWorkerLocalAssetBinding, + PortableWorkerRuntimeAdmission, PortableWorkerRuntimeCandidate, PortableWorkerRuntimeRegistry, inspect_runtime_candidate, ) from k1link.observatory.worker_agent import ( WORKER_006_CONTOUR_ID, - ObservatoryWorkerAgent, ObservatoryWorkerExecutorRegistration, - ObservatoryWorkerExecutorRegistry, ) -from k1link.observatory.worker_http_transport import ObservatoryWorkerHttpGateway from k1link.observatory.worker_service import ( InstalledObservatoryWorkerService, + ObservatoryWorkerExecutorBuildContext, + ObservatoryWorkerExecutorBuilderRegistration, ObservatoryWorkerServiceConfiguration, - load_observatory_worker_bearer_token, - require_ready_executor_coverage, + compose_installed_observatory_worker_service_from_builders, ) M49_WORKER_DEFINITIONS_FILE_ENV: Final = "MISSIONCORE_OBSERVATORY_WORKER_DEFINITIONS_FILE" @@ -268,6 +267,44 @@ class M49WorkerInstallationReceipt: return bindings +@dataclass(frozen=True, slots=True) +class _M49WorkerExecutorBuilder: + """Bind the already admitted fixed M4.9 installation to shared Worker ports.""" + + definition: PortableRunDefinition + candidate: PortableWorkerRuntimeCandidate + admission: PortableWorkerRuntimeAdmission + installation: M49PortableRunnerInstallation + + 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 M49WorkerCompositionError( + "M4.9 builder context differs from its admitted installation" + ) + adapter = compose_m49_portable_executor_adapter( + candidate=context.candidate, + definition=context.definition, + admission=self.admission, + source_transport=context.source_transport, + result_transport=context.result_transport, + installation=self.installation, + source_output_parent=context.work_root / "m49-portable" / "source-output", + created_at_utc=_utc_now, + ) + return ObservatoryWorkerExecutorRegistration( + identity=context.candidate.executor_identity(), + adapter=adapter, + ) + + def load_m49_worker_installation_receipt( path: Path, ) -> M49WorkerInstallationReceipt: @@ -397,9 +434,10 @@ def load_m49_worker_installation_receipt( def compose_installed_m49_worker_service( configuration: M49WorkerEntrypointConfiguration, *, + executor_builders: Sequence[ObservatoryWorkerExecutorBuilderRegistration] = (), http_transport: httpx.BaseTransport | None = None, ) -> InstalledObservatoryWorkerService: - """Compose one fixed M4.9 executor around one shared HTTP gateway.""" + """Compose M4.9 and any other ready installed profile into one Worker agent.""" if os.name != "posix": raise M49WorkerCompositionError("portable M4.9 Worker requires a POSIX runtime") @@ -436,50 +474,22 @@ def compose_installed_m49_worker_service( ), output_parent=configuration.worker.work_root / "m49-portable" / "runner-output", ) - token = load_observatory_worker_bearer_token(configuration.worker.bearer_token_file) - gateway: ObservatoryWorkerHttpGateway | None = None - try: - gateway = ObservatoryWorkerHttpGateway( - base_url=configuration.worker.base_url, - bearer_token=token, - work_root=configuration.worker.work_root, - transport=http_transport, - ) - adapter = compose_m49_portable_executor_adapter( - candidate=candidate, + m49_builder = ObservatoryWorkerExecutorBuilderRegistration( + setup_id=M49_WORKER_SETUP_ID, + builder=_M49WorkerExecutorBuilder( definition=definition, + candidate=candidate, admission=admission, - source_transport=gateway, - result_transport=gateway, installation=installation, - source_output_parent=( - configuration.worker.work_root / "m49-portable" / "source-output" - ), - created_at_utc=_utc_now, - ) - executors = ObservatoryWorkerExecutorRegistry( - ( - ObservatoryWorkerExecutorRegistration( - identity=candidate.executor_identity(), - adapter=adapter, - ), - ) - ) - require_ready_executor_coverage( - definitions=definitions, - executors=executors, - ) - return InstalledObservatoryWorkerService( - configuration=configuration.worker, - gateway=gateway, - agent=ObservatoryWorkerAgent(transport=gateway, executors=executors), - ) - except Exception: - if gateway is not None: - gateway.close() - raise - finally: - del token + ), + ) + return compose_installed_observatory_worker_service_from_builders( + configuration=configuration.worker, + definitions=definitions, + runtime_registry=runtime_registry, + builders=(m49_builder, *executor_builders), + http_transport=http_transport, + ) def run_installed_m49_worker( @@ -520,10 +530,11 @@ def _verify_ready_identity( definition: PortableRunDefinition, candidate: PortableWorkerRuntimeCandidate, ) -> None: - ready = definitions.ready_recorded_definitions() + ready_identities = { + (item.setup_id, item.definition_sha256) for item in definitions.ready_recorded_definitions() + } if ( - len(ready) != 1 - or ready[0].setup_id != M49_WORKER_SETUP_ID + (definition.setup_id, definition.definition_sha256) not in ready_identities or definition.setup_id != M49_WORKER_SETUP_ID or definition.executor.contour_id != WORKER_006_CONTOUR_ID or candidate.adapter_id != M49_WORKER_ADAPTER_ID @@ -531,7 +542,7 @@ def _verify_ready_identity( or not candidate.ready ): raise M49WorkerCompositionError( - "fixed M4.9 Worker requires exactly one ready M4.9 definition" + "fixed M4.9 Worker requires its exact ready M4.9 definition" ) diff --git a/src/k1link/observatory/worker_service.py b/src/k1link/observatory/worker_service.py index 503386f..2dbae4d 100644 --- a/src/k1link/observatory/worker_service.py +++ b/src/k1link/observatory/worker_service.py @@ -12,20 +12,30 @@ from __future__ import annotations import os import re import stat -from collections.abc import Callable, Mapping +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from pathlib import Path from threading import Event -from typing import Final +from typing import Final, Protocol from urllib.parse import urlsplit import httpx -from k1link.observatory.portable_run_definitions import PortableRunDefinitionRegistry +from k1link.observatory.portable_run_definitions import ( + PortableRunDefinition, + PortableRunDefinitionRegistry, +) +from k1link.observatory.portable_worker_runtime import ( + PortableWorkerResultPublisher, + PortableWorkerRuntimeCandidate, + PortableWorkerRuntimeRegistry, + PortableWorkerSourceMaterializer, +) from k1link.observatory.worker_agent import ( ObservatoryWorkerAgent, ObservatoryWorkerCycleReport, ObservatoryWorkerExecutorIdentity, + ObservatoryWorkerExecutorRegistration, ObservatoryWorkerExecutorRegistry, ObservatoryWorkerExecutorUnavailableError, ) @@ -37,9 +47,7 @@ from k1link.observatory.worker_http_transport import ( OBSERVATORY_WORKER_BASE_URL_ENV: Final = "MISSIONCORE_OBSERVATORY_WORKER_BASE_URL" OBSERVATORY_WORKER_TOKEN_FILE_ENV: Final = "MISSIONCORE_OBSERVATORY_WORKER_TOKEN_FILE" OBSERVATORY_WORKER_WORK_ROOT_ENV: Final = "MISSIONCORE_OBSERVATORY_WORKER_WORK_ROOT" -OBSERVATORY_WORKER_IDLE_POLL_SECONDS_ENV: Final = ( - "MISSIONCORE_OBSERVATORY_WORKER_IDLE_POLL_SECONDS" -) +OBSERVATORY_WORKER_IDLE_POLL_SECONDS_ENV: Final = "MISSIONCORE_OBSERVATORY_WORKER_IDLE_POLL_SECONDS" OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS_ENV: Final = ( "MISSIONCORE_OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS" ) @@ -53,12 +61,50 @@ DEFAULT_OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS: Final = 5.0 DEFAULT_OBSERVATORY_WORKER_MAX_TRANSPORT_FAILURES: Final = 12 _TOKEN = re.compile(r"^[A-Za-z0-9._:-]{32,512}$") +_SETUP_ID = re.compile(r"^[a-z][a-z0-9-]{2,95}$") class ObservatoryWorkerServiceError(RuntimeError): """Worker 006 cannot start without its exact local operational boundary.""" +class ObservatoryWorkerExecutorBuilder(Protocol): + """Installed code port that builds one reviewed executor before polling.""" + + def __call__( + self, + context: ObservatoryWorkerExecutorBuildContext, + ) -> ObservatoryWorkerExecutorRegistration: ... + + +@dataclass(frozen=True, slots=True) +class ObservatoryWorkerExecutorBuildContext: + """Fixed local inputs shared with one install-time executor builder.""" + + definition: PortableRunDefinition + candidate: PortableWorkerRuntimeCandidate + source_transport: PortableWorkerSourceMaterializer + result_transport: PortableWorkerResultPublisher + work_root: Path + + def __post_init__(self) -> None: + _absolute_path(self.work_root, "Worker build work root") + + +@dataclass(frozen=True, slots=True) +class ObservatoryWorkerExecutorBuilderRegistration: + """Local setup-to-builder binding; queued jobs cannot populate this map.""" + + setup_id: str + builder: ObservatoryWorkerExecutorBuilder + + def __post_init__(self) -> None: + if _SETUP_ID.fullmatch(self.setup_id) is None: + raise ValueError("Worker executor builder setup id is invalid") + if not callable(self.builder): + raise ValueError("Worker executor builder is not callable") + + @dataclass(frozen=True, slots=True) class ObservatoryWorkerServiceConfiguration: """Path-only service configuration; it never contains a plaintext secret.""" @@ -67,12 +113,8 @@ class ObservatoryWorkerServiceConfiguration: bearer_token_file: Path work_root: Path idle_poll_seconds: float = DEFAULT_OBSERVATORY_WORKER_IDLE_POLL_SECONDS - transport_backoff_seconds: float = ( - DEFAULT_OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS - ) - max_consecutive_transport_failures: int = ( - DEFAULT_OBSERVATORY_WORKER_MAX_TRANSPORT_FAILURES - ) + transport_backoff_seconds: float = DEFAULT_OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS + max_consecutive_transport_failures: int = DEFAULT_OBSERVATORY_WORKER_MAX_TRANSPORT_FAILURES def __post_init__(self) -> None: _validate_worker_base_url(self.base_url) @@ -191,9 +233,7 @@ def compose_installed_observatory_worker_service( """ require_ready_executor_coverage(definitions=definitions, executors=executors) - bearer_token = load_observatory_worker_bearer_token( - configuration.bearer_token_file - ) + bearer_token = load_observatory_worker_bearer_token(configuration.bearer_token_file) try: gateway = ObservatoryWorkerHttpGateway( base_url=configuration.base_url, @@ -212,6 +252,119 @@ def compose_installed_observatory_worker_service( ) +def compose_installed_observatory_worker_service_from_builders( + *, + configuration: ObservatoryWorkerServiceConfiguration, + definitions: PortableRunDefinitionRegistry, + runtime_registry: PortableWorkerRuntimeRegistry, + builders: Sequence[ObservatoryWorkerExecutorBuilderRegistration], + http_transport: httpx.BaseTransport | None = None, +) -> InstalledObservatoryWorkerService: + """Build complete ready-profile coverage before one agent can claim work. + + Builders are reviewed local installation ports. Mission Core jobs supply + neither builders nor executable locators. Every ready definition must have + one builder, and that builder must return the exact executor identity sealed + by the matching runtime candidate. Composition fails and closes transport + before constructing an agent if any ready profile is missing or mismatched. + """ + + bearer_token = load_observatory_worker_bearer_token(configuration.bearer_token_file) + gateway: ObservatoryWorkerHttpGateway | None = None + try: + gateway = ObservatoryWorkerHttpGateway( + base_url=configuration.base_url, + bearer_token=bearer_token, + work_root=configuration.work_root, + transport=http_transport, + ) + executors = build_ready_executor_registry( + definitions=definitions, + runtime_registry=runtime_registry, + builders=builders, + source_transport=gateway, + result_transport=gateway, + work_root=configuration.work_root, + ) + return InstalledObservatoryWorkerService( + configuration=configuration, + gateway=gateway, + agent=ObservatoryWorkerAgent(transport=gateway, executors=executors), + ) + except Exception: + if gateway is not None: + gateway.close() + raise + finally: + del bearer_token + + +def build_ready_executor_registry( + *, + definitions: PortableRunDefinitionRegistry, + runtime_registry: PortableWorkerRuntimeRegistry, + builders: Sequence[ObservatoryWorkerExecutorBuilderRegistration], + source_transport: PortableWorkerSourceMaterializer, + result_transport: PortableWorkerResultPublisher, + work_root: Path, +) -> ObservatoryWorkerExecutorRegistry: + """Build one exact registry covering every ready definition, or reject it.""" + + _absolute_path(work_root, "Worker build work root") + builders_by_setup: dict[str, ObservatoryWorkerExecutorBuilder] = {} + for registration in builders: + if registration.setup_id in builders_by_setup: + raise ObservatoryWorkerServiceError("Worker executor builder setup ids must be unique") + builders_by_setup[registration.setup_id] = registration.builder + + ready_definitions = definitions.ready_recorded_definitions() + ready_setup_ids = {definition.setup_id for definition in ready_definitions} + builder_setup_ids = set(builders_by_setup) + if ready_setup_ids - builder_setup_ids: + raise ObservatoryWorkerServiceError( + "a ready portable RunDefinition has no installed local builder" + ) + if builder_setup_ids - ready_setup_ids: + raise ObservatoryWorkerServiceError( + "an installed local builder targets a non-ready RunDefinition" + ) + + registrations: list[ObservatoryWorkerExecutorRegistration] = [] + for recorded_definition in ready_definitions: + definition = definitions.resolve( + recorded_definition.setup_id, + recorded_definition.definition_sha256, + ) + builder = builders_by_setup[definition.setup_id] + candidate = runtime_registry.resolve( + definition.setup_id, + definition.definition_sha256, + ) + if not candidate.ready: + raise ObservatoryWorkerServiceError( + "a ready portable RunDefinition has no ready local runtime candidate" + ) + expected_identity = candidate.executor_identity() + built = builder( + ObservatoryWorkerExecutorBuildContext( + definition=definition, + candidate=candidate, + source_transport=source_transport, + result_transport=result_transport, + work_root=work_root, + ) + ) + if built.identity != expected_identity: + raise ObservatoryWorkerServiceError( + "installed executor builder returned another exact identity" + ) + registrations.append(built) + + executors = ObservatoryWorkerExecutorRegistry(tuple(registrations)) + require_ready_executor_coverage(definitions=definitions, executors=executors) + return executors + + def require_ready_executor_coverage( *, definitions: PortableRunDefinitionRegistry, @@ -221,9 +374,7 @@ def require_ready_executor_coverage( ready = definitions.ready_recorded_definitions() if not ready: - raise ObservatoryWorkerServiceError( - "no portable RunDefinition has a sealed ready executor" - ) + raise ObservatoryWorkerServiceError("no portable RunDefinition has a sealed ready executor") identities: list[ObservatoryWorkerExecutorIdentity] = [] for definition in ready: identity = ObservatoryWorkerExecutorIdentity( @@ -254,9 +405,7 @@ def load_observatory_worker_bearer_token(path: Path) -> str: ) metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode): - raise ObservatoryWorkerServiceError( - "Worker bearer credential must be a regular file" - ) + raise ObservatoryWorkerServiceError("Worker bearer credential must be a regular file") if os.name != "posix": raise ObservatoryWorkerServiceError( "native Worker credential ACL verification is not available; " @@ -267,31 +416,23 @@ def load_observatory_worker_bearer_token(path: Path) -> str: "Worker bearer credential permissions are too broad" ) if not 32 <= metadata.st_size <= 512: - raise ObservatoryWorkerServiceError( - "Worker bearer credential format is invalid" - ) + raise ObservatoryWorkerServiceError("Worker bearer credential format is invalid") with os.fdopen(descriptor, "rb") as stream: descriptor = None payload = stream.read(513) except ObservatoryWorkerServiceError: raise except OSError as exc: - raise ObservatoryWorkerServiceError( - "Worker bearer credential is unavailable" - ) from exc + raise ObservatoryWorkerServiceError("Worker bearer credential is unavailable") from exc finally: if descriptor is not None: os.close(descriptor) try: token = payload.decode("ascii") except UnicodeDecodeError as exc: - raise ObservatoryWorkerServiceError( - "Worker bearer credential is not ASCII" - ) from exc + raise ObservatoryWorkerServiceError("Worker bearer credential is not ASCII") from exc if _TOKEN.fullmatch(token) is None: - raise ObservatoryWorkerServiceError( - "Worker bearer credential format is invalid" - ) + raise ObservatoryWorkerServiceError("Worker bearer credential format is invalid") return token diff --git a/tests/test_observatory_m49_worker_entrypoint.py b/tests/test_observatory_m49_worker_entrypoint.py index f734403..ab73a84 100644 --- a/tests/test_observatory_m49_worker_entrypoint.py +++ b/tests/test_observatory_m49_worker_entrypoint.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy import hashlib import json +from dataclasses import replace from pathlib import Path from threading import Event from typing import cast @@ -32,7 +33,10 @@ from k1link.observatory.portable_run_definitions import ( PortableRunDefinitionRegistry, canonical_sha256, ) -from k1link.observatory.portable_worker_runtime import PortableWorkerExecutorAdapter +from k1link.observatory.portable_worker_runtime import ( + PortableWorkerExecutorAdapter, + PortableWorkerRuntimeRegistry, +) from k1link.observatory.worker_http_transport import ObservatoryWorkerHttpGateway from k1link.observatory.worker_service import ObservatoryWorkerServiceConfiguration @@ -223,7 +227,7 @@ def _runtime_registry( receipt_payload: bytes, ) -> Path: definition = definitions.resolve_setup(service_module.M49_WORKER_SETUP_ID) - assets = [ + assets: list[dict[str, object]] = [ { "asset_id": M49_PORTABLE_COMPILED_RUNNER_ASSET_ID, "kind": "local-file", @@ -390,6 +394,36 @@ def test_fixed_m49_composition_uses_one_gateway_for_agent_source_and_result( assert gateway._client.is_closed # noqa: SLF001 +def test_fixed_m49_identity_allows_an_additional_ready_worker_profile( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + configuration, _receipt = _fixture(tmp_path, monkeypatch) + definitions = PortableRunDefinitionRegistry.from_file(configuration.definitions_file) + definition = definitions.resolve_setup(service_module.M49_WORKER_SETUP_ID) + runtime = PortableWorkerRuntimeRegistry.from_file( + configuration.runtime_registry_file, + definitions=definitions, + ) + candidate = runtime.resolve(definition.setup_id, definition.definition_sha256) + additional = replace( + definitions.ready_recorded_definitions()[0], + setup_id="lab-v1-eomt-ddrnet-portable-v1", + definition_id="lab-v1-eomt-ddrnet-portable", + definition_sha256="7" * 64, + ) + + class _DefinitionsWithAdditionalReadyProfile: + def ready_recorded_definitions(self): # type: ignore[no-untyped-def] + return (*definitions.ready_recorded_definitions(), additional) + + service_module._verify_ready_identity( # noqa: SLF001 + cast(PortableRunDefinitionRegistry, _DefinitionsWithAdditionalReadyProfile()), + definition, + candidate, + ) + + def test_fixed_m49_composition_rejects_receipt_asset_drift_before_gateway( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_observatory_worker_service.py b/tests/test_observatory_worker_service.py index b3d5634..11bc066 100644 --- a/tests/test_observatory_worker_service.py +++ b/tests/test_observatory_worker_service.py @@ -1,13 +1,15 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from threading import Event from typing import cast +import httpx import pytest from k1link.observatory.portable_run_definitions import PortableRunDefinitionRegistry +from k1link.observatory.portable_worker_runtime import PortableWorkerRuntimeRegistry from k1link.observatory.recorded_jobs import RecordedRunDefinition from k1link.observatory.worker_agent import ( ObservatoryWorkerCycleReport, @@ -20,8 +22,11 @@ from k1link.observatory.worker_agent import ( from k1link.observatory.worker_http_transport import ObservatoryWorkerHttpError from k1link.observatory.worker_service import ( InstalledObservatoryWorkerService, + ObservatoryWorkerExecutorBuildContext, + ObservatoryWorkerExecutorBuilderRegistration, ObservatoryWorkerServiceConfiguration, ObservatoryWorkerServiceError, + compose_installed_observatory_worker_service_from_builders, load_observatory_worker_bearer_token, require_ready_executor_coverage, ) @@ -55,6 +60,58 @@ class _ReadyDefinitions: def ready_recorded_definitions(self) -> tuple[RecordedRunDefinition, ...]: return self.definitions + def resolve(self, setup_id: str, definition_sha256: str) -> RecordedRunDefinition: + for definition in self.definitions: + if ( + definition.setup_id == setup_id + and definition.definition_sha256 == definition_sha256 + ): + return definition + raise AssertionError((setup_id, definition_sha256)) + + +@dataclass(frozen=True) +class _RuntimeCandidate: + setup_id: str + definition_sha256: str + identity: ObservatoryWorkerExecutorIdentity + ready: bool = True + + def executor_identity(self) -> ObservatoryWorkerExecutorIdentity: + return self.identity + + +@dataclass(frozen=True) +class _RuntimeRegistry: + candidates: tuple[_RuntimeCandidate, ...] + + def resolve(self, setup_id: str, definition_sha256: str) -> _RuntimeCandidate: + for candidate in self.candidates: + if candidate.setup_id == setup_id and candidate.definition_sha256 == definition_sha256: + return candidate + raise AssertionError((setup_id, definition_sha256)) + + +@dataclass +class _Builder: + identity: ObservatoryWorkerExecutorIdentity + contexts: list[ObservatoryWorkerExecutorBuildContext] + + def __call__( + self, + context: ObservatoryWorkerExecutorBuildContext, + ) -> ObservatoryWorkerExecutorRegistration: + self.contexts.append(context) + return ObservatoryWorkerExecutorRegistration(self.identity, _Executor()) + + +class _TrackingMockTransport(httpx.MockTransport): + closed = False + + def close(self) -> None: + self.closed = True + super().close() + class _Executor: def execute( @@ -73,6 +130,51 @@ def _identity(definition: RecordedRunDefinition) -> ObservatoryWorkerExecutorIde ) +def _two_ready_profiles() -> tuple[ + PortableRunDefinitionRegistry, + PortableWorkerRuntimeRegistry, + RecordedRunDefinition, + RecordedRunDefinition, +]: + first = _definition() + second = replace( + _definition(), + setup_id="portable-lab-v2", + definition_id="portable-lab-v2-definition", + definition_sha256="7" * 64, + executor_release_id="portable-lab-v2-worker", + executor_release_sha256="8" * 64, + executor_image_sha256="9" * 64, + model_manifest_sha256="a" * 64, + resource_profile_sha256="b" * 64, + ) + definitions = cast( + PortableRunDefinitionRegistry, + _ReadyDefinitions((first, second)), + ) + runtime_registry = cast( + PortableWorkerRuntimeRegistry, + _RuntimeRegistry( + tuple( + _RuntimeCandidate( + definition.setup_id, + definition.definition_sha256, + _identity(definition), + ) + for definition in (first, second) + ) + ), + ) + return definitions, runtime_registry, first, second + + +def _private_token(tmp_path: Path) -> Path: + token = tmp_path / "worker.token" + token.write_text("worker-006-test-bearer-token-000001", encoding="ascii") + token.chmod(0o600) + return token + + def _configuration(tmp_path: Path, **overrides: object) -> ObservatoryWorkerServiceConfiguration: values: dict[str, object] = { "base_url": "http://127.0.0.1:18080", @@ -111,9 +213,7 @@ def test_worker_token_loader_requires_private_exact_ascii_file(tmp_path: Path) - token.write_text("worker-006-test-bearer-token-000001", encoding="ascii") token.chmod(0o600) - assert load_observatory_worker_bearer_token(token) == ( - "worker-006-test-bearer-token-000001" - ) + assert load_observatory_worker_bearer_token(token) == ("worker-006-test-bearer-token-000001") token.chmod(0o644) with pytest.raises(ObservatoryWorkerServiceError, match="permissions"): @@ -156,6 +256,154 @@ def test_install_time_coverage_requires_each_ready_executor_identity() -> None: ) +def test_builder_composition_registers_all_ready_profiles_before_one_agent_claims( + tmp_path: Path, +) -> None: + definitions, runtime_registry, first, second = _two_ready_profiles() + _private_token(tmp_path) + contexts: list[ObservatoryWorkerExecutorBuildContext] = [] + first_builder = _Builder(_identity(first), contexts) + second_builder = _Builder(_identity(second), contexts) + requests: list[str] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request.url.path) + return httpx.Response(204) + + service = compose_installed_observatory_worker_service_from_builders( + configuration=_configuration(tmp_path), + definitions=definitions, + runtime_registry=runtime_registry, + builders=( + ObservatoryWorkerExecutorBuilderRegistration( + first.setup_id, + first_builder, + ), + ObservatoryWorkerExecutorBuilderRegistration( + second.setup_id, + second_builder, + ), + ), + http_transport=httpx.MockTransport(handle), + ) + + assert len(service.agent._executors.registrations) == 2 # noqa: SLF001 + assert len(contexts) == 2 + assert all(context.source_transport is service.gateway for context in contexts) + assert all(context.result_transport is service.gateway for context in contexts) + assert service.agent.run_once().state == "empty" + assert requests == ["/api/v1/worker/observatory/recorded-jobs/claims"] + service.close() + + +def test_builder_composition_fails_closed_before_claim_and_closes_gateway( + tmp_path: Path, +) -> None: + definitions, runtime_registry, first, _second = _two_ready_profiles() + _private_token(tmp_path) + contexts: list[ObservatoryWorkerExecutorBuildContext] = [] + requests: list[str] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request.url.path) + return httpx.Response(204) + + transport = _TrackingMockTransport(handle) + + with pytest.raises(ObservatoryWorkerServiceError, match="no installed local builder"): + compose_installed_observatory_worker_service_from_builders( + configuration=_configuration(tmp_path), + definitions=definitions, + runtime_registry=runtime_registry, + builders=( + ObservatoryWorkerExecutorBuilderRegistration( + first.setup_id, + _Builder(_identity(first), contexts), + ), + ), + http_transport=transport, + ) + + assert requests == [] + assert contexts == [] + assert transport.closed is True + + +def test_builder_composition_rejects_a_builder_for_a_non_ready_profile( + tmp_path: Path, +) -> None: + definition = _definition() + definitions = cast( + PortableRunDefinitionRegistry, + _ReadyDefinitions((definition,)), + ) + runtime_registry = cast( + PortableWorkerRuntimeRegistry, + _RuntimeRegistry( + ( + _RuntimeCandidate( + definition.setup_id, + definition.definition_sha256, + _identity(definition), + ), + ) + ), + ) + _private_token(tmp_path) + contexts: list[ObservatoryWorkerExecutorBuildContext] = [] + + with pytest.raises(ObservatoryWorkerServiceError, match="non-ready RunDefinition"): + compose_installed_observatory_worker_service_from_builders( + configuration=_configuration(tmp_path), + definitions=definitions, + runtime_registry=runtime_registry, + builders=( + ObservatoryWorkerExecutorBuilderRegistration( + definition.setup_id, + _Builder(_identity(definition), contexts), + ), + ObservatoryWorkerExecutorBuilderRegistration( + "lab-v1-eomt-ddrnet-portable-v1", + _Builder(_identity(definition), contexts), + ), + ), + ) + + assert contexts == [] + + +def test_builder_composition_rejects_identity_drift_before_claim(tmp_path: Path) -> None: + definition = _definition() + definitions = cast( + PortableRunDefinitionRegistry, + _ReadyDefinitions((definition,)), + ) + runtime_registry = _RuntimeRegistry( + ( + _RuntimeCandidate( + definition.setup_id, + definition.definition_sha256, + _identity(definition), + ), + ) + ) + _private_token(tmp_path) + mismatched = replace(_identity(definition), release_sha256="f" * 64) + + with pytest.raises(ObservatoryWorkerServiceError, match="another exact identity"): + compose_installed_observatory_worker_service_from_builders( + configuration=_configuration(tmp_path), + definitions=definitions, + runtime_registry=cast(PortableWorkerRuntimeRegistry, runtime_registry), + builders=( + ObservatoryWorkerExecutorBuilderRegistration( + definition.setup_id, + _Builder(mismatched, []), + ), + ), + ) + + @dataclass class _FakeGateway: closed: bool = False