databricks-solutions · matthewmoorcroft · Sep 1, 2026 · Sep 1, 2026
diff --git a/AGENTS.md b/AGENTS.md
@@ -17,7 +17,7 @@ bootstrap a self-contained virtual environment with pip via the `setup` skill or
 bash scripts/bootstrap.sh   # creates the venv, pip-installs requirements.txt, writes .migration-venv
 # then run plugin code with src/ on PYTHONPATH, using the interpreter from the marker file:
 PY="$(cat .migration-venv)"
-PYTHONPATH=src "$PY" -m flowx.adapter inputs discover
+PYTHONPATH=src "$PY" -m flowx.adapter inputs discover --source adf   # or --source airflow
 ```
 
 `bootstrap.sh` creates the venv at `/Workspace/Users/<current user>/.migration-skills` when running

diff --git a/Makefile b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: clean dev ci test integration fmt help docs-install docs-clean docs-build docs-serve lock-dependencies requirements precommit
+.PHONY: clean dev ci test integration integration-live fmt help docs-install docs-clean docs-build docs-serve lock-dependencies requirements precommit
 
 clean:
 	rm -rf .venv .pytest_cache .ruff_cache .mypy_cache __pycache__
@@ -15,6 +15,9 @@ test:
 	PYTHONPATH=src uv run pytest tests/unit -v
 
 integration:
+	PYTHONPATH=src uv run pytest tests/integration -v -m "not slow and not integration"
+
+integration-live:
 	PYTHONPATH=src uv run pytest tests/integration -v -m "not slow"
 
 fmt:

diff --git a/README.md b/README.md
@@ -1,34 +1,42 @@
 # flowx
 
-ADF to Databricks Lakeflow Jobs translator, delivered as agent skills.
+Orchestrator-to-Databricks Lakeflow Jobs translator, delivered as agent skills.
 
-flowx converts Azure Data Factory (ADF) pipeline definitions into Databricks Lakeflow Jobs packaged as Declarative Automation Bundles (DABs). It deterministically translates known activity types and falls back to agentic (LLM-assisted) translation for complex or rare types. flowx runs as a set of [agent skills](skills/) usable from Databricks Genie Code, Claude Code, or any tool that supports the Agent Skills standard.
+flowx converts a source orchestrator's pipelines — **Azure Data Factory (ADF)** or **Apache
+Airflow** — into Databricks Lakeflow Jobs packaged as Declarative Automation Bundles (DABs). It
+deterministically translates known activity/operator types and falls back to agentic (LLM-assisted)
+translation for complex or rare types. flowx runs as a set of [agent skills](skills/) usable from
+Databricks Genie Code, Claude Code, or any tool that supports the Agent Skills standard.
+
+Both sources emit the same source-neutral Pipeline IR, so the convert-configuration and package
+phases are shared; only discovery and translation are source-specific. Pick the source with
+`--source {adf,airflow}` (required for discover/convert; package is source-independent).
 
 ## Architecture
 
 ```
                          flowx Pipeline
                          ==================
 
-  ADF JSON (UC Volumes / Workspace)
-        |
-        v
-  +------------------+
-  |  1. DISCOVER     |    Parse ADF ARM/JSON exports
-  |  adf_loader.py   | -> Typed AST -> metadata/inventory.json
-  +------------------+
+  ADF ARM/JSON (UC Volumes / Workspace)   |   Airflow DAG .py files
+                          \               |              /
+                           v              v             v
+  +---------------------------------------------------------------+
+  |  1. DISCOVER   sources/<adf|airflow>/  -> metadata/inventory.json
+  |                (ADF: ARM/JSON parse; Airflow: static ast parse)
+  +---------------------------------------------------------------+
         |
         v
-  +------------------+
-  |  2. CONVERT      |    Registry dispatch + topological sort
-  |  engine.py       | -> Pipeline IR (deterministic + agentic gaps)
-  +------------------+
+  +---------------------------------------------------------------+
+  |  2. CONVERT    sources/<adf|airflow>/  -> shared Pipeline IR
+  |                (deterministic mappings + agentic gaps)
+  +---------------------------------------------------------------+
         |
         v
-  +------------------+
-  |  3. PACKAGE      |    IR -> DAB YAML + notebooks + setup scripts
-  |  dab_writer.py   | -> Deployable DABs project
-  +------------------+
+  +---------------------------------------------------------------+
+  |  3. PACKAGE    bundler/dab_writer.py (source-independent)
+  |                IR -> DAB YAML + notebooks + setup scripts
+  +---------------------------------------------------------------+
         |
         v
   databricks bundle validate / deploy
@@ -88,7 +96,7 @@ Run the end-to-end migration:
 Or run individual phases:
 
 ```
-/flowx:flowx-discover    # Parse ADF JSON, produce inventory + complexity report
+/flowx:flowx-discover    # Parse the source (ADF JSON / Airflow DAGs), produce inventory + complexity report
 /flowx:flowx-convert     # Deterministic + agentic translation
 /flowx:flowx-package     # Generate DABs project
 ```
@@ -160,13 +168,45 @@ agent using LLM-assisted reasoning from the activity's ARM JSON.
 | Script | LLM-assisted (agentic) |
 | Until | LLM-assisted (agentic) |
 
+## Supported Airflow Operators
+
+The Airflow source parses DAG `.py` modules **statically** (via `ast`, no Airflow install or DAG
+execution) and maps ~35 operator/sensor families to the shared IR. Highlights:
+
+- **Compute / scripts** — `PythonOperator` (callable → runnable notebook with transitive deps),
+  `BashOperator` / `SSHOperator` (incl. `spark-submit` lift), `SparkSubmitOperator`, the Databricks
+  provider operators, and SQL operators (`DatabricksSql*`, `SQLExecuteQueryOperator`, `HiveOperator`,
+  …) → `sql_task`.
+- **TaskFlow API** — `@dag` / `@task`; implicit XCom data flow lowers to `dbutils.jobs.taskValues`.
+  `@task.expand([literal])` → `for_each_task`; non-literal / `.partial().expand()` / `@task_group` →
+  a linked placeholder notebook that raises `NotImplementedError`.
+- **Sensors** — file/table/time sensors → job triggers or polling notebooks; `ExternalTaskSensor` →
+  cross-DAG wait; Http/Python/DateTime → polling tasks.
+- **dbt** — dbt CLI operators and astronomer-cosmos `DbtDag` / `DbtTaskGroup` → a dbt-factory job
+  (static per-node explosion by default, or PyDABs via `--dbt-mode pydabs`).
+- **Scheduling & semantics** — cron → Quartz, `timedelta` → periodic, `trigger_rule` → `run_if`,
+  `params={...}` → job parameters, `>>` / `<<` / `set_upstream` / TaskGroup edges.
+
+Operators without a deterministic mapping become a failing placeholder and are recorded in
+`gaps.json` for review. Eligible leaf gaps can use the fingerprint-bound resolver backed by the pinned [`airflow-to-dabs`](https://github.com/park-peter/airflow-to-dabs/tree/main/providers/flowx-gap-resolver) provider profile; flowx retains ownership of parsing, graph identity, policy, IR, and packaging. Full matrix:
+[`skills/flowx-convert/sources/airflow-coverage.md`](skills/flowx-convert/sources/airflow-coverage.md).
+
+Airflow discovery independently audits DAG declarations, task candidates, dependency declarations,
+DAG settings, mapped calls, and operator arguments before comparing them with captured IR. An
+included DAG is `verified` when every audited construct has a proven translation,
+`verified_with_gaps` when every unsupported construct is linked to a runnable-failure placeholder,
+or `failed` when reconciliation finds unexplained loss. Failed reconciliation exits nonzero and
+blocks package writes. `--exclude-dag <dag_id>` is repeatable; excluded DAGs emit no Job but remain
+visible with zero translated activities in inventory and coverage reporting. This guarantee applies
+to the supported static subset; flowx never imports or executes DAG modules.
+
 ## How It Works
 
 ### Phase 1: Discover
-Reads ADF JSON definitions from Unity Catalog volumes (or a `/Workspace` Git folder), normalizes ARM template format, parses into typed AST nodes, and classifies each activity as deterministic, agentic, or unsupported. Produces `metadata/inventory.json` and a per-pipeline complexity report at `metadata/profile_report.csv`.
+Parses the source into typed nodes and classifies each activity/operator as deterministic, agentic, or unsupported — ADF JSON from Unity Catalog volumes (or a `/Workspace` Git folder, normalizing ARM template format), or Airflow DAG `.py` modules read statically with `ast`. Airflow inventory includes audited/deterministic/agentic/failed/excluded counts, reconciliation status, stable finding fingerprints, translation-path coverage, and deterministic coverage. Produces `metadata/inventory.json` and a per-pipeline complexity report at `metadata/profile_report.csv`.
 
 ### Phase 2: Convert
-Applies deterministic translators via registry dispatch, resolves dependencies through topological sort, and threads immutable `TranslationContext` through control-flow visitors. Agentic gaps are flagged for LLM-assisted translation. Produces Pipeline IR.
+Applies deterministic translators (ADF activity registry / Airflow operator mapping), resolves dependencies, and records unresolved gaps. ADF supports its guided agentic translation workflow. Airflow supports a fingerprint-bound, explicitly reviewed leaf-gap workflow whose constrained provider output is replayed against an immutable deterministic baseline before packaging. Produces the shared Pipeline IR consumed unchanged by the package phase.
 
 ### Phase 3: Package
 Converts Pipeline IR into a deployable DABs project: `databricks.yml`, per-job YAML resource files, generated Python notebooks, and setup scripts for UC volumes, secrets, and connections.
@@ -180,7 +220,7 @@ flowx_output/
   databricks.yml              # Bundle configuration (package)
   resources/
     jobs/
-      <pipeline_name>.yml     # One job per ADF pipeline
+      <pipeline_name>.yml     # One Job per included ADF pipeline or Airflow DAG
   src/
     notebooks/
       <pipeline_name>/
@@ -209,11 +249,12 @@ for deployment (SDK notebook or CLI script) and Genie Code registration.
 ## Development
 
 ```bash
-make dev          # Install dependencies (uses uv)
-make test         # Run unit tests
-make integration  # Run integration tests
-make fmt          # Format + lint (ruff + mypy)
-make clean        # Remove build artifacts
+make dev               # Install dependencies (uses uv)
+make test              # Run unit tests
+make integration       # Run integration tests (excludes the live-Azure suite; gates CI)
+make integration-live  # Also run tests needing live ADF access (az login + factory access)
+make fmt               # Format + lint (ruff + mypy)
+make clean             # Remove build artifacts
 ```
 
 ### Prerequisites

diff --git a/app/README.md b/app/README.md
@@ -20,7 +20,8 @@ operation; `parameters` is its keyword-argument dict.
 | `inputs` | `adapter inputs` | List a phase's input prompts/defaults |
 | `discover` | `adapter discover` | Parse ADF JSON, classify activities |
 | `convert` | `adapter convert` | ADF activities → Databricks IR |
-| `merge_agentic` | `adapter convert --merge-agentic` | Merge agent-produced results into the report |
+| `merge_agentic` | `adapter convert --merge-agentic` | Merge ADF agent-produced results into the report |
+| `resolve_agentic` | `adapter resolve-agentic` | Prepare, stage, and apply reviewed Airflow leaf-gap resolutions |
 | `inspect` | `adapter inspect` | Surface pending translation options |
 | `apply_answers` | `adapter modify` | Apply answers → stamped IR |
 | `materialize_lookup` | `adapter materialize-lookup` | CSV → lookup-values JSON |
@@ -30,7 +31,7 @@ operation; `parameters` is its keyword-argument dict.
 | `record_results` | `adapter record-results` | Write coverage to a UC table |
 | `install_dashboard` | `adapter install-dashboard` | Publish the coverage dashboard |
 
-Example: `flowx(command="discover", parameters={"adf_source_path": "/Volumes/main/default/adf_export", "output_dir": "./out"})`.
+Example: `flowx(command="discover", parameters={"source": "adf", "adf_source_path": "/Volumes/main/default/adf_export", "output_dir": "./out"})`.
 
 Each command is a thin bridge over `python -m flowx.adapter` (the same entry point the agent
 skills use), then reads back the JSON/CSV artifacts each phase writes — so the MCP surface stays in

diff --git a/docs/content/docs/architecture.mdx b/docs/content/docs/architecture.mdx
@@ -28,7 +28,7 @@ Each activity is classified with a `TranslationStrategy`:
 * `AGENTIC` (LLM-assisted gaps)
 * `UNSUPPORTED`
 
-The reporting layer (`reporting/`) can write per-run coverage to a Unity Catalog table and publish an AI/BI dashboard.
+The reporting layer (`reporting/`) can write per-run coverage to a Unity Catalog table and publish an AI/BI dashboard. Airflow rows use independently audited candidates as the denominator and persist reconciliation status, failed/excluded counts, stable finding fingerprints, translation-path coverage, deterministic coverage, reviewed agentic outcomes, and mechanically validated code-attached coverage. Provider-authored code remains distinct from deterministic translation and requires human review.
 
 ## Two surfaces over one core
 
@@ -59,7 +59,7 @@ The unified `flowx.adapter` CLI is the single contract. Both surfaces go through
 | `mcp/runner.py`   | Subprocess bridge to `flowx.adapter` with artifact summarizers (for running translation without the `mcp` dependency) |
 | `mcp/__main__.py` | `python -m flowx.mcp` entry point (stdio default, `--http` for hosting)                                               |
 
-The `flowx` tool's `command` selects the adapter operation: `inputs`, `discover`, `convert`, `merge_agentic`, `inspect`, `apply_answers`, `materialize_lookup`, `workspace_paths`, `package`, `migrate`, `record_results`, and `install_dashboard` (with `parameters` carrying that command's arguments).
+The `flowx` tool's `command` selects the adapter operation: `inputs`, `discover`, `convert`, `merge_agentic` (ADF only), `resolve_agentic` (Airflow only), `inspect`, `apply_answers`, `materialize_lookup`, `workspace_paths`, `package`, `migrate`, `record_results`, and `install_dashboard` (with `parameters` carrying that command's arguments).
 
 ## Deployment topology
 
@@ -78,7 +78,7 @@ The MCP server runs in whichever transport fits the calling tool. This is chosen
                                                                        own service principal
 ```
 
-See [Installation](/docs/installation#running-flowx-as-an-mcp-server) for the exact commands and the [app README](https://github.com/databricks-solutions/flowx/tree/main/app) for deployment details.
+See [Installation](/flowx/docs/installation#running-flowx-as-an-mcp-server) for the exact commands and the [app README](https://github.com/databricks-solutions/flowx/tree/main/app) for deployment details.
 
 <Callout type="info" title="Inputs and outputs on a hosted app">
 A Databricks App can't read the user's workspace / UC Volume files (`/Volumes/...` is **not** auto-mounted). Two ways to get data in/out of the `flowx` tool:

diff --git a/docs/content/docs/configuration.mdx b/docs/content/docs/configuration.mdx
@@ -149,13 +149,12 @@ phase surfaces three optional inputs — `results_table`, `results_warehouse_id`
 
 - **`record-results`** writes one row **per pipeline per run** to the supplied Unity Catalog
   table (`catalog.schema.table`), combining the complexity columns above with the
-  deterministic/agentic/unsupported coverage breakdown. Every row is stamped with a shared
+  audited/deterministic/agentic/failed/excluded coverage breakdown, reconciliation and migration
+  status, finding fingerprints, translation-path coverage, deterministic coverage, unresolved agentic count, reviewed-resolution outcomes/provider version, and code-attached coverage. The corresponding result columns are `resolved_agentic_count`, `unresolved_agentic_count`, and `code_attached_coverage_pct`. Airflow's audited count remains the denominator even for failed or excluded candidates. Code-attached coverage counts deterministic tasks plus accepted `resolved` provider candidates; it means the generated code passed mechanical contract validation, not that its semantics were certified. Every row is stamped with a shared
   **`run_id`** (UUID), **`run_date`** (`CURRENT_TIMESTAMP()`), and **`run_by`**
   (`CURRENT_USER()`), so coverage is trackable across runs and users.
 - **`install-dashboard`** creates and publishes an AI/BI (Lakeview) dashboard over that table —
-  KPI counters (pipelines, coverage %, deterministic/agentic/unsupported activity totals), a
-  pipelines-by-complexity bar chart, a coverage-over-runs line, and a per-pipeline coverage
-  table.
+  KPI counters (pipelines, audited activities, and mechanically validated code-attached coverage), failed/excluded totals, a pipelines-by-complexity bar chart, a code-attached-coverage trend, and a per-pipeline table that retains translation-path and deterministic coverage.
 
 The SQL warehouse is auto-detected (preferring a running serverless warehouse) when
 `results_warehouse_id` is left blank. Both run via the Databricks SDK and degrade gracefully

diff --git a/docs/content/docs/guide.mdx b/docs/content/docs/guide.mdx
@@ -94,6 +94,15 @@ The bundle contains:
 Connection strings, credentials, and other protected configuration parameters are emitted as `SecretInstruction` steps that require [Databricks Secrets](https://docs.databricks.com/aws/en/security/secrets/). 
 Run the setup scripts to add any required secret values before deploying and running pipelines in your workspace.
 </Callout>
+
+When running with workspace auth (e.g. Genie Code), `package` can optionally persist this run's
+coverage to a Unity Catalog table — one row per pipeline stamped with a UUID `run_id`, `run_date`,
+and `run_by` (`record-results`) — and install a published AI/BI coverage dashboard over that table
+(`install-dashboard`). See [Configuration options](/flowx/docs/options) for details.
+
+For Airflow, `activities` is the independent source-audit count rather than the number of tasks the
+translator happened to emit. Reporting distinguishes deterministic, agentic, failed, and excluded
+candidates and carries reconciliation status, translation-path coverage, deterministic coverage, unresolved agentic outcomes, and mechanically validated code-attached coverage. Code attachment is not a certification that provider-authored code is semantically correct.
 </Step>
 
 <Step>

diff --git a/docs/content/docs/installation.mdx b/docs/content/docs/installation.mdx
@@ -159,10 +159,13 @@ The environment is created once and reused. No `uv` is required for plugin users
 Open Claude Code and ask *"What flowx skills do you have available?"*. You should see a list of skills (e.g. `flowx-setup`, 
 `flowx-migrate`). You can now run `/flowx:flowx-migrate`, `/flowx:flowx-discover`, and other flowx skills.
 
-<Callout type="warn" title="Troubleshooting missing dependencies">
-If calling a skill raises a `ModuleNotFoundError`, the virtual environment is missing or incomplete. Ensure Python is installed
-in your environment and that you have access to a Python package registry for installing dependencies, then re-run `/flowx:flowx-setup`.
-</Callout>
+If you hit a `ModuleNotFoundError` while running a phase, the venv is missing or incomplete — re-run `/flowx:flowx-setup`. Every Python command the skills run uses the interpreter recorded in `<plugin_dir>/.migration-venv`, with `src/` on `PYTHONPATH`:
+
+```bash
+export PYTHONPATH="<plugin_dir>/src"
+PY="$(cat <plugin_dir>/.migration-venv)"
+"$PY" -m flowx.adapter inputs discover --source adf   # or --source airflow
+```
 </Step>
 </Steps>
 

diff --git a/pyproject.toml b/pyproject.toml
@@ -81,10 +81,15 @@ markers = [
 cache-dir = ".venv/ruff-cache"
 target-version = "py312"
 line-length = 120
-exclude = ["templates/*"]
+exclude = ["templates/*", "tests/resources/airflow/review_repros/*"]
 
 [tool.ruff.lint]
 select = ["E", "F", "I"]
 
+[tool.ruff.lint.per-file-ignores]
+# Sample Airflow DAG fixtures are parsed statically, never executed; they
+# reference runtime globals (spark, dbutils) and Airflow imports by design.
+"tests/resources/airflow/*" = ["F821", "F401"]
+
 [tool.ruff.lint.isort]
 known-first-party = ["flowx"]