feat(worker): compose all ready observatory profiles

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,
+    compose_installed_observatory_worker_service_from_builders,
-    require_ready_executor_coverage,
 )
 
 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)
+    m49_builder = ObservatoryWorkerExecutorBuilderRegistration(
-    gateway: ObservatoryWorkerHttpGateway | None = None
+        setup_id=M49_WORKER_SETUP_ID,
-    try:
+        builder=_M49WorkerExecutorBuilder(
-        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,
             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"
+    )
-            ),
+    return compose_installed_observatory_worker_service_from_builders(
-            created_at_utc=_utc_now,
+        configuration=configuration.worker,
-        )
+        definitions=definitions,
-        executors = ObservatoryWorkerExecutorRegistry(
+        runtime_registry=runtime_registry,
-            (
+        builders=(m49_builder, *executor_builders),
-                ObservatoryWorkerExecutorRegistration(
+        http_transport=http_transport,
-                    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
 
 
 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
+        (definition.setup_id, definition.definition_sha256) not in ready_identities
-        or ready[0].setup_id != M49_WORKER_SETUP_ID
         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"
         )
 
 

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 = (
+OBSERVATORY_WORKER_IDLE_POLL_SECONDS_ENV: Final = "MISSIONCORE_OBSERVATORY_WORKER_IDLE_POLL_SECONDS"
-    "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 = (
+    transport_backoff_seconds: float = DEFAULT_OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS
-        DEFAULT_OBSERVATORY_WORKER_TRANSPORT_BACKOFF_SECONDS
+    max_consecutive_transport_failures: int = DEFAULT_OBSERVATORY_WORKER_MAX_TRANSPORT_FAILURES
-    )
-    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(
+    bearer_token = load_observatory_worker_bearer_token(configuration.bearer_token_file)
-        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(
+        raise ObservatoryWorkerServiceError("no portable RunDefinition has a sealed ready executor")
-            "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(
+            raise ObservatoryWorkerServiceError("Worker bearer credential must be a regular file")
-                "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(
+            raise ObservatoryWorkerServiceError("Worker bearer credential format is invalid")
-                "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(
+        raise ObservatoryWorkerServiceError("Worker bearer credential is unavailable") from exc
-            "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(
+        raise ObservatoryWorkerServiceError("Worker bearer credential is not ASCII") from exc
-            "Worker bearer credential is not ASCII"
-        ) from exc
     if _TOKEN.fullmatch(token) is None:
-        raise ObservatoryWorkerServiceError(
+        raise ObservatoryWorkerServiceError("Worker bearer credential format is invalid")
-            "Worker bearer credential format is invalid"
-        )
     return token
 
 

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,

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) == (
+    assert load_observatory_worker_bearer_token(token) == ("worker-006-test-bearer-token-000001")
-        "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