[FEATURE]: Agentic insights in inventory.json (discover phase)

[matthewmoorcroft]

TL;DR

Add an agent-authored insights key to metadata/inventory.json during the discover phase, giving the convert phase the two things it can't derive on its own: what each pipeline is trying to achieve + the Databricks pattern that maps to it, and how pipelines relate across the corpus (annotating #23's lineage edges). It's a new Step 5 in flowx-discover (agent authors JSON → calls flowx(command="enrich") → deterministic validate + merge), mirroring how the convert phase already folds agentic JSON back. Depends on #23. Produces the block only; a follow-up wires convert to consume it.

Validated on a representative multi-pipeline factory: the insights would have driven a ~5–6× smaller, more idiomatic conversion (one parameterized job vs. 6 cloned subtrees; Lakeflow Connect CDC vs. hand-rolled JDBC) — see Evidence.

---

The gap

The deterministic inventory + #23's lineage block is correct but flat: a list of typed activities and a set of edges. It can't answer, without judgment, the two questions that drive conversion quality:

1. What is each pipeline for, and what Databricks pattern fits? (e.g. "this is metadata-driven incremental extraction → collapse into one parameterized Lakeflow task with CDC" — read off the activity composition, not a field.)
2. How do pipelines interact in a way the converter must preserve? [FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23's edges say that A calls B / A writes a table B reads; they don't say how to assemble that in Databricks. The convert phase translates one pipeline at a time with no cross-pipeline view, so this is structurally invisible to it.

Design principle — annotate, don't rediscover. The inventory + lineage are the source of truth for facts; the agent adds only judgment and references pipelines/edges by their existing identifiers. Every insight is cheap to validate: named pipelines must exist, annotated edges must exist in lineage. Scope is deliberately factory / pipeline / relationship — no per-activity fields (the convert agent already handles that level well; see Evidence).

---

Schema — the insights key

enrich adds exactly one top-level key. Everything above it is deterministic and byte-for-byte untouched; all agent content lives inside insights and references deterministic elements by name.

{
  // deterministic — written by `discover`, untouched by `enrich`
  "pipelines": [ { "name": "...", "activities": [ { "name": "...", "type": "...", "strategy": "..." } ] } ],
  "lineage":   { "control_edges": [ ... ], "data_edges": [ ... ] },   // #23
  "summary":   { "...": "..." },

  // agent-authored — the ONLY key `enrich` adds (its presence = "enriched")
  "insights": {
    "overview": "The whole factory as one system: what it is, how the pieces fit, the one big migration steer spanning all pipelines.",

    "pipeline_insights": [               // per-pipeline judgment; SPARSE (omit pipelines with nothing to add)
      {
        "pipeline": "..._Orchestrator",  // FK → pipelines[].name (validated)
        "pattern_name": "Short handle, e.g. 'Metadata-driven incremental extraction orchestrator'.",
        "intent": "What the SOURCE pipeline is trying to achieve.",
        "databricks_pattern": "The TARGET pattern that maps to it, incl. 'do NOT transliterate' steers.",
        "recommended_databricks_features": ["Lakeflow Declarative Pipelines", "Delta control table"],
        "conversion_notes": ["CRITICAL: the 6 source-group loops are clones — emit ONE parameterized loop."]
      }
    ],

    "pipeline_relationships": [          // cross-pipeline judgment; annotates a #23 edge, never rediscovers it
      {
        "from_pipeline": "..._Orchestrator",  // FK → pipelines[].name (validated)
        "to_pipeline":   "..._Child_A",        // FK → pipelines[].name (validated)
        "lineage_edge": { "edge_type": "control", "edge_identity": "invoke_child_a" },  // must resolve to a real #23 edge
        "relationship_summary": "What the relationship IS (invocation / watermark hand-off / shared table).",
        "databricks_pattern": "How to assemble it in Databricks (what a per-pipeline converter can't see).",
        "risk_if_ignored": "The concrete failure mode, e.g. 'CHANGETABLE silently defaults to full load'."
      }
    ]
  }
}

All fields except the FKs (pipeline / from / to / lineage_edge) are optional and free-form; arrays are sparse.

lineage_edge is a typed reference, not prose — it binds a relationship to one deterministic #23 edge:

• edge_type: "control" (→ lineage.control_edges) or "data" (→ lineage.data_edges).
• edge_identity: echoed verbatim from that edge — the ExecutePipeline activity name (control) or the shared table/path (data).

Relationships key on (from, to, edge_type, edge_identity), so the same pipeline pair can carry multiple facets (e.g. one control invocation + one data watermark hand-off).

Validation on ingest: named pipelines ∈ inventory; each lineage_edge resolves to a real lineage edge; required fields present, unknown fields rejected; deterministic keys never mutated (a violation is flagged, never merged).

---

How the agent calls it

flowx(command="enrich", parameters={
    "output_dir": "./flowx_output",         # locates metadata/inventory.json
    "insights": { ... },                    # inline dict (hosted/Genie path) — OR —
    "insights_path": "/path/insights.json", # a readable path (local/CLI path)
})

Mirrors _cmd_discover/_cmd_merge_agentic (src/flowx/mcp/server.py:137,169). Returns {ok, process, ...} + a merge summary; on any validation failure returns ok:false with violations and does not write.

Step 5 loop in flowx-discover: read inventory.json (+ lineage, profile_report.csv) → author insights → call enrich → tool validates + merges into inventory.json → the (reworded) summary step presents intent/patterns. Default-on, skippable for fast/large runs (gate lives in the skill, not the tool).

---

Implementation — a two-pass write

Inlining into inventory.json stays safe because discover's serializer is unchanged and enrich only appends the insights key.

• Pass 1 — discover (unchanged): _inventory_to_dict (adf_loader.py:682) writes pure inventory.json, no insights. Golden diffs stay pristine.
• Pass 2 — enrich (new): read → validate → set insights key → re-serialize with the same json.dumps(indent=2) so the deterministic portion round-trips byte-identical. Idempotent.

Additions:

• Models (models/adf_ast.py, beside Inventory/[FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23's Lineage) — @dataclass(slots=True, kw_only=True):

  class PipelineInsight:      # pipeline (FK), pattern_name, intent, databricks_pattern,
                              # recommended_databricks_features: list, conversion_notes: list
  class LineageEdgeRef:       # edge_type, edge_identity
  class PipelineRelationship: # from_pipeline (FK), to_pipeline (FK), lineage_edge: LineageEdgeRef,
                              # relationship_summary, databricks_pattern, risk_if_ignored
  class Insights:             # overview, pipeline_insights: list, pipeline_relationships: list

• parser/pipeline_insights.py (new, sibling to [FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23's lineage.py, shaped like merge_agentic_results at engine.py:1632): load_insights → validate_insights (FK/edge checks) → merge_into_inventory (pure) → enrich_inventory (read/validate/merge/write-back).
• Adapter subcommand (adapter/__main__.py, modeled on record-results :341): standalone enrich (--output-dir, --insights/--insights-path) — standalone, not a discover flag, so it re-runs without re-parsing.
• MCP command (mcp/server.py): _cmd_enrich in the _COMMANDS map (:365).
• Skill (skills/flowx-discover/SKILL.md): add Step 5; reword the summary step to consume insights.
• Reporting (optional, reporting/coverage.py:57): optional intent/pattern column, gated on the key existing.

---

Evidence — representative migration

Ran discover on a representative multi-pipeline ADF factory (~10K-line ARM export), hand-authored insights, compared against the conversion convert actually produced. Names and paths below are anonymized/synthetic; the shape is representative.

• Discover: 6 pipelines, ~160 activities, 100% coverage. One XL master orchestrator (~120 activities) + 5 child extractors.
• Conversion produced (verified): ~80 tasks, 6 byte-identical ~200-line child templates, ~90 notebooks — a faithful 1:1 transliteration. Core Copy = SqlServerSource + CHANGETABLE → ParquetSink.
• Insights would have improved it: "collapse 6 loops + 5 clones into one parameterized job" → ~5–6× less generated surface the converter had no signal to attempt; named Lakeflow Connect CDC (higher-fidelity than the hand-rolled JDBC); flagged the CSV→Delta watermark coupling — invisible to a per-pipeline converter and exactly where a silent CDC bug hides.
• Counter-finding that set the scope: per-activity translations (GetMetadata→dbutils.fs.ls, dataflow→PySpark) were already good → no per-activity fields; the whole gap was pipeline/relationship-level.

Full authored insights for the representative factory (anonymized validated artifact)

{
  "insights": {
    "overview": "A single metadata-driven incremental data-extraction framework for SQL Server sources. One master orchestrator reads a control table of tables/views to extract, partitions the workload three ways, and drives six near-identical 'source group' loops that each resolve a per-group change-tracking watermark and fan out to templated child extractors. The five child pipelines are structural clones differentiated only by logging/version-update flags and by tables-vs-views. The whole factory should migrate as ONE parameterized Lakeflow job with a single parameterized child job — not as a 1:1 transliteration of the ADF activity graph.",

    "pipeline_insights": [
      {
        "pipeline": "ExtractOrchestrator",
        "pattern_name": "Metadata-driven incremental extraction orchestrator (watermark/CDC fan-out)",
        "intent": "Master orchestrator. Reads a control table (a source-table registry) enumerating every source table and view to extract, filters tables vs. views, and builds parameter arrays. A mapping data flow splits the input-params CSV into three near-equal partitions for parallelism. It then runs six near-identical 'source group' loops (group_1..group_6): each loop resolves the group's last change-tracking version (last SYS_CHANGE_VERSION, persisted as a per-group CSV in ADLS under a watermarks path), decides initial-load vs. incremental, and invokes the child extractor pipelines via ExecutePipeline. It also manages DB-version marker files (moving current→old, deleting stale) and loads timezone reference data. Extensive per-table success/failure logging is written back to ADLS.",
        "databricks_pattern": "This is textbook metadata-driven ingestion and must NOT be translated as six separate task subtrees. Collapse the six hand-cloned source-group loops into ONE parameterized Lakeflow task that iterates the control table (for-each over the table list), driven by the same control metadata. Persist the SYS_CHANGE_VERSION watermark in a Delta control table rather than per-group CSV files. The three-way CSV split is an ADF-runtime parallelism artifact with no Databricks equivalent — drop it and rely on Spark/Lakeflow partitioning instead of physically splitting the params file. The DB-version marker-file dance (current/old file existence + move) is a bespoke idempotency guard that maps to a Delta table checkpoint.",
        "recommended_databricks_features": [
          "Lakeflow Declarative Pipelines (for the CDC/watermark incremental logic)",
          "Lakeflow Connect SQL Server ingestion connector (native change tracking)",
          "Delta control/watermark table (replaces per-group CSV watermarks and DB-version marker files)",
          "Job-level for-each task (replaces the 6 cloned source-group loops)"
        ],
        "conversion_notes": [
          "CRITICAL: group_1..group_6 are byte-identical ~200-line clones differing only by numeric suffix — emit ONE parameterized loop, not six. A 1:1 port produces ~6x redundant surface.",
          "The watermark lives in ADLS CSV today; a faithful port preserves a fragile pattern. Recommend migrating it to a Delta control table as part of conversion.",
          "The core work is a SQL Server CHANGETABLE (Change Tracking) read — a full SELECT when lastVersionID=0, otherwise the CDC delta. This is the crux of the pipeline; get the initial-vs-incremental branch right."
        ]
      },
      {
        "pipeline": "Extractor_A",
        "pattern_name": "Templated leaf extractor — SQL Server Change Tracking → Parquet (with version-update + logging)",
        "intent": "Templated leaf extractor invoked by the master. Iterates a list of tables (ForEach) and, for each, runs the core Copy that reads SQL Server via CHANGETABLE(CHANGES ...) — a full SELECT when the passed lastVersionID=0, otherwise the change-tracking delta — and writes Parquet to ADLS. Tracks per-table success/failure into log files and updates the group's DB version marker. Extractor_A is the variant that DOES perform version-update and create-log-files steps.",
        "databricks_pattern": "The core Copy is a SQL Server Change-Tracking read landing to Parquet. Target the Lakeflow Connect SQL Server connector (native CDC) or, if kept as batch, a JDBC read of CHANGETABLE into a streaming table with MERGE. The per-table ForEach becomes a single parameterized notebook/task run over the table list. Crucially, the five child variants (A / AA / B / BB / Views) differ ONLY by logging/version-update flags and tables-vs-views — they must collapse into ONE parameterized child job with those differences expressed as parameters, not five near-identical jobs.",
        "recommended_databricks_features": [
          "Lakeflow Connect SQL Server connector (CHANGETABLE / change tracking)",
          "Parameterized child job (single job for all A/AA/B/BB/Views variants)",
          "Delta MERGE (apply the CDC delta idempotently)"
        ],
        "conversion_notes": [
          "lastVersionID is supplied by the master per group; the CHANGETABLE query keys off it — this parameter MUST flow through from the parent (see pipeline_relationships).",
          "Fold version-update and create-log-files into boolean parameters so A and B collapse into one job."
        ]
      },
      {
        "pipeline": "Extractor_AA",
        "pattern_name": "Templated leaf extractor — partition variant of Extractor_A",
        "intent": "Identical extractor structure to Extractor_A (ForEach over tables → CHANGETABLE Copy → Parquet, with version-update and log-file creation); the 'AA' partition variant. Exists only because ADF cannot parameterize a pipeline reference by partition.",
        "databricks_pattern": "Exact duplicate of Extractor_A's target pattern. Collapse A + AA into the single parameterized child job, keyed by partition parameter. Do not emit a second task tree.",
        "recommended_databricks_features": ["Parameterized child job (shared with Extractor_A)"],
        "conversion_notes": ["Pure clone of Extractor_A — no distinct logic. The only reason it exists separately is ADF's static pipeline references."]
      },
      {
        "pipeline": "Extractor_B",
        "pattern_name": "Templated leaf extractor — SQL Server Change Tracking → Parquet (logging only, no version bookkeeping)",
        "intent": "Leaf extractor variant WITHOUT the version-update / create-log-files steps: ForEach over tables → CHANGETABLE Copy → Parquet, with success/failure status logging only.",
        "databricks_pattern": "Same SQL Server Change-Tracking → Parquet core as Extractor_A, minus version bookkeeping. Fold into the single parameterized child job with logging/version-update exposed as boolean parameters (here both off).",
        "recommended_databricks_features": ["Parameterized child job (shared)", "Lakeflow Connect SQL Server connector"],
        "conversion_notes": ["Differs from Extractor_A only by the absence of version-update/log-file steps — a parameter toggle, not a separate job."]
      },
      {
        "pipeline": "Extractor_BB",
        "pattern_name": "Templated leaf extractor — partition variant of Extractor_B",
        "intent": "The 'BB' partition variant of Extractor_B — identical extractor structure, differing only by numeric suffix.",
        "databricks_pattern": "Exact duplicate of Extractor_B. Collapse into the shared parameterized child job keyed by partition.",
        "recommended_databricks_features": ["Parameterized child job (shared)"],
        "conversion_notes": ["Pure clone of Extractor_B."]
      },
      {
        "pipeline": "Extractor_Views",
        "pattern_name": "Templated leaf extractor — VIEW extraction (full read, no CDC)",
        "intent": "Leaf extractor specialized for VIEWS rather than tables: ForEach over a filtered view list → Copy → Parquet, with status logging. Views are read in full — there is no change-tracking watermark branch.",
        "databricks_pattern": "Full-read (non-CDC) JDBC/connector extraction of views to Parquet or Delta. Same parameterized-child-job pattern as the table extractors, with the CDC/watermark branch disabled (a 'views' mode parameter).",
        "recommended_databricks_features": ["Parameterized child job (shared, views mode)", "JDBC / Lakeflow Connect full read (no change tracking)"],
        "conversion_notes": ["The distinguishing trait is full-read views vs. CDC tables — represent as an extraction-mode parameter on the shared child job."]
      }
    ],

    "pipeline_relationships": [
      {
        "from_pipeline": "ExtractOrchestrator",
        "to_pipeline": "Extractor_A",
        "lineage_edge": { "edge_type": "control", "edge_identity": "invoke_extractor_a" },
        "relationship_summary": "Invocation. The master invokes the child extractor variants via ExecutePipeline from inside the six source-group loops — 18 invocations in total (3 partitions x 6 group loops) across the five child variants. Each invocation passes the group context, partition, table list, and the resolved lastVersionID.",
        "databricks_pattern": "Model the master as ONE job whose for-each task spawns runs of a SINGLE parameterized child job (run_job_task), passing partition + groupID + table-list + lastVersionID as parameters. The 18 ExecutePipeline edges collapse to one child job invoked with different parameters — not five child jobs wired 18 ways. This is the single biggest structural simplification available in this migration.",
        "risk_if_ignored": "A literal 1:1 port emits five near-identical child jobs and 18 explicit invocation edges, multiplying the maintenance surface ~5-6x and obscuring that this is one reusable extractor."
      },
      {
        "from_pipeline": "ExtractOrchestrator",
        "to_pipeline": "Extractor_A",
        "lineage_edge": { "edge_type": "data", "edge_identity": "{containerName}/{subDirectory}/watermarks/{groupID}.csv" },
        "relationship_summary": "Watermark hand-off. The master resolves each group's last SYS_CHANGE_VERSION by reading the per-group CSV under the watermarks path and passes it to the child as lastVersionID. The child's CHANGETABLE query keys off that value to decide full-load (lastVersionID=0) vs. incremental delta. The correctness of the entire incremental extraction depends on this value being read, passed, and updated consistently.",
        "databricks_pattern": "Replace the per-group CSV watermark file with a single shared Delta control table (keyed by groupID) that both the master reads and the extraction updates after a successful run. Pass lastVersionID into the child job as an explicit parameter. This centralizes the watermark, makes it transactional, and removes the fragile file-existence dance.",
        "risk_if_ignored": "This relationship is invisible to a per-pipeline converter — it translates the master and the child independently and cannot see that the watermark connects them. If the value isn't threaded through correctly, the CHANGETABLE branch silently defaults (e.g. re-runs a full load, or misses deltas) with no error — a silent data-correctness bug."
      }
    ]
  }
}

The two relationships share the same (from, to) pair but distinct lineage_edge facets — the case motivating the (from, to, edge_type, edge_identity) key.

---

Testing

Repo conventions (tests/unit/, fixtures under tests/resources/json/, precedent in test_merge_agentic.py). Output is non-deterministic → assert structure/schema, not prose:

• Validator: a good insights fixture parses; malformed ones reject — pipeline not in inventory, lineage_edge with no matching edge, missing required field, unknown field.
• lineage_edge binding: using pipeline_execute_pipeline_nested.json (known control edges), a matching lineage_edge validates, a non-matching one rejects.
• Two-pass invariant: enrich_inventory leaves deterministic keys byte-identical, adds only insights, is idempotent.
• No live LLM — stubbed fixture JSON only.

Scope

• In scope: Step 5 in flowx-discover; the insights schema + dataclasses; pipeline_insights.py (validate/merge); the enrich adapter subcommand + MCP command; the two-pass write; optional reporting column. Advisory; skippable.
• Out of scope — consuming insights in convert: a separate follow-up wires convert to read them (mirrors [FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23 produce → this issue/[FEATURE]: Ordered cross-pipeline deploy/run from control lineage (package phase) #24 consume). this issue only produces the block.
• Also out: new skill/phase; in-code LLM client; the [FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23 extraction itself; per-activity insights; domains grouping; per-edge narrative; [FEATURE]: Ordered cross-pipeline deploy/run from control lineage (package phase) #24 deploy-ordering.

Dependencies & open questions

• Depends on [FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23 — consumes its lineage block; edge_identity semantics must track [FEATURE]: Deterministic lineage & dependency tracking, surfaced in the inventory #23's final ControlEdge/DataEdge field names.
• Open: exact edge_identity grammar for control edges (activity name vs. composite); the auto-skip pipeline-count threshold; lineage_edge nested-object vs. flattened serialization (nested recommended).