From 819e3f8547b7602855874ea9454f7d0088ecab1d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 29 May 2026 09:48:07 +0000 Subject: [PATCH] Fix #5972: Allow multi-source OR listeners to re-fire in cyclic flows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a router emits a signal that should re-trigger an or_() listener that has already fired, the listener was permanently blocked by _fired_or_listeners. This created a catch-22: the clearing logic in _execute_single_listener (for cyclic re-execution) could never be reached because _find_triggered_methods blocked the listener first. Fix: Before processing router results as triggers, reset _fired_or_listeners entries for any OR listeners whose conditions match the router output signals. This preserves concurrent start-method protection (original trigger_method) while unblocking cyclic re-triggering through routers. Co-Authored-By: João --- lib/crewai/src/crewai/flow/flow.py | 35 ++++++++++ lib/crewai/tests/test_flow.py | 102 +++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) diff --git a/lib/crewai/src/crewai/flow/flow.py b/lib/crewai/src/crewai/flow/flow.py index 1ac8b9fde..a940798db 100644 --- a/lib/crewai/src/crewai/flow/flow.py +++ b/lib/crewai/src/crewai/flow/flow.py @@ -1271,6 +1271,34 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): with self._or_listeners_lock: self._fired_or_listeners.discard(listener_name) + def _reset_or_listeners_for_router_results( + self, router_results: list[FlowMethodName] + ) -> None: + """Reset fired OR listeners that could be triggered by router results. + + This enables cyclic flow re-triggering when a router emits a signal + that should re-trigger an OR listener that has already fired in the + current execution. Without this, _fired_or_listeners permanently blocks + the listener and the clearing logic in _execute_single_listener (which + handles cyclic re-execution) can never be reached. + + Only clears listeners matching router results, preserving the concurrent + start-method protection for the original trigger_method. + """ + with self._or_listeners_lock: + for listener_name in list(self._fired_or_listeners): + condition_data = self._listeners.get(listener_name) + if condition_data is None: + continue + if is_simple_flow_condition(condition_data): + _, methods = condition_data + if any(r in methods for r in router_results): + self._fired_or_listeners.discard(listener_name) + elif is_flow_condition_dict(condition_data): + all_methods = _extract_all_methods_recursive(condition_data) + if any(r in all_methods for r in router_results): + self._fired_or_listeners.discard(listener_name) + def _build_racing_groups(self) -> dict[frozenset[FlowMethodName], FlowMethodName]: """Identify groups of methods that race for the same OR listener. @@ -2847,6 +2875,13 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): # Now execute normal listeners for all router results and the original trigger all_triggers = [trigger_method, *router_results] + # For router results, clear _fired_or_listeners entries for any OR + # listeners that could be triggered by those results. This unblocks + # cyclic flows where a router emits a signal that should re-trigger an + # OR listener that already fired earlier in the same execution. + if router_results: + self._reset_or_listeners_for_router_results(router_results) + for current_trigger in all_triggers: if current_trigger: # Skip None results listeners_triggered = self._find_triggered_methods( diff --git a/lib/crewai/tests/test_flow.py b/lib/crewai/tests/test_flow.py index 56c14b85e..851768e5c 100644 --- a/lib/crewai/tests/test_flow.py +++ b/lib/crewai/tests/test_flow.py @@ -1737,6 +1737,108 @@ def test_cyclic_flow_multiple_or_listeners_fire_every_iteration(): ) +def test_or_listener_retriggers_via_different_router_signals(): + """Test that an or_() listener re-fires when a router emits different signals. + + Regression test for issue #5972: @listen(or_(A, B, C)) multi-source OR + listener only fires once, blocking cyclic flow re-triggering. + + Pattern: start → router emits "SignalA" → handler (or_ listener) fires → + handler's router emits "SignalB" → handler should fire AGAIN (but was + blocked by _fired_or_listeners). + """ + execution_order = [] + + class OrRetriggerFlow(Flow): + iteration = 0 + max_iterations = 3 + + @start() + def begin(self): + execution_order.append("begin") + + @router(begin) + def init_router(self): + execution_order.append("init_router") + return "SignalA" + + @listen(or_("SignalA", "SignalB")) + def handler(self): + self.iteration += 1 + execution_order.append(f"handler_{self.iteration}") + + @router(handler) + def handler_router(self): + execution_order.append(f"handler_router_{self.iteration}") + if self.iteration < self.max_iterations: + return "SignalB" + return "done" + + @listen("done") + def finish(self): + execution_order.append("finish") + + flow = OrRetriggerFlow() + flow.kickoff() + + # handler must fire max_iterations times (once for SignalA, rest for SignalB) + handler_events = [e for e in execution_order if e.startswith("handler_") and not e.startswith("handler_router")] + assert len(handler_events) == 3, ( + f"or_() listener 'handler' should fire every iteration via different " + f"router signals, got {len(handler_events)} fires: {execution_order}" + ) + + assert "finish" in execution_order, ( + f"Flow should have reached 'finish', got: {execution_order}" + ) + + +def test_or_listener_retriggers_via_same_router_signal(): + """Test that an or_() listener re-fires when a router emits the same signal repeatedly. + + Variant of issue #5972: the router always emits the SAME signal that + re-triggers the or_() listener. + """ + execution_order = [] + + class SameSignalFlow(Flow): + iteration = 0 + max_iterations = 3 + + @start() + def begin(self): + execution_order.append("begin") + + @router(begin) + def init_router(self): + return "process" + + @listen(or_("process", "other")) + def handler(self): + self.iteration += 1 + execution_order.append(f"handler_{self.iteration}") + + @router(handler) + def handler_router(self): + if self.iteration < self.max_iterations: + return "process" + return "exit" + + @listen("exit") + def finish(self): + execution_order.append("finish") + + flow = SameSignalFlow() + flow.kickoff() + + handler_events = [e for e in execution_order if e.startswith("handler_") and not e.startswith("handler_router")] + assert len(handler_events) == 3, ( + f"or_() listener 'handler' should re-fire via same router signal, " + f"got {len(handler_events)} fires: {execution_order}" + ) + assert "finish" in execution_order + + def test_cyclic_flow_works_with_persist_and_id_input(): """Cyclic router flows must complete all iterations when persistence is enabled and 'id' is passed in inputs.