Tighten weekly artifact boundaries

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

Author: Pigbibi

.github/workflows/rss_source_pipeline.yml

@@ -85,32 +85,6 @@ jobs:
             --watchlist "${WATCHLIST_PATH}" \
             --events data/output/rss_source_pipeline/source_events.csv \
             --output data/output/rss_source_pipeline/source_tracker.csv
-      - name: Build completed weekly producer artifact
-        env:
-          PERIOD_START: ${{ github.event.inputs.period_start || '' }}
-          AS_OF: ${{ github.event.inputs.as_of || '' }}
-          WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }}
-        run: |
-          set -euo pipefail
-          test "${GITHUB_RUN_ATTEMPT}" = "1"
-          mkdir -p data/output/political-event-weekly-v1
-          ARGS=(
-            --generated-at "$(date -u +%Y-%m-%dT%H:%M:%S.000000Z)"
-            --workflow-ref "${GITHUB_WORKFLOW_REF}"
-            --source-run-id "${GITHUB_RUN_ID}"
-            --producer-ref "${GITHUB_SHA}"
-            --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
-          )
-          if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then
-            ARGS+=(--scheduled-today "$(date -u +%F)" --run-mode scheduled)
-          else
-            test -n "${PERIOD_START}" && test -n "${AS_OF}"
-            ARGS+=(--period-start "${PERIOD_START}" --as-of "${AS_OF}" --run-mode manual)
-          fi
-          python scripts/build_weekly_artifact.py "${ARGS[@]}"
       - name: Publish live CSV outputs to repository
         env:
           COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }}
@@ -143,6 +117,32 @@ jobs:
             git commit -m "Update live RSS source events [skip ci]"
             git push
           fi
+      - name: Build completed weekly producer artifact
+        env:
+          PERIOD_START: ${{ github.event.inputs.period_start || '' }}
+          AS_OF: ${{ github.event.inputs.as_of || '' }}
+          WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }}
+        run: |
+          set -euo pipefail
+          test "${GITHUB_RUN_ATTEMPT}" = "1"
+          mkdir -p data/output/political-event-weekly-v1
+          ARGS=(
+            --generated-at "$(date -u +%Y-%m-%dT%H:%M:%S.000000Z)"
+            --workflow-ref "${GITHUB_WORKFLOW_REF}"
+            --source-run-id "${GITHUB_RUN_ID}"
+            --producer-ref "${GITHUB_SHA}"
+            --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
+          )
+          if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then
+            ARGS+=(--scheduled-today "$(date -u +%F)" --run-mode scheduled)
+          else
+            test -n "${PERIOD_START}" && test -n "${AS_OF}"
+            ARGS+=(--period-start "${PERIOD_START}" --as-of "${AS_OF}" --run-mode manual)
+          fi
+          python scripts/build_weekly_artifact.py "${ARGS[@]}"
       - name: Upload RSS source artifact
         uses: actions/upload-artifact@v7
         with:

docs/weekly_producer_artifact_contract.md

@@ -24,6 +24,11 @@ political_event_weekly.json
 weekly_manifest.json
 ```
 
+The producer filters `political_events.csv` to the locked inclusive date range
+before writing it; malformed `event_date` fails closed. The original source
+events/watchlist input snapshot digest remains in `period_lock.json`, while the
+filtered event bytes and row count are bound by the artifact manifest.
+
 The manifest binds the first four files by name, byte length and SHA-256. It also
 records CSV headers/row counts, period/as-of/generated-at, producer SHA,
 workflow ref, run id/attempt, source snapshot digest/provenance, and complete

src/political_event_tracking_research/rss_source_fetch.py

@@ -178,6 +178,7 @@ 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),
+        "complete": bool(statuses) and all(item.ok for item in statuses),
         "item_count": item_count,
         "feeds": [item.to_json() for item in statuses],
     }

src/political_event_tracking_research/weekly_artifact.py

@@ -93,7 +93,7 @@ def pairs(items: list[tuple[str, object]]) -> dict[str, object]:
     return value
 
 
-def _csv_snapshot(value: object, expected_header: tuple[str, ...], code: str) -> tuple[int, tuple[str, ...]]:
+def _csv_snapshot(value: object, expected_header: tuple[str, ...], code: str, *, allow_empty: bool = False) -> tuple[int, tuple[str, ...]]:
     if type(value) is not bytes:
         raise _invalid(code)
     try:
@@ -103,15 +103,44 @@ def _csv_snapshot(value: object, expected_header: tuple[str, ...], code: str) ->
         raise _invalid(code) from None
     if not rows or tuple(rows[0]) != expected_header:
         raise _invalid(f"{code}_header")
-    if len(rows) < 2 or any(len(row) != len(expected_header) for row in rows[1:]):
+    if (not allow_empty and len(rows) < 2) or any(len(row) != len(expected_header) for row in rows[1:]):
         raise _invalid(f"{code}_rows")
     return len(rows) - 1, expected_header
 
 
+def _filter_events(value: object, period_start: date, as_of: date) -> bytes:
+    if type(value) is not bytes:
+        raise _invalid("events_csv_invalid")
+    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")
+        event_date = row[1]
+        if type(event_date) is not str or len(event_date) != 10:
+            raise _invalid("events_date_invalid")
+        try:
+            parsed_date = date.fromisoformat(event_date)
+        except ValueError:
+            raise _invalid("events_date_invalid") from None
+        if period_start <= parsed_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 _feed_status(value: object) -> WeeklyFeedStatus:
     if not isinstance(value, Mapping):
         raise _invalid("feed_status_invalid")
-    required = ("feed_count", "successful_feed_count", "failed_feed_count")
+    required = ("feed_count", "successful_feed_count", "failed_feed_count", "complete")
     if any(key not in value for key in required):
         raise _invalid("feed_status_invalid")
     try:
@@ -121,7 +150,7 @@ def _feed_status(value: object) -> WeeklyFeedStatus:
             value["failed_feed_count"],
             value.get("stale_feed_count", 0),
             value.get("missing_feed_count", 0),
-            True,
+            value["complete"],
         )
     except (WeeklyContractError, TypeError, ValueError):
         raise _invalid("feed_status_incomplete") from None
@@ -173,7 +202,7 @@ def _file_metadata(name: str, value: bytes, *, role: str, row_count: int | None
 
 
 def _manifest_payload(lock: PoliticalEventWeeklyPeriodLockV1, contract: WeeklySourceContract, files: Mapping[str, bytes]) -> dict[str, object]:
-    event_rows, event_header = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid")
+    event_rows, event_header = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid", allow_empty=True)
     watch_rows, watch_header = _csv_snapshot(files[WATCHLIST_NAME], WATCHLIST_HEADER, "watchlist_csv_invalid")
     return {
         "manifest_version": MANIFEST_VERSION,
@@ -221,14 +250,16 @@ def build_weekly_artifact(
     generated_at = _timestamp(generated_at)
     if generated_at < datetime.combine(period_end, datetime.min.time(), timezone.utc):
         raise _invalid("generated_at_invalid")
-    event_rows, _ = _csv_snapshot(source_events, EVENT_HEADER, "events_csv_invalid")
+    raw_snapshot_digest = _snapshot_digest(source_events, watchlist)
+    period_events = _filter_events(source_events, period_start, as_of)
+    event_rows, _ = _csv_snapshot(period_events, EVENT_HEADER, "events_csv_invalid", allow_empty=True)
     watch_rows, _ = _csv_snapshot(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid")
     status = _feed_status(feed_status)
     if type(source_provenance) is not str or source_provenance != "official_rss_source_pipeline_v1":
         raise _invalid("source_provenance_invalid")
     if type(run_mode) is not str or run_mode not in {"scheduled", "manual"}:
         raise _invalid("run_mode_invalid")
-    source_snapshot_digest = _snapshot_digest(source_events, watchlist)
+    source_snapshot_digest = raw_snapshot_digest
     source_snapshot_id = f"rss_source_snapshot_{as_of:%Y%m%d}_{source_run_id}"
     try:
         lock = PoliticalEventWeeklyPeriodLockV1(
@@ -243,7 +274,7 @@ def build_weekly_artifact(
             source_snapshot_digest,
             source_provenance,
             (
-                SourceSnapshotArtifact(EVENTS_NAME, _sha256(source_events), event_rows),
+                SourceSnapshotArtifact(EVENTS_NAME, _sha256(period_events), event_rows),
                 SourceSnapshotArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_rows),
             ),
         )
@@ -256,7 +287,7 @@ def build_weekly_artifact(
             producer_ref,
             source_provenance,
             (
-                WeeklySourceArtifact(EVENTS_NAME, _sha256(source_events), event_rows),
+                WeeklySourceArtifact(EVENTS_NAME, _sha256(period_events), event_rows),
                 WeeklySourceArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_rows),
             ),
             status,
@@ -267,7 +298,7 @@ def build_weekly_artifact(
         raise _invalid("weekly_artifact_invalid") from None
     files: dict[str, bytes] = {
         PERIOD_LOCK_NAME: period_lock,
-        EVENTS_NAME: source_events,
+        EVENTS_NAME: period_events,
         WATCHLIST_NAME: watchlist,
         WEEKLY_NAME: weekly,
     }
@@ -293,10 +324,11 @@ def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]:
         raise _invalid("weekly_contract_invalid") from None
     if serialize_weekly_contract(contract) != files[WEEKLY_NAME]:
         raise _invalid("weekly_noncanonical")
-    event_rows, _ = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid")
+    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_rows, _ = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid", allow_empty=True)
     watch_rows, _ = _csv_snapshot(files[WATCHLIST_NAME], WATCHLIST_HEADER, "watchlist_csv_invalid")
-    if _snapshot_digest(files[EVENTS_NAME], files[WATCHLIST_NAME]) != lock.source_snapshot_digest:
-        raise _invalid("source_snapshot_digest_mismatch")
     expected_artifacts = (
         (EVENTS_NAME, _sha256(files[EVENTS_NAME]), event_rows),
         (WATCHLIST_NAME, _sha256(files[WATCHLIST_NAME]), watch_rows),
@@ -313,7 +345,6 @@ def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]:
         or lock.producer_ref != contract.producer_ref
         or lock.source_provenance != contract.source_provenance
         or lock.source_snapshot_id != expected_snapshot_id
-        or lock.source_snapshot_digest != _snapshot_digest(files[EVENTS_NAME], files[WATCHLIST_NAME])
         or lock.source_attempt != 1
     ):
         raise _invalid("period_contract_mismatch")

tests/test_rss_source_fetch.py

@@ -105,6 +105,7 @@ def fake_fetch(url: str) -> bytes:
     payload = json.loads(status.read_text(encoding="utf-8"))
     assert payload["successful_feed_count"] == 1
     assert payload["failed_feed_count"] == 1
+    assert payload["complete"] is False
     assert payload["feeds"][1]["feed_id"] == "bad"
     assert "RuntimeError" in payload["feeds"][1]["error"]
 

tests/test_weekly_artifact.py

@@ -27,6 +27,7 @@ def feed_status(**overrides: object) -> dict[str, object]:
         "failed_feed_count": 0,
         "stale_feed_count": 0,
         "missing_feed_count": 0,
+        "complete": True,
     }
     value.update(overrides)
     return value
@@ -104,6 +105,44 @@ def test_missing_or_malformed_csv_does_not_build(field: str) -> None:
         build(**{field: b"not,the,approved,header\n"})
 
 
+def test_events_are_filtered_to_locked_period_deterministically() -> None:
+    source = EVENTS + b"old,2026-07-05,MSFT,public_mention,neutral,high,https://example.test/old,note\n"
+    files = build(source_events=source)
+    assert files["political_events.csv"] == EVENTS
+    base_lock = json.loads(build()["period_lock.json"])
+    filtered_lock = json.loads(files["period_lock.json"])
+    assert filtered_lock["source_snapshot_digest"] != base_lock["source_snapshot_digest"]
+    manifest = json.loads(files["weekly_manifest.json"])
+    event_file = next(item for item in manifest["files"] if item["name"] == "political_events.csv")
+    assert event_file["row_count"] == 1
+
+
+def test_event_date_outside_period_is_filtered_including_next_monday() -> None:
+    source = EVENTS + b"next,2026-07-13,MSFT,public_mention,neutral,high,https://example.test/next,note\n"
+    files = build(source_events=source)
+    assert files["political_events.csv"] == EVENTS
+
+
+def test_malformed_event_date_fails_closed() -> None:
+    source = EVENTS + b"bad,not-a-date,MSFT,public_mention,neutral,high,https://example.test/bad,note\n"
+    with pytest.raises(WeeklyArtifactError) as error:
+        build(source_events=source)
+    assert error.value.code == "events_date_invalid"
+
+
+def test_zero_event_week_is_allowed() -> None:
+    files = build(source_events=b"event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n")
+    manifest = json.loads(files["weekly_manifest.json"])
+    event_file = next(item for item in manifest["files"] if item["name"] == "political_events.csv")
+    assert event_file["row_count"] == 0
+    assert files["political_events.csv"].endswith(b"\n")
+
+
+def test_incoming_complete_false_is_not_overridden() -> None:
+    with pytest.raises(WeeklyArtifactError):
+        build(feed_status=feed_status(complete=False))
+
+
 def test_manifest_file_tamper_is_rejected() -> None:
     files = build()
     tampered = dict(files)

tests/test_weekly_producer_workflow.py

@@ -0,0 +1,16 @@
+from pathlib import Path
+
+
+WORKFLOW = Path(__file__).parents[1] / ".github/workflows/rss_source_pipeline.yml"
+
+
+def test_legacy_live_publication_precedes_weekly_artifact_failure_boundary() -> None:
+    text = WORKFLOW.read_text(encoding="utf-8")
+    assert text.index("Publish live CSV outputs to repository") < text.index("Build completed weekly producer artifact")
+
+
+def test_dedicated_artifact_is_separate_from_legacy_source_artifact() -> None:
+    text = WORKFLOW.read_text(encoding="utf-8")
+    assert "path: data/output/rss_source_pipeline/" in text
+    assert "path: data/output/political-event-weekly-v1/" in text
+    assert text.index("Upload RSS source artifact") < text.index("Upload political weekly artifact")