From 7c551e647dcf61b0b39fd5c85505faa42fc55587 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 16 Jul 2026 01:14:10 +0800 Subject: [PATCH 1/2] Enforce immediate prior week provenance Co-Authored-By: Codex --- .github/workflows/rss_source_pipeline.yml | 141 ++++++++--- scripts/build_weekly_artifact.py | 55 ++++ scripts/validate_weekly_workflow.py | 48 ++++ .../weekly_artifact.py | 237 ++++++++++++++++++ .../workflow_boundary.py | 108 ++++++++ tests/test_immediate_prior_week.py | 144 +++++++++++ 6 files changed, 696 insertions(+), 37 deletions(-) create mode 100644 scripts/build_weekly_artifact.py create mode 100644 scripts/validate_weekly_workflow.py create mode 100644 src/political_event_tracking_research/weekly_artifact.py create mode 100644 src/political_event_tracking_research/workflow_boundary.py create mode 100644 tests/test_immediate_prior_week.py diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7bd252b..42633ac 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -23,18 +23,25 @@ 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: @@ -42,11 +49,82 @@ concurrency: 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 diff --git a/scripts/build_weekly_artifact.py b/scripts/build_weekly_artifact.py new file mode 100644 index 0000000..3ad4731 --- /dev/null +++ b/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() diff --git a/scripts/validate_weekly_workflow.py b/scripts/validate_weekly_workflow.py new file mode 100644 index 0000000..568dc0c --- /dev/null +++ b/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() diff --git a/src/political_event_tracking_research/weekly_artifact.py b/src/political_event_tracking_research/weekly_artifact.py new file mode 100644 index 0000000..6fe3671 --- /dev/null +++ b/src/political_event_tracking_research/weekly_artifact.py @@ -0,0 +1,237 @@ +"""Concrete five-file weekly artifact with run-relative period provenance.""" +from __future__ import annotations + +import csv +import hashlib +import io +import json +import re +from collections.abc import Mapping +from datetime import date, datetime, timedelta, timezone +from typing import Final + +from .workflow_boundary import WORKFLOW_REF + +ARTIFACT_NAME: Final = "political-event-weekly-v1" +RETENTION_DAYS: Final = 30 +ARTIFACT_FILES: Final = ("period_lock.json", "political_events.csv", "political_watchlist.csv", "political_event_weekly.json", "weekly_manifest.json") +EVENTS = "political_events.csv" +WATCHLIST = "political_watchlist.csv" +WEEKLY = "political_event_weekly.json" +LOCK = "period_lock.json" +MANIFEST = "weekly_manifest.json" +EVENT_HEADER: Final = ("event_id", "event_date", "symbol", "event_type", "direction", "confidence", "source_url", "notes") +WATCHLIST_HEADER: Final = ("symbol", "name", "bucket", "research_status", "thesis", "source_url") +_SHA = re.compile(r"^[0-9a-f]{40}$") + + +class WeeklyArtifactError(ValueError): + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _fail(code: str) -> WeeklyArtifactError: + return WeeklyArtifactError(code) + + +def _sha(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _snapshot_digest(*values: bytes) -> str: + h = hashlib.sha256() + for value in values: + h.update(len(value).to_bytes(8, "big")) + h.update(value) + return h.hexdigest() + + +def _json(value: object, code: str) -> bytes: + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _fail(code) from None + + +def _parse_json(wire: bytes, code: str) -> object: + if type(wire) is not bytes: + raise _fail(code) + + def pairs(items: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, item in items: + if key in result: + raise _fail(f"{code}_duplicate_key") + result[key] = item + return result + + try: + value = json.loads(wire.decode("utf-8"), object_pairs_hook=pairs) + except WeeklyArtifactError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _fail(code) from None + if _json(value, code) != wire: + raise _fail(f"{code}_noncanonical") + return value + + +def _csv_rows(value: bytes, header: tuple[str, ...], code: str, *, allow_empty: bool = False) -> list[list[str]]: + if type(value) is not bytes: + raise _fail(code) + try: + rows = list(csv.reader(io.StringIO(value.decode("utf-8"), newline=""))) + except (UnicodeError, csv.Error, ValueError, RecursionError): + raise _fail(code) from None + if not rows or tuple(rows[0]) != header or (not allow_empty and len(rows) < 2): + raise _fail(f"{code}_header") + if any(len(row) != len(header) for row in rows[1:]): + raise _fail(f"{code}_rows") + return rows + + +def _filter_events(source: bytes, start: date, as_of: date) -> bytes: + rows = _csv_rows(source, EVENT_HEADER, "events_csv_invalid", allow_empty=True) + chosen: list[list[str]] = [] + for row in rows[1:]: + try: + observed = date.fromisoformat(row[1]) + except (TypeError, ValueError): + raise _fail("events_date_invalid") from None + if observed.isoformat() != row[1]: + raise _fail("events_date_invalid") + if start <= observed <= as_of: + chosen.append(row) + output = io.StringIO(newline="") + writer = csv.writer(output, lineterminator="\n") + writer.writerow(EVENT_HEADER) + writer.writerows(chosen) + return output.getvalue().encode("utf-8") + + +def _status(value: object) -> dict[str, object]: + required = {"feed_count", "successful_feed_count", "failed_feed_count"} + optional = {"stale_feed_count", "missing_feed_count", "complete"} + if not isinstance(value, Mapping) or not required.issubset(value) or set(value) - required - optional: + raise _fail("feed_status_invalid") + counts = {key: value.get(key, 0) for key in required | {"stale_feed_count", "missing_feed_count"}} + if any(type(counts[key]) is not int or counts[key] < 0 for key in counts): + raise _fail("feed_status_invalid") + complete = value.get("complete", counts["failed_feed_count"] == 0 and counts["successful_feed_count"] == counts["feed_count"]) + if type(complete) is not bool: + raise _fail("feed_status_invalid") + if counts["feed_count"] <= 0 or counts["successful_feed_count"] != counts["feed_count"] or any(counts[key] for key in ("failed_feed_count", "stale_feed_count", "missing_feed_count")) or not complete: + raise _fail("feed_status_incomplete") + return {key: counts[key] for key in sorted(counts)} | {"complete": complete} + + +def _date(value: object, code: str) -> date: + if type(value) is not str: + raise _fail(code) + try: + parsed = date.fromisoformat(value) + except ValueError: + raise _fail(code) from None + if parsed.isoformat() != value: + raise _fail(code) + return parsed + + +def _manifest(lock: dict[str, object], weekly: dict[str, object], files: Mapping[str, bytes]) -> dict[str, object]: + event_rows = _csv_rows(files[EVENTS], EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_rows = _csv_rows(files[WATCHLIST], WATCHLIST_HEADER, "watchlist_csv_invalid") + return { + "manifest_version": "political_event_weekly_artifact_manifest.v1", + "artifact_name": ARTIFACT_NAME, + "retention_days": RETENTION_DAYS, + "schema_version": "1", + "contract_version": "political_event_weekly.v1", + "cadence": "weekly", + "period_start": weekly["period_start"], + "period_end_exclusive": weekly["period_end_exclusive"], + "as_of": weekly["as_of"], + "generated_at": weekly["generated_at"], + "workflow_ref": lock["workflow_ref"], + "source_run_id": lock["source_run_id"], + "source_attempt": 1, + "producer_ref": lock["producer_ref"], + "source_snapshot_digest": lock["source_snapshot_digest"], + "source_provenance": "official_rss_source_pipeline_v1", + "feed_status": weekly["feed_status"], + "source_inputs": [ + {"name": "source_events.csv", "length": lock["source_event_length"], "sha256": lock["source_event_sha256"]}, + {"name": WATCHLIST, "length": lock["source_watchlist_length"], "sha256": lock["source_watchlist_sha256"], "row_count": len(watch_rows) - 1}, + ], + "files": [ + {"name": LOCK, "length": len(files[LOCK]), "sha256": _sha(files[LOCK]), "role": "period_lock"}, + {"name": EVENTS, "length": len(files[EVENTS]), "sha256": _sha(files[EVENTS]), "row_count": len(event_rows) - 1, "header": list(EVENT_HEADER), "role": "filtered_events"}, + {"name": WATCHLIST, "length": len(files[WATCHLIST]), "sha256": _sha(files[WATCHLIST]), "row_count": len(watch_rows) - 1, "header": list(WATCHLIST_HEADER), "role": "watchlist"}, + {"name": WEEKLY, "length": len(files[WEEKLY]), "sha256": _sha(files[WEEKLY]), "role": "weekly_contract"}, + ], + } + + +def build_weekly_artifact(*, period_start: date, as_of: date, generated_at: datetime, workflow_ref: str, source_run_id: str, producer_ref: str, source_events: bytes, watchlist: bytes, feed_status: Mapping[str, object], run_mode: str) -> dict[str, bytes]: + if type(period_start) is not date or type(as_of) is not date or period_start.weekday() != 0 or workflow_ref != WORKFLOW_REF or run_mode not in {"scheduled", "manual"}: + raise _fail("period_contract_invalid") + try: + end = period_start + timedelta(days=7) + except OverflowError: + raise _fail("period_contract_invalid") from None + if as_of != end - timedelta(days=1): + raise _fail("period_mismatch") + if type(generated_at) is not datetime or generated_at.tzinfo != timezone.utc or generated_at < datetime.combine(end, datetime.min.time(), timezone.utc): + raise _fail("generated_at_invalid") + if type(source_run_id) is not str or not source_run_id.isdigit() or type(producer_ref) is not str or not _SHA.fullmatch(producer_ref): + raise _fail("source_identity_invalid") + filtered = _filter_events(source_events, period_start, as_of) + _csv_rows(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid") + status = _status(feed_status) + source_digest = _snapshot_digest(source_events, watchlist) + weekly: dict[str, object] = { + "schema_version": "1", "contract_version": "political_event_weekly.v1", "cadence": "weekly", + "period_start": period_start.isoformat(), "period_end_exclusive": end.isoformat(), "as_of": as_of.isoformat(), + "generated_at": generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"), "run_mode": run_mode, + "producer_ref": producer_ref, "source_provenance": "official_rss_source_pipeline_v1", "source_snapshot_digest": source_digest, + "feed_status": status, + } + lock: dict[str, object] = { + "lock_version": "pert.weekly.period_lock.v1", "calendar": "utc_iso_week_monday_sunday", + "period_start": period_start.isoformat(), "period_end_exclusive": end.isoformat(), "as_of": as_of.isoformat(), + "workflow_ref": workflow_ref, "source_run_id": source_run_id, "source_attempt": 1, "producer_ref": producer_ref, + "source_snapshot_digest": source_digest, "source_provenance": "official_rss_source_pipeline_v1", + "source_event_length": len(source_events), "source_event_sha256": _sha(source_events), + "source_watchlist_length": len(watchlist), "source_watchlist_sha256": _sha(watchlist), + "source_inputs": [{"name": "source_events.csv", "length": len(source_events), "sha256": _sha(source_events)}, {"name": WATCHLIST, "length": len(watchlist), "sha256": _sha(watchlist)}], + } + files: dict[str, bytes] = {LOCK: _json(lock, "period_lock_invalid"), EVENTS: filtered, WATCHLIST: watchlist, WEEKLY: _json(weekly, "weekly_contract_invalid")} + files[MANIFEST] = _json(_manifest(lock, weekly, files), "manifest_invalid") + return parse_weekly_artifact(files) + + +def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]: + if not isinstance(files, Mapping) or tuple(files) != ARTIFACT_FILES or any(type(files[name]) is not bytes for name in ARTIFACT_FILES): + raise _fail("artifact_file_set_invalid") + lock = _parse_json(files[LOCK], "period_lock_invalid") + weekly = _parse_json(files[WEEKLY], "weekly_contract_invalid") + manifest = _parse_json(files[MANIFEST], "manifest_invalid") + if not isinstance(lock, Mapping) or lock.get("lock_version") != "pert.weekly.period_lock.v1" or not isinstance(weekly, Mapping) or weekly.get("contract_version") != "political_event_weekly.v1" or not isinstance(manifest, Mapping): + raise _fail("artifact_contract_invalid") + start = _date(weekly.get("period_start"), "period_invalid") + as_of = _date(weekly.get("as_of"), "period_invalid") + end = _date(weekly.get("period_end_exclusive"), "period_invalid") + if start.weekday() != 0 or end != start + timedelta(days=7) or as_of != end - timedelta(days=1): + raise _fail("period_mismatch") + if lock.get("period_start") != weekly.get("period_start") or lock.get("as_of") != weekly.get("as_of") or lock.get("workflow_ref") != WORKFLOW_REF: + raise _fail("period_lock_mismatch") + filtered = _filter_events(files[EVENTS], start, as_of) + if filtered != files[EVENTS]: + raise _fail("events_period_mismatch") + _csv_rows(files[WATCHLIST], WATCHLIST_HEADER, "watchlist_csv_invalid") + if lock.get("source_snapshot_digest") != weekly.get("source_snapshot_digest"): + raise _fail("source_snapshot_mismatch") + expected_manifest = _manifest(dict(lock), dict(weekly), files) + if dict(manifest) != expected_manifest: + raise _fail("manifest_mismatch") + return {name: files[name] for name in ARTIFACT_FILES} diff --git a/src/political_event_tracking_research/workflow_boundary.py b/src/political_event_tracking_research/workflow_boundary.py new file mode 100644 index 0000000..6501fc1 --- /dev/null +++ b/src/political_event_tracking_research/workflow_boundary.py @@ -0,0 +1,108 @@ +"""Pure workflow boundary and immediate-prior-week contract.""" +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone + +REPOSITORY = "QuantStrategyLab/PoliticalEventTrackingResearch" +WORKFLOW_PATH = ".github/workflows/rss_source_pipeline.yml" +WORKFLOW_REF = f"{REPOSITORY}/{WORKFLOW_PATH}@refs/heads/main" +_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_SHA = re.compile(r"^[0-9a-f]{40}$") +_RUN_ID = re.compile(r"^[1-9][0-9]*$") + + +class WorkflowBoundaryError(ValueError): + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +@dataclass(frozen=True, slots=True) +class WorkflowRunEvidence: + period_start: date + period_end_exclusive: date + as_of: date + created_at: datetime + producer_ref: str + + +def _fail(code: str) -> WorkflowBoundaryError: + return WorkflowBoundaryError(code) + + +def _date(value: object) -> date: + if type(value) is not str or not _DATE.fullmatch(value): + raise _fail("manual_period_invalid") + try: + parsed = date.fromisoformat(value) + except ValueError: + raise _fail("manual_period_invalid") from None + if parsed.isoformat() != value: + raise _fail("manual_period_invalid") + return parsed + + +def _created_at(value: object) -> datetime: + if type(value) is not str or not value.endswith("Z"): + raise _fail("workflow_created_at_invalid") + try: + parsed = datetime.fromisoformat(value[:-1] + "+00:00") + except ValueError: + raise _fail("workflow_created_at_invalid") from None + if parsed.tzinfo != timezone.utc: + raise _fail("workflow_created_at_invalid") + return parsed + + +def _previous_week(created_at: datetime) -> tuple[date, date, date]: + if type(created_at) is not datetime or created_at.tzinfo != timezone.utc: + raise _fail("workflow_created_at_invalid") + try: + current_monday = created_at.date() - timedelta(days=created_at.date().weekday()) + end = current_monday + start = end - timedelta(days=7) + except (OverflowError, ValueError): + raise _fail("period_boundary_invalid") from None + return start, end, end - timedelta(days=1) + + +def validate_manual_period(period_start: object, as_of: object, *, run_created_at: datetime) -> tuple[date, date]: + if run_created_at is None: + raise _fail("workflow_created_at_missing") + start = _date(period_start) + observed_as_of = _date(as_of) + expected_start, expected_end, expected_as_of = _previous_week(run_created_at) + if (start, observed_as_of) != (expected_start, expected_as_of): + raise _fail("manual_period_not_immediate_prior") + return start, expected_as_of + + +def _validate_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, event: str) -> WorkflowRunEvidence: + if not isinstance(payload, Mapping) or type(run_id) is not str or not _RUN_ID.fullmatch(run_id) or workflow_ref != WORKFLOW_REF: + raise _fail("workflow_identity_invalid") + if type(payload.get("id")) is not int or payload["id"] != int(run_id): + raise _fail("workflow_identity_invalid") + if type(payload.get("run_attempt")) is not int or payload["run_attempt"] != 1 or payload.get("event") != event: + raise _fail("workflow_identity_invalid") + if payload.get("path") != WORKFLOW_PATH or payload.get("head_branch") != "main": + raise _fail("workflow_identity_invalid") + repo = payload.get("head_repository") + if not isinstance(repo, Mapping) or repo.get("full_name") != REPOSITORY: + raise _fail("workflow_identity_invalid") + producer_ref = payload.get("head_sha") + if type(producer_ref) is not str or not _SHA.fullmatch(producer_ref): + raise _fail("workflow_identity_invalid") + created_at = _created_at(payload.get("created_at")) + start, end, as_of = _previous_week(created_at) + return WorkflowRunEvidence(start, end, as_of, created_at, producer_ref) + + +def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> WorkflowRunEvidence: + return _validate_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) -> WorkflowRunEvidence: + return _validate_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="workflow_dispatch") diff --git a/tests/test_immediate_prior_week.py b/tests/test_immediate_prior_week.py new file mode 100644 index 0000000..f7ef6ec --- /dev/null +++ b/tests/test_immediate_prior_week.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import json +from datetime import date, datetime, timezone +from pathlib import Path + +import pytest + +from political_event_tracking_research.weekly_artifact import ( + ARTIFACT_FILES, + WeeklyArtifactError, + build_weekly_artifact, + parse_weekly_artifact, +) +from political_event_tracking_research.workflow_boundary import ( + WORKFLOW_REF, + WorkflowBoundaryError, + validate_manual_period, + validate_manual_run, + validate_scheduled_run, +) + + +EVENTS = ( + b"event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n" + b"e1,2026-07-08,AAPL,mention,neutral,high,https://example.test/e1,note\n" +) +WATCHLIST = b"symbol,name,bucket,research_status,thesis,source_url\nAAPL,Apple,watch,active,thesis,https://example.test/aapl\n" +STATUS = { + "feed_count": 2, + "successful_feed_count": 2, + "failed_feed_count": 0, + "stale_feed_count": 0, + "missing_feed_count": 0, + "complete": True, +} + + +def run_payload(*, created_at: str = "2026-07-13T12:45:00Z", attempt: int = 1) -> dict[str, object]: + return { + "id": 123, + "run_attempt": attempt, + "event": "workflow_dispatch", + "path": ".github/workflows/rss_source_pipeline.yml", + "head_branch": "main", + "head_sha": "a" * 40, + "created_at": created_at, + "head_repository": {"full_name": "QuantStrategyLab/PoliticalEventTrackingResearch"}, + } + + +def test_schedule_and_manual_use_same_previous_complete_week() -> None: + payload = {**run_payload(), "event": "schedule"} + scheduled = validate_scheduled_run(payload, run_id="123", workflow_ref=WORKFLOW_REF) + assert (scheduled.period_start, scheduled.as_of) == (date(2026, 7, 6), date(2026, 7, 12)) + manual = validate_manual_run({**payload, "event": "workflow_dispatch"}, run_id="123", workflow_ref=WORKFLOW_REF) + assert validate_manual_period("2026-07-06", "2026-07-12", run_created_at=manual.created_at) == ( + date(2026, 7, 6), + date(2026, 7, 12), + ) + + +@pytest.mark.parametrize(("period_start", "as_of"), [("2026-06-29", "2026-07-05"), ("2026-07-13", "2026-07-19"), ("2026-07-13", "2026-07-19")]) +def test_manual_history_current_or_future_is_rejected(period_start: str, as_of: str) -> None: + with pytest.raises(WorkflowBoundaryError) as error: + validate_manual_period(period_start, as_of, run_created_at=datetime(2026, 7, 13, 12, 45, tzinfo=timezone.utc)) + assert error.value.code in {"manual_period_mismatch", "manual_period_not_immediate_prior"} + + +def test_delayed_runner_uses_created_at_not_wall_clock() -> None: + payload = {**run_payload(created_at="2026-07-14T23:59:59Z"), "event": "schedule"} + evidence = validate_scheduled_run(payload, run_id="123", workflow_ref=WORKFLOW_REF) + assert evidence.period_start == date(2026, 7, 6) + + +def test_manual_identity_mismatch_fails_before_period_validation() -> None: + payload = run_payload() + with pytest.raises(WorkflowBoundaryError): + validate_manual_run(payload, run_id="124", workflow_ref=WORKFLOW_REF) + + +def test_attempt_two_and_wrong_workflow_fail_closed() -> None: + with pytest.raises(WorkflowBoundaryError): + validate_scheduled_run({**run_payload(attempt=2), "event": "schedule"}, run_id="123", workflow_ref=WORKFLOW_REF) + with pytest.raises(WorkflowBoundaryError): + validate_scheduled_run({**run_payload(), "event": "schedule"}, run_id="123", workflow_ref="wrong") + + +def _build(**overrides: object) -> dict[str, bytes]: + values: dict[str, object] = { + "period_start": date(2026, 7, 6), + "as_of": date(2026, 7, 12), + "generated_at": datetime(2026, 7, 13, 12, 45, tzinfo=timezone.utc), + "workflow_ref": WORKFLOW_REF, + "source_run_id": "123", + "producer_ref": "a" * 40, + "source_events": EVENTS, + "watchlist": WATCHLIST, + "feed_status": STATUS, + "run_mode": "manual", + } + values.update(overrides) + return build_weekly_artifact(**values) + + +def test_artifact_is_exact_five_files_and_round_trips() -> None: + files = _build() + assert tuple(files) == ARTIFACT_FILES + assert parse_weekly_artifact(files) == files + manifest = json.loads(files["weekly_manifest.json"]) + assert manifest["period_start"] == "2026-07-06" + assert manifest["as_of"] == "2026-07-12" + + +def test_out_of_period_rows_are_filtered_but_malformed_dates_fail() -> None: + source = EVENTS + b"old,2026-07-05,MSFT,mention,neutral,high,https://example.test/old,note\n" + assert b"old," not in _build(source_events=source)["political_events.csv"] + with pytest.raises(WeeklyArtifactError): + _build(source_events=EVENTS + b"bad,not-a-date,MSFT,mention,neutral,high,x,n\n") + + +def test_zero_event_week_and_incomplete_feed_are_distinct() -> None: + empty = _build(source_events=b"event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n") + assert empty["political_events.csv"].endswith(b"\n") + with pytest.raises(WeeklyArtifactError): + _build(feed_status={**STATUS, "complete": False}) + + +def test_manifest_tamper_and_file_set_fail_closed() -> None: + files = _build() + tampered = dict(files) + tampered["political_events.csv"] += b"e2,2026-07-09,MSFT,mention,neutral,high,x,n\n" + with pytest.raises(WeeklyArtifactError): + parse_weekly_artifact(tampered) + with pytest.raises(WeeklyArtifactError): + parse_weekly_artifact({**files, "extra": b"x"}) + + +def test_workflow_guard_and_legacy_upload_precede_weekly_build() -> None: + workflow = Path(__file__).parents[1].joinpath(".github/workflows/rss_source_pipeline.yml").read_text(encoding="utf-8") + assert workflow.index("Validate run identity and period before checkout") < workflow.index("actions/checkout@") + assert workflow.index("Upload RSS source artifact") < workflow.index("Build completed weekly producer artifact") + assert "github.event.inputs.period_start || ''" in workflow + assert "manual_period_not_immediate_prior" in workflow From aee50fd8fb2b92ea48c0d0eed2b47e941087fa11 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 16 Jul 2026 01:31:08 +0800 Subject: [PATCH 2/2] Close weekly workflow provenance review findings Co-Authored-By: Codex --- .github/workflows/rss_source_pipeline.yml | 21 +++- scripts/build_weekly_artifact.py | 2 + scripts/validate_weekly_workflow.py | 6 +- .../rss_source_fetch.py | 3 + .../weekly_artifact.py | 105 +++++++++++------- .../weekly_period_lock.py | 2 +- .../workflow_boundary.py | 15 +-- tests/test_immediate_prior_week.py | 27 ++++- tests/test_weekly_period_lock.py | 6 +- 9 files changed, 130 insertions(+), 57 deletions(-) diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 42633ac..bef82ed 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -60,6 +60,7 @@ jobs: period_end_exclusive: ${{ steps.guard.outputs.period_end_exclusive }} as_of: ${{ steps.guard.outputs.as_of }} producer_ref: ${{ steps.guard.outputs.producer_ref }} + source_attempt: ${{ steps.guard.outputs.source_attempt }} steps: - name: Validate run identity and period before checkout id: guard @@ -91,7 +92,8 @@ jobs: 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: + attempt = int(os.environ["RUN_ATTEMPT"]) + if not 1 <= attempt <= 9007199254740991 or payload.get("id") != int(os.environ["RUN_ID"]) or type(payload.get("run_attempt")) is not int or payload["run_attempt"] != attempt: 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") @@ -109,7 +111,7 @@ jobs: 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)): + for key, value in (("period_start", start.isoformat()), ("period_end_exclusive", end.isoformat()), ("as_of", as_of.isoformat()), ("producer_ref", producer_ref), ("source_attempt", str(attempt))): with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as output: output.write(f"{key}={value}\n") PY @@ -145,6 +147,7 @@ jobs: - 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 }} run: | set -euo pipefail if [ "${COMMIT_OUTPUTS}" != "true" ]; then echo "Live output commit disabled."; exit 0; fi @@ -158,7 +161,17 @@ jobs: git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" 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 + 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 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 + cleanup_git_auth + trap - EXIT + fi - name: Upload RSS source artifact uses: actions/upload-artifact@v7 with: @@ -172,7 +185,7 @@ jobs: 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}" + 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}" --source-attempt "${{ needs.validate-run-boundary.outputs.source_attempt }}" --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: diff --git a/scripts/build_weekly_artifact.py b/scripts/build_weekly_artifact.py index 3ad4731..866e30c 100644 --- a/scripts/build_weekly_artifact.py +++ b/scripts/build_weekly_artifact.py @@ -20,6 +20,7 @@ def main() -> None: 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("--source-attempt", required=True, type=int) parser.add_argument("--producer-ref", required=True) parser.add_argument("--source-events", required=True, type=Path) parser.add_argument("--watchlist", required=True, type=Path) @@ -34,6 +35,7 @@ def main() -> None: generated_at=datetime.fromisoformat(args.generated_at.replace("Z", "+00:00")), workflow_ref=args.workflow_ref, source_run_id=args.source_run_id, + source_attempt=args.source_attempt, producer_ref=args.producer_ref, source_events=args.source_events.read_bytes(), watchlist=args.watchlist.read_bytes(), diff --git a/scripts/validate_weekly_workflow.py b/scripts/validate_weekly_workflow.py index 568dc0c..e917ec5 100644 --- a/scripts/validate_weekly_workflow.py +++ b/scripts/validate_weekly_workflow.py @@ -31,13 +31,13 @@ def main() -> None: 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) + evidence = validate_scheduled_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref, run_attempt=args.run_attempt) 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) + evidence = validate_manual_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref, run_attempt=args.run_attempt) 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") + 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, "source_attempt": evidence.run_attempt}, 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): diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py index 98e6b13..2811623 100644 --- a/src/political_event_tracking_research/rss_source_fetch.py +++ b/src/political_event_tracking_research/rss_source_fetch.py @@ -178,6 +178,9 @@ def write_fetch_status(path: str | Path, statuses: list[FeedFetchStatus], *, ite "feed_count": len(statuses), "successful_feed_count": sum(1 for item in statuses if item.ok), "failed_feed_count": sum(1 for item in statuses if not item.ok), + "stale_feed_count": 0, + "missing_feed_count": 0, + "complete": all(item.ok for item in statuses), "item_count": item_count, "feeds": [item.to_json() for item in statuses], } diff --git a/src/political_event_tracking_research/weekly_artifact.py b/src/political_event_tracking_research/weekly_artifact.py index 6fe3671..02f848d 100644 --- a/src/political_event_tracking_research/weekly_artifact.py +++ b/src/political_event_tracking_research/weekly_artifact.py @@ -10,6 +10,21 @@ from datetime import date, datetime, timedelta, timezone from typing import Final +from .weekly_contract import ( + WeeklyContractError, + WeeklyFeedStatus, + WeeklySourceArtifact, + WeeklySourceContract, + parse_weekly_contract, + serialize_weekly_contract, +) +from .weekly_period_lock import ( + PeriodLockError, + PoliticalEventWeeklyPeriodLockV1, + SourceSnapshotArtifact, + parse_period_lock_bytes, + serialize_period_lock, +) from .workflow_boundary import WORKFLOW_REF ARTIFACT_NAME: Final = "political-event-weekly-v1" @@ -54,6 +69,13 @@ def _json(value: object, code: str) -> bytes: raise _fail(code) from None +def serialize_weekly_artifact_manifest(value: Mapping[str, object]) -> bytes: + """Authoritative canonical serializer for the dedicated artifact manifest.""" + if not isinstance(value, Mapping): + raise _fail("manifest_invalid") + return _json(dict(value), "manifest_invalid") + + def _parse_json(wire: bytes, code: str) -> object: if type(wire) is not bytes: raise _fail(code) @@ -110,20 +132,27 @@ def _filter_events(source: bytes, start: date, as_of: date) -> bytes: return output.getvalue().encode("utf-8") -def _status(value: object) -> dict[str, object]: - required = {"feed_count", "successful_feed_count", "failed_feed_count"} - optional = {"stale_feed_count", "missing_feed_count", "complete"} - if not isinstance(value, Mapping) or not required.issubset(value) or set(value) - required - optional: +_FETCH_STATUS_KEYS = {"generated_at", "feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count", "complete", "item_count", "feeds"} +_FEED_ENTRY_KEYS = {"feed_id", "feed_url", "ok", "item_count", "error"} + + +def _status(value: object) -> WeeklyFeedStatus: + if not isinstance(value, Mapping) or set(value) != _FETCH_STATUS_KEYS or not isinstance(value["feeds"], list) or len(value["feeds"]) != value.get("feed_count"): + raise _fail("feed_status_invalid") + if type(value["generated_at"]) is not str or type(value["item_count"]) is not int or value["item_count"] < 0: raise _fail("feed_status_invalid") - counts = {key: value.get(key, 0) for key in required | {"stale_feed_count", "missing_feed_count"}} + for feed in value["feeds"]: + if not isinstance(feed, Mapping) or set(feed) != _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 _fail("feed_status_invalid") + counts = {key: value[key] for key in ("feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count")} if any(type(counts[key]) is not int or counts[key] < 0 for key in counts): raise _fail("feed_status_invalid") - complete = value.get("complete", counts["failed_feed_count"] == 0 and counts["successful_feed_count"] == counts["feed_count"]) + complete = value["complete"] if type(complete) is not bool: raise _fail("feed_status_invalid") if counts["feed_count"] <= 0 or counts["successful_feed_count"] != counts["feed_count"] or any(counts[key] for key in ("failed_feed_count", "stale_feed_count", "missing_feed_count")) or not complete: raise _fail("feed_status_incomplete") - return {key: counts[key] for key in sorted(counts)} | {"complete": complete} + return WeeklyFeedStatus(counts["feed_count"], counts["successful_feed_count"], counts["failed_feed_count"], counts["stale_feed_count"], counts["missing_feed_count"], complete) def _date(value: object, code: str) -> date: @@ -154,14 +183,14 @@ def _manifest(lock: dict[str, object], weekly: dict[str, object], files: Mapping "generated_at": weekly["generated_at"], "workflow_ref": lock["workflow_ref"], "source_run_id": lock["source_run_id"], - "source_attempt": 1, + "source_attempt": lock["source_attempt"], "producer_ref": lock["producer_ref"], "source_snapshot_digest": lock["source_snapshot_digest"], "source_provenance": "official_rss_source_pipeline_v1", "feed_status": weekly["feed_status"], "source_inputs": [ - {"name": "source_events.csv", "length": lock["source_event_length"], "sha256": lock["source_event_sha256"]}, - {"name": WATCHLIST, "length": lock["source_watchlist_length"], "sha256": lock["source_watchlist_sha256"], "row_count": len(watch_rows) - 1}, + {"name": EVENTS, "length": len(files[EVENTS]), "sha256": _sha(files[EVENTS]), "row_count": len(event_rows) - 1}, + {"name": WATCHLIST, "length": len(files[WATCHLIST]), "sha256": _sha(files[WATCHLIST]), "row_count": len(watch_rows) - 1}, ], "files": [ {"name": LOCK, "length": len(files[LOCK]), "sha256": _sha(files[LOCK]), "role": "period_lock"}, @@ -172,7 +201,7 @@ def _manifest(lock: dict[str, object], weekly: dict[str, object], files: Mapping } -def build_weekly_artifact(*, period_start: date, as_of: date, generated_at: datetime, workflow_ref: str, source_run_id: str, producer_ref: str, source_events: bytes, watchlist: bytes, feed_status: Mapping[str, object], run_mode: str) -> dict[str, bytes]: +def build_weekly_artifact(*, period_start: date, as_of: date, generated_at: datetime, workflow_ref: str, source_run_id: str, source_attempt: int = 1, producer_ref: str, source_events: bytes, watchlist: bytes, feed_status: Mapping[str, object], run_mode: str) -> dict[str, bytes]: if type(period_start) is not date or type(as_of) is not date or period_start.weekday() != 0 or workflow_ref != WORKFLOW_REF or run_mode not in {"scheduled", "manual"}: raise _fail("period_contract_invalid") try: @@ -183,55 +212,51 @@ def build_weekly_artifact(*, period_start: date, as_of: date, generated_at: date raise _fail("period_mismatch") if type(generated_at) is not datetime or generated_at.tzinfo != timezone.utc or generated_at < datetime.combine(end, datetime.min.time(), timezone.utc): raise _fail("generated_at_invalid") - if type(source_run_id) is not str or not source_run_id.isdigit() or type(producer_ref) is not str or not _SHA.fullmatch(producer_ref): + if type(source_run_id) is not str or not source_run_id.isdigit() or type(source_attempt) is not int or not 1 <= source_attempt <= 2**53 - 1 or type(producer_ref) is not str or not _SHA.fullmatch(producer_ref): raise _fail("source_identity_invalid") filtered = _filter_events(source_events, period_start, as_of) _csv_rows(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid") status = _status(feed_status) - source_digest = _snapshot_digest(source_events, watchlist) - weekly: dict[str, object] = { - "schema_version": "1", "contract_version": "political_event_weekly.v1", "cadence": "weekly", - "period_start": period_start.isoformat(), "period_end_exclusive": end.isoformat(), "as_of": as_of.isoformat(), - "generated_at": generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"), "run_mode": run_mode, - "producer_ref": producer_ref, "source_provenance": "official_rss_source_pipeline_v1", "source_snapshot_digest": source_digest, - "feed_status": status, - } - lock: dict[str, object] = { - "lock_version": "pert.weekly.period_lock.v1", "calendar": "utc_iso_week_monday_sunday", - "period_start": period_start.isoformat(), "period_end_exclusive": end.isoformat(), "as_of": as_of.isoformat(), - "workflow_ref": workflow_ref, "source_run_id": source_run_id, "source_attempt": 1, "producer_ref": producer_ref, - "source_snapshot_digest": source_digest, "source_provenance": "official_rss_source_pipeline_v1", - "source_event_length": len(source_events), "source_event_sha256": _sha(source_events), - "source_watchlist_length": len(watchlist), "source_watchlist_sha256": _sha(watchlist), - "source_inputs": [{"name": "source_events.csv", "length": len(source_events), "sha256": _sha(source_events)}, {"name": WATCHLIST, "length": len(watchlist), "sha256": _sha(watchlist)}], - } - files: dict[str, bytes] = {LOCK: _json(lock, "period_lock_invalid"), EVENTS: filtered, WATCHLIST: watchlist, WEEKLY: _json(weekly, "weekly_contract_invalid")} - files[MANIFEST] = _json(_manifest(lock, weekly, files), "manifest_invalid") + source_digest = _snapshot_digest(filtered, watchlist) + artifacts = (WeeklySourceArtifact(EVENTS, _sha(filtered), len(_csv_rows(filtered, EVENT_HEADER, "events_csv_invalid", allow_empty=True)) - 1), WeeklySourceArtifact(WATCHLIST, _sha(watchlist), len(_csv_rows(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid")) - 1)) + contract = WeeklySourceContract(as_of, period_start, end, generated_at, run_mode, producer_ref, "official_rss_source_pipeline_v1", artifacts, status) + weekly_wire = serialize_weekly_contract(contract) + weekly = json.loads(weekly_wire) + lock_object = PoliticalEventWeeklyPeriodLockV1(period_start, end, as_of, workflow_ref, source_run_id, source_attempt, producer_ref, f"rss_source_snapshot_{as_of:%Y%m%d}_{source_run_id}_{source_attempt}", source_digest, "official_rss_source_pipeline_v1", (SourceSnapshotArtifact(EVENTS, _sha(filtered), len(_csv_rows(filtered, EVENT_HEADER, "events_csv_invalid", allow_empty=True)) - 1), SourceSnapshotArtifact(WATCHLIST, _sha(watchlist), len(_csv_rows(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid")) - 1))) + lock = json.loads(serialize_period_lock(lock_object)) + files: dict[str, bytes] = {LOCK: serialize_period_lock(lock_object), EVENTS: filtered, WATCHLIST: watchlist, WEEKLY: weekly_wire} + files[MANIFEST] = serialize_weekly_artifact_manifest(_manifest(dict(lock), weekly, files)) return parse_weekly_artifact(files) def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]: if not isinstance(files, Mapping) or tuple(files) != ARTIFACT_FILES or any(type(files[name]) is not bytes for name in ARTIFACT_FILES): raise _fail("artifact_file_set_invalid") - lock = _parse_json(files[LOCK], "period_lock_invalid") + try: + lock_object = parse_period_lock_bytes(files[LOCK]) + lock = json.loads(files[LOCK]) + except (PeriodLockError, TypeError, ValueError): + raise _fail("period_lock_invalid") from None weekly = _parse_json(files[WEEKLY], "weekly_contract_invalid") manifest = _parse_json(files[MANIFEST], "manifest_invalid") - if not isinstance(lock, Mapping) or lock.get("lock_version") != "pert.weekly.period_lock.v1" or not isinstance(weekly, Mapping) or weekly.get("contract_version") != "political_event_weekly.v1" or not isinstance(manifest, Mapping): + if not isinstance(lock, Mapping) or lock.get("lock_version") != "pert.weekly.period_lock.v1" or not isinstance(weekly, Mapping) or not isinstance(manifest, Mapping): raise _fail("artifact_contract_invalid") - start = _date(weekly.get("period_start"), "period_invalid") - as_of = _date(weekly.get("as_of"), "period_invalid") - end = _date(weekly.get("period_end_exclusive"), "period_invalid") - if start.weekday() != 0 or end != start + timedelta(days=7) or as_of != end - timedelta(days=1): + try: + contract = parse_weekly_contract(weekly) + except (WeeklyContractError, TypeError, ValueError): + raise _fail("weekly_contract_invalid") from None + start, as_of, end = contract.period_start, contract.as_of, contract.period_end_exclusive + if lock_object.source_attempt < 1: raise _fail("period_mismatch") - if lock.get("period_start") != weekly.get("period_start") or lock.get("as_of") != weekly.get("as_of") or lock.get("workflow_ref") != WORKFLOW_REF: + if lock_object.period_start != start or lock_object.as_of != as_of or lock_object.workflow_ref != WORKFLOW_REF or lock_object.producer_ref != contract.producer_ref: raise _fail("period_lock_mismatch") filtered = _filter_events(files[EVENTS], start, as_of) if filtered != files[EVENTS]: raise _fail("events_period_mismatch") _csv_rows(files[WATCHLIST], WATCHLIST_HEADER, "watchlist_csv_invalid") - if lock.get("source_snapshot_digest") != weekly.get("source_snapshot_digest"): + if lock_object.source_snapshot_digest != _snapshot_digest(files[EVENTS], files[WATCHLIST]): raise _fail("source_snapshot_mismatch") expected_manifest = _manifest(dict(lock), dict(weekly), files) - if dict(manifest) != expected_manifest: + if dict(manifest) != expected_manifest or serialize_weekly_artifact_manifest(dict(manifest)) != files[MANIFEST]: raise _fail("manifest_mismatch") return {name: files[name] for name in ARTIFACT_FILES} diff --git a/src/political_event_tracking_research/weekly_period_lock.py b/src/political_event_tracking_research/weekly_period_lock.py index da2803f..34de758 100644 --- a/src/political_event_tracking_research/weekly_period_lock.py +++ b/src/political_event_tracking_research/weekly_period_lock.py @@ -128,7 +128,7 @@ def __post_init__(self) -> None: _workflow_ref(self.workflow_ref) if type(self.source_run_id) is not str or not _RUN_ID_RE.fullmatch(self.source_run_id): raise _invalid("source_run_id_invalid") - if type(self.source_attempt) is not int or self.source_attempt != 1: + if type(self.source_attempt) is not int or not 1 <= self.source_attempt <= MAX_SAFE_JSON_INTEGER: raise _invalid("source_attempt_invalid") if type(self.producer_ref) is not str or not _SHA1_RE.fullmatch(self.producer_ref): raise _invalid("producer_ref_invalid") diff --git a/src/political_event_tracking_research/workflow_boundary.py b/src/political_event_tracking_research/workflow_boundary.py index 6501fc1..18f6750 100644 --- a/src/political_event_tracking_research/workflow_boundary.py +++ b/src/political_event_tracking_research/workflow_boundary.py @@ -27,6 +27,7 @@ class WorkflowRunEvidence: as_of: date created_at: datetime producer_ref: str + run_attempt: int def _fail(code: str) -> WorkflowBoundaryError: @@ -80,12 +81,12 @@ def validate_manual_period(period_start: object, as_of: object, *, run_created_a return start, expected_as_of -def _validate_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, event: str) -> WorkflowRunEvidence: +def _validate_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, event: str, run_attempt: object = 1) -> WorkflowRunEvidence: if not isinstance(payload, Mapping) or type(run_id) is not str or not _RUN_ID.fullmatch(run_id) or workflow_ref != WORKFLOW_REF: raise _fail("workflow_identity_invalid") if type(payload.get("id")) is not int or payload["id"] != int(run_id): raise _fail("workflow_identity_invalid") - if type(payload.get("run_attempt")) is not int or payload["run_attempt"] != 1 or payload.get("event") != event: + if type(run_attempt) is not int or not 1 <= run_attempt <= 2**53 - 1 or type(payload.get("run_attempt")) is not int or payload["run_attempt"] != run_attempt or payload.get("event") != event: raise _fail("workflow_identity_invalid") if payload.get("path") != WORKFLOW_PATH or payload.get("head_branch") != "main": raise _fail("workflow_identity_invalid") @@ -97,12 +98,12 @@ def _validate_run(payload: Mapping[str, object], *, run_id: object, workflow_ref raise _fail("workflow_identity_invalid") created_at = _created_at(payload.get("created_at")) start, end, as_of = _previous_week(created_at) - return WorkflowRunEvidence(start, end, as_of, created_at, producer_ref) + return WorkflowRunEvidence(start, end, as_of, created_at, producer_ref, run_attempt) -def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> WorkflowRunEvidence: - return _validate_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="schedule") +def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, run_attempt: object = 1) -> WorkflowRunEvidence: + return _validate_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="schedule", run_attempt=run_attempt) -def validate_manual_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> WorkflowRunEvidence: - return _validate_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="workflow_dispatch") +def validate_manual_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, run_attempt: object = 1) -> WorkflowRunEvidence: + return _validate_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="workflow_dispatch", run_attempt=run_attempt) diff --git a/tests/test_immediate_prior_week.py b/tests/test_immediate_prior_week.py index f7ef6ec..74c0b4c 100644 --- a/tests/test_immediate_prior_week.py +++ b/tests/test_immediate_prior_week.py @@ -27,12 +27,18 @@ ) WATCHLIST = b"symbol,name,bucket,research_status,thesis,source_url\nAAPL,Apple,watch,active,thesis,https://example.test/aapl\n" STATUS = { + "generated_at": "2026-07-13T12:45:00Z", "feed_count": 2, "successful_feed_count": 2, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True, + "item_count": 1, + "feeds": [ + {"feed_id": "one", "feed_url": "https://example.test/one", "ok": True, "item_count": 1, "error": ""}, + {"feed_id": "two", "feed_url": "https://example.test/two", "ok": True, "item_count": 0, "error": ""}, + ], } @@ -80,8 +86,10 @@ def test_manual_identity_mismatch_fails_before_period_validation() -> None: def test_attempt_two_and_wrong_workflow_fail_closed() -> None: + evidence = validate_scheduled_run({**run_payload(attempt=2), "event": "schedule"}, run_id="123", workflow_ref=WORKFLOW_REF, run_attempt=2) + assert evidence.run_attempt == 2 with pytest.raises(WorkflowBoundaryError): - validate_scheduled_run({**run_payload(attempt=2), "event": "schedule"}, run_id="123", workflow_ref=WORKFLOW_REF) + validate_scheduled_run({**run_payload(attempt=2), "event": "schedule"}, run_id="123", workflow_ref=WORKFLOW_REF, run_attempt=3) with pytest.raises(WorkflowBoundaryError): validate_scheduled_run({**run_payload(), "event": "schedule"}, run_id="123", workflow_ref="wrong") @@ -112,6 +120,12 @@ def test_artifact_is_exact_five_files_and_round_trips() -> None: assert manifest["as_of"] == "2026-07-12" +def test_attempt_is_bound_in_lock_and_manifest() -> None: + files = _build(source_attempt=3) + assert json.loads(files["period_lock.json"])["source_attempt"] == 3 + assert json.loads(files["weekly_manifest.json"])["source_attempt"] == 3 + + def test_out_of_period_rows_are_filtered_but_malformed_dates_fail() -> None: source = EVENTS + b"old,2026-07-05,MSFT,mention,neutral,high,https://example.test/old,note\n" assert b"old," not in _build(source_events=source)["political_events.csv"] @@ -136,9 +150,20 @@ def test_manifest_tamper_and_file_set_fail_closed() -> None: parse_weekly_artifact({**files, "extra": b"x"}) +@pytest.mark.parametrize("change", [{"complete": None}, {"unexpected": 1}, {"failed_feed_count": 1}]) +def test_fetch_status_shape_is_exact_and_complete(change: dict[str, object]) -> None: + status = dict(STATUS) + status.update(change) + with pytest.raises(WeeklyArtifactError): + _build(feed_status=status) + + def test_workflow_guard_and_legacy_upload_precede_weekly_build() -> None: workflow = Path(__file__).parents[1].joinpath(".github/workflows/rss_source_pipeline.yml").read_text(encoding="utf-8") assert workflow.index("Validate run identity and period before checkout") < workflow.index("actions/checkout@") assert workflow.index("Upload RSS source artifact") < workflow.index("Build completed weekly producer artifact") + assert 'git config --local http.https://github.com/.extraheader' in workflow + assert 'git config --local --unset-all http.https://github.com/.extraheader' in workflow + assert 'echo "${GITHUB_TOKEN}"' not in workflow assert "github.event.inputs.period_start || ''" in workflow assert "manual_period_not_immediate_prior" in workflow diff --git a/tests/test_weekly_period_lock.py b/tests/test_weekly_period_lock.py index c660991..ccd3981 100644 --- a/tests/test_weekly_period_lock.py +++ b/tests/test_weekly_period_lock.py @@ -82,7 +82,6 @@ def test_direct_value_object_canonicalizes_artifacts() -> None: ("period_end_exclusive", "2026-07-14"), ("as_of", "2026-07-11"), ("calendar", "utc_calendar_day"), - ("source_attempt", 2), ("source_snapshot_digest", "not-a-digest"), ("producer_ref", "A" * 40), ], @@ -98,6 +97,11 @@ def test_source_attempt_rejects_unsafe_integer_shapes(value: object) -> None: parse_period_lock(payload(source_attempt=value)) +@pytest.mark.parametrize("value", [1, 2, 3, 2**53 - 1]) +def test_source_attempt_accepts_safe_positive_integer(value: int) -> None: + assert parse_period_lock(payload(source_attempt=value)).source_attempt == value + + @pytest.mark.parametrize("path", ["", "/data/a.csv", "./a.csv", "a/../b.csv", "a//b.csv", "a\\b.csv", "é.csv", "a\n.csv"]) def test_source_artifact_paths_are_canonical_safe_posix(path: str) -> None: with pytest.raises(PeriodLockError):