fix(rag): replay evidence limitation integrity

docs/changelog.d/1135-rag-evidence-replay-integrity.md

@@ -0,0 +1,5 @@
+# RAG evidence limitation replay integrity
+
+## Fixed
+
+- Replay factory-derived RAG evidence limitation records before manifest or fingerprint projection so post-construction mutation fails closed before caller callbacks can execute.

python/fast_mlsirm/scoring/rag_evidence.py

@@ -84,17 +84,42 @@ def _content_dict(self) -> dict[str, Any]:
 
     @property
     def limitations_fingerprint(self) -> str:
-        """Return SHA-256 over the exact interpretation-limit contract."""
-        return artifact_digest(self)
+        """Return SHA-256 over the replay-validated interpretation-limit contract."""
+        replayed = _replay_limitations(self)
+        return artifact_digest(replayed)
 
     def to_dict(self) -> dict[str, Any]:
-        """Return canonical limitation content plus its deterministic digest."""
+        """Return replay-validated limitation content and deterministic digest."""
+        replayed = _replay_limitations(self)
         return {
-            **self._content_dict(),
-            "limitations_fingerprint": self.limitations_fingerprint,
+            **RAGEvidenceRegimeLimitations._content_dict(replayed),
+            "limitations_fingerprint": artifact_digest(replayed),
         }
 
 
+def _replay_limitations(
+    value: RAGEvidenceRegimeLimitations,
+) -> RAGEvidenceRegimeLimitations:
+    """Re-establish exact factory-derived state before manifest projection."""
+    if (
+        type(value) is not RAGEvidenceRegimeLimitations
+        or type(value.regime) is not RAGEvidenceRegime
+        or type(value.limitation_codes) is not tuple
+        or any(type(code) is not str for code in value.limitation_codes)
+        or value.limitation_codes != _REGIME_LIMITATIONS[value.regime]
+    ):
+        raise assessment_error(
+            "invalid_rag_evidence_limitations",
+            "$",
+            "RAG evidence limitations must retain factory-derived state",
+        )
+    return RAGEvidenceRegimeLimitations(
+        regime=value.regime,
+        limitation_codes=_REGIME_LIMITATIONS[value.regime],
+        _limitations_token=_RAG_EVIDENCE_LIMITATIONS_TOKEN,
+    )
+
+
 def rag_evidence_regime_limitations(
     regime: RAGEvidenceRegime | str,
 ) -> RAGEvidenceRegimeLimitations:

tests/test_scoring_rag_evidence_limitations.py

@@ -71,6 +71,34 @@ def test_limitations_contract_is_factory_sealed() -> None:
         )
 
 
+def test_limitations_replay_rejects_rebound_container_before_callbacks() -> None:
+    """Post-construction container rebinding cannot execute during replay."""
+    callbacks: list[str] = []
+
+    class HostileTuple(tuple):
+        def __iter__(self):  # type: ignore[override]
+            """Fail if package replay invokes caller-controlled iteration."""
+            callbacks.append("iter")
+            raise AssertionError("hostile limitation iteration executed")
+
+    for projection in (
+        lambda value: value.to_dict(),
+        lambda value: value.limitations_fingerprint,
+    ):
+        limits = rag_evidence_regime_limitations(RAGEvidenceRegime.RETRIEVED_CONTEXT)
+        object.__setattr__(
+            limits,
+            "limitation_codes",
+            HostileTuple(limits.limitation_codes),
+        )
+
+        with pytest.raises(AssessmentSpecError) as captured:
+            projection(limits)
+
+        assert captured.value.code == "invalid_rag_evidence_limitations"
+        assert callbacks == []
+
+
 def test_evidence_limitations_are_explicit_scoring_package_attributes() -> None:
     """Governed reporting callers should not need an internal-module import path."""
     assert scoring.RAGEvidenceRegimeLimitations is RAGEvidenceRegimeLimitations