Cut collection stream residue

.changeset/fan-out-fan-in.md

@@ -2,4 +2,4 @@
 "llama-index-workflows": minor
 ---
 
-Fan-out/fan-in now lives in step signatures: returning `list[E]` fans out a batch and taking `list[E]` joins it on close, with no manual cardinality threading. `Collect(Take(n))` releases a join early on the nth arrival, and `list[A | B]` joins a flat heterogeneous batch.
+Fan-out/fan-in now lives in step signatures: returning `list[E]` opens a finite collection stream and taking `list[E]` joins it on close. `Collect(Take(n))` releases a join early on the nth arrival, and `list[A | B]` joins a flat heterogeneous stream.

packages/llama-index-workflows/src/workflows/decorators.py

@@ -66,13 +66,12 @@ class StepConfig:
     # when the step declares a single ``list[E]`` parameter. The element types are
     # a tuple — ``list[Done]`` -> ``(Done,)``; a union flat list ``list[A | B]`` ->
     # ``(A, B)`` (every member routes to the step). The step buffers incoming
-    # members by innermost stream id and releases per ``collection_collect``.
-    collection_collect_param: tuple[str, tuple[Any, ...]] | None = None
-    # The resolved ``Collect`` marker for the collection collect parameter,
-    # carrying the release cardinality (``All`` / ``Take``). A bare
-    # ``list[E]`` parameter resolves to ``Collect()`` (``All``). None when the
-    # step is not a collection collect.
-    collection_collect: Collect | None = None
+    # members by innermost stream id and releases per ``collection_policy``.
+    collection_param: tuple[str, tuple[Any, ...]] | None = None
+    # The resolved ``Collect`` marker for the collection parameter. A bare
+    # ``list[E]`` parameter resolves to ``Collect()`` (``All``). None for steps
+    # without a collection parameter.
+    collection_policy: Collect | None = None
     role: StepRole = "step"
     # Only meaningful when role == "catch_error".
     # None means wildcard — covers any step not claimed by a scoped handler.
@@ -220,7 +219,7 @@ def make_step_function(
     # Collect-mode (heterogeneous fan-in): more than one event parameter. The
     # step accepts every declared event type for routing, and collects one of
     # each (in declaration order) before firing once. ``event_name`` keeps the
-    # first parameter for compatibility with single-event call paths.
+    # first parameter, which is still the carrier event passed into the wrapper.
     collect_params: list[tuple[str, Any]] | None = None
     if len(spec.accepted_events) > 1:
         collect_params = [
@@ -241,8 +240,8 @@ def make_step_function(
         skip_graph_checks=skip_graph_checks or [],
         collect_params=collect_params,
         is_fan_out=spec.is_fan_out,
-        collection_collect_param=spec.collection_collect_param,
-        collection_collect=spec.collection_collect,
+        collection_param=spec.collection_param,
+        collection_policy=spec.collection_policy,
     )
 
     return casted

packages/llama-index-workflows/src/workflows/representation/validate.py

@@ -632,9 +632,9 @@ def _stream_level_types_by_producer(
     """Return event types produced inside each returned-list stream level."""
 
     collects: dict[str, tuple[Any, ...]] = {
-        name: cfg.collection_collect_param[1]
+        name: cfg.collection_param[1]
         for name, cfg in steps.items()
-        if cfg.collection_collect_param is not None
+        if cfg.collection_param is not None
     }
 
     def same_level_types(
@@ -650,7 +650,7 @@ def _stream_level_types_by_producer(
             for name, cfg in steps.items():
                 if event_type not in cfg.accepted_events:
                     continue
-                if cfg.collection_collect_param is not None:
+                if cfg.collection_param is not None:
                     continue
                 if cfg.is_fan_out:
                     if name in guard:
@@ -672,15 +672,15 @@ def _stream_level_types_by_producer(
     }
 
 
-def _validate_collection_collect_bindings(steps: dict[str, StepConfig]) -> None:
+def _validate_collection_bindings(steps: dict[str, StepConfig]) -> None:
     """Require list[E] fan-in to bind to a static returned-list producer."""
 
     level_types_by_producer = _stream_level_types_by_producer(steps)
     errors: list[str] = []
     for step_name, cfg in steps.items():
-        if cfg.collection_collect_param is None:
+        if cfg.collection_param is None:
             continue
-        param_name, element_types = cfg.collection_collect_param
+        param_name, element_types = cfg.collection_param
         missing = [
             event_type
             for event_type in _event_types(element_types)
@@ -728,7 +728,7 @@ def _validate_workflow(
     stop_event_class = _ensure_stop_event_class(steps, workflow_cls_name)
 
     _validate_collect_param_slots(steps)
-    _validate_collection_collect_bindings(steps)
+    _validate_collection_bindings(steps)
 
     uses_hitl = _validate_event_connectivity(steps, start_event_class)
 

packages/llama-index-workflows/src/workflows/runtime/control_loop.py

@@ -863,8 +863,7 @@ def _process_step_result_tick(
                 for worker in state.workers.values():
                     worker.collected_events.clear()
                     worker.collected_waiters.clear()
-                # Drop all stream lineage: the run is over, so no stream can close
-                # and no collect can fire. Leaving it would leak into a snapshot.
+                # Drop open collection state; no release can fire after the run ends.
                 _clear_collection_state(state)
                 commands.append(CommandCompleteRun(result=result.result))
             elif isinstance(result.result, Event):
@@ -1099,9 +1098,6 @@ def _process_step_result_tick(
     )
 
     if not skip_accounting and is_fan_out and fan_out_stream_id is not None:
-        # Fan-out descent. The enclosing stream trades this source work item for
-        # one placeholder per collect bound to the child stream; the placeholder
-        # resolves when that collect emits its summary back to the parent scope.
         bindings = state.config.bindings_for_source(tick.step_name)
         accepting_binding_ids = tuple(binding.id for binding in bindings)
         seed = sum(_count_accepting_steps(state, type(m)) for m in emitted_non_stop)
@@ -1115,14 +1111,14 @@ def _process_step_result_tick(
             open_work_items=seed,
             closed_to_new_items=True,
         )
-        # The source work item is replaced by collect placeholders.
+        # The parent work item now waits for each child collection release.
         commands.extend(
             _apply_stream_work_delta(state, enclosing, len(accepting_binding_ids) - 1)
         )
         if seed == 0:
             commands.extend(_close_collection_stream(state, fan_out_stream_id))
     elif not skip_accounting:
-        # (b) Same-level resolution (1:1 step, or a collect step firing its
+        # Same-level resolution (1:1 step, or a collect step firing its
         # summary). Remove this work item and add its same-level successors: one
         # work item per accepting step per emitted event. A step that returns
         # None adds zero successors and simply leaves the set.
@@ -1271,7 +1267,7 @@ def _process_add_event_tick(
         if is_accepted and (tick.step_name is None or tick.step_name == step_name):
             handled = True
             worker_state = state.workers[step_name]
-            if worker_state.config.collection_collect_param is not None:
+            if worker_state.config.collection_param is not None:
                 if not tick.scope_path:
                     continue
                 stream_id = tick.scope_path[-1]
@@ -1482,7 +1478,7 @@ def _process_collection_stream_closed_tick(
     bindings = state.config.bindings_for_source(tick.source_step)
     for binding in bindings:
         worker_state = state.workers.get(binding.target_step)
-        if worker_state is None or worker_state.config.collection_collect_param is None:
+        if worker_state is None or worker_state.config.collection_param is None:
             continue
         key = _release_state_key(tick.stream_id, binding.id)
         release_state = state.collection_release_states.pop(

packages/llama-index-workflows/src/workflows/runtime/types/internal_state.py

@@ -353,9 +353,9 @@ def _binding_id(
 def _compute_collection_bindings(workflow: Workflow) -> dict[str, CollectionBinding]:
     steps = {name: fn._step_config for name, fn in workflow._get_steps().items()}
     collects: dict[str, tuple[Any, ...]] = {
-        name: cfg.collection_collect_param[1]
+        name: cfg.collection_param[1]
         for name, cfg in steps.items()
-        if cfg.collection_collect_param is not None
+        if cfg.collection_param is not None
     }
 
     def same_level_types(seed_types: Any, guard: frozenset[str]) -> set[type[Event]]:
@@ -369,7 +369,7 @@ def _compute_collection_bindings(workflow: Workflow) -> dict[str, CollectionBind
             for name, cfg in steps.items():
                 if t not in cfg.accepted_events:
                     continue
-                if cfg.collection_collect_param is not None:
+                if cfg.collection_param is not None:
                     continue
                 if cfg.is_fan_out:
                     if name in guard:
@@ -391,7 +391,7 @@ def _compute_collection_bindings(workflow: Workflow) -> dict[str, CollectionBind
             item_types = tuple(_event_types(collect_types))
             if not any(et in level_types for et in item_types):
                 continue
-            policy = steps[target_step].collection_collect
+            policy = steps[target_step].collection_policy
             if policy is None:
                 continue
             binding = CollectionBinding(

packages/llama-index-workflows/src/workflows/runtime/types/results.py

@@ -89,7 +89,7 @@ class StepWorkerState:
 
 @dataclass(frozen=True)
 class CollectionReleasePayload:
-    """List payload supplied to a collection collect step invocation."""
+    """List payload supplied to a collection step invocation."""
 
     binding_id: str
     stream_id: str

packages/llama-index-workflows/src/workflows/runtime/types/step_function.py

@@ -192,8 +192,8 @@ def as_step_worker_function(
             config = workflow._get_steps()[step_name]._step_config
             collected_binding: dict[str, Event] | None = None
             collection_binding: dict[str, list[Event]] | None = None
-            if config.collection_collect_param is not None:
-                param_name, _ = config.collection_collect_param
+            if config.collection_param is not None:
+                param_name, _ = config.collection_param
                 payload = state.collection_release_payload
                 collection_binding = {
                     param_name: list(payload.events if payload is not None else [])

packages/llama-index-workflows/src/workflows/utils.py

@@ -47,11 +47,11 @@ class StepSignatureSpec(BaseModel):
     # the step declares a single ``list[E]`` collect parameter, else None. The
     # element types are a tuple — ``list[Done]`` -> ``(Done,)``; a union flat list
     # ``list[A | B]`` -> ``(A, B)``.
-    collection_collect_param: tuple[str, tuple[Any, ...]] | None = None
-    # The resolved ``Collect`` marker for the collection collect parameter. A bare
-    # ``list[E]`` parameter resolves to ``Collect()`` (``All`` cardinality). None
-    # when the step is not a collection collect.
-    collection_collect: Any | None = None
+    collection_param: tuple[str, tuple[Any, ...]] | None = None
+    # The resolved ``Collect`` marker for the collection parameter. A bare
+    # ``list[E]`` parameter resolves to ``Collect()`` (``All`` cardinality).
+    # None for steps without a collection parameter.
+    collection_policy: Any | None = None
     # Fan-out producer: True when the return annotation is ``list[E]``.
     is_fan_out: bool = False
 
@@ -86,8 +86,8 @@ def inspect_signature(
     context_parameter = None
     context_state_type = None
     resources = []
-    collection_collect_param: tuple[str, tuple[Any, ...]] | None = None
-    collection_collect: Collect | None = None
+    collection_param: tuple[str, tuple[Any, ...]] | None = None
+    collection_policy: Collect | None = None
 
     # Inspect function parameters
     for name, t in sig.parameters.items():
@@ -150,15 +150,15 @@ def inspect_signature(
         if element_types is not None:
             marker = collect_marker if collect_marker is not None else Collect()
             _validate_collect_marker(marker, name)
-            if collection_collect_param is not None:
+            if collection_param is not None:
                 msg = (
-                    "A step may declare at most one collection (list[E]) collect "
-                    f"parameter, but found both {collection_collect_param[0]!r} and "
-                    f"{name!r}. Multi-slot collects are not supported."
+                    "A step may declare at most one list[E] collection parameter, "
+                    f"but found both {collection_param[0]!r} and {name!r}. "
+                    "Multi-slot collects are not supported."
                 )
                 raise WorkflowValidationError(msg)
-            collection_collect_param = (name, tuple(element_types))
-            collection_collect = marker
+            collection_param = (name, tuple(element_types))
+            collection_policy = marker
             accepted_events[name] = list(element_types)
             continue
         if collect_marker is not None:
@@ -185,8 +185,8 @@ def inspect_signature(
         context_parameter=context_parameter,
         context_state_type=context_state_type,
         resources=resources,
-        collection_collect_param=collection_collect_param,
-        collection_collect=collection_collect,
+        collection_param=collection_param,
+        collection_policy=collection_policy,
         is_fan_out=_is_fan_out_return(fn, localns=localns),
     )
 
@@ -270,11 +270,11 @@ def validate_step_signature(spec: StepSignatureSpec) -> None:
     if num_of_events == 0:
         msg = "Step signature must have at least one parameter annotated as type Event"
         raise WorkflowValidationError(msg)
-    if spec.collection_collect_param is not None and num_of_events > 1:
-        # A collection (list[E]) collect parameter alongside other event params is a
+    if spec.collection_param is not None and num_of_events > 1:
+        # A list[E] collection parameter alongside other event params is a
         # multi-slot collect, which this contract does not support.
         msg = (
-            "A collection (list[E]) collect parameter cannot be combined with other "
+            "A list[E] collection parameter cannot be combined with other "
             "event parameters. Use a single list[E] parameter, or multiple "
             "single-event parameters for a heterogeneous join."
         )

packages/llama-index-workflows/tests/runtime/test_state.py

@@ -61,7 +61,7 @@ def test_from_dict_auto_migrates_v1_in_progress_strings() -> None:
     attempt = worker.in_progress[0]
     assert attempt.event == event
     assert attempt.attempts == 0
-    # New lineage fields fall back to empty defaults.
+    # New stream fields fall back to empty defaults.
     assert attempt.scope_path == []
     assert result.stream_seq == 0
     assert result.streams == {}

packages/llama-index-workflows/tests/test_collect_algebra.py

@@ -97,9 +97,9 @@ def test_bare_list_infers_collect_all() -> None:
         return join
 
     cfg = _config_for(build)
-    assert cfg.collection_collect_param == ("events", (Done,))
-    assert cfg.collection_collect is not None
-    assert isinstance(cfg.collection_collect.cardinality, All)
+    assert cfg.collection_param == ("events", (Done,))
+    assert cfg.collection_policy is not None
+    assert isinstance(cfg.collection_policy.cardinality, All)
 
 
 def test_annotated_collect_is_synonym_for_bare_list() -> None:
@@ -111,9 +111,9 @@ def test_annotated_collect_is_synonym_for_bare_list() -> None:
         return join
 
     cfg = _config_for(build)
-    assert cfg.collection_collect_param == ("events", (Done,))
-    assert cfg.collection_collect is not None
-    assert isinstance(cfg.collection_collect.cardinality, All)
+    assert cfg.collection_param == ("events", (Done,))
+    assert cfg.collection_policy is not None
+    assert isinstance(cfg.collection_policy.cardinality, All)
 
 
 def test_annotated_take_cardinality_inferred() -> None:
@@ -125,8 +125,8 @@ def test_annotated_take_cardinality_inferred() -> None:
         return join
 
     cfg = _config_for(build)
-    assert cfg.collection_collect is not None
-    assert cfg.collection_collect.cardinality == Take(2)
+    assert cfg.collection_policy is not None
+    assert cfg.collection_policy.cardinality == Take(2)
 
 
 def test_union_flat_list_infers_all_member_types() -> None:
@@ -138,13 +138,13 @@ def test_union_flat_list_infers_all_member_types() -> None:
         return report
 
     cfg = _config_for(build)
-    assert cfg.collection_collect_param is not None
-    assert cfg.collection_collect_param[1] == (Done, Skipped)
+    assert cfg.collection_param is not None
+    assert cfg.collection_param[1] == (Done, Skipped)
     assert Done in cfg.accepted_events
     assert Skipped in cfg.accepted_events
 
 
-def test_single_event_param_is_not_collection_collect() -> None:
+def test_single_event_param_is_not_collection_param() -> None:
     def build(w: type[Workflow]) -> StepFunction:
         @free_step(workflow=w)
         async def work(ev: Task) -> Done:  # type: ignore[unused-ignore]
@@ -153,8 +153,8 @@ def test_single_event_param_is_not_collection_collect() -> None:
         return work
 
     cfg = _config_for(build)
-    assert cfg.collection_collect_param is None
-    assert cfg.collection_collect is None
+    assert cfg.collection_param is None
+    assert cfg.collection_policy is None
 
 
 # --------------------------------------------------------------------------- #
@@ -214,7 +214,7 @@ def test_two_list_params_rejected_as_multi_slot() -> None:
     class _W(Workflow):
         pass
 
-    with pytest.raises(WorkflowValidationError, match="at most one collection"):
+    with pytest.raises(WorkflowValidationError, match=r"at most one list\[E\]"):
 
         @free_step(workflow=_W)
         async def merge(a: list[Done], b: list[Skipped]) -> StopEvent:  # type: ignore[unused-ignore]

packages/llama-index-workflows/tests/test_collection_stream_liveness_model.py

@@ -2,12 +2,11 @@
 # Copyright (c) 2026 LlamaIndex Inc.
 """Model test for the fan-out/fan-in stream liveness rule.
 
-Drives a synthetic live set ``L(B)`` through the transitions the reducer applies
-— seed at open, same-level emission, collect delivery, fan-out placeholder, and
-close-on-empty — directly against the pure helpers, so the core accounting is
-verifiable without spinning a full workflow. The behavioral end-to-end coverage
-lives in ``test_stream_lineage_regressions.py``; this pins the invariant itself:
-**a stream closes exactly when its live work set empties, never before.**
+Drives a synthetic stream through the transitions the reducer applies: seed at
+open, same-scope emission, collect delivery, nested fan-out, and close-on-empty.
+End-to-end coverage lives in
+``test_collection_stream_regressions.py``; this pins the invariant itself:
+**a stream closes exactly when its open work set empties, never before.**
 """
 
 from __future__ import annotations
@@ -99,7 +98,7 @@ def test_close_fires_exactly_when_live_empties() -> None:
 def test_full_two_collect_stream_drains_to_close() -> None:
     """Walk #1's accounting: 3 Tasks, each work emits Done to two collects.
 
-    Seed L(B) = sum of accepting-step counts per member. Each work resolves
+    Seed open work = sum of accepting-step counts per member. Each work resolves
     (-1 + 2 successors); each of the 6 collect deliveries resolves (-1). The
     stream closes precisely on the last delivery, never earlier.
     """

packages/llama-index-workflows/tests/test_collection_stream_regressions.py

@@ -110,7 +110,7 @@ async def test_event_routed_to_step_and_join_keeps_full_stream() -> None:
 # ---------------------------------------------------------------------------
 
 
-async def test_retried_stream_member_keeps_lineage() -> None:
+async def test_retried_stream_member_keeps_scope() -> None:
     """A member that fails once and succeeds on retry still closes the stream."""
     attempts = {"n2": 0}
 
@@ -164,8 +164,8 @@ async def test_catch_error_recovery_closes_stream() -> None:
 
 
 # ---------------------------------------------------------------------------
-# Closed-collection collects use the normal worker capacity path. Overlapping closes
-# must queue instead of aliasing in-progress state.
+# Collection releases use the normal worker capacity path. Overlapping releases
+# must queue distinct payloads instead of aliasing in-progress state.
 # ---------------------------------------------------------------------------
 
 
@@ -282,7 +282,7 @@ async def test_nested_all_inner_dropped_terminates() -> None:
 
 # ---------------------------------------------------------------------------
 # Persistence: a snapshot taken mid-fan-out can resume without losing stream ids
-# or in-progress member lineage.
+# or in-progress member scope.
 # ---------------------------------------------------------------------------
 
 
@@ -457,7 +457,7 @@ async def test_send_event_extra_member_inside_stream_is_not_joined() -> None:
     assert result == [0, 1, 2], result
 
 
-async def test_send_event_only_inside_collection_collect_rejected() -> None:
+async def test_send_event_only_inside_collection_param_rejected() -> None:
     """A list[E] collect needs a returned-list producer binding."""
 
     class WF(Workflow):

packages/llama-index-workflows/tests/test_collection_streams.py

@@ -226,7 +226,7 @@ async def _run_recording_ticks(
     The in-memory runtime records every tick it reduces via ``on_tick`` and
     exposes them through the run adapter's ``replay()``. That recorded stream is
     exactly what persistence stores, so re-feeding it through
-    ``replay_ticks_stream`` must rebuild identical stream lineage.
+    ``replay_ticks_stream`` must rebuild identical stream scopes.
     """
     handler = wf.run()
     async for _ in handler.stream_events():

packages/llama-index-workflows/tests/test_heterogeneous_fan_in.py

@@ -104,7 +104,7 @@ def test_union_collect_param_rejected() -> None:
             return StopEvent(result="x")
 
 
-def test_list_event_param_accepted_as_collection_collect() -> None:
+def test_list_event_param_accepted_as_collection_param() -> None:
     """A single ``list[E]`` parameter is a collection-collect step."""
 
     class _ListWorkflow(Workflow):
@@ -115,9 +115,9 @@ def test_list_event_param_accepted_as_collection_collect() -> None:
         return StopEvent(result="x")
 
     cfg = collect._step_config
-    assert cfg.collection_collect_param is not None
-    assert cfg.collection_collect_param[0] == "events"
-    assert cfg.collection_collect_param[1] == (Header,)
+    assert cfg.collection_param is not None
+    assert cfg.collection_param[0] == "events"
+    assert cfg.collection_param[1] == (Header,)
     # The step routes on the element event type.
     assert Header in cfg.accepted_events
 
@@ -133,8 +133,8 @@ def test_list_union_event_param_accepted_as_flat_stream() -> None:
         return StopEvent(result="x")
 
     cfg = collect._step_config
-    assert cfg.collection_collect_param is not None
-    assert cfg.collection_collect_param[1] == (Header, Body)
+    assert cfg.collection_param is not None
+    assert cfg.collection_param[1] == (Header, Body)
     # Both member types route to the step.
     assert Header in cfg.accepted_events
     assert Body in cfg.accepted_events

packages/llama-index-workflows/tests/test_utils.py

@@ -324,16 +324,15 @@ def test_validate_step_signature_accepts_list_return() -> None:
 
 
 def test_validate_step_signature_rejects_stream_param_with_other_events() -> None:
-    # A list[E] collection collect parameter cannot be combined with additional event
-    # params.
+    # A list[E] collection parameter cannot be combined with additional event params.
     spec = StepSignatureSpec(
         accepted_events={"a": [StartEvent], "b": [StopEvent]},
         return_types=[StopEvent],
         context_parameter=None,
         context_state_type=None,
         resources=[],
-        collection_collect_param=("a", (StartEvent,)),
-        collection_collect=None,
+        collection_param=("a", (StartEvent,)),
+        collection_policy=None,
         is_fan_out=False,
     )
     with pytest.raises(WorkflowValidationError, match="cannot be combined with other"):