feat(ops): daily heartbeat + rehearsal review procedure

.github/workflows/scrape.yml

@@ -227,6 +227,35 @@ jobs:
           retention-days: 14
           if-no-files-found: warn
 
+      # Daily heartbeat: ONE comment per run on the ops-log issue (vars.OPS_ISSUE),
+      # empty runs included. A missing comment on a cron day is itself the alert
+      # (the cron did not fire). Fixed format; no PII (counts + a run URL only).
+      - name: Post ops heartbeat
+        if: always() && env.DRY_RUN != 'true' && vars.OPS_ISSUE != ''
+        env:
+          GH_TOKEN: ${{ secrets.SCRAPE_BOT_TOKEN || secrets.GITHUB_TOKEN }}
+        run: |
+          NEW=0; UPDATED=0; REVIEW=0; PUBLISHED=0; REJECTED=0; COST=0
+          FETCHED=0; PROCESSED=0; SKIPPED=0; EXTRACTED=0
+          [ -f logs/run_summary.env ] && . logs/run_summary.env
+          LEDGER="$(jq '.documents | length' data/_meta/processed.json 2>/dev/null || echo 0)"
+          STATUS="ok"; ERRS=""
+          if [ -f logs/gemini_cost.json ]; then
+            TRUNC="$(jq -r '.truncated_reason // empty' logs/gemini_cost.json)"
+            [ -n "$TRUNC" ] && STATUS="truncated:$TRUNC"
+            [ "$(jq -r '.aborted' logs/gemini_cost.json)" = "true" ] && STATUS="aborted"
+            ERRS="$(jq -r '(.error_samples // []) | join(" | ")' logs/gemini_cost.json | cut -c1-200)"
+          fi
+          [ "${{ job.status }}" = "failure" ] && STATUS="RUN FAILED"
+          INSCOPE=$((PUBLISHED + REVIEW))
+          URL="${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
+          BODY="$(printf '**%s** · [run](%s) · trigger: %s\n- docs: fetched **%s** · processed %s (skipped %s settled) · ledger **%s** settled\n- extract: %s extracted · %s in-scope · **%s** published · %s quarantined · %s out-of-scope\n- cost: $%s · status: **%s**%s' \
+            "$(date +%F)" "$URL" "${{ github.event_name }}" \
+            "$FETCHED" "$PROCESSED" "$SKIPPED" "$LEDGER" \
+            "$EXTRACTED" "$INSCOPE" "$PUBLISHED" "$REVIEW" "$REJECTED" \
+            "$COST" "$STATUS" "${ERRS:+ · errors: $ERRS}")"
+          gh issue comment "${{ vars.OPS_ISSUE }}" --body "$BODY"
+
       - name: Open issue on failure or large review queue
         if: always() && (failure() || fromJSON(steps.counts.outputs.review || '0') > 20)
         env:

CONTRIBUTING.md

@@ -152,6 +152,21 @@ and an unprotected side-branch avoids needing to.
   from now on under that fixed window). If you later widen the states or lookback,
   **delete the `ledger-state` branch** so those documents are re-examined.
 
+### Daily review procedure (rehearsal + cron period)
+
+Every scrape run posts one **heartbeat** comment to the pinned ops-log issue
+(number in the `OPS_ISSUE` repo variable) — cron and manual, empty or not.
+
+- **Review the heartbeat daily.** It carries date, run link, docs
+  (fetched/processed/skipped-settled/ledger settled), extract counts
+  (extracted/in-scope/published/quarantined/out-of-scope), cost, and status
+  (`ok`/`truncated`/`aborted`/`RUN FAILED`) with any provider `error_samples`.
+  No PII — counts and a run URL only.
+- **A cron day with no comment is itself the alert** — the run did not fire.
+- **Review the data-review PR whenever one exists.** The standing orders (empty
+  days, a publishable record, a guardrail failure, provider aborts) and the
+  launch-eligibility definition live in that pinned issue.
+
 ---
 
 ## 2. The data lifecycle

pipeline/__main__.py

@@ -231,6 +231,8 @@ def _write_logs(report: RunReport, logs_dir: Path, run_date: str) -> None:
     (logs_dir / "run_summary.env").write_text(
         f"NEW={report.new}\nUPDATED={report.updated}\nREVIEW={report.review}\n"
         f"PUBLISHED={report.published}\nREJECTED={report.rejected_out_of_scope}\n"
+        f"FETCHED={report.fetched}\nPROCESSED={report.processed}\n"
+        f"SKIPPED={report.skipped_settled}\nEXTRACTED={report.extracted}\n"
         f"COST={report.estimated_usd:.6f}\n",
         encoding="utf-8",
     )

pipeline/tests/test_orchestrator.py

@@ -44,6 +44,9 @@ def test_dry_run_end_to_end(tmp_path: Path) -> None:
 
     env = (tmp_path / "logs" / "run_summary.env").read_text()
     assert "NEW=1" in env and "REVIEW=0" in env
+    # Heartbeat fields present for the ops-log comment.
+    for key in ("FETCHED=", "PROCESSED=", "SKIPPED=", "EXTRACTED=", "COST="):
+        assert key in env
 
 
 def test_real_branch_with_injected_client_merges_sources(tmp_path: Path) -> None: