Enforce immediate prior week provenance

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

Author: Pigbibi

.github/workflows/rss_source_pipeline.yml

@@ -23,30 +23,108 @@ on:
         required: false
         default: "50"
         type: string
+      period_start:
+        description: "Must equal the immediately prior complete ISO-week Monday derived from this run."
+        required: true
+        type: string
+      as_of:
+        description: "Must equal the immediately prior complete ISO-week Sunday derived from this run."
+        required: true
+        type: string
       commit_outputs:
         description: "Commit generated live CSV outputs back to data/live."
         required: false
         default: "false"
         type: choice
-        options:
-          - "false"
-          - "true"
+        options: ["false", "true"]
   schedule:
-    - cron: "15 12 * * 6"
+    - cron: "15 12 * * 1"
 
 permissions:
+  actions: read
   contents: write
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref_name }}
   cancel-in-progress: false
 
 jobs:
+  validate-run-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
+    outputs:
+      period_start: ${{ steps.guard.outputs.period_start }}
+      period_end_exclusive: ${{ steps.guard.outputs.period_end_exclusive }}
+      as_of: ${{ steps.guard.outputs.as_of }}
+      producer_ref: ${{ steps.guard.outputs.producer_ref }}
+    steps:
+      - name: Validate run identity and period before checkout
+        id: guard
+        env:
+          GH_TOKEN: ${{ github.token }}
+          EVENT_NAME: ${{ github.event_name }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
+          WORKFLOW_REF: ${{ github.workflow_ref }}
+          PERIOD_START: ${{ github.event.inputs.period_start || '' }}
+          AS_OF: ${{ github.event.inputs.as_of || '' }}
+          API_URL: ${{ github.api_url }}
+        run: |
+          set -euo pipefail
+          payload="${RUNNER_TEMP}/pert-workflow-run.json"
+          curl --fail --silent --show-error \
+            -H "Accept: application/vnd.github+json" \
+            -H "Authorization: Bearer ${GH_TOKEN}" \
+            "${API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${RUN_ID}" > "${payload}"
+          python3 - "${payload}" <<'PY'
+          import json
+          import os
+          import sys
+          from datetime import datetime, timedelta, timezone
+
+          expected_repo = "QuantStrategyLab/PoliticalEventTrackingResearch"
+          expected_path = ".github/workflows/rss_source_pipeline.yml"
+          expected_ref = f"{expected_repo}/{expected_path}@refs/heads/main"
+          payload = json.load(open(sys.argv[1], encoding="utf-8"))
+          if os.environ["WORKFLOW_REF"] != expected_ref or os.environ["GITHUB_REF"] != "refs/heads/main":
+              raise SystemExit("workflow_identity_invalid")
+          if payload.get("id") != int(os.environ["RUN_ID"]) or type(payload.get("run_attempt")) is not int or payload["run_attempt"] != 1:
+              raise SystemExit("workflow_identity_invalid")
+          if payload.get("event") != os.environ["EVENT_NAME"] or payload.get("path") != expected_path or payload.get("head_branch") != "main":
+              raise SystemExit("workflow_identity_invalid")
+          if payload.get("head_repository", {}).get("full_name") != expected_repo:
+              raise SystemExit("workflow_identity_invalid")
+          producer_ref = payload.get("head_sha")
+          if type(producer_ref) is not str or producer_ref != os.environ["GITHUB_SHA"] or len(producer_ref) != 40 or any(char not in "0123456789abcdef" for char in producer_ref):
+              raise SystemExit("workflow_identity_invalid")
+          created = datetime.fromisoformat(payload["created_at"].replace("Z", "+00:00"))
+          if created.tzinfo != timezone.utc:
+              raise SystemExit("workflow_created_at_invalid")
+          current_monday = created.date() - timedelta(days=created.date().weekday())
+          start = current_monday - timedelta(days=7)
+          end = current_monday
+          as_of = end - timedelta(days=1)
+          if os.environ["EVENT_NAME"] == "workflow_dispatch" and (os.environ["PERIOD_START"], os.environ["AS_OF"]) != (start.isoformat(), as_of.isoformat()):
+              raise SystemExit("manual_period_not_immediate_prior")
+          for key, value in (("period_start", start.isoformat()), ("period_end_exclusive", end.isoformat()), ("as_of", as_of.isoformat()), ("producer_ref", producer_ref)):
+              with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as output:
+                  output.write(f"{key}={value}\n")
+          PY
+
   build-rss-source-events:
+    needs: validate-run-boundary
+    if: needs.validate-run-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
       - uses: actions/setup-python@v6
         with:
           python-version: "3.11"
@@ -61,55 +139,44 @@ jobs:
         run: |
           set -euo pipefail
           mkdir -p data/output/rss_source_pipeline
-          python scripts/fetch_rss_sources.py \
-            --feeds "${FEEDS_PATH}" \
-            --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
+          python scripts/fetch_rss_sources.py --feeds "${FEEDS_PATH}" --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
       - name: Publish live CSV outputs to repository
         env:
           COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }}
         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
           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
-          fi
+          git add data/live
+          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; 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
+      - name: Build completed weekly producer artifact
+        env:
+          WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }}
+          RUN_MODE: ${{ github.event_name == 'schedule' && 'scheduled' || 'manual' }}
+        run: |
+          set -euo pipefail
+          mkdir -p data/output/political-event-weekly-v1
+          python scripts/build_weekly_artifact.py --period-start "${{ needs.validate-run-boundary.outputs.period_start }}" --as-of "${{ needs.validate-run-boundary.outputs.as_of }}" --generated-at "$(date -u +%Y-%m-%dT%H:%M:%S.000000Z)" --workflow-ref "${GITHUB_WORKFLOW_REF}" --source-run-id "${GITHUB_RUN_ID}" --producer-ref "${{ needs.validate-run-boundary.outputs.producer_ref }}" --source-events data/output/rss_source_pipeline/source_events.csv --watchlist "${WATCHLIST_PATH}" --feed-status data/output/rss_source_pipeline/source_fetch_status.json --output-dir data/output/political-event-weekly-v1 --run-mode "${RUN_MODE}"
+      - name: Upload political weekly artifact
+        uses: actions/upload-artifact@v7
+        with:
+          name: political-event-weekly-v1
+          path: data/output/political-event-weekly-v1/
+          if-no-files-found: error
+          retention-days: 30

scripts/build_weekly_artifact.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Build the dedicated weekly artifact after the boundary guard."""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import date, datetime
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from political_event_tracking_research.weekly_artifact import WeeklyArtifactError, build_weekly_artifact  # noqa: E402
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--period-start", required=True)
+    parser.add_argument("--as-of", required=True)
+    parser.add_argument("--generated-at", required=True)
+    parser.add_argument("--workflow-ref", required=True)
+    parser.add_argument("--source-run-id", required=True)
+    parser.add_argument("--producer-ref", required=True)
+    parser.add_argument("--source-events", required=True, type=Path)
+    parser.add_argument("--watchlist", required=True, type=Path)
+    parser.add_argument("--feed-status", required=True, type=Path)
+    parser.add_argument("--output-dir", required=True, type=Path)
+    parser.add_argument("--run-mode", choices=("scheduled", "manual"), required=True)
+    args = parser.parse_args()
+    try:
+        files = build_weekly_artifact(
+            period_start=date.fromisoformat(args.period_start),
+            as_of=date.fromisoformat(args.as_of),
+            generated_at=datetime.fromisoformat(args.generated_at.replace("Z", "+00:00")),
+            workflow_ref=args.workflow_ref,
+            source_run_id=args.source_run_id,
+            producer_ref=args.producer_ref,
+            source_events=args.source_events.read_bytes(),
+            watchlist=args.watchlist.read_bytes(),
+            feed_status=json.loads(args.feed_status.read_text(encoding="utf-8")),
+            run_mode=args.run_mode,
+        )
+        args.output_dir.mkdir(parents=True, exist_ok=True)
+        if any(args.output_dir.iterdir()):
+            raise WeeklyArtifactError("artifact_output_not_empty")
+        for name, content in files.items():
+            (args.output_dir / name).write_bytes(content)
+    except WeeklyArtifactError as error:
+        raise SystemExit(error.code) from None
+    except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError, OverflowError):
+        raise SystemExit("weekly_artifact_input_invalid") from None
+
+
+if __name__ == "__main__":
+    main()

scripts/validate_weekly_workflow.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""Validate trusted run identity and the only permitted weekly period."""
+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.workflow_boundary import (  # noqa: E402
+    WorkflowBoundaryError,
+    validate_manual_period,
+    validate_manual_run,
+    validate_scheduled_run,
+)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--event", choices=("schedule", "workflow_dispatch"), required=True)
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--run-attempt", required=True, type=int)
+    parser.add_argument("--workflow-ref", required=True)
+    parser.add_argument("--period-start", required=False)
+    parser.add_argument("--as-of", required=False)
+    parser.add_argument("--run-payload", required=True, type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        payload = json.loads(args.run_payload.read_text(encoding="utf-8"))
+        if args.event == "schedule":
+            evidence = validate_scheduled_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref)
+            start, end, as_of = evidence.period_start, evidence.period_end_exclusive, evidence.as_of
+        else:
+            evidence = validate_manual_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref)
+            start, as_of = validate_manual_period(args.period_start, args.as_of, run_created_at=evidence.created_at)
+            end = start.fromordinal(start.toordinal() + 7)
+        args.output.write_text(json.dumps({"period_start": start.isoformat(), "period_end_exclusive": end.isoformat(), "as_of": as_of.isoformat(), "producer_ref": evidence.producer_ref}, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8")
+    except WorkflowBoundaryError as error:
+        raise SystemExit(error.code) from None
+    except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError):
+        raise SystemExit("workflow_boundary_invalid") from None
+
+
+if __name__ == "__main__":
+    main()