Stage 3: rewrite as Lakeflow Spark Declarative Pipelines

.gitignore

@@ -24,3 +24,8 @@ build/
 .env.local
 dist/
 .venv/
+
+# Session transcript — regenerate with scripts/export_chat.py.
+# Kept local: it is a working record, not part of the deliverable, and it
+# contains the full back-and-forth rather than the conclusions.
+docs/opus_chat.md

docs/architecture.md

@@ -1,97 +1,146 @@
-# Windmill Pipeline Architecture
+# Architecture
 
-## Overview
+Why the pipeline is shaped the way it is. The README covers what it does and how
+to run it; this covers the decisions and what was rejected.
 
-Windmill is a Databricks demo project implementing a medallion architecture (bronze/silver/gold) for renewable energy turbine data processing.
+## Medallion layers
 
-## Design Decisions
+| Layer | Holds | Rule |
+|---|---|---|
+| Bronze | Raw readings exactly as delivered, plus ingestion metadata | Never filtered, never corrected |
+| Silver | Deduplicated readings with per-field validity flags | Row-level judgements only |
+| Gold | Daily per-turbine statistics and anomaly flags | Anything needing a population |
 
-### 1. Medallion Architecture
-- **Bronze**: Raw data ingestion with minimal transformation
-- **Silver**: Cleaned, validated, and deduplicated data
-- **Gold**: Business-ready aggregations and analytics
+The split that matters is between **silver and gold**, and it is not about
+cleanliness — it is about how much context a check needs.
 
-**Why**: Industry standard for data lakes. Clear separation of concerns, easy to understand and audit.
+A check belongs in **silver** if a single row is enough to judge it: is
+`power_output` null, is it negative, is this turbine reporting from a file it
+does not belong to. No other row is required to answer.
 
-### 2. Databricks Asset Bundles (DABs)
-- Infrastructure as Code for entire pipeline
-- Single deployment command
-- Idempotent, version-controlled
+A check belongs in **gold** if it needs a population: is this turbine-day
+unusual *relative to that turbine's own history*. That cannot be answered one row
+at a time, and silver's watermark exists to bound deduplication state, not to
+define a statistical window.
 
-**Why**: Demo-friendly. Recruiter can deploy in one command. Direct engine (no Terraform dependency).
+## Decisions
 
-### 3. Serverless SQL
-- No compute management
-- Automatic scaling
-- Cost-effective for demos
+### Invalid readings are flagged, not dropped or imputed
 
-**Why**: Free tier accessible. No cluster configuration needed.
+Silver marks each field with a boolean and carries the row through unchanged.
+Gold filters on `is_reading_valid` before computing statistics.
 
-### 4. Data Volume for Landing Zone
-- CSV files uploaded to volume
-- Acts as staging area
-- Immutable source of truth
+Rejected: **forward-filling gaps**, which invents measurements that were never
+taken and makes a broken sensor look like a working one. Rejected: **dropping
+invalid rows**, which destroys the evidence someone needs to diagnose the sensor
+and makes "how much data did we lose?" unanswerable.
 
-**Why**: Separates raw data from processed tables. Easy to re-run pipeline.
+Quarantining in place gives gold clean statistics and keeps quality reporting
+possible. On the declarative branch the same rule appears as `expect_all`
+expectations — warn and keep, never `_or_drop` or `_or_fail`.
 
-## Data Flow
+### Anomalies are flagged, not removed
 
-```
-Raw CSV → Volume (dropzone) → Bronze Table → Silver Table → Gold Table
-                ↓                 ↓             ↓             ↓
-           [01_copy_data]  [02_bronze]    [03_silver]   [04_gold]
-```
+The task asks for turbines that have deviated to be *identified*. Filtering them
+out upstream would delete the deliverable. Gold emits `is_anomaly` alongside the
+bounds that produced it, so a reviewer can see why a day tripped.
 
-## Quality Checks
+### Baselines are computed over daily means, not raw readings
 
-### Bronze Layer
-- Schema validation
-- Deduplication (timestamp + turbine_id)
-- Metadata tracking (ingestion_timestamp, source_file)
+The most consequential decision here, and it was originally wrong.
 
-### Silver Layer
-- Forward-fill missing values (by turbine)
-- Outlier removal (>2 std dev from mean)
-- Metadata preservation
+Averaging 24 hourly readings shrinks the spread by √24 ≈ 5 — the standard error
+of the mean. A band built from reading-level standard deviation is therefore
+about five times too wide for a daily mean to ever escape. Measured on the seed
+data: reading-level σ ≈ 0.86 against daily-mean σ ≈ 0.17, giving a band of
+[1.30, 4.74] versus an actual daily range of [2.47, 3.38].
 
-### Gold Layer
-- 24-hour summary statistics (min, max, avg, stddev)
-- Power output anomaly detection
-- Efficiency score calculation
+The first implementation compared daily averages against reading-level spread and
+reported **zero anomalies across 465 turbine-days** — not because the data was
+clean, but because the test was mathematically incapable of firing. Full
+diagnosis in [`dev_seed_original_results.md`](dev_seed_original_results.md).
 
-## Scalability
+The rule: the baseline must be built from the same unit being tested against it.
 
-### Current (Demo)
-- 15 turbines, hourly data
-- CSV file-based ingestion
-- 3 separate jobs (can be orchestrated)
+### Baselines are per-turbine, not fleet-wide
 
-### Production Evolution
-1. Add job orchestration (Databricks Jobs)
-2. Delta Live Tables (DLT) for pipeline management
-3. Streaming ingestion (Kafka/Kinesis)
-4. Time-series analytics (materialized views)
-5. ML models (anomaly detection, forecasting)
+Turbines sit in different wind conditions. A fleet baseline conflates "this
+turbine is underperforming" with "this turbine is in a calmer spot", and would
+report a well-sited turbine as permanently anomalous. The task asks for deviation
+from *their* expected output.
 
-## Assumptions
+### Validity bounds are `>= 0`, not `> 0`
 
-1. **Data**: Hourly readings per turbine
-2. **Files**: 5 turbines per CSV, consistent schema
-3. **Missing Data**: Forward-fillable sensor gaps
-4. **Outliers**: >2 σ from mean = anomaly
-5. **Auth**: OAuth (OAuth preferred over PAT)
-6. **Cost**: FREE tier Databricks workspace
+A becalmed or idling turbine genuinely reports 0 MW in 0 m/s wind. Treating zero
+as invalid would discard real low-wind data silently. Guarded by unit tests that
+fail if the bound is ever tightened.
 
-## Testing
+## Dataset types (declarative branch)
+
+Bronze and silver are **streaming tables**: append-only, processing only what
+arrived since the last run.
+
+Gold is a **materialized view**. Streaming tables never revisit rows they have
+already emitted, so a daily aggregate built as one would go stale the moment late
+or corrected data reached silver. A materialized view recomputes. It reads silver
+with `spark.read`, not `readStream` — a streaming read would impose watermark and
+state constraints for no benefit, since the anomaly baseline spans the whole
+history rather than a window.
 
-- Unit tests for transformation logic (pytest)
-- Integration tests via Databricks notebooks
-- Data quality assertions in each layer
+## Environment separation
 
-## Future Enhancements
+Targets (`dev`, `test`, and any others added) run the same code against different
+data in separate schemas, isolated by a single `name_prefix` preset that applies
+to schema names as well as job names.
+
+This is what makes end-to-end testing possible. Unit tests prove the transformation
+logic; only a real run proves Auto Loader behaviour, checkpoints, deduplication
+across repeated loads, and the wiring between stages. Without isolation those go
+untested or get tested somewhere that matters.
+
+## Testing
 
-1. **Orchestration**: DABs job scheduling
-2. **Monitoring**: Databricks Jobs SQL alerts
-3. **ML**: MLflow for anomaly detection models
-4. **Real-time**: Structured Streaming for live data
-5. **API**: Databricks Apps for stakeholder dashboards
+Two layers, covering different failures, neither replacing the other.
+
+**Unit tests** (43, ~11s, no cluster) cover statistical behaviour and edge cases
+that are awkward to express as fixtures — null spread, single-day turbines,
+boundary values, per-turbine versus fleet-wide baselines. Possible because the
+transformation logic is packaged separately from the notebooks as pure
+DataFrame-in / DataFrame-out functions.
+
+**Integration** runs the whole pipeline against deliberately corrupted fixtures
+with known expected counts, generated deterministically and asserted against a
+manifest. Covers what unit tests cannot see: file ingestion, checkpoints,
+load-invariance, job wiring.
+
+See [`testing.md`](testing.md), including the mutation check that caught a
+regression guard which passed while the defect was present.
+
+## Scale
+
+Currently 15 turbines at hourly grain — 11,160 rows for a month. Nothing here is
+sized for that: Auto Loader ingests incrementally by file, silver processes only
+new bronze rows, and the volume of a real farm changes the runtime rather than
+the design.
+
+The parts that would need attention first at genuinely large scale:
+
+- **Gold recomputes the full baseline each run.** Fine at 465 turbine-days;
+  at millions it would want incremental refresh or a rolling window.
+- **Deduplication state** is bounded by a 2-day watermark. Higher throughput
+  means more state per micro-batch, and the watermark becomes a real tuning knob
+  rather than a formality.
+- **Liquid clustering** on `(turbine_id, date)` in gold, once table size makes
+  file pruning matter.
+
+## Known gaps
+
+- **Publish step of Write-Audit-Publish is not implemented** — there is no
+  production target to promote into.
+- **Schema drift is untested.** Auto Loader runs on a fixed schema; a file with a
+  renamed or extra column has an unobserved failure mode.
+- **Late-arriving data beyond the watermark** is dropped by design, but there is
+  no explicit test recording that as intentional.
+- **Sub-hourly readings** are detected (`has_sub_hourly_readings`) but not
+  corrected — the right correction depends on sensor semantics that cannot be
+  determined from the data. See the assumptions section of the README.

docs/declarative_pipeline.md

@@ -0,0 +1,150 @@
+# Declarative rewrite (Lakeflow Spark Declarative Pipelines)
+
+The job-based branches orchestrate three notebooks that each read a table, write
+a table, and manage their own checkpoint. This branch replaces them with one
+pipeline that declares the three datasets and lets the runtime work out the rest.
+
+The transformation logic is unchanged — both branches import the same
+`transform_silver` and `build_gold` from the `windmill` wheel, covered by the
+same 43 unit tests. Only the orchestration differs, which is what makes the two
+comparable.
+
+## What the rewrite removes
+
+| Concern | Job-based branches | This branch |
+|---|---|---|
+| Execution order | Three jobs run in sequence, by hand or by a scheduler | Derived from table references in the code |
+| Checkpoints | Explicit path per stream, hand-placed in a volume | Managed by the pipeline |
+| Triggers / output modes | `trigger(availableNow=True)`, `outputMode("append")` per write | Implicit in the dataset type |
+| Writes | `saveAsTable` / `toTable`, plus `overwriteSchema` when columns change | The decorator's return value is the table |
+| Table type | Everything is a plain managed Delta table | Streaming tables for append, materialized view for the aggregate |
+| Quality reporting | A separate job asserting counts after the fact | Expectations recorded per run in the event log |
+
+Three job YAML files and three notebooks are deleted; three transformation files
+and one pipeline YAML replace them.
+
+## Dataset types
+
+Bronze and silver are **streaming tables**: both are append-only, and both
+process only what arrived since the last run.
+
+Gold is a **materialized view**, deliberately. Streaming tables never revisit
+rows they have already emitted, so a daily aggregate computed as one would go
+stale as soon as late or corrected data landed in silver. A materialized view
+recomputes. The read is `spark.read.table`, not `readStream` — reading silver as
+a stream would impose watermark and state constraints for no benefit, since the
+anomaly baseline is computed over the whole history rather than a window.
+
+## Expectations
+
+Every expectation is `expect_all` — warn and keep. Never `_or_drop`, never
+`_or_fail`. That is the declarative statement of a decision made on the first
+branch: invalid readings are flagged, not discarded. Dropping them would hide
+sensor faults from the people who have to fix them, and failing the update would
+take the pipeline down over a handful of bad rows in an otherwise good delivery.
+
+The silver conditions reference the boolean columns the transform already
+produces (`is_power_output_valid`, and so on) rather than restating the rules in
+SQL. One source of truth: the rule cannot drift between the column and the
+expectation, and the column is unit tested in the package.
+
+### Measured against the corrupted fixtures
+
+| Expectation | Failed | Manifest expects |
+|---|---:|---:|
+| `turbine_id_in_expected_group` | 27 | 27 |
+| `power_output_present_and_non_negative` | 9 | 9 |
+| `wind_speed_present_and_non_negative` | 10 | 10 |
+| `wind_direction_within_compass_range` | 8 | 8 |
+| `source_file_maps_to_known_group` | 20 | 20 |
+| `timestamp_present` | 0 | 0 |
+| `turbine_id_present` | 0 | 0 |
+
+27 + 9 + 10 + 8 = 54, the manifest's `invalid_readings_total`. The expectations
+and the fixture generator arrive at the same numbers by completely independent
+routes, which is a stronger check than either alone.
+
+Gold's three expectations pass 465/465: every turbine-day has readings, every
+anomaly flag is decided rather than null, and no day contains sub-hourly
+readings.
+
+Reading the metrics back:
+
+```sql
+SELECT explode(from_json(
+         details:flow_progress.data_quality.expectations,
+         'array<struct<name:string,dataset:string,passed_records:bigint,failed_records:bigint>>'
+       )) AS e
+FROM event_log("<pipeline-id>")
+WHERE details:flow_progress.data_quality.expectations IS NOT NULL
+```
+
+Note the CLI's `list-pipeline-events --output json` flattens nested details and
+returns `flow_progress` as an empty object. Query the event log rather than the
+CLI for anything nested.
+
+## Results
+
+Identical to the job-based branches on every layer that matters:
+
+| Metric | Jobs | Pipeline |
+|---|---:|---:|
+| silver rows | 11,169 | 11,169 |
+| invalid readings | 54 | 54 |
+| `UNKNOWN` group rows | 20 | 20 |
+| turbine-days | 465 | 465 |
+| anomalies | 20 | 20 |
+
+The stage 1 validation job passes unchanged against pipeline-produced tables —
+it queries the same table names and asserts the same manifest, so it is a fair
+cross-check rather than a rewritten test.
+
+Bronze reads 33,534 because the inbox still held three loads when the pipeline
+first ran: `cloudFiles.cleanSource` archives on a retention delay, so previously
+ingested files had not yet moved, and the pipeline's checkpoint was new. Dedup
+collapsed them to the same 11,169 silver rows, which is the invariant that
+matters.
+
+## Migration constraint worth knowing
+
+**A pipeline cannot adopt an existing managed table.** The first run failed with:
+
+```
+Could not materialize `...`.`turbine_raw` because a MANAGED table already
+exists with that name.
+```
+
+The job-based branches had written plain Delta tables at those names. Converting
+them to pipeline-managed datasets is not an in-place operation — the tables must
+be dropped so the pipeline can create and own them. Fine here, where the data is
+regenerable fixtures, but on a real migration it means a cutover plan: either
+write the pipeline to new names and swap, or accept a rebuild window.
+
+## What this branch keeps
+
+`windmill_init` still seeds the dropzone, because getting files into the volume
+is not the pipeline's job.
+
+`windmill_validate` still runs, and is still worth having. Expectations report
+what the pipeline saw; the validation job asserts what it *should* have seen,
+against a manifest generated independently of the pipeline. They fail on
+different things — an expectation cannot notice that a whole file was skipped,
+and it cannot check that silver stayed invariant across repeated loads.
+
+## Trade-offs
+
+**In favour.** Much less orchestration code. The dependency graph is derived
+rather than maintained. Data quality is a first-class, per-run, queryable
+artifact instead of a bespoke job. Table types force an explicit answer to
+"should this recompute or append?".
+
+**Against.** The pipeline owns its tables, so migration is a cutover, not a
+switch. Debugging moves from reading a notebook top to bottom to reading the
+pipeline graph and the event log. Failures surface as flow errors, one level
+removed from the code that caused them. And the whole thing is Databricks
+specific in a way the job-based version — plain PySpark plus a scheduler — is
+not.
+
+Neither is strictly better. The declarative version is the stronger choice when
+the pipeline shape is stable and quality reporting matters; the job version is
+easier to reason about when the logic is still moving, and it ports.