Bind spreadsheet reuse to canonical concept existence

src/backend/integrations/internal_mcp/spreadsheet_record_tools.py

@@ -49,6 +49,12 @@ def _build_spreadsheet_record_materialisation_request(**kwargs):
     from ...services.spreadsheet_record_ingestion_service import (
         build_spreadsheet_record_materialisation_request,
     )
+    from ...services.source_processing_marker_service import (
+        find_existing_accessible_concept_ids,
+    )
+    from ...services.spreadsheet_materialisation_guard_service import (
+        build_spreadsheet_kr_materialisation_guard,
+    )
 
     record = kwargs.get("record")
     if not isinstance(record, Mapping):
@@ -57,7 +63,36 @@ def _build_spreadsheet_record_materialisation_request(**kwargs):
             "Missing validated spreadsheet record evidence.",
             details={"missing": ["record"]},
         )
-    return build_spreadsheet_record_materialisation_request(record=record)
+    guard_preview = build_spreadsheet_kr_materialisation_guard(record=record)
+    if not guard_preview.get("success"):
+        return guard_preview
+    guard_contract = guard_preview.get("materialisation_guard")
+    slots = (
+        guard_contract.get("concept_slots")
+        if isinstance(guard_contract, Mapping)
+        else None
+    )
+    candidate_ids = sorted(
+        {
+            str(concept_id).strip()
+            for slot in (slots if isinstance(slots, list) else [])
+            if isinstance(slot, Mapping)
+            for concept_id in (slot.get("allowed_existing_concept_ids") or [])
+            if isinstance(concept_id, str) and concept_id.strip()
+        }
+    )
+    try:
+        reusable_ids = find_existing_accessible_concept_ids(candidate_ids)
+    except Exception:
+        return make_error_response(
+            "spreadsheet_reuse_authority_read_failed",
+            "Could not bind idempotent reuse authority from canonical read-back.",
+            details={"candidate_count": len(candidate_ids)},
+        )
+    return build_spreadsheet_record_materialisation_request(
+        record=record,
+        reusable_existing_concept_ids=sorted(reusable_ids),
+    )
 
 
 def _compare_spreadsheet_record_batch(**kwargs):

src/backend/services/source_processing_marker_service.py

@@ -198,6 +198,18 @@ def _find_existing_concept_ids(concept_ids: Sequence[str]) -> set[str]:
     }
 
 
+def find_existing_accessible_concept_ids(
+    concept_ids: Sequence[str],
+) -> set[str]:
+    """Return exact accessible concept IDs using one bounded indexed read."""
+
+    from ..db.repositories.concepts_repository import ConceptsRepository
+
+    if ConceptsRepository.collection() is None:
+        raise RuntimeError("concept_store_unavailable")
+    return _find_existing_concept_ids(concept_ids)
+
+
 def _ensure_support_concepts(*, existing_ids: set[str] | None = None) -> list[str]:
     # These are global workflow-support concepts.  Resolve their exact IDs in
     # one bounded read instead of seven serial Atlas round trips on every
@@ -613,6 +625,7 @@ def record_source_processing_marker(
 __all__ = [
     "SOURCE_PROCESSING_EVIDENCE_PREDICATE_ID",
     "SOURCE_PROCESSING_MARKER_TYPE_ID",
+    "find_existing_accessible_concept_ids",
     "get_source_processing_marker",
     "record_source_processing_marker",
     "source_processing_marker_concept_id",

src/backend/services/spreadsheet_materialisation_guard_service.py

@@ -120,6 +120,7 @@ def _slot(
     identity: Any,
     stable_name: str | None = None,
     identity_namespace: str | None = None,
+    reusable_existing_concept_ids: set[str] | None = None,
 ) -> dict[str, Any]:
     namespace = identity_namespace or record_id
     token = _digest(
@@ -132,23 +133,51 @@ def _slot(
     )
     resolved_name = stable_name or f"{namespace}-{role}-{token[:16]}"
     stable_concept_id = canonicalise_vontology_concept_id(resolved_name)
+    if reusable_existing_concept_ids is None:
+        allowed_decisions = ["create", "reuse_existing"]
+        allowed_existing_concept_ids = (
+            [stable_concept_id] if stable_concept_id else []
+        )
+    elif stable_concept_id and stable_concept_id in reusable_existing_concept_ids:
+        allowed_decisions = ["reuse_existing"]
+        allowed_existing_concept_ids = [stable_concept_id]
+    else:
+        allowed_decisions = ["create"]
+        allowed_existing_concept_ids = []
     return {
         "key": f"slot_{token[:24]}",
         "stable_name": resolved_name,
         "target_kind": "instance",
         "parent_id": parent_id,
-        "allowed_decisions": ["create", "reuse_existing"],
-        "allowed_existing_concept_ids": (
-            [stable_concept_id] if stable_concept_id else []
-        ),
+        "allowed_decisions": allowed_decisions,
+        "allowed_existing_concept_ids": allowed_existing_concept_ids,
         "allow_unreferenced": False,
     }
 
 
 def build_spreadsheet_kr_materialisation_guard(
-    *, record: Mapping[str, Any]
+    *,
+    record: Mapping[str, Any],
+    reusable_existing_concept_ids: Sequence[str] | None = None,
 ) -> dict[str, Any]:
-    """Derive the exact additive-write envelope for one compiled record."""
+    """Derive the exact additive-write envelope for one compiled record.
+
+    When ``reusable_existing_concept_ids`` is provided, each slot is bound to
+    exactly one current-state decision: create when its deterministic concept
+    ID is absent, or reuse when canonical read-back confirms that ID exists.
+    ``None`` preserves the unbound authority template used by planning and
+    guard-shape tests.
+    """
+
+    reusable_ids = (
+        {
+            _text(concept_id)
+            for concept_id in reusable_existing_concept_ids
+            if _text(concept_id)
+        }
+        if reusable_existing_concept_ids is not None
+        else None
+    )
 
     record_id = _text(record.get("source_record_id"))
     version_id = _text(record.get("source_record_version_id"))
@@ -205,13 +234,15 @@ def build_spreadsheet_kr_materialisation_guard(
             parent_id=source_record_parent,
             identity="source-record",
             stable_name=record_id,
+            reusable_existing_concept_ids=reusable_ids,
         ),
         _slot(
             record_id=record_id,
             role="source-record-version",
             parent_id=source_version_parent,
             identity=record.get("record_fingerprint"),
             stable_name=version_id,
+            reusable_existing_concept_ids=reusable_ids,
         ),
     ]
     stable_identity_slots = {
@@ -245,6 +276,7 @@ def add_slot(
             parent_id=parent_id,
             identity=identity,
             identity_namespace=identity_namespace,
+            reusable_existing_concept_ids=reusable_ids,
         )
         existing = slots_by_key.get(str(created["key"]))
         if existing is not None:

src/backend/services/spreadsheet_record_ingestion_service.py

@@ -1957,7 +1957,9 @@ def _compile_spreadsheet_record_plan(
 
 
 def build_spreadsheet_record_materialisation_request(
-    *, record: Mapping[str, Any]
+    *,
+    record: Mapping[str, Any],
+    reusable_existing_concept_ids: Sequence[str] = (),
 ) -> dict[str, Any]:
     """Serialise one validated record as a bounded KR materialisation request."""
 
@@ -1983,7 +1985,10 @@ def build_spreadsheet_record_materialisation_request(
             "success": False,
             "error_code": "spreadsheet_record_identity_missing",
         }
-    guard = build_spreadsheet_kr_materialisation_guard(record=record)
+    guard = build_spreadsheet_kr_materialisation_guard(
+        record=record,
+        reusable_existing_concept_ids=reusable_existing_concept_ids,
+    )
     if not guard.get("success"):
         return guard
     guard_contract = guard["materialisation_guard"]
@@ -2038,7 +2043,9 @@ def build_spreadsheet_record_materialisation_request(
         "The materialisation_guard is trusted workflow authority. Every concept "
         "spec must use exactly one declared slot key, stable_name, target_kind, "
         "and parent_id with a slot-declared create or reuse_existing decision; "
-        "reuse is limited to the slot's exact stable concept ID. Every "
+        "the allowed decision has already been bound to canonical concept "
+        "existence read-back, and reuse is limited to the slot's exact stable "
+        "concept ID. Every "
         "relationship spec must match "
         "one declared rule. Do not add concepts or relationships outside it. "
         "The runtime validates this contract before any write and again after "

tests/backend/test_internal_mcp_spreadsheet_record_tools.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from src.backend.integrations.internal_mcp import spreadsheet_record_tools
+from src.backend.services import source_processing_marker_service
+from src.backend.services import spreadsheet_materialisation_guard_service
+from src.backend.services import spreadsheet_record_ingestion_service
+
+
+def _guard_preview() -> dict:
+    return {
+        "success": True,
+        "materialisation_guard": {
+            "concept_slots": [
+                {"allowed_existing_concept_ids": ["#V#candidate_b"]},
+                {
+                    "allowed_existing_concept_ids": [
+                        "#V#candidate_a",
+                        "#V#candidate_b",
+                    ]
+                },
+            ]
+        },
+    }
+
+
+def test_materialisation_request_binds_one_bounded_existence_read(
+    monkeypatch,
+) -> None:
+    captured: dict = {}
+    monkeypatch.setattr(
+        spreadsheet_materialisation_guard_service,
+        "build_spreadsheet_kr_materialisation_guard",
+        lambda **_kwargs: _guard_preview(),
+    )
+
+    def _find_existing(candidate_ids):
+        captured["candidate_ids"] = candidate_ids
+        return {"#V#candidate_b"}
+
+    monkeypatch.setattr(
+        source_processing_marker_service,
+        "find_existing_accessible_concept_ids",
+        _find_existing,
+    )
+
+    def _build_request(*, record, reusable_existing_concept_ids):
+        captured["record"] = record
+        captured["reusable_existing_concept_ids"] = (
+            reusable_existing_concept_ids
+        )
+        return {"success": True}
+
+    monkeypatch.setattr(
+        spreadsheet_record_ingestion_service,
+        "build_spreadsheet_record_materialisation_request",
+        _build_request,
+    )
+    record = {"ready_for_materialisation": True}
+
+    result = (
+        spreadsheet_record_tools._build_spreadsheet_record_materialisation_request(
+            record=record
+        )
+    )
+
+    assert result == {"success": True}
+    assert captured == {
+        "candidate_ids": ["#V#candidate_a", "#V#candidate_b"],
+        "record": record,
+        "reusable_existing_concept_ids": ["#V#candidate_b"],
+    }
+
+
+def test_materialisation_request_fails_closed_when_existence_read_fails(
+    monkeypatch,
+) -> None:
+    monkeypatch.setattr(
+        spreadsheet_materialisation_guard_service,
+        "build_spreadsheet_kr_materialisation_guard",
+        lambda **_kwargs: _guard_preview(),
+    )
+
+    def _raise_lookup(_candidate_ids):
+        raise TimeoutError("bounded lookup timed out")
+
+    monkeypatch.setattr(
+        source_processing_marker_service,
+        "find_existing_accessible_concept_ids",
+        _raise_lookup,
+    )
+
+    def _unexpected_build(**_kwargs):
+        raise AssertionError("write authority must not be built after lookup failure")
+
+    monkeypatch.setattr(
+        spreadsheet_record_ingestion_service,
+        "build_spreadsheet_record_materialisation_request",
+        _unexpected_build,
+    )
+
+    result = (
+        spreadsheet_record_tools._build_spreadsheet_record_materialisation_request(
+            record={"ready_for_materialisation": True}
+        )
+    )
+
+    assert result["success"] is False
+    assert result["error_code"] == "spreadsheet_reuse_authority_read_failed"
+    assert result["error_details"] == {"candidate_count": 2}

tests/backend/test_kr_materialisation_guard_service.py

@@ -145,6 +145,65 @@ def test_guard_accepts_exact_create_and_changed_record_reuse_plan() -> None:
     assert resolved["guard_passed"] is True
 
 
+def test_guard_enforces_canonical_existence_bound_decisions() -> None:
+    guard = _guard()
+    guard["concept_slots"][0]["allowed_decisions"] = ["create"]
+    guard["concept_slots"][0]["allowed_existing_concept_ids"] = []
+    guard["concept_slots"][1]["allowed_decisions"] = ["reuse_existing"]
+
+    accepted = validate_kr_materialisation_guard(
+        guard_contract=guard,
+        phase="plan",
+        concept_specs=_concept_specs(),
+        relationship_specs=_relationship_specs(),
+    )
+    assert accepted["guard_passed"] is True
+
+    invalid_reuse = _concept_specs()
+    invalid_reuse[0] = {
+        **invalid_reuse[0],
+        "decision": "reuse_existing",
+        "existing_concept_id": "#V#stable_candidate",
+    }
+    invalid_reuse[0].pop("concepts")
+    rejected_reuse = validate_kr_materialisation_guard(
+        guard_contract=guard,
+        phase="plan",
+        concept_specs=invalid_reuse,
+        relationship_specs=_relationship_specs(),
+    )
+    assert rejected_reuse["guard_passed"] is False
+    assert (
+        rejected_reuse["error_code"]
+        == "kr_materialisation_guard_concept_spec_rejected"
+    )
+
+    invalid_create = _concept_specs()
+    invalid_create[1] = {
+        **invalid_create[1],
+        "decision": "create",
+        "concepts": [
+            {
+                "name": "stable-programme",
+                "kind": "instance",
+                "description": "Updated programme evidence.",
+            }
+        ],
+    }
+    invalid_create[1].pop("existing_concept_id")
+    rejected_create = validate_kr_materialisation_guard(
+        guard_contract=guard,
+        phase="plan",
+        concept_specs=invalid_create,
+        relationship_specs=_relationship_specs(),
+    )
+    assert rejected_create["guard_passed"] is False
+    assert (
+        rejected_create["error_code"]
+        == "kr_materialisation_guard_concept_spec_rejected"
+    )
+
+
 def test_guard_normalises_predicate_id_alias_before_downstream_execution() -> None:
     relationship_specs = [
         {