fix(observatory): reject duplicate active portable computations atomically

src/k1link/observatory/portable_queue_binding.py

@@ -214,6 +214,7 @@ class PortableRecordedQueueBindingService:
         return self._queue.submit(
             preparation.intent(idempotency_key=idempotency_key),
             enqueue=enqueue,
+            reject_duplicate_computation=True,
         )
 
     def _resolve_definition(

src/k1link/observatory/recorded_jobs.py

@@ -239,6 +239,14 @@ class ObservatoryRecordedQueueConflictError(ObservatoryRecordedQueueError):
     """An idempotency identity or state transition conflicts with history."""
 
 
+class ObservatoryRecordedQueueDuplicateError(ObservatoryRecordedQueueConflictError):
+    """Another request already owns this exact recorded computation."""
+
+    def __init__(self, job_id: str) -> None:
+        self.job_id = job_id
+        super().__init__("an identical recorded computation already exists")
+
+
 class ObservatoryRecordedQueueNotFoundError(ObservatoryRecordedQueueError):
     """A requested job or live lease does not exist."""
 
@@ -1177,6 +1185,7 @@ class ObservatoryRecordedJobQueue:
         intent: ObservatoryRecordedJobIntent,
         *,
         enqueue: bool = False,
+        reject_duplicate_computation: bool = False,
     ) -> tuple[ObservatoryRecordedJob, bool]:
         """Accept one immutable identity and optionally queue it atomically."""
 
@@ -1199,16 +1208,32 @@ class ObservatoryRecordedJobQueue:
                     )
                     record = self._get_job(connection, record.job_id)
                 return record, False
+            definition = self._definitions.resolve(intent.setup_id, intent.definition_sha256)
+            identity_sha256 = _job_identity_sha256(intent, definition)
+            if reject_duplicate_computation:
+                # This check shares the INSERT transaction: two clients with
+                # different idempotency keys cannot race into duplicate jobs.
+                # A sealed result awaiting publication is not a reason to run
+                # the models again. Published-cache validity is a separate gate.
+                duplicate = connection.execute(
+                    "SELECT job_id FROM observatory_recorded_jobs "
+                    "WHERE identity_sha256 = ? AND ("
+                    "state IN ('accepted', 'queued', 'claimed', 'running', 'paused', "
+                    "'preemption-pending', 'reconciliation-required') OR "
+                    "(state = 'succeeded' AND publication_state IN ('pending', 'failed'))) "
+                    "ORDER BY created_at_utc DESC, job_id DESC LIMIT 1",
+                    (identity_sha256,),
+                ).fetchone()
+                if duplicate is not None:
+                    raise ObservatoryRecordedQueueDuplicateError(duplicate["job_id"])
             self._require_capacity(
                 connection,
                 table="observatory_recorded_jobs",
                 limit=self._max_jobs,
                 label="recorded job",
             )
-            definition = self._definitions.resolve(intent.setup_id, intent.definition_sha256)
             now = self._timestamp()
             job_id = f"observatory-run-{uuid4().hex}"
-            identity_sha256 = _job_identity_sha256(intent, definition)
             receipt_sha256 = _submission_receipt_sha256(
                 job_id=job_id,
                 request_sha256=intent.request_sha256,

src/k1link/web/observatory_api.py

@@ -53,6 +53,7 @@ from k1link.observatory.recorded_jobs import (
     ObservatoryRecordedJobQueue,
     ObservatoryRecordedQueueCapacityError,
     ObservatoryRecordedQueueConflictError,
+    ObservatoryRecordedQueueDuplicateError,
     ObservatoryRecordedQueueError,
     ObservatoryRecordedQueueNotFoundError,
 )
@@ -929,6 +930,14 @@ def build_observatory_router(
                         status_code=409,
                         detail="Portable-привязка источника не прошла проверку целостности.",
                     ) from exc
+                except ObservatoryRecordedQueueDuplicateError as exc:
+                    raise HTTPException(
+                        status_code=409,
+                        detail=(
+                            "Такой расчёт уже выполняется или ожидает публикации. "
+                            "Обновите список расчётов; повторный запуск не создан."
+                        ),
+                    ) from exc
                 except ObservatoryRecordedQueueConflictError as exc:
                     try:
                         raced = recorded_job_queue.get_by_idempotency_key(request.idempotency_key)

tests/test_observatory_portable_queue_binding.py

@@ -18,7 +18,10 @@ from k1link.observatory.portable_run_definitions import (
     PortableRunDefinitionUnavailableError,
     canonical_sha256,
 )
-from k1link.observatory.recorded_jobs import ObservatoryRecordedJobQueue
+from k1link.observatory.recorded_jobs import (
+    ObservatoryRecordedJobQueue,
+    ObservatoryRecordedQueueDuplicateError,
+)
 from k1link.observatory.source_admission import (
     PORTABLE_SOURCE_DOCUMENT_DIRECTORY,
     PortableRecordedSourceAdmission,
@@ -534,6 +537,16 @@ def test_admit_persists_exact_documents_and_submit_queues_same_identity(
     )
     assert queue is not None
     assert queue.list_jobs() == (job,)
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError) as caught:
+        service.submit(
+            source_session_id=SESSION_ID,
+            setup_id=definition.setup_id,
+            definition_sha256=definition.definition_sha256,
+            expected_check_sha256=checked.check_sha256,
+            idempotency_key="second-browser",
+        )
+    assert caught.value.job_id == job.job_id
+    assert queue.list_jobs() == (job,)
 
 
 def test_admit_rejects_catalog_change_since_check_without_writes_or_job(

tests/test_observatory_portable_setup_api.py

@@ -242,6 +242,7 @@ class _PortableBinding:
                 definition_sha256=definition_sha256,
             ),
             enqueue=True,
+            reject_duplicate_computation=True,
         )
 
 
@@ -299,6 +300,23 @@ def test_portable_api_check_sha_fences_ready_submission(tmp_path: Path) -> None:
     assert submitted.json()["state"] == "queued"
     assert submitted.json()["setup"]["setup_id"] == definition.setup_id
     assert binding.submit_count == 1
+    request = {
+        "schema_version": "missioncore.observatory-recorded-run-submit/v1",
+        "idempotency_key": "portable:lab-v1:source-005",
+        "source_session_id": SOURCE_SESSION_ID,
+        "setup_id": definition.setup_id,
+        "definition_sha256": definition.definition_sha256,
+        "check_sha256": binding.check_sha256,
+    }
+    repeated = client.post("/api/v1/observatory/runs", json=request)
+    assert repeated.status_code == 202
+    assert repeated.json()["job_id"] == submitted.json()["job_id"]
+    assert binding.submit_count == 1
+    request["idempotency_key"] = "second-browser"
+    duplicate = client.post("/api/v1/observatory/runs", json=request)
+    assert duplicate.status_code == 409
+    assert "повторный запуск не создан" in duplicate.json()["detail"]
+    assert len(queue.list_jobs()) == 1
 
 
 def test_portable_api_rejects_blocked_lab_executor_before_binding_submit(

tests/test_observatory_recorded_jobs.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import sqlite3
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import replace
 from pathlib import Path
+from threading import Barrier
 
 import pytest
 
@@ -17,6 +19,7 @@ from k1link.observatory.recorded_jobs import (
     ObservatoryRecordedPreemptionError,
     ObservatoryRecordedQueueBusyError,
     ObservatoryRecordedQueueConflictError,
+    ObservatoryRecordedQueueDuplicateError,
     ObservatoryRecordedQueueIntegrityError,
     ObservatoryRecordedQueueStaleClaimError,
     ObservatoryRecordedReconciliationRequest,
@@ -198,6 +201,160 @@ def _reconciliation_request(
     )
 
 
+@pytest.mark.parametrize("enqueue", [False, True])
+def test_portable_duplicate_guard_preserves_original_request(
+    tmp_path: Path,
+    enqueue: bool,
+) -> None:
+    queue = _queue(tmp_path)
+    first, created = queue.submit(
+        _intent(),
+        enqueue=enqueue,
+        reject_duplicate_computation=True,
+    )
+    assert created
+    repeated, created = queue.submit(
+        _intent(),
+        enqueue=enqueue,
+        reject_duplicate_computation=True,
+    )
+    assert repeated == first and not created
+    reopened = _queue(tmp_path)
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError) as caught:
+        reopened.submit(
+            _intent(idempotency_key="second-browser"),
+            enqueue=True,
+            reject_duplicate_computation=True,
+        )
+    assert caught.value.job_id == first.job_id
+    assert len(reopened.list_jobs()) == 1
+    # Reject, rather than silently rebinding a second idempotency key.
+    with pytest.raises(ObservatoryRecordedQueueConflictError):
+        reopened.submit(
+            _intent(source_catalog_sha256="0" * 64),
+            reject_duplicate_computation=True,
+        )
+
+
+@pytest.mark.parametrize(
+    "field",
+    [
+        "source_session_id",
+        "source_catalog_sha256",
+        "source_bundle_sha256",
+        "source_capability_manifest_sha256",
+        "setup_id",
+    ],
+)
+def test_duplicate_guard_keeps_distinct_sources_and_profiles(
+    tmp_path: Path,
+    field: str,
+) -> None:
+    queue = _queue(tmp_path)
+    queue.submit(_intent(), reject_duplicate_computation=True)
+    values = {
+        "source_session_id": "another-source",
+        "source_catalog_sha256": "0" * 64,
+        "source_bundle_sha256": "0" * 64,
+        "source_capability_manifest_sha256": "0" * 64,
+        "setup_id": "legacy-monolith-v1",
+    }
+    changes = {field: values[field]}
+    if field == "setup_id":
+        changes["definition_sha256"] = NON_CHECKPOINTABLE_DEFINITION_SHA
+    distinct = replace(_intent(idempotency_key="distinct-request"), **changes)
+    _, created = queue.submit(distinct, reject_duplicate_computation=True)
+    assert created and len(queue.list_jobs()) == 2
+
+
+@pytest.mark.parametrize("publication_failed", [False, True])
+def test_duplicate_guard_never_recomputes_pending_publication(
+    tmp_path: Path,
+    publication_failed: bool,
+) -> None:
+    queue = _queue(tmp_path)
+    job, claim = _running_job(queue)
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError):
+        queue.submit(
+            _intent(idempotency_key="while-running"),
+            reject_duplicate_computation=True,
+        )
+    queue.complete_for_publication(
+        job.job_id,
+        claim_token=claim.claim_token,
+        result_id="portable-result-001",
+        result_sha256=RESULT_SHA,
+    )
+    if publication_failed:
+        queue.mark_publication_failed(job.job_id, message="temporary publication outage")
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError):
+        queue.submit(
+            _intent(idempotency_key="while-publishing"),
+            reject_duplicate_computation=True,
+        )
+    assert len(queue.list_jobs()) == 1
+
+
+def test_duplicate_guard_allows_retry_after_computation_failure(tmp_path: Path) -> None:
+    queue = _queue(tmp_path)
+    job, claim = _running_job(queue)
+    queue.fail(
+        job.job_id,
+        claim_token=claim.claim_token,
+        error_code="executor-error",
+        message="bounded test failure",
+    )
+    retried, created = queue.submit(
+        _intent(idempotency_key="explicit-retry"),
+        enqueue=True,
+        reject_duplicate_computation=True,
+    )
+    assert created and retried.job_id != job.job_id
+
+
+def test_duplicate_guard_is_atomic_across_two_queue_instances(tmp_path: Path) -> None:
+    queues = (_queue(tmp_path), _queue(tmp_path))
+    barrier = Barrier(2)
+
+    def submit(index: int) -> str:
+        barrier.wait(timeout=5)
+        try:
+            job, created = queues[index].submit(
+                _intent(idempotency_key=f"browser-{index}"),
+                enqueue=True,
+                reject_duplicate_computation=True,
+            )
+            assert created
+            return job.job_id
+        except ObservatoryRecordedQueueDuplicateError as error:
+            return error.job_id
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = tuple(executor.map(submit, (0, 1)))
+    assert results[0] == results[1]
+    assert len(queues[0].list_jobs()) == 1
+
+
+def test_duplicate_guard_keeps_new_version_of_same_setup(tmp_path: Path) -> None:
+    queue = _queue(tmp_path)
+    queue.submit(_intent(), reject_duplicate_computation=True)
+    revised = replace(
+        _definitions().definitions[0],
+        definition_version=2,
+        definition_sha256="0" * 64,
+    )
+    reopened = ObservatoryRecordedJobQueue(
+        tmp_path,
+        definitions=RecordedRunDefinitionRegistry((revised,)),
+        clock=lambda: NOW,
+    )
+    _, created = reopened.submit(
+        _intent(idempotency_key="new-version", definition_sha256="0" * 64),
+        reject_duplicate_computation=True,
+    )
+    assert created and len(reopened.list_jobs()) == 2
+
+
 def test_submission_is_durable_exactly_idempotent_and_path_free(tmp_path: Path) -> None:
     queue = _queue(tmp_path)
     first, first_created = queue.submit(_intent())