Gate RSS live publication on complete fetches

Pigbibi · codex · Pigbibi · commit b0748fe27b67 · 2026-07-16T01:46:32.000+08:00
Co-Authored-By: Codex &lt;noreply@openai.com&gt;
diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml
@@ -4,17 +4,17 @@ on:
   workflow_dispatch:
     inputs:
       feeds_path:
-        description: "RSS feed config CSV path."
+        description: "Must remain the canonical feed configuration path."
         required: false
         default: "config/free_rss_feeds.csv"
         type: string
       aliases_path:
-        description: "Symbol alias CSV path."
+        description: "Must remain the canonical alias configuration path."
         required: false
         default: "config/core_us_equity_aliases.csv"
         type: string
       watchlist_path:
-        description: "Watchlist CSV path."
+        description: "Must remain the canonical watchlist path."
         required: false
         default: "data/live/political_watchlist.csv"
         type: string
@@ -28,9 +28,7 @@ on:
         required: false
         default: "false"
         type: choice
-        options:
-          - "false"
-          - "true"
+        options: ["false", "true"]
   schedule:
     - cron: "15 12 * * 6"
 
@@ -42,74 +40,112 @@ concurrency:
   cancel-in-progress: false
 
 jobs:
+  validate-workflow-boundary:
+    if: >-
+      github.repository == 'QuantStrategyLab/PoliticalEventTrackingResearch' &&
+      github.ref == 'refs/heads/main' &&
+      github.workflow_ref == 'QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Validate canonical workflow inputs
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+          FEEDS_PATH: ${{ github.event.inputs.feeds_path || 'config/free_rss_feeds.csv' }}
+          ALIASES_PATH: ${{ github.event.inputs.aliases_path || 'config/core_us_equity_aliases.csv' }}
+          WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }}
+        run: |
+          set -euo pipefail
+          python3 - <<'PY'
+          import os
+          expected = (
+              "config/free_rss_feeds.csv",
+              "config/core_us_equity_aliases.csv",
+              "data/live/political_watchlist.csv",
+          )
+          actual = tuple(os.environ[key] for key in ("FEEDS_PATH", "ALIASES_PATH", "WATCHLIST_PATH"))
+          if actual != expected:
+              raise SystemExit("workflow_input_path_invalid")
+          PY
+
   build-rss-source-events:
+    needs: validate-workflow-boundary
+    if: needs.validate-workflow-boundary.result == 'success'
     runs-on: ubuntu-latest
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
+        with:
+          repository: QuantStrategyLab/PoliticalEventTrackingResearch
+          ref: ${{ github.sha }}
+          persist-credentials: false
+      - name: Verify reviewed main checkout
+        env:
+          EXPECTED_SHA: ${{ github.sha }}
+        run: test "$(git rev-parse HEAD)" = "${EXPECTED_SHA}"
       - uses: actions/setup-python@v6
         with:
           python-version: "3.11"
       - name: Install package
         run: python -m pip install -e .
       - name: Fetch RSS sources and extract mentions
         env:
-          FEEDS_PATH: ${{ github.event.inputs.feeds_path || 'config/free_rss_feeds.csv' }}
-          ALIASES_PATH: ${{ github.event.inputs.aliases_path || 'config/core_us_equity_aliases.csv' }}
-          WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }}
           MAX_ITEMS_PER_FEED: ${{ github.event.inputs.max_items_per_feed || '50' }}
         run: |
           set -euo pipefail
           mkdir -p data/output/rss_source_pipeline
+          fetch_exit=0
           python scripts/fetch_rss_sources.py \
-            --feeds "${FEEDS_PATH}" \
+            --feeds config/free_rss_feeds.csv \
             --output data/output/rss_source_pipeline/source_items.csv \
             --max-items-per-feed "${MAX_ITEMS_PER_FEED}" \
             --continue-on-feed-error \
-            --status-output data/output/rss_source_pipeline/source_fetch_status.json
-          python scripts/extract_source_mentions.py \
-            --raw-items data/output/rss_source_pipeline/source_items.csv \
-            --aliases "${ALIASES_PATH}" \
-            --output data/output/rss_source_pipeline/source_events.csv
-          python scripts/build_tracker.py \
-            --watchlist "${WATCHLIST_PATH}" \
-            --events data/output/rss_source_pipeline/source_events.csv \
-            --output data/output/rss_source_pipeline/source_tracker.csv
+            --status-output data/output/rss_source_pipeline/source_fetch_status.json || fetch_exit=$?
+          if [ -f data/output/rss_source_pipeline/source_items.csv ]; then
+            python scripts/extract_source_mentions.py \
+              --raw-items data/output/rss_source_pipeline/source_items.csv \
+              --aliases config/core_us_equity_aliases.csv \
+              --output data/output/rss_source_pipeline/source_events.csv
+            python scripts/build_tracker.py \
+              --watchlist data/live/political_watchlist.csv \
+              --events data/output/rss_source_pipeline/source_events.csv \
+              --output data/output/rss_source_pipeline/source_tracker.csv
+          fi
+          printf 'fetch_exit=%s\n' "${fetch_exit}"
+      - name: Upload RSS source artifact
+        uses: actions/upload-artifact@v7
+        with:
+          name: rss-source-pipeline
+          path: data/output/rss_source_pipeline/
+          if-no-files-found: error
+      - name: Validate feed completeness
+        run: python scripts/validate_fetch_status.py --status data/output/rss_source_pipeline/source_fetch_status.json
       - name: Publish live CSV outputs to repository
         env:
           COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }}
+          GITHUB_TOKEN: ${{ github.token }}
+          EXPECTED_SHA: ${{ github.sha }}
         run: |
           set -euo pipefail
-          if [ "${COMMIT_OUTPUTS}" != "true" ]; then
-            echo "Live output commit disabled."
-            exit 0
-          fi
+          if [ "${COMMIT_OUTPUTS}" != "true" ]; then echo "Live output commit disabled."; exit 0; fi
+          test "$(git rev-parse HEAD)" = "${EXPECTED_SHA}"
           mkdir -p data/live
           cp data/output/rss_source_pipeline/source_items.csv data/live/source_items.csv
           cp data/output/rss_source_pipeline/source_events.csv data/live/source_events.csv
           cp data/output/rss_source_pipeline/source_events.csv data/live/political_events.csv
           cp data/output/rss_source_pipeline/source_tracker.csv data/live/source_tracker.csv
           cp data/output/rss_source_pipeline/source_fetch_status.json data/live/source_fetch_status.json
-          python scripts/write_live_manifest.py \
-            --base-dir . \
-            --output data/live/source_manifest.json \
-            data/live/source_fetch_status.json \
-            data/live/source_items.csv \
-            data/live/source_events.csv \
-            data/live/political_events.csv \
-            data/live/source_tracker.csv
+          python scripts/write_live_manifest.py --base-dir . --output data/live/source_manifest.json data/live/source_fetch_status.json data/live/source_items.csv data/live/source_events.csv data/live/political_events.csv data/live/source_tracker.csv
           git config user.name "github-actions[bot]"
           git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
           git add data/live/source_items.csv data/live/source_events.csv data/live/political_events.csv data/live/source_tracker.csv data/live/source_fetch_status.json data/live/source_manifest.json
           if git diff --cached --quiet; then
             echo "No live RSS output changes to commit."
           else
             git commit -m "Update live RSS source events [skip ci]"
-            git push
+            git config --local http.https://github.com/.extraheader "AUTHORIZATION: bearer ${GITHUB_TOKEN}"
+            cleanup_git_auth() { git config --local --unset-all http.https://github.com/.extraheader >/dev/null 2>&1 || true; }
+            trap cleanup_git_auth EXIT
+            git push origin HEAD:refs/heads/main
+            cleanup_git_auth
+            trap - EXIT
           fi
-      - name: Upload RSS source artifact
-        uses: actions/upload-artifact@v7
-        with:
-          name: rss-source-pipeline
-          path: data/output/rss_source_pipeline/
-          if-no-files-found: error
diff --git a/scripts/validate_fetch_status.py b/scripts/validate_fetch_status.py
@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Fail closed unless the existing RSS fetch status is complete."""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from political_event_tracking_research.rss_source_fetch import FetchStatusError, validate_fetch_status  # noqa: E402
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--status", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        complete = validate_fetch_status(json.loads(args.status.read_text(encoding="utf-8")))
+    except (FetchStatusError, OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError):
+        raise SystemExit("fetch_status_invalid") from None
+    if not complete:
+        raise SystemExit("fetch_incomplete")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py
@@ -48,6 +48,35 @@ def to_json(self) -> dict[str, object]:
         }
 
 
+class FetchStatusError(ValueError):
+    def __init__(self, code: str) -> None:
+        self.code = code
+        super().__init__(code)
+
+
+def validate_fetch_status(payload: object) -> bool:
+    required = {"generated_at", "feed_count", "successful_feed_count", "failed_feed_count", "item_count", "feeds"}
+    entry_keys = {"feed_id", "feed_url", "ok", "item_count", "error"}
+    if not isinstance(payload, dict) or set(payload) != required or not isinstance(payload["feeds"], list):
+        raise FetchStatusError("fetch_status_shape_invalid")
+    counters = ("feed_count", "successful_feed_count", "failed_feed_count", "item_count")
+    if type(payload["generated_at"]) is not str or any(type(payload[key]) is not int or payload[key] < 0 for key in counters):
+        raise FetchStatusError("fetch_status_counter_invalid")
+    feeds = payload["feeds"]
+    if len(feeds) != payload["feed_count"]:
+        raise FetchStatusError("fetch_status_counter_invalid")
+    successful = 0
+    total_items = 0
+    for feed in feeds:
+        if not isinstance(feed, dict) or set(feed) != entry_keys or type(feed["feed_id"]) is not str or type(feed["feed_url"]) is not str or type(feed["ok"]) is not bool or type(feed["item_count"]) is not int or feed["item_count"] < 0 or type(feed["error"]) is not str:
+            raise FetchStatusError("fetch_status_entry_invalid")
+        successful += int(feed["ok"])
+        total_items += feed["item_count"]
+    if successful != payload["successful_feed_count"] or payload["feed_count"] - successful != payload["failed_feed_count"] or total_items != payload["item_count"]:
+        raise FetchStatusError("fetch_status_counter_invalid")
+    return payload["failed_feed_count"] == 0
+
+
 def load_feed_config(path: str | Path) -> list[FeedConfig]:
     feeds: list[FeedConfig] = []
     for row in read_csv_rows(path):
diff --git a/src/political_event_tracking_research/workflow_boundary.py b/src/political_event_tracking_research/workflow_boundary.py
@@ -0,0 +1,25 @@
+"""Pure guards for the existing RSS workflow boundary."""
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+
+CANONICAL_FEEDS = "config/free_rss_feeds.csv"
+CANONICAL_ALIASES = "config/core_us_equity_aliases.csv"
+CANONICAL_WATCHLIST = "data/live/political_watchlist.csv"
+
+
+class PathBoundaryError(ValueError):
+    def __init__(self, code: str) -> None:
+        self.code = code
+        super().__init__(code)
+
+
+def validate_source_paths(feeds_path: object, aliases_path: object, watchlist_path: object) -> None:
+    expected = (CANONICAL_FEEDS, CANONICAL_ALIASES, CANONICAL_WATCHLIST)
+    actual = (feeds_path, aliases_path, watchlist_path)
+    if any(type(value) is not str or value != expected_value for value, expected_value in zip(actual, expected, strict=True)):
+        raise PathBoundaryError("workflow_input_path_invalid")
+    for value in actual:
+        path = PurePosixPath(value)
+        if path.is_absolute() or value != str(path) or any(part in {"", ".", ".."} for part in path.parts) or "\\" in value:
+            raise PathBoundaryError("workflow_input_path_invalid")
diff --git a/tests/test_weekly_workflow_w0.py b/tests/test_weekly_workflow_w0.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from political_event_tracking_research.rss_source_fetch import FetchStatusError, validate_fetch_status
+from political_event_tracking_research.workflow_boundary import PathBoundaryError, validate_source_paths
+
+
+def status(**overrides: object) -> dict[str, object]:
+    payload: dict[str, object] = {
+        "generated_at": "2026-07-16T00:00:00Z",
+        "feed_count": 2,
+        "successful_feed_count": 2,
+        "failed_feed_count": 0,
+        "item_count": 1,
+        "feeds": [
+            {"feed_id": "a", "feed_url": "https://a.example/feed", "ok": True, "item_count": 1, "error": ""},
+            {"feed_id": "b", "feed_url": "https://b.example/feed", "ok": True, "item_count": 0, "error": ""},
+        ],
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_fetch_status_validation_returns_complete_state() -> None:
+    assert validate_fetch_status(status()) is True
+    assert validate_fetch_status(status(failed_feed_count=1, successful_feed_count=1, feeds=[
+        {"feed_id": "a", "feed_url": "https://a.example/feed", "ok": True, "item_count": 1, "error": ""},
+        {"feed_id": "b", "feed_url": "https://b.example/feed", "ok": False, "item_count": 0, "error": "blocked"},
+    ])) is False
+
+
+@pytest.mark.parametrize("change", [{"complete": False}, {"failed_feed_count": 1}, {"unknown": 1}, {"feeds": []}])
+def test_fetch_status_shape_or_incompleteness_fails_closed(change: dict[str, object]) -> None:
+    with pytest.raises(FetchStatusError):
+        validate_fetch_status({**status(), **change})
+
+
+@pytest.mark.parametrize("value", ["config/other.csv", "../config/free_rss_feeds.csv", "/tmp/feeds.csv", "config\\feeds.csv", ""])
+def test_manual_source_paths_are_canonical_only(value: str) -> None:
+    with pytest.raises(PathBoundaryError):
+        validate_source_paths(value, "config/core_us_equity_aliases.csv", "data/live/political_watchlist.csv")
+
+
+def test_workflow_orders_debug_upload_gate_and_live_push() -> None:
+    workflow = Path(__file__).parents[1].joinpath(".github/workflows/rss_source_pipeline.yml").read_text(encoding="utf-8")
+    assert workflow.index("Validate canonical workflow inputs") < workflow.index("actions/checkout@")
+    assert workflow.index("Upload RSS source artifact") < workflow.index("Validate feed completeness")
+    assert workflow.index("Validate feed completeness") < workflow.index("Publish live CSV outputs")
+    assert "git push origin HEAD:refs/heads/main" in workflow
+    assert "git rev-parse HEAD" in workflow
+    assert "weekly" not in workflow.lower()
