#235: Airflow drift / signal-rot detection (run-over-run)

.claude/rules/airflow-integration.md

@@ -84,6 +84,24 @@ The second operator (sibling of #232), wrapping the no-LLM, read-only `signalfor
 
 Example DAG: `examples/airflow/signalforge_prune_existing_operator_dag.py` (`dag_id="signalforge_prune_existing_operator"`, single `signal_rot_monitor` task, templated `model`/`schema`/`as_of`, no Anthropic key referenced). Gated parse + `render_template_fields` tests in `tests/airflow/test_dag_parse.py`.
 
+## Drift / signal-rot detection (#235 DEC-001…019)
+
+Run-over-run drift detection — compare run N vs N-1 and page on **signal rot** (a test that was `kept` and is now `dropped:always-passes`) or a **grade regression**. The headline scheduled value-add of epic #228. Two surfaces, one shared airflow-free pure core. The durable patterns:
+
+- **Airflow-free pure core in `signalforge/airflow/drift.py`** (DEC-002) — eagerly importable, NO `from airflow`, eager-re-exported from `__init__.py` (alongside `result`/`runner`, NOT the lazy `__getattr__`). Carries `DriftReport` (+ `DriftArtifact`/`GradeRegression`/`SchemaShapeDelta`), the pure `compute_drift(*, previous_diff, current_diff, previous_grade=None, current_grade=None, as_of=None, grade_regression_threshold=0.05) -> DriftReport`, and the fail-soft loaders `load_diff_report`/`load_grade_report`/`parse_diff_report`. Reusable by the v0.8 GH Action (hoist candidate — see v0.8 note).
+- **`on_drift` is the run-over-run analogue of `on_flagged` — most-severe-wins, NOT a 5th axis precedent (DEC-006).** `decide_task_outcome(result, *, on_flagged="fail", on_drift="fail", drift: DriftReport | None = None)`. **Byte-identical when `drift is None`** (every #232/#233 caller + test unchanged — pinned). When `drift.alarming` on an exit-0 run, fold `on_drift` and return the MOST-SEVERE of {flagged-outcome, drift-outcome} (rank `FAIL_NO_RETRY > SKIP > SUCCESS`). Exit tiers 1/2/3 short-circuit before either policy. **`on_drift` keys on `DriftReport.alarming` (a property of an exit-0 run), never on the exit code** — same "decision layered on a successful run" rule as `on_flagged`. `TaskOutcome` stays 4-valued; no new exit tier; no new error class (config faults reuse `AirflowConfigError` tier 2).
+- **`alarming` = `newly_always_passes` OR `grade_regressions` only (DEC-005).** The OTHER categories (`newly_dropped`/`newly_kept`/`added_artifacts`/`removed_artifacts`/`schema_shape_changes`) are reported but informational — they don't page. A degraded or baseline report is **never** alarming by construction.
+- **Degrade-never-fail (DEC-013).** No prior file **OR a corrupt/unreadable/oversize prior** → `baseline=True` empty report + SUCCESS (the fail-soft `load_diff_report` returns `None` for absent AND malformed alike — a corrupt prior is NOT distinguished from a missing one, so it does NOT set `degrade_reason`); `model_unique_id` mismatch → `degrade_reason` + WARNING + SUCCESS (the ONE `degrade_reason` path, fired inside `compute_drift` after both diffs load); `--no-grade` → `grade_regressions=()`, tier-transition drift still computed. The dedicated operator's ONE hard error is a missing/unreadable CURRENT diff (`AirflowConfigError`). Mirrors the conservative-bias routing posture.
+- **Two surfaces (DEC-001).** (1) `detect_drift_against` flag on `SignalForgeGenerateOperator` (ergonomic: run + compare in one task; drift on the single-model path only — `--select` batch logs an INFO and skips drift); (2) dedicated `SignalForgeDriftOperator` (reads two `diff.json` sidecars downstream, branchable). The dedicated operator has NO `SignalForgeRunResult`, so its task state is driven by a pure `_drift_task_outcome(drift, on_drift)` that mirrors `decide_task_outcome`'s on_drift semantics; the airflow raise stays confined to `raise_for_outcome`.
+- **Templated-path history, persistence reuses the EXISTING writer (DEC-009).** `detect_drift_against="…/{{ macros.ds_add(ds,-1) }}/diff.json"` (prior) + `drift_history_dir="…/{{ ds }}"` (where THIS run persists its diff.json for tomorrow). **Persistence requires BOTH params** — `detect_drift_against` enables the drift step, `drift_history_dir` is the destination; with the latter unset the run computes drift but writes nothing. Persistence reuses `diff._sidecar.write_sidecar` (containment anchored to `drift_history_dir`, which may sit outside `project_dir`) — NO new fail-closed writer, NO new audit class.
+- **The prior-read path is operator-TRUSTED (DEC-008).** `load_diff_report` is symlink-loop-hardened + size-capped (10 MB, mirrors diff's `existing_schema` cap) but NOT project-contained — the prior sidecar legitimately lives outside `project_dir`. Reads are fail-soft (absent/corrupt/oversize → `None`).
+- **Schema-shape = column ADD/REMOVE only (DEC-004 Q4).** Derived from `artifact_id` prefixes — column SET mined from `column.<col>.<field>` (≥3 dotted parts) and `test.column.<col>.<type>` (≥4 parts); **enforce the arity** (a 2-/3-part malformed id must NOT be mined — a corrupt sidecar would otherwise pollute `schema_shape_changes`). Retype is OUT OF SCOPE (sidecars carry no types).
+- **Determinism (DEC-016).** `compute_drift` iterates `sorted` artifact_ids; transition tuples + column lists sorted; carries `previous_diff_hash`/`current_diff_hash` (project `blake2b-8` recipe over `model_dump_json(by_alias=True)` → canonical `json.dumps`). Same two sidecars + same `as_of` → byte-identical report. `DriftReport` is read-back-able → paired `StrictDriftReport(extra="forbid")` drift detector + committed fixture (`tests/fixtures/airflow/drift_report_v1.json`).
+- **XCom hygiene (DEC-015).** `DriftReport.to_xcom()` = per-category counts + transition lists (truncated `why`) + `alarming` + `as_of` + the two input hashes + `schema_version` + `degrade_reason`. No bulk sidecar text, no secrets. On the generate operator it nests under a `"drift"` key; the dedicated operator returns it directly.
+- **`--as-of` reproducibility carve-out** surfaced on both operators (the #171 precedent) so a comparison is reproducible at `(model, as_of)`.
+
+The `SignalForgeDriftOperator` follows the #232/#233 deferred-class pattern verbatim (module `__getattr__` + `find_spec("airflow")` + `functools.cache` factory + airflow-free placeholder + UNGATED skeleton test — required for the codecov patch gate). Example DAG `examples/airflow/signalforge_drift_monitor_dag.py` (`dag_id="signalforge_drift_monitor"`, both forms). Certified vs Airflow 2.10.4 (50 gated tests).
+
 ## `SignalForgeHook` (#234 DEC-001…016)
 
 The Airflow-native credential seam: a `SignalForgeHook(BaseHook)` keyed on a `signalforge_conn_id` so a DAG author configures SignalForge from one Airflow Connection (+ optional Variable) instead of inline per-task env. It **extends the two-layer split** (airflow-free core + shim-confined translator) to credential resolution, and threads `signalforge_conn_id` through BOTH operators. The durable patterns:
@@ -105,4 +123,4 @@ Example DAG: `examples/airflow/signalforge_hook_dag.py` (`dag_id="signalforge_ho
 
 ## Reference
 
-`plans/super/234-signalforge-hook.md` — DEC-001…DEC-016 (`SignalForgeHook`, Connection/Variable → profiles.yml + LLM key). `plans/super/233-prune-existing-operator.md` — DEC-001…DEC-008 (`SignalForgePruneExistingOperator`, the no-LLM sibling). `plans/super/232-generate-operator.md` — DEC-001…DEC-011 (`SignalForgeGenerateOperator`). `plans/super/231-result-task-state.md` — DEC-001…DEC-008. `plans/super/230-airflow-skeleton.md` — skeleton wiring. `docs/airflow-ops.md` — operator-facing contract + example DAGs (incl. the Airflow-native credentials section). `src/signalforge/airflow/{operators,hooks,result,runner,_resolve,_airflow_compat,__init__}.py`, `src/signalforge/__main__.py`, `src/signalforge/llm/providers.py` (`PROVIDER_ENV_VAR_KEYS`). `examples/airflow/{signalforge_generate_operator_dag,signalforge_prune_existing_operator_dag,signalforge_hook_dag}.py`. `tests/airflow/{test_operators,test_operators_helpers,test_hooks,test_resolve,test_dag_parse}.py`. See-Also: `cli-layer.md` (four-tier exit codes, the no-5th-tier rule, the `[airflow]` `errors.py`), `python-build.md` (`[airflow]` extra out of the dev group), `llm-drafter.md` (one-shim-per-vendor), `grade-layer.md`/`warehouse-adapters.md` (fail-soft vs fail-closed posture, `__repr__` redaction).
+`plans/super/235-drift-detection.md` — DEC-001…DEC-019 (run-over-run drift / signal-rot detection: airflow-free `compute_drift` + `DriftReport`, `on_drift` most-severe-wins, both surfaces, templated history). `plans/super/234-signalforge-hook.md` — DEC-001…DEC-016 (`SignalForgeHook`, Connection/Variable → profiles.yml + LLM key). `plans/super/233-prune-existing-operator.md` — DEC-001…DEC-008 (`SignalForgePruneExistingOperator`, the no-LLM sibling). `plans/super/232-generate-operator.md` — DEC-001…DEC-011 (`SignalForgeGenerateOperator`). `plans/super/231-result-task-state.md` — DEC-001…DEC-008. `plans/super/230-airflow-skeleton.md` — skeleton wiring. `docs/airflow-ops.md` — operator-facing contract + example DAGs (incl. the Airflow-native credentials section). `src/signalforge/airflow/{operators,hooks,result,runner,drift,_resolve,_airflow_compat,__init__}.py`, `src/signalforge/__main__.py`, `src/signalforge/llm/providers.py` (`PROVIDER_ENV_VAR_KEYS`). `examples/airflow/{signalforge_generate_operator_dag,signalforge_prune_existing_operator_dag,signalforge_hook_dag,signalforge_drift_monitor_dag}.py`. `tests/airflow/{test_operators,test_operators_helpers,test_hooks,test_resolve,test_dag_parse,test_drift_core,test_drift_loaders,test_drift_report_schema,test_drift_operators,test_result,test_skeleton}.py`. `tests/fixtures/airflow/{drift_report_v1.json,drift_pairs/}`. See-Also: `cli-layer.md` (four-tier exit codes, the no-5th-tier rule, the `[airflow]` `errors.py`), `python-build.md` (`[airflow]` extra out of the dev group), `llm-drafter.md` (one-shim-per-vendor), `grade-layer.md`/`warehouse-adapters.md` (fail-soft vs fail-closed posture, `__repr__` redaction).