Accept KR predicate_id compatibility alias

src/backend/services/kr_materialisation_guard_service.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 from collections import Counter
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping, MutableMapping, Sequence
 from typing import Any
 
 
@@ -332,6 +332,29 @@ def _relationship_reference(
     return ("fixed", concept_id), None
 
 
+def _normalise_relationship_predicate(
+    spec: Mapping[str, Any],
+    *,
+    invalid_code: str,
+    apply_alias: bool = False,
+) -> tuple[str | None, str | None]:
+    """Canonicalise the model-facing ``predicate_id`` alias before execution."""
+
+    predicate = _text(spec.get("predicate"))
+    predicate_id = _text(spec.get("predicate_id"))
+    if predicate and predicate_id and predicate != predicate_id:
+        return None, invalid_code
+    if predicate:
+        return predicate, None
+    if not predicate_id:
+        return "", None
+    if not isinstance(spec, MutableMapping):
+        return None, invalid_code
+    if apply_alias:
+        spec["predicate"] = predicate_id
+    return predicate_id, None
+
+
 def _matching_rule_ids(
     *,
     contract: Mapping[str, Any],
@@ -384,7 +407,12 @@ def _validate_relationship_specs(
         )
         if source_error or target_error or source is None or target is None:
             return None, source_error or target_error
-        predicate = _text(raw_spec.get("predicate"))
+        predicate, predicate_error = _normalise_relationship_predicate(
+            raw_spec,
+            invalid_code="kr_materialisation_guard_relationship_spec_invalid",
+        )
+        if predicate_error is not None or predicate is None:
+            return None, predicate_error
         matches = _matching_rule_ids(
             contract=contract,
             predicate=predicate,
@@ -430,6 +458,14 @@ def _validate_relationship_specs(
         referenced_fixed_ids
     ):
         return None, "kr_materialisation_guard_fixed_endpoint_missing"
+    for raw_spec in raw_specs:
+        if not isinstance(raw_spec, Mapping):
+            continue
+        _normalise_relationship_predicate(
+            raw_spec,
+            invalid_code="kr_materialisation_guard_relationship_spec_invalid",
+            apply_alias=True,
+        )
     return normalised, None
 
 
@@ -558,7 +594,18 @@ def validate_kr_materialisation_guard(
                 phase=phase_text,
             )
         source_id = _text(raw_spec.get("source_id"))
-        predicate = _text(raw_spec.get("predicate"))
+        predicate, predicate_error = _normalise_relationship_predicate(
+            raw_spec,
+            invalid_code=(
+                "kr_materialisation_guard_resolved_relationship_invalid"
+            ),
+        )
+        if predicate_error is not None or predicate is None:
+            return _reject(
+                predicate_error
+                or "kr_materialisation_guard_resolved_relationship_invalid",
+                phase=phase_text,
+            )
         target_id = _text(raw_spec.get("target_id"))
         triple = (source_id, predicate, target_id)
         if (
@@ -576,6 +623,14 @@ def validate_kr_materialisation_guard(
             "kr_materialisation_guard_resolved_relationship_rejected",
             phase=phase_text,
         )
+    for raw_spec in raw_resolved:
+        if not isinstance(raw_spec, Mapping):
+            continue
+        _normalise_relationship_predicate(
+            raw_spec,
+            invalid_code="kr_materialisation_guard_resolved_relationship_invalid",
+            apply_alias=True,
+        )
     return _pass(
         phase=phase_text,
         guard_applied=True,

tests/backend/test_kr_materialisation_guard_service.py

@@ -145,6 +145,85 @@ def test_guard_accepts_exact_create_and_changed_record_reuse_plan() -> None:
     assert resolved["guard_passed"] is True
 
 
+def test_guard_normalises_predicate_id_alias_before_downstream_execution() -> None:
+    relationship_specs = [
+        {
+            "source_key": "candidate",
+            "predicate_id": "#V#has_doctoral_programme",
+            "target_key": "programme",
+        }
+    ]
+    resolved_relationship_specs = [
+        {
+            "source_id": "#V#stable_candidate",
+            "predicate_id": "#V#has_doctoral_programme",
+            "target_id": "#V#stable_programme",
+        }
+    ]
+
+    plan = validate_kr_materialisation_guard(
+        guard_contract=_guard(),
+        phase="plan",
+        concept_specs=_concept_specs(),
+        relationship_specs=relationship_specs,
+    )
+    resolved = validate_kr_materialisation_guard(
+        guard_contract=_guard(),
+        phase="resolved_relationships",
+        concept_specs=_concept_specs(),
+        relationship_specs=relationship_specs,
+        concept_iteration_results=_concept_results(),
+        resolved_relationship_specs=resolved_relationship_specs,
+    )
+
+    assert plan["guard_passed"] is True
+    assert resolved["guard_passed"] is True
+    assert relationship_specs[0]["predicate"] == "#V#has_doctoral_programme"
+    assert (
+        resolved_relationship_specs[0]["predicate"]
+        == "#V#has_doctoral_programme"
+    )
+
+
+def test_guard_rejects_conflicting_predicate_aliases() -> None:
+    conflicting_plan = _relationship_specs()
+    conflicting_plan[0]["predicate_id"] = "#V#has_administrator"
+    plan = validate_kr_materialisation_guard(
+        guard_contract=_guard(),
+        phase="plan",
+        concept_specs=_concept_specs(),
+        relationship_specs=conflicting_plan,
+    )
+
+    conflicting_resolved = [
+        {
+            "source_id": "#V#stable_candidate",
+            "predicate": "#V#has_doctoral_programme",
+            "predicate_id": "#V#has_administrator",
+            "target_id": "#V#stable_programme",
+        }
+    ]
+    resolved = validate_kr_materialisation_guard(
+        guard_contract=_guard(),
+        phase="resolved_relationships",
+        concept_specs=_concept_specs(),
+        relationship_specs=_relationship_specs(),
+        concept_iteration_results=_concept_results(),
+        resolved_relationship_specs=conflicting_resolved,
+    )
+
+    assert plan["guard_passed"] is False
+    assert (
+        plan["error_code"]
+        == "kr_materialisation_guard_relationship_spec_invalid"
+    )
+    assert resolved["guard_passed"] is False
+    assert (
+        resolved["error_code"]
+        == "kr_materialisation_guard_resolved_relationship_invalid"
+    )
+
+
 def test_guard_rejects_injected_unrelated_concept_before_any_write() -> None:
     injected = _concept_specs() + [
         {