diff --git a/python/packages/core/agent_framework/_workflows/_edge.py b/python/packages/core/agent_framework/_workflows/_edge.py index 79ca09d990e..5a65b0563bf 100644 --- a/python/packages/core/agent_framework/_workflows/_edge.py +++ b/python/packages/core/agent_framework/_workflows/_edge.py @@ -863,11 +863,12 @@ def selection_func(message: Any, targets: list[str]) -> list[str]: for case in cases: if isinstance(case, SwitchCaseEdgeGroupDefault): return [case.target_id] - try: - if case.condition(message): - return [case.target_id] - except Exception as exc: # pragma: no cover - defensive logging - logger.warning("Error evaluating condition for case %s: %s", case.target_id, exc) + # Errors raised by a case predicate deliberately surface to the caller, matching + # `Edge.should_route`. Swallowing them would route the message to the default + # branch, turning a broken predicate into silent misrouting rather than a + # visible failure. `FanOutEdgeRunner` records the error on the edge-group span. + if case.condition(message): + return [case.target_id] raise RuntimeError("No matching case found in SwitchCaseEdgeGroup") target_ids = [case.target_id for case in cases] diff --git a/python/packages/core/agent_framework/_workflows/_edge_runner.py b/python/packages/core/agent_framework/_workflows/_edge_runner.py index e3bc14e64ab..7db1af2dbb3 100644 --- a/python/packages/core/agent_framework/_workflows/_edge_runner.py +++ b/python/packages/core/agent_framework/_workflows/_edge_runner.py @@ -255,6 +255,29 @@ async def send_message( source_span_ids=message.source_span_ids, ) as span: try: + # Drop messages this group cannot deliver before running the selection function, so + # a selection function is only ever asked about a message this group could route. + # `SingleEdgeRunner` applies the same ordering, checking the target and + # `_can_handle` before `Edge.should_route`. + # + # The target check matters beyond saving work: the runner fans every message out to + # all of the source's edge runners and cancels sibling deliveries when one raises, + # so evaluating a selection function for a message addressed to an executor outside + # this group could cancel that message's real delivery. + if message.target_id and message.target_id not in self._target_map: + span.set_attributes({ + OtelAttr.EDGE_GROUP_DELIVERED: False, + OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TARGET_MISMATCH.value, + }) + return False + + if not any(self._can_handle(target_id, message) for target_id in self._target_ids): + span.set_attributes({ + OtelAttr.EDGE_GROUP_DELIVERED: False, + OtelAttr.EDGE_GROUP_DELIVERY_STATUS: EdgeGroupDeliveryStatus.DROPPED_TYPE_MISMATCH.value, + }) + return False + selection_results = ( self._selection_func(message.data, self._target_ids) if self._selection_func else self._target_ids ) diff --git a/python/packages/core/tests/workflow/test_edge.py b/python/packages/core/tests/workflow/test_edge.py index 6d60abeb8a4..f32f6f9ec70 100644 --- a/python/packages/core/tests/workflow/test_edge.py +++ b/python/packages/core/tests/workflow/test_edge.py @@ -1389,6 +1389,113 @@ async def test_switch_case_edge_group_send_message() -> None: assert mock_send.call_count == 1 +def test_switch_case_edge_group_condition_error_propagates() -> None: + """A failing case condition surfaces instead of falling through to the default.""" + source = MockExecutor(id="source_executor") + target1 = MockExecutor(id="target_executor_1") + target2 = MockExecutor(id="target_executor_2") + + def broken_condition(message: MockMessage) -> bool: + raise ValueError("condition is broken") + + edge_group = SwitchCaseEdgeGroup( + source_id=source.id, + cases=[ + SwitchCaseEdgeGroupCase(condition=broken_condition, target_id=target1.id), + SwitchCaseEdgeGroupDefault(target_id=target2.id), + ], + ) + + assert edge_group._selection_func is not None # type: ignore[reportPrivateUsage] + with pytest.raises(ValueError, match="condition is broken"): + edge_group._selection_func(MockMessage(data=1), [target1.id, target2.id]) # type: ignore[reportPrivateUsage] + + +async def test_switch_case_edge_group_send_message_condition_error_propagates() -> None: + """A failing case condition surfaces from send_message and delivers nothing.""" + source = MockExecutor(id="source_executor") + target1 = MockExecutor(id="target_executor_1") + target2 = MockExecutor(id="target_executor_2") + + def broken_condition(message: MockMessage) -> bool: + raise ValueError("condition is broken") + + edge_group = SwitchCaseEdgeGroup( + source_id=source.id, + cases=[ + SwitchCaseEdgeGroupCase(condition=broken_condition, target_id=target1.id), + SwitchCaseEdgeGroupDefault(target_id=target2.id), + ], + ) + executors: dict[str, Executor] = {source.id: source, target1.id: target1, target2.id: target2} + edge_runner = create_edge_runner(edge_group, executors) + + message = WorkflowMessage(data=MockMessage(data=1), source_id=source.id) + + with patch("agent_framework._workflows._edge_runner.EdgeRunner._execute_on_target") as mock_send: + with pytest.raises(ValueError, match="condition is broken"): + await edge_runner.send_message(message, State(), InProcRunnerContext()) + + assert mock_send.call_count == 0 + + +async def test_fan_out_edge_group_send_message_with_target_outside_group() -> None: + """A message addressed to an executor outside the group never reaches the selection function.""" + source = MockExecutor(id="source_executor") + target1 = MockExecutor(id="target_executor_1") + target2 = MockExecutor(id="target_executor_2") + + def broken_condition(message: MockMessage) -> bool: + raise ValueError("condition is broken") + + edge_group = SwitchCaseEdgeGroup( + source_id=source.id, + cases=[ + SwitchCaseEdgeGroupCase(condition=broken_condition, target_id=target1.id), + SwitchCaseEdgeGroupDefault(target_id=target2.id), + ], + ) + executors: dict[str, Executor] = {source.id: source, target1.id: target1, target2.id: target2} + edge_runner = create_edge_runner(edge_group, executors) + + # RunnerImpl fans every message out to all of the source's edge runners, so a message + # addressed to an executor reached through a different edge still arrives here. It must be + # dropped as a target mismatch, not evaluated -- raising would cancel its real delivery. + message = WorkflowMessage(data=MockMessage(data=1), source_id=source.id, target_id="unrelated_executor") + + success = await edge_runner.send_message(message, State(), InProcRunnerContext()) + + assert success is False + + +async def test_fan_out_edge_group_send_message_with_unhandleable_data() -> None: + """A message no target can handle is dropped before the selection function runs.""" + source = MockExecutor(id="source_executor") + target1 = MockExecutor(id="target_executor_1") + target2 = MockExecutor(id="target_executor_2") + + calls: list[Any] = [] + + def selection_func(message: Any, targets: list[str]) -> list[str]: + calls.append(message) + return targets + + edge_group = FanOutEdgeGroup( + source_id=source.id, + target_ids=[target1.id, target2.id], + selection_func=selection_func, + ) + executors: dict[str, Executor] = {source.id: source, target1.id: target1, target2.id: target2} + edge_runner = create_edge_runner(edge_group, executors) + + message = WorkflowMessage(data="invalid_data", source_id=source.id) + + success = await edge_runner.send_message(message, State(), InProcRunnerContext()) + + assert success is False + assert calls == [] + + async def test_switch_case_edge_group_send_message_with_invalid_target() -> None: """Test sending a message through a switch case edge group with an invalid target.""" source = MockExecutor(id="source_executor")