Harden decision evidence immutability

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/political_event_tracking_research/feed_status_decision.py

@@ -6,6 +6,7 @@
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 from enum import Enum
+from types import MappingProxyType
 
 
 STATUS_VERSION = "pert.feed_status_decision.v1"
@@ -62,7 +63,7 @@ def wire(self) -> dict[str, str]:
 
 @dataclass(frozen=True)
 class StatusEvidence:
-    status: dict[str, object]
+    status: Mapping[str, object]
     canonical_bytes: bytes
 
 
@@ -133,8 +134,16 @@ def _parse_outcome(value: object) -> tuple[dict[str, object], list[_Row]]:
     return {"feed_id": feed_id, "feed_url": feed_url, "kind": kind, "state": state, "error_code": error}, rows
 
 
-def _row_key(row: _Row) -> tuple[str, str]:
-    return row.published_at, row.item_id
+def _row_key(row: _Row) -> tuple[str, ...]:
+    return tuple(getattr(row, key) for key in _ROW_KEYS)
+
+
+def _freeze(value: object) -> object:
+    if isinstance(value, dict):
+        return MappingProxyType({key: _freeze(child) for key, child in value.items()})
+    if isinstance(value, list):
+        return tuple(_freeze(child) for child in value)
+    return value
 
 
 def _digest(rows: list[_Row]) -> str:
@@ -216,7 +225,7 @@ def build_status_decision(outcomes: Iterable[Mapping[str, object]]) -> StatusDec
     if len(set(ids)) != len(ids):
         _fail("feed_duplicate")
     status = _build_status(parsed)
-    evidence = StatusEvidence(status, _canonical(status))
+    evidence = StatusEvidence(_freeze(status), _canonical(status))
     if status["failed_feed_count"]:
         kind = DecisionKind.HARD_FAIL
     elif status["quarantined_feed_count"]:

tests/test_feed_status_decision.py

@@ -101,3 +101,20 @@ def test_wire_is_canonical_and_deterministic() -> None:
     second = build_status_decision([outcome("a", "accepted"), outcome("b", "accepted")])
     assert first.evidence.canonical_bytes == second.evidence.canonical_bytes
     assert json.loads(first.evidence.canonical_bytes)["feeds"][0]["feed_id"] == "a"
+
+
+def test_status_evidence_is_deeply_immutable_and_bound_to_bytes() -> None:
+    result = build_status_decision([outcome("a", "accepted")])
+    with pytest.raises(TypeError):
+        result.evidence.status["feed_count"] = 99
+    with pytest.raises(TypeError):
+        result.evidence.status["feeds"][0]["feed_id"] = "tampered"
+    assert json.loads(result.evidence.canonical_bytes)["feed_count"] == result.evidence.status["feed_count"]
+
+
+def test_digest_sort_is_total_for_equal_published_at_and_item_id() -> None:
+    first = {**ROW, "text": "first"}
+    second = {**ROW, "text": "second"}
+    left = build_status_decision([outcome("a", "accepted", rows=[first, second])])
+    right = build_status_decision([outcome("a", "accepted", rows=[second, first])])
+    assert left.evidence.canonical_bytes == right.evidence.canonical_bytes