[Hardik K (hardikkaurani)]

Motivation & Context

In agent-framework (Python), standard edge conditions support asynchronous predicates via EdgeCondition: TypeAlias = Callable[[Any], bool | Awaitable[bool]], and Edge.should_route awaits the condition when it returns an awaitable.

However, switch-case edge groups (Case and SwitchCaseEdgeGroupCase) were restricted to synchronous predicates (Callable[[Any], bool]). In practice, when an asynchronous predicate was passed to Case, calling case.condition(message) produced a coroutine object. Because coroutine objects are truthy in Python, the first case would always immediately match regardless of input, leaving an unawaited coroutine and emitting RuntimeWarning: coroutine '...' was never awaited. Furthermore, FanOutEdgeRunner.send_message invoked self._selection_func(...) synchronously without awaiting awaitable selection results.

Routing in workflows frequently requires async operations (e.g. LLM classification, external API checks, or async database lookups). This PR enables async conditions in switch-case edge groups while maintaining full backward compatibility with synchronous conditions.

Description & Review Guide

• What are the major changes?

  1. Updated Case.condition and SwitchCaseEdgeGroupCase.condition typing from Callable[[Any], bool] to EdgeCondition (Callable[[Any], bool | Awaitable[bool]]).
  2. Updated FanOutEdgeGroup._selection_func and selection_func property to accept Callable[[Any, list[str]], list[str] | Awaitable[list[str]]] | None.
  3. Updated SwitchCaseEdgeGroup.__init__'s internal selection_func to evaluate cases sequentially: if a condition returns an awaitable, it seamlessly evaluates and awaits the result (and subsequent cases if necessary) before returning the target executor ID.
  4. Updated FanOutEdgeRunner.send_message to check if selection_results is awaitable (inspect.isawaitable(...)) and await it accordingly.
  5. Added comprehensive unit tests in packages/core/tests/workflow/test_edge.py covering async condition matching, async fallthrough to default, condition evaluation ordering with mixed sync/async predicates, async condition error propagation, and async fan-out selectors.
  6. Added workflow integration test in packages/core/tests/workflow/test_workflow.py verifying end-to-end switch-case routing with async conditions.

• What is the impact of these changes?

  • Workflow authors can now pass async functions (async def) or callables returning Awaitable[bool] to Case(condition=...) without errors or warnings.
  • Zero breaking changes: existing synchronous predicates continue to execute synchronously with identical behavior.

• What do you want reviewers to focus on?

  • The sequential evaluation in SwitchCaseEdgeGroup's selection function ensuring "first match wins" semantics are preserved whether conditions are sync or async.

Related Issue

Fixes #9642

AI Assistance

• No material AI assistance was used.
• This is an AI-assisted contribution. I reviewed, understood, and verified all submitted content and accept responsibility for it.

AI assistance details: Implementation of async condition support in switch-case edge groups and runners, comprehensive unit & workflow tests, and PR description.

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR links to an agreed issue with no competing open PR, or the Related Issue section documents a trivial-change or repository-automation exception.
• This is not a breaking change. If it is a breaking change, add the breaking change label (or add "[BREAKING]" to the title prefix, before or after any language prefix) - a workflow keeps the label and title prefix in sync automatically.

[Hardik K (hardikkaurani)]

Hi Jose Alvarez (@jpalvarezl),

I've opened this PR to resolve issue #9642 ("Support async conditions in switch-case edge groups").

Summary of Changes:

1. Type Definition Alignment: Updated Case.condition and SwitchCaseEdgeGroupCase to EdgeCondition (Callable[[Any], bool | Awaitable[bool]]), matching standard edge conditions.
2. Async Selection in Switch-Case: In SwitchCaseEdgeGroup, updated the selection function so that async predicates are properly awaited in order while preserving short-circuiting / first-match-wins semantics and raising if no case matches.
3. Runner Support: Updated FanOutEdgeRunner.send_message to await selection_func results when awaitable, properly resolving targets.
4. Testing & Validation:
   • Added 5 unit tests in test_edge.py covering async match, fallthrough to default, mixed sync/async precedence, and error propagation.
   • Added end-to-end integration test in test_workflow.py.
   • Verified with full test suite (1443 passed), ruff check, ruff format, and pyright (0 errors, 0 warnings).

Whenever you have a moment, please feel free to review. Thank you!

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has insufficient remaining quota.

[Bicheng (Kenneth) (BichengWang)]

Reviewed 4935175c1796688b7e60786caea2c9576571550e offline on macOS / Python 3.13.16. The edge, workflow, and serialization suites pass: 145 tests. Restoring the three implementation files to merge base c01db29b761196665f7f9eafcd1ec9b6c1d196ef makes five of the six added regression tests fail; the matching-first-case control still passes.

Eight additional workflow probes pass at this head: true/false results from futures and custom awaitables, cancellation without evaluating the next case, an error in a later async predicate, rejection of invalid async fan-out selections, and a synchronous match skipping the later async predicate. Changed-file Ruff lint and formatting checks also pass.

I found no blocking code issue in the diff. This is focused verification; I did not run the full workspace suite, type checkers, distribution builds, or live providers. Issue #9642 currently has no maintainer agreement comment or ready-for-implementation label, so the contribution approval prerequisite still needs maintainer confirmation.

[Hardik K (hardikkaurani)]

Hi Bicheng (Kenneth) (@BichengWang),

Thank you for the thorough review and for running the extensive offline verification probes across Python 3.13!

In response to your feedback, I have codified those exact probe scenarios into the automated test suite in commit 8021fb5 (python/packages/core/tests/workflow/test_edge.py):

1. Futures & Custom Awaitables (test_switch_case_edge_group_async_condition_future_and_custom_awaitable):
   • Verifies that asyncio.Future and custom awaitable objects implementing __await__ are properly awaited for boolean truthiness, correctly routing matches and falling through to default when non-matching.
2. Short-Circuiting Optimization (test_switch_case_edge_group_sync_match_skips_later_async_predicate):
   • Verifies that when an earlier synchronous condition matches, subsequent asynchronous predicates are never evaluated or awaited.
3. Cancellation Semantics (test_switch_case_edge_group_async_condition_cancellation):
   • Verifies that task cancellation during an async predicate cleanly raises asyncio.CancelledError without invoking subsequent cases.
4. Invalid Fan-Out Selection Validation (test_fan_out_edge_group_with_async_selection_func_invalid_result_raises):
   • Verifies that an async selection function returning invalid target IDs raises a RuntimeError, matching synchronous validation.

Validation Summary:

• Unit & Integration Tests: All 66 tests in test_edge.py and 64 tests in test_workflow.py pass cleanly.
• Lint & Formatting: ruff check packages/core and ruff format --check packages/core pass with 0 errors.
• Type Checking: pyright packages/core reports 0 errors, 0 warnings, 0 informations.

Regarding the issue prerequisite: issue #9642 is currently assigned to Eduard van Valkenburg (@eavanvalkenburg). Pinging Eduard van Valkenburg (@eavanvalkenburg) and Jose Alvarez (@jpalvarezl) for maintainer confirmation and review whenever convenient. Thank you!