From 2acffb4093baff080431ba9b5bd340d4415bb941 Mon Sep 17 00:00:00 2001 From: RachelWanggg Date: Thu, 17 Sep 2026 17:16:39 -0700 Subject: [PATCH 1/2] Python: surface switch-case condition errors instead of routing to default SwitchCaseEdgeGroup wrapped every case predicate in a bare `except Exception` that only logged a warning, so a predicate that raised was treated exactly like one that returned False: the message fell through to the default branch and the workflow completed successfully. A broken routing predicate became silently wrong routing rather than a visible failure. This contradicted `Edge.should_route`, which documents that predicate errors are deliberately allowed to surface "to avoid masking logic bugs", and made the existing `FanOutEdgeRunner` error path (span status EXCEPTION, re-raise) unreachable for switch-case groups, so the failures were missing from telemetry too. The block was marked `# pragma: no cover`. Also align `FanOutEdgeRunner` with `SingleEdgeRunner`, which checks `_can_handle` before evaluating a condition: drop messages no target can handle before running the selection function. Without this, removing the catch would turn an undeliverable message into a raised AttributeError instead of a type-mismatch drop. Fixes #8489 Co-Authored-By: Claude Opus 5 --- .../core/agent_framework/_workflows/_edge.py | 11 +-- .../_workflows/_edge_runner.py | 10 +++ .../packages/core/tests/workflow/test_edge.py | 78 +++++++++++++++++++ 3 files changed, 94 insertions(+), 5 deletions(-) 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..65cdc859814 100644 --- a/python/packages/core/agent_framework/_workflows/_edge_runner.py +++ b/python/packages/core/agent_framework/_workflows/_edge_runner.py @@ -255,6 +255,16 @@ async def send_message( source_span_ids=message.source_span_ids, ) as span: try: + # Drop messages no target can handle before running the selection function, so a + # selection function is only ever asked about a deliverable message. `SingleEdgeRunner` + # applies the same ordering by checking `_can_handle` before `Edge.should_route`. + 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..3d7daa28e76 100644 --- a/python/packages/core/tests/workflow/test_edge.py +++ b/python/packages/core/tests/workflow/test_edge.py @@ -1389,6 +1389,84 @@ 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_skips_selection_for_unhandleable_message() -> 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") From 6950a98534913ffc4cfb4ab2b0e4ac02641eb444 Mon Sep 17 00:00:00 2001 From: RachelWanggg Date: Thu, 17 Sep 2026 22:42:47 -0700 Subject: [PATCH 2/2] Python: drop target-mismatched messages before the fan-out selection function The runner fans every message out to all of the source's edge runners and gathers them with `gather_cancelling_siblings_on_error`. `FanOutEdgeRunner` checked `message.target_id` only after running the selection function, so a message addressed to an executor reached through a different edge still had this group's selection function evaluated against it - and a raise there cancelled the sibling runner that was actually delivering the message. Check `message.target_id` against `_target_map` first and return DROPPED_TARGET_MISMATCH, ahead of the `_can_handle` check, matching `SingleEdgeRunner`, which checks the target and then `_can_handle` before evaluating `Edge.should_route`. Co-Authored-By: Claude Opus 5 --- .../_workflows/_edge_runner.py | 19 ++++++++++-- .../packages/core/tests/workflow/test_edge.py | 31 ++++++++++++++++++- 2 files changed, 46 insertions(+), 4 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_edge_runner.py b/python/packages/core/agent_framework/_workflows/_edge_runner.py index 65cdc859814..7db1af2dbb3 100644 --- a/python/packages/core/agent_framework/_workflows/_edge_runner.py +++ b/python/packages/core/agent_framework/_workflows/_edge_runner.py @@ -255,9 +255,22 @@ async def send_message( source_span_ids=message.source_span_ids, ) as span: try: - # Drop messages no target can handle before running the selection function, so a - # selection function is only ever asked about a deliverable message. `SingleEdgeRunner` - # applies the same ordering by checking `_can_handle` before `Edge.should_route`. + # 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, diff --git a/python/packages/core/tests/workflow/test_edge.py b/python/packages/core/tests/workflow/test_edge.py index 3d7daa28e76..f32f6f9ec70 100644 --- a/python/packages/core/tests/workflow/test_edge.py +++ b/python/packages/core/tests/workflow/test_edge.py @@ -1439,7 +1439,36 @@ def broken_condition(message: MockMessage) -> bool: assert mock_send.call_count == 0 -async def test_fan_out_edge_group_skips_selection_for_unhandleable_message() -> None: +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")