Airflow: map discovery captures to shared source graphs

README.md

@@ -228,7 +228,7 @@ to the supported static subset; flowx never imports or executes DAG modules.
 ## How It Works
 
 ### Phase 1: Discover
-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`.
+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 discovery also persists the source-faithful shared graph at `metadata/source_graphs.json`, including source identities, dependencies, policies, groups, mappings, gaps, and lineage before target lowering. 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 (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.
@@ -258,6 +258,7 @@ flowx_output/
   metadata/
     inventory.json            # discover: activity inventory
     profile_report.csv        # discover: per-pipeline complexity report
+    source_graphs.json        # Airflow discover: source-faithful shared graphs and lineage
     <pipeline>.arm.json       # discover: verbatim original ADF/ARM source
     configuration.json        # modify: collected configuration answers
   .work/                      # transient intermediates (translation report, IR, gaps.json); pruned by package

docs/content/docs/guide.mdx

@@ -79,6 +79,7 @@ flowx_output/
 ├── metadata/
 │   ├── inventory.json      # discover: activity inventory
 │   ├── profile_report.csv  # profile: per-pipeline complexity report
+│   ├── source_graphs.json  # Airflow discover: source-faithful shared graphs
 │   ├── <pipeline>.arm.json # discover: verbatim original ADF/ARM pipeline source
 │   └── configuration.json  # modify: the collected configuration answers
 └── .work/                  # transient intermediates (translation report, IR, gaps.json); pruned by prepare

src/flowx/discovery_inventory.py

@@ -0,0 +1,110 @@
+"""Source-neutral projection of shared discovery graphs to ``inventory.json``."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from flowx.ir_serde import lineage_to_dict
+from flowx.models.discovery import ContainerNode, SourceGraph, SourceNode
+
+STRATEGY_PROPERTY = "strategy"
+INVENTORY_VISIBLE_PROPERTY = "inventory_visible"
+
+_DETERMINISTIC = "deterministic"
+_AGENTIC = "agentic"
+
+
+def build_source_inventory(
+    graphs: list[SourceGraph],
+    *,
+    source: str,
+    source_dir: str,
+    include_empty_pipelines: bool = True,
+) -> dict[str, Any]:
+    """Projects source graphs into the common discovery inventory shape.
+
+    Args:
+        graphs: Source-faithful workflow graphs to summarize.
+        source: Source-system identifier written to the inventory.
+        source_dir: Source directory label written to the inventory.
+        include_empty_pipelines: Whether workflows without visible activities remain in the inventory.
+
+    Returns:
+        The serializable inventory payload with per-pipeline activities and aggregate coverage.
+    """
+    pipeline_entries: list[dict[str, Any]] = []
+    deterministic = 0
+    agentic = 0
+    unsupported = 0
+
+    for graph in graphs:
+        flattened = [
+            node for node in _flatten_nodes(graph.tasks) if node.properties.get(INVENTORY_VISIBLE_PROPERTY, True)
+        ]
+        for node in flattened:
+            strategy = node.properties.get(STRATEGY_PROPERTY)
+            if strategy == _DETERMINISTIC:
+                deterministic += 1
+            elif strategy == _AGENTIC:
+                agentic += 1
+            else:
+                unsupported += 1
+        if flattened or include_empty_pipelines:
+            entry: dict[str, Any] = {
+                "name": graph.name,
+                "activities": [_activity_entry(node) for node in flattened],
+            }
+            if graph.lineage is not None:
+                entry["lineage"] = lineage_to_dict(graph.lineage)
+            pipeline_entries.append(entry)
+
+    total = deterministic + agentic + unsupported
+    coverage_pct = round((deterministic + agentic) / total * 100, 1) if total else 0.0
+    return {
+        "source": source,
+        "source_dir": source_dir,
+        "pipelines": pipeline_entries,
+        "summary": {
+            "pipeline_count": len(graphs),
+            "activity_count": total,
+            "deterministic_count": deterministic,
+            "agentic_count": agentic,
+            "unsupported_count": unsupported,
+            "coverage_pct": coverage_pct,
+        },
+    }
+
+
+def _flatten_nodes(nodes: list[SourceNode]) -> list[SourceNode]:
+    """Flattens container branches in source order."""
+    flattened: list[SourceNode] = []
+    for node in nodes:
+        flattened.append(node)
+        if isinstance(node, ContainerNode):
+            for children in node.branches.values():
+                flattened.extend(_flatten_nodes(children))
+    return flattened
+
+
+def _activity_entry(node: SourceNode) -> dict[str, Any]:
+    """Builds one additive per-activity inventory entry."""
+    entry: dict[str, Any] = {
+        "name": node.name if node.name is not None else node.task_key,
+        "type": node.native_type,
+        "strategy": node.properties.get(STRATEGY_PROPERTY),
+    }
+    upstream_names = [dependency.upstream for dependency in node.dependencies]
+    if upstream_names:
+        entry["depends_on"] = upstream_names
+    entry["original_type"] = node.native_type
+    entry["dependencies"] = [
+        {
+            "upstream": dependency.upstream,
+            "conditions": list(dependency.conditions),
+            "resolved": dependency.resolved,
+        }
+        for dependency in node.dependencies
+    ]
+    if node.raw is not None:
+        entry["raw"] = node.raw
+    return entry

src/flowx/sources/airflow/callable_notebook.py

@@ -19,7 +19,10 @@
 _AIRFLOW_IMPORT_ROOTS: frozenset[str] = frozenset({"airflow", "cosmos", "airflow_dbt"})
 
 
-def _enclosing_statements(module: ast.Module, func: ast.FunctionDef) -> list[ast.stmt]:
+def _enclosing_statements(
+    module: ast.Module,
+    func: ast.FunctionDef | ast.AsyncFunctionDef,
+) -> list[ast.stmt]:
     """Returns safe statements visible from the callable's enclosing function scopes."""
     scopes = [
         node
@@ -89,7 +92,9 @@ def _names_used(node: ast.AST) -> set[str]:
 
 
 def _closure(
-    func: ast.FunctionDef, defs: dict[str, ast.stmt], assigns: dict[str, ast.stmt]
+    func: ast.FunctionDef | ast.AsyncFunctionDef,
+    defs: dict[str, ast.stmt],
+    assigns: dict[str, ast.stmt],
 ) -> tuple[list[str], set[str]]:
     """Returns transitively-referenced module symbols (defs+assigns) in source order, plus all names used.
 
@@ -118,6 +123,36 @@ def _closure(
     return ordered, all_names
 
 
+def render_source_closure(func: ast.FunctionDef | ast.AsyncFunctionDef, source: str) -> str:
+    """Renders a callable and every statically resolved source dependency it uses.
+
+    Args:
+        func: Callable definition whose source closure is required.
+        source: Complete DAG module source used to resolve and slice dependencies.
+
+    Returns:
+        Source containing referenced imports, helpers, classes, constants, and the callable.
+    """
+    module = ast.parse(source)
+    enclosing_statements = _enclosing_statements(module, func)
+    definitions, assignments = _module_symbols(module, enclosing_statements)
+    imports = _import_bindings(module, enclosing_statements)
+    dependency_names, used_names = _closure(func, definitions, assignments)
+
+    import_nodes: dict[int, ast.stmt] = {}
+    for name in used_names:
+        binding = imports.get(name)
+        if binding is not None:
+            import_nodes[id(binding[0])] = binding[0]
+
+    nodes = [
+        *sorted(import_nodes.values(), key=lambda node: node.lineno),
+        *(definitions.get(name) or assignments[name] for name in dependency_names),
+        func,
+    ]
+    return "\n\n".join(segment for node in nodes if (segment := ast.get_source_segment(source, node)) is not None)
+
+
 def render_definitions(func: ast.FunctionDef, source: str, *, note: str) -> str:
     """Renders the callable's ``def`` plus its transitive deps as a notebook prelude (no invocation).
 

src/flowx/sources/airflow/discover.py

@@ -1,9 +1,9 @@
 """Airflow discover phase: parse DAGs into a classified inventory.
 
-Mirrors the ADF discover contract: writes ``metadata/inventory.json`` and
-``metadata/profile_report.csv`` under the shared output dir. Independently audited
-task candidates drive deterministic, agentic, failed, and excluded counts; emitted
-IR tasks remain available for the per-task inventory.
+Mirrors the ADF discover contract: writes ``metadata/inventory.json``,
+``metadata/source_graphs.json``, and ``metadata/profile_report.csv`` under the shared
+output dir. Shared source graphs drive per-task inventory while independently audited
+task candidates drive deterministic, agentic, failed, and excluded counts.
 Exposes ``main(argv)`` so the adapter runs it in-process, like the ADF loader.
 """
 
@@ -17,9 +17,13 @@
 from typing import Any
 
 from flowx.adapter.predicates import walk_activities
+from flowx.discovery_inventory import build_source_inventory
+from flowx.discovery_lineage import walk_nodes
+from flowx.discovery_serde import source_graph_to_dict
+from flowx.models.discovery import SourceGraph
 from flowx.models.ir import NotebookActivity, Pipeline, PlaceholderActivity
 from flowx.sources.adf.loader import clear_stale_outputs
-from flowx.sources.airflow.loader import load_pipelines
+from flowx.sources.airflow.loader import load_discovery_results
 
 logger = logging.getLogger(__name__)
 
@@ -42,12 +46,34 @@ def _classify(pipeline: Pipeline) -> list[dict[str, str]]:
     return items
 
 
-def build_inventory_dict(pipelines: list[Pipeline], source_dir: str) -> dict[str, Any]:
+def build_inventory_dict(
+    pipelines: list[Pipeline],
+    source_dir: str,
+    *,
+    graphs: list[SourceGraph] | None = None,
+) -> dict[str, Any]:
     """Builds the inventory.json payload matching the ADF discover shape."""
+    if graphs is not None and len(graphs) != len(pipelines):
+        raise ValueError("Airflow inventory requires exactly one source graph per pipeline")
+    base: dict[str, Any] = (
+        build_source_inventory(graphs, source="airflow", source_dir=source_dir)
+        if graphs is not None
+        else {"source": "airflow", "source_dir": source_dir, "pipelines": [], "summary": {}}
+    )
+    base_entries = list(base["pipelines"])
     pipeline_entries: list[dict[str, Any]] = []
     audited = deterministic = agentic = failed = excluded = 0
     for pipeline in pipelines:
-        items = _classify(pipeline)
+        base_entry = base_entries[len(pipeline_entries)] if len(base_entries) > len(pipeline_entries) else None
+        items = list(base_entry["activities"]) if base_entry is not None else _classify(pipeline)
+        if graphs is not None and base_entry is not None:
+            graph_nodes = [
+                node
+                for node in walk_nodes(graphs[len(pipeline_entries)].tasks)
+                if node.properties.get("inventory_visible", True)
+            ]
+            for item, node in zip(items, graph_nodes, strict=True):
+                item["task_key"] = node.task_key
         pipeline_audited = int(pipeline.audit.get("audited_activity_count", len(items)))
         pipeline_deterministic = int(
             pipeline.audit.get("deterministic_count", sum(1 for item in items if item["strategy"] == "deterministic"))
@@ -70,7 +96,8 @@ def build_inventory_dict(pipelines: list[Pipeline], source_dir: str) -> dict[str
         agentic += pipeline_agentic
         failed += pipeline_failed
         excluded += pipeline_excluded
-        pipeline_entries.append(
+        entry = dict(base_entry or {})
+        entry.update(
             {
                 "name": pipeline.name,
                 "activities": items,
@@ -87,6 +114,7 @@ def build_inventory_dict(pipelines: list[Pipeline], source_dir: str) -> dict[str
                 "transformations": pipeline.audit.get("transformations", []),
             }
         )
+        pipeline_entries.append(entry)
     coverage = round(100.0 * (deterministic + agentic) / audited, 1) if audited else 0.0
     deterministic_coverage = round(100.0 * deterministic / audited, 1) if audited else 0.0
     reconciliation_status = (
@@ -98,24 +126,25 @@ def build_inventory_dict(pipelines: list[Pipeline], source_dir: str) -> dict[str
         if pipelines and all(pipeline.migration_status == "excluded" for pipeline in pipelines)
         else "verified"
     )
-    return {
-        "source": "airflow",
-        "source_dir": source_dir,
-        "pipelines": pipeline_entries,
-        "summary": {
-            "pipeline_count": len(pipelines),
-            "activity_count": audited,
-            "audited_activity_count": audited,
-            "deterministic_count": deterministic,
-            "agentic_count": agentic,
-            "unsupported_count": 0,
-            "failed_count": failed,
-            "excluded_count": excluded,
-            "coverage_pct": coverage,
-            "deterministic_coverage_pct": deterministic_coverage,
-            "reconciliation_status": reconciliation_status,
-        },
-    }
+    base.update(
+        {
+            "pipelines": pipeline_entries,
+            "summary": {
+                "pipeline_count": len(pipelines),
+                "activity_count": audited,
+                "audited_activity_count": audited,
+                "deterministic_count": deterministic,
+                "agentic_count": agentic,
+                "unsupported_count": 0,
+                "failed_count": failed,
+                "excluded_count": excluded,
+                "coverage_pct": coverage,
+                "deterministic_coverage_pct": deterministic_coverage,
+                "reconciliation_status": reconciliation_status,
+            },
+        }
+    )
+    return base
 
 
 # Full profile column set the shared reporting.coverage / dashboard consume. Airflow has no
@@ -187,7 +216,9 @@ def main(argv: list[str] | None = None) -> int:
 
     logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
 
-    pipelines = load_pipelines(args.source_dir, pipeline=args.pipeline, exclude_dags=set(args.exclude_dag))
+    results = load_discovery_results(args.source_dir, pipeline=args.pipeline, exclude_dags=set(args.exclude_dag))
+    pipelines = [result.pipeline for result in results]
+    graphs = [result.graph for result in results]
     if not pipelines:
         logger.error("No Airflow DAGs found under %s (or none matched --pipeline).", args.source_dir)
         return 1
@@ -198,8 +229,14 @@ def main(argv: list[str] | None = None) -> int:
     metadata_dir = output_dir / "metadata"
     metadata_dir.mkdir(parents=True, exist_ok=True)
 
-    inventory = build_inventory_dict(pipelines, str(args.source_dir))
+    inventory = build_inventory_dict(pipelines, str(args.source_dir), graphs=graphs)
     (metadata_dir / "inventory.json").write_text(json.dumps(inventory, indent=2), encoding="utf-8")
+    source_graphs = {
+        "contract_version": "1",
+        "source": "airflow",
+        "graphs": [source_graph_to_dict(graph) for graph in graphs],
+    }
+    (metadata_dir / "source_graphs.json").write_text(json.dumps(source_graphs, indent=2), encoding="utf-8")
     _write_profile_csv(pipelines, metadata_dir / "profile_report.csv")
 
     summary = inventory["summary"]