Bundle packaging modes + ordered multi-bundle deploy

.claude-plugin/plugin.json

@@ -13,6 +13,7 @@
     "./skills/flowx-discover",
     "./skills/flowx-convert",
     "./skills/flowx-package",
+    "./skills/flowx-deploy",
     "./skills/flowx-migrate"
   ]
 }

AGENTS.md

@@ -58,7 +58,10 @@ intermediates under `.work/` (pruned by `package`).
 
 1. **Discover** -- Parse ADF JSON from UC volumes -> typed AST -> `metadata/inventory.json` + `metadata/profile_report.csv` + verbatim `metadata/<pipeline>.arm.json`
 2. **Convert** -- Registry dispatch + topological sort -> Pipeline IR (deterministic + agentic gaps); transient report at `.work/translation_report.json`
-3. **Package** -- IR -> DAB YAML + generated notebooks + setup scripts; prunes `.work/`
+3. **Package** -- IR -> DAB YAML + generated notebooks + setup scripts; prunes `.work/`. The
+   `--packaging-mode` flag (`per-pipeline` default / `single` / `per-group`) controls how a
+   multi-pipeline factory is laid out into bundles; a top-level `DEPLOY.md` records the suggested
+   callees-first deploy order for every mode.
 
 ### Key Patterns
 - `@dataclass(slots=True, kw_only=True)` for all models
@@ -80,7 +83,10 @@ intermediates under `.work/` (pruned by `package`).
 | `preparer/workflow_preparer.py` | Orchestrates activity preparers |
 | `preparer/code_generator.py` | Notebook code generation for activity types |
 | `preparer/activity_preparers/` | One module per activity type |
-| `bundler/dab_writer.py` | Generates databricks.yml, job YAML, resources |
+| `bundler/dab_writer.py` | Generates databricks.yml, job YAML, resources; groups pipelines into bundles per `--packaging-mode` |
+| `bundler/pipeline_graph.py` | Run Pipeline (ExecutePipeline) dependency graph: grouping (connected components) + deploy order (topo sort) |
+| `bundler/deploy_writer.py` | Renders the top-level `DEPLOY.md` (bundle layout + suggested deploy order) |
+| `bundler/deployer.py` | Ordered multi-bundle deploy: discovers bundles, deploys callees first, wires cross-bundle job ids |
 | `bundler/notebook_writer.py` | Writes generated notebooks to bundle |
 | `bundler/setup_generator.py` | Setup scripts for UC volumes, secrets, connections |
 | `reporting/coverage.py` | Builds per-pipeline coverage rows from `metadata/` |

skills/flowx-deploy/SKILL.md

@@ -0,0 +1,89 @@
+---
+name: flowx-deploy
+description: >
+  Deploy the per-pipeline Databricks Asset Bundles from a multi-pipeline flowx
+  migration in dependency order, resolving cross-bundle job ids automatically.
+  Local CLI only.
+triggers:
+  - "deploy bundles"
+  - "deploy in dependency order"
+  - "ordered deploy"
+  - "deploy flowx bundles"
+  - "deploy multi pipeline migration"
+---
+
+# Deploy per-pipeline flowx bundles in dependency order
+
+Deploy every bundle produced by a multi-pipeline migration, in the right order, wiring cross-bundle
+`ExecutePipeline` references automatically.
+
+## Context
+
+flowx emits **one bundle per ADF pipeline** under the output directory (`<output_dir>/<pipeline>/`).
+When pipeline A calls pipeline B via `ExecutePipeline`, the generated `run_job_task` in A's bundle
+references B — a job that lives in B's *own* bundle. flowx rewrites that out-of-bundle reference to
+`${var.<B>}` and declares a matching bundle variable, so each bundle is deploy-valid on its own; but
+the operator otherwise has to find B's numeric job id and pass it to A by hand.
+
+The package phase writes a top-level `DEPLOY.md` describing the bundle layout, cross-bundle
+dependencies, and the suggested callees-first deploy order. This skill is the **automated** form of
+those instructions — read `DEPLOY.md` for the human-readable version.
+
+This skill automates that:
+
+1. Discovers the bundles under the output directory (any immediate subdirectory with a
+   `databricks.yml`) — no manifest needed.
+2. Reads each bundle's job resource keys and its `${var.<callee>}` cross-bundle dependencies straight
+   from the generated `resources/*.yml`.
+3. Topologically sorts them (callees first) — a cyclic call graph is rejected with a clear error.
+4. Deploys each bundle with `databricks bundle deploy`.
+5. After each deploy, reads the deployed job id from `databricks bundle summary -o json` and injects
+   it into callers via `--var "<callee>=<id>"`.
+
+Because it captures and injects the **numeric job id** (not a name), dev-mode `[dev <user>]` job-name
+prefixes are irrelevant — it works identically for `dev` and `prod` targets.
+
+## Prerequisites
+
+- An output directory with the per-pipeline bundle subdirectories (from a multi-pipeline migration).
+- A working local `databricks` CLI with a configured profile / auth for the target workspace.
+
+> **Not available on Databricks serverless / Genie Code.** `databricks bundle deploy` and
+> `bundle summary` do not run on serverless compute, so this is a **local venv-CLI** (or web-terminal)
+> step only.
+
+## How to run
+
+Use the venv interpreter from the marker file (`<plugin_dir>/.migration-venv`) with `src/` on
+`PYTHONPATH`:
+
+```bash
+export PYTHONPATH="<plugin_dir>/src"
+PY="$(cat <plugin_dir>/.migration-venv)"
+
+# 1. Preview the deploy order and per-bundle commands without deploying:
+"$PY" -m flowx.adapter deploy --output-dir <output_dir> --target dev --dry-run
+
+# 2. Deploy for real:
+"$PY" -m flowx.adapter deploy --output-dir <output_dir> --target dev [--profile <profile>]
+```
+
+Flags:
+
+- `--output-dir` — directory holding the per-pipeline bundle subdirectories (default `./flowx_output`).
+- `--target` — bundle target to deploy (default `dev`).
+- `--profile` — Databricks CLI profile used for both `deploy` and `summary`.
+- `--dry-run` — print the dependency order and each `databricks bundle deploy …` command (with
+  `<callee>=<captured at deploy time>` placeholders), without deploying.
+- `--allow-missing-deps` — continue when a bundle references a callee that isn't present under the
+  output dir; that dependency's `--var` is skipped and must be set manually (see the bundle's
+  `SETUP.md`). Without this flag, a missing dependency is a hard error.
+
+## Behavior and failure handling
+
+- **Ordering:** callees always deploy before their callers. The order is deterministic.
+- **Deploy failure:** if any bundle's `databricks bundle deploy` fails, deployment stops immediately;
+  dependents are not deployed. The failing bundle and its stderr are printed.
+- **Job-id capture:** ids are read from `bundle summary -o json` at `.resources.jobs.<key>.id`. A
+  resource without a deployed job id (e.g. a Lakeflow pipeline resource) is skipped — no empty `--var`.
+- **Cycles:** a cyclic call graph cannot be ordered; the command errors out.

skills/flowx-package/SKILL.md

@@ -99,8 +99,26 @@ Ask the user for the following (provide defaults):
 | Bundle name | Name for the DABs project | derived from first pipeline name |
 | Target environments | Deployment targets to configure | `dev, staging, prod` |
 | Warehouse ID | SQL warehouse for SQL tasks (optional) | prompt if SQL tasks exist |
+| Packaging mode | How to lay out bundles for a multi-pipeline factory (`--packaging-mode`): `per-pipeline`, `single`, or `per-group`. See below. | `per-pipeline` |
 | Databricks CLI profile | Profile used to download workspace-resident notebooks / JARs / Python files (`--profile`). Required only when the bundle references absolute workspace paths. | resolved from `~/.databrickscfg` (auto-prompt if multiple) |
 
+**Packaging mode** (`--packaging-mode`, surfaced as the `packaging_mode` input) controls how a
+multi-pipeline migration is laid out. For a single-pipeline migration every mode is equivalent.
+
+- **`per-pipeline`** (default) — one Databricks Asset Bundle per ADF pipeline, each in its own
+  `<output_dir>/<pipeline>/` subdirectory. Cross-pipeline `ExecutePipeline` calls become
+  `${var.<callee>}` job-id references wired at deploy time.
+- **`single`** — every pipeline in one bundle at the output root. Intra-bundle `ExecutePipeline`
+  calls resolve directly via `${resources.jobs.<callee>.id}` (no deploy-time wiring needed).
+- **`per-group`** — pipelines grouped into bundles. By default (`--group-by inferred`) groups are
+  the connected components of the Run Pipeline call graph, so pipelines that call one another ship
+  together. Pass `--group-by spec --group-spec <path>` to use an explicit JSON/YAML mapping
+  (`{pipeline: group}` or `{group: [pipelines]}`); the `group_spec` input captures that path.
+
+Regardless of mode, a single top-level `DEPLOY.md` is written describing every bundle, its
+cross-bundle dependencies, and a suggested callees-first deploy order. Use the `flowx-deploy` skill
+(`python -m flowx.adapter deploy`) to deploy the bundles in that order automatically.
+
 ### Step 2.5 — Detect workspace artifacts and authenticate
 
 > **Databricks runtime (serverless / cluster):** Authentication is auto-configured
@@ -168,6 +186,8 @@ Execute the DAB writer:
   --catalog <catalog> \
   --schema <schema> \
   --bundle-name <bundle_name> \
+  [--packaging-mode per-pipeline|single|per-group] \
+  [--group-by inferred|spec] [--group-spec <path>] \
   [--profile <databricks-cli-profile>] \
   [--no-download-workspace-files] \
   [--keep-intermediates]
@@ -221,6 +241,7 @@ Show the user what was generated:
       create_secrets.py
       register_connections.py
   SETUP.md
+  DEPLOY.md                           # bundle layout + suggested deploy order (top level)
   metadata/                           # kept migration metadata (from discover + modify)
     inventory.json
     profile_report.csv
@@ -245,7 +266,7 @@ Emphasize that the user should review these scripts before running them, especia
 
 Briefly describe:
 - **databricks.yml** — The root bundle config with workspace, target environments (dev/staging/prod), and variable definitions. Variables are parameterized for environment-specific values (catalog, schema, warehouse).
-- **resources/*.yml** — One YAML file per Databricks Lakeflow Job (one per ADF pipeline). Each job contains tasks mapped from ADF activities, with dependencies matching the original ADF dependency chains.
+- **resources/*.yml** — One YAML file per Databricks Lakeflow Job. In `per-pipeline` mode each bundle holds one pipeline's job (plus any inner ForEach jobs); in `single`/`per-group` mode a bundle holds several pipelines' jobs side by side. Each job contains tasks mapped from ADF activities, with dependencies matching the original ADF dependency chains.
 - **src/notebooks/*.py** — Python notebooks for activities that translate to notebook_task. These contain the actual data movement or transformation logic.
 - **tests/*.py** — Skeleton test files for validating the migrated jobs.
 
@@ -279,6 +300,11 @@ Next Steps
 
 Recommend running `databricks bundle validate` first to catch any configuration issues before deployment.
 
+When the migration produced **multiple bundles** (`per-pipeline` or `per-group` mode), point the
+user at the top-level `DEPLOY.md` for the suggested callees-first deploy order, and recommend the
+`flowx-deploy` skill (`python -m flowx.adapter deploy --output-dir <output_dir>`) to deploy them in
+order and wire cross-bundle job ids automatically.
+
 ### Step 8 — (Optional) Persist coverage results and install a dashboard
 
 This step only applies when running with workspace auth (Genie Code, or a configured
@@ -336,7 +362,8 @@ All under the shared `<output_dir>`:
 | `resources/*.yml` | Job and pipeline YAML definitions |
 | `src/notebooks/*.py` | Generated notebooks |
 | `src/setup/*.py` | Infrastructure setup scripts |
-| `SETUP.md` | Human-readable setup instructions |
+| `SETUP.md` | Human-readable setup instructions (one per bundle) |
+| `DEPLOY.md` | Top-level bundle layout + suggested deploy order (all packaging modes) |
 | `metadata/inventory.json` | Activity inventory (from discover) |
 | `metadata/profile_report.csv` | Per-pipeline complexity report (from profile) |
 | `metadata/<pipeline>.arm.json` | Verbatim original ADF/ARM source (from discover) |

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``, ``inputs``, ``materialize-lookup``, ``workspace-paths``, ``record-results``,
+``install-dashboard``, and ``deploy`` -- so each agent turn runs as an independent process holding no
+session state across user prompts.
 """
 
 from __future__ import annotations
@@ -89,55 +89,27 @@ def main(argv: list[str] | None = None) -> int:
         return _run_record_results(args)
     if args.command == "install-dashboard":
         return _run_install_dashboard(args)
+    if args.command == "deploy":
+        return _run_deploy(args)
     parser.print_help(sys.stderr)
     return 2
 
 
-def _run_resolve_agentic(args: argparse.Namespace) -> int:
-    """Runs the fingerprint-bound agentic resolution workflow for Airflow leaf gaps."""
-    if args.source != "airflow":
-        print("resolve-agentic is not enabled for ADF; ADF uses the legacy merge path.", file=sys.stderr)
-        return 2
-    from flowx.agentic import (
-        AgenticContractError,
-        apply_airflow_resolutions,
-        prepare_airflow_resolutions,
-        stage_airflow_resolutions,
-    )
+def _run_deploy(args: argparse.Namespace) -> int:
+    """Implements ``deploy``: deploy per-pipeline bundles in dependency order.
 
-    try:
-        if args.action == "prepare":
-            if args.source_path is None or args.report is None:
-                print("resolve-agentic prepare requires --source-path and --report.", file=sys.stderr)
-                return 2
-            payload = prepare_airflow_resolutions(
-                source_path=args.source_path,
-                report_path=args.report,
-                output_dir=args.output_dir,
-                dbt_mode=args.dbt_mode,
-                gap_id=args.gap_id,
-            )
-        elif args.action == "stage":
-            payload = stage_airflow_resolutions(
-                output_dir=args.output_dir,
-                candidate_paths=args.candidate,
-                replace=args.replace,
-            )
-        else:
-            payload = apply_airflow_resolutions(
-                output_dir=args.output_dir,
-                accepted_gap_ids=args.accept_gap,
-                accept_all=args.accept_all,
-                review_complete=args.review_complete,
-                review_manifest_path=args.review_manifest,
-                reset=args.reset,
-                source_path=args.source_path,
-            )
-    except (AgenticContractError, OSError, json.JSONDecodeError) as error:
-        print(f"Agentic resolution failed: {error}", file=sys.stderr)
-        return 1
-    _emit_json(payload, None)
-    return 0
+    Local-CLI only — shells out to ``databricks bundle deploy`` / ``summary``, which are not
+    available on Databricks serverless / Genie Code. Returns the deployer's exit code.
+    """
+    from flowx.bundler.deployer import run as run_deploy
+
+    return run_deploy(
+        args.output_dir,
+        target=args.target,
+        profile=args.profile,
+        dry_run=args.dry_run,
+        allow_missing_deps=args.allow_missing_deps,
+    )
 
 
 def _run_record_results(args: argparse.Namespace) -> int:
@@ -527,10 +499,37 @@ def _build_parser() -> argparse.ArgumentParser:
         help="Workspace folder for the dashboard (defaults to the current user's home).",
     )
 
-    # Unified phase runners: `adapter <phase> --source <name> -- <flags>` routes discover/convert
-    # to the named source's phase module. --source is required for those phases (no default);
-    # package is source-independent. --source-path (and each source's own alias, e.g.
-    # --adf-source-path) normalise to --source-dir.
+    deploy = subparsers.add_parser(
+        "deploy",
+        help="Deploy per-pipeline bundles in dependency order, wiring cross-bundle job ids (local CLI).",
+    )
+    deploy.add_argument(
+        "--output-dir",
+        type=Path,
+        default=Path("./flowx_output"),
+        help="Directory holding the per-pipeline bundle subdirectories.",
+    )
+    deploy.add_argument("--target", type=str, default="dev", help="Bundle target to deploy (default: dev).")
+    deploy.add_argument("--profile", type=str, default=None, help="Databricks CLI profile for deploy and summary.")
+    deploy.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Print the dependency order and deploy commands without deploying.",
+    )
+    deploy.add_argument(
+        "--allow-missing-deps",
+        action="store_true",
+        help=(
+            "Order and attempt to deploy even when a bundle references a callee absent from the output "
+            "dir. The missing ${var.<callee>} is declared without a default, so that bundle's deploy "
+            "still fails until you supply the value manually (edit its databricks.yml default or "
+            "`databricks bundle deploy --var <callee>=<job_id>` per SETUP.md); this flag only unblocks "
+            "the ordering, not the deploy."
+        ),
+    )
+
+    # Unified phase runners: `adapter <phase> -- <flags>` forwards to the phase CLI (one entry point);
+    # --adf-source-path is accepted as an alias of the loader/translator --source-dir flag.
     for _phase in ("discover", "convert", "package"):
         _runner = subparsers.add_parser(
             _phase,

src/flowx/adapter/constants.py

@@ -42,6 +42,13 @@
 INPUT_RESULTS_TABLE: Final[str] = "results_table"
 INPUT_RESULTS_WAREHOUSE: Final[str] = "results_warehouse_id"
 INPUT_INSTALL_DASHBOARD: Final[str] = "install_dashboard"
+INPUT_PACKAGING_MODE: Final[str] = "packaging_mode"
+INPUT_GROUP_SPEC: Final[str] = "group_spec"
+
+# Packaging-mode answers accepted by the package phase's --packaging-mode flag.
+PACKAGING_MODE_PER_PIPELINE: Final[str] = "per-pipeline"
+PACKAGING_MODE_SINGLE: Final[str] = "single"
+PACKAGING_MODE_PER_GROUP: Final[str] = "per-group"
 
 LAKEFLOW_CONNECTOR_TYPE_QUERY_BASED: Final[str] = "query_based"
 LAKEFLOW_CONNECTOR_TYPE_CDC: Final[str] = "cdc"

src/flowx/adapter/session.py

@@ -20,11 +20,12 @@
     INPUT_BUNDLE_NAME,
     INPUT_CATALOG,
     INPUT_DATABRICKS_PROFILE,
-    INPUT_GLOBAL_PARAMETER_RESOLUTION,
+    INPUT_GROUP_SPEC,
     INPUT_INSTALL_DASHBOARD,
     INPUT_INVENTORY_PATH,
     INPUT_OUTPUT_BUNDLE_PATH,
     INPUT_OUTPUT_DIR,
+    INPUT_PACKAGING_MODE,
     INPUT_RESULTS_TABLE,
     INPUT_RESULTS_WAREHOUSE,
     INPUT_SCHEMA,
@@ -373,6 +374,31 @@ def _convert_options(source: str) -> tuple[MigrationInputOption, ...]:
         default="",
         required=False,
     ),
+    MigrationInputOption(
+        option_id=INPUT_PACKAGING_MODE,
+        prompt="How should pipelines be packaged into bundles?",
+        description=(
+            "One of ``per-pipeline`` (default — one Databricks Asset Bundle per ADF pipeline), "
+            "``single`` (all pipelines in one bundle), or ``per-group`` (group pipelines into "
+            "bundles by their Run Pipeline call graph, or by an explicit --group-spec). Forwarded "
+            "to ``package`` as ``--packaging-mode``. For a single-pipeline migration every mode is "
+            "equivalent."
+        ),
+        default="per-pipeline",
+        required=False,
+    ),
+    MigrationInputOption(
+        option_id=INPUT_GROUP_SPEC,
+        prompt="Path to a pipeline->group spec (only for --packaging-mode per-group with explicit groups)?",
+        description=(
+            "Optional. JSON/YAML file mapping pipelines to group names (``{pipeline: group}`` or "
+            "``{group: [pipelines]}``). When set, package is run with ``--packaging-mode per-group "
+            "--group-by spec --group-spec <path>``. Leave blank to infer groups from the Run "
+            "Pipeline call graph."
+        ),
+        default="",
+        required=False,
+    ),
     MigrationInputOption(
         option_id=INPUT_DATABRICKS_PROFILE,
         prompt="Databricks CLI profile?",