run_async runs a lazy variadic component early, once per partial input, when its sender can still run again #13284

Describe the bug

When a lazy variadic socket gets its input from a component that can still run again, Pipeline.run_async() (and run_async_generator()) runs the receiving component early, once per partial input. Pipeline.run() waits and runs it once with all the inputs. The final outputs differ, and every component downstream also runs twice in async mode.

The smallest case uses only built-in joiners: a BranchJoiner (greedy variadic) fed by two branches, one of them a step longer, followed by a ListJoiner (lazy variadic). No loops.

To Reproduce

import asyncio

from haystack import Pipeline, component
from haystack.components.joiners import BranchJoiner, ListJoiner


@component
class Step:
    """Stand-in for any component: passes its input through and records the call."""

    calls: list[str] = []

    def __init__(self, name: str):
        self.name = name

    @component.output_types(out=list[str])
    def run(self, items: list[str]):
        Step.calls.append(self.name)
        return {"out": items}


def build() -> Pipeline:
    p = Pipeline()
    p.add_component("source", Step("source"))
    p.add_component("slow_branch", Step("slow_branch"))
    p.add_component("branch_joiner", BranchJoiner(list[str]))
    p.add_component("list_joiner", ListJoiner(list[str]))
    p.add_component("sink", Step("sink"))
    p.connect("source.out", "slow_branch.items")
    p.connect("source.out", "branch_joiner.value")
    p.connect("slow_branch.out", "branch_joiner.value")
    p.connect("branch_joiner.value", "list_joiner.values")
    p.connect("list_joiner.values", "sink.items")
    return p


data = {"source": {"items": ["a"]}}
Step.calls.clear()
print("run:      ", build().run(data, include_outputs_from={"list_joiner"}), Step.calls)
Step.calls.clear()
print("run_async:", asyncio.run(build().run_async(data, include_outputs_from={"list_joiner"})), Step.calls)

run:       {'list_joiner': {'values': ['a', 'a']}, 'sink': {'out': ['a', 'a']}} ['source', 'slow_branch', 'sink']
run_async: {'list_joiner': {'values': ['a']}, 'sink': {'out': ['a']}} ['source', 'slow_branch', 'sink', 'sink']

The result is the same on every run, so this is not the completion-order nondeterminism from #13204.

Expected behavior

run_async() gives the same result as run(). branch_joiner runs twice, once per branch, and list_joiner runs once with both outputs, so sink runs once.

What happens

1. source runs. branch_joiner (HIGHEST priority) runs in isolation with source's output and sends it to list_joiner.
2. list_joiner is now READY: _calculate_priority returns READY when all_predecessors_executed(), and its only predecessor, branch_joiner, has executed once. run_async_generator schedules READY components together, so list_joiner runs at the same time as slow_branch, with one input.
3. slow_branch finishes, branch_joiner runs again, and list_joiner and sink run a second time. The second outputs replace the first ones in the pipeline outputs.

run() gives the expected result partly because of when it rebuilds its queue. It only refills the priority queue when _is_queue_stale(). So after branch_joiner's first run it pops slow_branch, which is still READY from the earlier fill, before it sees that list_joiner became READY. By the time it refills, branch_joiner has a second input and runs first (HIGHEST). The root cause looks shared: READY means "every predecessor has run at least once", not "no predecessor can still send". run() doesn't hit it here because of its scheduling order.

Additional context

• I found this by differential fuzzing run() against run_async() on random small acyclic pipelines (sum, router, optional-input and greedy-variadic components). 22 of 3,000 seeds differed, and every difference involved a lazy variadic or optional input downstream of a greedy-variadic component whose upstream was still running.
• A possible direction: treat a lazy variadic socket as READY only when none of its senders can still receive input. That is the reachability _is_pipeline_possibly_blocked / can_not_receive_inputs_from_pipeline already reason about. I'm happy to work on a PR if you think that's the right approach. It changes core scheduling, so I wanted to check first.

System:

• Haystack: main @ aca9fc2 (3.1.0rc0)
• Python 3.12.3, Linux (WSL2)

👋 Hello there! This issue will be handled internally and isn't open for external contributions. If you'd like to contribute, please take a look at issues labeled contributions welcome or good first issue. We'd really appreciate it!