Feat/streaming preemption uvloop

[blasscoc]

Problem: Workflow nodes today wait for each LLM turn to fully complete before the graph advances, and fan-out always waits for every branch. For search/retrieval-style work this wastes both output tokens (the model keeps generating a long answer we've already decided we don't need) and wall-clock time (downstream work can't start until generation ends, and losing branches keep running). Agents used in these pipelines often "waffle" or aren't perfectly aligned, but they're still perfectly usable if we can act on the useful part of the stream and drop the rest.

Solution: Three opt-in workflow primitives (plus one runtime switch) that let a graph react to a stream as it arrives. None change existing behavior unless you use them.

StreamingRouterNode — mid-stream preemption. A monitor callback inspects the SSE token stream and can commit an output and cancel the rest of generation the moment the answer is known (e.g. stop after a VERDICT: line instead of generating a 300-word summary you'll discard). Cancellation propagates cooperatively via aclosing.
SpeculativeRouterNode — speculative dispatch/overlap. As the agent streams a structured call, it repairs the still-truncated JSON (repair_json), dispatches a downstream target immediately, and keeps generating; when the finalized call arrives it verifies (keep on hit, cancel + re-run on miss). A combine hook lets the agent's full text remain a returned deliverable, so the overlap win is real and not from throwing output away. Use only for idempotent/cancel-safe targets.
FirstMatchNode — first-answer-wins fan-out. Races several branches, returns the first whose output satisfies a match predicate, and cancels the losers (search Recall@k).
enable_uvloop() — one-line runtime switch to run agents on the libuv event loop.
These are aimed squarely at search/fan-out tasks: read N sources in parallel, take the first good answer, don't keep summarizing the ones you're discarding, and overlap dependent steps.

Testing Plan
Unit Tests:

I have added or updated unit tests for my change.

All unit tests pass locally.
New/updated suites: test_streaming_router.py, test_speculative_router.py (incl. repair_json + combine), test_first_match_node.py, test_event_loop.py.

uv run pytest -q tests/unittests/workflow/test_streaming_router.py
tests/unittests/workflow/test_speculative_router.py
tests/unittests/workflow/test_first_match_node.py
tests/unittests/utils/test_event_loop.py
29 passed, 1 warning in 6.88s
Unit tests use scripted agents (no network) and cover: preemption cut points, speculation hit/miss + rollback, JSON repair edge cases, combine returning both agent text and target result, cross-branch cancellation, no-match fallback, and max_parallel.

Manual End-to-End (E2E) Tests:

Real-LLM integration tests (Vertex, gemini-3.5-flash-lite, whole documents, no chunking). They read config from the environment and skip unless GOOGLE_CLOUD_PROJECT is set:

uv run pytest -s -p no:cacheprovider
tests/integration/test_streaming_router_preemption_timing.py
tests/integration/test_speculative_router_chained_llm.py
tests/integration/test_streaming_router_tsla_10k.py
Both integration tests are written to be honest — the feature's output is a required deliverable, so a naive baseline gets no shortcut:

Preemption (5 arXiv papers in parallel): the relevant paper's full summary is kept and asserted in both runs; only the 4 irrelevant summaries are skipped. Result: 6.31s → 3.36s (1.88× faster), output tokens 3531 → 673 (5.25× fewer), cost −10.4% (−48.3% with context caching). Relevant summary preserved (465 vs 413 words).
Speculative chained LLM (planner → worker): planner rationale is a required output; both paths return {plan, answer}. Result: 6.11s → 3.52s, 2.59s saved (~72% of the worker call hidden behind the required rationale tail).
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. (N/A — no downstream dependencies.)
Additional context
All features are opt-in and additive; no existing node behavior changes. Samples are included under contributing/samples/workflows/ (streaming_route, fan_out_preempt, search_fanout_first_answer, speculative_tool) and a guide at docs/guides/workflow/streaming_preemption/. Speculative dispatch must be used with idempotent/cancel-safe targets only.