Fix three ADF -> Databricks control-flow / codegen bugs

src/flowx/bundler/dab_writer.py

@@ -2000,7 +2000,9 @@ def expand_terminals(condition_key: str, seen: set[str]) -> list[str]:
                 seen_keys.add(key)
                 deduped.append(dep)
             task["depends_on"] = deduped
-            task.setdefault("run_if", "AT_LEAST_ONE_SUCCESS")
+            # The untaken branch's terminals are SKIPPED, so NONE_FAILED lets the join run yet still propagates a
+            # taken-branch failure (ADF dependsOn [Succeeded]); AT_LEAST_ONE_SUCCESS let a sibling success mask it.
+            task.setdefault("run_if", "NONE_FAILED")
 
 
 _TASK_VALUE_REF = re.compile(r"\{\{tasks\.([^.]+)\.values\.[^}]+\}\}")
@@ -2671,6 +2673,7 @@ def pipeline_dict_to_ir(pipeline_dict: dict[str, Any]) -> tuple[Pipeline, list[d
         reconciliation_status=pipeline_dict.get("reconciliation_status"),
         migration_status=pipeline_dict.get("migration_status", "included"),
         audit=dict(pipeline_dict.get("audit") or {}),
+        synthesized_variable_init_keys=frozenset(pipeline_dict.get("synthesized_variable_init_keys") or ()),
     )
     return pipeline, parameters
 

src/flowx/ir_serde.py

@@ -70,6 +70,8 @@ def pipeline_to_dict(pipeline: Pipeline) -> dict[str, Any]:
     }
     if pipeline.bundle_variables:
         result["bundle_variables"] = pipeline.bundle_variables
+    if pipeline.synthesized_variable_init_keys:
+        result["synthesized_variable_init_keys"] = sorted(pipeline.synthesized_variable_init_keys)
     if pipeline.description is not None:
         result["description"] = pipeline.description
     if pipeline.timeout_seconds is not None:

src/flowx/models/ir.py

@@ -632,6 +632,10 @@ class Pipeline:
         audit: Source-audit counts and transformation ledger.
         bundle_variables: DAB bundle-variable declarations (name -> ``{"description", "default"}``)
             for factory globals hoisted under the ``bundle_variable`` resolution policy.
+        synthesized_variable_init_keys: Task keys of the SetVariable tasks the
+            translator synthesises for default-valued variables. Tracked
+            explicitly so dead-init pruning acts only on translator-generated
+            tasks and never touches a customer-authored task.
         reconciliation_status: Source-audit result for this pipeline.
         migration_status: Whether the pipeline is included or explicitly excluded.
         audit: Source-audit counts and transformation ledger.
@@ -651,6 +655,7 @@ class Pipeline:
     audit: dict[str, Any] = field(default_factory=dict)
     translation_configuration: TranslationConfiguration | None = None
     bundle_variables: dict[str, dict[str, Any]] = field(default_factory=dict)
+    synthesized_variable_init_keys: frozenset[str] = field(default_factory=frozenset)
 
 
 @dataclass(frozen=True, slots=True)

src/flowx/motifs/collapser.py

@@ -77,6 +77,7 @@ def collapse_motifs(
         migration_status=pipeline.migration_status,
         audit=dict(pipeline.audit),
         translation_configuration=pipeline.translation_configuration,
+        synthesized_variable_init_keys=pipeline.synthesized_variable_init_keys,
     )
 
 

src/flowx/preparer/workflow_preparer.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+import json
 from collections.abc import Iterable
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any
 
 from flowx.models.dab import DabNotebook, ParameterApproximation, SecretInstruction, SetupTask
@@ -336,9 +337,12 @@ def prepare_workflow(pipeline: Pipeline) -> PreparedWorkflow:
     # Updated in pipeline-declaration order so AppendVariable sees the
     # most-recent prior writer of each variable.
     variable_task_keys_map: dict[str, str] = {}
+    prepared_inits: dict[str, PreparedActivity] = {}
 
     for activity in pipeline.tasks:
         prepared = prepare_activity(activity, scope=scope, variable_task_keys=variable_task_keys_map)
+        if activity.task_key in pipeline.synthesized_variable_init_keys:
+            prepared_inits[activity.task_key] = prepared
         all_tasks.append(prepared.task)
         all_tasks.extend(prepared.extra_tasks)
         artifacts = merge_prepared_artifacts(artifacts, prepared)
@@ -359,6 +363,8 @@ def prepare_workflow(pipeline: Pipeline) -> PreparedWorkflow:
                 if original_key in task_key_remap:
                     dep["task_key"] = task_key_remap[original_key]
 
+    all_tasks, artifacts = _prune_dead_variable_inits(all_tasks, artifacts, prepared_inits)
+
     seen_secrets: set[tuple[str, str]] = set()
     unique_secrets: list[SecretInstruction] = []
     for secret in artifacts.secrets:
@@ -450,6 +456,133 @@ def prepare_workflow(pipeline: Pipeline) -> PreparedWorkflow:
     )
 
 
+def _prune_dead_variable_inits(
+    tasks: list[dict[str, Any]],
+    artifacts: PreparedArtifacts,
+    prepared_inits: dict[str, PreparedActivity],
+) -> tuple[list[dict[str, Any]], PreparedArtifacts]:
+    """Drops synthesised variable-init tasks that no real task reaches, with their notebooks and setup tasks.
+
+    Args:
+        tasks: Prepared parent-job task dicts.
+        artifacts: Artifacts accumulated while preparing *tasks*.
+        prepared_inits: Prepared output of each translator-synthesised ``_init_<var>`` task, by task key.
+
+    Returns:
+        The ``(tasks, artifacts)`` pair with every dead init task removed.
+
+    Notes:
+        The ADF translator emits an init task per default-valued variable so every read has a
+        writer to bind to; once an explicit SetVariable dominates every read, the init task is dead
+        weight that still runs and reads as a duplicate of the real setter.  Pruning runs here, on
+        the fully prepared tasks, because only now are all references final: the agentic merge has
+        happened, and preparers have materialised references the IR never shows (AppendVariable's
+        ``source_task_key``, ForEach inner-job parameters, notebook dispatch stubs).  Only
+        translator-synthesised keys are eligible, so a customer-authored task is never removed.
+    """
+    dead_keys = find_dead_variable_inits(tasks, artifacts, prepared_inits)
+    if not dead_keys:
+        return tasks, artifacts
+    dead_notebooks = {id(notebook) for key in dead_keys for notebook in prepared_inits[key].notebooks}
+    dead_setup_tasks = {id(setup_task) for key in dead_keys for setup_task in prepared_inits[key].setup_tasks}
+    pruned_artifacts = replace(
+        artifacts,
+        notebooks=tuple(notebook for notebook in artifacts.notebooks if id(notebook) not in dead_notebooks),
+        setup_tasks=tuple(setup_task for setup_task in artifacts.setup_tasks if id(setup_task) not in dead_setup_tasks),
+    )
+    return [task for task in tasks if task.get("task_key") not in dead_keys], pruned_artifacts
+
+
+def find_dead_variable_inits(
+    tasks: list[dict[str, Any]],
+    artifacts: PreparedArtifacts,
+    prepared_inits: dict[str, PreparedActivity],
+) -> frozenset[str]:
+    """Returns the synthesised variable-init task keys that no real task reaches.
+
+    Args:
+        tasks: Prepared parent-job task dicts.
+        artifacts: Artifacts accumulated while preparing *tasks*; their notebooks and inner
+            workflows (e.g. ForEach sub-jobs) are scanned too.
+        prepared_inits: Prepared output of each translator-synthesised ``_init_<var>`` task, by task key.
+
+    Returns:
+        The init task keys present in *tasks* that are safe to drop.
+
+    Notes:
+        A reference is any of the forms the preparers emit: an anchored task-value token
+        (``tasks.<key>.values``), a ``depends_on`` edge, a task parameter whose whole value is the
+        key (AppendVariable's ``source_task_key``), or a ``taskKey='<key>'`` read in generated
+        notebook code.  Matching these forms rather than a bare substring avoids a false keep when
+        the key text merely appears inside an unrelated key, path, or metadata field.  An init is
+        live when a real (non-init) task or notebook references it, directly or through a chain of
+        live inits, so a chain of inits that only feed each other is dropped as a whole.
+    """
+    init_keys = {task["task_key"] for task in tasks if task.get("task_key") in prepared_inits}
+    notebook_owners = {id(notebook): key for key in init_keys for notebook in prepared_inits[key].notebooks}
+    references = [
+        (task.get("task_key"), _task_references(task, init_keys))
+        for task in _iter_workflow_tasks(tasks, artifacts.inner_workflows)
+    ] + [
+        (notebook_owners.get(id(notebook)), {key for key in init_keys if _notebook_references(notebook.content, key)})
+        for notebook in artifacts.notebooks
+    ]
+    live_keys: set[str] = set()
+    pending = [key for owner, referenced in references if owner not in init_keys for key in referenced]
+    while pending:
+        key = pending.pop()
+        if key not in live_keys:
+            live_keys.add(key)
+            pending.extend(ref for owner, referenced in references if owner == key for ref in referenced)
+    return frozenset(init_keys - live_keys)
+
+
+_PARAMETER_FIELDS = (
+    ("notebook_task", "base_parameters"),
+    ("run_job_task", "job_parameters"),
+    ("spark_python_task", "parameters"),
+    ("spark_jar_task", "parameters"),
+    ("python_wheel_task", "parameters"),
+    ("python_wheel_task", "named_parameters"),
+    ("sql_task", "parameters"),
+)
+
+
+def _task_references(task: dict[str, Any], init_keys: set[str]) -> set[str]:
+    """Returns the init keys that *task* reads through a task-value token, a dependency, or a parameter."""
+    serialized_task = json.dumps(task, default=str)
+    whole_values = {value for value in _iter_reference_values(task) if isinstance(value, str)}
+    return {key for key in init_keys if f"tasks.{key}.values" in serialized_task or key in whole_values}
+
+
+def _iter_reference_values(task: dict[str, Any]) -> Iterable[Any]:
+    """Yields *task*'s ``depends_on`` keys and parameter values, including a ForEach body task's."""
+    for dependency in task.get("depends_on") or []:
+        yield dependency.get("task_key")
+    for task_type, field_name in _PARAMETER_FIELDS:
+        parameters = (task.get(task_type) or {}).get(field_name) or {}
+        yield from parameters.values() if isinstance(parameters, dict) else parameters
+    body_task = (task.get("for_each_task") or {}).get("task")
+    if body_task:
+        yield from _iter_reference_values(body_task)
+
+
+def _notebook_references(content: str, task_key: str) -> bool:
+    """Reports whether generated notebook *content* reads a task value published by *task_key*."""
+    return any(
+        token in content for token in (f"tasks.{task_key}.values", f"taskKey='{task_key}'", f'taskKey="{task_key}"')
+    )
+
+
+def _iter_workflow_tasks(
+    tasks: list[dict[str, Any]], inner_workflows: Iterable[PreparedWorkflow]
+) -> Iterable[dict[str, Any]]:
+    """Yields *tasks* followed by every task of the (recursively nested) inner workflows."""
+    yield from tasks
+    for workflow in inner_workflows:
+        yield from _iter_workflow_tasks(workflow.tasks, workflow.inner_workflows)
+
+
 def _detect_cross_foreach_variable_reads(activities: list[Activity]) -> list[SetupTask]:
     """Returns SetupTasks for variables mutated inside a ForEach but read outside.
 

src/flowx/sources/adf/translate.py

@@ -141,7 +141,7 @@ def translate_pipeline(
 
     # C-05 (VAREX-002): synthesise init SetVariable tasks for variables with a defaultValue, seeding
     # variable_cache so @variables('X') resolves to the init task's value instead of a self-referential dangler.
-    init_variable_activities, context = _build_variable_init_activities(pipeline, context)
+    init_variable_activities, context, synthesized_init_keys = _build_variable_init_activities(pipeline, context)
     translated_activities: list[Activity] = list(init_variable_activities)
 
     sorted_activities = _topological_visit(pipeline.activities)
@@ -183,6 +183,7 @@ def translate_pipeline(
         tasks=translated_activities,
         tags={"source": "adf", "pipeline": pipeline.name},
         schedule=schedule,
+        synthesized_variable_init_keys=synthesized_init_keys,
     )
 
     # Whole-IR expression rewrite: catches @{...} tokens the per-activity translators missed (raw SQL
@@ -772,7 +773,7 @@ def _normalize_timezone(tz: Any) -> str:
 def _build_variable_init_activities(
     pipeline: AdfPipeline,
     context: TranslationContext,
-) -> tuple[list[Activity], TranslationContext]:
+) -> tuple[list[Activity], TranslationContext, frozenset[str]]:
     """Synthesise init SetVariable IR tasks for variables carrying defaultValue.
 
     C-05 (VAREX-002): without an explicit ADF SetVariable activity, a
@@ -781,13 +782,18 @@ def _build_variable_init_activities(
     ``{{tasks.X.values.X}}`` reference.  This helper emits an init task
     per default-valued variable so the variable_cache carries a real
     setter task_key.
+
+    Also returns the exact set of synthesised init task keys so the
+    package-time pruning in :func:`prepare_workflow` can identify these tasks
+    explicitly and never act on a customer-authored task.
     """
     from flowx.parser.expression_parser import resolve_expression
 
     if not pipeline.variables:
-        return [], context
+        return [], context, frozenset()
 
     init_tasks: list[Activity] = []
+    init_keys: set[str] = set()
     for var_name, var_def in pipeline.variables.items():
         default = var_def.default_value
         if default is None:
@@ -833,12 +839,13 @@ def _build_variable_init_activities(
             required_parameters=required_parameters,
         )
         init_tasks.append(init_activity)
+        init_keys.add(task_key)
         # Register the synthesised setter so @variables('X') resolves to {{tasks._init_X.values.X}}; when
         # the value is itself a DAB ref (e.g. @utcNow()), inline it directly.
         dab_ref_value = variable_value if value_kind == "dab_ref" else None
         context = context.with_variable(var_name, task_key, dab_ref_value=dab_ref_value)
         context = context.with_activity(init_activity.name, init_activity)
-    return init_tasks, context
+    return init_tasks, context, frozenset(init_keys)
 
 
 def _topological_visit(activities: list[AdfActivity]) -> list[AdfActivity]:

tests/integration/azure/adf/cases/all-activity-types/expected/bundles/resources/all_activity_types.yml

@@ -3,12 +3,6 @@ resources:
     all_activity_types:
       name: all-activity-types
       tasks:
-      - task_key: _init_status
-        notebook_task:
-          notebook_path: ../src/notebooks/init_status.py
-          base_parameters:
-            variable_name: status
-            value: new
       - task_key: append-variable
         notebook_task:
           notebook_path: ../src/notebooks/append_variable.py

tests/integration/azure/adf/cases/all-activity-types/expected/ir/all-activity-types.json

@@ -21,6 +21,9 @@
     "pause_status": "PAUSED",
     "url": "<STORAGE_ACCOUNT_ID>"
   },
+  "synthesized_variable_init_keys": [
+    "_init_status"
+  ],
   "tags": {
     "pipeline": "all-activity-types",
     "source": "adf"

tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/resources/complex_orchestration.yml

@@ -3,12 +3,6 @@ resources:
     complex_orchestration:
       name: complex-orchestration
       tasks:
-      - task_key: _init_state
-        notebook_task:
-          notebook_path: ../src/notebooks/init_state.py
-          base_parameters:
-            variable_name: state
-            value: start
       - task_key: launch-child
         run_job_task:
           job_id: ${var.child_pipeline_job_id}

tests/integration/azure/adf/cases/complex-orchestration/expected/ir/complex-orchestration.json

@@ -11,6 +11,9 @@
     "quartz_cron_expression": "0 0 0 * * ?",
     "timezone_id": "UTC"
   },
+  "synthesized_variable_init_keys": [
+    "_init_state"
+  ],
   "tags": {
     "pipeline": "complex-orchestration",
     "source": "adf"

tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/resources/complex_orchestration.yml

@@ -3,12 +3,6 @@ resources:
     complex_orchestration:
       name: complex-orchestration
       tasks:
-      - task_key: _init_state
-        notebook_task:
-          notebook_path: ../src/notebooks/init_state.py
-          base_parameters:
-            variable_name: state
-            value: start
       - task_key: launch-child
         run_job_task:
           job_id: ${var.child_pipeline_job_id}