Harden weekly workflow boundary

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

Author: Pigbibi

.github/workflows/rss_source_pipeline.yml

@@ -54,10 +54,18 @@ concurrency:
 
 jobs:
   build-rss-source-events:
+    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
     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"
@@ -71,19 +79,10 @@ jobs:
         run: |
           set -euo pipefail
           RUN_PAYLOAD="${RUNNER_TEMP}/pert-workflow-run.json"
-          if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then
-            curl --fail --silent --show-error \
-              -H "Accept: application/vnd.github+json" \
-              -H "Authorization: Bearer ${GITHUB_TOKEN}" \
-              "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" > "${RUN_PAYLOAD}"
-          else
-            rm -f "${RUN_PAYLOAD}"
-          fi
-          if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then
-            PAYLOAD_ARGS=(--run-payload "${RUN_PAYLOAD}")
-          else
-            PAYLOAD_ARGS=()
-          fi
+          curl --fail --silent --show-error \
+            -H "Accept: application/vnd.github+json" \
+            -H "Authorization: Bearer ${GITHUB_TOKEN}" \
+            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" > "${RUN_PAYLOAD}"
           python scripts/validate_weekly_workflow.py \
             --event "${GITHUB_EVENT_NAME}" \
             --run-id "${GITHUB_RUN_ID}" \
@@ -92,7 +91,7 @@ jobs:
             --producer-ref "${GITHUB_SHA}" \
             --period-start "${PERIOD_START}" \
             --as-of "${AS_OF}" \
-            "${PAYLOAD_ARGS[@]}" \
+            --run-payload "${RUN_PAYLOAD}" \
             --output "${RUNNER_TEMP}/pert-period-context.json"
           python - <<'PY' "${RUNNER_TEMP}/pert-period-context.json" >> "${GITHUB_ENV}"
           import json, sys

scripts/validate_weekly_workflow.py

@@ -15,6 +15,7 @@
 from political_event_tracking_research.workflow_boundary import (  # noqa: E402
     WORKFLOW_REF,
     WorkflowBoundaryError,
+    validate_manual_run,
     validate_manual_period,
     validate_scheduled_run,
 )
@@ -37,18 +38,23 @@ def main() -> None:
     try:
         if args.run_attempt != 1:
             raise WorkflowBoundaryError("run_attempt_invalid")
-        if args.event == "schedule":
+        if args.event in {"schedule", "workflow_dispatch"}:
             if args.run_payload is None:
-                raise WorkflowBoundaryError("scheduled_run_payload_missing")
+                raise WorkflowBoundaryError("workflow_run_payload_missing")
             payload = json.loads(args.run_payload.read_text(encoding="utf-8"))
-            evidence = validate_scheduled_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref)
-            result = {
-                "period_start": evidence.period_start.isoformat(),
-                "period_end_exclusive": evidence.period_end_exclusive.isoformat(),
-                "as_of": evidence.as_of.isoformat(),
-                "scheduled_created_at": evidence.created_at.isoformat(timespec="seconds").replace("+00:00", "Z"),
-                "producer_ref": evidence.producer_ref,
-            }
+            if args.event == "schedule":
+                evidence = validate_scheduled_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref)
+                result = {
+                    "period_start": evidence.period_start.isoformat(),
+                    "period_end_exclusive": evidence.period_end_exclusive.isoformat(),
+                    "as_of": evidence.as_of.isoformat(),
+                    "scheduled_created_at": evidence.created_at.isoformat(timespec="seconds").replace("+00:00", "Z"),
+                    "producer_ref": evidence.producer_ref,
+                }
+            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)
+                result = {"period_start": start.isoformat(), "period_end_exclusive": (start + timedelta(days=7)).isoformat(), "as_of": as_of.isoformat(), "producer_ref": evidence.producer_ref}
         else:
             start, as_of = validate_manual_period(args.period_start, args.as_of)
             if args.workflow_ref != WORKFLOW_REF or type(args.producer_ref) is not str or not _SHA_RE.fullmatch(args.producer_ref):

src/political_event_tracking_research/weekly_artifact.py

@@ -187,14 +187,14 @@ def build_weekly_artifact(*, period_start: date, as_of: date, generated_at: date
         raise _invalid("generated_at_invalid")
     if source_provenance != "official_rss_source_pipeline_v1" or run_mode not in {"scheduled", "manual"}:
         raise _invalid("producer_contract_invalid")
-    raw_digest = _snapshot_digest(source_events, watchlist)
     period_events = _filter_events(source_events, period_start, as_of)
+    source_snapshot_digest = _snapshot_digest(period_events, watchlist)
     event_count, _ = _csv_snapshot(period_events, EVENT_HEADER, "events_csv_invalid", allow_empty=True)
     watch_count, _ = _csv_snapshot(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid")
     status = _status(feed_status)
     snapshot_id = f"rss_source_snapshot_{as_of:%Y%m%d}_{source_run_id}"
     try:
-        lock = PoliticalEventWeeklyPeriodLockV1(period_start, period_end, as_of, workflow_ref, source_run_id, 1, producer_ref, snapshot_id, raw_digest, source_provenance, (SourceSnapshotArtifact(EVENTS_NAME, _sha256(period_events), event_count), SourceSnapshotArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_count)))
+        lock = PoliticalEventWeeklyPeriodLockV1(period_start, period_end, as_of, workflow_ref, source_run_id, 1, producer_ref, snapshot_id, source_snapshot_digest, source_provenance, (SourceSnapshotArtifact(EVENTS_NAME, _sha256(period_events), event_count), SourceSnapshotArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_count)))
         contract = WeeklySourceContract(as_of, period_start, period_end, generated_at, run_mode, producer_ref, source_provenance, (WeeklySourceArtifact(EVENTS_NAME, _sha256(period_events), event_count), WeeklySourceArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_count)), status)
         files: dict[str, bytes] = {PERIOD_LOCK_NAME: serialize_period_lock(lock), EVENTS_NAME: period_events, WATCHLIST_NAME: watchlist, WEEKLY_NAME: serialize_weekly_contract(contract)}
     except (PeriodLockError, WeeklyContractError, TypeError, ValueError, OverflowError):
@@ -228,6 +228,8 @@ def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]:
     if tuple((item.path, item.sha256, item.row_count) for item in lock.source_artifacts) != expected or tuple((item.path, item.sha256, item.row_count) for item in contract.source_artifacts) != expected:
         raise _invalid("source_artifact_mismatch")
     expected_id = f"rss_source_snapshot_{contract.as_of:%Y%m%d}_{lock.source_run_id}"
+    if lock.source_snapshot_digest != _snapshot_digest(files[EVENTS_NAME], files[WATCHLIST_NAME]):
+        raise _invalid("source_snapshot_digest_mismatch")
     if lock.source_snapshot_id != expected_id or lock.source_attempt != 1 or lock.period_start != contract.period_start or lock.period_end_exclusive != contract.period_end_exclusive or lock.as_of != contract.as_of or lock.producer_ref != contract.producer_ref or lock.source_provenance != contract.source_provenance:
         raise _invalid("period_contract_mismatch")
     manifest_value = _parse_json(files[MANIFEST_NAME], "manifest_wire_invalid")

src/political_event_tracking_research/workflow_boundary.py

@@ -48,7 +48,7 @@ def _parse_date(value: object, code: str) -> date:
     return parsed
 
 
-def validate_manual_period(period_start: object, as_of: object) -> tuple[date, date]:
+def validate_manual_period(period_start: object, as_of: object, *, run_created_at: datetime | None = None) -> tuple[date, date]:
     start = _parse_date(period_start, "manual_period_invalid")
     end = _parse_date(as_of, "manual_period_invalid")
     try:
@@ -57,6 +57,8 @@ def validate_manual_period(period_start: object, as_of: object) -> tuple[date, d
         raise _invalid("manual_period_invalid") from None
     if start.weekday() != 0 or end != expected_end - timedelta(days=1):
         raise _invalid("manual_period_mismatch")
+    if run_created_at is not None and (type(run_created_at) is not datetime or run_created_at.tzinfo != timezone.utc or expected_end > run_created_at.date()):
+        raise _invalid("manual_period_incomplete")
     return start, end
 
 
@@ -72,12 +74,12 @@ def _parse_created_at(value: object) -> datetime:
     return parsed
 
 
-def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> ScheduledRunEvidence:
+def validate_workflow_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, event: str) -> ScheduledRunEvidence:
     if not isinstance(payload, Mapping) or type(run_id) is not str or not _RUN_ID_RE.fullmatch(run_id) or workflow_ref != WORKFLOW_REF:
         raise _invalid("scheduled_run_identity_invalid")
     if type(payload.get("id")) is not int or payload["id"] != int(run_id):
         raise _invalid("scheduled_run_identity_invalid")
-    if type(payload.get("run_attempt")) is not int or payload["run_attempt"] != 1 or payload.get("event") != "schedule":
+    if type(payload.get("run_attempt")) is not int or payload["run_attempt"] != 1 or payload.get("event") != event:
         raise _invalid("scheduled_run_identity_invalid")
     if payload.get("path") != WORKFLOW_PATH or payload.get("head_branch") != "main":
         raise _invalid("scheduled_run_identity_invalid")
@@ -91,3 +93,11 @@ def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, wor
     current_monday = created_at.date() - timedelta(days=created_at.date().weekday())
     period_start = current_monday - timedelta(days=7)
     return ScheduledRunEvidence(period_start, current_monday, current_monday - timedelta(days=1), created_at, producer_ref)
+
+
+def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> ScheduledRunEvidence:
+    return validate_workflow_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="schedule")
+
+
+def validate_manual_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> ScheduledRunEvidence:
+    return validate_workflow_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="workflow_dispatch")

tests/test_weekly_artifact.py

@@ -68,6 +68,16 @@ def test_malformed_event_date_and_tamper_fail_closed() -> None:
         parse_weekly_artifact(tampered)
 
 
+def test_source_snapshot_digest_is_recomputed_from_artifact_bytes() -> None:
+    files = build()
+    lock = json.loads(files["period_lock.json"])
+    lock["source_snapshot_digest"] = "b" * 64
+    files["period_lock.json"] = json.dumps(lock, sort_keys=True, separators=(",", ":")).encode()
+    with pytest.raises(WeeklyArtifactError) as error:
+        parse_weekly_artifact(files)
+    assert error.value.code == "source_snapshot_digest_mismatch"
+
+
 @pytest.mark.parametrize("feed", [status(complete=False), status(failed_feed_count=1, successful_feed_count=1), status(stale_feed_count=1), status(missing_feed_count=1)])
 def test_incomplete_feed_never_builds(feed: dict[str, object]) -> None:
     with pytest.raises(WeeklyArtifactError):

tests/test_workflow_boundary.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from datetime import date
+from datetime import date, datetime, timezone
 
 import pytest
 
@@ -29,6 +29,8 @@ def test_manual_guard_is_pure_and_strict() -> None:
         validate_manual_period("", "2026-07-19")
     with pytest.raises(WorkflowBoundaryError):
         validate_manual_period("2026-07-14", "2026-07-20")
+    with pytest.raises(WorkflowBoundaryError):
+        validate_manual_period("2026-07-20", "2026-07-26", run_created_at=datetime(2026, 7, 20, 0, 1, tzinfo=timezone.utc))
 
 
 def test_scheduled_run_uses_api_created_at_not_local_clock() -> None:
@@ -61,3 +63,11 @@ def test_scheduled_run_wrong_repository_fails_closed() -> None:
     payload = {**RUN, "head_repository": {"full_name": "other/repo"}}
     with pytest.raises(WorkflowBoundaryError):
         validate_scheduled_run(payload, run_id="12345", workflow_ref="QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main")
+
+
+def test_manual_run_binds_current_run_created_at() -> None:
+    payload = {**RUN, "event": "workflow_dispatch"}
+    from political_event_tracking_research.workflow_boundary import validate_manual_run
+
+    evidence = validate_manual_run(payload, run_id="12345", workflow_ref="QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main")
+    assert validate_manual_period("2026-07-13", "2026-07-19", run_created_at=evidence.created_at)[0] == date(2026, 7, 13)

tests/test_workflow_contract.py

@@ -10,6 +10,14 @@ def test_guard_and_legacy_upload_precede_weekly_build() -> None:
     assert text.index("Upload RSS source artifact") < text.index("Build completed weekly producer artifact")
 
 
+def test_privileged_job_has_server_side_main_gate_and_sha_checkout() -> None:
+    text = WORKFLOW.read_text(encoding="utf-8")
+    assert "github.ref == 'refs/heads/main'" in text
+    assert "github.workflow_ref == 'QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main'" in text
+    assert "ref: ${{ github.sha }}" in text
+    assert "persist-credentials: false" in text
+
+
 def test_schedule_uses_actions_api_and_no_period_wall_clock() -> None:
     text = WORKFLOW.read_text(encoding="utf-8")
     assert "actions: read" in text