feat(observatory): add installed package dispatch and durable publication

Checkpoint existing backend lifecycle changes. Focused verification found nine legacy fixture failures in portable LAB V1 executor/runtime tests; repair follows separately without rewriting this snapshot. ADR date retains its intentional Markdown hard break.
2026-09-02 00:58:37 +03:00
parent d655d6998d
commit a945d665dd
47 changed files with 6008 additions and 1034 deletions
@@ -47,6 +47,9 @@ from k1link.observatory.m49_queue_binding import (
    M49QueueBindingError,
    M49RecordedQueueBindingService,
 )
+from k1link.observatory.portable_publication_reconciler import (
+    PortablePublicationReconciler,
+)
 from k1link.observatory.portable_queue_binding import (
    PortableQueueBindingError,
    PortableRecordedQueueBindingService,
@@ -58,6 +61,7 @@ from k1link.observatory.portable_result_contract import (
 from k1link.observatory.portable_result_publisher import (
    resolve_published_portable_calculation_profile,
 )
+from k1link.observatory.portable_result_view import PortableResultViewService
 from k1link.observatory.portable_run_definitions import (
    PortableRunDefinitionRegistry,
    PortableRunDefinitionRegistryError,
@@ -478,6 +482,18 @@ except (PortableWorkerIntegrationError, OSError, ValueError) as exc:
    # Failure remains isolated from K1, Simulation and legacy LAB.
    OBSERVATORY_PORTABLE_WORKER_INTEGRATION = None
    OBSERVATORY_PORTABLE_WORKER_INTEGRATION_ERROR = str(exc)
+OBSERVATORY_PUBLICATION_RECONCILER = (
+    None
+    if (
+        OBSERVATORY_RECORDED_JOB_QUEUE is None
+        or OBSERVATORY_PORTABLE_WORKER_INTEGRATION is None
+    )
+    else PortablePublicationReconciler(
+        queue=OBSERVATORY_RECORDED_JOB_QUEUE,
+        artifact_transport=OBSERVATORY_PORTABLE_WORKER_INTEGRATION.artifact_transport,
+        result_publisher=OBSERVATORY_PORTABLE_WORKER_INTEGRATION.result_publisher,
+    )
+)
 OBSERVATORY_WORKER_API_GATE_ENABLED = OBSERVATORY_WORKER_LOCAL_ENABLED
 OBSERVATORY_WORKER_CLAIM_LEASE_READY = (
    OBSERVATORY_WORKER_API_GATE_ENABLED
@@ -539,6 +555,7 @@ try:
        registry=OBSERVATORY_PORTABLE_DEFINITION_REGISTRY,
        capability_probe=OBSERVATORY_PORTABLE_BINDING_SERVICE,
        dispatch_available=OBSERVATORY_WORKER_DISPATCH_READY,
+        equipment_capture_registry=session_store.equipment_capture_registry,
    )
    OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR = None
 except (
@@ -825,9 +842,22 @@ async def _recording_preparation_reconciler() -> None:
        await asyncio.sleep(2.0)


+async def _portable_result_publication_reconciler() -> None:
+    service = OBSERVATORY_PUBLICATION_RECONCILER
+    if service is None:
+        return
+    while True:
+        # Durable state remains pending/failed and is retried on the next
+        # bounded pass or through the explicit operator action.
+        with suppress(OSError, ValueError):
+            await asyncio.to_thread(service.run_once)
+        await asyncio.sleep(15.0)
+
+
@asynccontextmanager
 async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
    reconciler: asyncio.Task[None] | None = None
+    publication_reconciler: asyncio.Task[None] | None = None
    try:
        configure_scanner_diagnostics(session_store.data_dir / "logs")
        session_recording_preparation_manager.start()
@@ -841,6 +871,9 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
        # expensive on field captures. Start it immediately in the background
        # instead of holding the ASGI startup gate.
        reconciler = asyncio.create_task(_recording_preparation_reconciler())
+        publication_reconciler = asyncio.create_task(
+            _portable_result_publication_reconciler()
+        )
        yield
    finally:
        await map_gateway_proxy.close()
@@ -848,6 +881,10 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
            reconciler.cancel()
            with suppress(asyncio.CancelledError):
                await reconciler
+        if publication_reconciler is not None:
+            publication_reconciler.cancel()
+            with suppress(asyncio.CancelledError):
+                await publication_reconciler
        await asyncio.to_thread(session_recording_preparation_manager.close)
        await asyncio.to_thread(lidar_local_surface_read_service.close)
        plugin_environment.close()
@@ -1036,6 +1073,24 @@ app.include_router(
        portable_setup_projector=OBSERVATORY_PORTABLE_SETUP_PROJECTOR,
        portable_setup_projector_error=OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR,
        portable_binding_service=OBSERVATORY_PORTABLE_BINDING_SERVICE,
+        portable_result_view=(
+            None
+            if session_artifact_gateway is None
+            else PortableResultViewService(
+                sessions=session_store,
+                artifacts=session_artifact_gateway.store,
+            )
+        ),
+        portable_artifact_transport=(
+            None
+            if OBSERVATORY_PORTABLE_WORKER_INTEGRATION is None
+            else OBSERVATORY_PORTABLE_WORKER_INTEGRATION.artifact_transport
+        ),
+        portable_result_publisher=(
+            None
+            if OBSERVATORY_PORTABLE_WORKER_INTEGRATION is None
+            else OBSERVATORY_PORTABLE_WORKER_INTEGRATION.result_publisher
+        ),
    )
 )
 if OBSERVATORY_WORKER_DISPATCH_READY:
@@ -1,9 +1,10 @@
 from __future__ import annotations

-from typing import Any, Literal
+from typing import Annotated, Any, Literal

 from fastapi import APIRouter, HTTPException, Query, Response
 from fastapi import Path as ApiPath
+from fastapi.responses import JSONResponse
 from pydantic import BaseModel, ConfigDict, Field

 from k1link.observatory import (
@@ -22,12 +23,24 @@ from k1link.observatory.m49_queue_binding import (
    M49QueueBindingIntegrityError,
    M49RecordedQueueBindingService,
 )
+from k1link.observatory.portable_artifact_transport import (
+    PortableArtifactTransportError,
+    PortableObservatoryArtifactTransport,
+)
 from k1link.observatory.portable_queue_binding import (
    PortableQueueBindingError,
    PortableQueueBindingIntegrityError,
    PortableQueueBindingStaleCheckError,
    PortableRecordedQueueBindingService,
 )
+from k1link.observatory.portable_result_contract import PortableResultPublisherError
+from k1link.observatory.portable_result_publisher import (
+    PortableObservatoryResultPublisher,
+)
+from k1link.observatory.portable_result_view import (
+    PortableResultViewError,
+    PortableResultViewService,
+)
 from k1link.observatory.portable_run_definitions import (
    PortableRunDefinitionUnavailableError,
 )
@@ -158,10 +171,32 @@ def build_observatory_router(
    portable_setup_projector: PortableSetupProjector | PortableLabV1SetupProjector | None = None,
    portable_setup_projector_error: str | None = None,
    portable_binding_service: PortableRecordedQueueBindingService | None = None,
+    portable_result_view: PortableResultViewService | None = None,
+    portable_artifact_transport: PortableObservatoryArtifactTransport | None = None,
+    portable_result_publisher: PortableObservatoryResultPublisher | None = None,
 ) -> APIRouter:
    """Build bounded catalog-only mutations for typed Observatory projections."""

    router = APIRouter(tags=["observatory"])
+    if (portable_artifact_transport is None) != (portable_result_publisher is None):
+        raise ValueError("portable publication dependencies must be configured together")
+
+    if portable_result_view is not None:
+
+        @router.get("/api/v1/observatory/portable-results/{result_id}")
+        def get_portable_result_view(
+            result_id: Annotated[
+                str,
+                ApiPath(pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"),
+            ],
+        ) -> dict[str, object]:
+            try:
+                return portable_result_view.read(result_id)
+            except PortableResultViewError as exc:
+                raise HTTPException(
+                    status_code=409,
+                    detail="Portable-результат недоступен для универсального просмотра.",
+                ) from exc

    def source_summary(session_id: str) -> SessionSummary:
        try:
@@ -1092,6 +1127,80 @@ def build_observatory_router(
                    detail="Durable-очередь расчётов недоступна.",
                ) from exc

+        if (
+            portable_artifact_transport is not None
+            and portable_result_publisher is not None
+        ):
+
+            @router.post(
+                "/api/v1/observatory/runs/{job_id}/publication/retry",
+                response_model=None,
+            )
+            def retry_observatory_result_publication(
+                job_id: str = ApiPath(
+                    min_length=48,
+                    max_length=48,
+                    pattern=r"^observatory-run-[a-f0-9]{32}$",
+                ),
+            ) -> dict[str, object] | JSONResponse:
+                """Retry verification/publication only; never repeat compute."""
+
+                try:
+                    job = recorded_job_queue.get(job_id)
+                except ObservatoryRecordedQueueNotFoundError as exc:
+                    raise HTTPException(
+                        status_code=404,
+                        detail="Расчёт Обсерватории не найден.",
+                    ) from exc
+                except (ObservatoryRecordedQueueError, ValueError) as exc:
+                    raise HTTPException(
+                        status_code=503,
+                        detail="Durable-очередь расчётов недоступна.",
+                    ) from exc
+                if job.publication_state == "published":
+                    return job.as_dict()
+                if job.state != "succeeded" or job.publication_state not in {
+                    "pending",
+                    "failed",
+                }:
+                    raise HTTPException(
+                        status_code=409,
+                        detail="Результат не ожидает повторной публикации.",
+                    )
+                try:
+                    package_root = portable_artifact_transport.package_root_for_terminal(
+                        job
+                    )
+                    portable_result_publisher.publish(
+                        job=job,
+                        package_root=package_root,
+                    )
+                except PortableResultPublisherError as exc:
+                    message = (" ".join(str(exc).split()) or "Publication failed.")[:1_000]
+                    try:
+                        failed = recorded_job_queue.mark_publication_failed(
+                            job_id,
+                            message=message,
+                        )
+                    except (ObservatoryRecordedQueueError, ValueError) as queue_exc:
+                        raise HTTPException(
+                            status_code=503,
+                            detail="Состояние публикации не удалось сохранить.",
+                        ) from queue_exc
+                    return JSONResponse(status_code=202, content=failed.as_dict())
+                except PortableArtifactTransportError as exc:
+                    raise HTTPException(
+                        status_code=409,
+                        detail="Пакет результата недоступен для повторной публикации.",
+                    ) from exc
+                try:
+                    return recorded_job_queue.mark_published(job_id).as_dict()
+                except (ObservatoryRecordedQueueError, ValueError) as exc:
+                    raise HTTPException(
+                        status_code=503,
+                        detail="Состояние публикации не удалось сохранить.",
+                    ) from exc
+
    elif recorded_job_queue_error is not None:

        @router.post("/api/v1/observatory/runs")
@@ -20,9 +20,9 @@ from typing import Annotated, Final, Literal

 from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response
 from fastapi import Path as ApiPath
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, JSONResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator

 from k1link.observatory.portable_artifact_transport import (
    MAX_RESULT_MANIFEST_BYTES,
@@ -38,6 +38,7 @@ from k1link.observatory.portable_result_publisher import (
    PortableObservatoryResultPublisher,
 )
 from k1link.observatory.recorded_jobs import (
+    MAX_RECORDED_EXECUTOR_CAPABILITIES,
    ObservatoryRecordedCheckpointError,
    ObservatoryRecordedJobQueue,
    ObservatoryRecordedPreemptionError,
@@ -48,9 +49,12 @@ from k1link.observatory.recorded_jobs import (
    ObservatoryRecordedQueueIntegrityError,
    ObservatoryRecordedQueueNotFoundError,
    ObservatoryRecordedQueueStaleClaimError,
+    RecordedExecutorIdentity,
 )

-OBSERVATORY_WORKER_CLAIM_REQUEST_SCHEMA: Final = "missioncore.observatory-worker-claim-request/v1"
+OBSERVATORY_WORKER_CAPABILITY_CLAIM_REQUEST_SCHEMA: Final = (
+    "missioncore.observatory-worker-claim-request/v2"
+)
 OBSERVATORY_WORKER_START_REQUEST_SCHEMA: Final = "missioncore.observatory-worker-start-request/v1"
 OBSERVATORY_WORKER_RENEW_REQUEST_SCHEMA: Final = "missioncore.observatory-worker-renew-request/v1"
 OBSERVATORY_WORKER_CHECKPOINT_REQUEST_SCHEMA: Final = (
@@ -143,13 +147,36 @@ class _StrictWorkerRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)


+class ObservatoryWorkerExecutorCapability(_StrictWorkerRequest):
+    release_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
+    image_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
+    model_manifest_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
+    resource_profile_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
+
+    def recorded_identity(self) -> RecordedExecutorIdentity:
+        return RecordedExecutorIdentity(**self.model_dump())
+
+
 class ObservatoryWorkerClaimRequest(_StrictWorkerRequest):
-    schema_version: Literal["missioncore.observatory-worker-claim-request/v1"]
+    schema_version: Literal["missioncore.observatory-worker-claim-request/v2"]
    claim_request_id: str = Field(
        min_length=1,
        max_length=160,
        pattern=_CLAIM_REQUEST_ID_PATTERN,
    )
+    supported_executor_identities: tuple[ObservatoryWorkerExecutorCapability, ...] = Field(
+        max_length=MAX_RECORDED_EXECUTOR_CAPABILITIES,
+    )
+
+    @model_validator(mode="after")
+    def validate_capability_snapshot(self) -> ObservatoryWorkerClaimRequest:
+        identities = tuple(
+            capability.recorded_identity()
+            for capability in self.supported_executor_identities
+        )
+        if len(identities) != len(set(identities)):
+            raise ValueError("executor capabilities must be unique")
+        return self


 class ObservatoryWorkerStartRequest(_StrictWorkerRequest):
@@ -257,6 +284,10 @@ def build_observatory_worker_router(
            lambda: queue.claim_next(
                claimant_id=authentication.contour_id,
                claim_request_id=request.claim_request_id,
+                supported_executor_identities=tuple(
+                    capability.recorded_identity()
+                    for capability in request.supported_executor_identities
+                ),
            )
        )
        if claim is None:
@@ -303,11 +334,11 @@ def build_observatory_worker_router(
            )
        ).as_dict()

-    @router.post("/recorded-jobs/{job_id}/succeed")
+    @router.post("/recorded-jobs/{job_id}/succeed", response_model=None)
    def succeed_job(
        request: ObservatoryWorkerSucceedRequest,
        job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
-    ) -> dict[str, object]:
+    ) -> dict[str, object] | JSONResponse:
        if artifact_transport is not None:
            _artifact_call(
                lambda: artifact_transport.require_completed_for_success(
@@ -318,29 +349,76 @@ def build_observatory_worker_router(
                    claimant_id=authentication.contour_id,
                )
            )
-        succeeded = _queue_call(
-            lambda: queue.succeed(
-                job_id,
-                claim_token=request.claim_token,
-                result_id=request.result_id,
-                result_sha256=request.result_sha256,
-            )
-        )
        if artifact_transport is not None and result_publisher is not None:
+            succeeded = _queue_call(
+                lambda: queue.complete_for_publication(
+                    job_id,
+                    claim_token=request.claim_token,
+                    result_id=request.result_id,
+                    result_sha256=request.result_sha256,
+                )
+            )
+            if succeeded.publication_state == "published":
+                return succeeded.as_dict()
            package_root = _artifact_call(
                lambda: artifact_transport.package_root_for_terminal(succeeded)
            )
            try:
                result_publisher.publish(job=succeeded, package_root=package_root)
            except PortableResultPublisherError as exc:
+                publication_error = _publication_error(exc)
+                failed = _queue_call(
+                    lambda: queue.mark_publication_failed(
+                        job_id,
+                        message=publication_error,
+                    )
+                )
+                return JSONResponse(status_code=202, content=failed.as_dict())
+            return _queue_call(lambda: queue.mark_published(job_id)).as_dict()
+        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()
+
+    if artifact_transport is not None and result_publisher is not None:
+
+        @router.post(
+            "/recorded-jobs/{job_id}/publication/retry",
+            response_model=None,
+        )
+        def retry_publication(
+            job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+        ) -> dict[str, object] | JSONResponse:
+            job = _queue_call(lambda: queue.get(job_id))
+            if job.publication_state == "published":
+                return job.as_dict()
+            if job.state != "succeeded" or job.publication_state not in {
+                "pending",
+                "failed",
+            }:
                raise HTTPException(
-                    status_code=503,
-                    detail=(
-                        "Recorded result is sealed but its verified publication "
-                        "requires reconciliation."
-                    ),
-                ) from exc
-        return succeeded.as_dict()
+                    status_code=409,
+                    detail="Recorded result is not awaiting publication.",
+                )
+            package_root = _artifact_call(
+                lambda: artifact_transport.package_root_for_terminal(job)
+            )
+            try:
+                result_publisher.publish(job=job, package_root=package_root)
+            except PortableResultPublisherError as exc:
+                publication_error = _publication_error(exc)
+                failed = _queue_call(
+                    lambda: queue.mark_publication_failed(
+                        job_id,
+                        message=publication_error,
+                    )
+                )
+                return JSONResponse(status_code=202, content=failed.as_dict())
+            return _queue_call(lambda: queue.mark_published(job_id)).as_dict()

    @router.post("/recorded-jobs/{job_id}/fail")
    def fail_job(
@@ -650,6 +728,11 @@ def _raise(exc: Exception) -> None:
    raise exc


+def _publication_error(exc: PortableResultPublisherError) -> str:
+    message = " ".join(str(exc).split())
+    return (message or "Portable result publication failed.")[:1_000]
+
+
 async def _read_bounded_body(request: Request, maximum_bytes: int) -> bytes:
    content_length = request.headers.get("content-length")
    if content_length is not None: