Phase 1 stack #3: agentic insights / enrich

.claude-plugin/plugin.json

@@ -11,6 +11,7 @@
   "skills": [
     "./skills/flowx-setup",
     "./skills/flowx-discover",
+    "./skills/flowx-enrich",
     "./skills/flowx-convert",
     "./skills/flowx-package",
     "./skills/flowx-migrate"

skills/flowx-discover/SKILL.md

@@ -74,7 +74,28 @@ The inventory classifies every task into one of three strategies:
 - **Agentic** — requires LLM-assisted translation from the source definition.
 - **Unsupported** — no known translation path; needs manual intervention.
 
+## Step 3 — Enrich the inventory (default next step)
+
+Once the deterministic inventory is written, the standard flow **chains into enrichment**: you author
+a layer of judgment the parser cannot derive — a factory-wide architecture recommendation,
+per-pipeline intent + recommended Databricks patterns, and cross-pipeline relationships — and merge
+it back under a single additive `insights` key. flowx contains no LLM: you author the JSON, the
+library validates and merges it. The routing step (`flowx-route`) reads this block to present the
+agentic conversion option per pipeline group.
+
+**Continue with the `flowx-enrich` skill by default** — it guides authoring the insights and running
+`enrich`. Enrichment is additive and leaves every deterministic inventory key byte-identical, so it
+never destabilizes discover's output.
+
+**Deterministic-only skip path.** A standalone, headless discover — no agent, no LLM — is fully
+valid: `metadata/inventory.json` is complete and self-standing without an `insights` block, and
+`flowx-route` still recommends and records a plan from the deterministic structure alone. Skip
+`flowx-enrich` only when the caller explicitly asked for a deterministic-only pass, or when no human
+reader needs a migration narrative.
+
 ## Reference
 
 - `sources/adf.md` — Azure Data Factory discovery (ARM JSON, UC-volume download, complexity report)
 - `sources/airflow.md` — Apache Airflow discovery (DAG `.py` parsing, operator classification)
+- `flowx-enrich` skill — authoring the agentic `insights` layer and running `enrich` (the default
+  next step)

skills/flowx-enrich/SKILL.md

@@ -0,0 +1,109 @@
+---
+name: flowx-enrich
+description: >
+  Enrich the discover inventory with an agent-authored layer of judgment — a factory-wide
+  architecture recommendation, per-pipeline intent + recommended Databricks patterns, and
+  cross-pipeline relationships — then validate and merge it into inventory.json. The default next
+  step after flowx-discover and the input the routing step consumes.
+triggers:
+  - "enrich inventory"
+  - "enrich pipelines"
+  - "author insights"
+  - "agentic insights"
+  - "recommend databricks patterns"
+  - "annotate inventory"
+  - "enrich discover"
+---
+
+# Enrich the Inventory with Agentic Insights
+
+The deterministic discover pass records what each source workflow **is**; it cannot record what to
+**do** about it. That judgment — a factory-wide architectural recommendation, each pipeline's intent
+and recommended Databricks patterns, and how pipelines couple — is authored by **you, the agent**,
+and merged back into `metadata/inventory.json` under a single additive `insights` key.
+
+This is the standard step **between discover and route** in the flowx workflow. `flowx-discover`
+chains into this skill by default; the routing step (`flowx-route`) reads the `insights` block to
+present the agentic conversion option per connected component.
+
+## No LLM inside flowx — you author, the library validates and merges
+
+**There is no LLM inside flowx.** You author the insights JSON; the library (`enrich`) only
+*validates and merges* it — the same author → validate → merge, fingerprint-bound contract the
+agentic gap-resolution and routing paths use. That keeps the deterministic inventory trustworthy and
+every insight accountable: foreign keys must point at real pipelines, and every cross-pipeline edge
+is either an annotation of a proven lineage edge or an explicitly-flagged inference with cited
+evidence.
+
+`enrich` is **additive**: it merges only an `insights` block and leaves every existing inventory key
+byte-identical. It changes no conversion, IR, or routing decision on its own — the `insights` are
+descriptive data that `flowx-route` later consumes.
+
+## When to skip enrich (deterministic-only)
+
+Enrichment is on by default, but it is skippable. A standalone, deterministic-only discover — no
+agent, no LLM — is fully valid: `metadata/inventory.json` from discover is complete and self-standing
+without an `insights` block, and `flowx-route` still recommends and records a plan from the
+deterministic structure alone (the agentic option simply shows no recommended patterns). Skip enrich
+when the caller asked for a headless/deterministic pass, or when no human reader needs a migration
+narrative.
+
+## How to author (three steps)
+
+1. **Read the deterministic inventory.** Load `<output_dir>/metadata/inventory.json`. Note every
+   pipeline `name` (these are the only valid foreign keys), and each pipeline's `lineage` block — in
+   particular `lineage.control_edges`, each `{source_workflow, target_workflow, via_task_key}`. A
+   deterministic **control** relationship you annotate must match one of these exactly.
+2. **Read the source artifacts** you need to form judgment — the per-pipeline `raw` payloads in the
+   inventory, the ADF `metadata/<pipeline>.arm.json` provenance, or the DAG source — enough to state
+   each pipeline's *intent* and the Databricks patterns that fit. Ground every recommended pattern in
+   a **real, publicly-documented** Databricks capability; never invent a product name.
+3. **Author the insights JSON, then call `enrich`.** The library validates it against the inventory
+   and, only when clean, merges it in atomically. On any violation the inventory is left untouched
+   and you get the full list of problems to fix in one pass.
+
+See **`insights.md`** in this skill directory for the exact insights shape, every field, and the
+validation rules the library enforces.
+
+## Run enrich — MCP tool or venv CLI
+
+Run the **`setup`** skill first if you haven't. Both paths run the same validate-and-merge contract.
+
+- **MCP tool (Databricks Genie Code, or a local stdio registration):** call the single **`flowx`**
+  tool with `command="enrich"` and either inline insights or a file:
+
+  ```
+  flowx(command="enrich", parameters={"output_dir": "<dir>", "insights": { ... }})   # inline object
+  flowx(command="enrich", parameters={"output_dir": "<dir>", "insights_path": "<file>"})
+  ```
+
+  Provide **exactly one** of `insights` (inline object) or `insights_path`. `ok` reflects validation;
+  `result.violations` lists any problems. Run **no** `python3`/`$PY` commands on this path.
+
+- **venv CLI (local, no MCP server):** ensure the venv exists (`setup` / `bootstrap.sh`), then:
+
+  ```bash
+  export PYTHONPATH="<plugin_dir>/src"
+  PY="$(cat <plugin_dir>/.migration-venv)"
+  "$PY" -m flowx.adapter enrich --output-dir <dir> --insights-path insights.json
+  ```
+
+  Both `--output-dir` and `--insights-path` are required; `--out <file>` optionally writes the result
+  JSON to a file instead of stdout. **Exit code 0** means the insights merged; **exit code 1** prints
+  the violations JSON and leaves the inventory untouched.
+
+## Idempotency & safety
+
+`enrich` is atomic and idempotent: it replaces the whole `insights` block (never stacks), recomputes
+the `inventory_sha256` fingerprint from the deterministic inventory, and leaves every existing
+inventory key byte-identical. Re-running with the same insights rewrites the same bytes; re-running
+with different insights replaces the block. A validation failure writes nothing.
+
+## Next step
+
+After the inventory is enriched, continue with **`flowx-route`** to recommend and record a
+per-connected-component conversion route (deterministic vs. agentic), then convert and package.
+
+## Reference
+
+- `insights.md` — the insights shape, every field, and the validator's rules.

skills/flowx-enrich/insights.md

@@ -0,0 +1,92 @@
+# The agentic insights shape
+
+This is the reference for the `insights` JSON you author before calling `enrich`. The `flowx-enrich`
+SKILL.md covers the workflow (no-LLM contract, the three authoring steps, how to run `enrich`, and
+when a deterministic-only pass skips it); this file covers **what to write** and the rules the
+validator enforces.
+
+Author the insights when a human reader would benefit from a migration narrative — which pipelines
+collapse onto a managed capability, how the factory hangs together, what the risky couplings are.
+The `flowx-route` step reads this block to present the agentic conversion option per component.
+
+## The insights shape
+
+You author only these four fields (the library injects `schema_version` and an `inventory_sha256`
+fingerprint that binds your insights to the exact inventory they describe):
+
+```json
+{
+  "overview": "One short factory-wide narrative — what this collection of pipelines is.",
+  "system_recommendation": {
+    "headline": "The one decision a migrator must make before any per-pipeline work",
+    "recommended_patterns": [
+      {"pattern": "Lakeflow Connect SQL Server connector", "fit": "Replaces the child extractor family", "simplification_pattern": true},
+      {"pattern": "Parameterised Lakeflow Job", "fit": "Like-for-like orchestration fallback", "simplification_pattern": false}
+    ],
+    "cascade": ["5 child extractors -> managed connector pipelines"],
+    "decision_driver": "Is the Lakeflow Connect connector GA/approved for this source?"
+  },
+  "pipeline_insights": [
+    {
+      "pipeline": "IngestSalesforce",
+      "intent": "Land Salesforce objects into the bronze layer nightly",
+      "databricks_pattern": "Managed ingestion",
+      "recommended_patterns": [
+        {"pattern": "Lakeflow Connect", "fit": "Managed CDC ingestion replaces the copy loop", "simplification_pattern": true}
+      ],
+      "conversion_notes": ["Point the connector at the same source objects"],
+      "risk_if_ignored": "Bespoke extractor code and its watermark table carry forward"
+    }
+  ],
+  "pipeline_relationships": [
+    {
+      "from_pipeline": "Orchestrator",
+      "to_pipeline": "IngestSalesforce",
+      "lineage_edge": {"edge_type": "control", "edge_identity": "<via_task_key from a real control edge>"},
+      "relationship_summary": "Orchestrator invokes IngestSalesforce"
+    },
+    {
+      "from_pipeline": "IngestSalesforce",
+      "to_pipeline": "BuildMart",
+      "lineage_edge": {
+        "edge_type": "inferred",
+        "edge_identity": "shared table sales.curated",
+        "evidence": "Both notebooks read/write sales.curated, but the hand-off is inside notebook code the parser can't see",
+        "confidence": "medium"
+      }
+    }
+  ]
+}
+```
+
+### Rules the validator enforces
+
+- **Foreign keys.** Every `pipeline_insights[].pipeline` and every relationship
+  `from_pipeline` / `to_pipeline` must be a real pipeline name in the inventory.
+- **`recommended_patterns`** (per pipeline and system-wide): **1–4** patterns. The validator enforces
+  the count and that each has a non-empty `pattern` and `fit` and a boolean `simplification_pattern`;
+  it does **not** enforce ordering. Set `simplification_pattern: true` **only** for a distinctive
+  capability that collapses a whole legacy pattern (a managed connector, declarative `AUTO CDC`, Auto
+  Loader, system tables replacing a home-grown logging tier) — not for a like-for-like port. By
+  convention (not validated), order them best-first and list the `simplification_pattern: true` ones
+  ahead of like-for-like ports.
+- **`system_recommendation`** needs a non-empty `headline` and a `recommended_patterns` list;
+  `cascade` (non-empty strings) and `decision_driver` are optional.
+- **Relationship edges** come in two tiers:
+  - `control` — an **annotation** of a proven control edge. `edge_identity` must be the
+    `via_task_key` of a real `control_edges` entry whose `source_workflow`/`target_workflow` match
+    your `from_pipeline`/`to_pipeline`. Do **not** set `evidence`/`confidence` — the proven edge is
+    the evidence.
+  - `inferred` — a coupling the deterministic layer never found (data flow inside notebook code, an
+    external trigger, a shared table the parser didn't resolve). There is nothing to resolve
+    against, so `edge_identity` is your descriptor of the coupling and you **must** supply a non-empty
+    `evidence` string and a `confidence` of `high` / `medium` / `low`.
+  - There is no deterministic cross-pipeline **data** tier in v1: the deterministic data edges are
+    intra-pipeline and task-scoped, so a cross-pipeline data coupling rides the `inferred` tier.
+
+## Idempotency & safety
+
+`enrich` is atomic and idempotent: it replaces the whole `insights` block (never stacks), recomputes
+the fingerprint from the deterministic inventory, and leaves every existing inventory key
+byte-identical. Re-running with the same insights rewrites the same bytes; re-running with different
+insights replaces the block. A validation failure writes nothing.

src/flowx/adapter/__main__.py

@@ -1,9 +1,9 @@
 """Unified CLI entry point that the flowx skills and MCP tools drive via subprocesses.
 
 Exposes stateless subcommands -- the ``discover``/``convert``/``package`` phase runners plus
-``inspect``, ``modify``, ``resolve-agentic``, ``inputs``, ``materialize-lookup``, ``workspace-paths``,
-``record-results``, and ``install-dashboard`` -- so each agent turn runs as an independent process
-holding no session state across user prompts.
+``inspect``, ``modify``, ``resolve-agentic``, ``enrich``, ``inputs``, ``materialize-lookup``,
+``workspace-paths``, ``record-results``, and ``install-dashboard`` -- so each agent turn runs as an
+independent process holding no session state across user prompts.
 """
 
 from __future__ import annotations
@@ -85,6 +85,8 @@ def main(argv: list[str] | None = None) -> int:
         return _run_workspace_paths(args)
     if args.command == "resolve-agentic":
         return _run_resolve_agentic(args)
+    if args.command == "enrich":
+        return _run_enrich(args)
     if args.command == "record-results":
         return _run_record_results(args)
     if args.command == "install-dashboard":
@@ -140,6 +142,27 @@ def _run_resolve_agentic(args: argparse.Namespace) -> int:
     return 0
 
 
+def _run_enrich(args: argparse.Namespace) -> int:
+    """Implements ``enrich``: validate agent-authored insights and merge them into inventory.json.
+
+    Emits the enrich result JSON (``ok`` / ``violations`` / counts) to stdout so the caller can
+    surface every violation at once. Returns 0 when the insights merged cleanly, 1 on validation
+    failure (inventory left untouched) or when the inputs cannot be read.
+    """
+    from flowx.discovery_insights import enrich_inventory
+
+    try:
+        result = enrich_inventory(args.output_dir, insights_path=args.insights_path)
+    except FileNotFoundError as error:
+        print(str(error), file=sys.stderr)
+        return 1
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Failed to enrich inventory: {error}", file=sys.stderr)
+        return 1
+    _emit_json(result, args.out)
+    return 0 if result.get("ok") else 1
+
+
 def _run_record_results(args: argparse.Namespace) -> int:
     """Implements ``record-results``: write per-pipeline coverage to a UC table.
 
@@ -475,6 +498,29 @@ def _build_parser() -> argparse.ArgumentParser:
         help="Airflow dbt conversion mode used to reproduce the deterministic report during prepare.",
     )
 
+    enrich = subparsers.add_parser(
+        "enrich",
+        help="Validate agent-authored insights and merge them into inventory.json (additive, atomic).",
+    )
+    enrich.add_argument(
+        "--output-dir",
+        type=Path,
+        required=True,
+        help="Migration output directory (reads and rewrites metadata/inventory.json).",
+    )
+    enrich.add_argument(
+        "--insights-path",
+        type=Path,
+        required=True,
+        help="Path to the agent-authored insights JSON to validate and merge.",
+    )
+    enrich.add_argument(
+        "--out",
+        type=Path,
+        default=None,
+        help="Optional output file for the enrich result JSON; defaults to stdout.",
+    )
+
     record = subparsers.add_parser(
         "record-results",
         help="Write per-pipeline migration coverage for this run to a Unity Catalog table.",