#234: Airflow SignalForgeHook — Connection/Variable → profiles.yml + LLM key

.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 (hook, drift — `SignalForgeGenerateOperator` landed in #232, `SignalForgePruneExistingOperator` in #233):** 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 (drift — `SignalForgeGenerateOperator` landed in #232, `SignalForgePruneExistingOperator` in #233, `SignalForgeHook` in #234):** 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)
 
@@ -84,10 +84,25 @@ 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`.
 
+## `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:
+
+- **Airflow-free pure resolver is the heart (`_resolve.py`).** `resolve_connection(conn, *, variable_lookup, project_dir=None) -> HookResolution` is duck-typed (`conn.password` + `conn.extra_dejson`) with an INJECTED `variable_lookup` callable — so it carries NO `from airflow` import and is 100%-covered UNGATED. The gated `SignalForgeHook.get_conn()` is a thin wrapper that calls `self.get_connection(...)` + wires `variable_lookup` to `_airflow_compat.airflow_variable_get`, then delegates. **Mirror this for any future credential/config resolution: put the logic in a duck-typed, callable-injected pure function; keep the airflow-touching wrapper a thin gated shim.**
+- **The resolver is LENIENT; the consumer enforces requiredness (DEC-003).** `HookResolution(profiles_dir, provider, api_key, cache_scope)` (frozen, redacting `__repr__` — never the key or a field-name label) returns `None` for absent fields rather than raising. The **Generate** operator REQUIRES `provider`+`api_key` (raises `AirflowConfigError` else — it makes LLM calls); the **PruneExisting** operator requires NEITHER (DEC-016 — read-only, no LLM call, uses only `profiles_dir`). One lenient resolver serving two consumers with different needs is what avoids a second resolver. `_ConnectionExtra` is `extra="forbid"` + frozen, all-optional (`profiles_dir`/`provider`/`cache_scope`) — a typo key fails loud.
+- **`mask_secret` + `Variable.get` join the one shim (`_airflow_compat`).** `register_secret(value)` (lazy `from airflow.utils.log.secrets_masker import mask_secret` — stable across `apache-airflow>=2.8,<3`) and `airflow_variable_get(key) -> str | None` (coerces `Variable.get`'s `Any` to `str | None` so pyright doesn't widen) are the only new `from airflow` sites — confined to the shim, `# pragma: no cover` + `# type: ignore[import-not-found]`, the import-confinement scan unchanged. `hooks.py` carries NO module-scope airflow import (deferred-class construction, verbatim from #232: `__getattr__` + `functools.cache` + `find_spec("airflow")` + airflow-free `_SignalForgeHookAirflowMissing` placeholder).
+- **Closed `PROVIDER_ENV_VAR_KEYS` allowlist (DEC-005), home in `signalforge.llm.providers`.** `{anthropic: ANTHROPIC_API_KEY, openai: OPENAI_API_KEY, gemini: GOOGLE_API_KEY}` — sibling to `PROVIDER_DEFAULT_MODELS`/`PROVIDER_SKU_PREFIXES`, reusable by the v0.8 GH Action. Validated whenever `provider` is present (regardless of consumer): an unknown provider raises `AirflowConfigError` — the operator never derives an arbitrary env-var name from operator-supplied config (closes the arbitrary-env-var injection vector). `gemini → GOOGLE_API_KEY` (NOT `GEMINI_*`).
+- **The four leak-surface disciplines are each test-pinned (DEC-006/007).** (1) **Logs** — lazy-format `json.dumps` (the grep-gate now scans `signalforge.airflow`, DEC-013) carrying conn_id/provider/`profiles_dir_set` bool, never the key; `register_secret` called BEFORE any log/run. (2) **XCom** — counts + sidecar paths only; key held in a local, never on `self`, never returned. (3) **Rendered templates** — `signalforge_conn_id` is deliberately NOT in either operator's `template_fields`. (4) **`__repr__`** — hook + `HookResolution` redact. `mask_secret` is log-only (does NOT scrub XCom/templates) — the structural disciplines (2)(3)(4) are independent and necessary.
+- **Env injection at the operator seam, restored in `finally` (DEC-014/015).** Generate's `execute()` wraps `run_signalforge` in `_provider_key_env(env_var, api_key)` — a pure (airflow-free, ungated-testable) `os.environ` snapshot→inject→restore context manager (absent-before→delete-after; prior→restore, incl. empty-string; restores on exception). The runner's `_ISOLATED_ENV_KEYS` is UNCHANGED — the key restore belongs at the operator, not the runner. `invocation` default stays `in_process`; docs flag `subprocess` as the safe choice for concurrent multi-task workers (in-process shares `os.environ` + process-global stdout capture). `signalforge_conn_id=None` (default) ⇒ byte-identical to #232/#233.
+- **Shared helpers, reused not duplicated.** `_merge_with_resolution(param, extra)` (precedence param > Connection extra > default, DEC-012), `_resolve_hook(conn_id)`, and `_provider_key_env` are module-level in `operators.py`; PruneExisting reuses the first two for its `profiles_dir`-only path (NOT `_provider_key_env` — no key injection). **Decouple the profiles_dir-precedence piece from the key-injection piece so a no-LLM consumer reuses only what it needs.**
+- **On-disk profiles only for v0.7 (DEC-001); cost ceilings NOT in the Connection `extra` (DEC-011).** The hook resolves a `profiles_dir` → `--profiles-dir` (the existing `load_profile`/`from_profile` seam); synthesizing a `profiles.yml`/`DbtProfileTarget` from an Airflow Connection is deferred (re-deriving the #120 per-type validator is its own ticket). Cost ceilings have no CLI landing strip (#232 DEC-002 deferred `config_overrides`) so they are deliberately absent from `extra` — operators set them via the committed `signalforge.yml grade:` block. `AirflowConfigError` (tier 2) covers all hook misconfig — no new error class, no scan-7 churn.
+
+Example DAG: `examples/airflow/signalforge_hook_dag.py` (`dag_id="signalforge_hook"`, both operators via one `signalforge_conn_id`, no inline per-task env). Pure resolver tests ungated in `tests/airflow/test_resolve.py`; gated hook/operator/dag-parse tests certified against the real `.venv-airflow` rig (airflow 2.10.4).
+
 ## 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/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. `src/signalforge/airflow/{operators,result,runner,_airflow_compat,__init__}.py`, `src/signalforge/__main__.py`. `examples/airflow/{signalforge_generate_operator_dag,signalforge_prune_existing_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).
+`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).

.claude/rules/cli-layer.md

@@ -86,11 +86,11 @@ Every user-supplied path flows through `canonicalise_user_path(raw, project_dir)
 
 When introducing a new flag that takes a path, route it through `canonicalise_user_path` from the orchestrator. Don't trust the writer / loader to derive its own `project_dir`.
 
-## Logger grep gate covers 6 dirs (DEC-019)
+## Logger grep gate covers 11 dirs (DEC-019; `airflow` added by #234)
 
 Every `_LOGGER.{info,warning,debug,error}` call in `signalforge.cli.*` uses lazy-format with `json.dumps()` for any user-controlled string. Never f-string-interpolate — ANSI escapes in a model id or path would inject into log viewers; JSON encoding handles this; f-string interpolation does not.
 
-The grep gate at `tests/llm/test_logger_grep_gate.py` scans `src/signalforge/{llm, draft, prune, grade, diff, cli}` and rejects any `_LOGGER\.\w+\(f"` hit. Extend to a seventh dir only when a new pipeline package ships.
+The grep gate at `tests/llm/test_logger_grep_gate.py` scans the `_SCAN_SUBPACKAGES` set — `src/signalforge/{airflow, cli, demo, diff, draft, grade, llm, manifest, prune, safety, warehouse}` (11 dirs as of #234) — and rejects any `_LOGGER\.\w+\(f"` hit. Add the subpackage to `_SCAN_SUBPACKAGES` when a new package that emits logs ships (`airflow` was added by #234 ahead of the hook's logging).
 
 The CLI is the orchestration layer (NOT a stage-0 reader) so it IS allowed to emit logs. `setup_logging(verbose, quiet)` is the single config site: INFO default, `--verbose` → DEBUG, `--quiet` → WARNING.
 

.claude/rules/diff-renderer.md

@@ -157,7 +157,7 @@ No separate WARNING before raising typed errors — the exception IS the signal
 
 ## ANSI-safe lazy-format JSON logger + grep gate (DEC-019)
 
-Same rule as the other four pipeline layers (`safety-layer.md` DEC-022 / `llm-drafter.md` DEC-011 / `prune-engine.md` DEC-017 / `grade-layer.md` DEC-029). The grep gate at `tests/llm/test_logger_grep_gate.py` scans `src/signalforge/{llm,draft,prune,grade,diff,cli}` (6 dirs as of #9) and rejects any `_LOGGER\.\w+\(f"` hit.
+Same rule as the other four pipeline layers (`safety-layer.md` DEC-022 / `llm-drafter.md` DEC-011 / `prune-engine.md` DEC-017 / `grade-layer.md` DEC-029). The grep gate at `tests/llm/test_logger_grep_gate.py` scans the `_SCAN_SUBPACKAGES` set — `src/signalforge/{airflow, cli, demo, diff, draft, grade, llm, manifest, prune, safety, warehouse}` (11 dirs as of #234) — and rejects any `_LOGGER\.\w+\(f"` hit.
 
 ## Fail-closed writer AST defence (DEC-018)