From 1e5d7fd0588b7da1b83cc3aa068a02de229385a7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 16:04:21 +0800 Subject: [PATCH] feat: add retry-safe weekly period lock contract Co-Authored-By: Codex --- docs/weekly_period_lock_contract.md | 22 ++ .../weekly_period_lock.py | 236 ++++++++++++++++++ tests/test_weekly_period_lock.py | 135 ++++++++++ 3 files changed, 393 insertions(+) create mode 100644 docs/weekly_period_lock_contract.md create mode 100644 src/political_event_tracking_research/weekly_period_lock.py create mode 100644 tests/test_weekly_period_lock.py diff --git a/docs/weekly_period_lock_contract.md b/docs/weekly_period_lock_contract.md new file mode 100644 index 0000000..fb8359b --- /dev/null +++ b/docs/weekly_period_lock_contract.md @@ -0,0 +1,22 @@ +# `pert.weekly.period_lock.v1` + +This pure contract freezes the producer's original weekly period and source +snapshot for retry/rerun validation. It is not an Actions artifact acquisition +implementation and does not read the wall clock. + +- `period_start` is a Monday; `period_end_exclusive` is the next Monday; + `as_of` is the preceding Sunday. +- `source_attempt` is exactly `1`, identifying the original dispatch evidence. + A retry/rerun compares its expected lock to that original value and never + derives a new period from `run_attempt` or current time. +- `producer_ref`, source snapshot id/digest/provenance, and canonical sorted + source artifact path/sha256/row-count entries are part of the lock. +- `generated_at` is intentionally absent: a later producer build may use a new + actual UTC build timestamp without changing period or source identity. +- Wire is exact/canonical JSON with duplicate-key rejection and stable sanitized + errors. Unknown, missing, noncanonical, unsafe integer, invalid UTC-week, + tampered, or source-snapshot-mismatched values fail closed. + +The next integration slice must acquire the original lock and immutable input +snapshot before invoking the producer. Missing evidence must not fall back to +wall-clock derivation. diff --git a/src/political_event_tracking_research/weekly_period_lock.py b/src/political_event_tracking_research/weekly_period_lock.py new file mode 100644 index 0000000..da2803f --- /dev/null +++ b/src/political_event_tracking_research/weekly_period_lock.py @@ -0,0 +1,236 @@ +"""Pure immutable period/input lock for retry-safe weekly production.""" + +from __future__ import annotations + +import json +import re +import unicodedata +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import date, timedelta +from pathlib import PurePosixPath + +LOCK_VERSION = "pert.weekly.period_lock.v1" +CALENDAR = "utc_iso_week_monday_sunday" +MAX_SAFE_JSON_INTEGER = 2**53 - 1 + +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +_SHA1_RE = re.compile(r"^[0-9a-f]{40}$") +_RUN_ID_RE = re.compile(r"^[1-9][0-9]*$") +_PATH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$") +_IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$") +_WORKFLOW_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/@-]*$") +_LOCK_KEYS = frozenset( + { + "lock_version", + "calendar", + "period_start", + "period_end_exclusive", + "as_of", + "workflow_ref", + "source_run_id", + "source_attempt", + "producer_ref", + "source_snapshot_id", + "source_snapshot_digest", + "source_provenance", + "source_artifacts", + } +) +_ARTIFACT_KEYS = frozenset({"path", "sha256", "row_count"}) + + +class PeriodLockError(ValueError): + """Stable, sanitized period-lock contract error.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _invalid(code: str) -> PeriodLockError: + return PeriodLockError(code) + + +def _date(value: object) -> date: + if type(value) is not str or not _DATE_RE.fullmatch(value): + raise _invalid("period_date_invalid") + try: + return date.fromisoformat(value) + except ValueError: + raise _invalid("period_date_invalid") from None + + +def _safe_int(value: object, code: str) -> int: + if type(value) is not int or not 0 <= value <= MAX_SAFE_JSON_INTEGER: + raise _invalid(code) + return value + + +def _identifier(value: object, code: str) -> str: + if type(value) is not str or not value.isascii() or unicodedata.normalize("NFC", value) != value or not _IDENTIFIER_RE.fullmatch(value): + raise _invalid(code) + return value + + +def _path(value: object) -> str: + if type(value) is not str or not value.isascii() or unicodedata.normalize("NFC", value) != value or not _PATH_RE.fullmatch(value): + raise _invalid("source_snapshot_path_invalid") + parsed = PurePosixPath(value) + if parsed.is_absolute() or value != str(parsed) or value.endswith("/") or "//" in value or any(part in {"", ".", ".."} for part in parsed.parts): + raise _invalid("source_snapshot_path_invalid") + return value + + +def _workflow_ref(value: object) -> str: + if type(value) is not str or not value.isascii() or unicodedata.normalize("NFC", value) != value or not _WORKFLOW_REF_RE.fullmatch(value): + raise _invalid("workflow_ref_invalid") + return value + + +@dataclass(frozen=True, slots=True) +class SourceSnapshotArtifact: + path: str + sha256: str + row_count: int + + def __post_init__(self) -> None: + _path(self.path) + if type(self.sha256) is not str or not _SHA256_RE.fullmatch(self.sha256): + raise _invalid("source_snapshot_artifact_invalid") + _safe_int(self.row_count, "source_snapshot_artifact_invalid") + + +@dataclass(frozen=True, slots=True) +class PoliticalEventWeeklyPeriodLockV1: + period_start: date + period_end_exclusive: date + as_of: date + workflow_ref: str + source_run_id: str + source_attempt: int + producer_ref: str + source_snapshot_id: str + source_snapshot_digest: str + source_provenance: str + source_artifacts: tuple[SourceSnapshotArtifact, ...] + + def __post_init__(self) -> None: + if type(self.period_start) is not date or type(self.period_end_exclusive) is not date or type(self.as_of) is not date: + raise _invalid("period_date_invalid") + try: + expected_end = self.period_start + timedelta(days=7) + except OverflowError: + raise _invalid("period_boundary_invalid") from None + if self.period_start.weekday() != 0 or self.period_end_exclusive != expected_end or self.as_of != expected_end - timedelta(days=1): + raise _invalid("period_mismatch") + _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: + 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") + _identifier(self.source_snapshot_id, "source_snapshot_id_invalid") + if type(self.source_snapshot_digest) is not str or not _SHA256_RE.fullmatch(self.source_snapshot_digest): + raise _invalid("source_snapshot_digest_invalid") + _identifier(self.source_provenance, "source_provenance_invalid") + if not isinstance(self.source_artifacts, tuple) or not self.source_artifacts or any( + not isinstance(item, SourceSnapshotArtifact) for item in self.source_artifacts + ): + raise _invalid("source_snapshot_artifact_invalid") + ordered = tuple(sorted(self.source_artifacts, key=lambda item: item.path)) + if len({item.path for item in ordered}) != len(ordered): + raise _invalid("source_snapshot_duplicate") + object.__setattr__(self, "source_artifacts", ordered) + + +def parse_period_lock(value: Mapping[str, object]) -> PoliticalEventWeeklyPeriodLockV1: + if not isinstance(value, Mapping) or set(value) != _LOCK_KEYS: + raise _invalid("period_lock_shape_invalid") + if value["lock_version"] != LOCK_VERSION or value["calendar"] != CALENDAR: + raise _invalid("period_lock_version_invalid") + artifacts_value = value["source_artifacts"] + if not isinstance(artifacts_value, list) or not artifacts_value: + raise _invalid("source_snapshot_artifact_invalid") + artifacts: list[SourceSnapshotArtifact] = [] + for item in artifacts_value: + if not isinstance(item, Mapping) or set(item) != _ARTIFACT_KEYS: + raise _invalid("source_snapshot_artifact_invalid") + artifacts.append(SourceSnapshotArtifact(item["path"], item["sha256"], item["row_count"])) + return PoliticalEventWeeklyPeriodLockV1( + _date(value["period_start"]), + _date(value["period_end_exclusive"]), + _date(value["as_of"]), + _workflow_ref(value["workflow_ref"]), + value["source_run_id"], + _safe_int(value["source_attempt"], "source_attempt_invalid"), + value["producer_ref"], + value["source_snapshot_id"], + value["source_snapshot_digest"], + value["source_provenance"], + tuple(artifacts), + ) + + +def _payload(lock: PoliticalEventWeeklyPeriodLockV1) -> dict[str, object]: + if not isinstance(lock, PoliticalEventWeeklyPeriodLockV1): + raise _invalid("period_lock_type_invalid") + payload: dict[str, object] = { + "lock_version": LOCK_VERSION, + "calendar": CALENDAR, + "period_start": lock.period_start.isoformat(), + "period_end_exclusive": lock.period_end_exclusive.isoformat(), + "as_of": lock.as_of.isoformat(), + "workflow_ref": lock.workflow_ref, + "source_run_id": lock.source_run_id, + "source_attempt": lock.source_attempt, + "producer_ref": lock.producer_ref, + "source_snapshot_id": lock.source_snapshot_id, + "source_snapshot_digest": lock.source_snapshot_digest, + "source_provenance": lock.source_provenance, + "source_artifacts": [ + {"path": item.path, "sha256": item.sha256, "row_count": item.row_count} for item in lock.source_artifacts + ], + } + parse_period_lock(payload) + return payload + + +def serialize_period_lock(lock: PoliticalEventWeeklyPeriodLockV1) -> bytes: + try: + return json.dumps(_payload(lock), ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _invalid("period_lock_serialization_invalid") from None + + +def parse_period_lock_bytes(wire: bytes) -> PoliticalEventWeeklyPeriodLockV1: + if type(wire) is not bytes: + raise _invalid("period_lock_wire_invalid") + + 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("period_lock_duplicate_key") + result[key] = item + return result + + try: + value = json.loads(wire.decode("utf-8"), object_pairs_hook=pairs) + except PeriodLockError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _invalid("period_lock_wire_invalid") from None + lock = parse_period_lock(value) + if serialize_period_lock(lock) != wire: + raise _invalid("period_lock_noncanonical") + return lock + + +def assert_expected_period_lock(actual: PoliticalEventWeeklyPeriodLockV1, expected: PoliticalEventWeeklyPeriodLockV1) -> None: + if not isinstance(actual, PoliticalEventWeeklyPeriodLockV1) or not isinstance(expected, PoliticalEventWeeklyPeriodLockV1): + raise _invalid("period_lock_expected_invalid") + if actual != expected: + raise _invalid("period_lock_mismatch") diff --git a/tests/test_weekly_period_lock.py b/tests/test_weekly_period_lock.py new file mode 100644 index 0000000..c660991 --- /dev/null +++ b/tests/test_weekly_period_lock.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import json +from datetime import date + +import pytest + +from political_event_tracking_research.weekly_period_lock import ( + LOCK_VERSION, + PeriodLockError, + SourceSnapshotArtifact, + PoliticalEventWeeklyPeriodLockV1, + assert_expected_period_lock, + parse_period_lock, + parse_period_lock_bytes, + serialize_period_lock, +) + + +def artifact_payload(path: str = "data/live/source_events.csv") -> dict[str, object]: + return {"path": path, "sha256": "b" * 64, "row_count": 11} + + +def payload(**overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "lock_version": LOCK_VERSION, + "calendar": "utc_iso_week_monday_sunday", + "period_start": "2026-07-06", + "period_end_exclusive": "2026-07-13", + "as_of": "2026-07-12", + "workflow_ref": "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main", + "source_run_id": "29396791050", + "source_attempt": 1, + "producer_ref": "a" * 40, + "source_snapshot_id": "rss_source_snapshot_20260712", + "source_snapshot_digest": "c" * 64, + "source_provenance": "official_political_event_tracking_research_v1", + "source_artifacts": [artifact_payload()], + } + value.update(overrides) + return value + + +def test_valid_lock_is_canonical_and_round_trips() -> None: + lock = parse_period_lock(payload()) + wire = serialize_period_lock(lock) + assert parse_period_lock_bytes(wire) == lock + assert json.loads(wire) == payload() + + +def test_source_artifacts_are_sorted_and_permutation_stable() -> None: + items = [artifact_payload("z.csv"), artifact_payload("a.csv")] + first = parse_period_lock(payload(source_artifacts=items)) + second = parse_period_lock(payload(source_artifacts=list(reversed(items)))) + assert first == second + assert serialize_period_lock(first) == serialize_period_lock(second) + assert [item.path for item in first.source_artifacts] == ["a.csv", "z.csv"] + + +def test_direct_value_object_canonicalizes_artifacts() -> None: + first = SourceSnapshotArtifact("z.csv", "a" * 64, 2) + second = SourceSnapshotArtifact("a.csv", "b" * 64, 1) + lock = PoliticalEventWeeklyPeriodLockV1( + date(2026, 7, 6), + date(2026, 7, 13), + date(2026, 7, 12), + "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main", + "29396791050", + 1, + "a" * 40, + "rss_source_snapshot_20260712", + "c" * 64, + "official_political_event_tracking_research_v1", + (first, second), + ) + assert [item.path for item in lock.source_artifacts] == ["a.csv", "z.csv"] + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("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), + ], +) +def test_period_identity_and_original_attempt_are_strict(field: str, value: object) -> None: + with pytest.raises(PeriodLockError): + parse_period_lock(payload(**{field: value})) + + +@pytest.mark.parametrize("value", [-1, 2**53, True, "1"]) +def test_source_attempt_rejects_unsafe_integer_shapes(value: object) -> None: + with pytest.raises(PeriodLockError): + parse_period_lock(payload(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): + parse_period_lock(payload(source_artifacts=[artifact_payload(path)])) + + +def test_unknown_missing_and_duplicate_keys_fail_closed() -> None: + unknown = payload(extra=True) + with pytest.raises(PeriodLockError): + parse_period_lock(unknown) + missing = payload() + del missing["source_snapshot_digest"] + with pytest.raises(PeriodLockError): + parse_period_lock(missing) + duplicate = b'{"calendar":"utc_iso_week_monday_sunday","calendar":"utc_iso_week_monday_sunday"}' + with pytest.raises(PeriodLockError): + parse_period_lock_bytes(duplicate) + + +def test_noncanonical_wire_and_generated_at_are_rejected() -> None: + lock = parse_period_lock(payload()) + canonical = serialize_period_lock(lock) + with pytest.raises(PeriodLockError): + parse_period_lock_bytes(b" " + canonical) + with pytest.raises(PeriodLockError): + parse_period_lock(payload(generated_at="2026-07-13T00:00:00Z")) + + +def test_expected_lock_requires_exact_original_evidence() -> None: + actual = parse_period_lock(payload()) + assert_expected_period_lock(actual, actual) + with pytest.raises(PeriodLockError): + assert_expected_period_lock(actual, parse_period_lock(payload(source_run_id="29396791051"))) + with pytest.raises(PeriodLockError): + assert_expected_period_lock(actual, parse_period_lock(payload(source_snapshot_digest="d" * 64)))