We have two versions of a flow that only differ in the name of their steps (see flowspecs at the end). Both flows work completely fine locally. For use in argo, we deploy a fresh template with

argo-workflows create

and directly trigger it with

argo-workflows trigger

. spec1 works fine, all steps and the flow itself succeeds. For spec2, we get this error in the

join_risk_chunk

step:

Setting up task environment.
Downloading code package...
Code package downloaded.
Task is starting.
    Data store error:
    No completed attempts of the task was found for task 'DAGDebug/YXJnby1kYWdkZWJ1Zy1zcHQ0Zy9yaXNrL3QtMzY2ZTYwMGY=/risk_chunk/t-536b3977'
time="2025-11-06T14:43:05 UTC" level=info msg="sub-process exited" argo=true error="<nil>"
time="2025-11-06T14:43:05 UTC" level=info msg="no need to save parameter - on overlapping volume: /mnt/out/task_id" argo=true
Error: exit status 1

In previous work, we have also seen this error, but have been unable to reproduce with a minimal example:

Data missing: Some input datastores are missing. Expected: 33 Actual: 0

It is absolutely not clear to me what (combination of) things trigger this problem. If we rename the

join_risk_chunk

in the broken flowspec to

some_step_name

, it runs completely fine. Removing the conditional transition from the start step also fixes the issue (regardless of what the later join step is called). Working flowspec:

from metaflow import FlowSpec, step


class DAGDebug(FlowSpec):
    @step
    def start(self):
        self.next_step = "step2"
        self.next(
            {"foreach1": self.foreach1, "step2": self.step2},
            condition="next_step",
        )

    @step
    def foreach1(self):
        self.next(self.foreach_proc, foreach="grouped_work_list")

    @step
    def foreach_proc(self):
        self.next(self.join_1)

    @step
    def join_1(self, inputs):
        self.merge_artifacts(inputs)
        self.next(self.step2)

    @step
    def step2(self):
        self.next(self.step3)

    @step
    def step3(self):
        self.skip_step_4 = True
        self.next({True: self.foreach2, False: self.step4}, condition="skip_step_4")

    @step
    def step4(self):
        self.next(self.foreach2)

    @step
    def foreach2(self):
        self.risk_worklist = list(range(2))
        self.next(self.foreach_proc2, foreach="risk_worklist")

    @step
    def foreach_proc2(self):
        self.site_results = [1, 2, 3]
        self.next(self.join2)

    @step
    def join2(self, inputs):
        self.next(self.end)

    @step
    def end(self):
        pass


if __name__ == "__main__":
    DAGDebug()

Broken flowspec:

from metaflow import FlowSpec, step


class DAGDebug(FlowSpec):
    @step
    def start(self):
        self.next_step = "impacts"
        self.next(
            {"split_chunks": self.split_chunks, "impacts": self.impacts},
            condition="next_step",
        )

    @step
    def split_chunks(self):
        self.next(self.process_metrics_chunk, foreach="grouped_work_list")

    @step
    def process_metrics_chunk(self):
        self.next(self.join_process_metrics_chunk)

    @step
    def join_process_metrics_chunk(self, inputs):
        self.next(self.aggregate)

    @step
    def aggregate(self):
        self.next(self.impacts)

    @step
    def impacts(self):
        self.skip_evaluate = True
        self.next({True: self.risk, False: self.evaluate}, condition="skip_evaluate")

    @step
    def evaluate(self):
        self.next(self.risk)

    @step
    def risk(self):
        self.risk_worklist = list(range(2))
        self.next(self.risk_chunk, foreach="risk_worklist")

    @step
    def risk_chunk(self):
        self.site_results = [1, 2, 3]
        self.next(self.join_risk_chunk)

    @step
    def join_risk_chunk(self, inputs):
        self.next(self.end)

    @step
    def end(self):
        pass


if __name__ == "__main__":
    DAGDebug()

Just updated to latest and it's working like a charm. Thanks again for the rapid turnaround 🙂