Feat/streaming preemption uvloop

contributing/samples/workflows/fan_out_preempt/README.md

@@ -0,0 +1,72 @@
+# Parallel Fan-Out with Mid-Stream Preemption
+
+## Overview
+
+Read several data sources **in parallel**, and **stop reading any source the
+moment it becomes apparent it's irrelevant** — don't stream 100 pages of
+analysis when the model realizes on "page 3" that the source doesn't apply.
+
+Each source is analyzed by its own streaming agent wrapped in a
+`StreamingRouterNode`. A `monitor` predicate watches that branch's token
+stream; the instant the source declares itself irrelevant, the node commits an
+`{"relevant": false}` output and **cancels the rest of that generation** via
+ADK's cooperative `aclosing` cancellation. The other branches are untouched.
+
+A `JoinNode` fans the results back in and a synthesizer drops the irrelevant
+ones.
+
+## Why it's actually parallel (and fast)
+
+- The workflow scheduler launches every fan-out branch as its own
+  `asyncio.create_task` and awaits them together, so the reads truly overlap.
+- `enable_uvloop()` puts them on a libuv event loop. Install with
+  `pip install "google-adk[uvloop]"` (or set `ADK_UVLOOP=1`).
+- Preemption saves wall-clock by killing the tail of an irrelevant read
+  instead of paying for tokens nobody uses.
+
+## How the preemption works
+
+```python
+def monitor(view: StreamView) -> StreamDecision | None:
+    if view.text.lstrip().upper().startswith("IRRELEVANT"):
+        return StreamDecision(output={"source": source, "relevant": False})
+    return None  # keep reading
+
+StreamingRouterNode(name=f"reader_{source}", agent=reader, monitor=monitor,
+                    timeout=60)
+```
+
+- Returning a `StreamDecision` commits the output and (by default) cancels the
+  remaining generation for that branch.
+- Returning `None` keeps streaming; a relevant read finishes normally and its
+  final text becomes the branch's output.
+- `timeout=60` is a hard cap so a stuck/slow source can never hold up the join.
+
+## Graph
+
+```mermaid
+graph TD
+    START --> stash_query
+    stash_query --> reader_sharepoint
+    stash_query --> reader_havian
+    stash_query --> reader_wiki
+    stash_query --> reader_crm
+    stash_query --> reader_docs
+    reader_sharepoint --> join_sources
+    reader_havian --> join_sources
+    reader_wiki --> join_sources
+    reader_crm --> join_sources
+    reader_docs --> join_sources
+    join_sources --> synthesize
+```
+
+## Notes and limits
+
+- This cancels the **LLM analysis stream**. If the expensive work is the data
+  **fetch** itself (a network/tool call), that tool must be async and
+  cooperatively cancellable for the cancel to unwind it; consider a cheap
+  relevance pre-scan before the deep read.
+- `JoinNode` waits for **all** branches. Preemption just makes the irrelevant
+  ones return sooner. To proceed the moment the relevant subset is in (dropping
+  irrelevant branches from the join entirely), you'd need a custom any/first-N
+  join.

contributing/samples/workflows/fan_out_preempt/agent.py

@@ -0,0 +1,103 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Parallel fan-out with per-branch mid-stream preemption + libuv.
+
+Reads five data sources concurrently. Each source is analyzed by its own
+streaming agent. The instant an agent's stream reveals the source is
+irrelevant to the query, that branch is preempted -- the rest of its
+generation is cancelled -- while the other branches keep running. A JoinNode
+fans the results back in and a synthesizer drops the irrelevant ones.
+
+Each branch runs as its own asyncio task, so the reads truly overlap;
+``enable_uvloop()`` puts them on a faster libuv loop.
+"""
+
+from typing import Any
+
+from google.adk import Agent
+from google.adk import enable_uvloop
+from google.adk import Event
+from google.adk import Workflow
+from google.adk.workflow import JoinNode
+from google.adk.workflow import StreamDecision
+from google.adk.workflow import StreamingRouterNode
+from google.adk.workflow import StreamView
+
+enable_uvloop()
+
+SOURCES = ('sharepoint', 'havian', 'wiki', 'crm', 'docs')
+
+# Sentinel the reader emits as its first line when a source does not apply.
+IRRELEVANT = 'IRRELEVANT'
+
+
+def stash_query(node_input: str):
+  """Puts the user query in state so every reader can template it in."""
+  yield Event(state={'query': node_input})
+
+
+def _make_reader(source: str) -> StreamingRouterNode:
+  reader = Agent(
+      name=f'read_{source}',
+      instruction=(
+          f'You are reading the "{source}" source to answer: {{query}}.\n'
+          'If this source is clearly irrelevant to the query, your FIRST'
+          f' line must be exactly "{IRRELEVANT}". Otherwise, extract only the'
+          ' facts from this source that help answer the query.'
+      ),
+      output_key=f'{source}_result',
+  )
+
+  def monitor(view: StreamView) -> StreamDecision | None:
+    # As soon as the model declares irrelevance, stop reading this source.
+    if view.text.lstrip().upper().startswith(IRRELEVANT):
+      return StreamDecision(output={'source': source, 'relevant': False})
+    # Otherwise keep streaming; a relevant read finishes normally and its
+    # final text becomes this branch's output.
+    return None
+
+  return StreamingRouterNode(
+      name=f'reader_{source}',
+      agent=reader,
+      monitor=monitor,
+      # Bound the deep read so a stuck/slow source can never hold up the join.
+      timeout=60,
+  )
+
+
+readers = tuple(_make_reader(source) for source in SOURCES)
+join_sources = JoinNode(name='join_sources')
+
+
+async def synthesize(node_input: dict[str, Any]):
+  """Fan-in: drop the branches that preempted as irrelevant, then answer."""
+  relevant = {
+      name: result
+      for name, result in node_input.items()
+      if not (isinstance(result, dict) and result.get('relevant') is False)
+  }
+  skipped = sorted(set(node_input) - set(relevant))
+  yield Event(
+      message=(
+          f'Answer synthesized from {sorted(relevant)}.\n'
+          f'Skipped (irrelevant, preempted mid-read): {skipped}.'
+      ),
+  )
+
+
+root_agent = Workflow(
+    name='root_agent',
+    edges=[('START', stash_query, readers, join_sources, synthesize)],
+)

contributing/samples/workflows/search_fanout_first_answer/README.md

@@ -0,0 +1,75 @@
+# Search Fan-Out: Answer From the First Source That Has It
+
+## Overview
+
+The enterprise-search shape (Recall@k): a keyword search returns the top-k
+candidate sources — SharePoint, Confluence, CRM, a drive, email — and you don't
+know which one holds the answer. So you read them **in parallel** and answer the
+moment *any* one of them does, without waiting on (or paying to finish) the rest.
+
+This sample combines two mechanisms:
+
+1. **Per-branch preemption** — each source is read by a `StreamingRouterNode`
+   whose `monitor` stops that read as soon as the model can say "answer is here"
+   or "not here". A single irrelevant source never streams to the end.
+2. **Cross-branch first-answer-wins** — `FirstMatchNode` races the branches and,
+   the instant one returns `found=True`, **cancels the still-running siblings**
+   (tearing down their in-flight model calls) and returns that answer.
+
+`FirstMatchNode` is the complement of `JoinNode`: `JoinNode` waits for **all**
+predecessors; `FirstMatchNode` returns the **first** matching one and cancels the
+losers.
+
+## How the race works
+
+```python
+first_answer = FirstMatchNode(
+    name="first_answer",
+    nodes=[reader_sharepoint, reader_confluence, reader_crm, ...],
+    match=lambda r: isinstance(r, dict) and r.get("found"),
+    no_match_output={"found": False, "answer": "Not found in any source."},
+)
+```
+
+- Every branch is launched concurrently and handed the same input.
+- The first result the `match` predicate accepts wins; the rest are cancelled and
+  awaited (so their reads are actually torn down before the graph advances).
+- A branch that *fails* is logged and skipped, so one flaky source can't deny an
+  answer another source can give.
+- If no branch matches, the node yields `no_match_output`.
+
+## Why the cancellation is real
+
+Cancelling a branch propagates cooperatively: the `FirstMatchNode` cancels the
+branch's `asyncio` task → the dynamic scheduler's `await` on the child unwinds →
+`CancelledError` reaches the `StreamingRouterNode`'s `Aclosing` block → the SSE
+model call is closed. The loser stops **decoding** immediately.
+
+`enable_uvloop()` puts the concurrent reads on a libuv loop. Install with
+`pip install "google-adk[uvloop]"` (or set `ADK_UVLOOP=1`).
+
+## What it saves — and what it doesn't
+
+- **Saves:** the losers' output generation and the wall-clock of waiting on the
+  slowest branch. You answer at ~the speed of the fastest source that has it.
+- **Does not save:** the input already read. A branch that started still paid to
+  *prefill* its source. Preemption/racing is a **decode-side** win.
+- **To also cut prefill:** pass sources in rank order and set `max_parallel`
+  (e.g. `2`) so an early win short-circuits before lower-ranked sources are ever
+  read. With `max_parallel=1` it degrades to a cheap sequential gate.
+
+## Graph
+
+```mermaid
+graph TD
+    START --> stash_query
+    stash_query --> first_answer
+    first_answer --> respond
+    subgraph first_answer [FirstMatchNode: race, first-wins, cancel losers]
+        reader_sharepoint
+        reader_confluence
+        reader_crm
+        reader_gdrive
+        reader_email
+    end
+```

contributing/samples/workflows/search_fanout_first_answer/agent.py

@@ -0,0 +1,130 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Search fan-out that answers from the first source that has it (Recall@k).
+
+The classic enterprise-search shape: a keyword search returns the top-k candidate
+sources (SharePoint, Confluence, CRM, a drive, email...). You don't know which
+one holds the answer, so you read them **in parallel** -- and you want to answer
+the moment *any* one of them does, without waiting on (or paying to finish) the
+rest.
+
+Two mechanisms combine here:
+
+  1. Per-branch preemption -- each source is read by a ``StreamingRouterNode``
+     whose ``monitor`` stops that read as soon as the model can say "answer is
+     here" or "not here", so a single irrelevant source never streams to the end.
+  2. Cross-branch first-answer-wins -- ``FirstMatchNode`` races the branches and,
+     the instant one returns ``found=True``, cancels the still-running siblings
+     (tearing down their in-flight model calls) and returns that answer.
+
+``enable_uvloop()`` puts the concurrent reads on a libuv loop.
+
+Honest limit: branches that already started still paid to *read* their source
+(prefill). This saves the losers' *generation* and the wall-clock of waiting on
+them. To also avoid reading low-ranked sources, pass them in rank order and set
+``max_parallel`` so a win short-circuits before the tail is ever started.
+"""
+
+from google.adk import Agent
+from google.adk import enable_uvloop
+from google.adk import Event
+from google.adk import Workflow
+from google.adk.workflow import FirstMatchNode
+from google.adk.workflow import StreamDecision
+from google.adk.workflow import StreamingRouterNode
+from google.adk.workflow import StreamView
+
+enable_uvloop()
+
+# Top-k candidates as if returned by a keyword search, in rank order.
+SOURCES = ('sharepoint', 'confluence', 'crm', 'gdrive', 'email')
+
+_FOUND = 'FOUND:'
+_NOT_HERE = 'NOTHERE'
+
+
+def stash_query(node_input: str):
+  """Puts the user query in state so every reader can template it in."""
+  yield Event(state={'query': node_input})
+
+
+def _make_reader(source: str) -> StreamingRouterNode:
+  reader = Agent(
+      name=f'read_{source}',
+      instruction=(
+          f'You are searching the "{source}" source to answer: {{query}}.\n'
+          'Read only as far as you must. As soon as you can tell, output ONE'
+          ' line and nothing after it:\n'
+          f'  "{_FOUND} <the answer>"   if this source answers the question,'
+          ' or\n'
+          f'  "{_NOT_HERE}"             if it clearly does not.'
+      ),
+      output_key=f'{source}_answer',
+  )
+
+  def monitor(view: StreamView) -> StreamDecision | None:
+    upper = view.text.upper()
+    idx = upper.find(_FOUND)
+    if idx != -1:
+      line, sep, _ = view.text[idx + len(_FOUND) :].partition('\n')
+      if sep or line.strip():
+        # Answer found: commit it and preempt this branch's generation.
+        return StreamDecision(
+            output={'found': True, 'source': source, 'answer': line.strip()}
+        )
+    if _NOT_HERE in upper:
+      # Source is irrelevant: stop reading this one early.
+      return StreamDecision(output={'found': False, 'source': source})
+    return None
+
+  return StreamingRouterNode(
+      name=f'reader_{source}',
+      agent=reader,
+      monitor=monitor,
+      # Hard cap so a stuck/slow source can never hold up the race.
+      timeout=60,
+  )
+
+
+# The complement of JoinNode: race the readers, return the first that answers,
+# and cancel the losers mid-read.
+first_answer = FirstMatchNode(
+    name='first_answer',
+    nodes=[_make_reader(source) for source in SOURCES],
+    match=lambda r: isinstance(r, dict) and r.get('found'),
+    no_match_output={
+        'found': False,
+        'answer': 'Not found in any of the top sources.',
+    },
+    # Read all k at once. For rank-ordered prefill savings, set e.g.
+    # max_parallel=2 so a win short-circuits before lower ranks are read.
+)
+
+
+async def respond(node_input: dict):
+  if node_input and node_input.get('found'):
+    yield Event(
+        message=(
+            f"Answer (from {node_input['source']}): {node_input['answer']}"
+        )
+    )
+  else:
+    yield Event(message='No source contained the answer.')
+
+
+root_agent = Workflow(
+    name='root_agent',
+    edges=[('START', stash_query, first_answer, respond)],
+)