Pull replayed_events(): unwired stub that misleads on replay

ctx.replayed_events() returned [] on every runtime, including DBOS where
step bodies actually replay (nothing overrides the adapter method). A user
writing replay-dedup against it on a durable runtime gets silent no-dedup --
the exact duplicate-emission bug it advertises preventing, same footgun class
as AtLeast. Removed from Context, InternalContext, and the adapter base.
Reintroduce alongside the durable-runtime work that records prior emissions.

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

@@ -2,4 +2,4 @@
 "llama-index-workflows": minor
 ---
 
-Add the Collect selection algebra for batch fan-in. Annotate a list[E] join parameter with Collect(Take(n)) to release early on the nth arrival instead of waiting for the batch to close; bare list[E] stays Collect(All()). list[A | B] now collects a flat heterogeneous batch. New ctx.replayed_events() accessor exposes prior-run emissions for user-side dedup.
+Add the Collect selection algebra for batch fan-in. Annotate a list[E] join parameter with Collect(Take(n)) to release early on the nth arrival instead of waiting for the batch to close; bare list[E] stays Collect(All()). list[A | B] now collects a flat heterogeneous batch.

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

@@ -620,34 +620,6 @@ class Context(Generic[MODEL_T]):
         """
         self._require_internal(fn="write_event_to_stream").write_event_to_stream(ev)
 
-    def replayed_events(self) -> list[Event]:
-        """Events this step emitted on prior runs of the current execution.
-
-        A fan-out producer (`-> list[E]` / `-> AsyncIterator[E]`) re-runs from the
-        start on recovery, and the framework does not dedupe emissions — it
-        exposes the journal so the step can skip work it already did. "Same event"
-        is a domain concept (id, content hash, position in some stream), so the
-        dedupe policy is yours.
-
-        Returns an empty list on a first run and on runtimes that execute each
-        step body exactly once (the default in-memory runtime). Durable runtimes
-        that re-execute a step on recovery populate it with the prior emissions.
-
-        Examples:
-            ```python
-            @step
-            async def fan_out(self, ctx: Context, ev: StartEvent) -> AsyncIterator[Task]:
-                seen = {e.item_id for e in ctx.replayed_events()}
-                for item in source():
-                    if item.id not in seen:
-                        yield Task(item_id=item.id)
-            ```
-
-        Raises:
-            ContextStateError: If called outside of a running workflow step.
-        """
-        return self._require_internal(fn="replayed_events").replayed_events()
-
     async def _finalize_step(self) -> None:
         """Finalize step execution by awaiting background tasks.
 

packages/llama-index-workflows/src/workflows/context/internal_context.py

@@ -217,10 +217,6 @@ class InternalContext(Generic[MODEL_T]):
         if ev is not None:
             self._execute_task(self._internal_adapter.write_to_event_stream(ev))
 
-    def replayed_events(self) -> list[Event]:
-        """Events emitted by the current step on prior runs (for dedup)."""
-        return list(self._internal_adapter.replayed_events())
-
     def retry_info(self) -> RetryInfo:
         """Snapshot of the currently-executing step's retry state.
 

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

@@ -208,18 +208,6 @@ class InternalRunAdapter(ABC):
         """
         return False
 
-    def replayed_events(self) -> list[Event]:
-        """Events the current step emitted on prior runs of this execution.
-
-        Used by a fan-out producer (``list[E]`` / ``AsyncIterator[E]``) to skip
-        already-emitted work on replay — the framework does not dedupe emissions,
-        it exposes the journal and the user dedupes by domain id. Empty on a first
-        run and on runtimes that never replay step bodies (the in-memory runtime
-        runs each step exactly once); durable runtimes that re-execute a step on
-        recovery override this to return the prior emissions.
-        """
-        return []
-
     async def on_tick(self, tick: WorkflowTick) -> None:
         """
         Called whenever a tick event is processed by the control loop.

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

@@ -5,9 +5,9 @@
 Covers the public ``Collect`` / ``Cardinality`` API, signature inference for
 batch fan-in parameters (bare ``list[E]``, the ``Annotated[..., Collect()]``
 synonym, union flat lists, and the ``Take(n)`` marker), the validation errors
-that keep mode determination legible, ``Take(n)`` runtime release, and
-``ctx.replayed_events()``. ``AtLeast`` is v2 (it's a silent alias of ``Take``
-without v2 re-fire / cancellation), so it is intentionally not exported.
+that keep mode determination legible, and ``Take(n)`` runtime release.
+``AtLeast`` is v2 (it's a silent alias of ``Take`` without v2 re-fire /
+cancellation), so it is intentionally not exported.
 
 The cross-level (``at=``) and provenance (``from_=``) knobs and multi-slot joins
 are declared-but-deferred; their tests assert the clear validation error and
@@ -16,10 +16,10 @@ xfail the end-to-end examples.
 
 from __future__ import annotations
 
-from typing import Annotated, Any, AsyncIterator, Callable
+from typing import Annotated, Any, Callable
 
 import pytest
-from workflows import All, Cardinality, Collect, Context, Take, Workflow, step
+from workflows import All, Cardinality, Collect, Take, Workflow, step
 from workflows.decorators import StepConfig, StepFunction
 from workflows.decorators import step as free_step
 from workflows.errors import WorkflowValidationError
@@ -365,51 +365,6 @@ async def test_annotated_all_runs_like_bare_list() -> None:
     assert result == [0, 1, 2, 3]
 
 
-# --------------------------------------------------------------------------- #
-# ctx.replayed_events()
-# --------------------------------------------------------------------------- #
-
-
-async def test_replayed_events_empty_on_first_run() -> None:
-    """`ctx.replayed_events()` returns an empty list on a fresh run."""
-
-    seen: list[int] = []
-
-    class FanOut(Workflow):
-        @step
-        async def fan_out(self, ctx: Context, ev: StartEvent) -> AsyncIterator[Task]:
-            already = ctx.replayed_events()
-            seen.append(len(already))
-            assert already == []
-            for i in range(3):
-                yield Task(n=i)
-
-        @step
-        async def work(self, ev: Task) -> Done:
-            return Done(n=ev.n)
-
-        @step
-        async def join(self, events: list[Done]) -> StopEvent:
-            return StopEvent(result=len(events))
-
-    result = await _run(FanOut(timeout=10))
-    assert result == 3
-    assert seen == [0]
-
-
-def test_replayed_events_outside_step_raises() -> None:
-    from workflows.errors import ContextStateError
-
-    class Trivial(Workflow):
-        @step
-        async def go(self, ev: StartEvent) -> StopEvent:
-            return StopEvent(result="x")
-
-    ctx: Context = Context(Trivial())
-    with pytest.raises(ContextStateError):
-        ctx.replayed_events()
-
-
 # --------------------------------------------------------------------------- #
 # Deferred: cross-level scope, provenance, multi-slot (declared, not yet run)
 # --------------------------------------------------------------------------- #