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

AGENTS.md

@@ -75,6 +75,8 @@ intermediates under `.work/` (pruned by `package`).
 | `models/dab.py` | DAB output schema types |
 | `parser/adf_loader.py` | Parses ADF exports, produces `metadata/inventory.json` + `metadata/profile_report.csv` |
 | `parser/expression_parser.py` | Translates ADF expressions (@activity, @pipeline, @variables) |
+| `parser/dataset_resolvers.py` | Shared deterministic dataset-identity resolvers (schema.table / storage path), used by convert and discover |
+| `parser/lineage.py` | Deterministic control + data lineage extraction (`build_lineage`) surfaced in the inventory |
 | `translator/engine.py` | Registry dispatch, topological sort, context threading |
 | `translator/activity_translators/` | One module per deterministic activity type (16 total) |
 | `preparer/workflow_preparer.py` | Orchestrates activity preparers |

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.