Fix conditional input paths handling for Argo DAG generation

metaflow/plugins/argo/argo_workflows.py

@@ -1380,19 +1380,35 @@ def _visit(
                 )
             else:
                 # Every other node needs only input-paths
-                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
-                            ],
-                            # NOTE: We set zlibmin to infinite because zlib compression for the Argo input-paths breaks template value substitution.
-                            zlibmin=inf,
+                def _build_input_path(n):
+                    sanitized = self._sanitize(n)
+                    parent_node = self.graph[n]
+                    if self._is_conditional_node(parent_node):
+                        return (
+                            "argo-{{workflow.name}}/%s/"
+                            "{{=(tasks['%s'].status == 'Succeeded' "
+                            "? tasks['%s'].outputs.parameters['task-id'] : '')}}"
+                            % (n, sanitized, sanitized)
                         )
+                    else:
+                        return (
+                            "argo-{{workflow.name}}/%s/"
+                            "{{tasks.%s.outputs.parameters.task-id}}"
+                            % (n, sanitized)
+                        )
+                has_conditional_parent = any(
+                    self._is_conditional_node(self.graph[n]) for n in node.in_funcs
+                )
+                if has_conditional_parent:
+                    input_paths_value = ",".join(
+                        _build_input_path(n) for n in node.in_funcs
                     )
-                ]
+                else:
+                    input_paths_value = compress_list(
+                        [_build_input_path(n) for n in node.in_funcs],
+                        zlibmin=inf,
+                    )
+                parameters = [Parameter("input-paths").value(input_paths_value)]
                 # NOTE: Due to limitations with Argo Workflows Parameter size we
                 #       can not pass arbitrarily large lists of task id's to join tasks.
                 #       Instead we ensure that task id's for foreach tasks can be

metaflow/plugins/argo/conditional_input_paths.py

@@ -25,7 +25,16 @@ 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]
+    def _is_resolved_path(path):
+        if "{{" in path:
+            return False
+        parts = path.split("/")
+        # Paths from skipped conditional steps resolve to "argo-<run>/<step>/"
+        # (empty task-id segment), which must be filtered out.
+        if len(parts) < 3 or not parts[-1]:
+            return False
+        return True
+    trimmed = [path for path in paths if _is_resolved_path(path)]
 
     skippable_steps = [step for step in skippable_steps if step]
     skippable_step_set = set(skippable_steps)