Summary: Pete-owned work to map Airflow's pre-lowering source captures onto the shared SourceGraph and unified additive inventory; it is not built by the ADF workstream.
Depends on: #61/#84.
Blocks: #86 and Airflow use of shared discovery lineage/motifs/routing.
Current status
This is an open issue and is not implemented; owner is Pete.
Airflow currently has a split static loader but no shared discovery AST:
- src/flowx/sources/airflow/loader/api.py:load_airflow_dags/load_pipelines returns Databricks-facing Pipeline IR directly.
- src/flowx/sources/airflow/loader/captures.py:DagDeclaration/TaskCapture/EdgeCapture and visitor.py:_DagVisitor retain pre-lowering source information that can feed a faithful mapper.
- src/flowx/sources/airflow/loader/lowering.py:_load_airflow_module currently consumes those captures and lowers immediately to Pipeline/Activity.
- src/flowx/sources/airflow/discover.py:build_inventory_dict builds an Airflow-specific inventory from already-lowered IR, while sources/airflow/convert.py:main calls the same direct loader.
#84 defines the source-neutral SourceGraph contract but has not yet merged to main. #79's discovery_inventory.build_source_inventory exists only on integration/discovery and is designed for reuse once re-landed.
What needs to change
Pete should map Airflow's pre-lowering captures onto #84's exact SourceGraph/SourceNode/ContainerNode/GapNode contract, not invent another model and not derive discovery from target IR.
Populate:
- DAG identity, parameters, source-faithful schedule/default policy/timeout/tags;
- operator/TaskFlow native_type, raw source span/payload, trigger_rule as run_condition, task policy, groups/dynamic mapping as container structure, and dependency edges;
- best-effort data_reads/data_writes from Assets/inlets/outlets, SQL, and known operator arguments;
- cross-workflow ControlEdges for TriggerDagRun/Databricks run-job constructs and source-neutral discovery lineage.
Emit inventory through the same generic emitter as ADF, preserving Airflow's existing audit/reconciliation fields additively. Keep load_pipelines and current Airflow conversion unchanged until Pete's Phase 2 #86.
How to approach
Add an Airflow capture-to-SourceGraph mapper beside sources/airflow/loader, using captures.py, visitor.py, dag_discovery.py, and schedule.py before lowering.py loses source fidelity. Persist the full graph through discovery_serde.py and project inventory through discovery_inventory.py:build_source_inventory; layer Airflow audit/findings onto the generic base without changing the emitter's source-neutral core.
Use discovery_lineage.py:walk_nodes/build_graph_lineage and shared DataAsset/ControlEdge types. Add source-specific asset extraction in the Airflow package only.
Acceptance/verification:
- Airflow and ADF emit the same base top-level/per-pipeline inventory shape from build_source_inventory.
- Golden tests preserve every established Airflow audit/reconciliation/coverage field and existing reporting consumers.
- Capture fidelity tests retain operator FQN, source spans/raw, TaskFlow/groups/mapping, trigger rules, policies, schedules, and unsupported gaps.
- Lineage tests cover data assets and cross-DAG control calls without guessed identities.
- SourceGraph serde round-trips all Airflow extensions.
- Existing load_pipelines, Airflow convert/package, agentic leaf resolution, and all ADF paths remain green and unchanged.
This is additive discovery work owned by Pete; the ADF workstream must not touch or break Airflow's current path.
Summary: Pete-owned work to map Airflow's pre-lowering source captures onto the shared SourceGraph and unified additive inventory; it is not built by the ADF workstream.
Depends on: #61/#84.
Blocks: #86 and Airflow use of shared discovery lineage/motifs/routing.
Current status
This is an open issue and is not implemented; owner is Pete.
Airflow currently has a split static loader but no shared discovery AST:
#84 defines the source-neutral SourceGraph contract but has not yet merged to main. #79's discovery_inventory.build_source_inventory exists only on integration/discovery and is designed for reuse once re-landed.
What needs to change
Pete should map Airflow's pre-lowering captures onto #84's exact SourceGraph/SourceNode/ContainerNode/GapNode contract, not invent another model and not derive discovery from target IR.
Populate:
Emit inventory through the same generic emitter as ADF, preserving Airflow's existing audit/reconciliation fields additively. Keep load_pipelines and current Airflow conversion unchanged until Pete's Phase 2 #86.
How to approach
Add an Airflow capture-to-SourceGraph mapper beside sources/airflow/loader, using captures.py, visitor.py, dag_discovery.py, and schedule.py before lowering.py loses source fidelity. Persist the full graph through discovery_serde.py and project inventory through discovery_inventory.py:build_source_inventory; layer Airflow audit/findings onto the generic base without changing the emitter's source-neutral core.
Use discovery_lineage.py:walk_nodes/build_graph_lineage and shared DataAsset/ControlEdge types. Add source-specific asset extraction in the Airflow package only.
Acceptance/verification:
This is additive discovery work owned by Pete; the ADF workstream must not touch or break Airflow's current path.