Agentic insights: intent + cross-pipeline relationships in the discover inventory

skills/flowx-discover/sources/adf.md

@@ -83,6 +83,11 @@ Strategy Breakdown:
 Coverage:             95.7%
 ```
 
+Then, after the shared insights step has enriched `inventory.json`, surface the authored judgment so
+the user sees *what the factory does*, not just coverage numbers: print the factory `overview`, and
+for each `pipeline_insights` entry its `pattern_name` / `intent` and its top `recommended_patterns`
+(ranked simplification-first).
+
 ## Step 6 — Detail agentic activities
 
 For `agentic` activities, explain that each is translated by the agent using LLM-assisted reasoning
@@ -98,3 +103,47 @@ to a PySpark notebook.
 
 Tell the user where the metadata files were written (`<output_dir>/metadata/`), summarise the
 complexity sizes, and confirm they can proceed to `flowx-convert` with the same `<output_dir>`.
+
+## Insights — deep-dive & pattern vocabulary
+
+Reference for the shared agentic-insights step (parent `SKILL.md` Step 5, "Author and merge agentic
+insights"). Do this deep-dive before authoring insights for any ADF pipeline.
+
+**Deep-dive the ARM.** The inventory is a deterministic skeleton (types, strategy, control edges);
+the *why* and *how* — queries, Switch conditions, notebook paths, dataset parameters — live only in
+the verbatim ARM. The `metadata/` folder holds one `*.arm.json` file per pipeline; each is a **flat
+single-pipeline object** shaped `{"name": "<pipeline>", "properties": {"activities": [...], ...}}`
+(no `resources[]` array, no top-level `type`). To inspect a pipeline, **glob `metadata/*.arm.json`
+and match on each file's top-level `"name"` field** — do **not** construct a filename from the
+pipeline name (names are slugified and lossy, so a built path can miss or collide). Activities are
+under `properties.activities` (recurse into nested `ForEach`/`If`/`Switch` bodies). Read the ARM for
+any pipeline you write an insight or relationship about.
+
+**ADF constructs → Databricks** — a reference menu, NOT an allowlist; the target side uses current
+product names, so reach past it whenever a better or newer fit exists. Flag `simplification_pattern:
+true` only on entries that use a distinctive capability, never on the plain-orchestration fallback.
+
+| Pipeline does… | Simplifying target — `simplification_pattern: true` (rank first) | Fallback — `false` |
+|---|---|---|
+| Extract/Copy from a database (SQL Server, …) | **Lakeflow Connect** managed connector (change-tracking/CDC → Delta) | Auto Loader / JDBC read + `MERGE INTO` |
+| Incremental load via watermark | **Lakeflow Declarative Pipelines `AUTO CDC`** | Delta `MERGE INTO` + control table / `dbutils.jobs.taskValues` |
+| CDC / SQL Server change tracking | **Lakeflow Connect** or **`AUTO CDC`** | Structured Streaming over the change feed |
+| Land + process files | **Auto Loader** (`cloudFiles`, file-notification mode) | — |
+| Metadata-driven bulk copy (Lookup→ForEach→Copy) | **Lakeflow Connect** (multi-table) or a parameterized **Lakeflow Jobs** for-each task | — |
+| Parent/child `ExecutePipeline` fan-out | **Lakeflow Jobs** for-each task + run-job task + job parameters | — |
+| SCD Type 2 (data flow) | **Lakeflow Declarative Pipelines `AUTO CDC`** (SCD Type 2) | — |
+| Staged load + stored-proc transform | Spark write to **Delta** + post-load step | — |
+| REST API pagination | Python ingestion notebook (requests-based) | Lakeflow Connect SaaS connector if one fits |
+| Custom logging / observability tier | **system tables (`system.lakeflow.*`) + native job notifications + AI/BI dashboard** | — |
+| Run-state / control tables | Lakeflow job & task run state + `dbutils.jobs.taskValues` | — |
+| Clone family (many near-identical pipelines) | one **parameterized Lakeflow Job** invoked N times | — |
+
+**Emit current names, not legacy ones:** Lakeflow Jobs (was Databricks Workflows), Lakeflow
+Declarative Pipelines (was Delta Live Tables/DLT), `AUTO CDC` (was `APPLY CHANGES INTO`), Declarative
+Automation Bundles (was Databricks Asset Bundles), AI/BI dashboards (was Lakeview), `system.lakeflow`
+(was `system.workflow`).
+
+**Then author the insights (shared method).** With this deep-dive and pattern vocabulary in hand,
+author and merge the `insights` object by following the source-neutral "Author and merge agentic
+insights" step in the parent `SKILL.md`. The insight schema and the authoring method are shared
+across sources; only the deep-dive and the construct mappings above are ADF-specific.

skills/flowx-discover/sources/airflow.md

@@ -50,6 +50,11 @@ Total tasks:        8
 Coverage:           87.5%
 ```
 
+Then, after the shared insights step has enriched `inventory.json`, surface the authored judgment so
+the user sees *what the DAGs do*, not just coverage numbers: print the factory `overview`, and for
+each `pipeline_insights` entry its `pattern_name` / `intent` and its top `recommended_patterns`
+(ranked simplification-first).
+
 ## Step 5 — Detail agentic tasks
 
 For `agentic` tasks, name the operator that has no deterministic mapping yet (e.g. a custom or
@@ -69,3 +74,44 @@ that are **not** handled (dynamic TaskGroup mapping, shared multi-DAG bundle), s
 [`../../flowx-convert/sources/airflow-coverage.md`](../../flowx-convert/sources/airflow-coverage.md).
 Callables reading Airflow task context (`**context` / `ti`) or XCom, and runtime-branching
 decorators, are routed to placeholders for manual/agentic translation rather than converted.
+
+## Insights — deep-dive & pattern vocabulary
+
+Reference for the shared agentic-insights step (parent `SKILL.md` Step 5, "Author and merge agentic
+insights"). Do this deep-dive before authoring insights for any DAG.
+
+**Deep-dive the DAG source.** The inventory is a deterministic skeleton (task types, strategy,
+dependencies); the *why* and *how* live in the **DAG source** — the `.py` files under the
+`--source-path` you discovered from. Read the DAG module for any pipeline you write about: task
+callables (`PythonOperator` bodies), operator arguments, templated params, hooks / connections, and
+`set_upstream` / `>>` dependencies. Recurse into `TaskGroup`s and dynamically mapped (`.expand`)
+tasks. The parser already extracts operators, `>>` / `<<` edges, `schedule_interval`, and inline
+callables (see "How it works" above), so read the source for the intent the static parse can't
+capture — what a callable actually *does*, what a hook connects to, and why the tasks are ordered as
+they are.
+
+**Airflow operators → Databricks** — a reference menu, NOT an allowlist; the target side uses current
+product names, so reach past it whenever a better or newer fit exists. Flag `simplification_pattern:
+true` only on entries that use a distinctive capability, never on the plain-orchestration fallback.
+
+| DAG uses… | Simplifying target — `simplification_pattern: true` (rank first) | Fallback — `false` |
+|---|---|---|
+| DB extract via `MsSqlOperator` / `JdbcOperator` / custom hook | **Lakeflow Connect** managed connector (change-tracking/CDC → Delta) | JDBC read + `MERGE INTO` |
+| Incremental load w/ XCom or Variable watermark | **Lakeflow Declarative Pipelines `AUTO CDC`** | Delta `MERGE INTO` + `dbutils.jobs.taskValues` |
+| File sensor + load (`*FileSensor` → transform) | **Auto Loader** (`cloudFiles`, file-notification mode) | — |
+| `SparkSubmitOperator` / `DatabricksSubmitRunOperator` | native **Lakeflow Job** task (notebook / JAR / Python) | — |
+| `PythonOperator` glue / bespoke script | notebook or Python task in a **Lakeflow Job** | — |
+| `TriggerDagRunOperator` / `ExternalTaskSensor` fan-out | **Lakeflow Jobs** run-job task + job parameters | — |
+| Dynamic task mapping (`.expand`) over a list | **Lakeflow Jobs** for-each task | — |
+| `BashOperator` shelling out to a script | native task (notebook / Python) driven by job parameters | — |
+| Custom logging / observability via XComs or a side table | **system tables (`system.lakeflow.*`) + native job notifications + AI/BI dashboard** | — |
+
+**Emit current names, not legacy ones:** Lakeflow Jobs (was Databricks Workflows), Lakeflow
+Declarative Pipelines (was Delta Live Tables/DLT), `AUTO CDC` (was `APPLY CHANGES INTO`), Declarative
+Automation Bundles (was Databricks Asset Bundles), AI/BI dashboards (was Lakeview), `system.lakeflow`
+(was `system.workflow`).
+
+**Then author the insights (shared method).** With this deep-dive and pattern vocabulary in hand,
+author and merge the `insights` object by following the source-neutral "Author and merge agentic
+insights" step in the parent `SKILL.md`. The insight schema and the authoring method are shared
+across sources; only the deep-dive and the construct mappings above are Airflow-specific.

src/flowx/adapter/__main__.py

@@ -87,6 +87,8 @@ def main(argv: list[str] | None = None) -> int:
         return _run_resolve_agentic(args)
     if args.command == "record-results":
         return _run_record_results(args)
+    if args.command == "enrich":
+        return _run_enrich(args)
     if args.command == "install-dashboard":
         return _run_install_dashboard(args)
     parser.print_help(sys.stderr)
@@ -163,6 +165,51 @@ def _run_record_results(args: argparse.Namespace) -> int:
     return 0
 
 
+def _run_enrich(args: argparse.Namespace) -> int:
+    """Implements ``enrich``: validate + merge agent-authored insights into inventory.json.
+
+    Returns 0 on success, 1 on any failure (missing inventory, unreadable/absent/
+    both payload sources, or validation violations).
+    """
+    from flowx.parser.pipeline_insights import enrich_inventory
+
+    metadata_dir = args.output_dir / "metadata"
+    if not (metadata_dir / "inventory.json").exists():
+        print(f"No inventory.json under {metadata_dir}; run the discover phase first.", file=sys.stderr)
+        return 1
+
+    inline: dict[str, Any] | None = None
+    if args.insights is not None:
+        try:
+            inline = json.loads(args.insights)
+        except json.JSONDecodeError as error:
+            print(f"Invalid --insights JSON: {error}", file=sys.stderr)
+            return 1
+    if (inline is None) == (args.insights_path is None):
+        print("Provide exactly one of --insights (inline JSON) or --insights-path.", file=sys.stderr)
+        return 1
+
+    try:
+        result = enrich_inventory(args.output_dir, insights=inline, insights_path=args.insights_path)
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Failed to enrich inventory: {error}", file=sys.stderr)
+        return 1
+
+    if not result["ok"]:
+        for violation in result["violations"]:
+            print(f"  - {violation}", file=sys.stderr)
+        print(
+            f"Insights validation failed ({len(result['violations'])} violation(s)); inventory not modified.",
+            file=sys.stderr,
+        )
+        return 1
+    print(
+        f"Enriched inventory: {result['pipeline_insights']} pipeline insight(s), "
+        f"{result['relationships']} relationship(s)."
+    )
+    return 0
+
+
 def _run_install_dashboard(args: argparse.Namespace) -> int:
     """Implements ``install-dashboard``: create + publish the coverage dashboard.
 
@@ -498,6 +545,29 @@ def _build_parser() -> argparse.ArgumentParser:
         help="SQL warehouse id for the write. Auto-detected (prefers running serverless) when omitted.",
     )
 
+    enrich = subparsers.add_parser(
+        "enrich",
+        help="Validate and merge agent-authored insights into metadata/inventory.json.",
+    )
+    enrich.add_argument(
+        "--output-dir",
+        type=Path,
+        required=True,
+        help="Migration output directory (reads/writes metadata/inventory.json).",
+    )
+    enrich.add_argument(
+        "--insights-path",
+        type=Path,
+        default=None,
+        help="Path to a JSON file holding the insights object.",
+    )
+    enrich.add_argument(
+        "--insights",
+        type=str,
+        default=None,
+        help="Insights object as an inline JSON string (convenience for direct CLI use).",
+    )
+
     dashboard = subparsers.add_parser(
         "install-dashboard",
         help="Create and publish an AI/BI dashboard visualizing coverage from the results table.",

src/flowx/mcp/runner.py

@@ -204,15 +204,45 @@ def materialize_adf_definitions(definitions: dict[str, Any]) -> str:
     return str(base)
 
 
-def cleanup_materialized(source: str) -> None:
-    """Remove a temp tree created by :func:`materialize_adf_definitions`.
+def materialize_json(obj: Any) -> str:
+    """Write a JSON-serialisable object to a temp file and return its path.
+
+    Lets the MCP server pass an inline ``insights`` dict to the adapter's
+    ``enrich`` subcommand (which reads from ``--insights-path``). Clean up with
+    :func:`cleanup_materialized`.
+    """
+    fd, path = tempfile.mkstemp(prefix="flowx-insights-", suffix=".json")
+    with os.fdopen(fd, "w", encoding="utf-8") as handle:
+        json.dump(obj, handle)
+    return path
+
 
-    Accepts either the returned directory or the single-file path (whose parent temp dir is
-    removed). Only paths under the system temp dir are deleted, as a safety guard.
+_TEMP_DIR_PREFIXES = ("flowx-adf-", "flowx-vol-", "flowx-ws-")
+
+
+def cleanup_materialized(source: str) -> None:
+    """Remove a temp tree/file created by :func:`materialize_adf_definitions`,
+    :func:`download_volume_dir`, :func:`download_workspace_dir`, or :func:`materialize_json`.
+
+    Accepts a temp directory path (from the ``mkdtemp`` helpers), a single-file path
+    inside such a directory (the single ARM-template case, whose parent temp dir is
+    removed), or a standalone temp file created directly in the system temp root
+    (from :func:`materialize_json`, whose file alone is removed -- never its parent).
+    Only paths under the system temp dir and carrying one of our prefixes are deleted,
+    as a safety guard.
     """
+    tmp_root = str(Path(tempfile.gettempdir()).resolve())
     path = Path(source)
+    # A standalone temp file we created directly in the temp root (materialize_json):
+    # remove just the file -- never its parent, which is the shared system temp root.
+    if path.is_file() and path.name.startswith("flowx-insights-"):
+        if str(path.resolve()).startswith(tmp_root):
+            path.unlink(missing_ok=True)
+        return
+    # Otherwise the temp dir to remove is the path itself (a mkdtemp dir) or, for the
+    # single ARM-template case, the file's parent temp dir.
     target = path if path.is_dir() else path.parent
-    if str(target.resolve()).startswith(str(Path(tempfile.gettempdir()).resolve())):
+    if target.name.startswith(_TEMP_DIR_PREFIXES) and str(target.resolve()).startswith(tmp_root):
         shutil.rmtree(target, ignore_errors=True)
 
 

src/flowx/mcp/server.py

@@ -492,9 +492,36 @@ def _cmd_install_dashboard(p: dict[str, Any]) -> dict[str, Any]:
     return {"ok": result.ok, "result": runner.parse_stdout_json(result), "process": result.as_dict()}
 
 
+def _parse_enrich_violations(stderr: str) -> list[str]:
+    """Extract the '  - <violation>' lines the adapter's enrich prints on failure."""
+    return [line[4:] for line in stderr.splitlines() if line.startswith("  - ")]
+
+
+def _cmd_enrich(p: dict[str, Any]) -> dict[str, Any]:
+    output_dir = p.get("output_dir", "./flowx_output")
+    insights = p.get("insights")
+    insights_path = p.get("insights_path")
+    if insights is None and not insights_path:
+        return {"ok": False, "error": "Provide 'insights' (inline dict) or 'insights_path'."}
+    tmp: str | None = None
+    try:
+        if insights is not None:
+            tmp = runner.materialize_json(insights)
+            insights_path = tmp
+        args: list[Any] = ["enrich", "--output-dir", output_dir, "--insights-path", insights_path]
+        result = runner.run_adapter(args)
+        out = Path(output_dir)
+        violations = _parse_enrich_violations(result.stderr) if not result.ok else None
+        return _phase_result(result, out, inventory=runner.summarize_inventory(out), violations=violations)
+    finally:
+        if tmp:
+            runner.cleanup_materialized(tmp)
+
+
 _COMMANDS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
     "inputs": _cmd_inputs,
     "discover": _cmd_discover,
+    "enrich": _cmd_enrich,
     "convert": _cmd_convert,
     "merge_agentic": _cmd_merge_agentic,
     "resolve_agentic": _cmd_resolve_agentic,
@@ -539,18 +566,15 @@ def flowx(command: str, parameters: dict[str, Any] | None = None) -> dict[str, A
         Airflow reads ``airflow_source_path`` (a DAG .py file or directory). ``package`` is
         source-independent (it consumes the translation report).
 
-        - "inputs": phase(req: "discover"|"convert"|"package"), source(req for discover/convert) —
-          list a phase's input prompts.
-        - "discover": source(req), one ADF source key | airflow_source_path (req), output_dir,
-          pipeline, exclude_dag | exclude_dags (Airflow, repeatable list) — parse and audit definitions.
-        - "convert": source(req), (one ADF source key | airflow_source_path), output_dir, pipeline,
-          exclude_dag | exclude_dags (Airflow, repeatable list).
-        - "merge_agentic": source(req: "adf"), report_path(req), agentic_results_dir(req), output_path —
-          merge ADF agent results. Airflow's legacy name-based merge is disabled; use resolve_agentic.
-        - "resolve_agentic": source(req: "airflow"), action(req: prepare | stage | apply), output_dir,
-          airflow_source_path, report_path, gap_id, candidates, replace, accept_gap | accept_gaps, accept_all,
-          review_complete, review_manifest, reset —
-          prepare, stage, and explicitly apply fingerprint-bound Airflow leaf-gap resolutions.
+        - "inputs": phase(req: "discover"|"convert"|"package") — list a phase's input prompts.
+        - "discover": one of adf_volume_path | adf_workspace_path | adf_definitions | adf_source_path
+          (req), output_dir, pipeline — parse ADF JSON, classify activities.
+        - "enrich": output_dir(req), one of insights(inline dict) | insights_path — validate + merge
+          agent-authored insights into metadata/inventory.json (returns {ok:false, ...} without writing
+          on validation failure).
+        - "convert": output_dir, (adf_volume_path | adf_workspace_path | adf_definitions |
+          adf_source_path), pipeline.
+        - "merge_agentic": report_path(req), agentic_results_dir(req), output_path — merge agent results.
         - "inspect": report_path(req) — return the full translation-option schema (every option with
           a `show_when` condition) for the agent to walk locally. See "Collecting options" below.
         - "apply_answers": report_path(req), answers(req, list of "ID=VALUE"), output_dir, lookup_csv.

src/flowx/models/adf_ast.py

@@ -297,13 +297,13 @@ def get_pipeline(self, name: str | None) -> AdfPipeline | None:
         if not name:
             return None
         lowered = name.lower()
-        ci_fallback = None
+        exact = None
         for pipeline in self.pipelines:
             if pipeline.name == name:
                 return pipeline
-            if ci_fallback is None and pipeline.name.lower() == lowered:
-                ci_fallback = pipeline
-        return ci_fallback
+            if exact is None and pipeline.name.lower() == lowered:
+                exact = pipeline
+        return exact
 
 
 # ---------------------------------------------------------------------------