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

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
 
 
 # ---------------------------------------------------------------------------

src/flowx/models/insights.py

@@ -0,0 +1,162 @@
+"""Agentic insights (discover phase) -- agent-authored judgment merged into inventory.json.
+
+References pipelines by name. Its cross-pipeline edges are either ANNOTATIONS of a
+deterministic ``Lineage`` edge (control/data -- carry no facts of their own) or an
+agent-INFERRED coupling the deterministic layer could not see (e.g. data flow that
+happens inside notebook code, an external trigger, a message queue). Inferred edges
+must cite their evidence and a confidence level so they are never mistaken for proven
+lineage.
+
+These models are **source-neutral**: they describe the shape of the ``insights`` object
+the agent authors, independent of whether the pipelines were discovered from ADF or
+Airflow. The validate/merge engine in ``flowx.parser.pipeline_insights`` works on the
+raw dict form; these dataclasses document the contract and back the unit tests.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Literal
+
+
+@dataclass(slots=True, kw_only=True)
+class LineageEdgeRef:
+    """A typed reference from a PipelineRelationship to one cross-pipeline edge.
+
+    Two tiers:
+
+    * ``"control"`` / ``"data"`` -- an **annotation** of a deterministic edge.
+      ``edge_identity`` echoes that edge verbatim (``ControlEdge.activity_name``
+      for control, ``DataEdge.match_key`` for data) so enrichment can resolve it
+      against the inventory's ``lineage``. ``evidence`` / ``confidence`` are not
+      used (the deterministic edge *is* the evidence, confidence is implicitly
+      high) and must be omitted.
+    * ``"inferred"`` -- an agent-asserted coupling the deterministic layer did
+      not find. There is no lineage edge to resolve against, so ``edge_identity``
+      is an agent-authored descriptor of what couples the pipelines (e.g. a
+      shared table or asset name), and ``evidence`` (why the agent believes the
+      coupling exists) plus ``confidence`` are **required**. This tier stays
+      pattern-agnostic: it does not encode *why* the deterministic layer missed
+      the edge, so it generalises to couplings flowx cannot yet see.
+
+    Attributes:
+        edge_type: The tier -- ``"control"``, ``"data"``, or ``"inferred"``.
+        edge_identity: For ``"control"`` the ``ControlEdge.activity_name``; for
+            ``"data"`` the ``DataEdge.match_key`` (both echoed verbatim from a
+            real edge); for ``"inferred"`` an agent-authored descriptor of the
+            coupling.
+        evidence: Inferred edges only -- the observable basis for the asserted
+            coupling. Required for ``"inferred"``; must be omitted otherwise.
+        confidence: Inferred edges only -- ``"high"`` / ``"medium"`` / ``"low"``.
+            Required for ``"inferred"``; must be omitted otherwise.
+    """
+
+    edge_type: Literal["control", "data", "inferred"]
+    edge_identity: str
+    evidence: str | None = None
+    confidence: Literal["high", "medium", "low"] | None = None
+
+
+@dataclass(slots=True, kw_only=True)
+class RecommendedPattern:
+    """One ranked Databricks target pattern recommended for a pipeline.
+
+    A pipeline insight carries 1-4 of these, ordered best-first, drawn from the
+    agent's *holistic* read of the pipeline and grounded in publicly-documented
+    Databricks capabilities. ``simplification_pattern`` ranks the distinctive
+    capabilities that collapse a legacy pattern ahead of like-for-like ports and
+    plain building blocks.
+
+    Attributes:
+        pattern: The named, publicly-documented Databricks capability (e.g.
+            ``"Lakeflow Connect SQL Server connector"``). Never an invented name.
+        fit: One line on why it fits this pipeline / what custom logic it replaces.
+        simplification_pattern: ``True`` *only* when the pattern uses a **distinctive**
+            Databricks capability that collapses or eliminates a whole legacy
+            pattern -- a managed connector (Lakeflow Connect), declarative CDC
+            (``AUTO CDC``), Auto Loader, or system tables replacing a home-grown
+            logging tier. ``False`` for a like-for-like port AND for plain native
+            building blocks that merely re-home the same work (a bare parameterized
+            Lakeflow Job, a for-each/run-job orchestrator, a plain Delta control
+            table, ``MERGE INTO``) -- "runs on Databricks" is not a simplification,
+            so reserve this flag for the capability that makes the old pattern
+            *disappear*. Rank the ``True`` patterns first.
+    """
+
+    pattern: str
+    fit: str
+    simplification_pattern: bool
+
+
+@dataclass(slots=True, kw_only=True)
+class SystemRecommendation:
+    """The single top-level architectural decision spanning the whole factory.
+
+    Per-pipeline ``recommended_patterns`` are chosen *under* this decision: the
+    system-level branch you pick (e.g. adopt a managed connector for an entire
+    extraction family) cascades into what each pipeline becomes, so it is authored
+    first and the per-pipeline patterns are kept consistent with it. It captures
+    the payoff a reader cannot see from any single pipeline card.
+
+    Attributes:
+        headline: One line naming the decision a migrator must make before any
+            per-pipeline work (e.g. "Managed ingestion collapses the extraction
+            factory").
+        recommended_patterns: 1-4 whole-system target architectures, ordered
+            best-first (the simplifying/native branch first), each a
+            :class:`RecommendedPattern`. ``recommended_patterns[0]`` is the
+            recommended branch; later entries are the ranked fallbacks.
+        cascade: What choosing ``recommended_patterns[0]`` collapses or eliminates
+            across the whole system (e.g. "5 child extractors -> managed connector
+            pipelines", "version-watermark CSV -> gone"). Empty when the decision
+            does not cascade.
+        decision_driver: The gating question that selects the branch (e.g. "Is the
+            Lakeflow Connect SQL Server connector GA/approved for this source?");
+            omit when there is no single deciding factor.
+    """
+
+    headline: str
+    recommended_patterns: list[RecommendedPattern] = field(default_factory=list)
+    cascade: list[str] = field(default_factory=list)
+    decision_driver: str | None = None
+
+
+@dataclass(slots=True, kw_only=True)
+class PipelineInsight:
+    """Per-pipeline judgment; references a pipeline by name (foreign key)."""
+
+    pipeline: str
+    pattern_name: str | None = None
+    intent: str | None = None
+    databricks_pattern: str | None = None
+    recommended_patterns: list[RecommendedPattern] = field(default_factory=list)
+    conversion_notes: list[str] = field(default_factory=list)
+    risk_if_ignored: str | None = None
+
+
+@dataclass(slots=True, kw_only=True)
+class PipelineRelationship:
+    """Cross-pipeline judgment.
+
+    Either annotates one deterministic lineage edge (``lineage_edge.edge_type``
+    is ``"control"`` / ``"data"``) or records an agent-inferred coupling the
+    deterministic layer could not see (``"inferred"``). Both endpoints are always
+    real pipeline names validated against the inventory.
+    """
+
+    from_pipeline: str
+    to_pipeline: str
+    lineage_edge: LineageEdgeRef
+    relationship_summary: str | None = None
+    databricks_pattern: str | None = None
+    risk_if_ignored: str | None = None
+
+
+@dataclass(slots=True, kw_only=True)
+class Insights:
+    """Agent-authored insights merged into inventory.json under the ``insights`` key."""
+
+    overview: str | None = None
+    system_recommendation: SystemRecommendation | None = None
+    pipeline_insights: list[PipelineInsight] = field(default_factory=list)
+    pipeline_relationships: list[PipelineRelationship] = field(default_factory=list)