diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7bd252b..d8f2eca 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -23,6 +23,16 @@ on: required: false default: "50" type: string + period_start: + description: "Manual completed ISO-week Monday date." + required: false + default: "" + type: string + as_of: + description: "Manual completed ISO-week Sunday date." + required: false + default: "" + type: string commit_outputs: description: "Commit generated live CSV outputs back to data/live." required: false @@ -32,9 +42,10 @@ on: - "false" - "true" schedule: - - cron: "15 12 * * 6" + - cron: "15 12 * * 1" permissions: + actions: read contents: write concurrency: @@ -43,15 +54,51 @@ concurrency: jobs: build-rss-source-events: + if: >- + github.repository == 'QuantStrategyLab/PoliticalEventTrackingResearch' && + github.ref == 'refs/heads/main' && + github.workflow_ref == 'QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main' runs-on: ubuntu-latest timeout-minutes: 30 steps: - uses: actions/checkout@v6 + with: + repository: QuantStrategyLab/PoliticalEventTrackingResearch + ref: ${{ github.sha }} + persist-credentials: false - uses: actions/setup-python@v6 with: python-version: "3.11" - name: Install package run: python -m pip install -e . + - name: Validate workflow dispatch before side effects + env: + PERIOD_START: ${{ github.event.inputs.period_start || '' }} + AS_OF: ${{ github.event.inputs.as_of || '' }} + GITHUB_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + RUN_PAYLOAD="${RUNNER_TEMP}/pert-workflow-run.json" + curl --fail --silent --show-error \ + -H "Accept: application/vnd.github+json" \ + -H "Authorization: Bearer ${GITHUB_TOKEN}" \ + "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" > "${RUN_PAYLOAD}" + python scripts/validate_weekly_workflow.py \ + --event "${GITHUB_EVENT_NAME}" \ + --run-id "${GITHUB_RUN_ID}" \ + --run-attempt "${GITHUB_RUN_ATTEMPT}" \ + --workflow-ref "${GITHUB_WORKFLOW_REF}" \ + --producer-ref "${GITHUB_SHA}" \ + --period-start "${PERIOD_START}" \ + --as-of "${AS_OF}" \ + --run-payload "${RUN_PAYLOAD}" \ + --output "${RUNNER_TEMP}/pert-period-context.json" + python - <<'PY' "${RUNNER_TEMP}/pert-period-context.json" >> "${GITHUB_ENV}" + import json, sys + value = json.load(open(sys.argv[1], encoding="utf-8")) + for key in ("period_start", "period_end_exclusive", "as_of", "producer_ref"): + print(f"PERT_{key.upper()}={value[key]}") + PY - name: Fetch RSS sources and extract mentions env: FEEDS_PATH: ${{ github.event.inputs.feeds_path || 'config/free_rss_feeds.csv' }} @@ -113,3 +160,33 @@ jobs: name: rss-source-pipeline path: data/output/rss_source_pipeline/ if-no-files-found: error + - name: Build completed weekly producer artifact + env: + WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }} + run: | + set -euo pipefail + mkdir -p data/output/political-event-weekly-v1 + if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then + RUN_MODE=scheduled + else + RUN_MODE=manual + fi + python scripts/build_weekly_artifact.py \ + --period-start "${PERT_PERIOD_START}" \ + --as-of "${PERT_AS_OF}" \ + --generated-at "$(date -u +%Y-%m-%dT%H:%M:%S.000000Z)" \ + --workflow-ref "${GITHUB_WORKFLOW_REF}" \ + --source-run-id "${GITHUB_RUN_ID}" \ + --producer-ref "${PERT_PRODUCER_REF}" \ + --source-events data/output/rss_source_pipeline/source_events.csv \ + --watchlist "${WATCHLIST_PATH}" \ + --feed-status data/output/rss_source_pipeline/source_fetch_status.json \ + --output-dir data/output/political-event-weekly-v1 \ + --run-mode "${RUN_MODE}" + - name: Upload political weekly artifact + uses: actions/upload-artifact@v7 + with: + name: political-event-weekly-v1 + path: data/output/political-event-weekly-v1/ + if-no-files-found: error + retention-days: 30 diff --git a/docs/weekly_workflow_boundary_contract.md b/docs/weekly_workflow_boundary_contract.md new file mode 100644 index 0000000..4309b2d --- /dev/null +++ b/docs/weekly_workflow_boundary_contract.md @@ -0,0 +1,20 @@ +# Weekly Workflow Boundary Contract + +The RSS workflow validates dispatch identity before fetching, writing `data/live`, +committing, or uploading artifacts. + +- Manual runs require an exact completed ISO week: Monday `period_start` and + Sunday `as_of`. +- Scheduled runs fetch the current run from GitHub's workflow-run REST endpoint + using `actions:read`. The response must match the fixed repository, workflow + path, `refs/heads/main`, `event=schedule`, `run_id`, `run_attempt=1`, and a + full producer SHA. The immutable API `created_at` date determines the previous + complete UTC ISO week; runner `date -u` is not period authority. +- The existing `rss-source-pipeline` upload occurs before dedicated weekly build. + A weekly contract failure leaves that legacy/source artifact available while + still failing the workflow and preventing dedicated upload. +- Dedicated `political-event-weekly-v1` remains the exact five-file, 30-day, + complete-feed, digest-bound artifact defined by `weekly_artifact.py`. + +No new secrets, id-token, external store, identity registry, consumer, Pages, +publisher, or trading capability is introduced. diff --git a/scripts/build_weekly_artifact.py b/scripts/build_weekly_artifact.py new file mode 100644 index 0000000..dd96290 --- /dev/null +++ b/scripts/build_weekly_artifact.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Build the dedicated weekly artifact after dispatch validation.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import date, datetime +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from political_event_tracking_research.weekly_artifact import WeeklyArtifactError, build_weekly_artifact # noqa: E402 + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--period-start", required=True) + parser.add_argument("--as-of", required=True) + parser.add_argument("--generated-at", required=True) + parser.add_argument("--workflow-ref", required=True) + parser.add_argument("--source-run-id", required=True) + parser.add_argument("--producer-ref", required=True) + parser.add_argument("--source-events", required=True, type=Path) + parser.add_argument("--watchlist", required=True, type=Path) + parser.add_argument("--feed-status", required=True, type=Path) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--run-mode", choices=("scheduled", "manual"), required=True) + args = parser.parse_args() + try: + period_start = date.fromisoformat(args.period_start) + as_of = date.fromisoformat(args.as_of) + generated_at = datetime.fromisoformat(args.generated_at.replace("Z", "+00:00")) + feed_status = json.loads(args.feed_status.read_text(encoding="utf-8")) + files = build_weekly_artifact( + period_start=period_start, + as_of=as_of, + generated_at=generated_at, + workflow_ref=args.workflow_ref, + source_run_id=args.source_run_id, + producer_ref=args.producer_ref, + source_events=args.source_events.read_bytes(), + watchlist=args.watchlist.read_bytes(), + feed_status=feed_status, + source_provenance="official_rss_source_pipeline_v1", + run_mode=args.run_mode, + ) + args.output_dir.mkdir(parents=True, exist_ok=True) + if any(args.output_dir.iterdir()): + raise WeeklyArtifactError("artifact_output_not_empty") + for name, content in files.items(): + (args.output_dir / name).write_bytes(content) + except WeeklyArtifactError as exc: + raise SystemExit(exc.code) from None + except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError, OverflowError): + raise SystemExit("weekly_artifact_input_invalid") from None + + +if __name__ == "__main__": + main() diff --git a/scripts/validate_weekly_workflow.py b/scripts/validate_weekly_workflow.py new file mode 100644 index 0000000..ad91410 --- /dev/null +++ b/scripts/validate_weekly_workflow.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +"""Validate dispatch inputs before PERT fetch/live side effects.""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from datetime import timedelta +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from political_event_tracking_research.workflow_boundary import ( # noqa: E402 + WORKFLOW_REF, + WorkflowBoundaryError, + validate_manual_run, + validate_manual_period, + validate_scheduled_run, +) + +_SHA_RE = re.compile(r"^[0-9a-f]{40}$") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--event", choices=("schedule", "workflow_dispatch"), required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--run-attempt", required=True, type=int) + parser.add_argument("--workflow-ref", required=True) + parser.add_argument("--producer-ref", required=True) + parser.add_argument("--period-start") + parser.add_argument("--as-of") + parser.add_argument("--run-payload", type=Path) + parser.add_argument("--output", required=True, type=Path) + args = parser.parse_args() + try: + if args.run_attempt != 1: + raise WorkflowBoundaryError("run_attempt_invalid") + if args.event in {"schedule", "workflow_dispatch"}: + if args.run_payload is None: + raise WorkflowBoundaryError("workflow_run_payload_missing") + 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) + result = { + "period_start": evidence.period_start.isoformat(), + "period_end_exclusive": evidence.period_end_exclusive.isoformat(), + "as_of": evidence.as_of.isoformat(), + "scheduled_created_at": evidence.created_at.isoformat(timespec="seconds").replace("+00:00", "Z"), + "producer_ref": evidence.producer_ref, + } + else: + evidence = validate_manual_run(payload, run_id=args.run_id, workflow_ref=args.workflow_ref) + start, as_of = validate_manual_period(args.period_start, args.as_of, run_created_at=evidence.created_at) + result = {"period_start": start.isoformat(), "period_end_exclusive": (start + timedelta(days=7)).isoformat(), "as_of": as_of.isoformat(), "producer_ref": evidence.producer_ref} + else: + start, as_of = validate_manual_period(args.period_start, args.as_of) + if args.workflow_ref != WORKFLOW_REF or type(args.producer_ref) is not str or not _SHA_RE.fullmatch(args.producer_ref): + raise WorkflowBoundaryError("workflow_identity_invalid") + result = {"period_start": start.isoformat(), "period_end_exclusive": (start + timedelta(days=7)).isoformat(), "as_of": as_of.isoformat(), "producer_ref": args.producer_ref} + args.output.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8") + except WorkflowBoundaryError as exc: + raise SystemExit(exc.code) from None + except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError): + raise SystemExit("workflow_boundary_invalid") from None + + +if __name__ == "__main__": + main() 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..e5d820d --- /dev/null +++ b/src/political_event_tracking_research/weekly_artifact.py @@ -0,0 +1,238 @@ +"""Concrete five-file weekly artifact for the RSS producer.""" + +from __future__ import annotations + +import csv +import hashlib +import io +import json +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, +) + +ARTIFACT_NAME: Final = "political-event-weekly-v1" +RETENTION_DAYS: Final = 30 +MANIFEST_VERSION: Final = "political_event_weekly_artifact_manifest.v1" +PERIOD_LOCK_NAME: Final = "period_lock.json" +EVENTS_NAME: Final = "political_events.csv" +WATCHLIST_NAME: Final = "political_watchlist.csv" +WEEKLY_NAME: Final = "political_event_weekly.json" +MANIFEST_NAME: Final = "weekly_manifest.json" +ARTIFACT_FILES: Final = (PERIOD_LOCK_NAME, EVENTS_NAME, WATCHLIST_NAME, WEEKLY_NAME, MANIFEST_NAME) +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") + + +class WeeklyArtifactError(ValueError): + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _invalid(code: str) -> WeeklyArtifactError: + return WeeklyArtifactError(code) + + +def _sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _snapshot_digest(events: bytes, watchlist: bytes) -> str: + digest = hashlib.sha256() + for value in (events, watchlist): + digest.update(len(value).to_bytes(8, "big")) + digest.update(value) + return digest.hexdigest() + + +def _json_bytes(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 _invalid(code) from None + + +def _parse_json(wire: bytes, code: str) -> object: + if type(wire) is not bytes: + raise _invalid(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 _invalid(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 _invalid(code) from None + if _json_bytes(value, code) != wire: + raise _invalid(f"{code}_noncanonical") + return value + + +def _csv_snapshot(value: bytes, header: tuple[str, ...], code: str, *, allow_empty: bool = False) -> tuple[int, tuple[str, ...]]: + if type(value) is not bytes: + raise _invalid(code) + try: + rows = list(csv.reader(io.StringIO(value.decode("utf-8"), newline=""))) + except (UnicodeError, csv.Error, ValueError, RecursionError): + raise _invalid(code) from None + if not rows or tuple(rows[0]) != header: + raise _invalid(f"{code}_header") + if (not allow_empty and len(rows) < 2) or any(len(row) != len(header) for row in rows[1:]): + raise _invalid(f"{code}_rows") + return len(rows) - 1, header + + +def _filter_events(value: bytes, period_start: date, as_of: date) -> bytes: + try: + rows = list(csv.reader(io.StringIO(value.decode("utf-8"), newline=""))) + except (UnicodeError, csv.Error, ValueError, RecursionError): + raise _invalid("events_csv_invalid") from None + if not rows or tuple(rows[0]) != EVENT_HEADER: + raise _invalid("events_csv_header") + selected: list[list[str]] = [] + for row in rows[1:]: + if len(row) != len(EVENT_HEADER): + raise _invalid("events_csv_rows") + try: + event_date = date.fromisoformat(row[1]) + except (TypeError, ValueError): + raise _invalid("events_date_invalid") from None + if event_date.isoformat() != row[1]: + raise _invalid("events_date_invalid") + if period_start <= event_date <= as_of: + selected.append(row) + output = io.StringIO(newline="") + writer = csv.writer(output, lineterminator="\n") + writer.writerow(EVENT_HEADER) + writer.writerows(selected) + return output.getvalue().encode("utf-8") + + +def _status(value: object) -> WeeklyFeedStatus: + if not isinstance(value, Mapping) or any(key not in value for key in ("feed_count", "successful_feed_count", "failed_feed_count", "complete")): + raise _invalid("feed_status_invalid") + try: + return WeeklyFeedStatus( + value["feed_count"], value["successful_feed_count"], value["failed_feed_count"], + value.get("stale_feed_count", 0), value.get("missing_feed_count", 0), value["complete"], + ) + except (WeeklyContractError, TypeError, ValueError): + raise _invalid("feed_status_incomplete") from None + + +def _manifest(lock: PoliticalEventWeeklyPeriodLockV1, contract: WeeklySourceContract, files: Mapping[str, bytes]) -> dict[str, object]: + event_count, event_header = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_count, watch_header = _csv_snapshot(files[WATCHLIST_NAME], WATCHLIST_HEADER, "watchlist_csv_invalid") + return { + "manifest_version": MANIFEST_VERSION, + "artifact_name": ARTIFACT_NAME, + "retention_days": RETENTION_DAYS, + "schema_version": "1", + "contract_version": "political_event_weekly.v1", + "cadence": "weekly", + "period_start": contract.period_start.isoformat(), + "period_end_exclusive": contract.period_end_exclusive.isoformat(), + "as_of": contract.as_of.isoformat(), + "generated_at": contract.generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"), + "workflow_ref": lock.workflow_ref, + "producer_ref": lock.producer_ref, + "source_run_id": lock.source_run_id, + "source_attempt": lock.source_attempt, + "source_snapshot_id": lock.source_snapshot_id, + "source_snapshot_digest": lock.source_snapshot_digest, + "source_provenance": lock.source_provenance, + "feed_status": json.loads(serialize_weekly_contract(contract))["feed_status"], + "files": [ + {"name": PERIOD_LOCK_NAME, "role": "period_lock", "length": len(files[PERIOD_LOCK_NAME]), "sha256": _sha256(files[PERIOD_LOCK_NAME]), "row_count": None, "header": None}, + {"name": EVENTS_NAME, "role": "source_events", "length": len(files[EVENTS_NAME]), "sha256": _sha256(files[EVENTS_NAME]), "row_count": event_count, "header": list(event_header)}, + {"name": WATCHLIST_NAME, "role": "source_watchlist", "length": len(files[WATCHLIST_NAME]), "sha256": _sha256(files[WATCHLIST_NAME]), "row_count": watch_count, "header": list(watch_header)}, + {"name": WEEKLY_NAME, "role": "weekly_contract", "length": len(files[WEEKLY_NAME]), "sha256": _sha256(files[WEEKLY_NAME]), "row_count": None, "header": None}, + ], + } + + +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], source_provenance: str, 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: + raise _invalid("period_invalid") + try: + period_end = period_start + timedelta(days=7) + except OverflowError: + raise _invalid("period_invalid") from None + if as_of != period_end - timedelta(days=1): + raise _invalid("period_mismatch") + if type(generated_at) is not datetime or generated_at.tzinfo != timezone.utc or generated_at < datetime.combine(period_end, datetime.min.time(), timezone.utc): + raise _invalid("generated_at_invalid") + if source_provenance != "official_rss_source_pipeline_v1" or run_mode not in {"scheduled", "manual"}: + raise _invalid("producer_contract_invalid") + period_events = _filter_events(source_events, period_start, as_of) + source_snapshot_digest = _snapshot_digest(period_events, watchlist) + event_count, _ = _csv_snapshot(period_events, EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_count, _ = _csv_snapshot(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid") + status = _status(feed_status) + snapshot_id = f"rss_source_snapshot_{as_of:%Y%m%d}_{source_run_id}" + try: + lock = PoliticalEventWeeklyPeriodLockV1(period_start, period_end, as_of, workflow_ref, source_run_id, 1, producer_ref, snapshot_id, source_snapshot_digest, source_provenance, (SourceSnapshotArtifact(EVENTS_NAME, _sha256(period_events), event_count), SourceSnapshotArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_count))) + contract = WeeklySourceContract(as_of, period_start, period_end, generated_at, run_mode, producer_ref, source_provenance, (WeeklySourceArtifact(EVENTS_NAME, _sha256(period_events), event_count), WeeklySourceArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_count)), status) + files: dict[str, bytes] = {PERIOD_LOCK_NAME: serialize_period_lock(lock), EVENTS_NAME: period_events, WATCHLIST_NAME: watchlist, WEEKLY_NAME: serialize_weekly_contract(contract)} + except (PeriodLockError, WeeklyContractError, TypeError, ValueError, OverflowError): + raise _invalid("weekly_artifact_invalid") from None + files[MANIFEST_NAME] = _json_bytes(_manifest(lock, contract, files), "manifest_serialization_invalid") + return parse_weekly_artifact(files) + + +def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]: + if not isinstance(files, Mapping) or set(files) != set(ARTIFACT_FILES) or any(type(files[name]) is not bytes for name in ARTIFACT_FILES): + raise _invalid("artifact_file_set_invalid") + try: + lock = parse_period_lock_bytes(files[PERIOD_LOCK_NAME]) + except (PeriodLockError, TypeError, ValueError): + raise _invalid("period_lock_invalid") from None + weekly_value = _parse_json(files[WEEKLY_NAME], "weekly_wire_invalid") + if not isinstance(weekly_value, Mapping): + raise _invalid("weekly_shape_invalid") + try: + contract = parse_weekly_contract(weekly_value) + except (WeeklyContractError, TypeError, ValueError): + raise _invalid("weekly_contract_invalid") from None + if serialize_weekly_contract(contract) != files[WEEKLY_NAME]: + raise _invalid("weekly_noncanonical") + expected_events = _filter_events(files[EVENTS_NAME], contract.period_start, contract.as_of) + if expected_events != files[EVENTS_NAME]: + raise _invalid("events_period_mismatch") + event_count, _ = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_count, _ = _csv_snapshot(files[WATCHLIST_NAME], WATCHLIST_HEADER, "watchlist_csv_invalid") + expected = ((EVENTS_NAME, _sha256(files[EVENTS_NAME]), event_count), (WATCHLIST_NAME, _sha256(files[WATCHLIST_NAME]), watch_count)) + if tuple((item.path, item.sha256, item.row_count) for item in lock.source_artifacts) != expected or tuple((item.path, item.sha256, item.row_count) for item in contract.source_artifacts) != expected: + raise _invalid("source_artifact_mismatch") + expected_id = f"rss_source_snapshot_{contract.as_of:%Y%m%d}_{lock.source_run_id}" + if lock.source_snapshot_digest != _snapshot_digest(files[EVENTS_NAME], files[WATCHLIST_NAME]): + raise _invalid("source_snapshot_digest_mismatch") + if lock.source_snapshot_id != expected_id or lock.source_attempt != 1 or lock.period_start != contract.period_start or lock.period_end_exclusive != contract.period_end_exclusive or lock.as_of != contract.as_of or lock.producer_ref != contract.producer_ref or lock.source_provenance != contract.source_provenance: + raise _invalid("period_contract_mismatch") + manifest_value = _parse_json(files[MANIFEST_NAME], "manifest_wire_invalid") + if not isinstance(manifest_value, Mapping) or manifest_value != _manifest(lock, contract, files): + raise _invalid("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..41291e0 --- /dev/null +++ b/src/political_event_tracking_research/workflow_boundary.py @@ -0,0 +1,103 @@ +"""Side-effect-free validation of the PERT weekly workflow boundary.""" + +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 = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +_RUN_ID_RE = re.compile(r"^[1-9][0-9]*$") + + +class WorkflowBoundaryError(ValueError): + """Stable, sanitized dispatch-boundary error.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +@dataclass(frozen=True, slots=True) +class ScheduledRunEvidence: + period_start: date + period_end_exclusive: date + as_of: date + created_at: datetime + producer_ref: str + + +def _invalid(code: str) -> WorkflowBoundaryError: + return WorkflowBoundaryError(code) + + +def _parse_date(value: object, code: str) -> date: + if type(value) is not str or not _DATE_RE.fullmatch(value): + raise _invalid(code) + try: + parsed = date.fromisoformat(value) + except ValueError: + raise _invalid(code) from None + if parsed.isoformat() != value: + raise _invalid(code) + return parsed + + +def validate_manual_period(period_start: object, as_of: object, *, run_created_at: datetime | None = None) -> tuple[date, date]: + start = _parse_date(period_start, "manual_period_invalid") + end = _parse_date(as_of, "manual_period_invalid") + try: + expected_end = start + timedelta(days=7) + except OverflowError: + raise _invalid("manual_period_invalid") from None + if start.weekday() != 0 or end != expected_end - timedelta(days=1): + raise _invalid("manual_period_mismatch") + if run_created_at is not None and (type(run_created_at) is not datetime or run_created_at.tzinfo != timezone.utc or expected_end > run_created_at.date()): + raise _invalid("manual_period_incomplete") + return start, end + + +def _parse_created_at(value: object) -> datetime: + if type(value) is not str or not value.endswith("Z"): + raise _invalid("scheduled_run_created_at_invalid") + try: + parsed = datetime.fromisoformat(value[:-1] + "+00:00") + except ValueError: + raise _invalid("scheduled_run_created_at_invalid") from None + if parsed.tzinfo != timezone.utc: + raise _invalid("scheduled_run_created_at_invalid") + return parsed + + +def validate_workflow_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object, event: str) -> ScheduledRunEvidence: + if not isinstance(payload, Mapping) or type(run_id) is not str or not _RUN_ID_RE.fullmatch(run_id) or workflow_ref != WORKFLOW_REF: + raise _invalid("scheduled_run_identity_invalid") + if type(payload.get("id")) is not int or payload["id"] != int(run_id): + raise _invalid("scheduled_run_identity_invalid") + if type(payload.get("run_attempt")) is not int or payload["run_attempt"] != 1 or payload.get("event") != event: + raise _invalid("scheduled_run_identity_invalid") + if payload.get("path") != WORKFLOW_PATH or payload.get("head_branch") != "main": + raise _invalid("scheduled_run_identity_invalid") + repository = payload.get("head_repository") + if not isinstance(repository, Mapping) or repository.get("full_name") != REPOSITORY: + raise _invalid("scheduled_run_identity_invalid") + producer_ref = payload.get("head_sha") + if type(producer_ref) is not str or not _SHA_RE.fullmatch(producer_ref): + raise _invalid("scheduled_run_identity_invalid") + created_at = _parse_created_at(payload.get("created_at")) + current_monday = created_at.date() - timedelta(days=created_at.date().weekday()) + period_start = current_monday - timedelta(days=7) + return ScheduledRunEvidence(period_start, current_monday, current_monday - timedelta(days=1), created_at, producer_ref) + + +def validate_scheduled_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> ScheduledRunEvidence: + return validate_workflow_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="schedule") + + +def validate_manual_run(payload: Mapping[str, object], *, run_id: object, workflow_ref: object) -> ScheduledRunEvidence: + return validate_workflow_run(payload, run_id=run_id, workflow_ref=workflow_ref, event="workflow_dispatch") diff --git a/tests/test_weekly_artifact.py b/tests/test_weekly_artifact.py new file mode 100644 index 0000000..6621961 --- /dev/null +++ b/tests/test_weekly_artifact.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import json +from datetime import date, datetime, timezone + +import pytest + +from political_event_tracking_research.weekly_artifact import ( + WeeklyArtifactError, + build_weekly_artifact, + parse_weekly_artifact, +) + +EVENTS = b"event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\ne1,2026-07-08,AAPL,public_mention,neutral,high,https://example.test/e1,note\n" +WATCHLIST = b"symbol,name,bucket,research_status,thesis,source_url\nAAPL,Apple,named_mentioned,watchlist,thesis,https://example.test/aapl\n" + + +def status(**overrides: object) -> dict[str, object]: + result: dict[str, object] = {"feed_count": 2, "successful_feed_count": 2, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True} + result.update(overrides) + return result + + +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, 15, tzinfo=timezone.utc), + "workflow_ref": "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main", + "source_run_id": "12345", + "producer_ref": "a" * 40, + "source_events": EVENTS, + "watchlist": WATCHLIST, + "feed_status": status(), + "source_provenance": "official_rss_source_pipeline_v1", + "run_mode": "manual", + } + values.update(overrides) + return build_weekly_artifact(**values) + + +def test_exact_five_files_deterministic_and_readback() -> None: + first = build() + second = build() + assert tuple(first) == ("period_lock.json", "political_events.csv", "political_watchlist.csv", "political_event_weekly.json", "weekly_manifest.json") + assert first == second == parse_weekly_artifact(first) + manifest = json.loads(first["weekly_manifest.json"]) + assert manifest["artifact_name"] == "political-event-weekly-v1" + assert manifest["retention_days"] == 30 + + +def test_event_filtering_and_zero_event_are_deterministic() -> None: + source = EVENTS + b"old,2026-07-05,MSFT,public_mention,neutral,high,https://example.test/old,note\n" + b"next,2026-07-13,MSFT,public_mention,neutral,high,https://example.test/next,note\n" + filtered = build(source_events=source) + assert filtered["political_events.csv"] == EVENTS + 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") + assert next(item for item in json.loads(empty["weekly_manifest.json"])["files"] if item["name"] == "political_events.csv")["row_count"] == 0 + + +def test_malformed_event_date_and_tamper_fail_closed() -> None: + with pytest.raises(WeeklyArtifactError) as error: + build(source_events=EVENTS + b"bad,not-a-date,MSFT,public_mention,neutral,high,https://example.test/bad,note\n") + assert error.value.code == "events_date_invalid" + tampered = build() + tampered["political_events.csv"] += b"x" + with pytest.raises(WeeklyArtifactError): + parse_weekly_artifact(tampered) + + +def test_source_snapshot_digest_is_recomputed_from_artifact_bytes() -> None: + files = build() + lock = json.loads(files["period_lock.json"]) + lock["source_snapshot_digest"] = "b" * 64 + files["period_lock.json"] = json.dumps(lock, sort_keys=True, separators=(",", ":")).encode() + with pytest.raises(WeeklyArtifactError) as error: + parse_weekly_artifact(files) + assert error.value.code == "source_snapshot_digest_mismatch" + + +@pytest.mark.parametrize("feed", [status(complete=False), status(failed_feed_count=1, successful_feed_count=1), status(stale_feed_count=1), status(missing_feed_count=1)]) +def test_incomplete_feed_never_builds(feed: dict[str, object]) -> None: + with pytest.raises(WeeklyArtifactError): + build(feed_status=feed) diff --git a/tests/test_workflow_boundary.py b/tests/test_workflow_boundary.py new file mode 100644 index 0000000..94c9a6a --- /dev/null +++ b/tests/test_workflow_boundary.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from datetime import date, datetime, timezone + +import pytest + +from political_event_tracking_research.workflow_boundary import ( + WorkflowBoundaryError, + validate_manual_period, + validate_scheduled_run, +) + + +RUN = { + "id": 12345, + "run_attempt": 1, + "event": "schedule", + "path": ".github/workflows/rss_source_pipeline.yml", + "head_branch": "main", + "head_sha": "a" * 40, + "head_repository": {"full_name": "QuantStrategyLab/PoliticalEventTrackingResearch"}, + "created_at": "2026-07-20T12:15:00Z", +} + + +def test_manual_guard_is_pure_and_strict() -> None: + assert validate_manual_period("2026-07-13", "2026-07-19") == (date(2026, 7, 13), date(2026, 7, 19)) + with pytest.raises(WorkflowBoundaryError): + validate_manual_period("", "2026-07-19") + with pytest.raises(WorkflowBoundaryError): + validate_manual_period("2026-07-14", "2026-07-20") + with pytest.raises(WorkflowBoundaryError): + validate_manual_period("2026-07-20", "2026-07-26", run_created_at=datetime(2026, 7, 20, 0, 1, tzinfo=timezone.utc)) + + +def test_scheduled_run_uses_api_created_at_not_local_clock() -> None: + result = validate_scheduled_run(RUN, run_id="12345", workflow_ref="QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main") + assert result.period_start == date(2026, 7, 13) + assert result.as_of == date(2026, 7, 19) + assert result.producer_ref == "a" * 40 + + +@pytest.mark.parametrize( + "field,value", + [ + ("id", 12346), + ("run_attempt", 2), + ("event", "workflow_dispatch"), + ("path", ".github/workflows/other.yml"), + ("head_branch", "feature"), + ("head_sha", "not-sha"), + ("created_at", "not-time"), + ], +) +def test_scheduled_run_identity_tamper_fails_closed(field: str, value: object) -> None: + payload = dict(RUN) + payload[field] = value + with pytest.raises(WorkflowBoundaryError): + validate_scheduled_run(payload, run_id="12345", workflow_ref="QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main") + + +def test_scheduled_run_wrong_repository_fails_closed() -> None: + payload = {**RUN, "head_repository": {"full_name": "other/repo"}} + with pytest.raises(WorkflowBoundaryError): + validate_scheduled_run(payload, run_id="12345", workflow_ref="QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main") + + +def test_manual_run_binds_current_run_created_at() -> None: + payload = {**RUN, "event": "workflow_dispatch"} + from political_event_tracking_research.workflow_boundary import validate_manual_run + + evidence = validate_manual_run(payload, run_id="12345", workflow_ref="QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main") + assert validate_manual_period("2026-07-13", "2026-07-19", run_created_at=evidence.created_at)[0] == date(2026, 7, 13) diff --git a/tests/test_workflow_contract.py b/tests/test_workflow_contract.py new file mode 100644 index 0000000..2e169e9 --- /dev/null +++ b/tests/test_workflow_contract.py @@ -0,0 +1,33 @@ +from pathlib import Path + + +WORKFLOW = Path(__file__).parents[1] / ".github/workflows/rss_source_pipeline.yml" + + +def test_guard_and_legacy_upload_precede_weekly_build() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert text.index("Validate workflow dispatch before side effects") < text.index("Fetch RSS sources and extract mentions") + assert text.index("Upload RSS source artifact") < text.index("Build completed weekly producer artifact") + + +def test_privileged_job_has_server_side_main_gate_and_sha_checkout() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert "github.ref == 'refs/heads/main'" in text + assert "github.workflow_ref == 'QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main'" in text + assert "ref: ${{ github.sha }}" in text + assert "persist-credentials: false" in text + + +def test_schedule_uses_actions_api_and_no_period_wall_clock() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert "actions: read" in text + assert "actions/runs/${GITHUB_RUN_ID}" in text + assert "--run-payload" in text + assert "--scheduled-today" not in text + + +def test_dedicated_artifact_contract_is_fixed() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert 'name: political-event-weekly-v1' in text + assert 'retention-days: 30' in text + assert 'path: data/output/political-event-weekly-v1/' in text