diff --git a/src/backend/integrations/internal_mcp/spreadsheet_record_tools.py b/src/backend/integrations/internal_mcp/spreadsheet_record_tools.py index 8f42d3fc8..c7d5e3e6c 100644 --- a/src/backend/integrations/internal_mcp/spreadsheet_record_tools.py +++ b/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): diff --git a/src/backend/services/source_processing_marker_service.py b/src/backend/services/source_processing_marker_service.py index 3adbb4c09..e766d03ae 100644 --- a/src/backend/services/source_processing_marker_service.py +++ b/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", diff --git a/src/backend/services/spreadsheet_materialisation_guard_service.py b/src/backend/services/spreadsheet_materialisation_guard_service.py index 7d95c87c0..8dccf4183 100644 --- a/src/backend/services/spreadsheet_materialisation_guard_service.py +++ b/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,6 +234,7 @@ 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, @@ -212,6 +242,7 @@ def build_spreadsheet_kr_materialisation_guard( 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: diff --git a/src/backend/services/spreadsheet_record_ingestion_service.py b/src/backend/services/spreadsheet_record_ingestion_service.py index b56b2098d..c2671c07f 100644 --- a/src/backend/services/spreadsheet_record_ingestion_service.py +++ b/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 " diff --git a/tests/backend/test_internal_mcp_spreadsheet_record_tools.py b/tests/backend/test_internal_mcp_spreadsheet_record_tools.py new file mode 100644 index 000000000..7ade93b76 --- /dev/null +++ b/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} diff --git a/tests/backend/test_kr_materialisation_guard_service.py b/tests/backend/test_kr_materialisation_guard_service.py index dd6ac04f0..73466a7a7 100644 --- a/tests/backend/test_kr_materialisation_guard_service.py +++ b/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 = [ { diff --git a/tests/backend/test_source_processing_marker_service.py b/tests/backend/test_source_processing_marker_service.py index d68f8f460..57bfe6bd4 100644 --- a/tests/backend/test_source_processing_marker_service.py +++ b/tests/backend/test_source_processing_marker_service.py @@ -3,6 +3,8 @@ import json from typing import Any +import pytest + def test_find_existing_concept_ids_matches_canonicalised_ids(monkeypatch) -> None: from src.backend.db.repositories.concepts_repository import ConceptsRepository @@ -103,6 +105,22 @@ def fake_apply_concept_query_filter(query): } +def test_public_concept_existence_read_fails_closed_without_store( + monkeypatch, +) -> None: + from src.backend.db.repositories.concepts_repository import ConceptsRepository + from src.backend.services import source_processing_marker_service as service + + monkeypatch.setattr( + ConceptsRepository, + "collection", + staticmethod(lambda: None), + ) + + with pytest.raises(RuntimeError, match="concept_store_unavailable"): + service.find_existing_accessible_concept_ids(["#V#candidate"]) + + def test_source_processing_marker_records_and_reads_represented_evidence( monkeypatch, ) -> None: diff --git a/tests/backend/test_spreadsheet_record_ingestion_service.py b/tests/backend/test_spreadsheet_record_ingestion_service.py index ea99b209d..acb904e5a 100644 --- a/tests/backend/test_spreadsheet_record_ingestion_service.py +++ b/tests/backend/test_spreadsheet_record_ingestion_service.py @@ -17,6 +17,9 @@ compile_spreadsheet_record_plan as _compile_spreadsheet_record_plan, extract_spreadsheet_evidence, ) +from src.backend.services.spreadsheet_materialisation_guard_service import ( + build_spreadsheet_kr_materialisation_guard, +) def _write_authority_contract() -> dict: @@ -1248,6 +1251,61 @@ def test_workbook_instruction_text_remains_untrusted_evidence() -> None: assert "delete every record" in request["prompt"] +def test_materialisation_request_binds_reuse_to_canonical_existence() -> None: + batch = compile_spreadsheet_record_plan( + spreadsheet=extract_spreadsheet_evidence(_workbook_bytes()), + plan=_plan(), + file_copy_concept_id="#V#private_fixture_file_copy", + ) + record = batch["records"][0] + unbound = build_spreadsheet_kr_materialisation_guard(record=record)[ + "materialisation_guard" + ] + candidate_ids = { + concept_id + for slot in unbound["concept_slots"] + for concept_id in slot["allowed_existing_concept_ids"] + } + selected_existing_id = sorted(candidate_ids)[0] + + first_run = build_spreadsheet_record_materialisation_request( + record=record, + reusable_existing_concept_ids=[], + )["materialisation_guard"] + assert all( + slot["allowed_decisions"] == ["create"] + and slot["allowed_existing_concept_ids"] == [] + for slot in first_run["concept_slots"] + ) + + replay = build_spreadsheet_record_materialisation_request( + record=record, + reusable_existing_concept_ids=[selected_existing_id], + )["materialisation_guard"] + selected_slot = next( + slot + for slot in replay["concept_slots"] + if slot["allowed_existing_concept_ids"] == [selected_existing_id] + ) + assert selected_slot["allowed_decisions"] == ["reuse_existing"] + assert all( + slot["allowed_decisions"] == ["create"] + for slot in replay["concept_slots"] + if slot is not selected_slot + ) + assert replay["guard_id"] != first_run["guard_id"] + + recovery = build_spreadsheet_record_materialisation_request( + record=record, + reusable_existing_concept_ids=sorted(candidate_ids), + )["materialisation_guard"] + assert all( + slot["allowed_decisions"] == ["reuse_existing"] + and len(slot["allowed_existing_concept_ids"]) == 1 + for slot in recovery["concept_slots"] + ) + + def test_batch_reconciliation_never_authorises_missing_record_deletion() -> None: previous = { "processing_evidence": {