Python: wake workflow streaming on iteration completion

[Naveen (1aifanatic)]

Motivation & Context

The workflow runner waits up to 50 ms for another event before noticing that a superstep has finished. For fast executors, that polling delay dominates execution time. This implements the runtime follow-up to the bottleneck identified by Shivani . (@Shivani767) in #3686, without duplicating the proposed benchmark script.

Description & Review Guide

• What are the major changes? Wait for either iteration completion or the next event. Cancel and await outstanding tasks on cancellation or stream closure. Add regression tests for completion without timer delivery, live/tail event ordering on success and failure, and generator-close cleanup. Replace a concurrency test's sleep with explicit gates so the faster runner still exercises overlapping runs.
• What is the impact of these changes? Public APIs are unchanged. Two new regressions fail before the fix and pass afterward. The workflow suite passed 975 tests (4 skipped, 1 expected failure); the final focused runner/workflow run passed 94 tests. Ruff checks/formatting and Pyright checks of changed source and tests pass. Existing experimental/deprecation warnings remain. The entire cross-package test suite was not run locally.
• What do you want reviewers to focus on? Event/task ownership when iteration completion races event receipt, and cleanup when a consumer closes the stream. A local forward-only chain measurement on Windows 11/Python 3.13.11 (3 warmups, 10 measured runs, graph construction excluded) changed median times for 1/4/16 nodes from 117.287/288.462/994.765 ms to 0.360/0.892/3.016 ms. These numbers measure framework overhead without model/network calls and are not end-to-end agent latency claims.

Related Issue

Contributes to #3686. This addresses the reported polling bottleneck; it does not close the broader benchmark/performance-testing issue. No competing open runtime PR for this issue was found during preflight. Scope was proposed in the issue before implementation.

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 is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• 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.

[Copilot]

🟢 Approval recommended

The implementation handles completion and event races correctly, cleans up owned tasks, and includes focused regression coverage.

Pull request overview

Eliminates workflow superstep polling latency by waking on iteration completion or event arrival, with robust task cleanup.

Changes:

• Replaces timer polling with concurrent task waiting.
• Adds event-ordering, failure, completion, and cleanup regressions.
• Makes concurrency testing deterministic with explicit gates.

File summaries

File	Description
python/packages/core/agent_framework/_workflows/_runner.py	Implements completion-driven streaming and task cleanup.
python/packages/core/tests/workflow/test_runner.py	Covers wake-up, ordering, failures, and stream closure.
python/packages/core/tests/workflow/test_workflow.py	Removes timing dependence from concurrency testing.

Review details

• Files reviewed: 3/3 changed files
• Comments generated: 0
• Review effort level: Balanced

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.