Fix nested conditionals running skipped branches

In local execution, a conditional nested inside the active branch of
another conditional (inline, or inside a subworkflow called from that
branch) reused the parent's execution state, which was already marked
BRANCH_ACTIVE. Every branch of the nested conditional therefore ran its
tasks, even the ones whose condition was false.

Give every local conditional section its own execution state starting
in BRANCH_SKIPPED, so only the selected branch of a nested conditional
runs. Sections nested in a skipped branch still become
SkippedConditionalSection, as before.

Closes flyteorg/flyte#6426

Signed-off-by: Tianlang (Jackie) Chen <305351710+jackiectl2@users.noreply.github.com>

Author: jackiectl2

flytekit/core/context_manager.py

@@ -838,15 +838,13 @@ def enter_conditional_section(self) -> FlyteContext.Builder:
 
             if self.execution_state:
                 if self.execution_state.is_local_execution():
-                    if self.in_a_condition:
-                        if self.execution_state.branch_eval_mode == BranchEvalMode.BRANCH_SKIPPED:
-                            self.execution_state = self.execution_state.with_params()
-                    else:
-                        # In case of local workflow execution we should ensure a conditional section
-                        # is created so that skipped branches result in tasks not being executed
-                        self.execution_state = self.execution_state.with_params(
-                            branch_eval_mode=BranchEvalMode.BRANCH_SKIPPED
-                        )
+                    # In case of local workflow execution we should ensure a conditional section
+                    # is created so that skipped branches result in tasks not being executed.
+                    # This also applies to nested conditionals (inline, or inside a subworkflow called
+                    # from an active branch): they must not inherit the parent's active branch state.
+                    self.execution_state = self.execution_state.with_params(
+                        branch_eval_mode=BranchEvalMode.BRANCH_SKIPPED
+                    )
 
             self.in_a_condition = True
             return self

tests/flytekit/unit/core/test_conditions.py

@@ -494,6 +494,93 @@ def multiplier_2(my_input: float) -> float:
     assert res == 20
 
 
+def test_nested_condition_runs_only_selected_branch():
+    executed = []
+
+    @task
+    def inner_if(n: float) -> float:
+        executed.append("inner_if")
+        return n
+
+    @task
+    def inner_else(n: float) -> float:
+        executed.append("inner_else")
+        return -n
+
+    @task
+    def outer_else(n: float) -> float:
+        executed.append("outer_else")
+        return 0.0
+
+    @workflow
+    def wf(my_input: float) -> float:
+        return (
+            conditional("outer")
+            .if_(my_input > 0.1)
+            .then(
+                conditional("inner")
+                .if_(my_input < 0.5)
+                .then(inner_if(n=my_input))
+                .else_()
+                .then(inner_else(n=my_input))
+            )
+            .else_()
+            .then(outer_else(n=my_input))
+        )
+
+    assert wf(my_input=0.6) == -0.6
+    assert executed == ["inner_else"]
+
+    executed.clear()
+    assert wf(my_input=0.3) == 0.3
+    assert executed == ["inner_if"]
+
+    executed.clear()
+    assert wf(my_input=0.0) == 0.0
+    assert executed == ["outer_else"]
+
+
+def test_condition_in_subworkflow_called_from_branch():
+    # https://github.com/flyteorg/flyte/issues/6426
+    executed = []
+
+    @task
+    def do_test() -> bool:
+        executed.append("do_test")
+        return False
+
+    @task
+    def do_nothing1() -> bool:
+        executed.append("do_nothing1")
+        return True
+
+    @task
+    def do_nothing2() -> bool:
+        executed.append("do_nothing2")
+        return True
+
+    @task
+    def do_nothing3() -> bool:
+        executed.append("do_nothing3")
+        return True
+
+    @workflow
+    def sub_workflow() -> bool:
+        some_test = do_test()
+        return conditional("inner").if_(some_test.is_true()).then(do_nothing1()).else_().then(do_nothing2())
+
+    @workflow
+    def top_workflow(input_value: bool) -> bool:
+        return conditional("outer").if_(input_value.is_true()).then(sub_workflow()).else_().then(do_nothing3())  # type: ignore
+
+    assert top_workflow(input_value=True) is True
+    assert executed == ["do_test", "do_nothing2"]
+
+    executed.clear()
+    assert top_workflow(input_value=False) is True
+    assert executed == ["do_nothing3"]
+
+
 def test_echo_in_condition():
     echo1 = Echo(name="echo", inputs={"a": typing.Optional[float]})