feat(observatory): verify exact published result reuse

apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts

@@ -1,4 +1,5 @@
 import type {
+  ObservatoryLaboratoryPreservedResult,
   ObservatoryLaboratoryRunDefinition,
   ObservatoryLaboratorySetup,
   ObservatoryLaboratorySetupCatalog,
@@ -23,13 +24,14 @@ export function decodePortableCatalog(value: unknown): ObservatoryLaboratorySetu
   );
   exact(row.schema_version, PORTABLE_CATALOG_SCHEMA, "schema_version portable-каталога");
   observationAuthority(row.authority);
+  const sourceSessionId = text(row.source_session_id, "portable source_session_id");
   return {
-    sourceSessionId: text(row.source_session_id, "portable source_session_id"),
-    setups: array(row.setups, "portable setups").map(decodePortableSetup),
+    sourceSessionId,
+    setups: array(row.setups, "portable setups").map((item) => decodePortableSetup(item, sourceSessionId)),
   };
 }
 
-function decodePortableSetup(value: unknown): ObservatoryLaboratorySetup {
+function decodePortableSetup(value: unknown, sourceSessionId: string): ObservatoryLaboratorySetup {
   const row = record(value, "portable-сетап");
   exactKeys(row, [
     "authority", "description", "display_name", "executor", "existing_results",
@@ -39,6 +41,8 @@ function decodePortableSetup(value: unknown): ObservatoryLaboratorySetup {
   exact(row.origin, "portable-definition", "portable origin");
   observationAuthority(row.authority);
   decodePortableSourceRequirements(row.source_requirements);
+  const setupId = text(row.setup_id, "portable setup_id");
+  const runDefinition = decodePortableRunDefinition(row.run_definition);
 
   const compatibility = record(row.source_compatibility, "portable source_compatibility");
   exactKeys(
@@ -96,12 +100,12 @@ function decodePortableSetup(value: unknown): ObservatoryLaboratorySetup {
   ], "portable preflight");
   const preflightOutcome = oneOf(
     preflight.outcome,
-    ["ready", "blocked"] as const,
+    ["existing", "ready", "blocked"] as const,
     "portable preflight outcome",
   );
   const preflightAction = oneOf(
     preflight.action,
-    ["check", "blocked"] as const,
+    ["open-existing", "check", "blocked"] as const,
     "portable preflight action",
   );
   const submissionAllowed = boolean(
@@ -112,28 +116,37 @@ function decodePortableSetup(value: unknown): ObservatoryLaboratorySetup {
     submissionAllowed !== (preflightOutcome === "ready")
     || (preflightOutcome === "ready" && preflightAction !== "check")
     || (preflightOutcome === "blocked" && preflightAction !== "blocked")
+    || (preflightOutcome === "existing" && preflightAction !== "open-existing")
+    || (submissionAllowed && (!compatible || !executorReady))
   ) {
     throw new ObservatoryPortableSetupDecodeError(
       "Portable preflight: состояние запуска противоречиво.",
     );
   }
-  const existingResults = array(row.existing_results, "portable existing_results");
+  const existingResults = array(row.existing_results, "portable existing_results").map(
+    (item) => decodePortableResult(item, sourceSessionId, setupId, runDefinition),
+  );
   const existingResultIds = array(
     preflight.existing_result_ids,
     "portable existing_result_ids",
-  );
-  if (existingResults.length > 0 || existingResultIds.length > 0) {
+  ).map((item) => text(item, "portable existing result id"));
+  if (
+    (existingResults.length > 0) !== (preflightOutcome === "existing")
+    || existingResultIds.length !== existingResults.length
+    || new Set(existingResultIds).size !== existingResultIds.length
+    || existingResults.some((result, index) => result.resultId !== existingResultIds[index])
+  ) {
     throw new ObservatoryPortableSetupDecodeError(
-      "Portable result: проверяемая привязка результата к RunDefinition ещё не поддерживается.",
+      "Portable result: готовность не соответствует проверенным результатам.",
     );
   }
 
   return {
-    setupId: text(row.setup_id, "portable setup_id"),
+    setupId,
     displayName: text(row.display_name, "portable display_name"),
     description: text(row.description, "portable description"),
     origin: "portable-definition",
-    runDefinition: decodePortableRunDefinition(row.run_definition),
+    runDefinition,
     compatibility: {
       compatible,
       reasons: compatible
@@ -146,17 +159,58 @@ function decodePortableSetup(value: unknown): ObservatoryLaboratorySetup {
       reasonCode: executorReasonCode,
       reason: executorReason,
     },
-    preservedResults: [],
+    preservedResults: existingResults,
     preflight: {
       outcome: preflightOutcome,
       action: preflightAction,
       reason: text(preflight.reason, "portable preflight reason"),
       submissionAllowed,
-      existingResultIds: [],
+      existingResultIds,
     },
   };
 }
 
+function decodePortableResult(
+  value: unknown, sourceSessionId: string, setupId: string,
+  definition: ObservatoryLaboratoryRunDefinition,
+): ObservatoryLaboratoryPreservedResult {
+  const row = record(value, "portable result");
+  exactKeys(row, [
+    "result_id", "result_kind", "relation", "access", "created_at_utc",
+    "observatory_projection_available", "identity",
+  ], "portable result");
+  exact(row.result_kind, definition.resultKind, "portable result kind");
+  exact(row.relation, "exact-recorded-computation", "portable result relation");
+  exact(row.access, "observatory", "portable result access");
+  exact(row.observatory_projection_available, true, "portable result availability");
+  const identity = record(row.identity, "portable result identity");
+  exactKeys(identity, [
+    "job_id", "source_session_id", "source_catalog_sha256", "source_bundle_sha256",
+    "source_capability_manifest_sha256", "setup_id", "definition_sha256",
+    "package_sha256", "artifact_manifest_id",
+  ], "portable result identity");
+  exact(identity.source_session_id, sourceSessionId, "portable result source");
+  exact(identity.setup_id, setupId, "portable result setup");
+  exact(identity.definition_sha256, definition.definitionSha256, "portable result definition");
+  text(identity.job_id, "portable result job");
+  for (const key of [
+    "source_catalog_sha256", "source_bundle_sha256", "source_capability_manifest_sha256",
+    "definition_sha256", "package_sha256", "artifact_manifest_id",
+  ]) {
+    if (!SHA256.test(text(identity[key], `portable result ${key}`))) {
+      throw new ObservatoryPortableSetupDecodeError(`Portable result: некорректный ${key}.`);
+    }
+  }
+  return {
+    resultId: text(row.result_id, "portable result id"),
+    resultKind: text(row.result_kind, "portable result kind"),
+    relation: "exact-recorded-computation",
+    access: "observatory",
+    createdAtUtc: text(row.created_at_utc, "portable result created_at_utc"),
+    observatoryProjectionAvailable: true,
+  };
+}
+
 function decodePortableSourceRequirements(value: unknown): void {
   const row = record(value, "portable source_requirements");
   exactKeys(row, [

apps/control-station/test/observatoryLaboratorySetups.test.mjs

@@ -323,7 +323,7 @@ test("portable LAB V1 rejects an unbound existing result projection", async () =
         authority,
       }), { status: 200 }),
     }),
-    /значение изменилось|значение не поддерживается|проверяемая привязка результата/,
+    /значение изменилось|значение не поддерживается|обнаружены неизвестные поля/,
   );
 });
 
@@ -348,6 +348,76 @@ test("portable LAB V1 rejects heavyweight compatibility evidence", async () => {
   );
 });
 
+function cachedPortableSetup() {
+  const setup = portableSetup();
+  setup.existing_results = [{
+    result_id: "portable-result-001",
+    result_kind: setup.run_definition.result_kind,
+    relation: "exact-recorded-computation",
+    access: "observatory",
+    created_at_utc: "2026-09-03T00:00:00Z",
+    observatory_projection_available: true,
+    identity: {
+      job_id: "observatory-run-001",
+      source_session_id: "source-a",
+      source_catalog_sha256: "a".repeat(64),
+      source_bundle_sha256: "b".repeat(64),
+      source_capability_manifest_sha256: "c".repeat(64),
+      setup_id: setup.setup_id,
+      definition_sha256: setup.run_definition.definition_sha256,
+      package_sha256: "d".repeat(64),
+      artifact_manifest_id: "e".repeat(64),
+    },
+  }];
+  setup.preflight = {
+    outcome: "existing", action: "open-existing",
+    reason: "Точный результат проверен.", submission_allowed: false,
+    existing_result_ids: [setup.existing_results[0].result_id],
+  };
+  return setup;
+}
+
+function fetchCachedPortable(setup) {
+  return fetchObservatoryPortableLaboratorySetups("source-a", {
+    fetcher: async () => new Response(JSON.stringify({
+      schema_version: "missioncore.observatory-portable-setup-catalog/v2",
+      source_session_id: "source-a", setups: [setup], authority,
+    }), { status: 200 }),
+  });
+}
+
+test("portable exact cached result is readable without an installed executor", async () => {
+  const catalog = await fetchCachedPortable(cachedPortableSetup());
+  const setup = catalog.setups[0];
+  assert.equal(setup.executor.state, "not-installed");
+  assert.equal(setup.preflight.outcome, "existing");
+  assert.equal(setup.preflight.submissionAllowed, false);
+  assert.deepEqual(setup.preflight.existingResultIds, ["portable-result-001"]);
+  assert.equal(setup.preservedResults[0].access, "observatory");
+});
+
+for (const [label, change] of [
+  ["another source", (s) => { s.existing_results[0].identity.source_session_id = "source-b"; }],
+  ["another setup", (s) => { s.existing_results[0].identity.setup_id = "another-profile"; }],
+  ["another version", (s) => { s.existing_results[0].identity.definition_sha256 = "f".repeat(64); }],
+  ["bad package digest", (s) => { s.existing_results[0].identity.package_sha256 = "not-a-digest"; }],
+  ["unavailable artifact", (s) => { s.existing_results[0].observatory_projection_available = false; }],
+  ["missing binding", (s) => { delete s.existing_results[0].identity; }],
+  ["unrelated result ID", (s) => { s.preflight.existing_result_ids = ["different-result"]; }],
+  ["duplicate result", (s) => {
+    s.existing_results.push(s.existing_results[0]);
+    s.preflight.existing_result_ids.push(s.preflight.existing_result_ids[0]);
+  }],
+  ["contradictory action", (s) => { s.preflight.action = "check"; }],
+  ["cached but queueable", (s) => { s.preflight.submission_allowed = true; }],
+]) {
+  test(`portable cached result rejects ${label}`, async () => {
+    const setup = cachedPortableSetup();
+    change(setup);
+    await assert.rejects(fetchCachedPortable(setup));
+  });
+}
+
 test("Observatory preflight sends the exact selected definition and never submits a run", async () => {
   const selected = (await fetchObservatoryLaboratorySetups("source-a", {
     fetcher: async () => new Response(JSON.stringify({

src/k1link/observatory/portable_queue_binding.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 import hashlib
 import json
 import re
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Final
@@ -109,6 +110,7 @@ class PortableRecordedQueueBindingService:
         media_inspector: RecordedMediaInspector,
         definitions: PortableRunDefinitionRegistry,
         queue: ObservatoryRecordedJobQueue | None = None,
+        published_result_available: Callable[[ObservatoryRecordedJob], bool] | None = None,
     ) -> None:
         self.data_dir = data_dir.expanduser().resolve()
         if self.data_dir != session_store.data_dir:
@@ -123,6 +125,7 @@ class PortableRecordedQueueBindingService:
         self._media_inspector = media_inspector
         self._definitions = definitions
         self._queue = queue
+        self._published_result_available = published_result_available
 
     def probe(
         self,
@@ -215,6 +218,7 @@ class PortableRecordedQueueBindingService:
             preparation.intent(idempotency_key=idempotency_key),
             enqueue=enqueue,
             reject_duplicate_computation=True,
+            published_result_available=self._published_result_available,
         )
 
     def _resolve_definition(

src/k1link/observatory/portable_result_cache.py

@@ -0,0 +1,305 @@
+"""Verified recorded-result reuse without inference or source replay preparation."""
+
+from __future__ import annotations
+
+import hashlib
+import stat
+from collections import OrderedDict
+from dataclasses import fields
+from pathlib import Path
+from threading import RLock
+
+from k1link.artifact_gateway import ArtifactGatewayError, CentralArtifactStore
+from k1link.observatory.portable_result_contract import (
+    PORTABLE_RESULT_PACKAGE_SCHEMA,
+    RESULT_DOCUMENT_ROLE,
+    RESULT_PACKAGE_MANIFEST_ROLE,
+    PortableCalculationProfileRegistry,
+    PortableResultPackageManifest,
+    PortableResultPublisherError,
+    job_identity_document,
+    result_identity_document,
+    run_definition_document,
+    source_identity_document,
+)
+from k1link.observatory.portable_result_publisher import (
+    resolve_published_portable_calculation_profile,
+)
+from k1link.observatory.portable_run_definitions import (
+    PortableRunDefinition,
+    PortableRunDefinitionRegistry,
+    PortableRunDefinitionRegistryError,
+)
+from k1link.observatory.recorded_jobs import (
+    ObservatoryRecordedJob,
+    ObservatoryRecordedJobQueue,
+    ObservatoryRecordedQueueConflictError,
+)
+from k1link.sessions import SessionIntegrityError, SessionNotFoundError, SessionStore
+
+
+class PortableResultCacheCheckRequired(ObservatoryRecordedQueueConflictError):
+    """Refresh verifies a newly published/changed object outside the queue write lock."""
+
+
+class PortableResultCache:
+    """Join the durable publication receipt, Session binding and immutable CAS.
+
+    Content checks are memoized by filesystem fingerprint, not unconditional availability.
+    Every lookup rechecks metadata. First use/changed metadata streams SHA-256
+    with bounded memory; no videos, RRDs or model tensors are kept in RAM.
+    """
+
+    def __init__(
+        self,
+        *,
+        sessions: SessionStore,
+        artifacts: CentralArtifactStore,
+        queue: ObservatoryRecordedJobQueue,
+        definitions: PortableRunDefinitionRegistry,
+        calculation_profiles: PortableCalculationProfileRegistry,
+    ) -> None:
+        if sessions.data_dir != queue.data_dir:
+            raise ValueError("portable result cache and queue roots disagree")
+        self._sessions = sessions
+        self._artifacts = artifacts
+        self._queue = queue
+        self._definitions = definitions
+        self._profiles = calculation_profiles
+        self._verified: OrderedDict[tuple[Path, str], tuple[tuple[int, ...], bool]] = OrderedDict()
+        self._lock = RLock()
+
+    def find(
+        self,
+        source_session_id: str,
+        definition: PortableRunDefinition,
+    ) -> tuple[dict[str, object], ...]:
+        """Return the newest usable exact result; older evidence remains untouched."""
+
+        try:
+            source, catalog_sha256 = self._sessions.get_session_with_catalog_snapshot(
+                source_session_id
+            )
+        except (SessionNotFoundError, SessionIntegrityError):
+            return ()
+        if source.summary.lab is not None:
+            return ()
+        candidates = self._queue.published_results(
+            source_session_id=source_session_id,
+            source_catalog_sha256=catalog_sha256,
+            setup_id=definition.setup_id,
+            definition_sha256=definition.definition_sha256,
+        )
+        for job in candidates:
+            result = self._projection(job)
+            if result is not None:
+                return (result,)
+        return ()
+
+    def available(self, job: ObservatoryRecordedJob) -> bool:
+        """Queue INSERT fence: verify this row without re-entering the queue DB."""
+
+        if not self._lock.acquire(blocking=False):
+            raise PortableResultCacheCheckRequired("published result verification is in progress")
+        try:
+            return self._projection(job, warm_only=True) is not None
+        finally:
+            self._lock.release()
+
+    def _projection(
+        self,
+        job: ObservatoryRecordedJob,
+        *,
+        warm_only: bool = False,
+    ) -> dict[str, object] | None:
+        if (
+            job.state != "succeeded"
+            or job.publication_state != "published"
+            or job.result_id is None
+            or job.result_sha256 is None
+            or job.terminal_code != "result-sealed"
+            or job.terminal_claim_token_sha256 is None
+            or job.claim_generation < 1
+            or job.active_claim_token is not None
+            or job.active_claimant_id is not None
+        ):
+            return None
+        try:
+            definition = self._definitions.resolve(job.setup_id, job.definition_sha256)
+            recorded = definition.to_recorded_run_definition()
+            if any(
+                getattr(recorded, field.name) != getattr(job, field.name)
+                for field in fields(recorded)
+            ):
+                return None
+            summary = self._sessions.get_session(job.result_id).summary
+            profile = self._profiles.resolve(definition)
+            if (
+                resolve_published_portable_calculation_profile(
+                    summary,
+                    definitions=self._definitions,
+                    calculation_profiles=self._profiles,
+                )
+                != profile.as_dict()
+            ):
+                return None
+            binding = summary.lab
+            assert binding is not None
+            provenance = binding.provenance
+            published_package = provenance.get("result_package")
+            if (
+                provenance.get("job") != job_identity_document(job)
+                or provenance.get("source") != source_identity_document(job)
+                or not isinstance(published_package, dict)
+            ):
+                return None
+            manifest_id = published_package.get("artifact_manifest_id")
+            if not isinstance(manifest_id, str):
+                return None
+            manifest = self._artifacts.read_manifest(manifest_id)
+            package_member = manifest.member(RESULT_PACKAGE_MANIFEST_ROLE)
+            if (
+                package_member.sha256 != job.result_sha256
+                or package_member.media_type != "application/json"
+                or not 0 < package_member.byte_length <= 1024 * 1024
+            ):
+                return None
+            package_path = self._artifacts.object_path(package_member.sha256)
+            # This small contract is parsed on every lookup, never trusted by stat alone.
+            _fingerprint(package_path, package_member.byte_length)
+            with package_path.open("rb") as stream:
+                payload = stream.read(1024 * 1024 + 1)
+            package = PortableResultPackageManifest.from_bytes(payload)
+            if (
+                len(payload) != package_member.byte_length
+                or package.manifest_sha256 != job.result_sha256
+                or package.job != job_identity_document(job)
+                or package.source != source_identity_document(job)
+                or package.run_definition != run_definition_document(definition)
+                or package.result != result_identity_document(definition, job.result_id)
+                or manifest.artifact_type != "observatory-portable-result"
+                or manifest.subject_id != job.result_id
+                or dict(manifest.metadata)
+                != {
+                    "package-sha256": job.result_sha256,
+                    "package-identity-sha256": package.identity_sha256,
+                    "job-id": job.job_id,
+                    "job-identity-sha256": job.identity_sha256,
+                    "definition-sha256": job.definition_sha256,
+                    "source-bundle-sha256": job.source_bundle_sha256,
+                    "result-contract-sha256": definition.result_contract.contract_sha256,
+                    "calculation-profile-sha256": profile.identity_sha256,
+                }
+            ):
+                return None
+            result_document = next(
+                item for item in package.artifacts if item.role == RESULT_DOCUMENT_ROLE
+            )
+            if published_package != {
+                "schema_version": PORTABLE_RESULT_PACKAGE_SCHEMA,
+                "manifest_sha256": job.result_sha256,
+                "identity_sha256": package.identity_sha256,
+                "artifact_manifest_id": manifest_id,
+                "result_document_sha256": result_document.sha256,
+                "artifacts": [item.as_dict() for item in package.artifacts],
+            }:
+                return None
+            expected_members = {
+                item.role: (item.sha256, item.byte_length, item.media_type)
+                for item in package.artifacts
+            }
+            expected_members[RESULT_PACKAGE_MANIFEST_ROLE] = (
+                job.result_sha256,
+                len(payload),
+                "application/json",
+            )
+            if expected_members != {
+                item.role: (item.sha256, item.byte_length, item.media_type)
+                for item in manifest.members
+            }:
+                return None
+            for member in manifest.members:
+                self._verify_file(
+                    self._artifacts.object_path(member.sha256),
+                    member.sha256,
+                    member.byte_length,
+                    warm_only=warm_only,
+                )
+            return {
+                "result_id": job.result_id,
+                "result_kind": binding.result_kind,
+                "relation": "exact-recorded-computation",
+                "access": "observatory",
+                "created_at_utc": binding.run_created_at_utc,
+                "observatory_projection_available": True,
+                "identity": {
+                    "job_id": job.job_id,
+                    "source_session_id": job.source_session_id,
+                    "source_catalog_sha256": job.source_catalog_sha256,
+                    "source_bundle_sha256": job.source_bundle_sha256,
+                    "source_capability_manifest_sha256": job.source_capability_manifest_sha256,
+                    "setup_id": job.setup_id,
+                    "definition_sha256": job.definition_sha256,
+                    "package_sha256": job.result_sha256,
+                    "artifact_manifest_id": manifest_id,
+                },
+            }
+        except (
+            ArtifactGatewayError,
+            PortableResultPublisherError,
+            PortableRunDefinitionRegistryError,
+            SessionNotFoundError,
+            SessionIntegrityError,
+            OSError,
+            ValueError,
+            KeyError,
+            TypeError,
+        ):
+            # Missing/corrupt cache is not proof of computation coverage.
+            return None
+
+    def _verify_file(
+        self,
+        path: Path,
+        digest: str,
+        byte_length: int,
+        *,
+        warm_only: bool,
+    ) -> None:
+        with self._lock:
+            before = _fingerprint(path, byte_length)
+            key = (path, digest)
+            cached = self._verified.get(key)
+            if cached is not None and cached[0] == before:
+                self._verified.move_to_end(key)
+                if not cached[1]:
+                    raise ValueError("cached result object digest is invalid")
+                return
+            self._verified.pop(key, None)
+            if warm_only:
+                raise PortableResultCacheCheckRequired("refresh published result verification")
+            checksum = hashlib.sha256()
+            with path.open("rb") as stream:
+                while chunk := stream.read(1024 * 1024):
+                    checksum.update(chunk)
+            if _fingerprint(path, byte_length) != before:
+                raise ValueError("cached result object changed")
+            valid = checksum.hexdigest() == digest
+            self._verified[key] = (before, valid)
+            while len(self._verified) > 4096:
+                self._verified.popitem(last=False)
+            if not valid:
+                raise ValueError("cached result object digest is invalid")
+
+
+def _fingerprint(path: Path, byte_length: int) -> tuple[int, ...]:
+    metadata = path.lstat()
+    if not stat.S_ISREG(metadata.st_mode) or metadata.st_size != byte_length:
+        raise ValueError("cached result object is unavailable or not regular")
+    return (
+        metadata.st_dev,
+        metadata.st_ino,
+        metadata.st_size,
+        metadata.st_mtime_ns,
+        metadata.st_ctime_ns,
+    )

src/k1link/observatory/portable_setup_projection.py

@@ -22,6 +22,7 @@ from k1link.observatory.portable_run_definitions import (
     PortableRunDefinitionRegistry,
     PortableRunDefinitionRegistryError,
 )
+from k1link.observatory.recorded_jobs import ObservatoryRecordedQueueError
 from k1link.observatory.source_admission import (
     PortableRecordedSourceCapability,
     PortableSourceAdmissionError,
@@ -90,6 +91,14 @@ class PortableDefinitionCapabilityProbe(Protocol):
 type PortableSourceCapabilityProbe = Callable[[str], PortableRecordedSourceCapability]
 
 
+class PortablePublishedResultLookup(Protocol):
+    def find(
+        self,
+        source_session_id: str,
+        definition: PortableRunDefinition,
+    ) -> tuple[dict[str, object], ...]: ...
+
+
 @dataclass(frozen=True, slots=True)
 class _SourceCompatibility:
     compatible: bool
@@ -116,6 +125,7 @@ class PortableSetupProjector:
         capability_probe: PortableDefinitionCapabilityProbe,
         dispatch_available: bool = False,
         equipment_capture_registry: EquipmentCaptureRegistry | None = None,
+        result_cache: PortablePublishedResultLookup | None = None,
     ) -> None:
         if not hasattr(capability_probe, "probe"):
             raise PortableSetupProjectionError("portable source capability probe is unavailable")
@@ -123,6 +133,7 @@ class PortableSetupProjector:
         self._capability_probe = capability_probe
         self._dispatch_available = dispatch_available
         self._equipment_capture_registry = equipment_capture_registry
+        self._result_cache = result_cache
         for definition in registry.definitions:
             _validate_model_presentation(definition)
             if definition.authority.as_dict() != _OBSERVATION_ONLY_AUTHORITY:
@@ -166,8 +177,19 @@ class PortableSetupProjector:
         presentation = _presentation(definition)
         compatibility = self._probe_source(definition, source, presentation)
         executor = definition.executor
+        try:
+            existing = (
+                ()
+                if self._result_cache is None
+                else self._result_cache.find(source.session_id, definition)
+            )
+        except (ObservatoryRecordedQueueError, OSError) as exc:
+            raise PortableSetupProjectionError("portable result cache is unavailable") from exc
         submission_allowed = (
-            compatibility.compatible and executor.ready and self._dispatch_available
+            not existing
+            and compatibility.compatible
+            and executor.ready
+            and self._dispatch_available
         )
         return {
             "setup_id": definition.setup_id,
@@ -191,17 +213,23 @@ class PortableSetupProjector:
                 "reason_code": executor.reason_code,
                 "reason": executor.reason,
             },
-            "existing_results": [],
+            "existing_results": list(existing),
             "preflight": {
-                "outcome": "ready" if submission_allowed else "blocked",
-                "action": "check" if submission_allowed else "blocked",
-                "reason": self._preflight_reason(
+                "outcome": "existing" if existing else "ready" if submission_allowed else "blocked",
+                "action": "open-existing"
+                if existing
+                else "check"
+                if submission_allowed
+                else "blocked",
+                "reason": "Точный расчёт сохранён и проверен; доступен без запуска моделей."
+                if existing
+                else self._preflight_reason(
                     definition,
                     compatibility,
                     presentation,
                 ),
                 "submission_allowed": submission_allowed,
-                "existing_result_ids": [],
+                "existing_result_ids": [result["result_id"] for result in existing],
             },
             "authority": dict(_OBSERVATION_ONLY_AUTHORITY),
         }
@@ -214,10 +242,8 @@ class PortableSetupProjector:
     ) -> _SourceCompatibility:
         source_session_id = source.session_id
         if self._equipment_capture_registry is not None:
-            expected_profile = (
-                self._equipment_capture_registry.compatible_profile_for_requirements(
-                    definition.source_requirements.as_dict()
-                )
+            expected_profile = self._equipment_capture_registry.compatible_profile_for_requirements(
+                definition.source_requirements.as_dict()
             )
             if expected_profile is None:
                 raise PortableSetupProjectionError(
@@ -232,8 +258,7 @@ class PortableSetupProjector:
                     reason="У записи нет проверенной привязки к оборудованию и профилю записи.",
                 )
             if (
-                attestation.equipment_model_id
-                != expected_profile.equipment.equipment_model_id
+                attestation.equipment_model_id != expected_profile.equipment.equipment_model_id
                 or attestation.equipment_model_sha256
                 != expected_profile.equipment.equipment_model_sha256
             ):
@@ -245,8 +270,7 @@ class PortableSetupProjector:
                 )
             if (
                 attestation.capture_profile_id != expected_profile.capture_profile_id
-                or attestation.capture_profile_sha256
-                != expected_profile.capture_profile_sha256
+                or attestation.capture_profile_sha256 != expected_profile.capture_profile_sha256
             ):
                 return _SourceCompatibility(
                     compatible=False,

src/k1link/observatory/recorded_jobs.py

@@ -1202,6 +1202,7 @@ class ObservatoryRecordedJobQueue:
         *,
         enqueue: bool = False,
         reject_duplicate_computation: bool = False,
+        published_result_available: Callable[[ObservatoryRecordedJob], bool] | None = None,
     ) -> tuple[ObservatoryRecordedJob, bool]:
         """Accept one immutable identity and optionally queue it atomically."""
 
@@ -1242,6 +1243,20 @@ class ObservatoryRecordedJobQueue:
                 ).fetchone()
                 if duplicate is not None:
                     raise ObservatoryRecordedQueueDuplicateError(duplicate["job_id"])
+            if published_result_available is not None:
+                # Recheck under the INSERT lock: publication may have completed
+                # after catalog/preflight. The verifier must not re-enter this DB.
+                published = connection.execute(
+                    "SELECT * FROM observatory_recorded_jobs "
+                    "WHERE identity_sha256 = ? AND state = 'succeeded' "
+                    "AND publication_state = 'published' "
+                    "ORDER BY created_at_utc DESC, job_id DESC",
+                    (identity_sha256,),
+                ).fetchall()
+                for row in published:
+                    candidate = _job_from_row(row)
+                    if published_result_available(candidate):
+                        raise ObservatoryRecordedQueueDuplicateError(candidate.job_id)
             self._require_capacity(
                 connection,
                 table="observatory_recorded_jobs",
@@ -2091,6 +2106,27 @@ class ObservatoryRecordedJobQueue:
             ).fetchall()
         return tuple(_job_from_row(row) for row in rows)
 
+    def published_results(
+        self, *, source_session_id: str, source_catalog_sha256: str,
+        setup_id: str, definition_sha256: str,
+    ) -> tuple[ObservatoryRecordedJob, ...]:
+        """Exact cache candidates, never inferred from labels or a truncated job page."""
+
+        _validate_pattern(source_session_id, _SESSION_ID, "source session id")
+        _validate_pattern(setup_id, _IDENTIFIER, "setup id")
+        _validate_pattern(source_catalog_sha256, _SHA256, "source catalog sha256")
+        _validate_pattern(definition_sha256, _SHA256, "definition sha256")
+        with self._read_connection() as connection:
+            rows = connection.execute(
+                "SELECT * FROM observatory_recorded_jobs "
+                "WHERE source_session_id = ? AND source_catalog_sha256 = ? "
+                "AND setup_id = ? AND definition_sha256 = ? "
+                "AND state = 'succeeded' AND publication_state = 'published' "
+                "ORDER BY created_at_utc DESC, job_id DESC",
+                (source_session_id, source_catalog_sha256, setup_id, definition_sha256),
+            ).fetchall()
+        return tuple(_job_from_row(row) for row in rows)
+
     def request_live(self, intent: ObservatoryLiveLeaseIntent) -> tuple[ObservatoryLiveLease, bool]:
         """Close recorded admission without allowing a monolith to delay live K1."""
 
@@ -2746,6 +2782,18 @@ class ObservatoryRecordedJobQueue:
                 connection.executescript(_SCHEMA_SQL)
                 self._migrate_claim_lease_schema(connection)
                 self._migrate_publication_schema(connection)
+                # A legacy database may not have publication columns until
+                # the migration above. Never create the partial index earlier.
+                connection.execute(
+                    "CREATE INDEX IF NOT EXISTS observatory_recorded_jobs_published_source "
+                    "ON observatory_recorded_jobs "
+                    "(source_session_id, source_catalog_sha256, setup_id, definition_sha256) "
+                    "WHERE state = 'succeeded' AND publication_state = 'published'"
+                )
+                connection.execute(
+                    "CREATE INDEX IF NOT EXISTS observatory_recorded_jobs_computation "
+                    "ON observatory_recorded_jobs (identity_sha256)"
+                )
                 self._validate_schema(connection)
                 self._validate_existing_capacity(connection)
                 connection.commit()

src/k1link/web/app.py

@@ -54,6 +54,7 @@ from k1link.observatory.portable_queue_binding import (
     PortableQueueBindingError,
     PortableRecordedQueueBindingService,
 )
+from k1link.observatory.portable_result_cache import PortableResultCache
 from k1link.observatory.portable_result_contract import (
     PortableCalculationProfileRegistry,
     PortableResultContractValidatorRegistry,
@@ -538,24 +539,41 @@ else:
 OBSERVATORY_PORTABLE_BINDING_SERVICE: PortableRecordedQueueBindingService | None
 OBSERVATORY_PORTABLE_SETUP_PROJECTOR: PortableSetupProjector | None
 OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR: str | None
+OBSERVATORY_PORTABLE_RESULT_CACHE: PortableResultCache | None = None
 try:
     if OBSERVATORY_PORTABLE_DEFINITION_REGISTRY is None:
         raise PortableSetupProjectionError(
             OBSERVATORY_PORTABLE_DEFINITION_REGISTRY_ERROR
             or "portable definition registry is unavailable"
         )
+    if (
+        session_artifact_gateway is not None
+        and OBSERVATORY_RECORDED_JOB_QUEUE is not None
+        and OBSERVATORY_PORTABLE_CALCULATION_PROFILES is not None
+    ):
+        OBSERVATORY_PORTABLE_RESULT_CACHE = PortableResultCache(
+            sessions=session_store, artifacts=session_artifact_gateway.store,
+            queue=OBSERVATORY_RECORDED_JOB_QUEUE,
+            definitions=OBSERVATORY_PORTABLE_DEFINITION_REGISTRY,
+            calculation_profiles=OBSERVATORY_PORTABLE_CALCULATION_PROFILES,
+        )
     OBSERVATORY_PORTABLE_BINDING_SERVICE = PortableRecordedQueueBindingService(
         data_dir=session_store.data_dir,
         session_store=session_store,
         media_inspector=session_recorded_media_inspector,
         definitions=OBSERVATORY_PORTABLE_DEFINITION_REGISTRY,
         queue=OBSERVATORY_RECORDED_JOB_QUEUE,
+        published_result_available=(
+            None if OBSERVATORY_PORTABLE_RESULT_CACHE is None
+            else OBSERVATORY_PORTABLE_RESULT_CACHE.available
+        ),
     )
     OBSERVATORY_PORTABLE_SETUP_PROJECTOR = PortableSetupProjector(
         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,
+        result_cache=OBSERVATORY_PORTABLE_RESULT_CACHE,
     )
     OBSERVATORY_PORTABLE_SETUP_PROJECTOR_ERROR = None
 except (

src/k1link/web/observatory_api.py

@@ -33,6 +33,7 @@ from k1link.observatory.portable_queue_binding import (
     PortableQueueBindingStaleCheckError,
     PortableRecordedQueueBindingService,
 )
+from k1link.observatory.portable_result_cache import PortableResultCacheCheckRequired
 from k1link.observatory.portable_result_contract import PortableResultPublisherError
 from k1link.observatory.portable_result_publisher import (
     PortableObservatoryResultPublisher,
@@ -298,6 +299,29 @@ def build_observatory_router(
             )
         compatible = compatibility.get("compatible") is True
         executor_ready = executor.get("state") == "ready" and executor.get("ready") is True
+        projected_preflight = projected.get("preflight")
+        if (
+            isinstance(projected_preflight, dict)
+            and projected_preflight.get("outcome") == "existing"
+        ):
+            return {
+                "schema_version": OBSERVATORY_RUN_PREFLIGHT_SCHEMA,
+                "source_session_id": request.source_session_id,
+                "setup_id": request.setup_id,
+                "definition_sha256": expected_digest,
+                "check_sha256": None,
+                "outcome": "existing",
+                "submission_allowed": False,
+                "checks": [{
+                    "check_id": "published-result",
+                    "outcome": "pass",
+                    "reason_code": "exact-published-result-verified",
+                    "message": "Точный расчёт сохранён; повторный inference не требуется.",
+                }],
+                "existing_result_ids": projected_preflight["existing_result_ids"],
+                "executor": executor,
+                "authority": projected.get("authority", dict(_OBSERVATION_ONLY_AUTHORITY)),
+            }
         checked = None
         check_reason: str | None = None
         if (
@@ -895,6 +919,11 @@ def build_observatory_router(
                         status_code=409,
                         detail="Идентичность RunDefinition изменилась; повторите preflight.",
                     )
+                if portable_projection.get("existing_results"):
+                    raise HTTPException(
+                        status_code=409,
+                        detail="Точный расчёт уже сохранён. Обновите каталог и откройте результат.",
+                    )
                 if (
                     projected_executor.get("state") != "ready"
                     or projected_executor.get("ready") is not True
@@ -930,11 +959,16 @@ def build_observatory_router(
                         status_code=409,
                         detail="Portable-привязка источника не прошла проверку целостности.",
                     ) from exc
+                except PortableResultCacheCheckRequired as exc:
+                    raise HTTPException(
+                        status_code=409,
+                        detail="Опубликованный результат требует проверки. Обновите каталог.",
+                    ) from exc
                 except ObservatoryRecordedQueueDuplicateError as exc:
                     raise HTTPException(
                         status_code=409,
                         detail=(
-                            "Такой расчёт уже выполняется или ожидает публикации. "
+                            "Такой расчёт уже выполняется, ожидает публикации или сохранён. "
                             "Обновите список расчётов; повторный запуск не создан."
                         ),
                     ) from exc

tests/test_observatory_portable_result_publisher.py

@@ -3,14 +3,22 @@ from __future__ import annotations
 import copy
 import hashlib
 import json
+import os
+import sqlite3
 from collections.abc import Callable
 from dataclasses import replace
 from pathlib import Path
 from typing import cast
 
 import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
 
 from k1link.artifact_gateway import CentralArtifactStore
+from k1link.observatory.portable_result_cache import (
+    PortableResultCache,
+    PortableResultCacheCheckRequired,
+)
 from k1link.observatory.portable_result_contract import (
     OBSERVATORY_CALCULATION_PROFILE_SCHEMA,
     PORTABLE_RESULT_PACKAGE_IDENTITY_SCHEMA,
@@ -35,22 +43,27 @@ from k1link.observatory.portable_run_definitions import (
     PortableRunDefinitionRegistry,
     canonical_sha256,
 )
+from k1link.observatory.portable_setup_projection import PortableSetupProjector
 from k1link.observatory.recorded_jobs import (
     ObservatoryRecordedJob,
     ObservatoryRecordedJobIntent,
     ObservatoryRecordedJobQueue,
+    ObservatoryRecordedQueueDuplicateError,
+    ObservatoryRecordedQueueIntegrityError,
     RecordedRunDefinitionRegistry,
 )
 from k1link.observatory.source_admission import (
     PORTABLE_SOURCE_BUNDLE_SCHEMA,
     PORTABLE_SOURCE_CAPABILITY_SCHEMA,
     PORTABLE_SOURCE_DOCUMENT_DIRECTORY,
+    PortableSourceAdmissionIntegrityError,
 )
 from k1link.sessions import (
     ObservationArchiveSource,
     ObservationSessionCandidate,
     SessionStore,
 )
+from k1link.web.observatory_api import build_observatory_router
 
 REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
 REGISTRY_PATH = REPOSITORY_ROOT / "config" / "observatory-portable-run-definitions.json"
@@ -232,6 +245,7 @@ def _package(
     definition: PortableRunDefinition,
     result_id: str = RESULT_ID,
     accepted: bool = True,
+    extra_payload: bytes | None = None,
 ) -> tuple[Path, PortableResultPackageManifest]:
     result_document = {
         "schema_version": definition.result_contract.result_schema,
@@ -248,17 +262,32 @@ def _package(
         byte_length=len(result_bytes),
         sha256=hashlib.sha256(result_bytes).hexdigest(),
     )
+    extra = (
+        ()
+        if extra_payload is None
+        else (
+            PortableResultArtifact(
+                role="spatial-evidence",
+                relative_path="artifacts/spatial.bin",
+                media_type="application/octet-stream",
+                byte_length=len(extra_payload),
+                sha256=hashlib.sha256(extra_payload).hexdigest(),
+            ),
+        )
+    )
     package = PortableResultPackageManifest.create(
         job=job,
         definition=definition,
         result_id=result_id,
         created_at_utc=NOW,
-        artifacts=(artifact,),
+        artifacts=(artifact, *extra),
     )
     root = tmp_path / "packages" / package.manifest_sha256
     (root / "artifacts").mkdir(parents=True)
     (root / "manifest.json").write_bytes(package.canonical_bytes)
     (root / "artifacts" / "result.json").write_bytes(result_bytes)
+    if extra_payload is not None:
+        (root / "artifacts" / "spatial.bin").write_bytes(extra_payload)
     return root, package
 
 
@@ -320,6 +349,8 @@ def _fixture(
     *,
     result_id: str = RESULT_ID,
     accepted: bool = True,
+    extra_payload: bytes | None = None,
+    for_publication: bool = False,
 ) -> tuple[
     PortableRunDefinitionRegistry,
     PortableRunDefinition,
@@ -343,8 +374,10 @@ def _fixture(
         definition=definition,
         result_id=result_id,
         accepted=accepted,
+        extra_payload=extra_payload,
     )
-    succeeded = queue.succeed(
+    complete = queue.complete_for_publication if for_publication else queue.succeed
+    succeeded = complete(
         running.job_id,
         claim_token=claim_token,
         result_id=result_id,
@@ -410,15 +443,16 @@ def test_verified_package_publishes_immutable_binding_and_profile_provenance(
     assert view["viewer_capability"] == first.binding.replay_capability.as_dict()
 
     summary = store.get_session(RESULT_ID).summary
-    assert summary.display_name == (
-        "Portable result source · полный маршрут и воспроизведение"
-    )
+    assert summary.display_name == ("Portable result source · полный маршрут и воспроизведение")
     profiles = PortableCalculationProfileRegistry((_profile(definition),))
-    assert resolve_published_portable_calculation_profile(
-        summary,
-        definitions=registry,
-        calculation_profiles=profiles,
-    ) == _profile(definition).as_dict()
+    assert (
+        resolve_published_portable_calculation_profile(
+            summary,
+            definitions=registry,
+            calculation_profiles=profiles,
+        )
+        == _profile(definition).as_dict()
+    )
 
     assert summary.lab is not None
     drifted_provenance = copy.deepcopy(summary.lab.provenance)
@@ -586,9 +620,7 @@ def test_package_manifest_rejects_noncanonical_or_authority_elevating_documents(
         PortableResultPackageIntegrityError,
         match="not canonical JSON",
     ):
-        PortableResultPackageManifest.from_bytes(
-            json.dumps(manifest.as_dict(), indent=2).encode()
-        )
+        PortableResultPackageManifest.from_bytes(json.dumps(manifest.as_dict(), indent=2).encode())
 
 
 def test_legacy_canonical_result_namespace_cannot_be_republished(
@@ -614,3 +646,258 @@ def test_legacy_canonical_result_namespace_cannot_be_republished(
         publisher.publish(job=job, package_root=package_root)
 
     assert store.get_lab_instance(legacy_result_id) is None
+
+
+def _cache_fixture(tmp_path: Path, *, published: bool = True):
+    registry, definition, sessions, job, package_root = _fixture(
+        tmp_path,
+        extra_payload=b"synthetic spatial evidence",
+        for_publication=True,
+    )
+    queue = ObservatoryRecordedJobQueue(
+        sessions.data_dir,
+        definitions=RecordedRunDefinitionRegistry(registry.ready_recorded_definitions()),
+        clock=lambda: NOW,
+    )
+    artifacts = CentralArtifactStore(tmp_path / "central-artifacts", create=True)
+    publisher = _publisher(
+        tmp_path,
+        store=sessions,
+        registry=registry,
+        profile=_profile(definition),
+        validator=_validator,
+    )
+    if published:
+        publisher.publish(job=job, package_root=package_root)
+        job = queue.mark_published(job.job_id)
+    cache = PortableResultCache(
+        sessions=sessions,
+        artifacts=artifacts,
+        queue=queue,
+        definitions=registry,
+        calculation_profiles=PortableCalculationProfileRegistry((_profile(definition),)),
+    )
+    return cache, queue, sessions, artifacts, registry, definition, job, publisher, package_root
+
+
+def _retry_intent(job: ObservatoryRecordedJob, key: str = "another-operator-click"):
+    return ObservatoryRecordedJobIntent(
+        idempotency_key=key,
+        source_session_id=job.source_session_id,
+        source_catalog_sha256=job.source_catalog_sha256,
+        source_bundle_sha256=job.source_bundle_sha256,
+        source_capability_manifest_sha256=job.source_capability_manifest_sha256,
+        setup_id=job.setup_id,
+        definition_sha256=job.definition_sha256,
+    )
+
+
+def test_cache_exact_publication_roundtrip_and_atomic_duplicate_guard(tmp_path: Path) -> None:
+    cache, queue, sessions, artifacts, _, definition, job, _, _ = _cache_fixture(tmp_path)
+    results = cache.find(SOURCE_SESSION_ID, definition)
+    assert [result["result_id"] for result in results] == [RESULT_ID]
+    assert results[0]["identity"]["source_bundle_sha256"] == job.source_bundle_sha256
+    assert results[0]["identity"]["definition_sha256"] == definition.definition_sha256
+    view = PortableResultViewService(sessions=sessions, artifacts=artifacts).read(RESULT_ID)
+    assert view["result_id"] == RESULT_ID
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError) as duplicate:
+        queue.submit(_retry_intent(job), published_result_available=cache.available)
+    assert duplicate.value.job_id == job.job_id
+    # Same request remains idempotent even when the cache guard would reject a new click.
+    retried, created = queue.submit(
+        _retry_intent(job, job.idempotency_key),
+        published_result_available=cache.available,
+    )
+    assert retried.job_id == job.job_id and not created
+    assert len(queue.list_jobs()) == 1
+
+
+def test_cache_does_not_hide_another_source_snapshot_or_profile_version(tmp_path: Path) -> None:
+    cache, queue, sessions, _, _, definition, job, _, _ = _cache_fixture(tmp_path)
+    assert not cache.find("another-source", definition)
+    identity = definition.identity_document()
+    identity["version"] = definition.version + 1
+    newer = replace(
+        definition,
+        version=definition.version + 1,
+        definition_sha256=canonical_sha256(identity),
+    )
+    assert not cache.find(SOURCE_SESSION_ID, newer)
+    assert not queue.published_results(
+        source_session_id=SOURCE_SESSION_ID,
+        source_catalog_sha256=job.source_catalog_sha256,
+        setup_id=job.setup_id,
+        definition_sha256="f" * 64,
+    )
+    with sqlite3.connect(sessions.database_path) as connection:
+        connection.execute(
+            "UPDATE observation_sessions SET total_bytes = total_bytes + 1 WHERE session_id = ?",
+            (SOURCE_SESSION_ID,),
+        )
+    assert not cache.find(SOURCE_SESSION_ID, definition)
+    assert sessions.get_lab_instance(RESULT_ID) is not None
+
+
+@pytest.mark.parametrize("field", ["executor_image_sha256", "source_bundle_sha256"])
+def test_queue_rejects_mutated_job_identity_before_cache_lookup(tmp_path: Path, field: str) -> None:
+    cache, _, _, _, _, definition, job, _, _ = _cache_fixture(tmp_path)
+    assert cache.find(SOURCE_SESSION_ID, definition)
+    with pytest.raises(ObservatoryRecordedQueueIntegrityError):
+        replace(job, **{field: "f" * 64})
+
+
+@pytest.mark.parametrize("missing", [True, False])
+def test_cache_requires_intact_central_manifest(tmp_path: Path, missing: bool) -> None:
+    cache, _, _, artifacts, _, definition, _, _, _ = _cache_fixture(tmp_path)
+    result = cache.find(SOURCE_SESSION_ID, definition)[0]
+    path = artifacts.manifest_path(result["identity"]["artifact_manifest_id"])
+    if missing:
+        path.unlink()
+    else:
+        path.write_text("{}")
+    assert not cache.find(SOURCE_SESSION_ID, definition)
+
+
+@pytest.mark.parametrize("failure", ["missing", "same-size-corrupt", "symlink"])
+@pytest.mark.parametrize("role", ["result-document", "spatial-evidence", "result-package-manifest"])
+def test_cache_invalidates_any_changed_artifact_without_erasing_history(
+    tmp_path: Path,
+    failure: str,
+    role: str,
+) -> None:
+    cache, queue, sessions, artifacts, _, definition, job, _, _ = _cache_fixture(tmp_path)
+    result = cache.find(SOURCE_SESSION_ID, definition)[0]
+    manifest = artifacts.read_manifest(result["identity"]["artifact_manifest_id"])
+    path = artifacts.object_path(manifest.member(role).sha256)
+    original = path.read_bytes()
+    metadata = path.stat()
+    if failure == "missing":
+        path.unlink()
+    elif failure == "symlink":
+        target = tmp_path / "not-an-immutable-object"
+        target.write_bytes(original)
+        path.unlink()
+        path.symlink_to(target)
+    else:
+        path.write_bytes(b"!" * len(original))
+        os.utime(path, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
+    assert not cache.find(SOURCE_SESSION_ID, definition)
+    assert sessions.get_lab_instance(RESULT_ID) is not None
+    _, created = queue.submit(_retry_intent(job), published_result_available=cache.available)
+    assert created
+
+
+@pytest.mark.parametrize("field", ["job", "source", "run_definition", "result_package"])
+def test_cache_rejects_changed_publication_binding(tmp_path: Path, field: str) -> None:
+    cache, _, sessions, _, _, definition, _, _, _ = _cache_fixture(tmp_path)
+    assert cache.find(SOURCE_SESSION_ID, definition)
+    binding = sessions.get_lab_instance(RESULT_ID)
+    assert binding is not None
+    provenance = copy.deepcopy(binding.provenance)
+    provenance[field]["unexpected"] = "changed"
+    with sqlite3.connect(sessions.database_path) as connection:
+        connection.execute(
+            "UPDATE observation_lab_instances SET provenance_json = ? WHERE session_id = ?",
+            (json.dumps(provenance), RESULT_ID),
+        )
+    assert not cache.find(SOURCE_SESSION_ID, definition)
+
+
+def test_cache_publication_race_requires_refresh_without_hashing_under_queue_lock(
+    tmp_path: Path,
+) -> None:
+    cache, queue, _, _, _, definition, job, publisher, root = _cache_fixture(
+        tmp_path,
+        published=False,
+    )
+    assert not cache.find(SOURCE_SESSION_ID, definition)
+    publisher.publish(job=job, package_root=root)
+    # Sealed and even session-published are insufficient without the durable receipt.
+    assert not cache.find(SOURCE_SESSION_ID, definition)
+    queue.mark_published(job.job_id)
+    with pytest.raises(PortableResultCacheCheckRequired):
+        queue.submit(_retry_intent(job), published_result_available=cache.available)
+    assert len(queue.list_jobs()) == 1
+    assert cache.find(SOURCE_SESSION_ID, definition)
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError):
+        queue.submit(_retry_intent(job), published_result_available=cache.available)
+
+
+def test_cache_warm_refresh_does_not_reread_large_artifacts(tmp_path: Path, monkeypatch) -> None:
+    cache, _, _, artifacts, _, definition, _, _, _ = _cache_fixture(tmp_path)
+    result = cache.find(SOURCE_SESSION_ID, definition)[0]
+    manifest = artifacts.read_manifest(result["identity"]["artifact_manifest_id"])
+    spatial = artifacts.object_path(manifest.member("spatial-evidence").sha256)
+    original_open = Path.open
+
+    def guarded_open(path, *args, **kwargs):
+        assert path != spatial, "warm refresh must only stat an already verified large artifact"
+        return original_open(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "open", guarded_open)
+    assert cache.find(SOURCE_SESSION_ID, definition)
+
+
+def test_cache_api_existing_skips_source_preparation_submit_and_worker(tmp_path: Path) -> None:
+    cache, queue, sessions, artifacts, registry, definition, job, _, _ = _cache_fixture(tmp_path)
+
+    class NoCompute:
+        def probe(self, **_kwargs):
+            raise PortableSourceAdmissionIntegrityError("raw source temporarily offline")
+
+        def check(self, **_kwargs):
+            pytest.fail("cached preflight must not prepare the source")
+
+        def submit(self, **_kwargs):
+            pytest.fail("cached result must not create model work")
+
+    binding = NoCompute()
+    app = FastAPI()
+    app.include_router(
+        build_observatory_router(
+            sessions,
+            portable_setup_projector=PortableSetupProjector(
+                registry=registry,
+                capability_probe=binding,
+                dispatch_available=False,
+                result_cache=cache,
+            ),
+            portable_binding_service=binding,
+            recorded_job_queue=queue,
+            portable_result_view=PortableResultViewService(sessions=sessions, artifacts=artifacts),
+        )
+    )
+    client = TestClient(app)
+    catalog = client.get(
+        "/api/v1/observatory/portable-laboratory-setups",
+        params={
+            "source_session_id": SOURCE_SESSION_ID,
+        },
+    )
+    assert catalog.status_code == 200
+    setup = catalog.json()["setups"][0]
+    assert setup["source_compatibility"]["compatible"] is False
+    assert setup["preflight"]["outcome"] == "existing"
+    assert setup["preflight"]["submission_allowed"] is False
+    request = {
+        "schema_version": "missioncore.observatory-run-preflight-request/v1",
+        "source_session_id": SOURCE_SESSION_ID,
+        "setup_id": definition.setup_id,
+        "definition_sha256": definition.definition_sha256,
+    }
+    preflight = client.post("/api/v1/observatory/run-preflights", json=request)
+    assert preflight.status_code == 200
+    assert preflight.json()["outcome"] == "existing"
+    assert preflight.json()["check_sha256"] is None
+    assert preflight.json()["existing_result_ids"] == [RESULT_ID]
+    request.update(
+        {
+            "schema_version": "missioncore.observatory-recorded-run-submit/v1",
+            "idempotency_key": "new-click",
+            "check_sha256": "f" * 64,
+        }
+    )
+    assert client.post("/api/v1/observatory/runs", json=request).status_code == 409
+    request["idempotency_key"] = job.idempotency_key
+    assert client.post("/api/v1/observatory/runs", json=request).status_code == 202
+    assert len(queue.list_jobs()) == 1

tests/test_observatory_recorded_jobs.py

@@ -1057,6 +1057,30 @@ def test_operator_reconciliation_is_durable_idempotent_and_unblocks_queue(
     assert replacement.job.job_id == next_job.job_id
 
 
+def test_cache_indexes_follow_legacy_publication_column_migration(tmp_path: Path) -> None:
+    queue = _queue(tmp_path)
+    job, _ = queue.submit(_intent(), enqueue=True)
+    with sqlite3.connect(queue.database_path) as connection:
+        connection.execute("DROP INDEX observatory_recorded_jobs_published_source")
+        connection.execute("DROP INDEX observatory_recorded_jobs_computation")
+        for column in (
+            "publication_state", "publication_attempts", "publication_error", "published_at_utc",
+        ):
+            connection.execute(f"ALTER TABLE observatory_recorded_jobs DROP COLUMN {column}")
+    migrated = _queue(tmp_path)
+    assert migrated.get(job.job_id) == job
+    assert not migrated.published_results(
+        source_session_id=job.source_session_id, source_catalog_sha256=job.source_catalog_sha256,
+        setup_id=job.setup_id, definition_sha256=job.definition_sha256,
+    )
+    with sqlite3.connect(queue.database_path) as connection:
+        names = {
+            row[1] for row in connection.execute("PRAGMA index_list(observatory_recorded_jobs)")
+        }
+    assert "observatory_recorded_jobs_published_source" in names
+    assert "observatory_recorded_jobs_computation" in names
+
+
 def test_legacy_sqlite_claim_schema_migrates_without_reusing_old_token(
     tmp_path: Path,
 ) -> None: