Align feed status digest with emitted rows

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

Author: Pigbibi

src/political_event_tracking_research/feed_primitives.py

@@ -74,6 +74,10 @@ def _is_digest(value: object) -> bool:
     return type(value) is str and len(value) == 64 and all(char in "0123456789abcdef" for char in value)
 
 
+def _row_sort_key(row: PrimitiveRow) -> tuple[str, str]:
+    return row.published_at, row.item_id
+
+
 @dataclass(frozen=True, slots=True)
 class PrimitiveRow:
     item_id: str
@@ -156,7 +160,7 @@ def _snapshot_feed(value: object) -> dict[str, Any]:
 
 
 def _feed_wire(feed: Mapping[str, Any]) -> dict[str, object]:
-    rows = [row.to_mapping() for row in feed["rows"]]
+    rows = [row.to_mapping() for row in sorted(feed["rows"], key=_row_sort_key)]
     return {
         "feed_id": feed["feed_id"],
         "feed_url": feed["feed_url"],
@@ -189,7 +193,8 @@ def build_status(feed_records: Iterable[Mapping[str, object]]) -> dict[str, obje
     accepted = sum(item["state"] == "accepted" for item in records)
     failed = sum(item["state"] == "failed" for item in records)
     quarantined = sum(item["state"] == "quarantined" for item in records)
-    rows = [row.to_mapping() for item in records if item["state"] == "accepted" for row in item["rows"]]
+    accepted_rows = [row for item in records if item["state"] == "accepted" for row in item["rows"]]
+    rows = [row.to_mapping() for row in sorted(accepted_rows, key=_row_sort_key)]
     complete = accepted == len(records) and failed == 0 and quarantined == 0 and bool(rows)
     return {
         "status_version": STATUS_VERSION,

src/political_event_tracking_research/rss_source_fetch.py

@@ -126,8 +126,8 @@ def parse_feed_items(
         raise FeedXmlError("feed_xml_invalid") from None
     rows: list[dict[str, str]] = []
 
-    rss_items = root.findall("./channel/item")
-    if rss_items:
+    if root.tag == "rss":
+        rss_items = root.findall("./channel/item")
         for item in rss_items[:max_items]:
             title = child_text(item, ("title",))
             link = rss_item_link(item)

tests/test_feed_primitives.py

@@ -70,6 +70,12 @@ def test_failed_rows_are_excluded_and_mixed_status_is_not_complete() -> None:
     ) == status
 
 
+def test_digest_matches_emitted_row_order() -> None:
+    late = {**ROW, "item_id": "feed-a-2", "published_at": "2026-05-02T12:30:00Z"}
+    status = build_status([feed("a", rows=[late, ROW])])
+    assert status["aggregate_row_digest"] == build_status([feed("a", rows=[ROW, late])])["aggregate_row_digest"]
+
+
 def test_canonical_roundtrip_and_order_are_strict() -> None:
     status = build_status([feed("b"), feed("a")])
     wire = serialize_status(status)

tests/test_rss_source_fetch.py

@@ -198,26 +198,6 @@ def fake_fetch(_url: str) -> bytes:
         )
 
 
-def test_all_zero_entry_feeds_write_quarantine_status_without_hard_failure(tmp_path: Path) -> None:
-    feeds_path = tmp_path / "feeds.csv"
-    feeds_path.write_text(
-        "feed_id,feed_url,source_type,author\n"
-        "empty,https://example.invalid/empty.xml,official_remarks,Example\n",
-        encoding="utf-8",
-    )
-    feed_xml = b"<rss version='2.0'><channel/></rss>"
-
-    output = tmp_path / "source_items.csv"
-    status = tmp_path / "status.json"
-    rows = fetch_rss_sources(feeds_path, output, status_output=status, fetcher=lambda _url: feed_xml)
-
-    assert rows == []
-    assert output.read_text(encoding="utf-8") == "item_id,published_at,source_type,source_url,author,text\n"
-    payload = json.loads(status.read_text(encoding="utf-8"))
-    assert payload["quarantined_feed_count"] == 1
-    assert payload["accepted_row_count"] == 0
-    assert payload["publication_complete"] is False
-    assert payload["eligible_for_live_publication"] is False
 
 
 def test_all_zero_entry_feeds_write_quarantine_status_without_hard_failure(tmp_path: Path) -> None:
@@ -236,6 +216,7 @@ def test_all_zero_entry_feeds_write_quarantine_status_without_hard_failure(tmp_p
     assert rows == []
     assert output.read_text(encoding="utf-8") == "item_id,published_at,source_type,source_url,author,text\n"
     payload = json.loads(status.read_text(encoding="utf-8"))
+    assert payload["feeds"][0]["kind"] == "rss"
     assert payload["quarantined_feed_count"] == 1
     assert payload["accepted_row_count"] == 0
     assert payload["publication_complete"] is False