fix argo conditionals issue with new argo

Author: saikonen

metaflow/plugins/argo/argo_workflows.py

@@ -1225,6 +1225,20 @@ def _skippable_input_steps_in_dag_order(self, node):
             reverse=True,
         )
 
+    def _input_path_ref(self, node_name):
+        sanitized = self._sanitize(node_name)
+        pred_node = self.graph[node_name]
+        if self._is_conditional_node(pred_node) or pred_node.type == "split-switch":
+            return (
+                "argo-{{workflow.name}}/%s/"
+                "{{=tasks['%s']?.outputs?.parameters['task-id'] ?? 'SKIPPED'}}"
+                % (node_name, sanitized)
+            )
+        return "argo-{{workflow.name}}/%s/{{tasks.%s.outputs.parameters.task-id}}" % (
+            node_name,
+            sanitized,
+        )
+
     def _is_recursive_node(self, node):
         return node.name in self.recursive_nodes
 
@@ -1378,11 +1392,7 @@ def _visit(
                 parameters = [
                     Parameter("input-paths").value(
                         compress_list(
-                            [
-                                "argo-{{workflow.name}}/%s/{{tasks.%s.outputs.parameters.task-id}}"
-                                % (n, self._sanitize(n))
-                                for n in node.in_funcs
-                            ],
+                            [self._input_path_ref(n) for n in node.in_funcs],
                             # NOTE: We set zlibmin to infinite because zlib compression for the Argo input-paths breaks template value substitution.
                             zlibmin=inf,
                         )

metaflow/plugins/argo/conditional_input_paths.py

@@ -25,7 +25,9 @@ def generate_input_paths(input_paths, skippable_steps):
     # strip these out of the list.
 
     # all pathspecs of leading steps that executed.
-    trimmed = [path for path in paths if not "{{" in path]
+    trimmed = [
+        path for path in paths if "{{" not in path and not path.endswith("/SKIPPED")
+    ]
 
     skippable_steps = [step for step in skippable_steps if step]
     skippable_step_set = set(skippable_steps)

test/unit/test_argo_conditional_input_paths.py

@@ -75,6 +75,10 @@ def _unresolved_task_path(step_name):
     )
 
 
+def _skipped_task_path(step_name):
+    return "%s/%s/SKIPPED" % (RUN_ID, step_name)
+
+
 def test_chain_skip_fallback_uses_latest_executed_split_switch(chain_skip_argo):
     node = chain_skip_argo.graph["end"]
     assert node.in_funcs == ["start", "step2", "step3"]
@@ -91,6 +95,20 @@ def test_chain_skip_fallback_uses_latest_executed_split_switch(chain_skip_argo):
     assert _decode_input_paths(result) == [_task_path("step2")]
 
 
+def test_chain_skip_with_skipped_sentinel(chain_skip_argo):
+    """Same as above but with SKIPPED sentinel (Argo v3.7.11+ behavior)."""
+    node = chain_skip_argo.graph["end"]
+    skippable_steps = chain_skip_argo._skippable_input_steps_in_dag_order(node)
+
+    input_paths = _encode_input_paths(
+        [_task_path("start"), _task_path("step2"), _skipped_task_path("step3")]
+    )
+
+    result = generate_input_paths(input_paths, skippable_steps)
+
+    assert _decode_input_paths(result) == [_task_path("step2")]
+
+
 @pytest.mark.parametrize(
     "paths, skippable_steps, expected",
     [
@@ -105,8 +123,30 @@ def test_chain_skip_fallback_uses_latest_executed_split_switch(chain_skip_argo):
             [_task_path("branch")],
         ),
         ([_task_path("step"), _task_path("step2")], ["step"], [_task_path("step2")]),
+        (
+            [_task_path("start"), _skipped_task_path("branch")],
+            ["start"],
+            [_task_path("start")],
+        ),
+        (
+            [_task_path("start"), _task_path("step2"), _skipped_task_path("step3")],
+            ["step2", "start"],
+            [_task_path("step2")],
+        ),
+        (
+            [_skipped_task_path("left"), _task_path("right")],
+            [],
+            [_task_path("right")],
+        ),
+    ],
+    ids=[
+        "normal_join",
+        "non_skippable_executed",
+        "exact_step_name",
+        "skipped_non_skippable",
+        "skipped_with_skippable_fallback",
+        "skipped_no_skippable_steps",
     ],
-    ids=["normal_join", "non_skippable_executed", "exact_step_name"],
 )
 def test_generate_input_paths_filters_by_exact_step_name(
     paths, skippable_steps, expected