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

src/flowx/bundler/dab_writer.py

@@ -1390,7 +1390,17 @@ 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")
+            # A join after an ADF IfCondition must reproduce the container's
+            # ``dependsOn [Succeeded]`` semantics: it runs when the *taken*
+            # branch did not fail, and skips when it did.  Because exactly one
+            # branch executes, the other branch's terminals are SKIPPED.
+            # ``NONE_FAILED`` treats SKIPPED as non-failing (so the untaken
+            # branch never blocks the join) but treats FAILED as blocking (so a
+            # failure in the taken branch propagates).  ``AT_LEAST_ONE_SUCCESS``
+            # is wrong here: a sibling success (e.g. a ForEach inputs-bridge)
+            # masks a taken-branch failure, letting the join run when ADF would
+            # have skipped it.
+            task.setdefault("run_if", "NONE_FAILED")
 
 
 _TASK_VALUE_REF = re.compile(r"\{\{tasks\.([^.]+)\.values\.[^}]+\}\}")

src/flowx/models/ir.py

@@ -616,11 +616,10 @@ class Pipeline:
         tasks: Ordered list of translated activities.
         tags: System and user-defined tags.
         not_translatable: Entries describing properties that could not be translated.
-        bundle_variables: DAB bundle-variable declarations (name -> ``{"description", "default"}``)
-            for factory globals hoisted under the ``bundle_variable`` resolution policy.
-        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.
+        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.
     """
 
     name: str
@@ -636,7 +635,7 @@ class Pipeline:
     migration_status: str = "included"
     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

@@ -73,10 +73,7 @@ def collapse_motifs(
         tasks=new_tasks,
         tags=pipeline.tags,
         not_translatable=pipeline.not_translatable,
-        reconciliation_status=pipeline.reconciliation_status,
-        migration_status=pipeline.migration_status,
-        audit=dict(pipeline.audit),
-        translation_configuration=pipeline.translation_configuration,
+        synthesized_variable_init_keys=pipeline.synthesized_variable_init_keys,
     )