#232: Airflow SignalForgeGenerateOperator

.claude/rules/airflow-integration.md

@@ -13,7 +13,7 @@ The integration splits cleanly so the "act on the graded diff" logic is unit-tes
   - `run_signalforge(argv, *, project_dir, invocation="in_process"|"subprocess", timeout_seconds=None) -> SignalForgeRunResult` — builds/normalises argv, runs the pipeline, parses the result. Does NOT compute the outcome (the operator calls `decide_task_outcome`).
 - **Shim-confined translator** — `_airflow_compat.raise_for_outcome(outcome, *, message)`. The ONLY new `from airflow.exceptions import ...` site (lazy, inside the body, `# type: ignore[import-not-found]`, `# pragma: no cover`), per the one-shim-per-vendor rule (`llm-drafter.md` §"One SDK seam"). Maps `FAIL_NO_RETRY`→`AirflowFailException` (no retry), `SKIP`→`AirflowSkipException`, `FAIL_RETRYABLE`→`AirflowException` (retryable — Airflow's `retries`/`retry_delay` apply), `SUCCESS`→return. Deliberately NOT re-exported from the package top — the operator calls it via `_airflow_compat`.
 
-**Rule for the remaining epic-#228 children (operators, hook, drift):** put the decision in the pure core, the airflow-exception raise in the shim. Never re-derive the outcome by string-matching; carry the typed `TaskOutcome`. Never grow a fifth exit tier.
+**Rule for the remaining epic-#228 children (hook, drift — `SignalForgeGenerateOperator` landed in #232):** put the decision in the pure core, the airflow-exception raise in the shim. Never re-derive the outcome by string-matching; carry the typed `TaskOutcome`. Never grow a fifth exit tier.
 
 ## Exit → TaskOutcome → Airflow (the #231 contract table)
 
@@ -60,10 +60,21 @@ The diff-stdout parse and the `grade.json` / sidecar-path reads are **fail-SOFT*
 - Airflow-touching tests (`raise_for_outcome`, DAG-parse) carry `@pytest.mark.airflow` + **in-test** `pytest.importorskip("airflow")` (NOT module-scope) and are deselected by default — mirrors `tests/airflow/test_dag_parse.py` and the #229 gating convention ([[signalforge-airflow-local-e2e]]).
 - **Certify the airflow-touching paths against the real `.venv-airflow` rig** (#229) before closing — workers can't (airflow isn't in the feature worktree). Non-destructive: `SF_RUN_AIRFLOW=1 PYTHONPATH="$PWD/src" /path/to/.venv-airflow/bin/python -m pytest tests/airflow -m airflow --no-cov` (PYTHONPATH-shadow, no editable reinstall — avoids repointing the rig per [[ralph-editable-install-race]]). #231's translator + refactored example DAG were certified this way (6 passed vs airflow 2.10.4).
 
+## `SignalForgeGenerateOperator` (#232 DEC-001…011)
+
+The first real operator (replaces the #230 stub). It wraps `signalforge generate` (single model + `--select` batch). The durable patterns the next operator/hook child must mirror:
+
+- **Pure/gated structural split (DEC-011) — the codecov-patch-gate enabler.** `execute()` requires airflow (it subclasses `BaseOperator`), so it is gated and outside default coverage. ALL decision logic lives in airflow-free PURE module-level helpers tested UNGATED: `_build_generate_argv` (params→CLI argv), `_validate_operator_config` (DEC-009 guards), `_resolve_select_models` (manifest load + `select_models`, mapping `ManifestError`/`SelectorParseError`/zero-match → `AirflowConfigError`), `_aggregate_batch_result` (DEC-008). `execute()` is a thin wire-up marked `# pragma: no cover`. **A new operator puts ~all its logic in ungated pure helpers; only the airflow-touching wire-up is gated.**
+- **Deferred class construction keeps the module airflow-free (DEC-011 mechanics).** `operators.py` must NOT do module-scope `class X(make_base_operator())` — the ungated helper tests import from this module and the no-eager-import + confinement gates run ungated, so a module-scope subclass would eagerly require airflow. Instead a module-level PEP 562 `__getattr__` resolves `SignalForgeGenerateOperator` on access; resolution branches on `importlib.util.find_spec("airflow")` (which imports nothing) — airflow present → cached factory builds the real `BaseOperator` subclass via `make_base_operator()`; airflow absent → an airflow-free placeholder whose `__init__` raises `ModuleNotFoundError`. Net: `import signalforge.airflow` and `from signalforge.airflow.operators import <helper>` stay airflow-free; constructing the operator without airflow fails loud. No `from airflow` line in `operators.py`; the base comes from `make_base_operator()`, the raise from `_airflow_compat.raise_for_outcome`.
+- **`write=False` → `--dry-run` (DEC-003).** The safe scheduled default writes nothing; XCom is built from stdout JSON; sidecar paths are `None`. `write=True` → `--write` and sidecar paths populate. Concurrent same-`project_dir` runs collide on sidecars (O_TRUNC last-writer-wins) only under `write=True` — dry-run writes nothing so there's no collision.
+- **`--select` batch: operator loops per model (DEC-001/007/008/010).** The raw `run_signalforge(--select)` seam reflects only the LAST model's sidecar; the operator overcomes that by resolving the selector itself and calling `run_signalforge` once per matched model. It forces `--cache-scope project` when ≥2 models match (DEC-007) so Anthropic's *server-side* prompt cache amortises the byte-identical project prefix across the separate in-process calls within TTL. XCom = `{"models": [per-model to_xcom()…], "aggregate": aggregate.to_xcom()}`; the single Airflow task state is driven by the aggregate (`exit_code = max` over per-model, then `decide_task_outcome` once). The N manifest reloads (one per looped call) are <1% overhead vs per-model pipeline cost.
+- **`config_overrides` deferred (DEC-002).** The grade cost/time ceilings are `signalforge.yml grade:` knobs with no CLI-flag landing strip and no `--config` override; v0.7 surfaces them via the committed `signalforge.yml`, not an operator param. A future `--config` flag or per-run config overlay would re-open this.
+- **`invocation` param defaults to `in_process` (DEC-004)**; `subprocess` is the clean-isolation choice for concurrent workers (in-process `redirect_stdout` is process-global). No new error class, no exit-code-table change — `AirflowConfigError` (tier 2) already covers misconfiguration. The existing import-confinement scan already gates `operators.py`; no new AST scan.
+
 ## v0.8 note
 
 `run_signalforge` is airflow-free and meant for the v0.8 GitHub Action too. If/when that lands, consider hoisting `result.py`/`runner.py` to a neutral package (e.g. `signalforge.automation`) so the GH Action doesn't import from a package named `airflow`. Out of scope for v0.7 — the modules are airflow-free so `import signalforge.airflow.runner` works without airflow today.
 
 ## Reference
 
-`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 DAG. `src/signalforge/airflow/{result,runner,_airflow_compat,__init__}.py`, `src/signalforge/__main__.py`. `tests/airflow/`. 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).
+`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. `src/signalforge/airflow/{operators,result,runner,_airflow_compat,__init__}.py`, `src/signalforge/__main__.py`. `examples/airflow/signalforge_generate_operator_dag.py`. `tests/airflow/{test_operators,test_operators_helpers,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).