From 8a0649ad58c2da2f54b6a4662f22a5a6498cc46b Mon Sep 17 00:00:00 2001 From: Devendra Kushwah Date: Sun, 29 Mar 2026 20:20:30 +0530 Subject: [PATCH 1/3] Fix conditional input paths handling in Argo DAG generation --- metaflow/plugins/argo/argo_workflows.py | 38 +++++++++++++------ .../plugins/argo/conditional_input_paths.py | 9 ++++- 2 files changed, 35 insertions(+), 12 deletions(-) diff --git a/metaflow/plugins/argo/argo_workflows.py b/metaflow/plugins/argo/argo_workflows.py index b7b25c8c69d..580e3f98aba 100644 --- a/metaflow/plugins/argo/argo_workflows.py +++ b/metaflow/plugins/argo/argo_workflows.py @@ -1346,19 +1346,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 diff --git a/metaflow/plugins/argo/conditional_input_paths.py b/metaflow/plugins/argo/conditional_input_paths.py index 13013b69472..7ddabe17274 100644 --- a/metaflow/plugins/argo/conditional_input_paths.py +++ b/metaflow/plugins/argo/conditional_input_paths.py @@ -18,7 +18,14 @@ 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("/") + if len(parts) < 3 or not parts[-1]: + return False + return True + trimmed = [path for path in paths if _is_resolved_path(path)] # If the input-path is from a conditional, we want to pick the one that is last-in-line in the DAG. # The order of graph parsing ensures that the steps are in reverse order of occurrence, so the first one is the latest. From 1ac98d96db92941bf340fceee27bb1dba7413c7b Mon Sep 17 00:00:00 2001 From: Devendra kushwah <105185014+devendrakushwah80@users.noreply.github.com> Date: Sun, 7 Jun 2026 22:23:33 +0530 Subject: [PATCH 2/3] Update metaflow/plugins/argo/argo_workflows.py Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- metaflow/plugins/argo/argo_workflows.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/metaflow/plugins/argo/argo_workflows.py b/metaflow/plugins/argo/argo_workflows.py index 580e3f98aba..723ae2862d0 100644 --- a/metaflow/plugins/argo/argo_workflows.py +++ b/metaflow/plugins/argo/argo_workflows.py @@ -1373,7 +1373,7 @@ def _build_input_path(n): 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. From d57f6f9986228ba95c69d2384ebfc2fb1a77fe39 Mon Sep 17 00:00:00 2001 From: Devendra kushwah <105185014+devendrakushwah80@users.noreply.github.com> Date: Sun, 7 Jun 2026 22:23:43 +0530 Subject: [PATCH 3/3] Update metaflow/plugins/argo/conditional_input_paths.py Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- metaflow/plugins/argo/conditional_input_paths.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/metaflow/plugins/argo/conditional_input_paths.py b/metaflow/plugins/argo/conditional_input_paths.py index 7ddabe17274..0b961129f11 100644 --- a/metaflow/plugins/argo/conditional_input_paths.py +++ b/metaflow/plugins/argo/conditional_input_paths.py @@ -22,6 +22,8 @@ def _is_resolved_path(path): if "{{" in path: return False parts = path.split("/") + # Paths from skipped conditional steps resolve to "argo-//" + # (empty task-id segment), which must be filtered out. if len(parts) < 3 or not parts[-1]: return False return True