Add pipeline reliability framework: survival guardrails, schema-drift detection, month-end ops runbooks

- Risk guardrails, schema-drift detection, structured logging, upstream registry
- Gold gate build + streaming dead-letter replay
- Month-end critical-path job, ops SQL scorecards, incident runbook & postmortem template
- Governance/contract/AI-risk CI check scripts and reliability test suites

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Mitchell-MC

.github/workflows/ci.yml

@@ -13,6 +13,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
 
       - uses: actions/setup-python@v5
         with:
@@ -22,10 +24,19 @@ jobs:
         run: python -m pip install -e ".[dev]"
 
       - name: Lint
-        run: ruff check src tests
+        run: ruff check src tests scripts
 
       - name: Unit tests
-        run: pytest -v
+        run: pytest -v --cov=src/aml_lakehouse --cov-report=term-missing --cov-fail-under=35
+
+      - name: AI code safety audit
+        run: python scripts/check_ai_risk_patterns.py
+
+      - name: Governance policy checks
+        run: python scripts/check_governance_policies.py
+
+      - name: Contract risk check
+        run: python scripts/check_contract_risk.py
 
       - name: Validate .sql files parse into well-formed statements
         # Lightweight pre-deploy check: every .sql file under src/ must split into at least

.gitignore

@@ -14,6 +14,8 @@ venv/
 .pytest_cache/
 .mypy_cache/
 .ruff_cache/
+.coverage
+htmlcov/
 build/
 dist/
 

README.md

@@ -105,7 +105,10 @@ for both assumptions in full.
 5. **Build Gold**: run `gold/build_gold.sql`
 6. **Streaming**: `databricks bundle run streaming_validation_run --target dev` (drips
    AMLSim transactions through a paced producer into the Structured Streaming consumer)
-7. **Dashboard**: open `bi/AML_Dashboard.pbip` in Power BI Desktop — see
+7. **Month-end critical path**: `databricks bundle run month_end_critical_path --target dev`
+  (runs guarded publish flow with reconciliation gate checks)
+8. **Dead-letter replay (as needed)**: `databricks bundle run replay_txn_dead_letter --target dev`
+9. **Dashboard**: open `bi/AML_Dashboard.pbip` in Power BI Desktop — see
    [bi/README.md](bi/README.md)
 
 ## Project status
@@ -155,5 +158,13 @@ for both assumptions in full.
   duplicate-amplification incident: detection, root cause, fix, verification
 - [docs/05_architecture.md](docs/05_architecture.md) — architecture diagram and layer
   responsibilities
+- [docs/06_pipeline_survival_framework.md](docs/06_pipeline_survival_framework.md) —
+  fail-loud coding standards, CI gatekeepers, and governance-as-code checklist
+- [docs/07_month_end_incident_runbook.md](docs/07_month_end_incident_runbook.md) —
+  month-end triage, stakeholder communication cadence, and recovery checklist
+- [docs/08_postmortem_template.md](docs/08_postmortem_template.md) —
+  incident postmortem template with mandatory reliability-upgrade action
 - [dataset_inspection_notes.md](dataset_inspection_notes.md) — real dataset schemas, row
   counts, and every bug found during ingestion
+- [resources/ops/README.md](resources/ops/README.md) — operational SQL query pack for
+  stale-impact triage, owner handoffs, schema drift, and SLA/cost scorecards

docs/01_nonfunctional_requirements.md

@@ -63,3 +63,14 @@ controls.
   explainable and reproducible).
 - Ingestion metadata (source, timestamp, schema version, row/file counts) is immutable and
   queryable per batch, forming the audit trail for "what data was known at decision time."
+
+## Enforced controls
+
+- **Fail-loud coding standard**: broad or silent exception handlers are blocked in CI via
+  `scripts/check_ai_risk_patterns.py`.
+- **Data contract guardrails**: required-column/non-empty/invalid-ratio checks are enforced
+  with `src/aml_lakehouse/common/risk_guardrails.py`.
+- **Structured incident telemetry**: batch and streaming jobs emit JSON logs with row counts,
+  lag, and quality metrics.
+- **Governance-as-code checks**: CI runs `scripts/check_governance_policies.py` to catch
+  secret patterns and blocked PII field names in SQL assets before merge.

docs/03_schema_contracts.md

@@ -42,7 +42,7 @@ text + `_error_reason`, `_ingested_at`) rather than failing the batch — see Ph
 | `silver.entity_alias` | one row per alias | `entity_id`, `alias_name`, `alias_normalized` | Normalized = uppercased, punctuation-stripped, whitespace-collapsed — the actual normalization function is a Phase 3 deliverable |
 | `silver.entity_address` | one row per address | `entity_id`, `address_line`, `city`, `country` | Parsed out of OFAC's combined `city_state_zip_etc` and OpenSanctions' free-text `addresses` |
 | `silver.account` | one row per account | `account_id` (= AMLSim `nodeid`) | `synthetic_holder_name`, `synthetic_country`, `synthetic_dob` (the account's own fabricated identity attributes from the synthetic identity-linkage step — **not** a FK to `silver.entity`, see below), `init_balance`, `opened_date` (synthetic, derived from the date anchor) |
-| `silver.transaction` | one row per transaction | `transaction_id` (surrogate) | `source_account_id`, `target_account_id`, `amount`, `currency` (assumed USD), `event_time` (derived from synthetic date anchor + AMLSim `time` step) |
+| `silver.transaction` | one row per transaction | `transaction_id` (surrogate) | `source_account_id`, `target_account_id`, `amount`, `currency` (assumed USD), `event_time` (derived from synthetic date anchor + AMLSim `time` step), `event_date` (partition column for pruning) |
 | `silver.match_candidate` | one row per candidate match | `match_id` | `account_id`, `entity_id` (the watchlist entity), `match_confidence_band` (`exact` / `strong` / `weak`), `match_score`, `matched_on` (name/alias/dob/country), `explanation` (structured — which fields matched and how) |
 | `silver.elliptic_txn_risk` | one row per Elliptic tx | `txId` | `class`, `time_step`, `in_degree`, `out_degree` (from `bronze.elliptic_txs_edgelist`), `illicit_neighbor_ratio` (share of 1-hop neighbors labeled illicit — the interpretable signal this dataset can actually offer, since the 165 raw features can't be feature-engineered further). Real SQL self-join/window-function work, standalone graph-topology analytics — not natively joinable to `silver.account`, see the bridge table below. |
 | `silver.network_risk_reference` | one row per `account_id` that received a synthetic linkage | `account_id` | `network_risk_exposure` (0-1, sampled from the `illicit_neighbor_ratio` distribution in `silver.elliptic_txn_risk`), `linkage_note` (fixed string documenting this is a simulated bridge, not a real Bitcoin-to-remittance-account linkage) — see caveat below |
@@ -80,7 +80,8 @@ linkage is disclosed.
 | `gold.entity_risk_profile` | one row per entity/account, per `as_of_date` | `entity_id`, `account_id`, `as_of_date` | `watchlist_match_summary` (struct), behavioral features: `velocity_1d/7d/30d`, `counterparty_diversity_30d`, `burst_flag`, `round_dollar_ratio`, `risky_counterparty_exposure`, `network_risk_exposure` (from `silver.network_risk_reference`, nullable — only populated for accounts that received the synthetic Elliptic linkage); `composite_risk_score`, `score_band`, `top_contributing_factors` (array of struct: factor name, value, weight/contribution) |
 | `gold.prioritized_alert_queue` | one row per active alert | `alert_id` | `entity_id`, `account_id`, `priority_rank`, `composite_risk_score`, `escalation_reason` (`sanctions_hard_override` vs. `behavioral_threshold`), `status` (`new`/`reviewed`/`suppressed`), `generated_at` |
 | `gold.watchlist_match_summary` | one row per watchlist entity | `watchlist_entity_id` | `matched_account_count`, `highest_confidence_band`, `jurisdictions`, `risk_type` |
-| `gold.ops_control` | one row per pipeline batch | `pipeline_name`, `batch_id` | `last_successful_batch_id`, `processed_row_count`, `expected_row_count`, `lag_seconds`, `freshness_status` (`fresh`/`stale`/`degraded`), `dead_letter_count`, `checkpoint_offset` — the silent-failure/ops layer from Phase 6/8 |
+| `gold.ops_control` | one row per pipeline batch | `pipeline_name`, `batch_id` | `business_process`, `impacted_table`, `last_successful_batch_id`, `last_good_batch_age_minutes`, `processed_row_count`, `expected_row_count`, `lag_seconds`, `freshness_status` (`fresh`/`stale`/`degraded`), `dead_letter_count`, `checkpoint_offset`, `upstream_owner`, `upstream_contact`, `runtime_seconds`, `estimated_cloud_cost_usd` — actionable business-impact alerting + ops/cost scorecard |
+| `gold.month_end_reconciliation_serving` | one row per entity risk profile in serving mode | `entity_id`, `account_id`, `as_of_date` | Adds serving metadata (`as_of_batch_id`, `is_stale`, `last_good_batch_age_minutes`) so finance consumers always know freshness/lineage of month-end outputs |
 
 `top_contributing_factors` and `escalation_reason` exist specifically to satisfy the audit
 requirement in `docs/01_nonfunctional_requirements.md` — every ranked alert must show its

docs/06_pipeline_survival_framework.md

@@ -0,0 +1,65 @@
+# Pipeline Survival Framework (2026)
+
+This checklist codifies the production-safety standards used in this repository. It exists
+to prevent "green but wrong" pipelines: jobs that run successfully while silently damaging
+trust, cost, or compliance.
+
+## 1) Fail Loud by Default
+
+- Broad exception handlers (`except Exception`, bare `except`) are blocked in CI by
+  `scripts/check_ai_risk_patterns.py`.
+- Only recoverable cases should be handled (for example, rate-limit retries); everything
+  else must fail immediately with traceable context.
+
+## 2) Enforce Data Contracts at Boundaries
+
+- Critical jobs validate required input columns and non-empty source datasets.
+- Contract violations raise `DataContractError` from
+  `src/aml_lakehouse/common/risk_guardrails.py`.
+
+## 3) Use Circuit Breakers for Data Quality
+
+- Streaming ingestion enforces a hard invalid-row ratio cap (`MAX_INVALID_RATIO`) and aborts
+  the micro-batch when exceeded.
+- This prevents bad data from propagating silently downstream.
+
+## 4) Require Structured Incident Logging
+
+- Batch and streaming jobs emit JSON structured events (timestamp, pipeline, batch,
+  row counts, quality/circuit-breaker details) via
+  `src/aml_lakehouse/common/structured_logging.py`.
+
+## 5) Make Idempotency Non-Negotiable
+
+- Streaming writes use deterministic natural-key `MERGE` semantics (`_merge_append`) so
+  restarts/replays do not duplicate facts.
+- Silver and Gold table builds are deterministic and rerunnable.
+
+## 6) Enforce CI/CD Gatekeepers
+
+- CI blocks merges when lint, tests, SQL statement validation, AI safety audit, governance
+  checks, or coverage threshold fail.
+- Coverage floor is enforced in `.github/workflows/ci.yml`.
+
+## 7) Test Failure Paths Explicitly
+
+- Unit tests must include fail-on-bad-data behavior, not just happy-path outcomes.
+- `tests/test_risk_guardrails.py` validates contract and circuit-breaker failures.
+
+## 8) Tie Engineering to Business Outcomes
+
+- Every pipeline run records operational metrics to `gold.ops_control` for before/after
+  benchmarking and trend monitoring.
+- Optimization and reliability changes should be justified in terms of cost, freshness,
+  and incident reduction.
+
+## 9) Audit AI-Generated Code Before Merge
+
+- Automated static checks review risky code patterns commonly emitted by LLMs.
+- Human review still verifies architecture, scalability, and cost-aware behavior.
+
+## 10) Treat Governance as Code
+
+- CI policy checks detect secret leaks and blocked PII field patterns before merge.
+- Data classification and auditability requirements remain defined in
+  `docs/01_nonfunctional_requirements.md`.

docs/07_month_end_incident_runbook.md

@@ -0,0 +1,40 @@
+# Month-End Incident Runbook
+
+Use this when finance-facing reconciliation outputs are stale, blank, or failed during close.
+
+## 0-15 minutes: Triage and communication
+
+- Answer three questions first: what failed, impact, who must know now.
+- Send first stakeholder update immediately: "Issue acknowledged, impact under assessment, next update in 30 minutes."
+- Identify failed step(s), upstream dependency status, and last successful run timestamp.
+
+## 15-45 minutes: Investigation
+
+- Confirm whether failure mode is schema drift, late-arriving data, permissions, or bad joins.
+- Compare current schema to last known schema snapshot (`gold.ops_schema_drift`).
+- Validate business impact:
+  - Is `gold.month_end_reconciliation_serving` stale?
+  - Is `as_of_batch_id` behind expected close window?
+
+## 45+ minutes: Recovery
+
+- Apply a temporary patch with explicit audit trail.
+- Re-run month-end critical path job: `month_end_critical_path`.
+- If invalid records caused failure, quarantine and replay via `replay_txn_dead_letter`.
+- Validate reconciliation gate metrics and source/target tolerances before confirming recovery.
+
+## Required status update template
+
+- Incident: month-end reconciliation pipeline degraded/failed
+- Impacted table(s): <table>
+- Business impact: <close blocked / stale but serving>
+- Last good batch: <batch id>
+- ETA to next update: <time>
+- Current owner: <team/contact>
+
+## Recovery completion checklist
+
+- Month-end outputs are fresh or explicitly marked stale with known reason.
+- `gold.ops_control` has run metadata for the recovery batch.
+- Schema drift entry captured in `gold.ops_schema_drift`.
+- Postmortem opened using `docs/08_postmortem_template.md`.

docs/08_postmortem_template.md

@@ -0,0 +1,65 @@
+# Incident Postmortem Template
+
+## Summary
+
+- Incident ID:
+- Date/Time (UTC):
+- Duration:
+- Severity:
+- Owner:
+- Impacted business process:
+
+## What happened
+
+- Trigger event:
+- Detection channel:
+- First customer/stakeholder impact observed:
+- Last known good batch:
+
+## Timeline
+
+- T0:
+- T+15:
+- T+30:
+- T+60:
+- Recovery complete:
+
+## Root cause
+
+- Primary technical cause:
+- Contributing causes:
+- Why existing controls did not prevent it:
+
+## Detection and response quality
+
+- Was alert actionable?
+- Was ownership clear?
+- Did communication cadence meet runbook expectations?
+
+## Corrective actions
+
+- Immediate remediation:
+- Long-term prevention:
+
+## Mandatory reliability upgrade (required)
+
+At least one systemic control must be added/updated from this incident:
+
+- [ ] Contract validation
+- [ ] Schema drift guard
+- [ ] Reconciliation check
+- [ ] Alert quality/business-impact routing
+- [ ] Idempotent replay/recovery automation
+- [ ] Documentation/runbook update
+
+## Verification
+
+- Test evidence for fix:
+- Rollback plan validated:
+- CI checks updated:
+
+## Sign-off
+
+- Engineering:
+- Analytics/Finance stakeholder:
+- Compliance (if applicable):

resources/dashboards/README.md

@@ -66,3 +66,15 @@ No `network_risk_exposure` / Elliptic-sourced widget yet, and the alert-queue ta
 yet show `top_contributing_factors` (an `array<struct>` column — same Power Query-style
 "needs an explode/flatten step" caveat as `WatchlistMatchSummary.jurisdictions` in the Power
 BI project). Straightforward to add as a further dataset/widget; not done here for time.
+
+## Ops triage companion
+
+For incident response and month-end readiness checks, use the operational SQL pack in
+`resources/ops/` (especially `month_end_status.sql`, `owner_handoff_queue.sql`, and
+`schema_drift_breaking_changes.sql`). This lets on-call and finance teams answer freshness,
+impact, and ownership questions immediately even before dashboard updates are published.
+
+The Lakeview dashboard now also includes native Operational Health tables for:
+
+- month-end publish/serving status (`table_month_end_status`)
+- ownership-based escalation queue (`table_owner_handoff_queue`)