Bind publication evidence to source CSV

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

Author: Pigbibi

.github/workflows/rss_source_pipeline.yml

@@ -52,7 +52,7 @@ jobs:
       - uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/PoliticalEventTrackingResearch
-          ref: refs/heads/main
+          ref: ${{ github.sha }}
           persist-credentials: false
       - name: Verify reviewed main checkout and canonical paths
         env:
@@ -76,7 +76,7 @@ jobs:
       - uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/PoliticalEventTrackingResearch
-          ref: refs/heads/main
+          ref: ${{ github.sha }}
           persist-credentials: false
       - name: Verify reviewed main checkout
         env:

scripts/validate_publish_input_evidence.py

@@ -2,15 +2,14 @@
 from __future__ import annotations
 
 import argparse
-import csv
 from pathlib import Path
 import sys
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
-from political_event_tracking_research.feed_status_canonical_h2c import read_status  # noqa: E402
 from political_event_tracking_research.publish_input_policy import (  # noqa: E402
     build_publication_evidence,
+    recompute_source_items_binding,
     read_input_policy_evidence,
     read_publication_evidence,
 )
@@ -30,19 +29,17 @@ def main() -> None:
         evidence = read_publication_evidence(args.evidence.read_bytes())
         source_bytes = args.source_items.read_bytes()
         status_bytes = args.status.read_bytes()
-        status = read_status(status_bytes)
-        with args.source_items.open(newline="", encoding="utf-8") as handle:
-            actual_row_count = sum(1 for _ in csv.DictReader(handle))
+        actual_row_count, aggregate_digest, status_eligible = recompute_source_items_binding(source_bytes, status_bytes)
         if actual_row_count != evidence["source_items_row_count"]:
             raise ValueError("source_items_row_count_mismatch")
         actual = build_publication_evidence(
             policy,
             root=args.root,
             source_items_bytes=source_bytes,
             status_bytes=status_bytes,
-            status_eligible=status["eligible_for_live_publication"],
+            status_eligible=status_eligible,
             source_items_row_count=actual_row_count,
-            aggregate_row_digest=status["aggregate_row_digest"],
+            aggregate_row_digest=aggregate_digest,
         )
         if actual != evidence:
             raise ValueError("publication_evidence_mismatch")

scripts/write_publish_input_evidence.py

@@ -2,15 +2,14 @@
 from __future__ import annotations
 
 import argparse
-import csv
 from pathlib import Path
 import sys
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
-from political_event_tracking_research.feed_status_canonical_h2c import read_status  # noqa: E402
 from political_event_tracking_research.publish_input_policy import (  # noqa: E402
     build_publication_evidence,
+    recompute_source_items_binding,
     read_input_policy_evidence,
     read_publication_evidence,
     serialize_publication_evidence,
@@ -29,17 +28,15 @@ def main() -> None:
         policy = read_input_policy_evidence(args.policy.read_bytes())
         source_bytes = args.source_items.read_bytes()
         status_bytes = args.status.read_bytes()
-        status = read_status(status_bytes)
-        with args.source_items.open(newline="", encoding="utf-8") as handle:
-            row_count = sum(1 for _ in csv.DictReader(handle))
+        row_count, aggregate_digest, status_eligible = recompute_source_items_binding(source_bytes, status_bytes)
         evidence = build_publication_evidence(
             policy,
             root=args.root,
             source_items_bytes=source_bytes,
             status_bytes=status_bytes,
-            status_eligible=status["eligible_for_live_publication"],
+            status_eligible=status_eligible,
             source_items_row_count=row_count,
-            aggregate_row_digest=status["aggregate_row_digest"],
+            aggregate_row_digest=aggregate_digest,
         )
         if read_publication_evidence(evidence) != evidence:
             raise ValueError("publication_evidence_readback_invalid")

src/political_event_tracking_research/feed_status_canonical_h2c.py

@@ -102,6 +102,11 @@ def _digest(rows: list[dict[str, str]]) -> str:
     return hashlib.sha256(payload).hexdigest()
 
 
+def digest_rows(rows: list[dict[str, str]]) -> str:
+    """Return the H2C aggregate digest for an already validated row snapshot."""
+    return _digest(rows)
+
+
 def _parse_outcome(value: object) -> tuple[dict[str, object], list[dict[str, str]]]:
     data = _mapping(value, _OUTCOME_KEYS, "outcome_invalid")
     kind = data["kind"]

src/political_event_tracking_research/publish_input_policy.py

@@ -1,12 +1,16 @@
 from __future__ import annotations
 
+import csv
 import hashlib
+import io
 import json
 import re
 import stat
 from collections.abc import Mapping
 from pathlib import Path
 
+from .feed_status_canonical_h2c import EMPTY_DIGEST, digest_rows, read_status
+
 
 PUBLISH_MAX_ITEMS_PER_FEED = 50
 MAX_DEBUG_ITEMS_PER_FEED = 10_000
@@ -19,6 +23,7 @@
 WORKFLOW_PATH = ".github/workflows/rss_source_pipeline.yml"
 WORKFLOW_REF = f"{REPOSITORY}/{WORKFLOW_PATH}@refs/heads/main"
 POLICY_VERSION = "pert.rss_publish_input.v1"
+SOURCE_ITEM_FIELDS = ("item_id", "published_at", "source_type", "source_url", "author", "text")
 _SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 _SHA1_RE = re.compile(r"^[0-9a-f]{40}$")
 _POLICY_KEYS = frozenset(
@@ -58,6 +63,40 @@ def _fail(code: str) -> None:
     raise PublishInputPolicyError(code)
 
 
+def recompute_source_items_binding(source_items_bytes: bytes, status_bytes: bytes) -> tuple[int, str, bool]:
+    if type(source_items_bytes) is not bytes or type(status_bytes) is not bytes:
+        _fail("publication_evidence_invalid")
+    try:
+        reader = csv.DictReader(io.StringIO(source_items_bytes.decode("utf-8"), newline=""))
+        if tuple(reader.fieldnames or ()) != SOURCE_ITEM_FIELDS:
+            _fail("source_items_schema_invalid")
+        rows: list[dict[str, str]] = []
+        for row in reader:
+            if set(row) != set(SOURCE_ITEM_FIELDS) or any(type(value) is not str for value in row.values()):
+                _fail("source_items_schema_invalid")
+            rows.append({key: row[key] for key in SOURCE_ITEM_FIELDS})
+        canonical = io.StringIO(newline="")
+        writer = csv.DictWriter(canonical, fieldnames=SOURCE_ITEM_FIELDS, lineterminator="\n")
+        writer.writeheader()
+        writer.writerows(rows)
+        if canonical.getvalue().encode("utf-8") != source_items_bytes:
+            _fail("source_items_noncanonical")
+        status = read_status(status_bytes)
+    except (csv.Error, UnicodeError, ValueError):
+        _fail("source_items_invalid")
+    row_count = len(rows)
+    aggregate_digest = digest_rows(rows) if rows else EMPTY_DIGEST
+    derived_eligible = status["failed_feed_count"] == 0 and status["quarantined_feed_count"] == 0
+    if (
+        row_count != status["accepted_row_count"]
+        or aggregate_digest != status["aggregate_row_digest"]
+        or status["publication_complete"] is not derived_eligible
+        or status["eligible_for_live_publication"] is not derived_eligible
+    ):
+        _fail("source_items_status_mismatch")
+    return row_count, aggregate_digest, derived_eligible
+
+
 def _mapping(value: object, keys: frozenset[str], code: str) -> dict[str, object]:
     if not isinstance(value, Mapping):
         _fail(code)

tests/test_publish_input_policy.py

@@ -13,7 +13,9 @@
     build_publication_evidence,
     read_input_policy_evidence,
     read_publication_evidence,
+    recompute_source_items_binding,
 )
+from political_event_tracking_research.feed_status_canonical_h2c import build_decision
 
 
 def write_inputs(root: Path) -> None:
@@ -171,6 +173,62 @@ def test_debug_evidence_can_readback_without_becoming_publishable(tmp_path: Path
     assert evidence["status_eligible_for_live_publication"] is True
 
 
+def test_source_items_binding_recomputes_count_digest_and_publishability() -> None:
+    row = {
+        "item_id": "a-1",
+        "published_at": "2026-05-01T12:30:00Z",
+        "source_type": "official",
+        "source_url": "https://example.test/a",
+        "author": "",
+        "text": "event",
+    }
+    status = build_decision(
+        [
+            {
+                "feed_id": "feed-a",
+                "feed_url": "https://example.test/feed-a",
+                "kind": "rss2",
+                "state": "accepted",
+                "rows": [row],
+                "error_code": None,
+            }
+        ]
+    )
+    source = (
+        "item_id,published_at,source_type,source_url,author,text\n"
+        "a-1,2026-05-01T12:30:00Z,official,https://example.test/a,,event\n"
+    ).encode()
+    count, digest, eligible = recompute_source_items_binding(source, status.status_bytes)
+    assert (count, digest, eligible) == (1, json.loads(status.status_bytes)["aggregate_row_digest"], True)
+
+
+def test_source_items_binding_rejects_count_or_digest_drift() -> None:
+    status = build_decision(
+        [
+            {
+                "feed_id": "feed-a",
+                "feed_url": "https://example.test/feed-a",
+                "kind": "rss2",
+                "state": "accepted",
+                "rows": [
+                    {
+                        "item_id": "a-1",
+                        "published_at": "2026-05-01T12:30:00Z",
+                        "source_type": "official",
+                        "source_url": "https://example.test/a",
+                        "author": "",
+                        "text": "event",
+                    }
+                ],
+                "error_code": None,
+            }
+        ]
+    )
+    tampered = b"item_id,published_at,source_type,source_url,author,text\na-1,2026-05-01T12:30:00Z,official,https://example.test/a,,changed\n"
+    with pytest.raises(PublishInputPolicyError, match="source_items_status_mismatch"):
+        recompute_source_items_binding(tampered, status.status_bytes)
+
+
 def test_workflow_guard_and_evidence_precede_fetch_and_publish() -> None:
     workflow = Path(__file__).parents[1].joinpath(".github/workflows/rss_source_pipeline.yml").read_text(
         encoding="utf-8"
@@ -182,4 +240,5 @@ def test_workflow_guard_and_evidence_precede_fetch_and_publish() -> None:
     assert workflow.index("Upload RSS source artifact") < workflow.index("Build and validate publication evidence")
     assert workflow.index("Build and validate publication evidence") < workflow.index("Publish live CSV outputs")
     assert '--max-items-per-feed "${MAX_ITEMS_PER_FEED}"' in workflow
+    assert workflow.count("ref: ${{ github.sha }}") == 2
     assert "git push origin HEAD:refs/heads/main" in workflow