From 1e4ddc2cff927cbd37b94d22e1866f99e0f1b08a Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 3 Sep 2026 09:14:54 +0300 Subject: [PATCH] feat(observatory): verify exact published result reuse --- .../portableLaboratorySetupDecoder.ts | 80 ++++- .../test/observatoryLaboratorySetups.test.mjs | 72 +++- .../observatory/portable_queue_binding.py | 4 + .../observatory/portable_result_cache.py | 305 +++++++++++++++++ .../observatory/portable_setup_projection.py | 52 ++- src/k1link/observatory/recorded_jobs.py | 48 +++ src/k1link/web/app.py | 18 + src/k1link/web/observatory_api.py | 36 +- ...t_observatory_portable_result_publisher.py | 313 +++++++++++++++++- tests/test_observatory_recorded_jobs.py | 24 ++ 10 files changed, 910 insertions(+), 42 deletions(-) create mode 100644 src/k1link/observatory/portable_result_cache.py diff --git a/apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts b/apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts index 0483218..7cf25c8 100644 --- a/apps/control-station/src/core/observatory/portableLaboratorySetupDecoder.ts +++ b/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, [ diff --git a/apps/control-station/test/observatoryLaboratorySetups.test.mjs b/apps/control-station/test/observatoryLaboratorySetups.test.mjs index 99e25af..1377ebf 100644 --- a/apps/control-station/test/observatoryLaboratorySetups.test.mjs +++ b/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({ diff --git a/src/k1link/observatory/portable_queue_binding.py b/src/k1link/observatory/portable_queue_binding.py index 75f9ce1..d2a8ff6 100644 --- a/src/k1link/observatory/portable_queue_binding.py +++ b/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( diff --git a/src/k1link/observatory/portable_result_cache.py b/src/k1link/observatory/portable_result_cache.py new file mode 100644 index 0000000..d8a8766 --- /dev/null +++ b/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, + ) diff --git a/src/k1link/observatory/portable_setup_projection.py b/src/k1link/observatory/portable_setup_projection.py index 544ecec..353401d 100644 --- a/src/k1link/observatory/portable_setup_projection.py +++ b/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, diff --git a/src/k1link/observatory/recorded_jobs.py b/src/k1link/observatory/recorded_jobs.py index 2e00dea..63c530e 100644 --- a/src/k1link/observatory/recorded_jobs.py +++ b/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() diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 373b6dc..bce9e54 100644 --- a/src/k1link/web/app.py +++ b/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 ( diff --git a/src/k1link/web/observatory_api.py b/src/k1link/web/observatory_api.py index 180e938..e3ff871 100644 --- a/src/k1link/web/observatory_api.py +++ b/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 diff --git a/tests/test_observatory_portable_result_publisher.py b/tests/test_observatory_portable_result_publisher.py index 5dc1ffc..688d0d5 100644 --- a/tests/test_observatory_portable_result_publisher.py +++ b/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 diff --git a/tests/test_observatory_recorded_jobs.py b/tests/test_observatory_recorded_jobs.py index 7dee2f9..43a8783 100644 --- a/tests/test_observatory_recorded_jobs.py +++ b/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: