diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index e41c463..48b905b 100644 --- a/.claude-plugin/plugin.json +++ b/.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" diff --git a/skills/flowx-discover/SKILL.md b/skills/flowx-discover/SKILL.md index b27e6f0..047e89d 100644 --- a/skills/flowx-discover/SKILL.md +++ b/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) diff --git a/skills/flowx-enrich/SKILL.md b/skills/flowx-enrich/SKILL.md new file mode 100644 index 0000000..e441cba --- /dev/null +++ b/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 `/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/.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": "", "insights": { ... }}) # inline object + flowx(command="enrich", parameters={"output_dir": "", "insights_path": ""}) + ``` + + 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="/src" + PY="$(cat /.migration-venv)" + "$PY" -m flowx.adapter enrich --output-dir --insights-path insights.json + ``` + + Both `--output-dir` and `--insights-path` are required; `--out ` 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. diff --git a/skills/flowx-enrich/insights.md b/skills/flowx-enrich/insights.md new file mode 100644 index 0000000..93db744 --- /dev/null +++ b/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": ""}, + "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. diff --git a/src/flowx/adapter/__main__.py b/src/flowx/adapter/__main__.py index 38545be..a2bb308 100644 --- a/src/flowx/adapter/__main__.py +++ b/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.", diff --git a/src/flowx/discovery_insights.py b/src/flowx/discovery_insights.py new file mode 100644 index 0000000..04adebc --- /dev/null +++ b/src/flowx/discovery_insights.py @@ -0,0 +1,462 @@ +"""Validate agent-authored insights against the discover inventory, then merge them in. + +The discover phase writes a purely deterministic ``metadata/inventory.json`` (source, +pipelines, per-pipeline ``lineage``, summary). An external agent then *authors* an +``insights`` object -- its judgment about factory-wide architecture, per-pipeline intent +and recommended Databricks patterns, and cross-pipeline relationships (see +:mod:`flowx.models.insights`). This module *enriches* the inventory: it validates the +authored JSON against the real inventory and, **only when clean**, adds a single additive +``insights`` key while leaving every existing key byte-identical. + +There is **no LLM here** -- the tool only validates and merges, mirroring the +author-then-validate-merge contract :mod:`flowx.agentic` uses for gap resolution. That +keeps the deterministic inventory trustworthy and every insight accountable: + +* every ``pipeline`` and every relationship endpoint must be a real pipeline in the + inventory (foreign-key validation); +* a ``control`` relationship edge must resolve to a real ``ControlEdge`` in the inventory's + ``lineage`` -- the full ``(from, to, via_task_key)`` triple, so the annotation connects + exactly the pipelines it claims, not merely some edge that shares a ``via_task_key``; +* an ``inferred`` edge has nothing to resolve against, so it must instead carry a non-empty + ``evidence`` string and a ``confidence`` level. + +The merge is **atomic** (temp file + ``os.replace``) and **idempotent**: it recomputes the +``inventory_sha256`` fingerprint from the deterministic inventory (with any prior ``insights`` +stripped) and *replaces* the whole ``insights`` block, so re-running with the same authored +insights rewrites byte-identical bytes and never stacks. The library owns ``schema_version`` +and ``inventory_sha256``; authored insights carrying either are rejected as unknown keys. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +from flowx.models.insights import ( + CONFIDENCE_LEVELS, + MAX_RECOMMENDED_PATTERNS, + SCHEMA_VERSION, +) + +# The single additive top-level key insights merge into. +INSIGHTS_KEY = "insights" + +# Library-owned keys injected on merge; authored insights must not supply them. +_SCHEMA_VERSION_KEY = "schema_version" +_FINGERPRINT_KEY = "inventory_sha256" + +_INSIGHTS_TOP_KEYS = {"overview", "system_recommendation", "pipeline_insights", "pipeline_relationships"} +_INSIGHT_KEYS = { + "pipeline", + "pattern_name", + "intent", + "databricks_pattern", + "recommended_patterns", + "conversion_notes", + "risk_if_ignored", +} +_RECOMMENDED_PATTERN_KEYS = {"pattern", "fit", "simplification_pattern"} +_SYSTEM_RECOMMENDATION_KEYS = {"headline", "recommended_patterns", "cascade", "decision_driver"} +_RELATIONSHIP_KEYS = { + "from_pipeline", + "to_pipeline", + "lineage_edge", + "relationship_summary", + "databricks_pattern", + "risk_if_ignored", +} +_EDGE_KEYS = {"edge_type", "edge_identity", "evidence", "confidence"} +_EDGE_TYPES = ("control", "inferred") + + +# --------------------------------------------------------------------------- # +# Inventory projections used for foreign-key + lineage-edge resolution. +# --------------------------------------------------------------------------- # + + +def _pipeline_names(inventory: dict[str, Any]) -> set[str]: + """The set of real pipeline names in the inventory (the foreign-key domain).""" + return { + str(pipeline["name"]) + for pipeline in inventory.get("pipelines", []) + if isinstance(pipeline, dict) and pipeline.get("name") is not None + } + + +def _control_edge_triples(inventory: dict[str, Any]) -> set[tuple[str, str, str]]: + """Real control edges as ``(source_workflow, target_workflow, via_task_key)`` triples. + + The unified inventory places lineage **per pipeline** (one block beside each pipeline's + ``activities``), so every pipeline's ``lineage.control_edges`` are gathered into one set. + Resolving on the full triple -- not the bare ``via_task_key`` -- pins a relationship to a + *specific* edge: a callee is often invoked from several callers, so a ``via_task_key`` + alone could match an edge between the wrong pair. + """ + triples: set[tuple[str, str, str]] = set() + for pipeline in inventory.get("pipelines", []): + if not isinstance(pipeline, dict): + continue + lineage = pipeline.get("lineage") or {} + for edge in lineage.get("control_edges", []): + if ( + isinstance(edge, dict) + and edge.get("source_workflow") is not None + and edge.get("target_workflow") is not None + and edge.get("via_task_key") is not None + ): + triples.add((str(edge["source_workflow"]), str(edge["target_workflow"]), str(edge["via_task_key"]))) + return triples + + +# --------------------------------------------------------------------------- # +# Validation. All violations are collected (never fail-fast) so the authoring +# agent can fix every problem in one pass. +# --------------------------------------------------------------------------- # + + +def validate_insights(raw: Any, inventory: dict[str, Any]) -> list[str]: + """Validate an authored insights payload against the inventory. + + Returns a list of human-readable violation strings; an empty list means the insights are + valid. Never raises on a malformed payload -- a non-dict payload is reported as a violation + so the caller can surface it the same way as every other problem. + """ + if not isinstance(raw, dict): + return [f"insights must be a JSON object, got {type(raw).__name__}"] + + violations: list[str] = [] + for key in sorted(set(raw) - _INSIGHTS_TOP_KEYS): + hint = " (set by the library, not the author)" if key in (_SCHEMA_VERSION_KEY, _FINGERPRINT_KEY) else "" + violations.append(f"unknown top-level key: {key!r}{hint}") + + overview = raw.get("overview") + if overview is not None and (not isinstance(overview, str) or not overview.strip()): + violations.append("'overview' must be a non-empty string when present") + + if "system_recommendation" in raw: + violations.extend(_validate_system_recommendation(raw["system_recommendation"])) + + names = _pipeline_names(inventory) + control_triples = _control_edge_triples(inventory) + + violations.extend(_validate_pipeline_insights(raw.get("pipeline_insights", []), names)) + violations.extend(_validate_relationships(raw.get("pipeline_relationships", []), names, control_triples)) + return violations + + +def _validate_pipeline_insights(insights: Any, names: set[str]) -> list[str]: + """Validate the ``pipeline_insights`` list: shape, unknown fields, and the pipeline FK.""" + if not isinstance(insights, list): + return ["'pipeline_insights' must be a list"] + violations: list[str] = [] + for index, item in enumerate(insights): + loc = f"pipeline_insights[{index}]" + if not isinstance(item, dict): + violations.append(f"{loc} must be an object") + continue + for key in sorted(set(item) - _INSIGHT_KEYS): + violations.append(f"{loc}: unknown field {key!r}") + name = item.get("pipeline") + if not name: + violations.append(f"{loc}: missing required field 'pipeline'") + elif name not in names: + violations.append(f"{loc}: pipeline {name!r} not in inventory") + if "recommended_patterns" in item: + violations.extend(_validate_recommended_patterns(item["recommended_patterns"], loc)) + return violations + + +def _validate_relationships( + relationships: Any, + names: set[str], + control_triples: set[tuple[str, str, str]], +) -> list[str]: + """Validate the ``pipeline_relationships`` list: endpoints (FK) + each ``lineage_edge``.""" + if not isinstance(relationships, list): + return ["'pipeline_relationships' must be a list"] + violations: list[str] = [] + for index, relationship in enumerate(relationships): + loc = f"pipeline_relationships[{index}]" + if not isinstance(relationship, dict): + violations.append(f"{loc} must be an object") + continue + for key in sorted(set(relationship) - _RELATIONSHIP_KEYS): + violations.append(f"{loc}: unknown field {key!r}") + from_pipeline = relationship.get("from_pipeline") + to_pipeline = relationship.get("to_pipeline") + for endpoint, value in (("from_pipeline", from_pipeline), ("to_pipeline", to_pipeline)): + if not value: + violations.append(f"{loc}: missing required field {endpoint!r}") + elif value not in names: + violations.append(f"{loc}: {endpoint} {value!r} not in inventory") + violations.extend( + _validate_edge(relationship.get("lineage_edge"), loc, from_pipeline, to_pipeline, control_triples) + ) + return violations + + +def _validate_edge( + edge: Any, + loc: str, + from_pipeline: Any, + to_pipeline: Any, + control_triples: set[tuple[str, str, str]], +) -> list[str]: + """Validate one ``lineage_edge`` reference. + + A ``control`` edge annotates a deterministic edge: the full ``(from, to, edge_identity)`` + triple must resolve against the inventory's lineage and the inferred-only ``evidence`` / + ``confidence`` keys must be **absent entirely** (an explicit ``null`` is still a violation -- + the deterministic edge *is* the evidence). An ``inferred`` edge asserts a coupling the + deterministic layer never found: nothing to resolve, but a non-empty ``evidence`` string + and a ``confidence`` level are required instead. + + Every problem on the edge is collected (never fail-fast), so a single edge that is wrong in + several ways -- e.g. an ``inferred`` edge with both an invalid identity and missing evidence + -- surfaces all its errors in one pass, matching the rest of the validator. + """ + if edge is None: + return [f"{loc}: missing required field 'lineage_edge'"] + if not isinstance(edge, dict): + return [f"{loc}.lineage_edge must be an object"] + problems: list[str] = [] + for key in sorted(set(edge) - _EDGE_KEYS): + problems.append(f"{loc}.lineage_edge: unknown field {key!r}") + + edge_type = edge.get("edge_type") + identity = edge.get("edge_identity") + if edge_type not in _EDGE_TYPES: + problems.append(f"{loc}.lineage_edge: edge_type must be 'control' or 'inferred', got {edge_type!r}") + if not isinstance(identity, str) or not identity: + problems.append(f"{loc}.lineage_edge: edge_identity must be a non-empty string") + + # Tier-specific checks run independently of the type/identity checks above so every problem + # on the edge is reported together rather than masked by an early return. + if edge_type == "inferred": + problems.extend(_validate_inferred_edge(edge, loc)) + elif edge_type == "control": + # Annotation tier: the inferred-only fields must be ABSENT (key not present), not merely + # non-null -- an explicit `evidence: null` / `confidence: null` is still a violation. + for inferred_only in ("evidence", "confidence"): + if inferred_only in edge: + problems.append(f"{loc}.lineage_edge: {inferred_only!r} is only valid on an 'inferred' edge") + # Resolve the full triple only when the identity and both endpoints are usable strings + # (a bad identity / endpoint is already reported here or by the caller). + if ( + isinstance(identity, str) + and identity + and isinstance(from_pipeline, str) + and isinstance(to_pipeline, str) + and (from_pipeline, to_pipeline, identity) not in control_triples + ): + problems.append( + f"{loc}.lineage_edge: control edge {identity!r} does not resolve to a lineage edge " + f"from {from_pipeline!r} to {to_pipeline!r}" + ) + return problems + + +def _validate_inferred_edge(edge: dict[str, Any], loc: str) -> list[str]: + """Validate the inferred-only fields: a non-empty ``evidence`` string + a ``confidence`` level.""" + problems: list[str] = [] + evidence = edge.get("evidence") + if not isinstance(evidence, str) or not evidence.strip(): + problems.append(f"{loc}.lineage_edge: an 'inferred' edge requires a non-empty 'evidence' string") + confidence = edge.get("confidence") + if confidence not in CONFIDENCE_LEVELS: + levels = ", ".join(repr(level) for level in CONFIDENCE_LEVELS) + problems.append( + f"{loc}.lineage_edge: an 'inferred' edge requires 'confidence' in {{{levels}}}, got {confidence!r}" + ) + return problems + + +def _validate_recommended_patterns(value: Any, loc: str) -> list[str]: + """Validate a ranked ``recommended_patterns`` list. + + When present it must hold 1-:data:`MAX_RECOMMENDED_PATTERNS` objects, ordered best-first. + Each requires a non-empty ``pattern`` and ``fit`` string and a boolean + ``simplification_pattern``. Shared by a pipeline's list and the system recommendation's + (``loc`` distinguishes them). All problems are collected. + """ + field_loc = f"{loc}.recommended_patterns" + if not isinstance(value, list): + return [f"{field_loc} must be a list"] + if not value: + return [ + f"{field_loc} must contain 1-{MAX_RECOMMENDED_PATTERNS} patterns when present " + f"(omit the field instead of sending an empty list)" + ] + problems: list[str] = [] + if len(value) > MAX_RECOMMENDED_PATTERNS: + problems.append(f"{field_loc} has {len(value)} patterns; at most {MAX_RECOMMENDED_PATTERNS} are allowed") + for index, pattern in enumerate(value): + pattern_loc = f"{field_loc}[{index}]" + if not isinstance(pattern, dict): + problems.append(f"{pattern_loc} must be an object") + continue + for key in sorted(set(pattern) - _RECOMMENDED_PATTERN_KEYS): + problems.append(f"{pattern_loc}: unknown field {key!r}") + for required in ("pattern", "fit"): + text = pattern.get(required) + if not isinstance(text, str) or not text.strip(): + problems.append(f"{pattern_loc}: {required!r} must be a non-empty string") + # A JSON bool parses to a Python bool; reject ints/strings so 1 / "yes" don't slip through. + if not isinstance(pattern.get("simplification_pattern"), bool): + problems.append( + f"{pattern_loc}: 'simplification_pattern' must be a boolean (true/false), " + f"got {type(pattern.get('simplification_pattern')).__name__}" + ) + return problems + + +def _validate_system_recommendation(value: Any) -> list[str]: + """Validate the optional top-level ``system_recommendation`` object. + + The one whole-factory decision, authored before per-pipeline insights. When present it must + be an object with a non-empty ``headline`` and a ``recommended_patterns`` ranked list. + ``cascade`` (non-empty strings) and ``decision_driver`` (the gating question) are optional. + All problems are collected. + """ + loc = "system_recommendation" + if not isinstance(value, dict): + return [f"{loc} must be an object"] + problems: list[str] = [] + for key in sorted(set(value) - _SYSTEM_RECOMMENDATION_KEYS): + problems.append(f"{loc}: unknown field {key!r}") + headline = value.get("headline") + if not isinstance(headline, str) or not headline.strip(): + problems.append(f"{loc}: 'headline' must be a non-empty string") + if "recommended_patterns" not in value: + problems.append(f"{loc}: missing required field 'recommended_patterns'") + else: + problems.extend(_validate_recommended_patterns(value["recommended_patterns"], loc)) + cascade = value.get("cascade") + if cascade is not None and ( + not isinstance(cascade, list) or not all(isinstance(item, str) and item.strip() for item in cascade) + ): + problems.append(f"{loc}: 'cascade' must be a list of non-empty strings when present") + driver = value.get("decision_driver") + if driver is not None and (not isinstance(driver, str) or not driver.strip()): + problems.append(f"{loc}: 'decision_driver' must be a non-empty string when present") + return problems + + +# --------------------------------------------------------------------------- # +# Loading, fingerprinting, and the atomic idempotent merge. +# --------------------------------------------------------------------------- # + + +def load_insights(*, insights: dict[str, Any] | None = None, insights_path: Path | None = None) -> dict[str, Any]: + """Return the raw authored insights dict from exactly one source (inline or file). + + Raises: + ValueError: if neither or both sources are provided. + """ + if (insights is None) == (insights_path is None): + raise ValueError("provide exactly one of 'insights' (inline dict) or 'insights_path'") + if insights is not None: + return insights + assert insights_path is not None # guaranteed by the guard above + return json.loads(insights_path.read_text(encoding="utf-8")) + + +def _base_inventory(inventory: dict[str, Any]) -> dict[str, Any]: + """The deterministic inventory with any previously-merged ``insights`` block stripped. + + Fingerprinting and re-serialisation both work off this so re-enriching an already-enriched + inventory is stable: the fingerprint reflects only the deterministic layer, never a prior + ``insights`` block. + """ + return {key: value for key, value in inventory.items() if key != INSIGHTS_KEY} + + +def inventory_fingerprint(inventory: dict[str, Any]) -> str: + """A stable SHA-256 over the deterministic inventory (any ``insights`` block excluded). + + Canonicalised with ``sort_keys`` so the digest is independent of key insertion order -- + it binds the authored insights to *what the inventory says*, not to a particular byte layout. + """ + import hashlib + + canonical = json.dumps(_base_inventory(inventory), sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(canonical).hexdigest() + + +def merge_into_inventory(inventory: dict[str, Any], raw: dict[str, Any]) -> dict[str, Any]: + """Return a new inventory dict with exactly one additive ``insights`` key. + + Does not mutate the input and performs no I/O. Existing keys are preserved in their original + order and values, so re-serialising them is byte-identical to the deterministic write. The + ``insights`` block is the authored content plus the library-owned ``schema_version`` and + ``inventory_sha256`` fingerprint, and *replaces* any prior block (idempotent). + """ + base = _base_inventory(inventory) + block: dict[str, Any] = {_SCHEMA_VERSION_KEY: SCHEMA_VERSION, _FINGERPRINT_KEY: inventory_fingerprint(base)} + for key in ("overview", "system_recommendation", "pipeline_insights", "pipeline_relationships"): + if key in raw: + block[key] = raw[key] + base[INSIGHTS_KEY] = block + return base + + +def _write_inventory_atomic(path: Path, inventory: dict[str, Any]) -> None: + """Write the inventory JSON atomically, matching the deterministic write's formatting. + + Uses ``json.dumps(..., indent=2)`` with no trailing newline -- exactly how the discover + phase writes ``inventory.json`` -- so every pre-existing key stays byte-identical. The temp + file + ``os.replace`` keeps the on-disk inventory intact if the process dies mid-write. + """ + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text(json.dumps(inventory, indent=2), encoding="utf-8") + os.replace(temporary, path) + + +def enrich_inventory( + output_dir: Path, + *, + insights: dict[str, Any] | None = None, + insights_path: Path | None = None, +) -> dict[str, Any]: + """Validate authored insights against the inventory, then merge them in on success. + + Reads ``/metadata/inventory.json``, validates the authored insights, and -- only + when there are no violations -- writes the merged inventory back atomically (adding just the + additive ``insights`` key, every existing key byte-unchanged). On any validation failure the + inventory file is left untouched. + + Provide the authored insights via exactly one of ``insights`` (an inline dict) or + ``insights_path`` (a JSON file). + + Returns a result dict ``{"ok", "violations", "inventory_sha256", "pipeline_insights", + "relationships"}``. ``ok`` is ``False`` (and the file untouched) when there are violations. + + Raises: + FileNotFoundError: when ``inventory.json`` does not exist (run discover first). + ValueError: when neither or both insight sources are provided, or the inventory file is + not a JSON object. + """ + inventory_path = Path(output_dir) / "metadata" / "inventory.json" + if not inventory_path.exists(): + raise FileNotFoundError(f"No inventory.json under {inventory_path.parent}; run the discover phase first.") + inventory = json.loads(inventory_path.read_text(encoding="utf-8")) + if not isinstance(inventory, dict): + raise ValueError(f"inventory.json must contain a JSON object, got {type(inventory).__name__}") + + raw = load_insights(insights=insights, insights_path=insights_path) + violations = validate_insights(raw, inventory) + if violations: + return {"ok": False, "violations": violations, "pipeline_insights": 0, "relationships": 0} + + merged = merge_into_inventory(inventory, raw) + _write_inventory_atomic(inventory_path, merged) + return { + "ok": True, + "violations": [], + "inventory_sha256": merged[INSIGHTS_KEY][_FINGERPRINT_KEY], + "pipeline_insights": len(raw.get("pipeline_insights", [])), + "relationships": len(raw.get("pipeline_relationships", [])), + } diff --git a/src/flowx/mcp/server.py b/src/flowx/mcp/server.py index 36c67fc..0651640 100644 --- a/src/flowx/mcp/server.py +++ b/src/flowx/mcp/server.py @@ -51,6 +51,7 @@ def _transport_security() -> TransportSecuritySettings: Typical flow (ADF shown; swap source + source-path for Airflow): flowx("inputs", {"phase": "discover", "source": "adf"}) # learn a phase's inputs flowx("discover", {"source": "adf", "adf_source_path": "...", "output_dir": "..."}) + flowx("enrich", {"output_dir": "...", "insights": {...}}) # optional: merge agent-authored insights flowx("convert", {"source": "adf", "output_dir": "..."}) flowx("inspect", {"report_path": "/.work/translation_report.json"}) flowx("apply_answers", {"report_path": "...", "answers": ["id=value"], "output_dir": "..."}) @@ -296,6 +297,35 @@ def _cmd_resolve_agentic(p: dict[str, Any]) -> dict[str, Any]: return {"ok": result.ok, "process": result.as_dict(), **extra} +def _cmd_enrich(p: dict[str, Any]) -> dict[str, Any]: + """Validate agent-authored insights and merge them into inventory.json. + + Accepts the insights either inline as ``insights`` (a JSON object) or via ``insights_path`` + (a file the server can read); exactly one is required. Inline insights are staged to a temp + file so the same ``enrich`` CLI contract runs on both paths. The returned ``ok`` reflects + *validation* success -- ``result.violations`` lists every problem when it is False, and the + inventory is left untouched on any failure. + """ + output_dir = p.get("output_dir", "./flowx_output") + insights = p.get("insights") + insights_path = p.get("insights_path") + if (insights is None) == (insights_path is None): + return {"ok": False, "error": "provide exactly one of 'insights' (inline object) or 'insights_path'."} + + def _run(path: str) -> dict[str, Any]: + result = runner.run_adapter(["enrich", "--output-dir", output_dir, "--insights-path", path]) + payload = runner.parse_stdout_json(result) + ok = bool(isinstance(payload, dict) and payload.get("ok")) + return {"ok": ok, "result": payload, "process": result.as_dict()} + + if insights_path is not None: + return _run(str(insights_path)) + with tempfile.TemporaryDirectory(prefix="flowx-insights-") as temporary: + inline_path = Path(temporary) / "insights.json" + inline_path.write_text(json.dumps(insights, indent=2), encoding="utf-8") + return _run(str(inline_path)) + + def _cmd_inspect(p: dict[str, Any]) -> dict[str, Any]: args: list[Any] = ["inspect", p["report_path"]] for answer in p.get("answers") or []: @@ -498,6 +528,7 @@ def _cmd_install_dashboard(p: dict[str, Any]) -> dict[str, Any]: "convert": _cmd_convert, "merge_agentic": _cmd_merge_agentic, "resolve_agentic": _cmd_resolve_agentic, + "enrich": _cmd_enrich, "inspect": _cmd_inspect, "apply_answers": _cmd_apply_answers, "materialize_lookup": _cmd_materialize_lookup, @@ -551,6 +582,11 @@ def flowx(command: str, parameters: dict[str, Any] | None = None) -> dict[str, A 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. + - "enrich": output_dir(req), one of insights(inline object) | insights_path(req) — validate + agent-authored discover insights against inventory.json and merge them under one additive + `insights` key (atomic, idempotent). `ok` reflects validation; `result.violations` lists any + problems and the inventory is left untouched on failure. Author the insights by reading + inventory.json + the source artifacts first (see the flowx-discover skill's insights guide). - "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. diff --git a/src/flowx/models/insights.py b/src/flowx/models/insights.py new file mode 100644 index 0000000..253093c --- /dev/null +++ b/src/flowx/models/insights.py @@ -0,0 +1,207 @@ +"""Agentic insights (discover phase) -- agent-authored judgment merged into ``inventory.json``. + +The deterministic discover layer captures what a source workflow *is*; these models +capture what to *do* about it -- the judgment the deterministic pass can never derive: +a factory-wide architectural recommendation, per-pipeline intent + recommended +Databricks patterns, and how pipelines couple. An external agent *authors* this object +by reading the inventory and the source artifacts; the library only validates and merges +it (see :mod:`flowx.discovery_insights`). There is **no LLM in the tool** -- the same +author-then-validate-merge contract :mod:`flowx.agentic` uses for gap resolution. + +These models are **source-neutral**: they describe the shape of the ``insights`` object +independent of whether the pipelines came from ADF or Airflow, because the inventory they +attach to is itself standardised across sources via the shared discovery AST (#61/#62). +The validate/merge engine works on the raw dict form; these dataclasses document the +contract and back the unit tests. + +Cross-pipeline couplings come in two accountable tiers (:class:`LineageEdgeRef`): + +* ``control`` -- an **annotation** of a deterministic control edge already proven in the + inventory's per-pipeline ``lineage`` block (an ``ExecutePipeline`` / run-job invocation). + It carries no facts of its own; enrichment resolves it against a real ``ControlEdge``. +* ``inferred`` -- a coupling the deterministic layer could not see (e.g. data flow buried + in notebook code, an external trigger, a shared table the parser never resolved). There + is nothing to resolve against, so it must cite ``evidence`` and a ``confidence`` level and + is never mistaken for proven lineage. + +Deterministic cross-pipeline **data** edges are intentionally *not* a tier here: the shared +``DataEdge`` is task-endpoint and intra-pipeline only, so a cross-pipeline data coupling has +no deterministic edge to annotate and must ride the ``inferred`` tier until cross-pipeline +data lineage lands in a later issue. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Literal + +# The insights schema version stamped onto the merged block. Bump on any +# backwards-incompatible change to the authored shape. +SCHEMA_VERSION = "1" + +# Cap on a ranked ``recommended_patterns`` list (per pipeline and system-wide). A short, +# ranked shortlist keeps the recommendation legible; an unbounded list is noise. +MAX_RECOMMENDED_PATTERNS = 4 + +# The confidence levels an ``inferred`` edge may carry. +CONFIDENCE_LEVELS: tuple[str, ...] = ("high", "medium", "low") + + +@dataclass(slots=True, kw_only=True) +class LineageEdgeRef: + """A typed reference from a :class:`PipelineRelationship` to how two pipelines couple. + + Two tiers, validated differently by :mod:`flowx.discovery_insights`: + + * ``"control"`` -- an **annotation** of a deterministic control edge. ``edge_identity`` + echoes that edge's ``ControlEdge.via_task_key`` verbatim so enrichment can resolve the + full ``(from_pipeline, to_pipeline, edge_identity)`` triple against the inventory's + ``lineage``. ``evidence`` / ``confidence`` are unused (the proven edge *is* the + evidence) and must be omitted. + * ``"inferred"`` -- a coupling the deterministic layer never found. 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 name), and ``evidence`` (why the agent + believes the coupling exists) plus ``confidence`` are **required**. + + Attributes: + edge_type: The tier -- ``"control"`` or ``"inferred"``. + edge_identity: For ``"control"`` the ``ControlEdge.via_task_key`` 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", "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 or the whole factory. + + A recommendation carries 1-:data:`MAX_RECOMMENDED_PATTERNS` of these, ordered best-first, + drawn from the agent's holistic read 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 / 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. + 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-:data:`MAX_RECOMMENDED_PATTERNS` whole-system target + architectures, ordered best-first (the simplifying/native branch first). + ``recommended_patterns[0]`` is the recommended branch; later entries are ranked + fallbacks. + cascade: What choosing ``recommended_patterns[0]`` collapses or eliminates across the + whole system (e.g. "5 child extractors -> managed connector pipelines"). 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 (a foreign key to the inventory). + + Attributes: + pipeline: Name of the pipeline this insight annotates. Must be a real pipeline in the + inventory (validated on enrichment). + pattern_name: A short label naming the pipeline's recognised shape, when one applies. + intent: What the pipeline is really trying to accomplish, in business terms. + databricks_pattern: The single headline Databricks pattern the pipeline maps to. + recommended_patterns: 1-:data:`MAX_RECOMMENDED_PATTERNS` ranked target patterns, + best-first, chosen under the factory-wide :class:`SystemRecommendation`. + conversion_notes: Concrete notes a migrator should heed when converting. + risk_if_ignored: What breaks or degrades if the recommendation is not followed. + """ + + 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; both endpoints are real inventory pipeline names. + + Either annotates one deterministic control edge (``lineage_edge.edge_type == "control"``) + or records an agent-inferred coupling the deterministic layer could not see + (``"inferred"``). + + Attributes: + from_pipeline: Source endpoint -- the caller (control) or upstream (inferred) pipeline. + to_pipeline: Target endpoint -- the callee (control) or downstream (inferred) pipeline. + lineage_edge: The typed edge reference describing and grounding the coupling. + relationship_summary: One line on how the two pipelines relate. + databricks_pattern: The Databricks construct that should carry this coupling. + risk_if_ignored: What breaks if the coupling is dropped in the migration. + """ + + 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 additive ``insights`` key. + + The agent authors only these four content fields. The library injects the ``schema_version`` + and the ``inventory_sha256`` fingerprint on merge (see :mod:`flowx.discovery_insights`), so + the authored insights stay bound to the exact deterministic inventory they describe. + + Attributes: + overview: A short factory-wide narrative -- what this collection of pipelines is. + system_recommendation: The one whole-factory architectural decision. + pipeline_insights: Per-pipeline judgments, one per pipeline the agent chose to annotate. + pipeline_relationships: Cross-pipeline couplings (control annotations + inferred). + """ + + 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) diff --git a/tests/unit/test_discovery_insights.py b/tests/unit/test_discovery_insights.py new file mode 100644 index 0000000..f0153fb --- /dev/null +++ b/tests/unit/test_discovery_insights.py @@ -0,0 +1,479 @@ +"""Tests for agent-authored discover insights: models, validation, and the atomic merge. + +The inventory fixtures are built by hand through the source-agnostic emitter +(:func:`flowx.discovery_inventory.build_source_inventory`) -- no ADF, no Airflow -- so the +insights engine is proven against the standardised inventory shape, control-edge lineage and +all, exactly as it will see it in production. +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Any + +import pytest + +from flowx.adapter.__main__ import main as adapter_cli_main +from flowx.discovery_insights import ( + INSIGHTS_KEY, + enrich_inventory, + inventory_fingerprint, + load_insights, + merge_into_inventory, + validate_insights, +) +from flowx.discovery_inventory import STRATEGY_PROPERTY, build_source_inventory +from flowx.models.discovery import CONCEPT_NOTEBOOK, CONCEPT_RUN_WORKFLOW, SourceGraph, SourceNode +from flowx.models.insights import ( + Insights, + LineageEdgeRef, + PipelineInsight, + PipelineRelationship, + RecommendedPattern, + SystemRecommendation, +) +from flowx.models.ir import ControlEdge, Lineage + +# --------------------------------------------------------------------------- # +# Fixtures: a two-pipeline factory where "parent" invokes "child" via a proven +# control edge, plus a standalone "sibling" for inferred-coupling tests. +# --------------------------------------------------------------------------- # + + +def _node(task_key: str, native_type: str, concept: str = CONCEPT_NOTEBOOK) -> SourceNode: + return SourceNode( + source_id=task_key, + task_key=task_key, + concept=concept, + source="unit", + name=task_key, + native_type=native_type, + properties={STRATEGY_PROPERTY: "deterministic"}, + raw={"name": task_key, "type": native_type}, + ) + + +def _inventory() -> dict[str, Any]: + parent = SourceGraph( + name="parent", + source="unit", + tasks=[_node("call_child", "ExecutePipeline", CONCEPT_RUN_WORKFLOW)], + lineage=Lineage( + control_edges=[ControlEdge(source_workflow="parent", target_workflow="child", via_task_key="call_child")] + ), + ) + child = SourceGraph(name="child", source="unit", tasks=[_node("load", "Notebook")]) + sibling = SourceGraph(name="sibling", source="unit", tasks=[_node("export", "Notebook")]) + return build_source_inventory([parent, child, sibling], source="unit", source_dir="/tmp/src") + + +def _valid_insights() -> dict[str, Any]: + return { + "overview": "A parent orchestrates a child extractor; a sibling exports downstream.", + "system_recommendation": { + "headline": "Collapse the extraction factory onto a managed connector", + "recommended_patterns": [ + { + "pattern": "Lakeflow Connect SQL Server connector", + "fit": "Replaces the child extractor", + "simplification_pattern": True, + }, + { + "pattern": "Parameterised Lakeflow Job", + "fit": "Like-for-like orchestration", + "simplification_pattern": False, + }, + ], + "cascade": ["child extractor -> managed connector pipeline"], + "decision_driver": "Is the Lakeflow Connect connector GA for this source?", + }, + "pipeline_insights": [ + { + "pipeline": "child", + "intent": "Extract a table into the lake", + "databricks_pattern": "Managed ingestion", + "recommended_patterns": [ + {"pattern": "Lakeflow Connect", "fit": "Managed CDC ingestion", "simplification_pattern": True} + ], + "conversion_notes": ["Point the connector at the same source"], + "risk_if_ignored": "Bespoke extractor code carries forward", + } + ], + "pipeline_relationships": [ + { + "from_pipeline": "parent", + "to_pipeline": "child", + "lineage_edge": {"edge_type": "control", "edge_identity": "call_child"}, + "relationship_summary": "parent runs child", + }, + { + "from_pipeline": "child", + "to_pipeline": "sibling", + "lineage_edge": { + "edge_type": "inferred", + "edge_identity": "shared table sales.curated", + "evidence": "Both notebooks read/write sales.curated in their code", + "confidence": "medium", + }, + }, + ], + } + + +def _write_inventory(output_dir: Path, inventory: dict[str, Any]) -> Path: + """Write inventory.json exactly as the discover phase does (json.dumps(indent=2), no newline).""" + metadata = output_dir / "metadata" + metadata.mkdir(parents=True, exist_ok=True) + path = metadata / "inventory.json" + path.write_text(json.dumps(inventory, indent=2), encoding="utf-8") + return path + + +# --------------------------------------------------------------------------- # +# Models. +# --------------------------------------------------------------------------- # + + +def test_models_construct_and_document_the_contract() -> None: + insights = Insights( + overview="o", + system_recommendation=SystemRecommendation( + headline="h", + recommended_patterns=[RecommendedPattern(pattern="p", fit="f", simplification_pattern=True)], + ), + pipeline_insights=[PipelineInsight(pipeline="child", intent="i")], + pipeline_relationships=[ + PipelineRelationship( + from_pipeline="parent", + to_pipeline="child", + lineage_edge=LineageEdgeRef(edge_type="control", edge_identity="call_child"), + ) + ], + ) + assert insights.pipeline_insights[0].pipeline == "child" + assert insights.pipeline_relationships[0].lineage_edge.edge_type == "control" + + +# --------------------------------------------------------------------------- # +# Validation: success. +# --------------------------------------------------------------------------- # + + +def test_valid_insights_pass_validation() -> None: + assert validate_insights(_valid_insights(), _inventory()) == [] + + +# --------------------------------------------------------------------------- # +# Validation: failure modes. +# --------------------------------------------------------------------------- # + + +def test_non_dict_payload_is_a_violation() -> None: + violations = validate_insights([1, 2, 3], _inventory()) + assert violations == ["insights must be a JSON object, got list"] + + +def test_unknown_pipeline_reference_in_insight() -> None: + raw = _valid_insights() + raw["pipeline_insights"][0]["pipeline"] = "ghost" + violations = validate_insights(raw, _inventory()) + assert any("pipeline 'ghost' not in inventory" in v for v in violations) + + +def test_unknown_pipeline_reference_in_relationship_endpoint() -> None: + raw = _valid_insights() + raw["pipeline_relationships"][0]["to_pipeline"] = "ghost" + violations = validate_insights(raw, _inventory()) + assert any("to_pipeline 'ghost' not in inventory" in v for v in violations) + + +def test_control_edge_not_matching_inventory_lineage() -> None: + raw = _valid_insights() + # Wrong via_task_key -- no such control edge from parent to child. + raw["pipeline_relationships"][0]["lineage_edge"]["edge_identity"] = "not_a_real_task" + violations = validate_insights(raw, _inventory()) + assert any("does not resolve to a lineage edge from 'parent' to 'child'" in v for v in violations) + + +def test_control_edge_with_wrong_endpoints_does_not_resolve() -> None: + raw = _valid_insights() + # The via_task_key is real, but between parent->child, not child->sibling. + raw["pipeline_relationships"][0]["from_pipeline"] = "child" + raw["pipeline_relationships"][0]["to_pipeline"] = "sibling" + violations = validate_insights(raw, _inventory()) + assert any("call_child' does not resolve" in v for v in violations) + + +def test_control_edge_may_not_carry_evidence_or_confidence() -> None: + raw = _valid_insights() + raw["pipeline_relationships"][0]["lineage_edge"]["evidence"] = "nope" + raw["pipeline_relationships"][0]["lineage_edge"]["confidence"] = "high" + violations = validate_insights(raw, _inventory()) + assert any("'evidence' is only valid on an 'inferred' edge" in v for v in violations) + assert any("'confidence' is only valid on an 'inferred' edge" in v for v in violations) + + +def test_control_edge_rejects_present_but_null_evidence_or_confidence() -> None: + """The inferred-only keys are forbidden by PRESENCE -- an explicit ``null`` is still a violation. + + A deterministic control edge annotates a proven lineage edge, so it must not carry these keys + at all; ``evidence: null`` / ``confidence: null`` must not slip through as "not set". + """ + raw = _valid_insights() + raw["pipeline_relationships"][0]["lineage_edge"]["evidence"] = None + raw["pipeline_relationships"][0]["lineage_edge"]["confidence"] = None + violations = validate_insights(raw, _inventory()) + assert any("'evidence' is only valid on an 'inferred' edge" in v for v in violations) + assert any("'confidence' is only valid on an 'inferred' edge" in v for v in violations) + + +def test_inferred_edge_requires_evidence_and_confidence() -> None: + raw = _valid_insights() + edge = raw["pipeline_relationships"][1]["lineage_edge"] + del edge["evidence"] + edge["confidence"] = "certain" + violations = validate_insights(raw, _inventory()) + assert any("requires a non-empty 'evidence' string" in v for v in violations) + assert any("requires 'confidence' in" in v for v in violations) + + +def test_inferred_edge_aggregates_identity_and_evidence_errors() -> None: + """A single edge wrong in several ways surfaces ALL its errors, never fail-fast. + + An inferred edge with both an invalid (empty) ``edge_identity`` AND missing + evidence/confidence must report the identity error *and* the evidence/confidence errors in + one pass -- not just the first. + """ + raw = _valid_insights() + raw["pipeline_relationships"][1]["lineage_edge"] = {"edge_type": "inferred", "edge_identity": ""} + violations = validate_insights(raw, _inventory()) + assert any("edge_identity must be a non-empty string" in v for v in violations) + assert any("requires a non-empty 'evidence' string" in v for v in violations) + assert any("requires 'confidence' in" in v for v in violations) + + +def test_unknown_edge_type_is_rejected() -> None: + raw = _valid_insights() + raw["pipeline_relationships"][0]["lineage_edge"]["edge_type"] = "data" + violations = validate_insights(raw, _inventory()) + assert any("edge_type must be 'control' or 'inferred'" in v for v in violations) + + +def test_unknown_top_level_key_including_library_owned_fields() -> None: + raw = _valid_insights() + raw["schema_version"] = "1" + raw["nonsense"] = True + violations = validate_insights(raw, _inventory()) + assert any("unknown top-level key: 'schema_version' (set by the library, not the author)" in v for v in violations) + assert any("unknown top-level key: 'nonsense'" in v for v in violations) + + +def test_recommended_patterns_cap_and_shape() -> None: + raw = _valid_insights() + raw["pipeline_insights"][0]["recommended_patterns"] = [ + {"pattern": f"p{i}", "fit": "f", "simplification_pattern": False} for i in range(5) + ] + violations = validate_insights(raw, _inventory()) + assert any("at most 4 are allowed" in v for v in violations) + + +def test_recommended_pattern_simplification_flag_must_be_boolean() -> None: + raw = _valid_insights() + raw["pipeline_insights"][0]["recommended_patterns"][0]["simplification_pattern"] = "yes" + violations = validate_insights(raw, _inventory()) + assert any("'simplification_pattern' must be a boolean" in v for v in violations) + + +def test_empty_recommended_patterns_list_is_rejected() -> None: + raw = _valid_insights() + raw["pipeline_insights"][0]["recommended_patterns"] = [] + violations = validate_insights(raw, _inventory()) + assert any("must contain 1-4 patterns when present" in v for v in violations) + + +def test_system_recommendation_requires_headline_and_patterns() -> None: + raw = _valid_insights() + raw["system_recommendation"] = {"cascade": ["x"]} + violations = validate_insights(raw, _inventory()) + assert any("'headline' must be a non-empty string" in v for v in violations) + assert any("missing required field 'recommended_patterns'" in v for v in violations) + + +# --------------------------------------------------------------------------- # +# Merge: fingerprint, additive key, byte-compat, atomicity, idempotency. +# --------------------------------------------------------------------------- # + + +def test_merge_adds_single_additive_block_with_fingerprint_and_schema_version() -> None: + inventory = _inventory() + merged = merge_into_inventory(inventory, _valid_insights()) + # Original keys are untouched and one additive key is appended, last. + assert list(merged) == ["source", "source_dir", "pipelines", "summary", "insights"] + block = merged[INSIGHTS_KEY] + assert block["schema_version"] == "1" + assert block["inventory_sha256"] == inventory_fingerprint(inventory) + assert "overview" in block and "pipeline_relationships" in block + # merge does not mutate the input. + assert INSIGHTS_KEY not in inventory + + +def test_enrich_writes_additive_block_and_leaves_existing_keys_byte_identical(tmp_path: Path) -> None: + inventory = _inventory() + path = _write_inventory(tmp_path, inventory) + original_bytes = path.read_bytes() + + result = enrich_inventory(tmp_path, insights=_valid_insights()) + assert result["ok"] is True + assert result["pipeline_insights"] == 1 + assert result["relationships"] == 2 + + enriched = json.loads(path.read_text(encoding="utf-8")) + assert INSIGHTS_KEY in enriched + # Every existing key is byte-identical: strip the additive block and re-serialise. + stripped = {k: v for k, v in enriched.items() if k != INSIGHTS_KEY} + assert json.dumps(stripped, indent=2).encode("utf-8") == original_bytes + + +def test_enrich_leaves_inventory_untouched_on_validation_failure(tmp_path: Path) -> None: + path = _write_inventory(tmp_path, _inventory()) + original_bytes = path.read_bytes() + + raw = _valid_insights() + raw["pipeline_insights"][0]["pipeline"] = "ghost" + result = enrich_inventory(tmp_path, insights=raw) + + assert result["ok"] is False + assert any("ghost" in v for v in result["violations"]) + assert path.read_bytes() == original_bytes # untouched + + +def test_enrich_is_idempotent_and_replaces_prior_block(tmp_path: Path) -> None: + path = _write_inventory(tmp_path, _inventory()) + + first = enrich_inventory(tmp_path, insights=_valid_insights()) + after_first = path.read_bytes() + # Re-running with the same authored insights rewrites byte-identical content. + second = enrich_inventory(tmp_path, insights=_valid_insights()) + assert path.read_bytes() == after_first + assert first["inventory_sha256"] == second["inventory_sha256"] + + # Enriching with different insights replaces (not stacks) the block; fingerprint unchanged + # because the deterministic base is the same. + changed = _valid_insights() + changed["overview"] = "A different narrative" + third = enrich_inventory(tmp_path, insights=changed) + enriched = json.loads(path.read_text(encoding="utf-8")) + assert enriched[INSIGHTS_KEY]["overview"] == "A different narrative" + assert third["inventory_sha256"] == first["inventory_sha256"] + # Still exactly one insights block. + assert list(enriched).count(INSIGHTS_KEY) == 1 + + +def test_fingerprint_ignores_any_prior_insights_block() -> None: + inventory = _inventory() + base_fingerprint = inventory_fingerprint(inventory) + enriched = merge_into_inventory(inventory, _valid_insights()) + # Fingerprinting the already-enriched inventory yields the same digest (insights excluded). + assert inventory_fingerprint(enriched) == base_fingerprint + + +def test_enrich_raises_when_inventory_missing(tmp_path: Path) -> None: + with pytest.raises(FileNotFoundError): + enrich_inventory(tmp_path, insights=_valid_insights()) + + +def test_load_insights_requires_exactly_one_source(tmp_path: Path) -> None: + with pytest.raises(ValueError): + load_insights() + with pytest.raises(ValueError): + load_insights(insights={}, insights_path=tmp_path / "x.json") + path = tmp_path / "insights.json" + path.write_text(json.dumps({"overview": "hi"}), encoding="utf-8") + assert load_insights(insights_path=path) == {"overview": "hi"} + + +# --------------------------------------------------------------------------- # +# CLI wiring. +# --------------------------------------------------------------------------- # + + +def test_cli_enrich_success(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + _write_inventory(tmp_path, _inventory()) + insights_path = tmp_path / "insights.json" + insights_path.write_text(json.dumps(_valid_insights()), encoding="utf-8") + + code = adapter_cli_main(["enrich", "--output-dir", str(tmp_path), "--insights-path", str(insights_path)]) + assert code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True and payload["relationships"] == 2 + + +def test_cli_enrich_validation_failure_returns_1_and_emits_violations( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + _write_inventory(tmp_path, _inventory()) + raw = _valid_insights() + raw["pipeline_relationships"][0]["lineage_edge"]["edge_identity"] = "nope" + insights_path = tmp_path / "insights.json" + insights_path.write_text(json.dumps(raw), encoding="utf-8") + + code = adapter_cli_main(["enrich", "--output-dir", str(tmp_path), "--insights-path", str(insights_path)]) + assert code == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is False and payload["violations"] + + +# --------------------------------------------------------------------------- # +# MCP wiring. +# --------------------------------------------------------------------------- # + + +def test_mcp_enrich_inline_insights(monkeypatch, tmp_path: Path) -> None: + server = pytest.importorskip("flowx.mcp.server") + runner = pytest.importorskip("flowx.mcp.runner") + + captured: dict[str, Any] = {} + + class _Result: + ok = True + returncode = 0 + + def __init__(self, stdout: str) -> None: + self.stdout = stdout + self.stderr = "" + + def as_dict(self) -> dict[str, Any]: + return {"returncode": 0, "stdout": self.stdout, "stderr": ""} + + def fake_run_adapter(args: list[Any]) -> Any: + captured["args"] = [str(a) for a in args] + # The inline dict is staged to a temp file that the CLI would read; here just echo success. + return _Result(json.dumps({"ok": True, "violations": [], "pipeline_insights": 1, "relationships": 2})) + + monkeypatch.setattr(runner, "run_adapter", fake_run_adapter) + + out = server._cmd_enrich({"output_dir": str(tmp_path), "insights": _valid_insights()}) + assert out["ok"] is True + assert out["result"]["relationships"] == 2 + # The handler forwarded a real --insights-path (the staged temp file) to the CLI. + assert captured["args"][0] == "enrich" + assert "--insights-path" in captured["args"] + + +def test_mcp_enrich_requires_exactly_one_source(tmp_path: Path) -> None: + server = pytest.importorskip("flowx.mcp.server") + both = server._cmd_enrich({"output_dir": str(tmp_path), "insights": {}, "insights_path": "x.json"}) + neither = server._cmd_enrich({"output_dir": str(tmp_path)}) + assert both["ok"] is False and "exactly one" in both["error"] + assert neither["ok"] is False and "exactly one" in neither["error"] + + +# Ensure the deep-copied fixtures never share mutable state between tests. +def test_fixture_isolation() -> None: + a = _valid_insights() + b = _valid_insights() + a["pipeline_insights"][0]["pipeline"] = "mutated" + assert b["pipeline_insights"][0]["pipeline"] == "child" + assert copy.deepcopy(a) == a