Merge branch 'simp/r3-03-rth' into simp/r3-03

agent/async_utils.py

@@ -1,24 +1,10 @@
 """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.
+``asyncio.run_coroutine_threadsafe`` can raise ``RuntimeError`` (loop closed during a
+shutdown race); the coroutine is then never awaited or closed, which triggers a
+"coroutine was never awaited" RuntimeWarning and leaks its frame. The helpers here
+close the coroutine on scheduling failure. ``future.result()`` failures are deliberately
+NOT handled: once the loop accepts the coroutine its lifecycle belongs to the loop.
 """
 from __future__ import annotations
 
@@ -41,25 +27,14 @@ def safe_schedule_threadsafe(
 ) -> 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.).
+    Returns the Future on success, or ``None`` if the loop is missing or scheduling
+    raised; in every failure path the coroutine is closed. Callers keep full control
+    over the returned future (``.result(timeout=...)``, callbacks, fire-and-forget).
     """
     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:
+        if loop is None:
+            raise RuntimeError("loop is None")
         return asyncio.run_coroutine_threadsafe(coro, loop)
     except Exception as exc:
         if asyncio.iscoroutine(coro):
@@ -69,15 +44,9 @@ def safe_schedule_threadsafe(
 
 
 def consume_detached_task_result(task: "asyncio.Future[Any]") -> None:
-    """Retrieve a detached task's result without surfacing cancellation.
-
-    Used as an ``add_done_callback`` on tasks that were cancelled and
-    detached (e.g. an adapter close path that swallows ``CancelledError``
-    past its teardown deadline). Observing ``task.exception()`` prevents
-    "exception was never retrieved" noise on the event loop; cancellation
-    and any terminal error are deliberately swallowed — the task's owner
-    already gave up on it.
-    """
+    """``add_done_callback`` for cancelled-and-detached tasks: observe the exception so the
+    loop does not log "exception was never retrieved"; cancellation and terminal errors
+    are swallowed because the task's owner already gave up on it."""
     try:
         task.exception()
     except (asyncio.CancelledError, Exception):

agent/aux_accounting.py

@@ -1,27 +1,11 @@
 """Ambient session-accounting context for auxiliary LLM calls.
 
-Auxiliary calls (vision, compression, title generation, web_extract,
-session_search, ...) funnel through ``agent.auxiliary_client`` which has no
-session handle — so their token usage was historically discarded, leaving
-dashboard analytics blind to aux model spend (issue #23270).
-
-Instead of threading ``session_db``/``session_id`` parameters through every
-aux call site, the agent loop publishes them here (mirroring the Nous Portal
-conversation context in ``agent.portal_tags``) and the auxiliary client
-records usage at its single response-validation chokepoint.
-
-ContextVar semantics give us the right isolation for free:
-
-* concurrent agents in one process (gateway sessions, delegate subagents)
-  never see each other's accounting context;
-* worker threads spawned via ``tools.thread_context.propagate_context_to_thread``
-  (MoA fan-out, background review) inherit the parent turn's context;
-* asyncio tasks inherit the context of the code that created them.
-
-MoA reference/aggregator slots are explicitly EXCLUDED from recording:
-``agent/conversation_loop.py`` already folds MoA advisor usage and cost into
-the main loop's ``update_token_counts`` delta, so recording them here would
-double-count (see ``_EXCLUDED_TASKS``).
+Aux calls (vision, compression, title generation, web_extract, session_search, ...) go
+through ``agent.auxiliary_client`` which has no session handle, so their usage was
+historically discarded. The agent loop publishes ``(session_db, session_id)`` here
+(mirroring ``agent.portal_tags``) and the aux client records usage at its single
+response-validation chokepoint. ContextVar semantics isolate concurrent agents, propagate
+to worker threads via ``tools.thread_context`` and to asyncio tasks automatically.
 """
 
 from __future__ import annotations
@@ -33,22 +17,17 @@ from typing import Any, Optional
 logger = logging.getLogger(__name__)
 
 # (session_db, session_id) for the active agent turn, or None outside one.
-_accounting: ContextVar[Optional[tuple]] = ContextVar(
-    "aux_accounting_context", default=None
-)
+_accounting: ContextVar[Optional[tuple]] = ContextVar("aux_accounting_context", default=None)
 
-# Aux tasks whose usage is already accounted by the main loop — recording
-# them here would double-count. MoA advisor/aggregator usage is folded into
-# conversation_loop's update_token_counts delta (tokens AND cost).
+# MoA advisor/aggregator usage is already folded into conversation_loop's
+# update_token_counts delta (tokens AND cost); recording it here would double-count.
 _EXCLUDED_TASKS = frozenset({"moa_reference", "moa_aggregator"})
 
 
 def set_accounting_context(session_db: Any, session_id: Optional[str]):
-    """Publish the active session's accounting handles for aux usage recording.
+    """Publish the active session's accounting handles; returns the token for ``reset_accounting_context``.
 
-    Called by the agent loop at turn entry. Returns the ContextVar token so
-    callers can ``reset_accounting_context(token)`` on turn exit. Publishing
-    ``None`` handles (no DB / no session id) clears the context.
+    ``None`` handles (no DB / no session id) clear the context.
     """
     if session_db is None or not session_id:
         return _accounting.set(None)
@@ -72,17 +51,10 @@ def record_aux_usage(
 ) -> None:
     """Record an auxiliary response's token usage against the ambient session.
 
-    Called from the auxiliary client's response-validation chokepoint. Strictly
-    best-effort: any failure is swallowed (accounting must never break an aux
-    call). No-ops when:
-
-    * no accounting context is published (call is outside any agent turn),
-    * the task is main-loop-accounted (MoA slots — see ``_EXCLUDED_TASKS``),
-    * the response carries no usage object.
-
-    The model is read from ``response.model`` (accurate even after the aux
-    client's provider-fallback chains); *provider*/*base_url* reflect the
-    originally-resolved route and are best-effort.
+    Strictly best-effort (accounting must never break an aux call). No-ops outside an
+    agent turn, for main-loop-accounted tasks (``_EXCLUDED_TASKS``), or without usage.
+    The model is read from ``response.model`` (accurate after aux provider fallback);
+    *provider*/*base_url* reflect the originally-resolved route.
     """
     try:
         if not task or task in _EXCLUDED_TASKS:
@@ -104,18 +76,14 @@ def record_aux_usage(
             or usage.reasoning_tokens
         ):
             return
-
         model = str(getattr(response, "model", "") or "") or "unknown"
         estimated_cost = None
         try:
-            cost = estimate_usage_cost(
-                model, usage, provider=provider, base_url=base_url
-            )
+            cost = estimate_usage_cost(model, usage, provider=provider, base_url=base_url)
             if cost.amount_usd is not None:
                 estimated_cost = float(cost.amount_usd)
         except Exception:
             logger.debug("Aux usage cost estimation failed", exc_info=True)
-
         session_db.record_auxiliary_usage(
             session_id,
             task,