[alexnastetsky]

Port of databricks-field-eng/flowx#17 onto the restructured codebase (translator/engine.py is now sources/adf/translate.py + ir_serde.py). Supersedes #40.

1. ForEach inputs-bridge notebooks carry the # Databricks notebook source marker so bundle validate accepts the notebook_task.
   • Already fixed on main by the render_bridge_notebook consolidation. No source change here — this PR only adds a guard test.
2. A join after an IfCondition (or Switch, which lowers to condition tasks) uses run_if: NONE_FAILED so a failure in the taken branch propagates (AT_LEAST_ONE_SUCCESS let a sibling success mask it).
3. Synthesised _init_<var> tasks that nothing references are pruned.

Changes

Where pruning runs. #17 pruned during IR serialisation. Here it runs at the end of prepare_workflow, which both packaging paths share. Only at that point are all references final:

• the agentic merge has happened, so an agent-written task may read {{tasks._init_X.values.X}};
• the preparers have created references the IR never shows: AppendVariable's source_task_key, ForEach inner-job parameters, and taskValues.get(taskKey=...) reads in generated notebooks.

Pruning on the IR would drop an init that an AppendVariable still needs.

How the pruned tasks are identified.

• The translator records the exact synthesised keys in Pipeline.synthesized_variable_init_keys.
• That set is serialised into translation_report.json only when non-empty, so Airflow reports are unchanged.
• It is rehydrated by pipeline_dict_to_ir.
• Only those keys are eligible for pruning, so a customer-authored task is never touched.

What counts as a reference. Only these structured forms count:

• a tasks.<key>.values token;
• a depends_on edge;
• a parameter whose whole value is the key;
• a taskKey='<key>' read in notebook code.

So a sibling key like _init_x_2, or a path like /Shared/_init_x_report, never keeps a dead _init_x alive. An init read by another init stays live. A pruned init also drops its notebook and setup tasks.

A dead init still appears in translation_report.json; it is removed only from the generated bundle.

Tests

• manually tested

• added unit tests

• added integration tests

• tests/unit/test_control_flow_fixes.py covers:

  • reference matching;
  • end-to-end translate → report → package for: a dominated init (pruned along with its notebook), a read init, an init feeding an AppendVariable, and an init read by a merged agentic task.

• Results:

  • pytest tests/unit: 1470 passed. The 2 failures in test_airflow_provider_sync.py are environmental (gpg signing in the fixture's git commit) and also fail on main.
  • pytest tests/integration -m "not slow and not integration": 141 passed.
  • ruff and mypy are clean.

• Before/after comparison: I packaged every ADF fixture in tests/resources/json on main and on this branch.

  • 24 _init_ tasks and their notebooks are gone. A substring search of main's packaged output finds no other mention of any of them, so they were dead.
  • Two joins changed from AT_LEAST_ONE_SUCCESS to NONE_FAILED.
  • Nothing else differs.

This pull request and its description were written by Isaac.

[ghanse]

Looks good overall. The join fix and the report round-trip are solid. One gap in pruning plus a few test holes are noted inline.

[ghanse]

LGTM, opened #194 to be fixed in a follow-up PR