[FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory

[matthewmoorcroft]

Problem

flowx already parses which pipeline calls which (ExecutePipeline) and which dataset each activity reads/writes (Copy source/sink, Lookup, Delete, GetMetadata) — but the discover inventory (metadata/inventory.json) throws that structure away. It records only per-pipeline, intra-pipeline activity DAGs; there is no cross-pipeline call graph and no dataset producer→consumer graph anywhere.

This issue adds a purely deterministic (no LLM) lineage extractor that assembles two graphs from the already-parsed AdfDefinitions and surfaces them in the inventory under a new lineage block:

1. Control lineage — ExecutePipeline caller→callee call graph.
2. Data lineage — dataset write→read (producer→consumer) edges, joined on resolved physical identity (schema.table or storage path) so two differently-named datasets pointing at the same table match.

This is the ground-truth substrate that #25 (agentic enrichment) and #24 (cross-pipeline deploy/run ordering) consume. Both are out of scope here.

Proposed solution

Follow repo conventions (AGENTS.md): @dataclass(slots=True, kw_only=True) models, deterministic extraction in the parser layer.

Models (src/flowx/models/adf_ast.py, beside Inventory):

@dataclass(slots=True, kw_only=True)
class ControlEdge:
    caller_pipeline: str
    callee_pipeline: str          # resolved ExecutePipeline referenceName (raw name if unresolved)
    activity_name: str
    wait_on_completion: bool = True

@dataclass(slots=True, kw_only=True)
class DataEdge:
    dataset_name: str
    identity: str | None          # resolved schema.table / storage path, or None
    producer_pipeline: str
    producer_activity: str
    consumer_pipeline: str
    consumer_activity: str

@dataclass(slots=True, kw_only=True)
class Lineage:
    control_edges: list[ControlEdge] = field(default_factory=list)
    data_edges: list[DataEdge] = field(default_factory=list)

Add lineage: Lineage | None = None to Inventory.

Extractor — new src/flowx/parser/lineage.py with build_lineage(definitions) -> Lineage:

• Control edges: walk each pipeline's activity tree (recursing into ForEach/If/Switch) and, for each ExecutePipeline, read the callee referenceName + waitOnCompletion. Resolve callees case-insensitively against the pipeline set; record unresolved callees (partial export) with their raw name rather than dropping them.
• Data edges: derive direction-aware (dataset_name, identity, direction) triples — sink/outputs ⇒ producer, source/inputs + Lookup/Delete/GetMetadata ⇒ consumer. Join producers→consumers on resolved identity (fall back to dataset_name when identity is None). Never guess: unresolvable identity ⇒ identity: null.

Design decision: shared resolver module (composition) — committed

The physical-identity resolvers already exist in copy.py but run only at convert time. We commit to composition — extract them once into a shared module both phases import — not duplication into lineage.py.

• New parser-layer module src/flowx/parser/dataset_resolvers.py becomes the single home for identity resolution: resolve_table_reference, resolve_dataset_path, resolve_dataset_linked_service_name, dataset_props, plus one composition entry point resolve_dataset_identity(dataset_ref, definitions, context=None) -> str | None.
• Both translator/copy.py (convert) and the new parser/lineage.py (discover) import it. copy.py's call sites are unchanged apart from imports.

Why this is safe and clean (verified):

• No import cycle: parser/ does not import translator/, and copy.py already imports from parser/, so a shared module in the parser layer is importable by both.
• Runs at discover time: the resolvers' only convert dependency is TranslationContext, and every field of it has a default, so TranslationContext() constructs empty. Lineage passes an empty context; literals/defaults resolve, runtime-only expressions yield non-literals ⇒ identity: null.
• Behavior-preserving: the extraction is a pure relocation; copy.py's existing convert/copy tests staying green is the proof.

Codegen-specific helpers (SinkPathInfo, _resolve_path_info, _LOCATION_URL_TEMPLATE, volume sanitisation) stay in copy.py.

Wiring: build_inventory calls build_lineage; _inventory_to_dict adds a sibling "lineage" key (always emits empty lists, never null) plus a new top-level "schema_version" marker.

JSON shape:

{
  "schema_version": "2",
  "pipelines": [ /* unchanged */ ],
  "summary": { /* unchanged */ },
  "lineage": {
    "control_edges": [
      {"caller_pipeline": "ParentPipeline", "callee_pipeline": "ChildPipeline_A",
       "activity_name": "Run Child A", "wait_on_completion": true}
    ],
    "data_edges": [
      {"dataset_name": "ds_curated_orders", "identity": "curated.orders",
       "producer_pipeline": "ChildPipeline_A", "producer_activity": "data extraction",
       "consumer_pipeline": "ParentPipeline", "consumer_activity": "Lookup latest"}
    ]
  }
}

Scope / non-goals

• In scope: deterministic control + data lineage in inventory.json; the model dataclasses; the extractor module; the shared resolver extraction; wiring into build_inventory/_inventory_to_dict; schema_version; tests.
• Out of scope: agentic enrichment / "holistic view" ([FEATURE]: Agentic insights in inventory.json (discover phase) #25); consuming lineage for cross-pipeline deploy/run ordering ([FEATURE]: Ordered cross-pipeline deploy/run from control lineage (package phase) #24); column-level lineage and expression-level dataflow inside notebooks.

Testing

Deterministic, golden-graph style (matches tests/unit/test_adf_loader.py):

• Control: assert the exact caller→callee edge set from pipeline_execute_pipeline_nested.json (3 ExecutePipeline activities); add a fixture with an ExecutePipeline nested in a ForEach/If to lock in recursion; assert an unresolved callee is recorded.
• Data: assert sink=producer / source=consumer direction, and that two differently-named datasets pointing at the same table join on identity; unresolvable identity ⇒ null and still joins on name.
• Serialization: _inventory_to_dict emits schema_version + lineage, with empty lists (not null) when there are no edges.
• The shared-resolver extraction is behavior-preserving: the existing convert/copy suite stays green.

Notes / risks

• Identity resolution can be parameterized (@dataset().X); when not deterministically resolvable, emit identity: null.
• Unresolved callees (partial exports) are recorded, not dropped, so the graph surfaces missing dependencies.
• Cycles are tolerated (the graph is descriptive, not sorted); only [FEATURE]: Ordered cross-pipeline deploy/run from control lineage (package phase) #24's ordering consumer needs cycle handling.
• Backward compatibility: the new top-level keys are additive; reporting/coverage.py reads only pipelines/summary and is unaffected.

Relationship: deterministic foundation for #25 (agentic enrichment) and #24 (deploy ordering); both depend on this and are tracked separately.