fix(async): close unscheduled coroutines in all threadsafe bridges

acp_adapter/events.py

@@ -31,10 +31,17 @@ def _send_update(
     update: Any,
 ) -> None:
     """Fire-and-forget an ACP session update from a worker thread."""
+    from agent.async_utils import safe_schedule_threadsafe
+
+    future = safe_schedule_threadsafe(
+        conn.session_update(session_id, update),
+        loop,
+        logger=logger,
+        log_message="Failed to send ACP update",
+    )
+    if future is None:
+        return
     try:
-        future = asyncio.run_coroutine_threadsafe(
-            conn.session_update(session_id, update), loop
-        )
         future.result(timeout=5)
     except Exception:
         logger.debug("Failed to send ACP update", exc_info=True)

acp_adapter/permissions.py

@@ -111,21 +111,28 @@ def _callback(
         allow_permanent: bool = True,
         **_: object,
     ) -> str:
+        from agent.async_utils import safe_schedule_threadsafe
+
         options = _build_permission_options(allow_permanent=allow_permanent)
 
-        future = None
+        tool_call = _build_permission_tool_call(command, description)
+        coro = request_permission_fn(
+            session_id=session_id,
+            tool_call=tool_call,
+            options=options,
+        )
+        future = safe_schedule_threadsafe(
+            coro, loop,
+            logger=logger,
+            log_message="Permission request: failed to schedule on loop",
+        )
+        if future is None:
+            return "deny"
+
         try:
-            tool_call = _build_permission_tool_call(command, description)
-            coro = request_permission_fn(
-                session_id=session_id,
-                tool_call=tool_call,
-                options=options,
-            )
-            future = asyncio.run_coroutine_threadsafe(coro, loop)
             response = future.result(timeout=timeout)
         except (FutureTimeout, Exception) as exc:
-            if future is not None:
-                future.cancel()
+            future.cancel()
             logger.warning("Permission request timed out or failed: %s", exc)
             return "deny"
 

agent/async_utils.py

@@ -0,0 +1,68 @@
+"""Async/sync bridging helpers.
+
+The codebase has ~30 sites that schedule a coroutine onto an event loop from a
+worker thread via :func:`asyncio.run_coroutine_threadsafe`.  That function can
+raise :class:`RuntimeError` (e.g. the loop was closed during a shutdown race),
+and when it does the coroutine object is never awaited and never closed —
+which triggers a ``"coroutine '<name>' was never awaited"`` RuntimeWarning and
+leaks the coroutine's frame until GC.
+
+:func:`safe_schedule_threadsafe` wraps the call, closes the coroutine on
+scheduling failure, and returns ``None`` (instead of a half-formed future) so
+callers can branch cleanly:
+
+    fut = safe_schedule_threadsafe(coro, loop)
+    if fut is None:
+        return  # or fallback behavior
+    fut.result(timeout=5)
+
+The helper deliberately does NOT also handle ``future.result()`` failures —
+that is a separate concern.  Once the loop has accepted the coroutine, its
+lifecycle belongs to the loop, not the scheduling thread.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from concurrent.futures import Future
+from typing import Any, Coroutine, Optional
+
+
+_DEFAULT_LOGGER = logging.getLogger(__name__)
+
+
+def safe_schedule_threadsafe(
+    coro: Coroutine[Any, Any, Any],
+    loop: Optional[asyncio.AbstractEventLoop],
+    *,
+    logger: Optional[logging.Logger] = None,
+    log_message: str = "Failed to schedule coroutine on loop",
+    log_level: int = logging.DEBUG,
+) -> Optional[Future]:
+    """Schedule ``coro`` on ``loop`` from a sync context, leak-safe.
+
+    Returns the :class:`concurrent.futures.Future` on success, or ``None`` if
+    the loop is missing or :func:`asyncio.run_coroutine_threadsafe` raised
+    (e.g. the loop was closed during a shutdown race).  In all failure paths
+    the coroutine is :meth:`close`-d so it does not trigger
+    ``"coroutine was never awaited"`` warnings or leak its frame.
+
+    Callers retain full control over what to do with the returned future
+    (call ``.result(timeout=...)``, attach ``add_done_callback``, ignore it
+    fire-and-forget, etc.).
+    """
+    log = logger if logger is not None else _DEFAULT_LOGGER
+
+    if loop is None:
+        if asyncio.iscoroutine(coro):
+            coro.close()
+        log.log(log_level, "%s: loop is None", log_message)
+        return None
+
+    try:
+        return asyncio.run_coroutine_threadsafe(coro, loop)
+    except Exception as exc:
+        if asyncio.iscoroutine(coro):
+            coro.close()
+        log.log(log_level, "%s: %s", log_message, exc)
+        return None

agent/lsp/manager.py

@@ -107,9 +107,14 @@ def run(self, coro, *, timeout: Optional[float] = None) -> Any:
 
         Returns the coroutine's result, or raises its exception.
         """
+        from agent.async_utils import safe_schedule_threadsafe
         if self._loop is None:
+            if asyncio.iscoroutine(coro):
+                coro.close()
             raise RuntimeError("background loop not started")
-        fut: ConcurrentFuture = asyncio.run_coroutine_threadsafe(coro, self._loop)
+        fut = safe_schedule_threadsafe(coro, self._loop)
+        if fut is None:
+            raise RuntimeError("background loop not running")
         try:
             return fut.result(timeout=timeout)
         except Exception:

cron/scheduler.py

@@ -464,7 +464,14 @@ def _send_media_via_adapter(
             else:
                 coro = adapter.send_document(chat_id=chat_id, file_path=media_path, metadata=metadata)
 
-            future = asyncio.run_coroutine_threadsafe(coro, loop)
+            from agent.async_utils import safe_schedule_threadsafe
+            future = safe_schedule_threadsafe(coro, loop)
+            if future is None:
+                logger.warning(
+                    "Job '%s': cannot send media %s, gateway loop unavailable",
+                    job.get("id", "?"), media_path,
+                )
+                return
             try:
                 result = future.result(timeout=30)
             except TimeoutError:
@@ -585,22 +592,26 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 text_to_send = cleaned_delivery_content.strip()
                 adapter_ok = True
                 if text_to_send:
-                    future = asyncio.run_coroutine_threadsafe(
+                    from agent.async_utils import safe_schedule_threadsafe
+                    future = safe_schedule_threadsafe(
                         runtime_adapter.send(chat_id, text_to_send, metadata=send_metadata),
                         loop,
                     )
-                    try:
-                        send_result = future.result(timeout=60)
-                    except TimeoutError:
-                        future.cancel()
-                        raise
-                    if send_result and not getattr(send_result, "success", True):
-                        err = getattr(send_result, "error", "unknown")
-                        logger.warning(
-                            "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
-                            job["id"], platform_name, chat_id, err,
-                        )
-                        adapter_ok = False  # fall through to standalone path
+                    if future is None:
+                        adapter_ok = False
+                    else:
+                        try:
+                            send_result = future.result(timeout=60)
+                        except TimeoutError:
+                            future.cancel()
+                            raise
+                        if send_result and not getattr(send_result, "success", True):
+                            err = getattr(send_result, "error", "unknown")
+                            logger.warning(
+                                "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
+                                job["id"], platform_name, chat_id, err,
+                            )
+                            adapter_ok = False  # fall through to standalone path
 
                 # Send extracted media files as native attachments via the live adapter
                 if adapter_ok and media_files:

gateway/platforms/feishu.py

@@ -2273,11 +2273,7 @@ def _on_message_event(self, data: Any) -> None:
                     daemon=True,
                 ).start()
             return
-        future = asyncio.run_coroutine_threadsafe(
-            self._handle_message_event_data(data),
-            loop,
-        )
-        future.add_done_callback(self._log_background_failure)
+        self._submit_on_loop(loop, self._handle_message_event_data(data))
 
     def _enqueue_pending_inbound_event(self, data: Any) -> bool:
         """Append an event to the pending-inbound queue.
@@ -2353,16 +2349,12 @@ def _drain_pending_inbound_events(self) -> None:
                     dispatched = 0
                     requeue: List[Any] = []
                     for event in batch:
-                        try:
-                            fut = asyncio.run_coroutine_threadsafe(
-                                self._handle_message_event_data(event),
-                                loop,
-                            )
-                            fut.add_done_callback(self._log_background_failure)
+                        if self._submit_on_loop(
+                            loop, self._handle_message_event_data(event)
+                        ):
                             dispatched += 1
-                        except RuntimeError:
-                            # Loop closed between check and submit — requeue
-                            # and poll again.
+                        else:
+                            # Loop closed/unavailable — requeue and poll again.
                             requeue.append(event)
                     if requeue:
                         with self._pending_inbound_lock:
@@ -2466,11 +2458,10 @@ def _on_drive_comment_event(self, data: Any) -> None:
         if not self._loop_accepts_callbacks(loop):
             logger.warning("[Feishu] Dropping drive comment event before adapter loop is ready")
             return
-        future = asyncio.run_coroutine_threadsafe(
-            handle_drive_comment_event(self._client, data, self_open_id=self._bot_open_id),
+        self._submit_on_loop(
             loop,
+            handle_drive_comment_event(self._client, data, self_open_id=self._bot_open_id),
         )
-        future.add_done_callback(self._log_background_failure)
 
     def _on_reaction_event(self, event_type: str, data: Any) -> None:
         """Route user reactions on bot messages as synthetic text events."""
@@ -2498,11 +2489,7 @@ def _on_reaction_event(self, event_type: str, data: Any) -> None:
             or bool(getattr(loop, "is_closed", lambda: False)())
         ):
             return
-        future = asyncio.run_coroutine_threadsafe(
-            self._handle_reaction_event(event_type, data),
-            loop,
-        )
-        future.add_done_callback(self._log_background_failure)
+        self._submit_on_loop(loop, self._handle_reaction_event(event_type, data))
 
     def _on_card_action_trigger(self, data: Any) -> Any:
         """Handle card-action callback from the Feishu SDK (synchronous).
@@ -2548,11 +2535,14 @@ def _loop_accepts_callbacks(loop: Any) -> bool:
 
     def _submit_on_loop(self, loop: Any, coro: Any) -> bool:
         """Schedule background work on the adapter loop with shared failure logging."""
-        try:
-            future = asyncio.run_coroutine_threadsafe(coro, loop)
-        except Exception:
-            coro.close()
-            logger.warning("[Feishu] Failed to schedule background callback work", exc_info=True)
+        from agent.async_utils import safe_schedule_threadsafe
+        future = safe_schedule_threadsafe(
+            coro, loop,
+            logger=logger,
+            log_message="[Feishu] Failed to schedule background callback work",
+            log_level=logging.WARNING,
+        )
+        if future is None:
             return False
         future.add_done_callback(self._log_background_failure)
         return True