Add deterministic weekly workflow boundary

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

Author: Pigbibi

.github/workflows/rss_source_pipeline.yml

@@ -23,6 +23,16 @@ on:
         required: false
         default: "50"
         type: string
+      period_start:
+        description: "Manual completed ISO-week Monday date."
+        required: false
+        default: ""
+        type: string
+      as_of:
+        description: "Manual completed ISO-week Sunday date."
+        required: false
+        default: ""
+        type: string
       commit_outputs:
         description: "Commit generated live CSV outputs back to data/live."
         required: false
@@ -32,9 +42,10 @@ on:
           - "false"
           - "true"
   schedule:
-    - cron: "15 12 * * 6"
+    - cron: "15 12 * * 1"
 
 permissions:
+  actions: read
   contents: write
 
 concurrency:
@@ -52,6 +63,43 @@ jobs:
           python-version: "3.11"
       - name: Install package
         run: python -m pip install -e .
+      - name: Validate workflow dispatch before side effects
+        env:
+          PERIOD_START: ${{ github.event.inputs.period_start || '' }}
+          AS_OF: ${{ github.event.inputs.as_of || '' }}
+          GITHUB_TOKEN: ${{ github.token }}
+        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
+          python scripts/validate_weekly_workflow.py \
+            --event "${GITHUB_EVENT_NAME}" \
+            --run-id "${GITHUB_RUN_ID}" \
+            --run-attempt "${GITHUB_RUN_ATTEMPT}" \
+            --workflow-ref "${GITHUB_WORKFLOW_REF}" \
+            --producer-ref "${GITHUB_SHA}" \
+            --period-start "${PERIOD_START}" \
+            --as-of "${AS_OF}" \
+            "${PAYLOAD_ARGS[@]}" \
+            --output "${RUNNER_TEMP}/pert-period-context.json"
+          python - <<'PY' "${RUNNER_TEMP}/pert-period-context.json" >> "${GITHUB_ENV}"
+          import json, sys
+          value = json.load(open(sys.argv[1], encoding="utf-8"))
+          for key in ("period_start", "period_end_exclusive", "as_of", "producer_ref"):
+              print(f"PERT_{key.upper()}={value[key]}")
+          PY
       - name: Fetch RSS sources and extract mentions
         env:
           FEEDS_PATH: ${{ github.event.inputs.feeds_path || 'config/free_rss_feeds.csv' }}
@@ -113,3 +161,33 @@ jobs:
           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: |
+          set -euo pipefail
+          mkdir -p data/output/political-event-weekly-v1
+          if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then
+            RUN_MODE=scheduled
+          else
+            RUN_MODE=manual
+          fi
+          python scripts/build_weekly_artifact.py \
+            --period-start "${PERT_PERIOD_START}" \
+            --as-of "${PERT_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 "${PERT_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

docs/weekly_workflow_boundary_contract.md

@@ -0,0 +1,20 @@
+# Weekly Workflow Boundary Contract
+
+The RSS workflow validates dispatch identity before fetching, writing `data/live`,
+committing, or uploading artifacts.
+
+- Manual runs require an exact completed ISO week: Monday `period_start` and
+  Sunday `as_of`.
+- Scheduled runs fetch the current run from GitHub's workflow-run REST endpoint
+  using `actions:read`. The response must match the fixed repository, workflow
+  path, `refs/heads/main`, `event=schedule`, `run_id`, `run_attempt=1`, and a
+  full producer SHA. The immutable API `created_at` date determines the previous
+  complete UTC ISO week; runner `date -u` is not period authority.
+- The existing `rss-source-pipeline` upload occurs before dedicated weekly build.
+  A weekly contract failure leaves that legacy/source artifact available while
+  still failing the workflow and preventing dedicated upload.
+- Dedicated `political-event-weekly-v1` remains the exact five-file, 30-day,
+  complete-feed, digest-bound artifact defined by `weekly_artifact.py`.
+
+No new secrets, id-token, external store, identity registry, consumer, Pages,
+publisher, or trading capability is introduced.

scripts/build_weekly_artifact.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Build the dedicated weekly artifact after dispatch validation."""
+
+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(description=__doc__)
+    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:
+        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"))
+        feed_status = json.loads(args.feed_status.read_text(encoding="utf-8"))
+        files = build_weekly_artifact(
+            period_start=period_start,
+            as_of=as_of,
+            generated_at=generated_at,
+            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=feed_status,
+            source_provenance="official_rss_source_pipeline_v1",
+            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 exc:
+        raise SystemExit(exc.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,65 @@
+#!/usr/bin/env python3
+"""Validate dispatch inputs before PERT fetch/live side effects."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from datetime import timedelta
+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
+    WORKFLOW_REF,
+    WorkflowBoundaryError,
+    validate_manual_period,
+    validate_scheduled_run,
+)
+
+_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    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("--producer-ref", required=True)
+    parser.add_argument("--period-start")
+    parser.add_argument("--as-of")
+    parser.add_argument("--run-payload", type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        if args.run_attempt != 1:
+            raise WorkflowBoundaryError("run_attempt_invalid")
+        if args.event == "schedule":
+            if args.run_payload is None:
+                raise WorkflowBoundaryError("scheduled_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,
+            }
+        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):
+                raise WorkflowBoundaryError("workflow_identity_invalid")
+            result = {"period_start": start.isoformat(), "period_end_exclusive": (start + timedelta(days=7)).isoformat(), "as_of": as_of.isoformat(), "producer_ref": args.producer_ref}
+        args.output.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8")
+    except WorkflowBoundaryError as exc:
+        raise SystemExit(exc.code) from None
+    except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError):
+        raise SystemExit("workflow_boundary_invalid") from None
+
+
+if __name__ == "__main__":
+    main()