feat(observatory): add portable LAB V1 foundation

src/k1link/observatory/portable_queue_binding.py

@@ -0,0 +1,303 @@
+"""Portable recorded-source admission bound to the durable Observatory queue.
+
+The binding resolves a server-owned portable RunDefinition before inspecting a
+session.  It therefore cannot write source contracts or jobs for an unavailable
+executor.  A successful check is fenced by a content digest; admission repeats
+the source read and persists only if the exact checked snapshot still holds.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+from k1link.observatory.portable_run_definitions import (
+    PortableRunDefinition,
+    PortableRunDefinitionRegistry,
+)
+from k1link.observatory.recorded_jobs import (
+    ObservatoryRecordedJob,
+    ObservatoryRecordedJobIntent,
+    ObservatoryRecordedJobQueue,
+    RecordedRunDefinition,
+)
+from k1link.observatory.source_admission import (
+    PortableRecordedSourceAdmission,
+    PortableRecordedSourceCapability,
+    PortableSourceAdmissionStaleError,
+    RecordedK1SourceAdmissionService,
+)
+from k1link.sessions.media import RecordedMediaInspector
+from k1link.sessions.store import SessionStore
+
+PORTABLE_QUEUE_BINDING_SCHEMA: Final = "missioncore.observatory-portable-queue-binding/v1"
+_SHA256: Final = re.compile(r"^[a-f0-9]{64}$")
+_IDEMPOTENCY_KEY: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$")
+_AUTHORITY: Final = {
+    "commands_enabled": False,
+    "actuation_allowed": False,
+    "navigation_or_safety_accepted": False,
+    "production_accepted": False,
+}
+
+
+class PortableQueueBindingError(RuntimeError):
+    """A portable source cannot be bound to the recorded queue."""
+
+
+class PortableQueueBindingIntegrityError(PortableQueueBindingError):
+    """A registry, source admission, or queue identity disagrees."""
+
+
+class PortableQueueBindingStaleCheckError(PortableQueueBindingError):
+    """The source changed after the operator-visible compatibility check."""
+
+
+@dataclass(frozen=True, slots=True)
+class PortableRecordedRunPreparation:
+    """Path-free immutable bridge from source admission to a queue intent."""
+
+    definition: RecordedRunDefinition
+    source: PortableRecordedSourceAdmission
+
+    def __post_init__(self) -> None:
+        if self.definition.source_adapter_sha256 != self.source.source_adapter_sha256:
+            raise PortableQueueBindingIntegrityError(
+                "portable registry and source admission adapter identities disagree"
+            )
+
+    @property
+    def check_sha256(self) -> str:
+        return _sha256(self.identity_document())
+
+    def identity_document(self) -> dict[str, object]:
+        return {
+            "schema_version": PORTABLE_QUEUE_BINDING_SCHEMA,
+            "source": self.source.as_dict(),
+            "definition": _definition_document(self.definition),
+            "authority": dict(_AUTHORITY),
+        }
+
+    def intent(self, *, idempotency_key: str) -> ObservatoryRecordedJobIntent:
+        _pattern(idempotency_key, _IDEMPOTENCY_KEY, "idempotency key")
+        return ObservatoryRecordedJobIntent(
+            idempotency_key=idempotency_key,
+            source_session_id=self.source.source_session_id,
+            source_catalog_sha256=self.source.source_catalog_sha256,
+            source_bundle_sha256=self.source.source_bundle_sha256,
+            source_capability_manifest_sha256=(self.source.source_capability_manifest_sha256),
+            setup_id=self.definition.setup_id,
+            definition_sha256=self.definition.definition_sha256,
+        )
+
+    def as_dict(self) -> dict[str, object]:
+        return {**self.identity_document(), "check_sha256": self.check_sha256}
+
+
+class PortableRecordedQueueBindingService:
+    """Resolve, check, admit, and submit portable recorded computations."""
+
+    def __init__(
+        self,
+        *,
+        data_dir: Path,
+        session_store: SessionStore,
+        media_inspector: RecordedMediaInspector,
+        definitions: PortableRunDefinitionRegistry,
+        queue: ObservatoryRecordedJobQueue | None = None,
+    ) -> None:
+        self.data_dir = data_dir.expanduser().resolve()
+        if self.data_dir != session_store.data_dir:
+            raise PortableQueueBindingIntegrityError(
+                "portable binding and SessionStore roots disagree"
+            )
+        if queue is not None and self.data_dir != queue.data_dir:
+            raise PortableQueueBindingIntegrityError(
+                "portable binding and recorded queue roots disagree"
+            )
+        self._session_store = session_store
+        self._media_inspector = media_inspector
+        self._definitions = definitions
+        self._queue = queue
+
+    def probe(
+        self,
+        *,
+        source_session_id: str,
+        setup_id: str,
+        definition_sha256: str,
+    ) -> PortableRecordedSourceCapability:
+        """Return cheap authoritative compatibility without replay preparation."""
+
+        portable, recorded = self._resolve_definition(setup_id, definition_sha256)
+        capability = self._source_service(portable).probe(source_session_id)
+        if recorded.source_adapter_sha256 != capability.source_adapter_sha256:
+            raise PortableQueueBindingIntegrityError(
+                "portable registry and source capability adapter identities disagree"
+            )
+        return capability
+
+    def check(
+        self,
+        *,
+        source_session_id: str,
+        setup_id: str,
+        definition_sha256: str,
+    ) -> PortableRecordedRunPreparation:
+        """Verify one source/setup combination without persistent writes."""
+
+        portable, recorded = self._resolve_definition(setup_id, definition_sha256)
+        source_service = self._source_service(portable)
+        return self._bind(recorded, source_service.check(source_session_id))
+
+    def admit(
+        self,
+        *,
+        source_session_id: str,
+        setup_id: str,
+        definition_sha256: str,
+        expected_check_sha256: str,
+    ) -> PortableRecordedRunPreparation:
+        """Persist exact checked contracts and reject changed source state."""
+
+        _digest(expected_check_sha256, "expected portable check sha256")
+        portable, recorded = self._resolve_definition(setup_id, definition_sha256)
+        source_service = self._source_service(portable)
+        checked = self._bind(recorded, source_service.check(source_session_id))
+        if checked.check_sha256 != expected_check_sha256:
+            raise PortableQueueBindingStaleCheckError(
+                "portable source or definition changed after its check"
+            )
+        try:
+            source = source_service.admit(
+                source_session_id,
+                expected_admission_sha256=checked.source.identity_sha256,
+            )
+        except PortableSourceAdmissionStaleError as exc:
+            raise PortableQueueBindingStaleCheckError(
+                "portable source changed while admission was being committed"
+            ) from exc
+        admitted = self._bind(recorded, source)
+        if admitted.check_sha256 != expected_check_sha256:
+            raise PortableQueueBindingStaleCheckError(
+                "portable source changed while admission was being committed"
+            )
+        return admitted
+
+    def submit(
+        self,
+        *,
+        source_session_id: str,
+        setup_id: str,
+        definition_sha256: str,
+        expected_check_sha256: str,
+        idempotency_key: str,
+        enqueue: bool = True,
+    ) -> tuple[ObservatoryRecordedJob, bool]:
+        """Admit and atomically submit the resulting path-free queue intent."""
+
+        if self._queue is None:
+            raise PortableQueueBindingIntegrityError("portable recorded queue is unavailable")
+        preparation = self.admit(
+            source_session_id=source_session_id,
+            setup_id=setup_id,
+            definition_sha256=definition_sha256,
+            expected_check_sha256=expected_check_sha256,
+        )
+        return self._queue.submit(
+            preparation.intent(idempotency_key=idempotency_key),
+            enqueue=enqueue,
+        )
+
+    def _resolve_definition(
+        self,
+        setup_id: str,
+        definition_sha256: str,
+    ) -> tuple[PortableRunDefinition, RecordedRunDefinition]:
+        portable = self._definitions.resolve(setup_id, definition_sha256)
+        # This conversion is deliberately first: not-installed executors fail
+        # before the SessionStore, immutable document store, or queue is touched.
+        recorded = portable.to_recorded_run_definition()
+        if self._queue is not None:
+            try:
+                queue_definition = self._queue.resolve_definition(
+                    setup_id,
+                    definition_sha256,
+                )
+            except Exception as exc:
+                raise PortableQueueBindingIntegrityError(
+                    "portable definition is absent from the recorded queue"
+                ) from exc
+            if queue_definition != recorded:
+                raise PortableQueueBindingIntegrityError(
+                    "portable registry and recorded queue definitions disagree"
+                )
+        return portable, recorded
+
+    def _source_service(
+        self,
+        definition: PortableRunDefinition,
+    ) -> RecordedK1SourceAdmissionService:
+        return RecordedK1SourceAdmissionService(
+            data_dir=self.data_dir,
+            session_store=self._session_store,
+            media_inspector=self._media_inspector,
+            requirements=definition.to_source_admission_requirements(),
+            prepare_media=False,
+        )
+
+    @staticmethod
+    def _bind(
+        definition: RecordedRunDefinition,
+        source: PortableRecordedSourceAdmission,
+    ) -> PortableRecordedRunPreparation:
+        return PortableRecordedRunPreparation(definition=definition, source=source)
+
+
+def _definition_document(definition: RecordedRunDefinition) -> dict[str, object]:
+    return {
+        "setup_id": definition.setup_id,
+        "definition_id": definition.definition_id,
+        "definition_version": definition.definition_version,
+        "definition_sha256": definition.definition_sha256,
+        "source_adapter_id": definition.source_adapter_id,
+        "source_adapter_version": definition.source_adapter_version,
+        "source_adapter_sha256": definition.source_adapter_sha256,
+        "executor_release_id": definition.executor_release_id,
+        "executor_release_sha256": definition.executor_release_sha256,
+        "executor_image_sha256": definition.executor_image_sha256,
+        "model_release_ids": list(definition.model_release_ids),
+        "model_manifest_sha256": definition.model_manifest_sha256,
+        "resource_profile_id": definition.resource_profile_id,
+        "resource_profile_sha256": definition.resource_profile_sha256,
+        "checkpoint_policy": definition.checkpoint_policy,
+        "allowed_checkpoints": list(definition.allowed_checkpoints),
+    }
+
+
+def _canonical_json(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        allow_nan=False,
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode("utf-8")
+
+
+def _sha256(value: object) -> str:
+    return hashlib.sha256(_canonical_json(value)).hexdigest()
+
+
+def _digest(value: object, label: str) -> str:
+    return _pattern(value, _SHA256, label)
+
+
+def _pattern(value: object, pattern: re.Pattern[str], label: str) -> str:
+    if not isinstance(value, str) or pattern.fullmatch(value) is None:
+        raise ValueError(f"{label} is invalid")
+    return value

src/k1link/observatory/portable_setup_projection.py

@@ -0,0 +1,234 @@
+"""UI-ready Observatory projection for the portable LAB V1 definition.
+
+This module deliberately does not extend the legacy setup registry and does
+not submit work.  It projects two independent facts for one selected source:
+
+* the result of a lightweight recorded-source capability probe;
+* the executor state sealed by the portable RunDefinition;
+
+Keeping those facts separate prevents a compatible new recording from being
+described as incompatible merely because the executor is not installed yet.
+Historical vegetation-shadow results belong only to the legacy catalog and
+never become exact results of the generic portable v2 definition.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Final, Protocol, runtime_checkable
+
+from k1link.observatory.portable_run_definitions import (
+    PortableRunDefinition,
+    PortableRunDefinitionRegistry,
+)
+from k1link.observatory.source_admission import (
+    PortableRecordedSourceCapability,
+    PortableSourceAdmissionError,
+)
+from k1link.sessions.models import SessionSummary
+
+PORTABLE_LABORATORY_SETUP_CATALOG_SCHEMA: Final = (
+    "missioncore.observatory-portable-setup-catalog/v2"
+)
+PORTABLE_LAB_V1_SETUP_ID: Final = "lab-v1-eomt-ddrnet-portable-v1"
+PORTABLE_LAB_V1_DISPLAY_NAME: Final = "LAB V1 · EoMT Cityscapes Large 1024 + DDRNet-39"
+
+_SOURCE_ID: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+_OBSERVATION_ONLY_AUTHORITY: Final = {
+    "commands_enabled": False,
+    "actuation_allowed": False,
+    "navigation_or_safety_accepted": False,
+    "production_accepted": False,
+}
+
+_MODEL_PRESENTATION: Final = {
+    "eomt-cityscapes-large-1024-v1": "EoMT Cityscapes Large 1024",
+    "lab-v1-ddrnet-39-goose-fine-64-v1": "DDRNet-39",
+}
+
+
+class PortableSetupProjectionError(RuntimeError):
+    """The portable setup cannot be projected without weakening its contract."""
+
+
+@runtime_checkable
+class PortableSourceCapabilityProbeService(Protocol):
+    """A definition-bound lightweight source-capability service."""
+
+    def probe(self, source_session_id: str) -> PortableRecordedSourceCapability:
+        """Probe one source without preparing media or persisting documents."""
+
+
+type PortableSourceCapabilityProbe = (
+    Callable[[str], PortableRecordedSourceCapability] | PortableSourceCapabilityProbeService
+)
+
+
+@dataclass(frozen=True, slots=True)
+class _SourceCompatibility:
+    compatible: bool
+    capability: PortableRecordedSourceCapability | None
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "outcome": "pass" if self.compatible else "blocked",
+            "compatible": self.compatible,
+            "reason": (
+                "Запись соответствует требованиям EoMT + DDRNet."
+                if self.compatible
+                else "Запись не соответствует требованиям EoMT + DDRNet."
+            ),
+        }
+
+
+class PortableLabV1SetupProjector:
+    """Project the generic portable LAB V1 setup for one selected source."""
+
+    def __init__(
+        self,
+        *,
+        registry: PortableRunDefinitionRegistry,
+        capability_probe: PortableSourceCapabilityProbe,
+    ) -> None:
+        self._definition = _resolve_lab_v1_definition(registry)
+        if not isinstance(
+            capability_probe,
+            PortableSourceCapabilityProbeService,
+        ) and not callable(capability_probe):
+            raise PortableSetupProjectionError("portable source capability probe is unavailable")
+        self._capability_probe = capability_probe
+        _validate_model_presentation(self._definition)
+        if self._definition.authority.as_dict() != _OBSERVATION_ONLY_AUTHORITY:
+            raise PortableSetupProjectionError("portable LAB V1 authority is not observation-only")
+
+    def catalog(self, source: SessionSummary) -> dict[str, object]:
+        """Return a one-setup v2 catalog projection for ``source``."""
+
+        return {
+            "schema_version": PORTABLE_LABORATORY_SETUP_CATALOG_SCHEMA,
+            "source_session_id": source.session_id,
+            "setups": [self.project(source)],
+            "authority": dict(_OBSERVATION_ONLY_AUTHORITY),
+        }
+
+    def project(self, source: SessionSummary) -> dict[str, object]:
+        """Return a strict, observation-only setup projection."""
+
+        _validate_source_id(source.session_id)
+        compatibility = self._probe_source(source.session_id)
+        definition = self._definition
+        executor = definition.executor
+        return {
+            "setup_id": definition.setup_id,
+            "display_name": PORTABLE_LAB_V1_DISPLAY_NAME,
+            "description": (
+                "Проверка записанной K1-сессии моделями EoMT и DDRNet; только наблюдение."
+            ),
+            "origin": "portable-definition",
+            "source_requirements": definition.source_requirements.as_dict(),
+            "run_definition": {
+                "definition_id": definition.definition_id,
+                "version": definition.version,
+                "definition_sha256": definition.definition_sha256,
+                "result_schema": definition.result_contract.result_schema,
+                "result_kind": definition.result_contract.result_kind,
+                "models": _project_models(definition),
+            },
+            "source_compatibility": compatibility.as_dict(),
+            "executor": {
+                "contour_id": executor.contour_id,
+                "state": executor.state,
+                "ready": executor.ready,
+                "reason": executor.reason,
+            },
+            "existing_results": [],
+            "preflight": {
+                "outcome": "blocked",
+                "action": "blocked",
+                "reason": self._preflight_reason(compatibility),
+                "submission_allowed": False,
+                "existing_result_ids": [],
+            },
+            "authority": dict(_OBSERVATION_ONLY_AUTHORITY),
+        }
+
+    def _probe_source(self, source_session_id: str) -> _SourceCompatibility:
+        try:
+            if isinstance(
+                self._capability_probe,
+                PortableSourceCapabilityProbeService,
+            ):
+                capability = self._capability_probe.probe(source_session_id)
+            else:
+                capability = self._capability_probe(source_session_id)
+        except PortableSourceAdmissionError:
+            return _SourceCompatibility(compatible=False, capability=None)
+        if not isinstance(capability, PortableRecordedSourceCapability):
+            raise PortableSetupProjectionError("capability probe returned an invalid result")
+        if capability.source_session_id != source_session_id:
+            raise PortableSetupProjectionError(
+                "capability probe is bound to another source session"
+            )
+        if capability.source_adapter_sha256 != self._definition.source_adapter.contract_sha256:
+            raise PortableSetupProjectionError("capability probe uses another source adapter")
+        return _SourceCompatibility(compatible=True, capability=capability)
+
+    def _preflight_reason(self, compatibility: _SourceCompatibility) -> str:
+        if not compatibility.compatible:
+            return "Запись не соответствует требованиям этого сетапа."
+        if not self._definition.executor.ready:
+            return "Вычислительный контур LAB V1 пока недоступен."
+        return (
+            "Server-side проверка definition/check SHA и постановка portable "
+            "LAB V1 в очередь пока недоступны."
+        )
+
+
+def _resolve_lab_v1_definition(
+    registry: PortableRunDefinitionRegistry,
+) -> PortableRunDefinition:
+    matching = tuple(
+        definition
+        for definition in registry.definitions
+        if definition.setup_id == PORTABLE_LAB_V1_SETUP_ID
+    )
+    if len(matching) != 1:
+        raise PortableSetupProjectionError("portable LAB V1 definition is unavailable or ambiguous")
+    return matching[0]
+
+
+def _validate_model_presentation(definition: PortableRunDefinition) -> None:
+    releases = {model.release_id: model for model in definition.models}
+    if set(releases) != set(_MODEL_PRESENTATION):
+        raise PortableSetupProjectionError(
+            "portable LAB V1 model set does not match its presentation contract"
+        )
+    eomt = releases["eomt-cityscapes-large-1024-v1"]
+    ddrnet = releases["lab-v1-ddrnet-39-goose-fine-64-v1"]
+    if (
+        eomt.model_id != "tue-mps/cityscapes_semantic_eomt_large_1024"
+        or ddrnet.architecture != "ddrnet_39"
+    ):
+        raise PortableSetupProjectionError(
+            "portable LAB V1 model identities do not match their presentation"
+        )
+
+
+def _project_models(definition: PortableRunDefinition) -> list[dict[str, object]]:
+    by_release = {model.release_id: model for model in definition.models}
+    return [
+        {
+            "name": _MODEL_PRESENTATION[release_id],
+            "release_id": release_id,
+            "model_id": by_release[release_id].model_id,
+            "architecture": by_release[release_id].architecture,
+        }
+        for release_id in _MODEL_PRESENTATION
+    ]
+
+
+def _validate_source_id(source_session_id: str) -> None:
+    if _SOURCE_ID.fullmatch(source_session_id) is None:
+        raise PortableSetupProjectionError("source session id is invalid")

src/k1link/observatory/recorded_jobs.py

@@ -34,16 +34,10 @@ from uuid import uuid4
 from k1link.artifacts import utc_now_iso
 
 OBSERVATORY_RECORDED_JOB_SCHEMA: Final = "missioncore.observatory-recorded-job/v1"
-OBSERVATORY_RECORDED_JOB_REQUEST_SCHEMA: Final = (
-    "missioncore.observatory-recorded-job-request/v1"
-)
-OBSERVATORY_RECORDED_CLAIM_SCHEMA: Final = (
-    "missioncore.observatory-recorded-job-claim/v1"
-)
+OBSERVATORY_RECORDED_JOB_REQUEST_SCHEMA: Final = "missioncore.observatory-recorded-job-request/v1"
+OBSERVATORY_RECORDED_CLAIM_SCHEMA: Final = "missioncore.observatory-recorded-job-claim/v1"
 OBSERVATORY_LIVE_LEASE_SCHEMA: Final = "missioncore.observatory-live-k1-lease/v1"
-OBSERVATORY_LIVE_LEASE_REQUEST_SCHEMA: Final = (
-    "missioncore.observatory-live-k1-lease-request/v1"
-)
+OBSERVATORY_LIVE_LEASE_REQUEST_SCHEMA: Final = "missioncore.observatory-live-k1-lease-request/v1"
 RECORDED_JOB_DATABASE_NAME: Final = "observatory-recorded-jobs.sqlite3"
 MAX_RECORDED_JOBS: Final = 10_000
 MAX_RECORDED_CLAIM_RECEIPTS: Final = 50_000
@@ -67,9 +61,7 @@ type RecordedJobState = Literal[
     "reconciliation-required",
 ]
 type CheckpointPolicy = Literal["cooperative", "non-checkpointable"]
-type LiveLeaseState = Literal[
-    "pending", "active", "completed", "failed", "cancelled"
-]
+type LiveLeaseState = Literal["pending", "active", "completed", "failed", "cancelled"]
 type LiveTerminalOutcome = Literal["completed", "failed", "cancelled"]
 
 _JOB_ID = re.compile(r"^observatory-run-[a-f0-9]{32}$")
@@ -80,9 +72,7 @@ _SESSION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
 _IDENTIFIER = re.compile(r"^[a-z][a-z0-9-]{2,95}$")
 _CHECKPOINT_ID = re.compile(r"^[a-z][a-z0-9-]{2,95}$")
 _SHA256 = re.compile(r"^[a-f0-9]{64}$")
-_TERMINAL_STATES: Final = frozenset(
-    {"succeeded", "failed", "reconciliation-required"}
-)
+_TERMINAL_STATES: Final = frozenset({"succeeded", "failed", "reconciliation-required"})
 _OPEN_LIVE_STATES: Final = frozenset({"pending", "active"})
 _AUTHORITY: Final = {
     "commands_enabled": False,
@@ -299,8 +289,7 @@ class RecordedRunDefinitionRegistry:
 
     def __post_init__(self) -> None:
         keys = [
-            (definition.setup_id, definition.definition_sha256)
-            for definition in self.definitions
+            (definition.setup_id, definition.definition_sha256) for definition in self.definitions
         ]
         version_keys = [
             (definition.definition_id, definition.definition_version)
@@ -358,9 +347,7 @@ class ObservatoryRecordedJobIntent:
                 "source_session_id": self.source_session_id,
                 "source_catalog_sha256": self.source_catalog_sha256,
                 "source_bundle_sha256": self.source_bundle_sha256,
-                "source_capability_manifest_sha256": (
-                    self.source_capability_manifest_sha256
-                ),
+                "source_capability_manifest_sha256": (self.source_capability_manifest_sha256),
                 "setup_id": self.setup_id,
                 "definition_sha256": self.definition_sha256,
             }
@@ -471,18 +458,12 @@ class ObservatoryRecordedJob:
                 "recorded-job preemption marker is invalid"
             )
         if self.claim_generation < 0:
-            raise ObservatoryRecordedQueueIntegrityError(
-                "recorded-job claim generation is invalid"
-            )
+            raise ObservatoryRecordedQueueIntegrityError("recorded-job claim generation is invalid")
         _validate_optional_pattern(self.active_claim_token, _TOKEN, "active claim token")
         _validate_optional_pattern(self.active_claimant_id, _IDENTIFIER, "claimant id")
-        _validate_optional_pattern(
-            self.last_checkpoint_id, _CHECKPOINT_ID, "last checkpoint id"
-        )
+        _validate_optional_pattern(self.last_checkpoint_id, _CHECKPOINT_ID, "last checkpoint id")
         if not isinstance(self.restart_from_zero, bool):
-            raise ObservatoryRecordedQueueIntegrityError(
-                "recorded-job restart marker is invalid"
-            )
+            raise ObservatoryRecordedQueueIntegrityError("recorded-job restart marker is invalid")
         _validate_optional_pattern(
             self.preemption_receipt_sha256,
             _SHA256,
@@ -513,9 +494,7 @@ class ObservatoryRecordedJob:
                 "session_id": self.source_session_id,
                 "catalog_sha256": self.source_catalog_sha256,
                 "bundle_sha256": self.source_bundle_sha256,
-                "capability_manifest_sha256": (
-                    self.source_capability_manifest_sha256
-                ),
+                "capability_manifest_sha256": (self.source_capability_manifest_sha256),
                 "adapter": {
                     "adapter_id": self.source_adapter_id,
                     "version": self.source_adapter_version,
@@ -661,9 +640,7 @@ class ObservatoryLiveLease:
             and self.terminal_request_sha256 is not None
             and self.terminated_at_utc is not None
         ):
-            raise ObservatoryRecordedQueueIntegrityError(
-                "live terminal identity is incomplete"
-            )
+            raise ObservatoryRecordedQueueIntegrityError("live terminal identity is incomplete")
 
     def as_dict(self) -> dict[str, object]:
         return {
@@ -730,9 +707,7 @@ class ObservatoryNonCheckpointableCancellationRequest:
                 "schema_version": OBSERVATORY_LIVE_LEASE_REQUEST_SCHEMA,
                 "cancellation_request_id": self.cancellation_request_id,
                 "job_id": self.job_id,
-                "claim_token_sha256": hashlib.sha256(
-                    self.claim_token.encode()
-                ).hexdigest(),
+                "claim_token_sha256": hashlib.sha256(self.claim_token.encode()).hexdigest(),
                 "claim_generation": self.claim_generation,
                 "live_trigger_id": self.live_trigger_id,
                 "executor_release_sha256": self.executor_release_sha256,
@@ -781,11 +756,7 @@ class ObservatoryNonCheckpointableCancellationReceipt:
         _validate_pattern(self.cancellation_id, _IDEMPOTENCY_KEY, "cancellation id")
         _validate_digest(self.request_sha256, "cancellation request sha256")
         _validate_digest(self.receipt_sha256, "cancellation receipt sha256")
-        if not (
-            self.resources_released
-            and self.staging_discarded
-            and self.restart_from_zero
-        ):
+        if not (self.resources_released and self.staging_discarded and self.restart_from_zero):
             raise ValueError(
                 "cancellation receipt must release resources, discard staging, "
                 "and require restart from zero"
@@ -819,17 +790,13 @@ class ObservatoryRecordedJobQueue:
         *,
         definitions: RecordedRunDefinitionRegistry,
         clock: Callable[[], str] = utc_now_iso,
-        non_checkpointable_preemptor: (
-            ObservatoryNonCheckpointablePreemptor | None
-        ) = None,
+        non_checkpointable_preemptor: (ObservatoryNonCheckpointablePreemptor | None) = None,
         max_jobs: int = MAX_RECORDED_JOBS,
         max_claim_receipts: int = MAX_RECORDED_CLAIM_RECEIPTS,
         max_live_leases: int = MAX_LIVE_LEASES,
     ) -> None:
         _validate_quota(max_jobs, MAX_RECORDED_JOBS, "recorded job")
-        _validate_quota(
-            max_claim_receipts, MAX_RECORDED_CLAIM_RECEIPTS, "claim receipt"
-        )
+        _validate_quota(max_claim_receipts, MAX_RECORDED_CLAIM_RECEIPTS, "claim receipt")
         _validate_quota(max_live_leases, MAX_LIVE_LEASES, "live lease")
         self.data_dir = data_dir.expanduser().resolve()
         self.database_path = self.data_dir / RECORDED_JOB_DATABASE_NAME
@@ -842,6 +809,15 @@ class ObservatoryRecordedJobQueue:
         self._lock = threading.RLock()
         self._initialize()
 
+    def resolve_definition(
+        self,
+        setup_id: str,
+        definition_sha256: str,
+    ) -> RecordedRunDefinition:
+        """Expose the immutable server allowlist without exposing queue internals."""
+
+        return self._definitions.resolve(setup_id, definition_sha256)
+
     def submit(
         self,
         intent: ObservatoryRecordedJobIntent,
@@ -875,9 +851,7 @@ class ObservatoryRecordedJobQueue:
                 limit=self._max_jobs,
                 label="recorded job",
             )
-            definition = self._definitions.resolve(
-                intent.setup_id, intent.definition_sha256
-            )
+            definition = self._definitions.resolve(intent.setup_id, intent.definition_sha256)
             now = self._timestamp()
             job_id = f"observatory-run-{uuid4().hex}"
             identity_sha256 = _job_identity_sha256(intent, definition)
@@ -973,8 +947,7 @@ class ObservatoryRecordedJobQueue:
         request_sha256 = _claim_request_sha256(claimant_id, claim_request_id)
         with self._transaction() as connection:
             receipt = connection.execute(
-                "SELECT * FROM observatory_recorded_claim_receipts "
-                "WHERE claim_request_id = ?",
+                "SELECT * FROM observatory_recorded_claim_receipts WHERE claim_request_id = ?",
                 (claim_request_id,),
             ).fetchone()
             if receipt is not None:
@@ -1062,11 +1035,7 @@ class ObservatoryRecordedJobQueue:
                 raise ObservatoryRecordedQueueConflictError(
                     f"cannot start recorded job from {job.state}"
                 )
-            state = (
-                "paused"
-                if self._open_live_lease_row(connection) is not None
-                else "running"
-            )
+            state = "paused" if self._open_live_lease_row(connection) is not None else "running"
             connection.execute(
                 "UPDATE observatory_recorded_jobs SET state = ?, "
                 "preemption_requested = ?, updated_at_utc = ? WHERE job_id = ?",
@@ -1179,14 +1148,11 @@ class ObservatoryRecordedJobQueue:
         with self._read_connection() as connection:
             return self._get_job(connection, job_id)
 
-    def get_by_idempotency_key(
-        self, idempotency_key: str
-    ) -> ObservatoryRecordedJob:
+    def get_by_idempotency_key(self, idempotency_key: str) -> ObservatoryRecordedJob:
         _validate_pattern(idempotency_key, _IDEMPOTENCY_KEY, "idempotency key")
         with self._read_connection() as connection:
             row = connection.execute(
-                "SELECT * FROM observatory_recorded_jobs "
-                "WHERE idempotency_key = ?",
+                "SELECT * FROM observatory_recorded_jobs WHERE idempotency_key = ?",
                 (idempotency_key,),
             ).fetchone()
         if row is None:
@@ -1226,9 +1192,7 @@ class ObservatoryRecordedJobQueue:
             ).fetchall()
         return tuple(_job_from_row(row) for row in rows)
 
-    def request_live(
-        self, intent: ObservatoryLiveLeaseIntent
-    ) -> tuple[ObservatoryLiveLease, bool]:
+    def request_live(self, intent: ObservatoryLiveLeaseIntent) -> tuple[ObservatoryLiveLease, bool]:
         """Close recorded admission without allowing a monolith to delay live K1."""
 
         created = False
@@ -1357,9 +1321,7 @@ class ObservatoryRecordedJobQueue:
                     "scheduler did not release non-checkpointable replay resources"
                 ) from exc
             if (
-                not isinstance(
-                    receipt, ObservatoryNonCheckpointableCancellationReceipt
-                )
+                not isinstance(receipt, ObservatoryNonCheckpointableCancellationReceipt)
                 or receipt.request_sha256 != request.request_sha256
             ):
                 with self._transaction() as connection:
@@ -1423,9 +1385,7 @@ class ObservatoryRecordedJobQueue:
         """Release live priority only on an exact explicit terminal trigger."""
 
         _validate_pattern(lease_id, _LEASE_ID, "live lease id")
-        _validate_pattern(
-            terminal_trigger_id, _IDEMPOTENCY_KEY, "live terminal trigger id"
-        )
+        _validate_pattern(terminal_trigger_id, _IDEMPOTENCY_KEY, "live terminal trigger id")
         if outcome not in ("completed", "failed", "cancelled"):
             raise ValueError("live terminal outcome is invalid")
         terminal_request_sha256 = _sha256(
@@ -1500,9 +1460,7 @@ class ObservatoryRecordedJobQueue:
         with self._read_connection() as connection:
             row = self._open_live_lease_row(connection)
         if row is None:
-            return ObservatoryRecordedAdmissionGate(
-                blocked=False, lease_id=None, lease_state=None
-            )
+            return ObservatoryRecordedAdmissionGate(blocked=False, lease_id=None, lease_state=None)
         state = str(row["state"])
         if state == "pending":
             lease_state: Literal["pending", "active"] = "pending"
@@ -1556,8 +1514,7 @@ class ObservatoryRecordedJobQueue:
                     f"cannot terminate recorded job from {job.state}"
                 )
             if state == "succeeded" and (
-                job.preemption_requested
-                or self._open_live_lease_row(connection) is not None
+                job.preemption_requested or self._open_live_lease_row(connection) is not None
             ):
                 raise ObservatoryRecordedQueueConflictError(
                     "cannot publish recorded success while live preemption is open"
@@ -1599,14 +1556,10 @@ class ObservatoryRecordedJobQueue:
             job=self._get_job(connection, job_id),
         )
 
-    def _require_active_claim(
-        self, job: ObservatoryRecordedJob, claim_token: str
-    ) -> None:
+    def _require_active_claim(self, job: ObservatoryRecordedJob, claim_token: str) -> None:
         _validate_pattern(claim_token, _TOKEN, "claim token")
         if job.active_claim_token != claim_token:
-            raise ObservatoryRecordedQueueStaleClaimError(
-                "recorded-job claim token is stale"
-            )
+            raise ObservatoryRecordedQueueStaleClaimError("recorded-job claim token is stale")
 
     def _cancellation_request(
         self,
@@ -1642,8 +1595,7 @@ class ObservatoryRecordedJobQueue:
     ) -> None:
         with self._transaction() as connection:
             row = connection.execute(
-                "SELECT * FROM observatory_recorded_preemptions "
-                "WHERE cancellation_request_id = ?",
+                "SELECT * FROM observatory_recorded_preemptions WHERE cancellation_request_id = ?",
                 (request.cancellation_request_id,),
             ).fetchone()
             if row is None:
@@ -1704,9 +1656,7 @@ class ObservatoryRecordedJobQueue:
                 (receipt.receipt_sha256, now, request.cancellation_request_id),
             )
 
-    def _get_job(
-        self, connection: sqlite3.Connection, job_id: str
-    ) -> ObservatoryRecordedJob:
+    def _get_job(self, connection: sqlite3.Connection, job_id: str) -> ObservatoryRecordedJob:
         row = connection.execute(
             "SELECT * FROM observatory_recorded_jobs WHERE job_id = ?", (job_id,)
         ).fetchone()
@@ -1724,17 +1674,12 @@ class ObservatoryRecordedJobQueue:
             raise ObservatoryRecordedQueueNotFoundError(lease_id)
         return _live_lease_from_row(row)
 
-    def _open_live_lease_row(
-        self, connection: sqlite3.Connection
-    ) -> sqlite3.Row | None:
+    def _open_live_lease_row(self, connection: sqlite3.Connection) -> sqlite3.Row | None:
         rows = connection.execute(
-            "SELECT * FROM observatory_live_leases "
-            "WHERE state IN ('pending', 'active') LIMIT 2"
+            "SELECT * FROM observatory_live_leases WHERE state IN ('pending', 'active') LIMIT 2"
         ).fetchall()
         if len(rows) > 1:
-            raise ObservatoryRecordedQueueIntegrityError(
-                "multiple live K1 leases are open"
-            )
+            raise ObservatoryRecordedQueueIntegrityError("multiple live K1 leases are open")
         return None if not rows else rows[0]
 
     def _require_capacity(
@@ -1788,9 +1733,7 @@ class ObservatoryRecordedJobQueue:
                 "SELECT name FROM pragma_table_info(?)", (table,)
             ).fetchall()
             if len(columns) != column_count:
-                raise ObservatoryRecordedQueueIntegrityError(
-                    f"{table} schema is incompatible"
-                )
+                raise ObservatoryRecordedQueueIntegrityError(f"{table} schema is incompatible")
         indexes = connection.execute(
             "SELECT name FROM sqlite_master WHERE type = 'index' "
             "AND name = 'observatory_one_open_live_lease'"
@@ -1816,14 +1759,10 @@ class ObservatoryRecordedJobQueue:
             ),
         ):
             count = len(
-                connection.execute(
-                    f"SELECT 1 FROM {table} LIMIT ?", (limit + 1,)
-                ).fetchall()
+                connection.execute(f"SELECT 1 FROM {table} LIMIT ?", (limit + 1,)).fetchall()
             )
             if count > limit:
-                raise ObservatoryRecordedQueueCapacityError(
-                    f"{label} quota is exceeded"
-                )
+                raise ObservatoryRecordedQueueCapacityError(f"{label} quota is exceeded")
 
     def _validate_storage_paths(self, *, require_database: bool = False) -> None:
         paths = (
@@ -1834,9 +1773,7 @@ class ObservatoryRecordedJobQueue:
         total_bytes = 0
         for path in paths:
             if path.is_symlink() or (path.exists() and not path.is_file()):
-                raise ObservatoryRecordedQueueIntegrityError(
-                    "recorded-job queue path is unsafe"
-                )
+                raise ObservatoryRecordedQueueIntegrityError("recorded-job queue path is unsafe")
             if path.exists():
                 total_bytes += path.stat().st_size
         if require_database and not self.database_path.is_file():
@@ -1874,9 +1811,7 @@ class ObservatoryRecordedJobQueue:
         except ObservatoryRecordedQueueError:
             raise
         except (OSError, sqlite3.Error) as exc:
-            raise ObservatoryRecordedQueueIntegrityError(
-                "recorded-job queue read failed"
-            ) from exc
+            raise ObservatoryRecordedQueueIntegrityError("recorded-job queue read failed") from exc
 
     @contextmanager
     def _connect(self) -> Iterator[sqlite3.Connection]:
@@ -1889,9 +1824,7 @@ class ObservatoryRecordedJobQueue:
         try:
             connection.execute("PRAGMA foreign_keys = ON")
             connection.execute("PRAGMA synchronous = FULL")
-            connection.execute(
-                f"PRAGMA busy_timeout = {_SQLITE_BUSY_TIMEOUT_MILLISECONDS}"
-            )
+            connection.execute(f"PRAGMA busy_timeout = {_SQLITE_BUSY_TIMEOUT_MILLISECONDS}")
             connection.execute("PRAGMA journal_mode = WAL")
             yield connection
         finally:
@@ -1944,9 +1877,7 @@ def _job_from_row(row: sqlite3.Row) -> ObservatoryRecordedJob:
             source_session_id=row["source_session_id"],
             source_catalog_sha256=row["source_catalog_sha256"],
             source_bundle_sha256=row["source_bundle_sha256"],
-            source_capability_manifest_sha256=row[
-                "source_capability_manifest_sha256"
-            ],
+            source_capability_manifest_sha256=row["source_capability_manifest_sha256"],
             setup_id=row["setup_id"],
             definition_id=row["definition_id"],
             definition_version=row["definition_version"],
@@ -1982,9 +1913,7 @@ def _job_from_row(row: sqlite3.Row) -> ObservatoryRecordedJob:
             updated_at_utc=row["updated_at_utc"],
         )
     except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
-        raise ObservatoryRecordedQueueIntegrityError(
-            "stored recorded job is invalid"
-        ) from exc
+        raise ObservatoryRecordedQueueIntegrityError("stored recorded job is invalid") from exc
 
 
 def _live_lease_from_row(row: sqlite3.Row) -> ObservatoryLiveLease:
@@ -2005,9 +1934,7 @@ def _live_lease_from_row(row: sqlite3.Row) -> ObservatoryLiveLease:
             terminated_at_utc=row["terminated_at_utc"],
         )
     except (IndexError, KeyError, TypeError, ValueError) as exc:
-        raise ObservatoryRecordedQueueIntegrityError(
-            "stored live K1 lease is invalid"
-        ) from exc
+        raise ObservatoryRecordedQueueIntegrityError("stored live K1 lease is invalid") from exc
 
 
 def _job_identity_sha256(
@@ -2019,9 +1946,7 @@ def _job_identity_sha256(
             "source_session_id": intent.source_session_id,
             "source_catalog_sha256": intent.source_catalog_sha256,
             "source_bundle_sha256": intent.source_bundle_sha256,
-            "source_capability_manifest_sha256": (
-                intent.source_capability_manifest_sha256
-            ),
+            "source_capability_manifest_sha256": (intent.source_capability_manifest_sha256),
             "setup_id": definition.setup_id,
             "definition_id": definition.definition_id,
             "definition_version": definition.definition_version,
@@ -2100,11 +2025,7 @@ def _sha256(value: object) -> str:
 
 
 def _validate_quota(value: object, maximum: int, label: str) -> None:
-    if (
-        not isinstance(value, int)
-        or isinstance(value, bool)
-        or not 1 <= value <= maximum
-    ):
+    if not isinstance(value, int) or isinstance(value, bool) or not 1 <= value <= maximum:
         raise ValueError(f"{label} quota is invalid")
 
 
@@ -2122,9 +2043,7 @@ def _validate_pattern(value: object, pattern: re.Pattern[str], label: str) -> No
         raise ValueError(f"{label} is invalid")
 
 
-def _validate_optional_pattern(
-    value: object | None, pattern: re.Pattern[str], label: str
-) -> None:
+def _validate_optional_pattern(value: object | None, pattern: re.Pattern[str], label: str) -> None:
     if value is not None:
         _validate_pattern(value, pattern, label)
 
@@ -2146,11 +2065,7 @@ def _validate_timestamp(value: object, label: str) -> None:
         parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
     except ValueError as exc:
         raise ValueError(f"{label} is invalid") from exc
-    if (
-        parsed.tzinfo is None
-        or parsed.utcoffset() != timedelta(0)
-        or not value.endswith("Z")
-    ):
+    if parsed.tzinfo is None or parsed.utcoffset() != timedelta(0) or not value.endswith("Z"):
         raise ValueError(f"{label} must use UTC")
 
 

src/k1link/observatory/source_admission.py

@@ -0,0 +1,978 @@
+"""Capability-based admission for portable recorded K1 laboratory runs.
+
+The portable setup identity is deliberately independent from a session name.
+This module binds one concrete SessionStore snapshot to an allowlisted K1
+source profile and emits path-free, content-addressed documents for the Worker
+boundary.  It does not enqueue work or expose host filesystem locations.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import secrets
+import stat
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+from k1link.sessions.media import (
+    CAMERA_ARCHIVE_SCHEMA,
+    MAX_MEDIA_SUMMARY_BYTES,
+    MAX_SAFE_INTEGER,
+    RecordedMediaEpoch,
+    RecordedMediaInspector,
+    RecordedMediaManifest,
+)
+from k1link.sessions.models import (
+    RecordedMediaArtifact,
+    ReplayCommand,
+    SessionArtifact,
+    SessionDetail,
+    SessionSource,
+)
+from k1link.sessions.store import SessionStore
+
+PORTABLE_SOURCE_BUNDLE_SCHEMA: Final = "missioncore.portable-recorded-source-bundle/v1"
+PORTABLE_SOURCE_CAPABILITY_SCHEMA: Final = "missioncore.portable-recorded-source-capability/v1"
+PORTABLE_SOURCE_ADAPTER_SCHEMA: Final = "missioncore.portable-source-adapter/v1"
+PORTABLE_SOURCE_DOCUMENT_DIRECTORY: Final = "observatory-portable-source-contracts"
+
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_IDENTIFIER = re.compile(r"^[a-z][a-z0-9.-]{2,127}$")
+_SOURCE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+_MEDIA_TYPE = re.compile(r'^[a-z0-9.+-]+/[a-z0-9.+-]+(?:; codecs="[A-Za-z0-9.]+")?$')
+_EPOCH = re.compile(r"^epoch-([1-9][0-9]*)$")
+_ALLOWED_MODALITIES: Final = frozenset({"point-cloud", "trajectory", "video"})
+_SEALED_ARTIFACT_STATES: Final = frozenset({"verified", "validated-structure"})
+_AUTHORITY: Final = {
+    "commands_enabled": False,
+    "actuation_allowed": False,
+    "navigation_or_safety_accepted": False,
+    "production_accepted": False,
+}
+
+
+class PortableSourceAdmissionError(RuntimeError):
+    """A recorded session does not satisfy the portable source contract."""
+
+
+class PortableSourceAdmissionIntegrityError(PortableSourceAdmissionError):
+    """A catalog, media, or persisted source identity changed."""
+
+
+class PortableSourceNotPreparedError(PortableSourceAdmissionError):
+    """A compatible source has no previously validated media sidecar."""
+
+
+class PortableSourceAdmissionStaleError(PortableSourceAdmissionIntegrityError):
+    """The catalog changed across a checked admission boundary."""
+
+
+@dataclass(frozen=True, slots=True)
+class RecordedK1SourceRequirements:
+    """Typed matcher owned by a portable RunDefinition.
+
+    ``camera_init_sha256`` is the exact ISO-BMFF initialization segment for the
+    admitted codec/resolution profile.  Width and height are asserted by that
+    immutable media-profile attestation; arbitrary init segments are rejected.
+    Calibration remains an external, digest-bound rig profile until recordings
+    carry their own calibration snapshot.
+    """
+
+    adapter_id: str
+    adapter_version: int
+    plugin_id: str
+    archive_id: str
+    required_modalities: tuple[str, ...]
+    camera_source_id: str
+    camera_semantic_channel_id: str
+    camera_media_type: str
+    camera_init_sha256: str
+    expected_width: int
+    expected_height: int
+    calibration_slot: str
+    calibration_sha256: str
+    require_single_camera_epoch: bool = True
+
+    def __post_init__(self) -> None:
+        _pattern(self.adapter_id, _IDENTIFIER, "source adapter id")
+        if (
+            not isinstance(self.adapter_version, int)
+            or isinstance(self.adapter_version, bool)
+            or self.adapter_version < 1
+        ):
+            raise ValueError("source adapter version is invalid")
+        _pattern(self.plugin_id, _IDENTIFIER, "device plugin id")
+        _pattern(self.archive_id, _IDENTIFIER, "archive id")
+        if (
+            not self.required_modalities
+            or len(set(self.required_modalities)) != len(self.required_modalities)
+            or not set(self.required_modalities).issubset(_ALLOWED_MODALITIES)
+            or "video" not in self.required_modalities
+        ):
+            raise ValueError("portable source modalities are invalid")
+        _pattern(self.camera_source_id, _SOURCE_ID, "camera source id")
+        _pattern(
+            self.camera_semantic_channel_id,
+            _IDENTIFIER,
+            "camera semantic channel id",
+        )
+        _pattern(self.camera_media_type, _MEDIA_TYPE, "camera media type")
+        _digest(self.camera_init_sha256, "camera init sha256")
+        if (
+            not isinstance(self.expected_width, int)
+            or isinstance(self.expected_width, bool)
+            or not isinstance(self.expected_height, int)
+            or isinstance(self.expected_height, bool)
+            or not 1 <= self.expected_width <= 16_384
+            or not 1 <= self.expected_height <= 16_384
+        ):
+            raise ValueError("camera dimensions are invalid")
+        _pattern(self.calibration_slot, _SOURCE_ID, "calibration slot")
+        _digest(self.calibration_sha256, "calibration sha256")
+        if self.require_single_camera_epoch is not True:
+            raise ValueError("portable source v1 requires one camera epoch")
+
+    def adapter_document(self) -> dict[str, object]:
+        return {
+            "schema_version": PORTABLE_SOURCE_ADAPTER_SCHEMA,
+            "adapter_id": self.adapter_id,
+            "version": self.adapter_version,
+            "source": {
+                "plugin_id": self.plugin_id,
+                "archive_id": self.archive_id,
+                "required_modalities": list(self.required_modalities),
+                "camera_source_id": self.camera_source_id,
+                "camera_semantic_channel_id": self.camera_semantic_channel_id,
+            },
+            "camera_profile": {
+                "media_type": self.camera_media_type,
+                "init_sha256": self.camera_init_sha256,
+                "width": self.expected_width,
+                "height": self.expected_height,
+                "attestation": "exact-isobmff-init-sha256",
+            },
+            "calibration": {
+                "slot": self.calibration_slot,
+                "sha256": self.calibration_sha256,
+                "binding": "external-rig-profile",
+            },
+            "camera_epoch_policy": "exactly-one-complete-epoch",
+            "authority": dict(_AUTHORITY),
+        }
+
+    @property
+    def adapter_sha256(self) -> str:
+        return _sha256(self.adapter_document())
+
+
+@dataclass(frozen=True, slots=True)
+class PortableRecordedSourceCapability:
+    """Cheap catalog/summary attestation used by compatibility surfaces.
+
+    This is intentionally not a replay admission.  In particular it carries
+    no timeline or content-addressed source bundle because proving those facts
+    requires the prepared replay sidecar and a full archive validation.
+    """
+
+    source_session_id: str
+    source_catalog_sha256: str
+    source_adapter_sha256: str
+    camera_segment_count: int
+
+    def __post_init__(self) -> None:
+        _pattern(self.source_session_id, _SOURCE_ID, "source session id")
+        _digest(self.source_catalog_sha256, "source catalog sha256")
+        _digest(self.source_adapter_sha256, "source adapter sha256")
+        if (
+            not isinstance(self.camera_segment_count, int)
+            or isinstance(self.camera_segment_count, bool)
+            or not 1 <= self.camera_segment_count <= MAX_SAFE_INTEGER
+        ):
+            raise ValueError("camera segment count is invalid")
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": "missioncore.portable-recorded-source-probe/v1",
+            "source_session_id": self.source_session_id,
+            "source_catalog_sha256": self.source_catalog_sha256,
+            "source_adapter_sha256": self.source_adapter_sha256,
+            "camera_segment_count": self.camera_segment_count,
+            "authority": dict(_AUTHORITY),
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class PortableRecordedSourceAdmission:
+    source_session_id: str
+    source_catalog_sha256: str
+    source_bundle_sha256: str
+    source_capability_manifest_sha256: str
+    source_adapter_sha256: str
+    frame_count: int
+    timeline_start_seconds: float
+    timeline_end_seconds: float
+    camera_generation_sha256: str
+    source_bundle: bytes
+    capability_manifest: bytes
+
+    def __post_init__(self) -> None:
+        _pattern(self.source_session_id, _SOURCE_ID, "source session id")
+        for label, value in (
+            ("source catalog sha256", self.source_catalog_sha256),
+            ("source bundle sha256", self.source_bundle_sha256),
+            (
+                "source capability manifest sha256",
+                self.source_capability_manifest_sha256,
+            ),
+            ("source adapter sha256", self.source_adapter_sha256),
+            ("camera generation sha256", self.camera_generation_sha256),
+        ):
+            _digest(value, label)
+        if self.frame_count < 1 or self.timeline_end_seconds <= self.timeline_start_seconds:
+            raise ValueError("portable source timeline is invalid")
+        if hashlib.sha256(self.source_bundle).hexdigest() != self.source_bundle_sha256:
+            raise ValueError("source bundle bytes do not match their identity")
+        if (
+            hashlib.sha256(self.capability_manifest).hexdigest()
+            != self.source_capability_manifest_sha256
+        ):
+            raise ValueError("capability bytes do not match their identity")
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": "missioncore.portable-recorded-source-admission/v1",
+            "source_session_id": self.source_session_id,
+            "source_catalog_sha256": self.source_catalog_sha256,
+            "source_bundle_sha256": self.source_bundle_sha256,
+            "source_capability_manifest_sha256": (self.source_capability_manifest_sha256),
+            "source_adapter_sha256": self.source_adapter_sha256,
+            "camera": {
+                "generation_sha256": self.camera_generation_sha256,
+                "frame_count": self.frame_count,
+                "timeline_start_seconds": self.timeline_start_seconds,
+                "timeline_end_seconds": self.timeline_end_seconds,
+            },
+            "authority": dict(_AUTHORITY),
+        }
+
+    @property
+    def identity_sha256(self) -> str:
+        """Content identity used to fence check-to-admit transitions."""
+
+        return _sha256(self.as_dict())
+
+
+class RecordedK1SourceAdmissionService:
+    """Bind compatible recorded K1 sessions without trusting their labels."""
+
+    def __init__(
+        self,
+        *,
+        data_dir: Path,
+        session_store: SessionStore,
+        media_inspector: RecordedMediaInspector,
+        requirements: RecordedK1SourceRequirements,
+        prepare_media: bool = True,
+    ) -> None:
+        self.data_dir = data_dir.expanduser().resolve()
+        if self.data_dir != session_store.data_dir:
+            raise ValueError("portable admission and SessionStore roots disagree")
+        self._session_store = session_store
+        self._media_inspector = media_inspector
+        self._prepare_media = prepare_media
+        self.requirements = requirements
+        self._document_root = self.data_dir / PORTABLE_SOURCE_DOCUMENT_DIRECTORY
+
+    def probe(self, source_session_id: str) -> PortableRecordedSourceCapability:
+        """Prove cheap setup compatibility without replay/media preparation.
+
+        The probe reads one atomic catalog snapshot, the catalog's single
+        recorded-video handle, and one bounded canonical camera summary.  It
+        deliberately never calls ``prepare_replay`` or ``RecordedMediaInspector``
+        and never enters the archive's index or segment directory.
+        """
+
+        _pattern(source_session_id, _SOURCE_ID, "source session id")
+        try:
+            detail, catalog_sha256 = self._session_store.get_session_with_catalog_snapshot(
+                source_session_id
+            )
+            recorded_media = self._session_store.list_recorded_media(source_session_id)
+        except Exception as exc:
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded source capability could not be resolved"
+            ) from exc
+        _digest(catalog_sha256, "source catalog sha256")
+        selected_sources = self._verify_catalog(
+            detail,
+            expected_session_id=source_session_id,
+        )
+        media_artifact = self._verify_recorded_media_artifact(
+            detail=detail,
+            camera_source=selected_sources["video"],
+            recorded_media=recorded_media,
+        )
+        segment_count = self._probe_camera_summary(media_artifact)
+        return PortableRecordedSourceCapability(
+            source_session_id=detail.summary.session_id,
+            source_catalog_sha256=catalog_sha256,
+            source_adapter_sha256=self.requirements.adapter_sha256,
+            camera_segment_count=segment_count,
+        )
+
+    def check(self, source_session_id: str) -> PortableRecordedSourceAdmission:
+        return self._prepare(
+            source_session_id,
+            persist=False,
+            prepare_media=False,
+        )
+
+    def admit(
+        self,
+        source_session_id: str,
+        *,
+        expected_admission_sha256: str | None = None,
+    ) -> PortableRecordedSourceAdmission:
+        """Persist only the source identity most recently admitted by a check.
+
+        ``expected_admission_sha256`` is optional for existing internal callers.
+        Portable queue binding supplies it so a changed SessionStore snapshot is
+        rejected before either immutable source document is written.
+        """
+
+        if expected_admission_sha256 is not None:
+            _digest(expected_admission_sha256, "expected source admission sha256")
+        return self._prepare(
+            source_session_id,
+            persist=True,
+            prepare_media=self._prepare_media,
+            expected_admission_sha256=expected_admission_sha256,
+        )
+
+    def _prepare(
+        self,
+        source_session_id: str,
+        *,
+        persist: bool,
+        prepare_media: bool,
+        expected_admission_sha256: str | None = None,
+    ) -> PortableRecordedSourceAdmission:
+        _pattern(source_session_id, _SOURCE_ID, "source session id")
+        try:
+            detail, catalog_sha256 = self._session_store.get_session_with_catalog_snapshot(
+                source_session_id
+            )
+            replay = self._session_store.prepare_replay(source_session_id)
+            recorded_media = self._session_store.list_recorded_media(source_session_id)
+        except Exception as exc:
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded source could not be resolved"
+            ) from exc
+        _digest(catalog_sha256, "source catalog sha256")
+        selected_sources = self._verify_catalog(
+            detail,
+            expected_session_id=source_session_id,
+        )
+        camera_source = selected_sources["video"]
+        media_artifact = self._verify_recorded_media_artifact(
+            detail=detail,
+            camera_source=camera_source,
+            recorded_media=recorded_media,
+        )
+        self._verify_replay(
+            detail=detail,
+            selected_sources=selected_sources,
+            replay=replay,
+        )
+        try:
+            media = (
+                self._media_inspector.inspect(media_artifact, replay)
+                if prepare_media
+                else self._media_inspector.restore_prepared(media_artifact, replay)
+            )
+        except Exception as exc:
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera manifest failed validation"
+            ) from exc
+        if media is None:
+            raise PortableSourceNotPreparedError("recorded camera manifest has not been prepared")
+        epoch = self._verify_media(
+            media,
+            expected_session_id=detail.summary.session_id,
+            media_artifact=media_artifact,
+        )
+        source_bundle_document = self._source_bundle_document(
+            detail=detail,
+            catalog_sha256=catalog_sha256,
+            selected_sources=selected_sources,
+            replay=replay,
+            media=media,
+        )
+        source_bundle = _canonical_json(source_bundle_document)
+        source_bundle_sha256 = hashlib.sha256(source_bundle).hexdigest()
+        capability_document = self._capability_document(
+            detail=detail,
+            catalog_sha256=catalog_sha256,
+            source_bundle_sha256=source_bundle_sha256,
+            selected_sources=selected_sources,
+            media=media,
+        )
+        capability_manifest = _canonical_json(capability_document)
+        capability_sha256 = hashlib.sha256(capability_manifest).hexdigest()
+        admission = PortableRecordedSourceAdmission(
+            source_session_id=detail.summary.session_id,
+            source_catalog_sha256=catalog_sha256,
+            source_bundle_sha256=source_bundle_sha256,
+            source_capability_manifest_sha256=capability_sha256,
+            source_adapter_sha256=self.requirements.adapter_sha256,
+            frame_count=len(epoch.segments),
+            timeline_start_seconds=epoch.timeline_start_seconds,
+            timeline_end_seconds=epoch.timeline_end_seconds,
+            camera_generation_sha256=media.generation_sha256,
+            source_bundle=source_bundle,
+            capability_manifest=capability_manifest,
+        )
+        if (
+            expected_admission_sha256 is not None
+            and admission.identity_sha256 != expected_admission_sha256
+        ):
+            raise PortableSourceAdmissionStaleError(
+                "recorded source changed after its admission check"
+            )
+        if persist:
+            try:
+                final_detail, final_catalog_sha256 = (
+                    self._session_store.get_session_with_catalog_snapshot(source_session_id)
+                )
+            except Exception as exc:
+                raise PortableSourceAdmissionIntegrityError(
+                    "recorded source catalog could not be rechecked"
+                ) from exc
+            _digest(final_catalog_sha256, "source catalog sha256")
+            if final_catalog_sha256 != catalog_sha256 or final_detail != detail:
+                raise PortableSourceAdmissionStaleError(
+                    "recorded source catalog changed during admission"
+                )
+            _write_immutable_document(
+                self._document_root,
+                source_bundle_sha256,
+                source_bundle,
+            )
+            _write_immutable_document(
+                self._document_root,
+                capability_sha256,
+                capability_manifest,
+            )
+        return admission
+
+    def _verify_catalog(
+        self,
+        detail: SessionDetail,
+        *,
+        expected_session_id: str,
+    ) -> dict[str, SessionSource]:
+        expected = self.requirements
+        summary = detail.summary
+        if (
+            summary.session_id != expected_session_id
+            or summary.lab is not None
+            or summary.status != "ready"
+            or not summary.replayable
+            or detail.plugin_id != expected.plugin_id
+            or detail.archive_id != expected.archive_id
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "session is not an admitted recorded K1 source"
+            )
+        if not set(expected.required_modalities).issubset(
+            summary.modalities
+        ) or summary.source_count != len(detail.sources):
+            raise PortableSourceAdmissionIntegrityError(
+                "required recorded source modalities are unavailable"
+            )
+        artifacts = {artifact.artifact_id: artifact for artifact in detail.artifacts}
+        selected: dict[str, SessionSource] = {}
+        for modality in expected.required_modalities:
+            matches = tuple(source for source in detail.sources if source.modality == modality)
+            if len(matches) != 1:
+                raise PortableSourceAdmissionIntegrityError(
+                    "recorded modality does not have one canonical source"
+                )
+            source = matches[0]
+            if source.status != "recorded" or not source.seekable:
+                raise PortableSourceAdmissionIntegrityError(
+                    "recorded source is not sealed and seekable"
+                )
+            artifact = artifacts.get(source.artifact_id)
+            if artifact is None or artifact.integrity_status not in _SEALED_ARTIFACT_STATES:
+                raise PortableSourceAdmissionIntegrityError(
+                    "recorded source artifact is not sealed"
+                )
+            if modality != "video" and artifact.sha256 is None:
+                raise PortableSourceAdmissionIntegrityError(
+                    "spatial source artifact has no content identity"
+                )
+            selected[modality] = source
+        camera = selected["video"]
+        recorded_video_artifacts = tuple(
+            artifact for artifact in detail.artifacts if artifact.kind == "recorded-video"
+        )
+        if (
+            camera.source_id != expected.camera_source_id
+            or camera.semantic_channel_id != expected.camera_semantic_channel_id
+            or len(recorded_video_artifacts) != 1
+            or recorded_video_artifacts[0].artifact_id != camera.artifact_id
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera capability does not match the setup"
+            )
+        return selected
+
+    def _verify_recorded_media_artifact(
+        self,
+        *,
+        detail: SessionDetail,
+        camera_source: SessionSource,
+        recorded_media: tuple[RecordedMediaArtifact, ...],
+    ) -> RecordedMediaArtifact:
+        artifacts = {artifact.artifact_id: artifact for artifact in detail.artifacts}
+        catalog_artifact = artifacts.get(camera_source.artifact_id)
+        matching_media = tuple(
+            item for item in recorded_media if item.artifact_id == camera_source.artifact_id
+        )
+        if (
+            catalog_artifact is None
+            or catalog_artifact.kind != "recorded-video"
+            or len(recorded_media) != 1
+            or len(matching_media) != 1
+        ):
+            raise PortableSourceAdmissionIntegrityError("recorded camera artifact is unavailable")
+        media_artifact = matching_media[0]
+        if (
+            media_artifact.session_id != detail.summary.session_id
+            or media_artifact.artifact_id != catalog_artifact.artifact_id
+            or media_artifact.byte_length != catalog_artifact.byte_length
+            or media_artifact.byte_length < 1
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera artifact identity disagrees with the catalog"
+            )
+        _pattern(
+            media_artifact.public_source_id,
+            _SOURCE_ID,
+            "recorded camera public source id",
+        )
+        return media_artifact
+
+    def _verify_replay(
+        self,
+        *,
+        detail: SessionDetail,
+        selected_sources: dict[str, SessionSource],
+        replay: ReplayCommand,
+    ) -> None:
+        if (
+            replay.session_id != detail.summary.session_id
+            or replay.plugin_id != detail.plugin_id
+            or replay.plugin_id != self.requirements.plugin_id
+            or replay.speed != 1.0
+            or replay.loop
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "spatial replay is bound to another source contract"
+            )
+        catalog_artifacts = {artifact.artifact_id: artifact for artifact in detail.artifacts}
+        replay_ids = tuple(artifact.artifact_id for artifact in replay.artifacts)
+        if not replay_ids or len(set(replay_ids)) != len(replay_ids):
+            raise PortableSourceAdmissionIntegrityError("spatial replay members are not unique")
+        for replay_artifact in replay.artifacts:
+            catalog_artifact = catalog_artifacts.get(replay_artifact.artifact_id)
+            if (
+                catalog_artifact is None
+                or catalog_artifact.integrity_status not in _SEALED_ARTIFACT_STATES
+                or replay_artifact.media_type != catalog_artifact.media_type
+                or replay_artifact.file_byte_length != catalog_artifact.byte_length
+                or not 1 <= replay_artifact.replay_byte_length <= replay_artifact.file_byte_length
+                or replay_artifact.expected_sha256 != catalog_artifact.sha256
+            ):
+                raise PortableSourceAdmissionIntegrityError(
+                    "spatial replay member disagrees with the catalog"
+                )
+        required_spatial_artifact_ids = {
+            selected_sources[modality].artifact_id
+            for modality in self.requirements.required_modalities
+            if modality != "video"
+        }
+        if (
+            replay.primary_artifact_id not in replay_ids
+            or not required_spatial_artifact_ids.issubset(replay_ids)
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "required spatial artifacts are absent from replay"
+            )
+
+    def _probe_camera_summary(
+        self,
+        media_artifact: RecordedMediaArtifact,
+    ) -> int:
+        summary, epoch_ordinal = _read_canonical_camera_summary(media_artifact.source_path)
+        segment_count = summary.get("segment_count")
+        if (
+            summary.get("schema_version") != CAMERA_ARCHIVE_SCHEMA
+            or summary.get("source_id") != self.requirements.camera_source_id
+            or summary.get("codec_epoch") != epoch_ordinal
+            or summary.get("status") != "complete"
+            or summary.get("synchronization") != "host-arrival-best-effort"
+            or summary.get("failure_code") is not None
+            or summary.get("init_sha256") != self.requirements.camera_init_sha256
+            or not isinstance(segment_count, int)
+            or isinstance(segment_count, bool)
+            or not 1 <= segment_count <= MAX_SAFE_INTEGER
+            or summary.get("entry_count") != segment_count
+            or summary.get("media_segment_count") != segment_count
+            or summary.get("commit_policy") != "per-segment-fsync"
+            or summary.get("artifacts")
+            != {
+                "init": "init.mp4",
+                "segments": "segments",
+                "index": "index.jsonl",
+            }
+        ):
+            raise PortableSourceAdmissionIntegrityError("recorded camera summary is incompatible")
+        return segment_count
+
+    def _verify_media(
+        self,
+        manifest: RecordedMediaManifest,
+        *,
+        expected_session_id: str,
+        media_artifact: RecordedMediaArtifact,
+    ) -> RecordedMediaEpoch:
+        expected = self.requirements
+        if (
+            manifest.session_id != expected_session_id
+            or manifest.session_id != media_artifact.session_id
+            or manifest.artifact_id != media_artifact.artifact_id
+            or manifest.public_source_id != media_artifact.public_source_id
+            or manifest.byte_length != media_artifact.byte_length
+            or len(manifest.epochs) != 1
+            or manifest.synchronization != "host-arrival-best-effort"
+            or not _SHA256.fullmatch(manifest.generation_sha256)
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera epoch topology is unsupported"
+            )
+        epoch = manifest.epochs[0]
+        if (
+            epoch.media_type != expected.camera_media_type
+            or epoch.init_sha256 != expected.camera_init_sha256
+            or not epoch.segments
+            or epoch.timeline_end_seconds <= epoch.timeline_start_seconds
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera media profile is incompatible"
+            )
+        return epoch
+
+    def _source_bundle_document(
+        self,
+        *,
+        detail: SessionDetail,
+        catalog_sha256: str,
+        selected_sources: dict[str, SessionSource],
+        replay: ReplayCommand,
+        media: RecordedMediaManifest,
+    ) -> dict[str, object]:
+        epoch = media.epochs[0]
+        artifacts = {artifact.artifact_id: artifact for artifact in detail.artifacts}
+        return {
+            "schema_version": PORTABLE_SOURCE_BUNDLE_SCHEMA,
+            "source_session_id": detail.summary.session_id,
+            "source_catalog_sha256": catalog_sha256,
+            "plugin_id": detail.plugin_id,
+            "archive_id": detail.archive_id,
+            "source_adapter": {
+                "id": self.requirements.adapter_id,
+                "version": self.requirements.adapter_version,
+                "sha256": self.requirements.adapter_sha256,
+            },
+            "sources": [
+                {
+                    **selected_sources[modality].as_dict(),
+                    "artifact": _artifact_document(
+                        artifacts[selected_sources[modality].artifact_id]
+                    ),
+                }
+                for modality in self.requirements.required_modalities
+            ],
+            "spatial_replay": {
+                "primary_artifact_id": replay.primary_artifact_id,
+                "members": [
+                    {
+                        "artifact_id": artifact.artifact_id,
+                        "media_type": artifact.media_type,
+                        "byte_length": artifact.file_byte_length,
+                        "replay_byte_length": artifact.replay_byte_length,
+                        "sha256": artifact.expected_sha256,
+                    }
+                    for artifact in replay.artifacts
+                ],
+                "timeline_origin_epoch_ns": replay.timeline_origin_epoch_ns,
+                "timeline_origin_monotonic_ns": replay.timeline_origin_monotonic_ns,
+            },
+            "camera": {
+                "artifact_id": media.artifact_id,
+                "public_source_id": media.public_source_id,
+                "generation_sha256": media.generation_sha256,
+                "synchronization": media.synchronization,
+                "epoch": {
+                    "ordinal": epoch.ordinal,
+                    "media_type": epoch.media_type,
+                    "init": {
+                        "byte_length": epoch.init_byte_length,
+                        "sha256": epoch.init_sha256,
+                    },
+                    "timeline_start_seconds": epoch.timeline_start_seconds,
+                    "timeline_end_seconds": epoch.timeline_end_seconds,
+                    "segments": [
+                        {
+                            "sequence": segment.sequence,
+                            "byte_length": segment.byte_length,
+                            "sha256": segment.sha256,
+                            "random_access": segment.random_access,
+                            "end_time_seconds": segment.end_time_seconds,
+                        }
+                        for segment in epoch.segments
+                    ],
+                },
+            },
+            "authority": dict(_AUTHORITY),
+        }
+
+    def _capability_document(
+        self,
+        *,
+        detail: SessionDetail,
+        catalog_sha256: str,
+        source_bundle_sha256: str,
+        selected_sources: dict[str, SessionSource],
+        media: RecordedMediaManifest,
+    ) -> dict[str, object]:
+        epoch = media.epochs[0]
+        return {
+            "schema_version": PORTABLE_SOURCE_CAPABILITY_SCHEMA,
+            "source_session_id": detail.summary.session_id,
+            "source_catalog_sha256": catalog_sha256,
+            "source_bundle_sha256": source_bundle_sha256,
+            "source_adapter_sha256": self.requirements.adapter_sha256,
+            "modalities": [
+                {
+                    "modality": modality,
+                    "source_id": selected_sources[modality].source_id,
+                    "semantic_channel_id": (selected_sources[modality].semantic_channel_id),
+                    "seekable": True,
+                }
+                for modality in self.requirements.required_modalities
+            ],
+            "camera_profile": {
+                "media_type": epoch.media_type,
+                "init_sha256": epoch.init_sha256,
+                "width": self.requirements.expected_width,
+                "height": self.requirements.expected_height,
+                "profile_attestation": "exact-isobmff-init-sha256",
+                "generation_sha256": media.generation_sha256,
+                "frame_count": len(epoch.segments),
+                "timeline_start_seconds": epoch.timeline_start_seconds,
+                "timeline_end_seconds": epoch.timeline_end_seconds,
+            },
+            "calibration": {
+                "slot": self.requirements.calibration_slot,
+                "sha256": self.requirements.calibration_sha256,
+                "binding": "external-rig-profile",
+            },
+            "authority": dict(_AUTHORITY),
+        }
+
+
+def _read_canonical_camera_summary(
+    source_path: Path,
+) -> tuple[dict[str, object], int]:
+    """Read one direct epoch summary through bounded no-follow descriptors."""
+
+    source_descriptor = -1
+    epoch_descriptor = -1
+    summary_descriptor = -1
+    try:
+        directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+        source_descriptor = os.open(source_path, directory_flags)
+        if not stat.S_ISDIR(os.fstat(source_descriptor).st_mode):
+            raise PortableSourceAdmissionIntegrityError("recorded camera source is not a directory")
+        epoch_name: str | None = None
+        epoch_ordinal: int | None = None
+        with os.scandir(source_descriptor) as entries:
+            for entry in entries:
+                match = _EPOCH.fullmatch(entry.name)
+                if (
+                    epoch_name is not None
+                    or match is None
+                    or not entry.is_dir(follow_symlinks=False)
+                ):
+                    raise PortableSourceAdmissionIntegrityError(
+                        "recorded camera source does not have one canonical epoch"
+                    )
+                epoch_name = entry.name
+                epoch_ordinal = int(match.group(1))
+        if epoch_name is None or epoch_ordinal is None:
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera source does not have one canonical epoch"
+            )
+        epoch_descriptor = os.open(
+            epoch_name,
+            directory_flags,
+            dir_fd=source_descriptor,
+        )
+        if not stat.S_ISDIR(os.fstat(epoch_descriptor).st_mode):
+            raise PortableSourceAdmissionIntegrityError("recorded camera epoch is not a directory")
+        summary_descriptor = os.open(
+            "summary.json",
+            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0),
+            dir_fd=epoch_descriptor,
+        )
+        before = os.fstat(summary_descriptor)
+        if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= MAX_MEDIA_SUMMARY_BYTES:
+            raise PortableSourceAdmissionIntegrityError("recorded camera summary is outside bounds")
+        chunks: list[bytes] = []
+        remaining = before.st_size
+        while remaining:
+            chunk = os.read(summary_descriptor, remaining)
+            if not chunk:
+                break
+            chunks.append(chunk)
+            remaining -= len(chunk)
+        payload = b"".join(chunks)
+        after = os.fstat(summary_descriptor)
+        if len(payload) != before.st_size or (
+            after.st_dev,
+            after.st_ino,
+            after.st_size,
+            after.st_mtime_ns,
+        ) != (
+            before.st_dev,
+            before.st_ino,
+            before.st_size,
+            before.st_mtime_ns,
+        ):
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera summary changed while it was read"
+            )
+        try:
+            value = json.loads(payload)
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise PortableSourceAdmissionIntegrityError(
+                "recorded camera summary is invalid"
+            ) from exc
+        if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
+            raise PortableSourceAdmissionIntegrityError("recorded camera summary is not an object")
+        return value, epoch_ordinal
+    except PortableSourceAdmissionIntegrityError:
+        raise
+    except OSError as exc:
+        raise PortableSourceAdmissionIntegrityError(
+            "recorded camera summary is unavailable"
+        ) from exc
+    finally:
+        for descriptor in (
+            summary_descriptor,
+            epoch_descriptor,
+            source_descriptor,
+        ):
+            if descriptor >= 0:
+                os.close(descriptor)
+
+
+def _artifact_document(artifact: SessionArtifact) -> dict[str, object]:
+    return {
+        "artifact_id": artifact.artifact_id,
+        "kind": artifact.kind,
+        "media_type": artifact.media_type,
+        "byte_length": artifact.byte_length,
+        "sha256": artifact.sha256,
+        "integrity_status": artifact.integrity_status,
+    }
+
+
+def _write_immutable_document(root: Path, digest: str, payload: bytes) -> None:
+    _digest(digest, "document sha256")
+    if hashlib.sha256(payload).hexdigest() != digest:
+        raise PortableSourceAdmissionIntegrityError(
+            "immutable document payload has another identity"
+        )
+    root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    metadata = root.lstat()
+    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+        raise PortableSourceAdmissionIntegrityError("portable source document root is unsafe")
+    destination = root / f"{digest}.json"
+    if destination.exists():
+        existing = destination.read_bytes()
+        if existing != payload:
+            raise PortableSourceAdmissionIntegrityError(
+                "immutable source document identity collided"
+            )
+        return
+    temporary = root / f".tmp-{secrets.token_hex(16)}"
+    descriptor = os.open(
+        temporary,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+        0o600,
+    )
+    published = False
+    try:
+        with os.fdopen(descriptor, "wb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, destination)
+        _fsync_directory(root)
+        published = True
+    finally:
+        if not published:
+            with suppress(FileNotFoundError):
+                temporary.unlink()
+
+
+def _fsync_directory(path: Path) -> None:
+    descriptor = os.open(path, os.O_RDONLY)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def _canonical_json(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def _sha256(value: object) -> str:
+    return hashlib.sha256(_canonical_json(value)).hexdigest()
+
+
+def _digest(value: object, label: str) -> str:
+    if not isinstance(value, str) or _SHA256.fullmatch(value) is None:
+        raise ValueError(f"{label} is invalid")
+    return value
+
+
+def _pattern(value: object, pattern: re.Pattern[str], label: str) -> str:
+    if not isinstance(value, str) or pattern.fullmatch(value) is None:
+        raise ValueError(f"{label} is invalid")
+    return value

src/k1link/observatory/worker_agent.py

@@ -0,0 +1,695 @@
+"""Transport-agnostic Worker 006 core for sealed Observatory jobs.
+
+The agent accepts only the durable, path-free recorded-job projection.  It
+validates every identity sealed by Mission Core, resolves an executor from a
+local four-digest allowlist, and passes a typed job to that executor.  Server
+payloads can never provide commands, paths, environment variables, images, or
+other executable instructions.
+
+Network authentication, polling cadence, local CAS resolution, ML runtimes,
+and deployment are deliberately outside this module.  They are supplied by a
+transport and an executor adapter at the Worker boundary.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import threading
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Annotated, Final, Literal, Protocol
+from uuid import uuid4
+
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+from k1link.observatory.recorded_jobs import (
+    OBSERVATORY_RECORDED_CLAIM_SCHEMA,
+    OBSERVATORY_RECORDED_JOB_REQUEST_SCHEMA,
+    OBSERVATORY_RECORDED_JOB_SCHEMA,
+)
+
+WORKER_006_CONTOUR_ID: Final = "worker-006"
+MAX_EXECUTOR_FAILURE_MESSAGE_LENGTH: Final = 512
+
+_JOB_ID_PATTERN: Final = r"^observatory-run-[a-f0-9]{32}$"
+_CLAIM_TOKEN_PATTERN: Final = r"^[a-f0-9]{64}$"
+_CLAIM_REQUEST_ID_PATTERN: Final = r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$"
+_SESSION_ID_PATTERN: Final = r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
+_IDENTIFIER_PATTERN: Final = r"^[a-z][a-z0-9-]{2,95}$"
+_SHA256_PATTERN: Final = r"^[a-f0-9]{64}$"
+_TIMESTAMP_PATTERN: Final = (
+    r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}"
+    r"(?:\.[0-9]{1,9})?(?:Z|[+-][0-9]{2}:[0-9]{2})$"
+)
+_CLAIM_REQUEST_ID = re.compile(_CLAIM_REQUEST_ID_PATTERN)
+
+type WorkerCycleState = Literal[
+    "empty",
+    "deferred",
+    "succeeded",
+    "failed",
+    "rejected",
+]
+type RecordedJobWireState = Literal[
+    "accepted",
+    "queued",
+    "claimed",
+    "running",
+    "paused",
+    "preemption-pending",
+    "succeeded",
+    "failed",
+    "reconciliation-required",
+]
+
+Sha256 = Annotated[str, Field(pattern=_SHA256_PATTERN)]
+Identifier = Annotated[
+    str,
+    Field(min_length=3, max_length=96, pattern=_IDENTIFIER_PATTERN),
+]
+SessionId = Annotated[
+    str,
+    Field(min_length=1, max_length=128, pattern=_SESSION_ID_PATTERN),
+]
+Timestamp = Annotated[
+    str,
+    Field(min_length=20, max_length=64, pattern=_TIMESTAMP_PATTERN),
+]
+
+
+class ObservatoryWorkerAgentError(RuntimeError):
+    """Base error for the local Worker 006 agent core."""
+
+
+class ObservatoryWorkerAgentBusyError(ObservatoryWorkerAgentError):
+    """A second poll cycle attempted to overlap the active claim."""
+
+
+class ObservatoryWorkerClaimRejectedError(ObservatoryWorkerAgentError):
+    """A transport supplied an unknown, spoofed, or corrupted claim."""
+
+
+class ObservatoryWorkerExecutorUnavailableError(ObservatoryWorkerAgentError):
+    """No local adapter matches the exact sealed executor identity."""
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryWorkerExecutorIdentity:
+    """The only identity that may select executable Worker code."""
+
+    release_sha256: str
+    image_sha256: str
+    model_manifest_sha256: str
+    resource_profile_sha256: str
+
+    def __post_init__(self) -> None:
+        for label, value in (
+            ("executor release", self.release_sha256),
+            ("executor image", self.image_sha256),
+            ("model manifest", self.model_manifest_sha256),
+            ("resource profile", self.resource_profile_sha256),
+        ):
+            if re.fullmatch(_SHA256_PATTERN, value) is None:
+                raise ValueError(f"{label} SHA-256 is invalid")
+
+
+@dataclass(frozen=True, slots=True)
+class SealedObservatoryRecordedJob:
+    """Validated path-free job passed to one local executor adapter."""
+
+    job_id: str
+    request_sha256: str
+    identity_sha256: str
+    source_session_id: str
+    source_catalog_sha256: str
+    source_bundle_sha256: str
+    source_capability_manifest_sha256: str
+    source_adapter_id: str
+    source_adapter_version: int
+    source_adapter_sha256: str
+    setup_id: str
+    definition_id: str
+    definition_version: int
+    definition_sha256: str
+    executor_release_id: str
+    executor_identity: ObservatoryWorkerExecutorIdentity
+    model_release_ids: tuple[str, ...]
+    resource_profile_id: str
+    checkpoint_policy: Literal["cooperative", "non-checkpointable"]
+    allowed_checkpoints: tuple[str, ...]
+    claim_generation: int
+    restart_from_zero: bool
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryWorkerExecutionResult:
+    """Content-addressed result identity returned by a local executor."""
+
+    result_id: str
+    result_sha256: str
+
+    def __post_init__(self) -> None:
+        if re.fullmatch(_SESSION_ID_PATTERN, self.result_id) is None:
+            raise ValueError("Worker result id is invalid")
+        if re.fullmatch(_SHA256_PATTERN, self.result_sha256) is None:
+            raise ValueError("Worker result SHA-256 is invalid")
+
+
+class ObservatoryWorkerExecutor(Protocol):
+    """A local implementation selected only by its sealed digest tuple."""
+
+    def execute(
+        self,
+        job: SealedObservatoryRecordedJob,
+    ) -> ObservatoryWorkerExecutionResult: ...
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryWorkerExecutorRegistration:
+    identity: ObservatoryWorkerExecutorIdentity
+    adapter: ObservatoryWorkerExecutor
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryWorkerExecutorRegistry:
+    """In-memory local executor allowlist; it never accepts server code."""
+
+    registrations: tuple[ObservatoryWorkerExecutorRegistration, ...]
+
+    def __post_init__(self) -> None:
+        identities = [registration.identity for registration in self.registrations]
+        if len(identities) != len(set(identities)):
+            raise ValueError("Worker executor identities must be unique")
+
+    def resolve(
+        self,
+        identity: ObservatoryWorkerExecutorIdentity,
+    ) -> ObservatoryWorkerExecutor:
+        for registration in self.registrations:
+            if registration.identity == identity:
+                return registration.adapter
+        raise ObservatoryWorkerExecutorUnavailableError(
+            "exact executor identity is not locally allowlisted"
+        )
+
+
+class ObservatoryWorkerTransport(Protocol):
+    """State-transition port implemented by HTTP, IPC, or a test transport."""
+
+    def claim_next(
+        self,
+        *,
+        claimant_id: str,
+        claim_request_id: str,
+    ) -> Mapping[str, object] | None: ...
+
+    def start(
+        self,
+        *,
+        claimant_id: str,
+        job_id: str,
+        claim_token: str,
+    ) -> Mapping[str, object]: ...
+
+    def succeed(
+        self,
+        *,
+        claimant_id: str,
+        job_id: str,
+        claim_token: str,
+        result_id: str,
+        result_sha256: str,
+    ) -> Mapping[str, object]: ...
+
+    def fail(
+        self,
+        *,
+        claimant_id: str,
+        job_id: str,
+        claim_token: str,
+        error_code: str,
+        message: str,
+    ) -> Mapping[str, object]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryWorkerCycleReport:
+    state: WorkerCycleState
+    claim_request_id: str
+    job_id: str | None = None
+    result_id: str | None = None
+    failure_code: str | None = None
+
+
+class _StrictPayload(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)
+
+
+class _AuthorityPayload(_StrictPayload):
+    commands_enabled: Literal[False]
+    actuation_allowed: Literal[False]
+    navigation_or_safety_accepted: Literal[False]
+    production_accepted: Literal[False]
+
+
+class _SourceAdapterPayload(_StrictPayload):
+    adapter_id: Identifier
+    version: int = Field(ge=1)
+    adapter_sha256: Sha256
+
+
+class _SourcePayload(_StrictPayload):
+    session_id: SessionId
+    catalog_sha256: Sha256
+    bundle_sha256: Sha256
+    capability_manifest_sha256: Sha256
+    adapter: _SourceAdapterPayload
+
+
+class _SetupPayload(_StrictPayload):
+    setup_id: Identifier
+    definition_id: Identifier
+    definition_version: int = Field(ge=1)
+    definition_sha256: Sha256
+
+
+class _ExecutorPayload(_StrictPayload):
+    release_id: Identifier
+    release_sha256: Sha256
+    image_sha256: Sha256
+    model_release_ids: list[Identifier] = Field(max_length=32)
+    learned_models: list[Identifier] = Field(max_length=32)
+    model_manifest_sha256: Sha256
+    resource_profile_id: Identifier
+    resource_profile_sha256: Sha256
+
+
+class _CheckpointPolicyPayload(_StrictPayload):
+    mode: Literal["cooperative", "non-checkpointable"]
+    allowed_checkpoints: list[Identifier] = Field(max_length=64)
+    last_checkpoint_id: Identifier | None
+
+
+class _PriorityPayload(_StrictPayload):
+    class_: Literal["recorded"] = Field(alias="class")
+    rank: Literal[100]
+    server_owned: Literal[True]
+
+
+class _ResultPayload(_StrictPayload):
+    result_id: SessionId
+    sha256: Sha256
+
+
+class _TerminalPayload(_StrictPayload):
+    code: Identifier
+    message: str = Field(min_length=1, max_length=1_000)
+
+
+class _RecordedJobPayload(_StrictPayload):
+    schema_version: Literal["missioncore.observatory-recorded-job/v1"]
+    job_id: str = Field(pattern=_JOB_ID_PATTERN)
+    idempotency_key: str = Field(
+        min_length=1,
+        max_length=160,
+        pattern=_CLAIM_REQUEST_ID_PATTERN,
+    )
+    request_sha256: Sha256
+    identity_sha256: Sha256
+    submission_receipt_sha256: Sha256
+    source: _SourcePayload
+    setup: _SetupPayload
+    executor: _ExecutorPayload
+    checkpoint_policy: _CheckpointPolicyPayload
+    priority: _PriorityPayload
+    state: RecordedJobWireState
+    preemption_requested: bool
+    restart_from_zero: bool
+    preemption_receipt_sha256: Sha256 | None
+    claim_generation: int = Field(ge=0)
+    result: _ResultPayload | None
+    terminal: _TerminalPayload | None
+    created_at_utc: Timestamp
+    updated_at_utc: Timestamp
+    authority: _AuthorityPayload
+
+
+class _RecordedClaimPayload(_StrictPayload):
+    schema_version: Literal["missioncore.observatory-recorded-job-claim/v1"]
+    claim_request_id: str = Field(
+        min_length=1,
+        max_length=160,
+        pattern=_CLAIM_REQUEST_ID_PATTERN,
+    )
+    request_sha256: Sha256
+    claimant_id: Identifier
+    claim_token: str = Field(pattern=_CLAIM_TOKEN_PATTERN)
+    job: _RecordedJobPayload
+
+
+@dataclass(frozen=True, slots=True)
+class _ValidatedClaim:
+    claim_request_id: str
+    claim_token: str
+    job: SealedObservatoryRecordedJob
+
+
+class ObservatoryWorkerAgent:
+    """Runs at most one sealed recorded-job claim at a time on Worker 006."""
+
+    def __init__(
+        self,
+        *,
+        transport: ObservatoryWorkerTransport,
+        executors: ObservatoryWorkerExecutorRegistry,
+        claim_request_id_factory: Callable[[], str] | None = None,
+    ) -> None:
+        self._transport = transport
+        self._executors = executors
+        self._claim_request_id_factory = claim_request_id_factory or _default_claim_request_id
+        self._cycle_lock = threading.Lock()
+
+    def run_once(self) -> ObservatoryWorkerCycleReport:
+        """Claim, validate, execute, and seal one durable job if available."""
+
+        if not self._cycle_lock.acquire(blocking=False):
+            raise ObservatoryWorkerAgentBusyError("Worker 006 already owns an active claim cycle")
+        try:
+            return self._run_once_locked()
+        finally:
+            self._cycle_lock.release()
+
+    def _run_once_locked(self) -> ObservatoryWorkerCycleReport:
+        claim_request_id = self._claim_request_id_factory()
+        if _CLAIM_REQUEST_ID.fullmatch(claim_request_id) is None:
+            raise ValueError("Worker claim request id is invalid")
+        payload = self._transport.claim_next(
+            claimant_id=WORKER_006_CONTOUR_ID,
+            claim_request_id=claim_request_id,
+        )
+        if payload is None:
+            return ObservatoryWorkerCycleReport(
+                state="empty",
+                claim_request_id=claim_request_id,
+            )
+        try:
+            claim = _validate_claim(payload, claim_request_id=claim_request_id)
+        except ObservatoryWorkerClaimRejectedError:
+            return ObservatoryWorkerCycleReport(
+                state="rejected",
+                claim_request_id=claim_request_id,
+                failure_code="claim-rejected",
+            )
+
+        try:
+            adapter = self._executors.resolve(claim.job.executor_identity)
+        except ObservatoryWorkerExecutorUnavailableError:
+            failure_code = "executor-not-allowlisted"
+            acknowledgement = self._transport.fail(
+                claimant_id=WORKER_006_CONTOUR_ID,
+                job_id=claim.job.job_id,
+                claim_token=claim.claim_token,
+                error_code=failure_code,
+                message="Exact executor identity is not installed on Worker 006.",
+            )
+            _validate_transition_acknowledgement(
+                acknowledgement,
+                expected_job=claim.job,
+                expected_state="failed",
+            )
+            return ObservatoryWorkerCycleReport(
+                state="failed",
+                claim_request_id=claim_request_id,
+                job_id=claim.job.job_id,
+                failure_code=failure_code,
+            )
+
+        started_payload = self._transport.start(
+            claimant_id=WORKER_006_CONTOUR_ID,
+            job_id=claim.job.job_id,
+            claim_token=claim.claim_token,
+        )
+        started = _validate_transition_acknowledgement(
+            started_payload,
+            expected_job=claim.job,
+            expected_state=("running", "paused"),
+        )
+        if started.state == "paused":
+            return ObservatoryWorkerCycleReport(
+                state="deferred",
+                claim_request_id=claim_request_id,
+                job_id=claim.job.job_id,
+            )
+
+        try:
+            result = adapter.execute(claim.job)
+            if not isinstance(result, ObservatoryWorkerExecutionResult):
+                raise TypeError("executor returned an unknown result contract")
+        except Exception as exc:
+            failure_code = "executor-error"
+            acknowledgement = self._transport.fail(
+                claimant_id=WORKER_006_CONTOUR_ID,
+                job_id=claim.job.job_id,
+                claim_token=claim.claim_token,
+                error_code=failure_code,
+                message=_bounded_executor_failure(exc),
+            )
+            _validate_transition_acknowledgement(
+                acknowledgement,
+                expected_job=claim.job,
+                expected_state="failed",
+            )
+            return ObservatoryWorkerCycleReport(
+                state="failed",
+                claim_request_id=claim_request_id,
+                job_id=claim.job.job_id,
+                failure_code=failure_code,
+            )
+
+        acknowledgement = self._transport.succeed(
+            claimant_id=WORKER_006_CONTOUR_ID,
+            job_id=claim.job.job_id,
+            claim_token=claim.claim_token,
+            result_id=result.result_id,
+            result_sha256=result.result_sha256,
+        )
+        succeeded = _validate_transition_acknowledgement(
+            acknowledgement,
+            expected_job=claim.job,
+            expected_state="succeeded",
+        )
+        if succeeded.result is None or (
+            succeeded.result.result_id != result.result_id
+            or succeeded.result.sha256 != result.result_sha256
+        ):
+            raise ObservatoryWorkerClaimRejectedError(
+                "Worker success acknowledgement changed result identity"
+            )
+        return ObservatoryWorkerCycleReport(
+            state="succeeded",
+            claim_request_id=claim_request_id,
+            job_id=claim.job.job_id,
+            result_id=result.result_id,
+        )
+
+
+def _validate_claim(
+    payload: Mapping[str, object],
+    *,
+    claim_request_id: str,
+) -> _ValidatedClaim:
+    try:
+        claim = _RecordedClaimPayload.model_validate(dict(payload))
+        if claim.claim_request_id != claim_request_id:
+            raise ObservatoryWorkerClaimRejectedError("Worker claim request identity changed")
+        if claim.claimant_id != WORKER_006_CONTOUR_ID:
+            raise ObservatoryWorkerClaimRejectedError("Worker claim belongs to another claimant")
+        expected_claim_request_sha256 = _sha256_document(
+            {
+                "schema_version": OBSERVATORY_RECORDED_CLAIM_SCHEMA,
+                "claim_request_id": claim_request_id,
+                "claimant_id": WORKER_006_CONTOUR_ID,
+            }
+        )
+        if claim.request_sha256 != expected_claim_request_sha256:
+            raise ObservatoryWorkerClaimRejectedError("Worker claim request digest changed")
+        if claim.job.state != "claimed" or claim.job.claim_generation < 1:
+            raise ObservatoryWorkerClaimRejectedError(
+                "Worker claim job is not in a claimed generation"
+            )
+        if claim.job.result is not None or claim.job.terminal is not None:
+            raise ObservatoryWorkerClaimRejectedError(
+                "Worker claim already carries a terminal outcome"
+            )
+        job = _seal_job(claim.job)
+    except ObservatoryWorkerClaimRejectedError:
+        raise
+    except (TypeError, ValueError, ValidationError) as exc:
+        raise ObservatoryWorkerClaimRejectedError(
+            "Worker claim payload violates the sealed protocol"
+        ) from exc
+    return _ValidatedClaim(
+        claim_request_id=claim.claim_request_id,
+        claim_token=claim.claim_token,
+        job=job,
+    )
+
+
+def _seal_job(payload: _RecordedJobPayload) -> SealedObservatoryRecordedJob:
+    if payload.executor.learned_models != payload.executor.model_release_ids:
+        raise ObservatoryWorkerClaimRejectedError("Worker claim learned-model identities changed")
+    if len(set(payload.executor.model_release_ids)) != len(payload.executor.model_release_ids):
+        raise ObservatoryWorkerClaimRejectedError("Worker claim model identities are not unique")
+    if len(set(payload.checkpoint_policy.allowed_checkpoints)) != len(
+        payload.checkpoint_policy.allowed_checkpoints
+    ):
+        raise ObservatoryWorkerClaimRejectedError(
+            "Worker claim checkpoint identities are not unique"
+        )
+    if (
+        payload.checkpoint_policy.mode == "cooperative"
+        and not payload.checkpoint_policy.allowed_checkpoints
+    ) or (
+        payload.checkpoint_policy.mode == "non-checkpointable"
+        and payload.checkpoint_policy.allowed_checkpoints
+    ):
+        raise ObservatoryWorkerClaimRejectedError("Worker claim checkpoint policy is inconsistent")
+
+    expected_request_sha256 = _sha256_document(
+        {
+            "schema_version": OBSERVATORY_RECORDED_JOB_REQUEST_SCHEMA,
+            "idempotency_key": payload.idempotency_key,
+            "source_session_id": payload.source.session_id,
+            "source_catalog_sha256": payload.source.catalog_sha256,
+            "source_bundle_sha256": payload.source.bundle_sha256,
+            "source_capability_manifest_sha256": (payload.source.capability_manifest_sha256),
+            "setup_id": payload.setup.setup_id,
+            "definition_sha256": payload.setup.definition_sha256,
+        }
+    )
+    if payload.request_sha256 != expected_request_sha256:
+        raise ObservatoryWorkerClaimRejectedError("Worker recorded-job request identity changed")
+    expected_identity_sha256 = _sha256_document(
+        {
+            "schema_version": OBSERVATORY_RECORDED_JOB_SCHEMA,
+            "source_session_id": payload.source.session_id,
+            "source_catalog_sha256": payload.source.catalog_sha256,
+            "source_bundle_sha256": payload.source.bundle_sha256,
+            "source_capability_manifest_sha256": (payload.source.capability_manifest_sha256),
+            "setup_id": payload.setup.setup_id,
+            "definition_id": payload.setup.definition_id,
+            "definition_version": payload.setup.definition_version,
+            "definition_sha256": payload.setup.definition_sha256,
+            "source_adapter_id": payload.source.adapter.adapter_id,
+            "source_adapter_version": payload.source.adapter.version,
+            "source_adapter_sha256": payload.source.adapter.adapter_sha256,
+            "executor_release_id": payload.executor.release_id,
+            "executor_release_sha256": payload.executor.release_sha256,
+            "executor_image_sha256": payload.executor.image_sha256,
+            "model_release_ids": payload.executor.model_release_ids,
+            "model_manifest_sha256": payload.executor.model_manifest_sha256,
+            "resource_profile_id": payload.executor.resource_profile_id,
+            "resource_profile_sha256": payload.executor.resource_profile_sha256,
+            "checkpoint_policy": payload.checkpoint_policy.mode,
+            "allowed_checkpoints": payload.checkpoint_policy.allowed_checkpoints,
+        }
+    )
+    if payload.identity_sha256 != expected_identity_sha256:
+        raise ObservatoryWorkerClaimRejectedError("Worker recorded-job execution identity changed")
+    expected_submission_receipt_sha256 = _sha256_document(
+        {
+            "schema_version": OBSERVATORY_RECORDED_JOB_SCHEMA,
+            "job_id": payload.job_id,
+            "request_sha256": payload.request_sha256,
+            "identity_sha256": payload.identity_sha256,
+            "state": "accepted",
+            "created_at_utc": payload.created_at_utc,
+        }
+    )
+    if payload.submission_receipt_sha256 != expected_submission_receipt_sha256:
+        raise ObservatoryWorkerClaimRejectedError("Worker recorded-job submission receipt changed")
+
+    return SealedObservatoryRecordedJob(
+        job_id=payload.job_id,
+        request_sha256=payload.request_sha256,
+        identity_sha256=payload.identity_sha256,
+        source_session_id=payload.source.session_id,
+        source_catalog_sha256=payload.source.catalog_sha256,
+        source_bundle_sha256=payload.source.bundle_sha256,
+        source_capability_manifest_sha256=(payload.source.capability_manifest_sha256),
+        source_adapter_id=payload.source.adapter.adapter_id,
+        source_adapter_version=payload.source.adapter.version,
+        source_adapter_sha256=payload.source.adapter.adapter_sha256,
+        setup_id=payload.setup.setup_id,
+        definition_id=payload.setup.definition_id,
+        definition_version=payload.setup.definition_version,
+        definition_sha256=payload.setup.definition_sha256,
+        executor_release_id=payload.executor.release_id,
+        executor_identity=ObservatoryWorkerExecutorIdentity(
+            release_sha256=payload.executor.release_sha256,
+            image_sha256=payload.executor.image_sha256,
+            model_manifest_sha256=payload.executor.model_manifest_sha256,
+            resource_profile_sha256=payload.executor.resource_profile_sha256,
+        ),
+        model_release_ids=tuple(payload.executor.model_release_ids),
+        resource_profile_id=payload.executor.resource_profile_id,
+        checkpoint_policy=payload.checkpoint_policy.mode,
+        allowed_checkpoints=tuple(payload.checkpoint_policy.allowed_checkpoints),
+        claim_generation=payload.claim_generation,
+        restart_from_zero=payload.restart_from_zero,
+    )
+
+
+def _validate_transition_acknowledgement(
+    payload: Mapping[str, object],
+    *,
+    expected_job: SealedObservatoryRecordedJob,
+    expected_state: RecordedJobWireState | tuple[RecordedJobWireState, ...],
+) -> _RecordedJobPayload:
+    try:
+        acknowledgement = _RecordedJobPayload.model_validate(dict(payload))
+        sealed = _seal_job(acknowledgement)
+    except ObservatoryWorkerClaimRejectedError:
+        raise
+    except (TypeError, ValueError, ValidationError) as exc:
+        raise ObservatoryWorkerClaimRejectedError(
+            "Worker transition acknowledgement violates the sealed protocol"
+        ) from exc
+    states = (expected_state,) if isinstance(expected_state, str) else expected_state
+    if acknowledgement.state not in states:
+        raise ObservatoryWorkerClaimRejectedError(
+            "Worker transition acknowledgement has an unexpected state"
+        )
+    if (
+        sealed.job_id != expected_job.job_id
+        or sealed.identity_sha256 != expected_job.identity_sha256
+        or sealed.claim_generation != expected_job.claim_generation
+    ):
+        raise ObservatoryWorkerClaimRejectedError(
+            "Worker transition acknowledgement changed job identity"
+        )
+    return acknowledgement
+
+
+def _default_claim_request_id() -> str:
+    return f"worker-006:{uuid4().hex}"
+
+
+def _bounded_executor_failure(exc: Exception) -> str:
+    detail = " ".join(str(exc).split())
+    message = f"Executor adapter raised {type(exc).__name__}."
+    if detail:
+        message = f"{message} {detail}"
+    return message[:MAX_EXECUTOR_FAILURE_MESSAGE_LENGTH]
+
+
+def _sha256_document(document: Mapping[str, object]) -> str:
+    payload = json.dumps(
+        document,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode()
+    return hashlib.sha256(payload).hexdigest()

src/k1link/web/app.py

@@ -47,10 +47,20 @@ from k1link.observatory.m49_queue_binding import (
     M49QueueBindingError,
     M49RecordedQueueBindingService,
 )
+from k1link.observatory.portable_run_definitions import (
+    PortableRunDefinitionRegistry,
+    PortableRunDefinitionRegistryError,
+)
+from k1link.observatory.portable_setup_projection import (
+    PORTABLE_LAB_V1_SETUP_ID,
+    PortableLabV1SetupProjector,
+    PortableSetupProjectionError,
+)
 from k1link.observatory.recorded_jobs import (
     ObservatoryRecordedJobQueue,
     ObservatoryRecordedQueueError,
 )
+from k1link.observatory.source_admission import RecordedK1SourceAdmissionService
 from k1link.sessions import (
     MaterializedRecording,
     RecordedCameraFrameService,
@@ -160,6 +170,11 @@ from k1link.web.map_api import (
 )
 from k1link.web.map_view_api import build_map_view_router
 from k1link.web.observatory_api import build_observatory_router
+from k1link.web.observatory_worker_api import (
+    ObservatoryWorkerAuthentication,
+    build_observatory_worker_router,
+    load_observatory_worker_authentication,
+)
 from k1link.web.plugin_catalog import DevicePluginCatalog, PluginCatalogError
 from k1link.web.plugin_runtime import (
     STATE_READ_ACTION_ID,
@@ -245,6 +260,25 @@ plugin_environment = load_installed_device_plugins(REPOSITORY_ROOT)
 plugin_catalog: DevicePluginCatalog = plugin_environment.catalog
 plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher
 session_store = SessionStore(REPOSITORY_ROOT)
+
+
+def _load_optional_observatory_worker_authentication(
+    recorded_job_queue: ObservatoryRecordedJobQueue | None,
+    *,
+    token_path: Path,
+) -> tuple[ObservatoryWorkerAuthentication | None, str | None]:
+    """Load the optional Worker credential without widening app startup risk."""
+
+    if recorded_job_queue is None:
+        return None, "Observatory recorded-job queue is unavailable"
+    try:
+        return load_observatory_worker_authentication(token_path), None
+    except ValueError as exc:
+        # Worker pull transport is optional.  A missing or unsafe credential
+        # disables only this router; K1, Simulation and legacy LAB still start.
+        return None, str(exc)
+
+
 OBSERVATORY_RUN_PREPARATION_LEDGER: ObservatoryRunPreparationLedger | None
 OBSERVATORY_RUN_PREPARATION_LEDGER_ERROR: str | None
 (
@@ -262,9 +296,7 @@ try:
         session_store=session_store,
         setup_registry=OBSERVATORY_LABORATORY_SETUP_REGISTRY,
         config=M49QueueBindingConfig.from_file(
-            REPOSITORY_ROOT
-            / "config"
-            / "observatory-m49-recorded-queue-binding.json"
+            REPOSITORY_ROOT / "config" / "observatory-m49-recorded-queue-binding.json"
         ),
     )
     OBSERVATORY_RECORDED_JOB_QUEUE = ObservatoryRecordedJobQueue(
@@ -279,6 +311,17 @@ except (M49QueueBindingError, ObservatoryRecordedQueueError, OSError, ValueError
     OBSERVATORY_RECORDED_BINDING_SERVICE = None
     OBSERVATORY_RECORDED_JOB_QUEUE = None
     OBSERVATORY_RECORDED_JOB_QUEUE_ERROR = str(exc)
+OBSERVATORY_WORKER_TOKEN_PATH = session_store.data_dir / "worker-auth" / "observatory-worker.token"
+OBSERVATORY_WORKER_CLAIM_LEASE_READY = False
+OBSERVATORY_WORKER_VERIFIED_RESULT_PUBLISHER_READY = False
+OBSERVATORY_WORKER_PRODUCTION_API_ENABLED = False
+OBSERVATORY_WORKER_AUTHENTICATION: ObservatoryWorkerAuthentication | None
+OBSERVATORY_WORKER_API_ERROR: str | None
+OBSERVATORY_WORKER_AUTHENTICATION = None
+OBSERVATORY_WORKER_API_ERROR = (
+    "Worker pull API is hard-disabled until claim leases and a verified "
+    "Observatory result publisher are implemented and accepted"
+)
 simulation_project_store = SimulationProjectStore(session_store.data_dir)
 simulation_project_service = SimulationProjectService(simulation_project_store)
 session_artifact_gateway = configured_artifact_gateway(session_store.data_dir)
@@ -293,6 +336,39 @@ session_recording_materializer = SessionRecordingMaterializer(
 session_recorded_media_inspector = RecordedMediaInspector(
     session_store.data_dir / "recorded-media-preparations"
 )
+OBSERVATORY_PORTABLE_SETUP_PROJECTOR: PortableLabV1SetupProjector | None
+OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR: str | None
+try:
+    portable_definition_registry = PortableRunDefinitionRegistry.from_file(
+        REPOSITORY_ROOT / "config" / "observatory-portable-run-definitions.json"
+    )
+    portable_lab_v1_definition = next(
+        definition
+        for definition in portable_definition_registry.definitions
+        if definition.setup_id == PORTABLE_LAB_V1_SETUP_ID
+    )
+    portable_source_capability_service = RecordedK1SourceAdmissionService(
+        data_dir=session_store.data_dir,
+        session_store=session_store,
+        media_inspector=session_recorded_media_inspector,
+        requirements=portable_lab_v1_definition.to_source_admission_requirements(),
+    )
+    OBSERVATORY_PORTABLE_SETUP_PROJECTOR = PortableLabV1SetupProjector(
+        registry=portable_definition_registry,
+        capability_probe=portable_source_capability_service,
+    )
+    OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR = None
+except (
+    PortableRunDefinitionRegistryError,
+    PortableSetupProjectionError,
+    OSError,
+    StopIteration,
+    ValueError,
+) as exc:
+    # Portable LAB V1 is an optional observation-only slice.  A drifted
+    # registry cannot affect K1, Simulation, legacy LAB, or the exact M49 queue.
+    OBSERVATORY_PORTABLE_SETUP_PROJECTOR = None
+    OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR = str(exc)
 _ffmpeg = _resolve_media_tool("ffmpeg")
 _ffprobe = _resolve_media_tool("ffprobe")
 session_recorded_camera_frame_service = (
@@ -765,8 +841,23 @@ app.include_router(
         recorded_binding_service=OBSERVATORY_RECORDED_BINDING_SERVICE,
         recorded_job_queue=OBSERVATORY_RECORDED_JOB_QUEUE,
         recorded_job_queue_error=OBSERVATORY_RECORDED_JOB_QUEUE_ERROR,
+        portable_setup_projector=OBSERVATORY_PORTABLE_SETUP_PROJECTOR,
+        portable_setup_projector_error=OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR,
     )
 )
+if (
+    OBSERVATORY_WORKER_PRODUCTION_API_ENABLED
+    and OBSERVATORY_WORKER_CLAIM_LEASE_READY
+    and OBSERVATORY_WORKER_VERIFIED_RESULT_PUBLISHER_READY
+    and OBSERVATORY_RECORDED_JOB_QUEUE is not None
+    and OBSERVATORY_WORKER_AUTHENTICATION is not None
+):
+    app.include_router(
+        build_observatory_worker_router(
+            OBSERVATORY_RECORDED_JOB_QUEUE,
+            authentication=OBSERVATORY_WORKER_AUTHENTICATION,
+        )
+    )
 app.include_router(
     build_environment_router(root_provider=lambda: session_store.data_dir / "ui-environment")
 )
@@ -1066,10 +1157,7 @@ app.include_router(
         spatial_evidence_provider=m48_raw_evidence_reader,
         evaluation_runner=LABORATORY_RUNNER,
         evaluation_receipt_root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "laboratory-run-receipts"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "laboratory-run-receipts"
         ),
     )
 )
@@ -1093,33 +1181,21 @@ app.include_router(
 app.include_router(
     build_m49_tgs_fail_closed_router(
         root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "m49"
-            / "tgs-fail-closed-results"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "m49" / "tgs-fail-closed-results"
         ),
     )
 )
 app.include_router(
     build_m49_tgs_full_shadow_router(
         root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "m49"
-            / "tgs-full-shadow-results"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "m49" / "tgs-full-shadow-results"
         ),
     )
 )
 app.include_router(
     build_vegetation_shadow_lab_router(
         root_provider=lambda: (
-            REPOSITORY_ROOT
-            / ".runtime"
-            / "compute-experiments"
-            / "lab-v1-vegetation"
-            / "results"
+            REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "lab-v1-vegetation" / "results"
         ),
         canonical_recording_provider=_canonical_lab_recording_source,
         camera_frame_provider=(
@@ -1128,9 +1204,7 @@ app.include_router(
             else None
         ),
         jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
-        rerun_overlay_cache_root=(
-            session_store.data_dir / "laboratory-rerun-overlays"
-        ),
+        rerun_overlay_cache_root=(session_store.data_dir / "laboratory-rerun-overlays"),
         ffmpeg_path=_ffmpeg,
     )
 )

src/k1link/web/observatory_api.py

@@ -22,6 +22,10 @@ from k1link.observatory.m49_queue_binding import (
     M49QueueBindingIntegrityError,
     M49RecordedQueueBindingService,
 )
+from k1link.observatory.portable_setup_projection import (
+    PortableLabV1SetupProjector,
+    PortableSetupProjectionError,
+)
 from k1link.observatory.recorded_jobs import (
     ObservatoryRecordedJobQueue,
     ObservatoryRecordedQueueCapacityError,
@@ -32,24 +36,24 @@ from k1link.observatory.recorded_jobs import (
 from k1link.sessions import SessionIntegrityError, SessionNotFoundError, SessionStore
 from k1link.sessions.models import SessionSummary
 
-OBSERVATORY_PROJECTION_SCHEMA: Literal[
+OBSERVATORY_PROJECTION_SCHEMA: Literal["missioncore.observatory-lab-projection/v1"] = (
     "missioncore.observatory-lab-projection/v1"
-] = "missioncore.observatory-lab-projection/v1"
-OBSERVATORY_RENAME_SCHEMA: Literal[
+)
+OBSERVATORY_RENAME_SCHEMA: Literal["missioncore.observatory-lab-projection-rename/v1"] = (
     "missioncore.observatory-lab-projection-rename/v1"
-] = "missioncore.observatory-lab-projection-rename/v1"
+)
 OBSERVATORY_RUN_PREFLIGHT_REQUEST_SCHEMA: Literal[
     "missioncore.observatory-run-preflight-request/v1"
 ] = "missioncore.observatory-run-preflight-request/v1"
-OBSERVATORY_RUN_PREFLIGHT_SCHEMA: Literal[
+OBSERVATORY_RUN_PREFLIGHT_SCHEMA: Literal["missioncore.observatory-run-preflight/v1"] = (
     "missioncore.observatory-run-preflight/v1"
-] = "missioncore.observatory-run-preflight/v1"
+)
 OBSERVATORY_RECORDED_RUN_SUBMIT_SCHEMA: Literal[
     "missioncore.observatory-recorded-run-submit/v1"
 ] = "missioncore.observatory-recorded-run-submit/v1"
-OBSERVATORY_RECORDED_JOB_LIST_SCHEMA: Literal[
+OBSERVATORY_RECORDED_JOB_LIST_SCHEMA: Literal["missioncore.observatory-recorded-job-list/v1"] = (
     "missioncore.observatory-recorded-job-list/v1"
-] = "missioncore.observatory-recorded-job-list/v1"
+)
 
 _OBSERVATION_ONLY_AUTHORITY: dict[str, bool] = {
     "commands_enabled": False,
@@ -64,9 +68,7 @@ class _StrictApiModel(BaseModel):
 
 
 class ObservatoryProjectionRenameRequest(_StrictApiModel):
-    schema_version: Literal[
-        "missioncore.observatory-lab-projection-rename/v1"
-    ]
+    schema_version: Literal["missioncore.observatory-lab-projection-rename/v1"]
     display_name: str = Field(min_length=1, max_length=160)
 
 
@@ -92,9 +94,7 @@ class ObservatoryRunPreflightRequest(_StrictApiModel):
 
 
 class ObservatoryRunPreparationRequest(_StrictApiModel):
-    schema_version: Literal[
-        "missioncore.observatory-run-preparation-request/v1"
-    ]
+    schema_version: Literal["missioncore.observatory-run-preparation-request/v1"]
     idempotency_key: str = Field(
         min_length=1,
         max_length=160,
@@ -142,6 +142,8 @@ def build_observatory_router(
     recorded_binding_service: M49RecordedQueueBindingService | None = None,
     recorded_job_queue: ObservatoryRecordedJobQueue | None = None,
     recorded_job_queue_error: str | None = None,
+    portable_setup_projector: PortableLabV1SetupProjector | None = None,
+    portable_setup_projector_error: str | None = None,
 ) -> APIRouter:
     """Build bounded catalog-only mutations for typed Observatory projections."""
 
@@ -212,6 +214,41 @@ def build_observatory_router(
                 available.add(result_id)
         return frozenset(available)
 
+    if portable_setup_projector is not None:
+
+        @router.get("/api/v1/observatory/portable-laboratory-setups")
+        def list_observatory_portable_laboratory_setups(
+            source_session_id: str = Query(
+                min_length=1,
+                max_length=128,
+                pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$",
+            ),
+        ) -> dict[str, object]:
+            source = source_summary(source_session_id)
+            try:
+                return portable_setup_projector.catalog(source)
+            except PortableSetupProjectionError as exc:
+                raise HTTPException(
+                    status_code=503,
+                    detail="Portable-каталог LAB V1 нарушил контракт целостности.",
+                ) from exc
+
+    elif portable_setup_projector_error is not None:
+
+        @router.get("/api/v1/observatory/portable-laboratory-setups")
+        def unavailable_observatory_portable_laboratory_setups(
+            source_session_id: str = Query(
+                min_length=1,
+                max_length=128,
+                pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$",
+            ),
+        ) -> None:
+            del source_session_id
+            raise HTTPException(
+                status_code=503,
+                detail="Portable-каталог LAB V1 недоступен.",
+            )
+
     if setup_registry is not None:
 
         @router.get("/api/v1/observatory/laboratory-setups")
@@ -225,9 +262,7 @@ def build_observatory_router(
             source = source_summary(source_session_id)
             return setup_registry.catalog(
                 source,
-                available_observatory_result_ids=available_observatory_results(
-                    source_session_id
-                ),
+                available_observatory_result_ids=available_observatory_results(source_session_id),
             )
 
         @router.post("/api/v1/observatory/run-preflights")
@@ -254,9 +289,7 @@ def build_observatory_router(
             )
             definition = projected["run_definition"]
             expected_digest = (
-                definition.get("definition_sha256")
-                if isinstance(definition, dict)
-                else None
+                definition.get("definition_sha256") if isinstance(definition, dict) else None
             )
             if request.definition_sha256 != expected_digest:
                 raise HTTPException(
@@ -276,8 +309,7 @@ def build_observatory_router(
             binding_service = recorded_binding_service
             exact_queue_setup = (
                 binding_service is not None
-                and request.setup_id
-                == binding_service.config.setup.setup_id
+                and request.setup_id == binding_service.config.setup.setup_id
             )
             if (
                 compatible
@@ -303,9 +335,7 @@ def build_observatory_router(
                 {
                     "check_id": "source-compatibility",
                     "outcome": "pass" if compatible else "fail",
-                    "reason_code": (
-                        "source-compatible" if compatible else "source-incompatible"
-                    ),
+                    "reason_code": ("source-compatible" if compatible else "source-incompatible"),
                     "message": (
                         "Источник точно совместим с сохранённым сетапом."
                         if compatible
@@ -323,11 +353,7 @@ def build_observatory_router(
                 {
                     "check_id": "executor",
                     "outcome": (
-                        "not-applicable"
-                        if existing
-                        else "pass"
-                        if queue_binding_ready
-                        else "fail"
+                        "not-applicable" if existing else "pass" if queue_binding_ready else "fail"
                     ),
                     "reason_code": (
                         "existing-result-does-not-require-executor"
@@ -349,13 +375,7 @@ def build_observatory_router(
                 },
                 {
                     "check_id": "durable-queue",
-                    "outcome": (
-                        "not-applicable"
-                        if existing
-                        else "pass"
-                        if queueable
-                        else "fail"
-                    ),
+                    "outcome": ("not-applicable" if existing else "pass" if queueable else "fail"),
                     "reason_code": (
                         "existing-result-does-not-require-queue"
                         if existing
@@ -378,9 +398,7 @@ def build_observatory_router(
                 "source_session_id": request.source_session_id,
                 "setup_id": request.setup_id,
                 "definition_sha256": expected_digest,
-                "outcome": (
-                    "existing" if existing else "queueable" if queueable else "blocked"
-                ),
+                "outcome": ("existing" if existing else "queueable" if queueable else "blocked"),
                 "submission_allowed": queueable,
                 "checks": checks,
                 "existing_result_ids": preflight.get("existing_result_ids", []),
@@ -421,9 +439,7 @@ def build_observatory_router(
             request: ObservatoryRunPreparationRequest,
         ) -> dict[str, object]:
             try:
-                existing = run_preparation_ledger.get_by_idempotency_key(
-                    request.idempotency_key
-                )
+                existing = run_preparation_ledger.get_by_idempotency_key(request.idempotency_key)
                 request_sha256 = observatory_run_preparation_request_sha256(
                     idempotency_key=request.idempotency_key,
                     source_session_id=request.source_session_id,
@@ -634,9 +650,7 @@ def build_observatory_router(
             request: ObservatoryRecordedRunSubmitRequest,
         ) -> dict[str, object]:
             try:
-                existing_job = recorded_job_queue.get_by_idempotency_key(
-                    request.idempotency_key
-                )
+                existing_job = recorded_job_queue.get_by_idempotency_key(request.idempotency_key)
             except ObservatoryRecordedQueueNotFoundError:
                 existing_job = None
             except (ObservatoryRecordedQueueError, ValueError) as exc:

src/k1link/web/observatory_worker_api.py

@@ -0,0 +1,343 @@
+"""Authenticated pull transport for the durable Observatory recorded-job queue.
+
+The transport is deliberately narrower than an execution API.  Worker callers
+can claim a server-sealed RunDefinition and advance its durable state, but they
+cannot supply commands, paths, environment variables, container images, or
+priority.  Those execution identities remain part of the queue-owned job.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import os
+import re
+import stat
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Annotated, Final, Literal
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Response
+from fastapi import Path as ApiPath
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from pydantic import BaseModel, ConfigDict, Field
+
+from k1link.observatory.recorded_jobs import (
+    ObservatoryRecordedCheckpointError,
+    ObservatoryRecordedJobQueue,
+    ObservatoryRecordedPreemptionError,
+    ObservatoryRecordedQueueBusyError,
+    ObservatoryRecordedQueueCapacityError,
+    ObservatoryRecordedQueueConflictError,
+    ObservatoryRecordedQueueError,
+    ObservatoryRecordedQueueIntegrityError,
+    ObservatoryRecordedQueueNotFoundError,
+    ObservatoryRecordedQueueStaleClaimError,
+)
+
+OBSERVATORY_WORKER_CLAIM_REQUEST_SCHEMA: Final = "missioncore.observatory-worker-claim-request/v1"
+OBSERVATORY_WORKER_START_REQUEST_SCHEMA: Final = "missioncore.observatory-worker-start-request/v1"
+OBSERVATORY_WORKER_CHECKPOINT_REQUEST_SCHEMA: Final = (
+    "missioncore.observatory-worker-checkpoint-request/v1"
+)
+OBSERVATORY_WORKER_SUCCEED_REQUEST_SCHEMA: Final = (
+    "missioncore.observatory-worker-succeed-request/v1"
+)
+OBSERVATORY_WORKER_FAIL_REQUEST_SCHEMA: Final = "missioncore.observatory-worker-fail-request/v1"
+OBSERVATORY_WORKER_CONTOUR_HEADER: Final = "X-Mission-Core-Contour-Id"
+
+_IDENTIFIER = re.compile(r"^[a-z][a-z0-9-]{2,95}$")
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_JOB_ID_PATTERN = r"^observatory-run-[a-f0-9]{32}$"
+_CLAIM_TOKEN_PATTERN = r"^[a-f0-9]{64}$"
+_CLAIM_REQUEST_ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$"
+_SESSION_ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
+_IDENTIFIER_PATTERN = r"^[a-z][a-z0-9-]{2,95}$"
+_WORKER_BEARER = HTTPBearer(auto_error=False)
+_TOKEN = re.compile(r"^[A-Za-z0-9._:-]{32,512}$")
+
+
+@dataclass(frozen=True, slots=True)
+class ObservatoryWorkerAuthentication:
+    """Server-owned Worker identity and the SHA-256 of its bearer secret."""
+
+    bearer_token_sha256: str
+    contour_id: str = "worker-006"
+
+    def __post_init__(self) -> None:
+        if _SHA256.fullmatch(self.bearer_token_sha256) is None:
+            raise ValueError("Worker bearer token SHA-256 is invalid")
+        if _IDENTIFIER.fullmatch(self.contour_id) is None:
+            raise ValueError("Worker contour id is invalid")
+
+
+def load_observatory_worker_authentication(
+    token_path: Path,
+    *,
+    contour_id: str = "worker-006",
+) -> ObservatoryWorkerAuthentication:
+    """Load one local Worker credential without retaining its plaintext.
+
+    The credential file is an operator/deployment concern.  Mission Core only
+    retains its SHA-256 in the router configuration and refuses symlinks,
+    non-regular files, or group/other permissions.
+    """
+
+    candidate = token_path.expanduser().absolute()
+    descriptor: int | None = None
+    try:
+        descriptor = os.open(
+            candidate,
+            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 ValueError("Worker bearer credential must be a regular file")
+        if metadata.st_mode & 0o077:
+            raise ValueError("Worker bearer credential permissions are too broad")
+        if not 32 <= metadata.st_size <= 512:
+            raise ValueError("Worker bearer credential format is invalid")
+        with os.fdopen(descriptor, "rb") as stream:
+            descriptor = None
+            payload = stream.read(513)
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError("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 ValueError("Worker bearer credential is not ASCII") from exc
+    if _TOKEN.fullmatch(token) is None:
+        raise ValueError("Worker bearer credential format is invalid")
+    return ObservatoryWorkerAuthentication(
+        bearer_token_sha256=hashlib.sha256(payload).hexdigest(),
+        contour_id=contour_id,
+    )
+
+
+class _StrictWorkerRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+class ObservatoryWorkerClaimRequest(_StrictWorkerRequest):
+    schema_version: Literal["missioncore.observatory-worker-claim-request/v1"]
+    claim_request_id: str = Field(
+        min_length=1,
+        max_length=160,
+        pattern=_CLAIM_REQUEST_ID_PATTERN,
+    )
+
+
+class ObservatoryWorkerStartRequest(_StrictWorkerRequest):
+    schema_version: Literal["missioncore.observatory-worker-start-request/v1"]
+    claim_token: str = Field(pattern=_CLAIM_TOKEN_PATTERN)
+
+
+class ObservatoryWorkerCheckpointRequest(_StrictWorkerRequest):
+    schema_version: Literal["missioncore.observatory-worker-checkpoint-request/v1"]
+    claim_token: str = Field(pattern=_CLAIM_TOKEN_PATTERN)
+    checkpoint_id: str = Field(
+        min_length=3,
+        max_length=96,
+        pattern=_IDENTIFIER_PATTERN,
+    )
+
+
+class ObservatoryWorkerSucceedRequest(_StrictWorkerRequest):
+    schema_version: Literal["missioncore.observatory-worker-succeed-request/v1"]
+    claim_token: str = Field(pattern=_CLAIM_TOKEN_PATTERN)
+    result_id: str = Field(
+        min_length=1,
+        max_length=128,
+        pattern=_SESSION_ID_PATTERN,
+    )
+    result_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
+
+
+class ObservatoryWorkerFailRequest(_StrictWorkerRequest):
+    schema_version: Literal["missioncore.observatory-worker-fail-request/v1"]
+    claim_token: str = Field(pattern=_CLAIM_TOKEN_PATTERN)
+    error_code: str = Field(
+        min_length=3,
+        max_length=96,
+        pattern=_IDENTIFIER_PATTERN,
+    )
+    message: str = Field(min_length=1, max_length=1_000)
+
+
+def build_observatory_worker_router(
+    queue: ObservatoryRecordedJobQueue,
+    *,
+    authentication: ObservatoryWorkerAuthentication,
+) -> APIRouter:
+    """Build the bounded Worker pull/state-transition router.
+
+    ``authentication`` contains only a token digest.  The plaintext bearer
+    secret exists transiently while FastAPI parses one request, is immediately
+    hashed, and is compared to the configured digest in constant time.
+    """
+
+    def require_configured_worker(
+        credentials: Annotated[
+            HTTPAuthorizationCredentials | None,
+            Depends(_WORKER_BEARER),
+        ],
+        contour_id: Annotated[
+            str | None,
+            Header(alias=OBSERVATORY_WORKER_CONTOUR_HEADER),
+        ] = None,
+    ) -> None:
+        if credentials is None or credentials.scheme.lower() != "bearer":
+            raise _unauthorized()
+        token = credentials.credentials
+        if not token or len(token) > 512:
+            raise _unauthorized()
+        supplied_sha256 = hashlib.sha256(token.encode("utf-8")).hexdigest()
+        if not hmac.compare_digest(
+            supplied_sha256,
+            authentication.bearer_token_sha256,
+        ):
+            raise _unauthorized()
+        if contour_id is None or not hmac.compare_digest(
+            contour_id,
+            authentication.contour_id,
+        ):
+            raise HTTPException(
+                status_code=403,
+                detail="Worker contour identity was rejected.",
+            )
+
+    router = APIRouter(
+        prefix="/api/v1/worker/observatory",
+        tags=["observatory-worker"],
+        dependencies=[Depends(require_configured_worker)],
+    )
+
+    @router.post("/recorded-jobs/claims", response_model=None)
+    def claim_next(
+        request: ObservatoryWorkerClaimRequest,
+    ) -> dict[str, object] | Response:
+        claim = _queue_call(
+            lambda: queue.claim_next(
+                claimant_id=authentication.contour_id,
+                claim_request_id=request.claim_request_id,
+            )
+        )
+        if claim is None:
+            return Response(status_code=204)
+        return claim.as_dict()
+
+    @router.get("/recorded-jobs/{job_id}")
+    def get_job(
+        job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+    ) -> dict[str, object]:
+        return _queue_call(lambda: queue.get(job_id)).as_dict()
+
+    @router.post("/recorded-jobs/{job_id}/start")
+    def start_job(
+        request: ObservatoryWorkerStartRequest,
+        job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+    ) -> dict[str, object]:
+        return _queue_call(lambda: queue.start(job_id, claim_token=request.claim_token)).as_dict()
+
+    @router.post("/recorded-jobs/{job_id}/checkpoint")
+    def checkpoint_job(
+        request: ObservatoryWorkerCheckpointRequest,
+        job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+    ) -> dict[str, object]:
+        return _queue_call(
+            lambda: queue.checkpoint(
+                job_id,
+                claim_token=request.claim_token,
+                checkpoint_id=request.checkpoint_id,
+            )
+        ).as_dict()
+
+    @router.post("/recorded-jobs/{job_id}/succeed")
+    def succeed_job(
+        request: ObservatoryWorkerSucceedRequest,
+        job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+    ) -> dict[str, object]:
+        return _queue_call(
+            lambda: queue.succeed(
+                job_id,
+                claim_token=request.claim_token,
+                result_id=request.result_id,
+                result_sha256=request.result_sha256,
+            )
+        ).as_dict()
+
+    @router.post("/recorded-jobs/{job_id}/fail")
+    def fail_job(
+        request: ObservatoryWorkerFailRequest,
+        job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+    ) -> dict[str, object]:
+        return _queue_call(
+            lambda: queue.fail(
+                job_id,
+                claim_token=request.claim_token,
+                error_code=request.error_code,
+                message=request.message,
+            )
+        ).as_dict()
+
+    return router
+
+
+def _unauthorized() -> HTTPException:
+    return HTTPException(
+        status_code=401,
+        detail="Worker bearer credential was rejected.",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+
+def _queue_call[T](operation: Callable[[], T]) -> T:
+    try:
+        return operation()
+    except ObservatoryRecordedQueueNotFoundError as exc:
+        raise HTTPException(
+            status_code=404,
+            detail="Recorded job was not found.",
+        ) from exc
+    except ObservatoryRecordedQueueStaleClaimError as exc:
+        raise HTTPException(
+            status_code=409,
+            detail="Recorded-job claim is stale.",
+        ) from exc
+    except ObservatoryRecordedCheckpointError as exc:
+        raise HTTPException(
+            status_code=409,
+            detail="Recorded-job checkpoint was rejected.",
+        ) from exc
+    except ObservatoryRecordedQueueConflictError as exc:
+        raise HTTPException(
+            status_code=409,
+            detail="Recorded-job transition conflicts with durable state.",
+        ) from exc
+    except (ObservatoryRecordedQueueBusyError, ObservatoryRecordedPreemptionError) as exc:
+        raise HTTPException(
+            status_code=409,
+            detail="Recorded-job resources are reserved for live K1 work.",
+        ) from exc
+    except ObservatoryRecordedQueueCapacityError as exc:
+        raise HTTPException(
+            status_code=503,
+            detail="Recorded-job queue capacity is unavailable.",
+            headers={"Retry-After": "5"},
+        ) from exc
+    except ObservatoryRecordedQueueIntegrityError as exc:
+        raise HTTPException(
+            status_code=503,
+            detail="Recorded-job queue integrity is unavailable.",
+        ) from exc
+    except ObservatoryRecordedQueueError as exc:
+        raise HTTPException(
+            status_code=503,
+            detail="Recorded-job queue is unavailable.",
+        ) from exc