Feat/streaming preemption uvloop

[blasscoc]

Link to Issue or Description of Change

1. Link to an existing issue:

Closes:
Related:

2. Description of the change:

Problem:

ADK workflows currently advance only after an LlmAgent produces its final,
non-partial event. For routing and classification tasks, the model may produce
the actionable decision early in the streamed response, but the workflow still
waits for the remaining generation to finish.

This adds unnecessary latency and output-token usage.

Separately, network-bound ADK workloads can benefit from uvloop, but users
currently need to configure the asyncio event-loop policy themselves.

Solution:

This change adds two opt-in capabilities:

google.adk.enable_uvloop()

An idempotent helper that installs the uvloop event-loop policy before the
first Runner.run(), asyncio.run(), or other event-loop initialization.

The synchronous Runner.run() path can also enable it through:

ADK_UVLOOP=1

uvloop remains an optional dependency installed with:

pip install "google-adk[uvloop]"

Existing asyncio behavior is unchanged unless explicitly enabled. uvloop
is not supported on Windows.

StreamingRouterNode

A workflow node that runs a wrapped agent in SSE streaming mode and forwards
each streamed delta to a caller-supplied monitor.

When the monitor returns a StreamDecision, the node immediately commits
the route and output. By default, it then closes the model stream so the
scheduler can advance without waiting for the remainder of the response.

Stream closure propagates through the asynchronous-generator cleanup chain
and cooperatively cancels the in-flight model call where supported by the
provider.

The change also includes:

Unit tests covering routing, streaming, cleanup, cancellation, and uvloop
configuration.
Runnable streaming_route and fan_out_preempt samples.
A workflow guide.
An optional real-model arXiv benchmark.
An optional real-model Tesla Form 10-K benchmark using SEC EDGAR.

In a measured Vertex AI benchmark using gemini-3.5-flash-lite to classify five
complete arXiv papers in parallel, early stream preemption produced identical
classifications while using approximately 28 times fewer output tokens and
finishing approximately 3.5 times faster.

Preemption avoids unnecessary output generation. It does not avoid processing
the input context or its initial prefill cost.

Testing Plan

Please describe the tests that you ran to verify your changes. This is required
for all PRs that are not small documentation or typo fixes.

Unit Tests:

I have added or updated unit tests for my change.

All unit tests pass locally.

Unit-test coverage includes:

Installing the uvloop event-loop policy.
Repeated calls to enable_uvloop().
Behavior when the optional uvloop dependency is missing.
Enabling uvloop through ADK_UVLOOP=1.
Preserving the default asyncio behavior when not enabled.
Passing streamed deltas to the routing monitor.
Continuing until the monitor returns a decision.
Committing the first StreamDecision.
Propagating route and output values.
Closing the upstream stream after an early decision.
Opting out of early stream closure.
Normal completion without an early decision.
Monitor and wrapped-agent exceptions.
Consumer cancellation and asynchronous-generator cleanup.
Fan-out branches preempting independently.
Preventing duplicate commits.

Relevant tests:

pytest

Pytest result:

Full local test suite:

tox

Result:

Formatting and lint checks:

pre-commit run --all-files

Result:

Manual End-to-End (E2E) Tests:

uvloop

Install the optional dependency:

pip install -e ".[uvloop]"

Run an ADK application with explicit activation:

import google.adk

google.adk.enable_uvloop()

Construct and run the ADK Runner.

Run the synchronous runner using environment-based activation:

ADK_UVLOOP=1 python .py

Verify that:

The application runs successfully.
The uvloop policy is installed when explicitly enabled.
Repeated activation does not fail.
The default asyncio policy remains unchanged when uvloop is not enabled.
Missing optional dependencies produce an actionable error.

Observed output:

Streaming routing sample

Run:

python

Verify that:

Partial SSE events reach the monitor.
The route is committed as soon as the decision is complete.
The next workflow node starts before the model would otherwise finish its
response.
The upstream stream closes cleanly.

Observed output:

Fan-out preemption sample

Run:

python

Verify that:

Branches stream independently.
Each branch commits when its own decision becomes available.
Closing one branch does not cancel sibling branches.
All expected branch results are returned.

Observed output:

Optional arXiv real-model benchmark

Prerequisites:

Vertex AI credentials.
Access to the configured Gemini model.
Network access to retrieve the source documents.

Run:

pytest -v -s

The benchmark compares:

Streaming each response to completion.
Closing each stream as soon as the verdict is detected.

It reports:

Classification results.
Wall-clock duration.
Input-token usage.
Output-token usage.
Estimated cost under the documented assumptions.

Observed result:

Optional Tesla Form 10-K benchmark

Prerequisites:

Vertex AI credentials.
Access to the configured Gemini model.
SEC EDGAR network access.
A descriptive SEC User-Agent.

Run:

pytest -v -s

The test retrieves the latest available Tesla Form 10-K, submits the complete
filing in a single streaming request, and compares full generation with
preemption immediately after the VERDICT line.

The test skips when model credentials or SEC network access are unavailable.

Observed result:

Checklist

I have read the CONTRIBUTING.md document.

I have performed a self-review of my own code.

I have commented my code, particularly in hard-to-understand areas.

I have added tests that prove my fix is effective or that my feature works.

New and existing unit tests pass locally with my changes.

I have manually tested my changes end-to-end.

Any dependent changes have been merged and published in downstream modules.

Additional context

Stream preemption reduces output generation after the required decision appears.
It does not eliminate input processing or prefill latency.

Cancellation is cooperative and provider-dependent. Closing the local stream
allows ADK to stop consuming output and propagate asynchronous-generator
cleanup, but a provider may not always terminate remote generation
instantaneously.

The benchmark results are workload-specific and should not be interpreted as
universal performance guarantees.

uvloop is optional, process-wide, and must be enabled before the application
creates its event loop. It is unavailable on Windows.