Python: Surface switch-case condition errors instead of routing to the default branch

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]

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
                 )

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")