From 1a98bfbe16da4156f3bf6baa3e66cebd8ea4416f Mon Sep 17 00:00:00 2001 From: Adrian Lyjak Date: Sat, 30 May 2026 12:08:38 -0400 Subject: [PATCH] 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 +- .../src/workflows/context/context.py | 28 ---------- .../src/workflows/context/internal_context.py | 4 -- .../src/workflows/runtime/types/plugin.py | 12 ---- .../tests/test_collect_algebra.py | 55 ++----------------- 5 files changed, 6 insertions(+), 95 deletions(-) diff --git a/.changeset/fan-out-fan-in-collect-algebra.md b/.changeset/fan-out-fan-in-collect-algebra.md index f76dcaf5..e60d2675 100644 --- a/.changeset/fan-out-fan-in-collect-algebra.md +++ b/.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. diff --git a/packages/llama-index-workflows/src/workflows/context/context.py b/packages/llama-index-workflows/src/workflows/context/context.py index 05f40429..0dc79a23 100644 --- a/packages/llama-index-workflows/src/workflows/context/context.py +++ b/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. diff --git a/packages/llama-index-workflows/src/workflows/context/internal_context.py b/packages/llama-index-workflows/src/workflows/context/internal_context.py index 2f39ba97..204a19cb 100644 --- a/packages/llama-index-workflows/src/workflows/context/internal_context.py +++ b/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. diff --git a/packages/llama-index-workflows/src/workflows/runtime/types/plugin.py b/packages/llama-index-workflows/src/workflows/runtime/types/plugin.py index 9a80f6fa..ccd820db 100644 --- a/packages/llama-index-workflows/src/workflows/runtime/types/plugin.py +++ b/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. diff --git a/packages/llama-index-workflows/tests/test_collect_algebra.py b/packages/llama-index-workflows/tests/test_collect_algebra.py index 7d0acf95..9bc7b71c 100644 --- a/packages/llama-index-workflows/tests/test_collect_algebra.py +++ b/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) # --------------------------------------------------------------------------- #