#235: Airflow drift / signal-rot detection (run-over-run) - #245
Merged
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
✅ Files skipped from review due to trivial changes (1)
🚧 Files skipped from review as they are similar to previous changes (4)
📝 WalkthroughWalkthroughAdds run-over-run drift/signal-rot detection to SignalForge Airflow integration via a pure Pydantic drift core ( ChangesDrift / Signal-Rot Detection Feature
Sequence Diagram(s)sequenceDiagram
participant DAG as Airflow DAG
participant GenOp as SignalForgeGenerateOperator
participant DriftOp as SignalForgeDriftOperator
participant Core as compute_drift (drift.py)
participant FS as Filesystem (diff.json sidecars)
participant Outcome as decide_task_outcome
rect rgba(100, 149, 237, 0.5)
Note over DAG,Outcome: Ergonomic single-operator flow
DAG->>GenOp: execute(context)
GenOp->>FS: run signalforge CLI, parse stdout → current diff
GenOp->>FS: load_diff_report(detect_drift_against) → prior diff
GenOp->>Core: compute_drift(prior, current, grades)
Core-->>GenOp: DriftReport
GenOp->>FS: persist diff.json/grade.json to drift_history_dir
GenOp->>Outcome: decide_task_outcome(result, on_drift, drift)
Outcome-->>DAG: SUCCESS / AirflowSkipException / AirflowFailException
end
rect rgba(144, 238, 144, 0.5)
Note over DAG,Outcome: Two-task generate + dedicated drift gate
DAG->>GenOp: execute(context), on_drift="succeed"
GenOp->>FS: persist diff.json sidecar
GenOp-->>DAG: XCom with drift nested under "drift" key
DAG->>DriftOp: execute(context)
DriftOp->>FS: load_diff_report(previous_diff_path)
DriftOp->>FS: load_diff_report(current_diff_path)
DriftOp->>Core: compute_drift(prior, current)
Core-->>DriftOp: DriftReport
DriftOp->>Outcome: _drift_task_outcome(report, on_drift)
Outcome-->>DAG: SUCCESS / AirflowSkipException / AirflowFailException
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related issues
Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
Comment |
There was a problem hiding this comment.
Pull request overview
Adds run-over-run drift detection to the signalforge.airflow integration so scheduled DAG runs can compare run N vs N-1 and emit a structured drift report (signal-rot + grade-regression paging, plus informational tier/schema-shape deltas). This extends the existing Airflow result→task-state contract to incorporate an optional drift verdict while keeping the drift core Airflow-free and reusable.
Changes:
- Introduces an Airflow-free drift core (
compute_drift -> DriftReport) with fail-soft sidecar/stdout loaders and XCom-safe serialization. - Adds two Airflow surfaces: drift-on-generate (
detect_drift_against/drift_history_dir/on_drift) and a dedicatedSignalForgeDriftOperator. - Adds extensive ungated + gated tests, committed fixtures, an example DAG, and docs/rules updates.
Reviewed changes
Copilot reviewed 22 out of 22 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/signalforge/airflow/drift.py | New Airflow-free drift core (DriftReport, compute_drift) + fail-soft loaders. |
| src/signalforge/airflow/operators.py | Wires drift into SignalForgeGenerateOperator and adds deferred SignalForgeDriftOperator + pure helpers. |
| src/signalforge/airflow/result.py | Extends decide_task_outcome with optional drift policy (on_drift) and most-severe-wins combine. |
| src/signalforge/airflow/init.py | Eagerly re-exports drift core; adds lazy export for the new drift operator. |
| tests/airflow/test_drift_core.py | Ungated unit tests for drift classification, determinism, alarming semantics, and to_xcom. |
| tests/airflow/test_drift_loaders.py | Ungated tests pinning fail-soft loader behavior (absent/corrupt/oversize/symlink-loop). |
| tests/airflow/test_drift_report_schema.py | Ungated strict-schema drift detector + fixture validation for DriftReport. |
| tests/airflow/test_drift_operators.py | Gated Airflow execute() tests for both drift surfaces and outcome mapping. |
| tests/airflow/test_operators_helpers.py | Ungated tests for new pure operator helpers (build_drift_report, drift validation/path helpers, outcomes). |
| tests/airflow/test_result.py | Extends result/outcome tests to cover drift policy interactions. |
| tests/airflow/test_skeleton.py | Ungated skeleton test ensuring drift operator name is resolvable without importing Airflow. |
| tests/airflow/test_airflow_no_eager_import.py | Extends no-eager-import test to include the drift operator. |
| tests/airflow/test_dag_parse.py | Updates DagBag parse coverage to include the new example drift-monitor DAG and templating. |
| tests/fixtures/airflow/drift_report_v1.json | Committed fixture for DriftReport schema-stability detector. |
| tests/fixtures/airflow/drift_pairs/signal_rot_prev_diff.json | Committed engineered “prev” diff fixture for signal-rot/grade-regression comparisons. |
| tests/fixtures/airflow/drift_pairs/signal_rot_curr_diff.json | Committed engineered “curr” diff fixture for signal-rot/grade-regression comparisons. |
| tests/fixtures/airflow/drift_pairs/signal_rot_prev_grade.json | Committed engineered “prev” grade fixture for regression comparisons. |
| tests/fixtures/airflow/drift_pairs/signal_rot_curr_grade.json | Committed engineered “curr” grade fixture for regression comparisons. |
| examples/airflow/signalforge_drift_monitor_dag.py | New example DAG demonstrating both drift surfaces (ergonomic + branchable). |
| docs/airflow-ops.md | Adds operator-facing documentation for drift detection, policies, and example usage. |
| .claude/rules/airflow-integration.md | Captures integration patterns/decisions for drift detection. |
| plans/super/235-drift-detection.md | Detailed implementation plan / DEC log for #235. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…ction # Conflicts: # .claude/rules/airflow-integration.md # docs/airflow-ops.md # src/signalforge/airflow/operators.py # tests/airflow/test_dag_parse.py # tests/airflow/test_operators_helpers.py # tests/airflow/test_skeleton.py
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/signalforge/airflow/operators.py (1)
264-343:⚠️ Potential issue | 🟠 Major | ⚡ Quick winValidate
grade_regression_thresholdbefore computing drift.
grade_regression_thresholdis accepted from both operator surfaces but never validated. Invalid values (negative/NaN/inf) can silently misclassify regressions and produce incorrect alarming behavior.Proposed fix
@@ +import math @@ +def _validate_grade_regression_threshold(value: float) -> None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise AirflowConfigError( + "`grade_regression_threshold` must be a finite number >= 0." + ) + numeric = float(value) + if not math.isfinite(numeric) or numeric < 0: + raise AirflowConfigError( + f"`grade_regression_threshold` must be a finite number >= 0 (got {value!r})." + ) @@ def _validate_operator_config( @@ - drift_history_dir: str | None = None, + drift_history_dir: str | None = None, + grade_regression_threshold: float = 0.05, ) -> None: @@ + _validate_grade_regression_threshold(grade_regression_threshold) @@ def _validate_drift_config( @@ - on_drift: str, + on_drift: str, + grade_regression_threshold: float, ) -> None: @@ + _validate_grade_regression_threshold(grade_regression_threshold)Also applies to: 627-669, 877-880, 1524-1527
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/signalforge/airflow/operators.py` around lines 264 - 343, The parameter `grade_regression_threshold` is accepted from operator surfaces but lacks validation, allowing invalid values like negative numbers, NaN, or infinity to silently cause incorrect regression classification and alarming behavior. Add `grade_regression_threshold` as a parameter to the `_validate_operator_config` function signature, then add validation logic that ensures the value is either None/unset or is a valid positive number (check that it is a numeric type, is not negative, and is not NaN or infinity). Apply the same validation pattern at all other locations where this parameter is accepted from operator configurations, mirroring the validation approach used for other numeric or optional parameters in the function.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@examples/airflow/signalforge_drift_monitor_dag.py`:
- Around line 178-195: The `SignalForgeGenerateOperator` task with
`task_id="generate"` currently only sets `on_drift="succeed"` but does not
explicitly set `on_flagged`, which means it inherits the default
`on_flagged="fail"`. In this branchable pattern where generate should never fail
and instead let the downstream dedicated operator handle the gating, add
`on_flagged="succeed"` to the operator configuration to ensure flagged runs also
succeed rather than blocking downstream execution of the `drift_check` task.
In `@src/signalforge/airflow/operators.py`:
- Around line 334-335: The whitespace-only string handling is inconsistent
across multiple locations in the file. The validation logic in
_validate_operator_config (lines 334-335) correctly treats whitespace-only
strings as empty by stripping them, but the runtime checks that use
self.detect_drift_against and self.drift_history_dir at lines 902-904,
1024-1025, and 1144-1145 treat whitespace-only strings as truthy/enabled. Apply
the same whitespace-stripping check at each of these runtime locations: update
the conditions to check not just if the attribute exists, but also verify it is
not None and not just whitespace (using the same pattern as the validation:
isinstance check and .strip() call). This ensures whitespace-only paths are
consistently treated as disabled everywhere.
---
Outside diff comments:
In `@src/signalforge/airflow/operators.py`:
- Around line 264-343: The parameter `grade_regression_threshold` is accepted
from operator surfaces but lacks validation, allowing invalid values like
negative numbers, NaN, or infinity to silently cause incorrect regression
classification and alarming behavior. Add `grade_regression_threshold` as a
parameter to the `_validate_operator_config` function signature, then add
validation logic that ensures the value is either None/unset or is a valid
positive number (check that it is a numeric type, is not negative, and is not
NaN or infinity). Apply the same validation pattern at all other locations where
this parameter is accepted from operator configurations, mirroring the
validation approach used for other numeric or optional parameters in the
function.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a523b04b-4603-4e02-86d3-10c74d1cad09
📒 Files selected for processing (22)
.claude/rules/airflow-integration.mddocs/airflow-ops.mdexamples/airflow/signalforge_drift_monitor_dag.pyplans/super/235-drift-detection.mdsrc/signalforge/airflow/__init__.pysrc/signalforge/airflow/drift.pysrc/signalforge/airflow/operators.pysrc/signalforge/airflow/result.pytests/airflow/test_airflow_no_eager_import.pytests/airflow/test_dag_parse.pytests/airflow/test_drift_core.pytests/airflow/test_drift_loaders.pytests/airflow/test_drift_operators.pytests/airflow/test_drift_report_schema.pytests/airflow/test_operators_helpers.pytests/airflow/test_result.pytests/airflow/test_skeleton.pytests/fixtures/airflow/drift_pairs/signal_rot_curr_diff.jsontests/fixtures/airflow/drift_pairs/signal_rot_curr_grade.jsontests/fixtures/airflow/drift_pairs/signal_rot_prev_diff.jsontests/fixtures/airflow/drift_pairs/signal_rot_prev_grade.jsontests/fixtures/airflow/drift_report_v1.json
…rmalization, degrade docs, branchable example on_flagged)
Owner
Author
PR Review SummaryAll 7 review threads addressed — 7 fixed, 0 false positives. Pushed in Fixed (7 items)
Tests6 new ungated tests in Validation
|
This was referenced Jun 17, 2026
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #235. Part of epic #228 (v0.7 Airflow); builds on #231 (result/XCom contract) and #232 (
SignalForgeGenerateOperator).What
Run-over-run drift detection: compare a SignalForge run (N) against its prior run (N-1) and emit a structured drift report — the headline scheduled value-add that turns a SignalForge DAG into a schema-drift / signal-rot monitor.
keptand is nowdropped: always-passes(the alarm).newly_dropped/newly_kept), schema-shape (column add/remove), all reported; only signal-rot + grade-regression page.Shape
signalforge.airflow.drift—compute_drift(...) -> DriftReport+ fail-soft loaders; 100% ungated coverage; reusable by the v0.8 GitHub Action.detect_drift_againstflag onSignalForgeGenerateOperator(run + compare in one task) and a dedicatedSignalForgeDriftOperator(reads two sidecars downstream, branchable).on_drift(fail/skip/succeed) is the run-over-run analogue ofon_flagged— combined via most-severe-wins; byte-identical when no drift is passed (noTaskOutcome/exit-tier changes, no new error class).detect_drift_against+drift_history_dirwith{{ ds }}/{{ prev_ds }}); persistence reuses the existing fail-closedwrite_sidecar(no new writer, no new audit class).--no-grade. Deterministic (sorted iteration + twoblake2b-8input hashes), reproducible at(model, as_of).Tests & certification
decide_task_outcomeextension / deferred-operator skeleton — ungated;StrictDriftReportschema-stability drift detector + committed fixture.@pytest.mark.airflowexecute() tests (both surfaces) + example DAG (signalforge_drift_monitor_dag.py) + DagBag-parse.artifact_idarity bug (+ regression test) and two docs-accuracy items.Docs
docs/airflow-ops.md— new "Drift / signal-rot detection" section (ticket A8). Working conventions captured in.claude/rules/airflow-integration.md.Plan:
plans/super/235-drift-detection.md(19 DECs). Devolved + built via beads epic (8 stories).Summary by CodeRabbit
New Features
on_drift).Documentation
Tests & Examples