refactor: add source-neutral discovery graph contract and shared lineage core

[matthewmoorcroft]

Intent

Define one source-neutral discovery graph contract (SourceGraph) that ADF and Airflow can both project into before any Databricks IR. It is persisted as a versioned, hashed source_graphs.json, and one shared inventory.json builder serves every source. This is Phase 1 groundwork for lineage-aware routing. It is additive, and nothing in discover, convert or package calls it yet.

What changed

• models/discovery.py: SourceGraph, SourceNode, ContainerNode and GapNode, plus schedule, parameter, policy and dependency specs. A graph carries raw, properties and extensions; a node carries raw and properties.
• models/lineage.py, ir_serde.py: source-neutral lineage types (DataAsset, ControlEdge, DataEdge, MotifAnnotation, Lineage), shared by the discovery graph and the IR, and their serialisers. models/ir.py re-exports them.
• lineage.py, discovery_lineage.py: shared lineage derivation.
  • Control edges have no self-edges, are deduplicated, and keep unresolved callees with resolved=False.
  • Data edges match on a resolved physical identity first, then on signature; a signature-only match carries no identity.
  • Container branches are walked fully, including Switch defaults and ForEach and Until bodies.
• discovery_serde.py: the SourceGraph JSON round trip.
  • Node kinds are kept, and an empty PolicySpec comes back as PolicySpec().
  • The source_graphs.json envelope has a contract version, a hash per graph and a document hash, all checked on read.
• discovery_inventory.py: build_source_inventory, the single source-agnostic inventory.json emitter. It writes the source, each activity's task_key, dependencies and raw, a per-pipeline lineage block, detected motifs (members left in place) and an optional source_graphs_sha256.
• Greg's review: a named callee missing from the export is now unresolved when the caller passes known_workflows; the source_graphs.json reader requires both hashes, and both reader and writer refuse graphs from another source; an unknown node_type raises; data edges are documented as matching assets, not run order.

How it was tested

• Unit tests for the model, the serde (round trip and hash verification), both lineage tiers and the inventory emitter. The full unit suite, ruff and mypy pass in CI.
• ADF discover, convert and package on the 40 fixture pipelines: bundles and translation reports are byte-identical before and after.
• Codex cross-review, read-only, twice. Round 1 found two issues, the signature-tier identity and the empty PolicySpec round trip. Both are fixed in bbd00f1 with regression tests that fail on the previous code. Round 2 found none.

Merge order

Merge #84 first, then #94, which then retargets to main. #95, #96 and #97 follow in order.

Peter's #108 rebases onto main afterwards. Its only conflicts are the expected add/add on discovery_inventory.py and tests/unit/test_discovery_inventory.py. Take #84's copy of discovery_inventory.py. In the test file, take #84's copy but keep #108's test_emitter_has_no_source_specific_imports, which #84 does not carry.

This pull request and its description were written by Isaac.

[ghanse]

Looking good, requested a few changes. The contract and shared lineage core are clean and well tested. Two things should be fixed before merge: a named but missing callee still gets marked as resolved, and the hash check can be skipped by deleting the hash fields. The other comments are smaller.

[ghanse]

Any call with a name gets resolved=True, even when that pipeline isn't in the export. So a call to a missing pipeline looks resolved. Could the caller pass the set of known workflows?

resolved=target_name in known_workflows,

[ghanse]

Both hash checks are skipped when the field is missing, so deleting graph_sha256 and document_sha256 gets an edited file past verification. Since v1 always writes them, I think the reader should require them:

if recorded_graph_hashes is None:
    raise ValueError("source_graphs document is missing graph_sha256")

[ghanse]

Non-blocking: this pairs every writer with every reader of the same asset, no matter the run order. So two parallel writers, or a writer that runs after the reader, still get a "proven" hand-off. It's worth a short comment saying so, plus tests for multiple-writer and reversed-order cases.

[ghanse]

Non-blocking: an unknown node_type quietly becomes a SourceNode and drops any extra fields. Raising a ValueError for anything other than the three known kinds would be safer.

[ghanse]

Small suggestion: check that each graph's source matches the document's source. Right now a document labeled adf can hold Airflow graphs and still pass every hash check.

[matthewmoorcroft]

The requested changes are applied at fb38c14 (fix commit 6d41efb, plus follow-ups):

1. Missing callee: control_edges_from_calls, build_graph_control_edges, build_graph_lineage and with_graph_lineage take a keyword-only known_workflows. A named callee not in that set comes back with resolved=False. The signature and rule match the one feat(adf): save source graphs with dataset lineage at ADF discover #94 uses.
2. Hash bypass: the source_graphs.json reader now requires both graph_sha256 and document_sha256 and raises when either is missing, using your suggested check.
3. Run order: added a comment in data_edges_from_endpoints and on DataEdge saying the pairing matches assets, not run order. Added tests for two parallel writers and for a writer that runs after its reader.
4. Unknown node_type: anything other than SourceNode, ContainerNode or GapNode now raises ValueError.
5. Source mismatch: the reader raises when a graph's source differs from the document's source. The writer (source_graphs_document / write_source_graphs) now raises the same error, so a mismatched document cannot be saved in the first place.

Each fix has a test that failed on the previous code, except point 3, whose tests pin the existing behaviour as you asked.

[ghanse]

Looking good. Found a few additional items we should address before merging.

[ghanse]

Thanks, this works when known_workflows is passed, but it still defaults to None, and then any named call counts as resolved again. Nothing in src/ passes it yet, so that's the path real code takes. Could we make it required, or treat None as "unknown"?

resolved=bool(target_name) and known_workflows is not None and target_name in known_workflows,

[ghanse]

The docstring says document_sha256 covers the whole envelope, but it only hashes contract_version, source and graphs. Extra top-level keys get through without being hashed or rejected. Either reject unknown keys on read or fix the docstring.

doc["extra"] = {"injected": True}
source_graphs_from_document(doc)  # still passes

[ghanse]

A node with no strategy, or one we don't recognize, counts toward unsupported_count, but its activity entry still shows the raw value (None or whatever it was). So the summary and the activity list disagree. Could the entry say "unsupported" too?

[ghanse]

with_graph_lineage swaps in a brand-new Lineage, so any motifs already on the graph are quietly lost. Worth keeping them, plus a test that calls it after motifs are attached.

motifs = graph.lineage.motifs if graph.lineage else []

[ghanse]

Suggestion: _flatten_nodes does the same walk as walk_nodes in discovery_lineage.py. Could this just be list(walk_nodes(nodes)), so the two can't drift apart?

[matthewmoorcroft]

The second round of changes is applied at bec41c5:

1. known_workflows=None now means unknown, using your line: a named callee is resolved only when known_workflows is given and contains it. I kept the argument optional rather than required, since nothing reads resolved yet and the one real caller already passes the export's workflow names; without them every edge is now unresolved.
2. The source_graphs.json reader now rejects any top-level key outside contract_version, source, graphs and the two hashes, so nothing can ride along unhashed. No writer adds other keys, so this breaks no compatibility path. The docstring now says document_sha256 covers contract_version, source and graphs.
3. A node with a missing or unrecognised strategy is now listed as unsupported in its activity entry, matching the summary count. Both read the same helper.
4. with_graph_lineage keeps motifs already on the graph, with a test that attaches motifs first.
5. The inventory now walks nodes with walk_nodes; _flatten_nodes is gone.

Points 1 to 4 each have a test that failed on the previous code. Point 5 is a refactor covered by the existing ordering test.

I also went over the whole diff again, against the themes from both of your reviews here and on #94, and fixed a few more of the same kind:

• an empty identity no longer joins two assets;
• a weak name match needs the same asset_type when both sides set one;
• a non-string callee marker is an empty, unresolved target, not str(value);
• reading back a missing or non-boolean resolved gives unresolved, and a non-boolean wait_for_completion gives None;
• inventory_visible hides a node only when it is exactly False;
• branches or reason on the wrong kind of node raises;
• a malformed source_graphs.json raises ValueError instead of crashing.

Each has a test that failed before the change.

[ghanse]

Good groundwork. The lineage rules and hash checks are solid and well tested, and I didn't find any bugs. Add the nested-container tests before the source PRs build on this.

[ghanse]

This says every graph and node has properties / extensions, but SourceNode only has properties and raw. Please fix the wording here and in the PR description so later mappers don't go looking for a node extensions field.

[ghanse]

An identity match skips the asset_type check that the signature match does, so a "table" and a "file" with the same identity still get linked. Either check the type here too or note why it's fine.

[ghanse]

Suggestion: The dedupe key leaves out wait_for_completion, so two edges that differ only in that setting quietly become one, and the first one wins. Fine for now, but add wait to the key or say that the first one wins.

[ghanse]

There are no tests for an invoke inside a nested container, or a container inside a container. The code handles both, so a test should validate the behavior.

[ghanse]

Suggestion: This module docstring reads like a design doc. A short summary would do, since the details already live on the functions (same goes for the top of models/discovery.py).

[ghanse]

Suggestion: Discovery runs before IR but imports DataAsset / Lineage from models.ir. Reusing them is right, but maybe move them to a small shared module so "IR" doesn't end up meaning two things.

[matthewmoorcroft]

Thanks. All six are addressed at 5ebdfcb:

1. Nested containers (r4321169393): added tests for an invoke two containers deep, a data edge between nodes nested in different containers, walk order through a container inside a container, and nested inventory flattening. They pass on the old code, as expected, since the code already handled both. 78763b2.
2. Identity match and asset_type (r4321169384): the identity tier now refuses a match when both sides set asset_type and the types differ, the same as the signature tier, with a test that failed before. One side unknown still matches. 78763b2, docstrings in 5ebdfcb.
3. Node fields (r4321169381): the docstring now says a graph carries raw, properties and extensions, and a node carries raw and properties. The PR description says the same. 78763b2.
4. Dedupe key (r4321169388): wait_for_completion is now part of the key, so calls that differ only in waiting stay separate edges; exact repeats still collapse. Test added. 78763b2.
5. Module docstrings (r4321169397): discovery_inventory.py and models/discovery.py now open with a short summary, and the details moved to build_source_inventory. 78763b2.
6. Shared module (r4321169402): DataAsset, ControlEdge, DataEdge, MotifAnnotation and Lineage now live in flowx/models/lineage.py. flowx.models.ir re-exports them, so the stacked PRs keep working unchanged, with a test that both names point at the same classes. 78763b2.

With these in, could you approve #84 when you have a moment? Then we merge the stack in order: #84 first, then #94 to #97, each with its own round-three fixes.