feat: support graceful abort and foreground-to-background subagent handoff

Add abort signal propagation through tool execution layers, allowing
concurrent and sequential tool runs to be cancelled cleanly. Implement
automatic handoff of interrupted foreground subagents to background
tasks so they can continue running after user interruption. Fix UI
event loop shutdown to cancel pending tasks before async generators,
and defer background notification responses when the interactive prompt
is active to avoid rendering races.

Generated with Ripperdoc

Co-Authored-By: Ripperdoc

Author: jstzwj

ripperdoc/cli/ui/rich_ui/rendering.py

@@ -27,6 +27,9 @@ def _summarize_subagent_progress_content(content: Any) -> str:
         return str(content)
 
     message_payload = getattr(content, "message", None)
+    metadata = getattr(message_payload, "metadata", None) or {}
+    if metadata.get("hook_additional_context"):
+        return ""
     body = getattr(message_payload, "content", None) if message_payload is not None else None
     if isinstance(body, str):
         return body

ripperdoc/cli/ui/rich_ui/session.py

@@ -1519,6 +1519,22 @@ def _schedule_background_notification_response(
         """Trigger a follow-up assistant response for an idle task notification."""
         if self._loop.is_closed():
             return
+        # When the interactive prompt is active, avoid launching an automatic
+        # query that would race prompt_toolkit redraw and spinner rendering.
+        # Defer it into pending messages so it is consumed on the next user turn.
+        prompt_app = getattr(getattr(self, "_prompt_session", None), "app", None)
+        if (
+            not self._query_in_progress
+            and prompt_app is not None
+            and bool(getattr(prompt_app, "is_running", False))
+            and self.query_context is not None
+        ):
+            self.query_context.pending_message_queue.enqueue_text(
+                agent_message,
+                metadata=metadata,
+            )
+            self._request_prompt_redraw()
+            return
 
         async def _run() -> None:
             try:
@@ -1974,7 +1990,13 @@ def _run_async(self, coro: Any) -> Any:
         if threading.current_thread() is self._loop_thread:
             raise RuntimeError("_run_async cannot be called from the UI event loop thread")
         future = asyncio.run_coroutine_threadsafe(coro, self._loop)
-        return future.result()
+        try:
+            return future.result()
+        except KeyboardInterrupt:
+            # If the caller interrupts while waiting, propagate cancellation to the
+            # coroutine so async generators are not left running into shutdown.
+            future.cancel()
+            raise
 
     def run_async(self, coro: Any) -> Any:
         """Public wrapper for running coroutines on the UI event loop."""
@@ -2432,11 +2454,21 @@ def _shutdown_event_loop_thread(self) -> None:
         if self._loop.is_closed():
             return
 
-        async def _shutdown_asyncgens_only() -> None:
-            await asyncio.get_running_loop().shutdown_asyncgens()
+        async def _drain_tasks_and_shutdown_asyncgens() -> None:
+            loop = asyncio.get_running_loop()
+            current = asyncio.current_task(loop=loop)
+            pending = [
+                task for task in asyncio.all_tasks(loop)
+                if task is not current and not task.done()
+            ]
+            for task in pending:
+                task.cancel()
+            if pending:
+                await asyncio.gather(*pending, return_exceptions=True)
+            await loop.shutdown_asyncgens()
 
         try:
-            self._run_async(_shutdown_asyncgens_only())
+            self._run_async(_drain_tasks_and_shutdown_asyncgens())
         except (RuntimeError, asyncio.CancelledError, concurrent.futures.TimeoutError):
             pass
 

ripperdoc/core/query/loop.py

@@ -1438,7 +1438,11 @@ async def _process_iteration_assistant_message(
         return
 
     if prepared.prepared_calls:
-        async for message in _run_tools_concurrently(prepared.prepared_calls, prepared.tool_results):
+        async for message in _run_tools_concurrently(
+            prepared.prepared_calls,
+            prepared.tool_results,
+            abort_signal=query_context.abort_controller,
+        ):
             yield message
 
     _apply_skill_context_updates(prepared.tool_results, query_context, context)
@@ -1491,6 +1495,7 @@ async def _process_iteration_assistant_message(
                 async for message in _run_tools_concurrently(
                     auto_prepared.prepared_calls,
                     auto_prepared.tool_results,
+                    abort_signal=query_context.abort_controller,
                 ):
                     yield message
 

ripperdoc/core/query/tools.py

@@ -456,21 +456,64 @@ def _group_tool_calls_by_concurrency(prepared_calls: List[Dict[str, Any]]) -> Li
 
 
 async def _execute_tools_sequentially(
-    items: List[Dict[str, Any]], tool_results: List[UserMessage]
+    items: List[Dict[str, Any]],
+    tool_results: List[UserMessage],
+    abort_signal: Optional[asyncio.Event] = None,
 ) -> AsyncGenerator[Union[UserMessage, ProgressMessage], None]:
     """Run tool generators one by one."""
     for item in items:
         gen = item.get("generator")
         if not gen:
             continue
-        async for message in gen:
+        if abort_signal is None:
+            async for message in gen:
+                if isinstance(message, UserMessage):
+                    tool_results.append(message)
+                yield message
+            continue
+
+        while True:
+            if abort_signal.is_set():
+                await gen.aclose()
+                return
+
+            next_item = asyncio.create_task(gen.__anext__())
+            abort_waiter = asyncio.create_task(abort_signal.wait())
+            done, pending = await asyncio.wait(
+                {next_item, abort_waiter},
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+
+            for pending_task in pending:
+                pending_task.cancel()
+                try:
+                    await pending_task
+                except asyncio.CancelledError:
+                    pass
+
+            if abort_waiter in done and abort_signal.is_set():
+                next_item.cancel()
+                try:
+                    await next_item
+                except (asyncio.CancelledError, StopAsyncIteration):
+                    pass
+                await gen.aclose()
+                return
+
+            try:
+                message = next_item.result()
+            except StopAsyncIteration:
+                break
+
             if isinstance(message, UserMessage):
                 tool_results.append(message)
             yield message
 
 
 async def _execute_tools_in_parallel(
-    items: List[Dict[str, Any]], tool_results: List[UserMessage]
+    items: List[Dict[str, Any]],
+    tool_results: List[UserMessage],
+    abort_signal: Optional[asyncio.Event] = None,
 ) -> AsyncGenerator[Union[UserMessage, ProgressMessage], None]:
     """Run tool generators concurrently."""
     logger.debug("[query] _execute_tools_in_parallel ENTER: %d items", len(items))
@@ -482,42 +525,65 @@ async def _execute_tools_in_parallel(
         len(generators),
         tool_names,
     )
-    async for message in _run_concurrent_tool_uses(generators, tool_names, tool_results):
+    async for message in _run_concurrent_tool_uses(
+        generators,
+        tool_names,
+        tool_results,
+        abort_signal=abort_signal,
+    ):
         yield message
     logger.debug("[query] _execute_tools_in_parallel DONE")
 
 
 async def _run_tools_concurrently(
-    prepared_calls: List[Dict[str, Any]], tool_results: List[UserMessage]
+    prepared_calls: List[Dict[str, Any]],
+    tool_results: List[UserMessage],
+    abort_signal: Optional[asyncio.Event] = None,
 ) -> AsyncGenerator[Union[UserMessage, ProgressMessage], None]:
     """Run tools grouped by concurrency safety (parallel for safe groups)."""
     for group in _group_tool_calls_by_concurrency(prepared_calls):
         if group["is_concurrency_safe"]:
             logger.debug(
                 f"[query] Executing {len(group['items'])} concurrency-safe tool(s) in parallel"
             )
-            async for message in _execute_tools_in_parallel(group["items"], tool_results):
+            async for message in _execute_tools_in_parallel(
+                group["items"],
+                tool_results,
+                abort_signal=abort_signal,
+            ):
                 yield message
         else:
             logger.debug(
                 f"[query] Executing {len(group['items'])} tool(s) sequentially (not concurrency safe)"
             )
-            async for message in _run_tools_serially(group["items"], tool_results):
+            async for message in _run_tools_serially(
+                group["items"],
+                tool_results,
+                abort_signal=abort_signal,
+            ):
                 yield message
 
 
 async def _run_tools_serially(
-    prepared_calls: List[Dict[str, Any]], tool_results: List[UserMessage]
+    prepared_calls: List[Dict[str, Any]],
+    tool_results: List[UserMessage],
+    abort_signal: Optional[asyncio.Event] = None,
 ) -> AsyncGenerator[Union[UserMessage, ProgressMessage], None]:
     """Run all tools sequentially (helper for clarity)."""
-    async for message in _execute_tools_sequentially(prepared_calls, tool_results):
+    async for message in _execute_tools_sequentially(
+        prepared_calls,
+        tool_results,
+        abort_signal=abort_signal,
+    ):
         yield message
 
 
 async def _run_concurrent_tool_uses(
     generators: List[AsyncGenerator[Union[UserMessage, ProgressMessage], None]],
     tool_names: List[str],
     tool_results: List[UserMessage],
+    *,
+    abort_signal: Optional[asyncio.Event] = None,
 ) -> AsyncGenerator[Union[UserMessage, ProgressMessage], None]:
     """Drain multiple tool generators concurrently and stream outputs with overall timeout."""
     overall_timeout_sec = _resolve_concurrent_timeout_sec(tool_names)
@@ -601,8 +667,41 @@ async def _consume(
                 logger.debug(
                     "[query] _run_concurrent_tool_uses: waiting for queue.get(), active=%d", active
                 )
+                abort_waiter: Optional[asyncio.Task[bool]] = None
+                queue_waiter: Optional[asyncio.Task[Optional[Union[UserMessage, ProgressMessage]]]] = None
                 try:
-                    message = await _queue_get_with_timeout(queue, overall_timeout_sec)
+                    queue_waiter = asyncio.create_task(
+                        _queue_get_with_timeout(queue, overall_timeout_sec)
+                    )
+                    if abort_signal is not None:
+                        abort_waiter = asyncio.create_task(abort_signal.wait())
+                        done, pending = await asyncio.wait(
+                            {queue_waiter, abort_waiter},
+                            return_when=asyncio.FIRST_COMPLETED,
+                        )
+                        for pending_task in pending:
+                            pending_task.cancel()
+                            try:
+                                await pending_task
+                            except asyncio.CancelledError:
+                                pass
+
+                        if abort_waiter in done and abort_signal.is_set():
+                            logger.info(
+                                "[query] Abort signal set; cancelling %d concurrent tool task(s)",
+                                len(tasks),
+                            )
+                            for task in tasks:
+                                if not task.done():
+                                    task.cancel()
+                            break
+
+                        if queue_waiter not in done:
+                            # Defensive: should not happen, but keep behavior deterministic.
+                            continue
+                        message = queue_waiter.result()
+                    else:
+                        message = await queue_waiter
                 except asyncio.TimeoutError:
                     logger.error(
                         "[query] Concurrent tool execution timed out waiting for messages"
@@ -612,6 +711,19 @@ async def _consume(
                         if not task.done():
                             task.cancel()
                     raise
+                finally:
+                    if abort_waiter is not None and not abort_waiter.done():
+                        abort_waiter.cancel()
+                        try:
+                            await abort_waiter
+                        except asyncio.CancelledError:
+                            pass
+                    if queue_waiter is not None and not queue_waiter.done():
+                        queue_waiter.cancel()
+                        try:
+                            await queue_waiter
+                        except asyncio.CancelledError:
+                            pass
 
                 logger.debug(
                     "[query] _run_concurrent_tool_uses: got message type=%s, active=%d",