flowx-convert

name	flowx-convert
description	Translate parsed ADF pipeline AST into Databricks IR (intermediate representation). Runs deterministic translators for known activity types, then performs agentic (LLM-assisted) translation for the remaining gaps.
triggers	["translate ADF","convert ADF","translate pipelines","convert pipelines","run translation"]

Convert ADF to Databricks IR

Convert the parsed ADF inventory into Databricks intermediate representation (IR) using deterministic translators for known types and agentic fallback for unknown types.

Context

This is phase 2 of the flowx migration workflow. It consumes the ADF source (profiled by the discover skill) and produces a translation report — a transient intermediate under <output_dir>/.work/ — that the package skill uses to generate Databricks Declarative Automation Bundles. It shares the single migration <output_dir> with the other phases.

The translation follows a deterministic-first strategy:

1. Activities with known, well-defined mappings are translated by built-in Python translators
2. Activities that require interpretation, expression conversion, or lack a Python translator are handled by agentic (LLM-assisted) translation performed by the agent

How to run this skill — MCP tools or venv CLI

This phase runs one of two ways; run the setup skill first if you haven't.

• MCP tool (Databricks Genie Code, or a local stdio registration) — the only path in Genie Code: call the single flowx tool (one command per step) and run no python3/$PY/bash commands. The "$PY" -m … snippets in the steps below are the local-CLI fallback only — ignore them on this path. Map the steps to:

  flowx(command="convert", parameters={"output_dir": "<dir>", "pipeline": "<optional>"})
  # convert reuses the discovered output_dir on the server; only pass "adf_definitions" (inline ARM
  # JSON) if you are converting without a prior discover on this server.
  flowx(command="inspect", parameters={"report_path": "<dir>/.work/translation_report.json", "answers": [...]})
  flowx(command="apply_answers", parameters={"report_path": "...", "answers": ["id=value", ...], "output_dir": "<dir>", "lookup_csv": "<optional>"})
  flowx(command="merge_agentic", parameters={"report_path": "...", "agentic_results_dir": "<dir>", "output_path": "<optional>"})
  

  Use the tool results in place of reading the files directly. command="merge_agentic" covers the agentic --merge-agentic step shown later in this skill.

• venv CLI (local, no MCP server): ensure the venv exists (setup Path B / bootstrap.sh), then run the commands below with the venv interpreter (from the marker file <plugin_dir>/.migration-venv) and src/ on PYTHONPATH (use $PY anywhere a command shows python3):

  export PYTHONPATH="<plugin_dir>/src"
  PY="$(cat <plugin_dir>/.migration-venv)"
  "$PY" -m flowx.adapter convert --output-dir <dir>
  

  If Python or pip is missing, bootstrap.sh prints a warning telling the user what to install — relay it and stop until they have Python 3.12+ and pip.

Workflow

Follow these steps in order:

Step 0 — Gather phase inputs

Run the adapter inputs subcommand so the agent surfaces the free-text options the phase needs (inventory path, ADF source dir, output directory):

"$PY" -m flowx.adapter inputs convert

The JSON response carries the prompts and defaults; collect answers from the user (or fall back to the defaults). Keep them in conversation context — the same shared <output_dir> is used by every phase.

Step 1 — Locate the inventory

The discover phase wrote <output_dir>/metadata/inventory.json (and profile_report.csv). If the shared <output_dir> is not already in conversation context, ask the user:

Which migration output directory did the discover phase use? (default: ./flowx_output)

Validate <output_dir>/metadata/inventory.json exists and is well-formed.

Step 2 — Run deterministic translation

Execute the translation engine on all deterministic activities:

# Unified runner (recommended): `"$PY" -m flowx.adapter convert ...`
# forwards to the engine below; --adf-source-path aliases --source-dir.
"$PY" -m flowx.translator.engine \
  --source-dir <adf_source_dir> \
  --output-dir <output_dir> \
  [--pipeline <pipeline_name>] \
  [--global-parameter-resolution literal|bundle_variable]

Where:

• <adf_source_dir> is the original ADF JSON directory (the same --source-dir used by discover)
• <output_dir> is the shared migration output directory (default: ./flowx_output) — the same one discover used
• <pipeline_name> (optional) — when provided, translates only the named pipeline. Always pass --pipeline when the user has specified a specific pipeline to migrate, matching the value passed to the discover phase.
• --global-parameter-resolution (optional, default literal) — how @pipeline().globalParameters.X references resolve, applied to every pipeline. literal bakes the factory value in as a literal; bundle_variable emits ${var.X} and declares the global as a DAB bundle variable whose default is the factory value, so it can be set at deploy time (--var X=… or a per-target override) instead of being hard-coded into pipeline/activity bodies. Globals referenced inside generated notebook code are bridged through the task's base_parameters so ${var.X} still resolves. See SETUP.md for the list of hoisted variables and a plaintext-secret caveat.

The translation report and intermediate IR are written to the transient <output_dir>/.work/ folder (translation_report.json, per-pipeline IR, gaps.json). These are consumed by the steps below and the package phase, then pruned — they are not kept artifacts.

Step 3 — Read the translation report

Read <output_dir>/.work/translation_report.json. It has this structure:

{
  "inventory_path": "/path/to/inventory.json",
  "generated_at": "2026-04-07T12:30:00Z",
  "translations": [
    {
      "pipeline": "ETL_Main",
      "activity": "CopyFromBlob",
      "type": "Copy",
      "strategy": "deterministic",
      "status": "translated",
      "ir": {
        "task_key": "copy_from_blob",
        "task_type": "notebook_task",
        "notebook_path": "notebooks/copy_from_blob.py",
        "parameters": { "source": "abfss://...", "target": "..." }
      }
    },
    {
      "pipeline": "ETL_Main",
      "activity": "TransformData",
      "type": "ExecuteDataFlow",
      "strategy": "agentic",
      "status": "pending",
      "raw_activity_json": { "...": "..." }
    }
  ],
  "summary": {
    "total": 47,
    "deterministic_translated": 35,
    "agentic_pending": 10,
    "failed": 2
  }
}

Step 4 — Handle agentic gaps

For each translation with "status": "pending" and "strategy": "agentic", perform LLM-assisted translation from the activity's ARM JSON, routing by activity type.

Every agentic gap in the translation report carries the activity's full ADF/ARM JSON under raw_activity_json (engine field raw_definition), and the generated placeholder notebook embeds the same JSON in a fenced json block. This holds for nested activities too — an Until inside an IfCondition / Switch / ForEach is reported as its own gap. Always translate from this ARM JSON.

Until activities (agent-based handler): Databricks Lakeflow Jobs have no native repeat-until loop, so translate the Until from its ARM JSON into a single Python notebook task implementing a bounded polling loop. From the embedded JSON, read:

• typeProperties.expression — the ADF exit condition (e.g. @or(equals(variables('jobStatus'),'succeeded'), equals(variables('jobStatus'),'failed'))); convert it into the Python while not (<condition>): guard.
• typeProperties.timeout — wrap the loop in a wall-clock deadline (time.monotonic()), raising on timeout.
• typeProperties.activities — the loop body (e.g. a Wait, a polling WebActivity, a SetVariable that captures the next status); translate each child inline so the whole loop runs in one notebook. Read the loop variables from dbutils.widgets, surface the final state as a task value, and write the result over the placeholder notebook's raise NotImplementedError cell. Perform the translation directly from the same ARM JSON.

ExecuteDataFlow activities: Translate the data flow directly from the raw activity JSON and associated data flow definition, using:

• The raw typeProperties from the ADF activity
• The data flow JSON definition (if available in the source directory under dataflow/)
• The linked service configurations for source/sink connections
• Target catalog and schema for the SDP pipeline or PySpark notebook output

Control flow activities (Switch, Until, Wait, Filter, AppendVariable): Translate the control-flow activity directly from the raw activity JSON, using:

• The full pipeline JSON containing the activity
• Any nested activities within the control flow
• Variable definitions from the pipeline
• The desired Databricks task type mapping

Stored procedures and external calls (SqlServerStoredProcedure, AzureFunction, WebHook, Custom): Translate the activity directly from the raw activity JSON, using:

• The linked service configuration for the target system
• Connection details and authentication method
• Any parameters or request bodies

Complex expressions: If any activity (deterministic or agentic) contains ADF expressions that the deterministic translator could not resolve, translate them directly, using:

• The raw expression string (e.g., @pipeline().parameters.inputPath)
• The expression context (pipeline parameters, variables, activity outputs)
• The target format (Python f-string, Spark SQL, task parameter reference)

Trigger definitions: Translate the trigger directly, using:

• The trigger JSON definition
• The associated pipeline references
• Target: Databricks job schedule configuration (quartz_cron_expression, periodic, or file_arrival)

Step 5 — Collect agentic results

Each resolved agentic gap produces one translation result. Write them into <output_dir>/agentic_results/ as one JSON file per activity (the filename is arbitrary, e.g. <pipeline>__<activity>.json). Each file MUST use this schema:

{
  "activity_name": "<the placeholder activity name, exactly as in the report>",
  "pipeline": "<pipeline name>",
  "task": {
    "type": "NotebookActivity",
    "name": "<activity name>",
    "task_key": "<task key>",
    "notebook_path": "/Workspace/.../your_translated_notebook"
  }
}

• activity_name (required) — matches the name of the placeholder task in the report (the merge locates it by name, recursing into IfCondition / ForEach / Switch containers, so nested gaps like an Until are found).
• pipeline (optional) — only needed to disambiguate multi-pipeline reports.
• task (required) — the replacement IR task. The most portable form is a NotebookActivity whose notebook_path points at a notebook you have written to the workspace; the package phase references it directly. task_key and depends_on are inherited from the placeholder when omitted, so dependency edges are preserved.

Step 6 — Merge agentic results

Fold the results into the translation report (placeholders are replaced in place):

"$PY" -m flowx.translator.engine \
  --merge-agentic \
  --report <output_dir>/.work/translation_report.json \
  --agentic-results <agentic_results_dir>

Equivalently via the unified runner: "$PY" -m flowx.adapter convert --merge-agentic --report <output_dir>/.work/translation_report.json --agentic-results <dir>. Add --output <path> to write a copy instead of overwriting the report. The command exits non-zero if any result could not be matched to a placeholder.

This updates <output_dir>/.work/translation_report.json with the agentic results merged in, changing their status from pending to translated (or failed if the agentic skill could not produce a result).

Step 6.1 — Gather just-in-time translation configuration

Run inspect once to get the full option schema, then drive the whole question chain yourself — do not re-run inspect per follow-up:

"$PY" -m flowx.adapter inspect <output_dir>/.work/translation_report.json

It returns every option the report can raise, each annotated with a show_when condition:

{"pipelines": [{"pipeline_name": "...", "options": [
  {"option_id": "notify_destination", "prompt": "...", "rationale": "...",
   "choices": [{"value": "...", "label": "...", "description": "..."}],
   "free_text": false, "default": "keep", "show_when": []},
  {"option_id": "notify_slack_url", "prompt": "...", "free_text": true, "default": "",
   "show_when": [{"option_id": "notify_destination", "in": ["slack"]}]}
]}]}

Walk it locally:

1. Ask an option only when its show_when is satisfied — every clause {option_id, in:[values]} must match an answer you've already collected (empty show_when = always ask). So notify_slack_url surfaces only after notify_destination=slack; the metadata-driven access/size/lookup_tool chain surfaces only after metadata_driven_consolidate=consolidate, etc. Present each option's prompt/rationale and choices; honor the default.
2. Validate each answer against choices (a free_text option — empty choices — accepts any value; blank skips an optional one).
3. Perform data actions inline when an answer calls for it — e.g. when metadata_driven_lookup_tool=have, run the lookup query with your database tool to get the rows.
4. When every applicable option is answered, apply them in one modify call (Step 6.2) with all answers as --answer OPTION_ID=VALUE flags. modify validates every answer server-side.

Activity→Notify (activity_and_notify) motifs. When any activity (Copy, Notebook, Lookup, stored procedure, …) is followed by notification Web activities, the adapter raises notify_destination: keep (default) leaves the Web activities to translate directly — nothing is collapsed. Any other value (email, slack, teams, pagerduty, webhook)