@@ -324,11 +324,19 @@ async def run_async(
   ) -> AsyncGenerator[Event, None]:
     """Runs the flow."""
     while True:
+      # `is True` rather than truthiness: tests drive agents with mocked
+      # contexts, whose is_aborted is a truthy Mock rather than False.
+      if invocation_context.is_aborted is True:
+        break
       last_event = None
       async with Aclosing(self._run_one_step_async(invocation_context)) as agen:
         async for event in agen:
           last_event = event
           yield event
+          if invocation_context.is_aborted is True:
+            break
+      if invocation_context.is_aborted is True:
+        break
       if not last_event or last_event.is_final_response() or last_event.partial:
         if last_event and last_event.partial:
           logger.warning('The last event is partial, which is not expected.')
 
@@ -550,6 +550,7 @@ async def _run_node_async(
       yield_user_message: bool = False,
       node: BaseNode | None = None,
       session: Optional[Session] = None,
+      abort_signal: Optional[asyncio.Event] = None,
   ) -> AsyncGenerator[Event, None]:
     """Run a BaseNode through NodeRunner.
 
@@ -569,6 +570,7 @@ async def _run_node_async(
             yield_user_message=yield_user_message,
             node=node,
             session=session,
+            abort_signal=abort_signal,
         )
     ) as agen:
       async for event in agen:
@@ -755,52 +757,99 @@ async def _consume_event_queue(
     """Consume events from ic._event_queue until done_sentinel."""
     event_queue: asyncio.Queue[_EventQueueItem] | None = ic._event_queue
     assert event_queue is not None
-    while True:
-      event_or_done, processed_signal = await event_queue.get()
-      if event_or_done is done_sentinel:
-        break
-      if not isinstance(event_or_done, Event):
-        raise TypeError(
-            f'Unexpected node event queue item: {type(event_or_done).__name__}'
-        )
-      event = event_or_done
-      output_event = await self._process_event_with_plugin_callbacks(
-          invocation_context=ic,
-          event=event,
-      )
+    if ic.is_aborted and event_queue.empty():
+      return
+    abort_task: asyncio.Task[bool] | None = None
+    get_task: asyncio.Task[_EventQueueItem] | None = None
+    try:
+      while True:
+        # Anything still queued after abort is dropped rather than drained,
+        # except an event that a node produced before checking abort in the
+        # same turn. A producer parked in _enqueue_event is released by
+        # _cleanup_root_task, not by this loop.
+        if ic.is_aborted and event_queue.empty():
+          break
+        if not event_queue.empty():
+          event_or_done, processed_signal = event_queue.get_nowait()
+        else:
+          if abort_task is None or abort_task.done():
+            abort_task = asyncio.create_task(
+                ic._abort_signal.wait()  # pylint: disable=protected-access
+            )
+          get_task = asyncio.create_task(event_queue.get())
+          done, _ = await asyncio.wait(
+              [get_task, abort_task], return_when=asyncio.FIRST_COMPLETED
+          )
+          if not get_task.done():
+            # Let get_task finish its step if event_queue.put() already resolved
+            # its internal getter future in the same event-loop turn as abort().
+            await asyncio.sleep(0)
+          if not get_task.done():
+            break
+          event_or_done, processed_signal = get_task.result()
+          get_task = None
+        if event_or_done is done_sentinel:
+          break
+        if not isinstance(event_or_done, Event):
+          raise TypeError(
+              'Unexpected node event queue item:'
+              f' {type(event_or_done).__name__}'
+          )
+        event = event_or_done
 
-      # When an LlmAgent node uses ``message_as_output`` (no
-      # ``output_schema``), the wrapper sets both ``event.content``
-      # (the model's text) AND ``event.output`` (the same text) to
-      # signal that the message IS the node's output.  Clear
-      # ``event.output`` on a copy here so downstream renderers don't
-      # surface the same text twice.  Task-mode agents set
-      # ``event.output`` from the ``finish_task`` FC args without
-      # ``message_as_output``, so this clearing doesn't affect them.
-      if not output_event.partial:
-        if (
-            output_event.node_info
-            and output_event.node_info.message_as_output
-            and output_event.content is not None
-        ):
-          output_event = output_event.model_copy()
-          output_event.output = None
-        await self.session_service.append_event(
-            session=ic.session, event=output_event
+        async def _process_and_clear(invocation_context, ev_arg):
+          nonlocal event
+          ev = await self._process_event_with_plugin_callbacks(
+              invocation_context=invocation_context,
+              event=ev_arg,
+          )
+          if not ev.partial:
+            if (
+                ev.node_info
+                and ev.node_info.message_as_output
+                and ev.content is not None
+            ):
+              ev = ev.model_copy()
+              ev.output = None
+          event = ev
+          return ev
+
+        output_event = await _process_and_clear(
+            invocation_context=ic,
+            ev_arg=event,
         )
-      yield output_event
 
-      if isinstance(processed_signal, asyncio.Event):
-        processed_signal.set()
+        if not event.partial:
+          await self.session_service.append_event(
+              session=ic.session, event=output_event
+          )
+        yield output_event
+
+        if isinstance(processed_signal, asyncio.Event):
+          processed_signal.set()
+    finally:
+      pending_tasks = [
+          t for t in (get_task, abort_task) if t is not None and not t.done()
+      ]
+      for t in pending_tasks:
+        t.cancel()
+      if pending_tasks:
+        await asyncio.wait(pending_tasks)
 
   async def _cleanup_root_task(
-      self, task: asyncio.Task[None], node_name: str
+      self,
+      task: asyncio.Task[None],
+      node_name: str,
   ) -> None:
     """Cancel the root task if still running, then await it.
 
     The task may still be running if the caller stopped iterating
     early (e.g., break in async for). In that case we must cancel
     to avoid a leaked task.
+
+    Args:
+      task: The root task to cancel and await.
+      node_name: The name of the root node, used for logging.
     """
     cancelled_by_cleanup = False
     if not task.done():
@@ -1037,6 +1086,7 @@ async def run_async(
       state_delta: Optional[dict[str, Any]] = None,
       run_config: Optional[RunConfig] = None,
       yield_user_message: bool = False,
+      abort_signal: Optional[asyncio.Event] = None,
   ) -> AsyncGenerator[Event, None]:
     """Main entry method to run the agent in this runner.
 
@@ -1056,6 +1106,7 @@ async def run_async(
       run_config: The run config for the agent.
       yield_user_message: If True, yield the user message event before
         agent/node events.
+      abort_signal: Optional asyncio.Event to cancel the invocation.
 
     Yields:
       The events generated by the agent.
@@ -1144,6 +1195,7 @@ async def run_async(
               yield_user_message=yield_user_message,
               node=agent_to_run,
               session=session,
+              abort_signal=abort_signal,
           )
       ) as agen:
         async for event in agen:
@@ -1164,6 +1216,7 @@ async def run_async(
               state_delta=state_delta,
               run_config=run_config,
               yield_user_message=yield_user_message,
+              abort_signal=abort_signal,
           )
       ) as agen:
         async for event in agen:
@@ -1213,6 +1266,7 @@ async def _run_with_trace(
               run_config=run_config,
               state_delta=state_delta,
               invocation_id=invocation_id,
+              abort_signal=abort_signal,
           )
         else:
           invocation_id = self._resolve_invocation_id(
@@ -1228,6 +1282,7 @@ async def _run_with_trace(
                 new_message=new_message,
                 run_config=run_config,
                 state_delta=state_delta,
+                abort_signal=abort_signal,
             )
           else:
             invocation_context = (
@@ -1237,6 +1292,7 @@ async def _run_with_trace(
                     invocation_id=invocation_id,
                     run_config=run_config,
                     state_delta=state_delta,
+                    abort_signal=abort_signal,
                 )
             )
             active_agent = invocation_context.agent
@@ -1249,15 +1305,21 @@ async def _run_with_trace(
               # already final.
               return
 
+        invocation_context._abort_state.loop = asyncio.get_running_loop()  # pylint: disable=protected-access
+
         async def execute(
             ctx: InvocationContext,
         ) -> AsyncGenerator[Event, None]:
+          if ctx.is_aborted:
+            return
           active_agent = ctx.agent
           if not isinstance(active_agent, BaseAgent):
             raise RuntimeError('Agent execution has no active BaseAgent.')
           async with aclosing(active_agent.run_async(ctx)) as agen:
             async for event in agen:
               yield event
+              if ctx.is_aborted:
+                break
 
         async with aclosing(
             _with_caller_context(
@@ -1272,14 +1334,24 @@ async def execute(
         ) as agen:
           async for event in agen:
             yield event
+            if invocation_context.is_aborted:
+              break
         # Run compaction after all events are yielded from the agent.
         # (We don't compact in the middle of an invocation, we only compact at
         # the end of an invocation.)
-        await self._run_post_invocation_compaction(
-            session=invocation_context.session,
-            skip_token_compaction=(invocation_context.token_compaction_checked),
-        )
+        if not invocation_context.is_aborted:
+          await self._run_post_invocation_compaction(
+              session=invocation_context.session,
+              skip_token_compaction=(
+                  invocation_context.token_compaction_checked
+              ),
+          )
 
+    # For BaseAgent root agents running via _run_with_trace, events flow
+    # through _run_with_trace which breaks cleanly upon abort.
+    # Note that BaseNode/Workflow root targets are routed earlier via
+    # _run_node_async, where Workflow and NodeRunner manage task cancellation
+    # and queue-draining internally.
     async with aclosing(_run_with_trace(new_message, invocation_id)) as agen:
       async for event in agen:
         yield event
@@ -1821,6 +1893,7 @@ async def _setup_context_for_new_invocation(
       run_config: RunConfig,
       state_delta: Optional[dict[str, Any]],
       invocation_id: Optional[str] = None,
+      abort_signal: Optional[asyncio.Event] = None,
   ) -> InvocationContext:
     """Sets up the context for a new invocation.
 
@@ -1830,6 +1903,7 @@ async def _setup_context_for_new_invocation(
       run_config: The run config of the agent.
       state_delta: Optional state changes to apply to the session.
       invocation_id: Optional invocation identifier.
+      abort_signal: Optional abort signal to cancel this invocation.
 
     Returns:
       The invocation context for the new invocation.
@@ -1841,6 +1915,11 @@ async def _setup_context_for_new_invocation(
         run_config=run_config,
         invocation_id=invocation_id,
     )
+    if abort_signal is not None:
+      # Attached here rather than passed to `_new_invocation_context`, whose
+      # signature subclasses override. Safe at this point because nothing has
+      # derived a sub-context from this one yet.
+      invocation_context._attach_abort_signal(abort_signal)
     # Step 2: Handle new message, by running callbacks and appending to
     # session.
     await self._handle_new_message(
@@ -1868,6 +1947,7 @@ async def _setup_context_for_resumed_invocation(
       invocation_id: str,
       run_config: RunConfig,
       state_delta: Optional[dict[str, Any]],
+      abort_signal: Optional[asyncio.Event] = None,
   ) -> InvocationContext:
     """Sets up the context for a resumed invocation.
 
@@ -1877,6 +1957,7 @@ async def _setup_context_for_resumed_invocation(
       invocation_id: The invocation id to resume.
       run_config: The run config of the agent.
       state_delta: Optional state changes to apply to the session.
+      abort_signal: Optional abort signal to cancel this invocation.
 
     Returns:
       The invocation context for the resumed invocation.
@@ -1903,6 +1984,10 @@ async def _setup_context_for_resumed_invocation(
         run_config=run_config,
         invocation_id=invocation_id,
     )
+    if abort_signal is not None:
+      # See `_setup_context_for_new_invocation` for why this is attached
+      # after construction rather than passed to the factory.
+      invocation_context._attach_abort_signal(abort_signal)
     # Step 3: Maybe handle new message.
     if new_message:
       await self._handle_new_message(
@@ -1955,6 +2040,12 @@ def _new_invocation_context(
   ) -> InvocationContext:
     """Creates a new invocation context.
 
+    This is an extension point that subclasses override to build their own
+    context type, so its signature is kept stable. A caller-supplied abort
+    signal is attached to the returned context by the caller rather than
+    threaded through here, because adding a parameter would raise TypeError in
+    every existing override.
+
     Args:
         session: The session for the context.
         invocation_id: The invocation id for the context.
 
@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 from typing import Any
 from typing import Optional
@@ -314,6 +315,13 @@ async def run_async(
           update={'streaming_mode': StreamingMode.NONE}
       )
 
+    # Aborting the caller also aborts the wrapped agent. The isinstance check
+    # is intentional: tests may hand in a mocked caller context whose signal is
+    # a Mock, and the nested runner would read that as already aborted.
+    abort_signal = invocation_context._abort_signal
+    if not isinstance(abort_signal, asyncio.Event):
+      abort_signal = None
+
     last_content = None
     last_error_message = None
     last_grounding_metadata = None
@@ -323,6 +331,7 @@ async def run_async(
             session_id=session.id,
             new_message=content,
             run_config=nested_run_config,
+            abort_signal=abort_signal,
         )
     ) as agen:
       async for event in agen: