Add fix for double send when waiter event and accepted event match (#351)

.changeset/rich-lizards-sing.md

@@ -0,0 +1,5 @@
+---
+"llama-index-workflows": patch
+---
+
+Add fix for double send when waiter event and accepted event match

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

@@ -834,7 +834,36 @@ def _process_add_event_tick(
     handled = False
     if isinstance(tick.event, StartEvent):
         state.is_running = True
+
+    # First, check if the event resolves any waiters. Track which steps were
+    # woken via waiter resolution so we don't also route the event to them
+    # as a normal accepted event (which would cause duplicate processing).
+    waiter_resolved_steps: set[str] = set()
     for step_name, step_config in state.config.steps.items():
+        wait_conditions = state.workers[step_name].collected_waiters
+        for wait_condition in wait_conditions:
+            is_match = type(tick.event) is wait_condition.waiting_for_event
+            is_match = is_match and all(
+                getattr(tick.event, k, None) == v
+                for k, v in wait_condition.requirements.items()
+            )
+            if is_match:
+                handled = True
+                waiter_resolved_steps.add(step_name)
+                wait_condition.resolved_event = tick.event
+                subcommands = _add_or_enqueue_event(
+                    EventAttempt(event=wait_condition.event),
+                    step_name,
+                    state.workers[step_name],
+                    now_seconds,
+                )
+                commands.extend(subcommands)
+
+    # Then route to accepting steps, skipping any that were already woken
+    # via waiter resolution above.
+    for step_name, step_config in state.config.steps.items():
+        if step_name in waiter_resolved_steps:
+            continue
         is_accepted = type(tick.event) in step_config.accepted_events
         if is_accepted and (tick.step_name is None or tick.step_name == step_name):
             handled = True
@@ -849,27 +878,6 @@ def _process_add_event_tick(
                 now_seconds,
             )
             commands.extend(subcommands)
-
-    # separately, check if the event is a waiting event, and if so, update the waiting event state
-    # and set the resolved event. Add the original event as a command
-    for step_name, step_config in state.config.steps.items():
-        wait_conditions = state.workers[step_name].collected_waiters
-        for wait_condition in wait_conditions:
-            is_match = type(tick.event) is wait_condition.waiting_for_event
-            is_match = is_match and all(
-                getattr(tick.event, k, None) == v
-                for k, v in wait_condition.requirements.items()
-            )
-            if is_match:
-                handled = True
-                wait_condition.resolved_event = tick.event
-                subcommands = _add_or_enqueue_event(
-                    EventAttempt(event=wait_condition.event),
-                    step_name,
-                    state.workers[step_name],
-                    now_seconds,
-                )
-                commands.extend(subcommands)
     if not handled:
         # InputRequiredEvent subclasses are intentionally designed to be handled
         # externally by human consumers, not by workflow steps. Don't emit

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

@@ -1011,3 +1011,72 @@ async def test_simultaneous_retries_with_same_delay(
     assert result.result == "both_succeeded"
     assert wf.step_a_attempts == 2
     assert wf.step_b_attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_external_event_not_double_routed_when_waiter_exists(
+    test_plugin: MockRunAdapter,
+) -> None:
+    """Regression test: an external event that resolves a wait_for_event waiter
+    should NOT also be routed to another step that accepts the same event type.
+
+    Before the fix, the accepting step would run twice — once from normal
+    routing and once from the waiter waking up and re-emitting the event.
+    """
+
+    class ExternalInput(Event):
+        value: str
+
+    step_run_count = 0
+
+    class DoubleRouteWorkflow(Workflow):
+        @step
+        async def kickoff(self, ev: StartEvent) -> ExternalInput:
+            return ExternalInput(value="init")
+
+        @step
+        async def handle_input(self, ev: ExternalInput, ctx: Context) -> StopEvent:
+            # This step accepts ExternalInput AND waits for ExternalInput.
+            # wait_for_event works by raising an exception on first call,
+            # then the control loop re-runs the step after the waiter resolves.
+            # So this step runs twice normally: once to register the waiter,
+            # once after resolution. The bug caused a THIRD run via normal
+            # event routing of the external event to this step.
+            nonlocal step_run_count
+            step_run_count += 1
+            result = await ctx.wait_for_event(
+                ExternalInput,
+                waiter_event=InputRequiredEvent(),
+            )
+            return StopEvent(result=f"got_{result.value}")
+
+    wf = DoubleRouteWorkflow(timeout=2.0)
+    task = asyncio.create_task(
+        run_control_loop(
+            workflow=wf,
+            start_event=StartEvent(),
+            test_runtime=test_plugin,
+        )
+    )
+
+    # Wait for the waiter to be registered
+    async for event in test_plugin.stream_published_events():
+        if isinstance(event, InputRequiredEvent):
+            break
+
+    # Send the external event — this resolves the waiter on handle_input.
+    # Without the fix, handle_input would ALSO get ExternalInput via normal
+    # accepted_events routing, causing a second execution.
+    await test_plugin.send_event(TickAddEvent(event=ExternalInput(value="hello")))
+
+    result = await asyncio.wait_for(task, timeout=2.0)
+    assert isinstance(result, StopEvent)
+    assert result.result == "got_hello", (
+        f"Expected waiter resolution result, got '{result.result}'"
+    )
+    # Step runs twice: once to register the waiter (raises WaitingForEvent),
+    # once after waiter resolution (returns the result). Without the fix,
+    # it would run a third time from the external event being routed directly.
+    assert step_run_count == 2, (
+        f"handle_input should run exactly twice, but ran {step_run_count} times"
+    )