#232: Airflow SignalForgeGenerateOperator - #241
Merged
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)
📝 WalkthroughWalkthroughReplaces the ChangesSignalForgeGenerateOperator Full Implementation
Sequence Diagram(s)sequenceDiagram
participant DAG as Airflow DAG
participant Op as SignalForgeGenerateOperator.execute()
participant Validate as _validate_operator_config
participant Resolve as _resolve_select_models
participant Build as _build_generate_argv
participant RunSF as run_signalforge
participant Agg as _aggregate_batch_result
participant Outcome as decide_task_outcome / raise_for_outcome
DAG->>Op: execute(context)
Op->>Validate: project_dir, model, select, on_flagged
Validate-->>Op: ok or AirflowConfigError
alt --select batch
Op->>Resolve: project_dir, select expr
Resolve-->>Op: sorted tuple of unique_ids
loop per unique_id
Op->>Build: model=unique_id, write, flags
Build-->>Op: argv
Op->>RunSF: argv
RunSF-->>Op: SignalForgeRunResult
end
Op->>Agg: sequence of results
Agg-->>Op: aggregated SignalForgeRunResult
else single model
Op->>Build: model, write, flags
Build-->>Op: argv
Op->>RunSF: argv
RunSF-->>Op: SignalForgeRunResult
end
Op->>Outcome: result, on_flagged
Outcome-->>Op: AirflowSkipException / AirflowFailException / XCom dict
Op-->>DAG: XCom payload or raises exception
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Suggested labels
🚥 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 |
…fig-validation helpers (#232 US-001) Add two pure, airflow-free module-level helpers to signalforge.airflow.operators: - _build_generate_argv(...) maps operator params to a 'signalforge generate' CLI argv list: positional <model> XOR --select <expr>, always --project-dir + --format json, write=False->--dry-run / write=True->--write (DEC-003), plus optional --no-grade / --as-of / --cache-scope / --profiles-dir. - _validate_operator_config(...) raises AirflowConfigError (DEC-009) on empty project_dir, model/select mutex violations, leading-dash argv-injection, and on_flagged outside {fail,skip,succeed}. No 'from airflow' import (confinement scan stays green); no I/O. The NotImplementedError stub class is unchanged (US-003 owns the real operator). New ungated test file covers every branch.
…_result helpers (#232 US-002) Add two airflow-free pure helpers to signalforge.airflow.operators: - _resolve_select_models(project_dir, select) -> tuple[str, ...]: loads the dbt manifest and resolves a --select expression to sorted unique_ids, mapping ManifestError / SelectorParseError / zero-match to AirflowConfigError (carrying the source .remediation, chaining via from exc) (DEC-001/006). - _aggregate_batch_result(results) -> SignalForgeRunResult: rolls up per-model batch results into one aggregate (max exit_code, summed counts, unioned model_unique_ids, mean of non-None grades, summed non-None durations, None sidecar paths, empty stdout/stderr) (DEC-008). Extend tests/airflow/test_operators_helpers.py (ungated) covering every branch: selector resolution against the dbt_project_multi fixture (valid, multi-atom union, parse-error, zero-match, missing/invalid project_dir) and batch aggregation (max-exit, count sums, mean-grade with/without None, duration sums, single-result, cleared sidecars, empty-input ValueError).
US-003) Replace the #230 stub with the real generate operator via deferred class construction: a module-level PEP 562 __getattr__ resolves the SignalForgeGenerateOperator name through a find_spec-guarded factory. - Airflow present: the cached factory builds the real BaseOperator subclass (base obtained via the one shim make_base_operator(); the from-airflow import stays confined there, out of module scope). - Airflow absent: the name resolves WITHOUT importing airflow (find_spec does not execute the module) to an airflow-free placeholder whose construction raises ModuleNotFoundError. Attribute access stays airflow-free so the no-eager-import / import-confinement gates keep passing; only construction of the real operator needs airflow. execute() dispatches single-model (build argv -> run_signalforge -> decide_task_outcome -> raise_for_outcome -> to_xcom) vs --select batch (resolve selector to model ids, loop once per model, force --cache-scope project for a >=2-model batch when unset, aggregate, return {"models", "aggregate"}). __init__ + execute both run _validate_operator_config (DAG-parse fail-fast + rendered-value dash guard). template_fields per DEC-005. Update the skeleton test to assert airflow-free access + construction-raises under no-airflow; add gated tests/airflow/test_operators.py (pytest.mark.airflow + in-test importorskip) covering the decision table, argv shape, batch loop + forced project cache scope, and construction-time validation. Default suite green (3920 passed, 123 deselected); operators.py 100% covered (gated factory + real class body pragma'd); no-eager-import + import-confinement gates pass.
…ender-template test
Add examples/airflow/signalforge_generate_operator_dag.py — a single-task
drift monitor (dag_id signalforge_generate_operator, task_id drift_monitor)
built with SignalForgeGenerateOperator directly. write=False (dry-run default),
project_dir/select/on_flagged sourced from env-then-Variable with safe parse-time
fallbacks. Demonstrates template_fields: select={{ params.select }} and
as_of={{ ds }}.
Extend tests/airflow/test_dag_parse.py (gated, airflow marker + per-test
importorskip): parametrize _load_example_dag by dag_id; add a parse test for the
new DAG (import_errors == {}, task_ids == {drift_monitor}) and a
render_template_fields test asserting select/as_of render the injected context.
Existing PythonOperator DAG test untouched.
…rtion) Add a SignalForgeGenerateOperator section to docs/airflow-ops.md covering the param->CLI-flag table (with template_fields), write=False=--dry-run (DEC-003), on_flagged branches (DEC-008), the invocation concurrency caveat (DEC-004), cost/time guardrails via committed signalforge.yml (DEC-002), --select batch per-model XCom + aggregate task-state (DEC-001/008/010), the concurrent project_dir sidecar caveat, and a minimal usage example. Refine the --select batch section to distinguish the raw seam (last-writer-wins) from the operator (loops per model). Update the status note + caveats to reflect the operator has landed. mkdocs nav already lists airflow-ops.md (no nav change).
…-integration.md (rule portion)
…lass, manifest-id guard comment, 1-model batch test Review findings fixed: - functools.cache replaces manual global+TOCTOU caching of the deferred operator class - comment: manifest-resolved unique_ids are trusted (no leading-dash guard needed) - new gated test: --select resolving to exactly 1 model (no cache-scope force, batch XCom shape) Certified: 22 passed against airflow 2.10.4.
wjduenow
changed the title
wjduenow
marked this pull request as ready for review
June 16, 2026 05:58
Owner
Author
Implementation landed (Ralph run)All 7 stories from the plan shipped on this branch:
Validation
Decisions worth a look (vetoable)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
docs/airflow-ops.md (1)
8-16: ⚡ Quick winKeep
docs/airflow-ops.mdfocused on current behavior, not shipped-status narration.The “has landed / roadmap / epic-issue” phrasing reads like release-history narrative. Move that historical framing to
CHANGELOG.md/plans/super/*/.claude/rules/*, and keep this page strictly operational.As per coding guidelines: “Do not re-narrate shipped work; those three places own the historical record.”
Also applies to: 408-411
🤖 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 `@docs/airflow-ops.md` around lines 8 - 16, Remove the release-history narrative from the operations documentation in docs/airflow-ops.md. Specifically, eliminate the phrases "has landed (epic `#228`, issue `#232`)", "remains on the roadmap", and other shipped-status descriptions that narrate development history rather than document current behavior. Instead, restructure the content to focus purely on what the operators (SignalForgeGenerateOperator and related helpers) do operationally: describe their contracts, parameters, and usage patterns. Move any historical context about what has shipped, what is planned, or which features landed in which version to CHANGELOG.md, plans/super/*, or .claude/rules/* where that narrative belongs. Apply this same refactoring to lines 408-411 where similar roadmap/shipped-status language appears.Source: Coding guidelines
🤖 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 `@docs/airflow-ops.md`:
- Line 199: The internal anchor links in the markdown file are using
automatically generated slugs that don't resolve reliably, breaking in-page
navigation. Fix the broken anchor references at lines 199, 255, 269, and 340 by
adding explicit, stable HTML id attributes (using markdown syntax like
`{`#custom-id`}` or HTML `<a id="custom-id"></a>`) to the target section headers,
then update the corresponding link fragments to reference these explicit IDs
instead of relying on auto-generated slugs. This ensures that intra-document
navigation works consistently regardless of header text variations.
In `@src/signalforge/airflow/operators.py`:
- Around line 445-464: In the batch loop where run_signalforge is called for
each model_id, the results being collected may contain per-model sidecar path
fields (diff_sidecar_path, grade_sidecar_path) that are misleading because in
write mode all models write to the same fixed project paths rather than stable
per-model locations. Review the SignalForgeRunResult type and the results being
appended to the results list, then remove or exclude the per-model sidecar path
fields from the returned data structure to avoid implying that each model has
its own stable artifact at those paths. The sidecar paths should not be returned
as per-model data when they are actually shared/overwritten across models.
- Around line 148-152: The validation logic in the mutex check at lines 148-152
only verifies that model and select are not None, but does not reject empty
strings or whitespace-only values, allowing blank values to bypass validation
and be passed downstream. Update the condition to also check that the values are
not empty or whitespace by using appropriate string validation methods (such as
checking if the string is falsy after stripping whitespace) alongside the None
checks, and raise AirflowConfigError for blank/whitespace values. Apply the same
validation fix to the related validation code at lines 154-163 to ensure
consistent rejection of blank model or select values across all validation
points.
In `@tests/airflow/test_skeleton.py`:
- Around line 83-100: The assertion in
test_generate_operator_access_is_airflow_free_and_construction_requires_airflow
that checks sys.modules is not deterministic because it inspects the global
state without first normalizing prior airflow imports. Before the assertion that
validates no airflow modules are loaded, remove any pre-existing airflow-related
modules from sys.modules to ensure a clean slate for the test. This way the
assertion will only reflect the behavior of the lazy import mechanism itself,
not prior test pollution or plugin side effects.
---
Nitpick comments:
In `@docs/airflow-ops.md`:
- Around line 8-16: Remove the release-history narrative from the operations
documentation in docs/airflow-ops.md. Specifically, eliminate the phrases "has
landed (epic `#228`, issue `#232`)", "remains on the roadmap", and other
shipped-status descriptions that narrate development history rather than
document current behavior. Instead, restructure the content to focus purely on
what the operators (SignalForgeGenerateOperator and related helpers) do
operationally: describe their contracts, parameters, and usage patterns. Move
any historical context about what has shipped, what is planned, or which
features landed in which version to CHANGELOG.md, plans/super/*, or
.claude/rules/* where that narrative belongs. Apply this same refactoring to
lines 408-411 where similar roadmap/shipped-status language appears.
🪄 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: 2854db83-6690-46da-9773-f4d9e3bcb13e
📒 Files selected for processing (9)
.claude/rules/airflow-integration.mddocs/airflow-ops.mdexamples/airflow/signalforge_generate_operator_dag.pyplans/super/232-generate-operator.mdsrc/signalforge/airflow/operators.pytests/airflow/test_dag_parse.pytests/airflow/test_operators.pytests/airflow/test_operators_helpers.pytests/airflow/test_skeleton.py
There was a problem hiding this comment.
Pull request overview
This PR implements the first real Apache Airflow operator integration for SignalForge by replacing the prior skeleton SignalForgeGenerateOperator with a lazily-resolved BaseOperator subclass that can run signalforge generate for a single model or a --select batch, while preserving the “no eager Airflow import” contract.
Changes:
- Implemented
SignalForgeGenerateOperatorwith airflow-free helper functions and a gated Airflow-dependentexecute()path (including batch looping + aggregation). - Added ungated helper unit tests plus gated operator execution and DAG-parse/render tests.
- Added an operator-based example DAG and updated Airflow ops documentation and integration rules to reflect the new recommended surface.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
src/signalforge/airflow/operators.py |
Adds airflow-free argv/config/select/aggregation helpers plus lazy operator class construction and execution wiring. |
tests/airflow/test_skeleton.py |
Updates the skeleton test to validate airflow-free attribute access and Airflow-required construction behavior. |
tests/airflow/test_operators.py |
New gated tests covering operator execute() behavior, argv construction, outcome translation, and batch semantics. |
tests/airflow/test_operators_helpers.py |
New ungated tests covering pure helper functions for argv building, validation, selector resolution, and batch aggregation. |
tests/airflow/test_dag_parse.py |
Extends gated DAG-parse tests to include the operator-based example and template rendering assertions. |
examples/airflow/signalforge_generate_operator_dag.py |
Adds a single-task operator-first drift monitor example DAG. |
docs/airflow-ops.md |
Documents the new operator, parameter→flag mappings, dry-run semantics, and batch/XCom behavior. |
.claude/rules/airflow-integration.md |
Updates integration rules to include the shipped operator patterns and constraints. |
plans/super/232-generate-operator.md |
Adds the #232 plan document describing decisions and implementation structure. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- _validate_operator_config: reject blank/non-str model/select before the mutex (model="" no longer bypasses 'exactly one' and emits an empty positional) - _build_generate_argv: treat empty cache_scope as absent (no '--cache-scope ""') - batch XCom: null per-model sidecar paths (unstable under shared O_TRUNC sidecar) - test_skeleton: snapshot sys.modules before/after for a deterministic no-eager assertion - docs: de-link fragile intra-page anchors (duplicate headings); clarify flagged-exits-0 only under grade.fail_on_below_threshold=false + ungated tests for the new branches; certified 22 passed vs airflow 2.10.4
Owner
Author
PR Review SummaryAll 8 review threads addressed in Fixed (7 items)
New ungated tests cover the blank/non-str/empty- Documented (1 item)
|
wjduenow
added a commit
that referenced
this pull request
Jun 16, 2026
This was referenced Jun 16, 2026
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.
Summary
Implements #232 —
SignalForgeGenerateOperator(epic #228, builds on #231). Plan:plans/super/232-generate-operator.md(Phase: devolved → implemented).All 7 plan stories landed + Quality Gate + a PR-review pass (CodeRabbit + Copilot). See the "Implementation landed" and "PR Review Summary" comments for detail.
Changes
src/signalforge/airflow/operators.py— real operator (deferred construction; airflow-free import, airflow-required construction), pure helpers,execute()wiring.examples/airflow/signalforge_generate_operator_dag.py— operator example DAG.tests/airflow/— ungated pure-helper tests (100% covered) + gatedexecute()/DAG-parse/render tests.docs/airflow-ops.md,.claude/rules/airflow-integration.md— operator docs + conventions.Testing
3927 passed, ruff/format/pyright clean,operators.py100% covered..venv-airflow):22 passed, 1 skipped(live-warehouse e2e needsSF_RUN_BQ).Compounding update
.claude/rules/airflow-integration.md§ "SignalForgeGenerateOperator" + memorysignalforge-airflow-operator-pattern.Summary by CodeRabbit
New Features
SignalForgeGenerateOperatorfor Airflow drift monitoring, supporting single-model runs and--selectbatch workflows with per-model execution and aggregated outcomes.on_flaggedtask behavior to control fail/skip/succeed based on flagged artifacts, plus read-only “dry-run” semantics for write prevention.Documentation
--select/aggregation semantics, and updated caveats.Tests