diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7bd252b..bef82ed 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,84 @@ 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 }} + source_attempt: ${{ steps.guard.outputs.source_attempt }} + 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") + 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") + 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), ("source_attempt", str(attempt))): + 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,51 +141,36 @@ 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' }} + GITHUB_TOKEN: ${{ github.token }} 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 + 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 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 @@ -113,3 +178,18 @@ 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_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}" --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: + 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..866e30c --- /dev/null +++ b/scripts/build_weekly_artifact.py @@ -0,0 +1,57 @@ +#!/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("--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) + 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, + source_attempt=args.source_attempt, + 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..e917ec5 --- /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, 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, 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, "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): + raise SystemExit("workflow_boundary_invalid") from None + + +if __name__ == "__main__": + main() diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py 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 new file mode 100644 index 0000000..02f848d --- /dev/null +++ b/src/political_event_tracking_research/weekly_artifact.py @@ -0,0 +1,262 @@ +"""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 .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" +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 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) + + 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") + + +_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") + 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["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 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: + 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": 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": 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"}, + {"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, 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: + 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(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(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") + 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 not isinstance(manifest, Mapping): + raise _fail("artifact_contract_invalid") + 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_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_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 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 new file mode 100644 index 0000000..18f6750 --- /dev/null +++ b/src/political_event_tracking_research/workflow_boundary.py @@ -0,0 +1,109 @@ +"""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 + run_attempt: int + + +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, 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(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") + 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, run_attempt) + + +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, 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 new file mode 100644 index 0000000..74c0b4c --- /dev/null +++ b/tests/test_immediate_prior_week.py @@ -0,0 +1,169 @@ +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 = { + "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": ""}, + ], +} + + +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: + 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, run_attempt=3) + 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_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"] + 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"}) + + +@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):