fix(fallback): normalize exception metadata before retry decisions

Provider exceptions may expose numeric error_type or error_code values. Convert those fields at the response boundary before fallback consumers apply string operations, and cover the behavior with a regression test.

nanobot/agent/tools/session_messages.py

@@ -13,6 +13,8 @@ from dataclasses import dataclass
 from typing import Any, Protocol
 from uuid import uuid4
 
+from loguru import logger
+
 from nanobot.agent.tools.base import Tool, ToolResult, tool_parameters
 from nanobot.agent.tools.context import RequestContext, ToolContext, current_request_context
 from nanobot.agent.tools.schema import (
@@ -325,11 +327,20 @@ class SendSessionMessageTool(Tool):
         def expire() -> None:
             task = asyncio.create_task(self._expire_pending_reply(key, pending))
             self._expiry_tasks.add(task)
-            task.add_done_callback(self._expiry_tasks.discard)
+            task.add_done_callback(self._on_expiry_task_done)
 
         schedule = self._schedule_later or asyncio.get_running_loop().call_later
         pending.timer = schedule(float(timeout_seconds), expire)
 
+    def _on_expiry_task_done(self, task: asyncio.Task[None]) -> None:
+        self._expiry_tasks.discard(task)
+        if task.cancelled():
+            return
+        try:
+            task.result()
+        except Exception:
+            logger.exception("Session reply timeout delivery failed")
+
     async def _expire_pending_reply(
         self,
         key: tuple[str, str],

nanobot/channels/signal/runtime.py

@@ -862,6 +862,7 @@ class SignalChannel(BaseChannel):
                 return False, chat_id
             if (
                 self.config.group.policy == "allowlist"
+                and "*" not in self.config.group.allow_from
                 and chat_id not in self.config.group.allow_from
             ):
                 self.logger.info(
@@ -1061,6 +1062,9 @@ class SignalChannel(BaseChannel):
     def _sender_matches_allowlist(cls, sender_id: str, allow_list: list[str]) -> bool:
         """Return True if any normalized variant of sender_id is on allow_list.
 
+        A ``"*"`` entry allows every sender, matching the channel-wide
+        allowlist contract.
+
         Both ``sender_id`` and each allow_list entry can be a single
         identifier or a pipe-joined composite of several (e.g.
         ``"+1234567890|uuid-abc"``); both sides are split on ``|`` and each
@@ -1070,6 +1074,8 @@ class SignalChannel(BaseChannel):
         """
         if not allow_list:
             return False
+        if "*" in allow_list:
+            return True
         sender_variants: set[str] = set()
         for part in str(sender_id).split("|"):
             sender_variants.update(cls._normalize_signal_id(part))

nanobot/channels/signal/tests/test_signal_channel.py

@@ -790,6 +790,14 @@ class TestHandleDataMessageDM:
         await ch._handle_receive_notification(params)
         assert len(handled) == 1
 
+    @pytest.mark.asyncio
+    async def test_dm_allowlist_wildcard_preserves_content(self):
+        ch, handled = self._make_dm_channel(policy="allowlist", allow_from=["*"])
+        params = _dm_envelope(source_number="+19995550001", message="wildcard DM")
+        await ch._handle_receive_notification(params)
+        assert len(handled) == 1
+        assert handled[0]["content"] == "wildcard DM"
+
     @pytest.mark.asyncio
     async def test_dm_allowlist_rejected_triggers_pairing(self):
         # Denied DM senders go through super()._handle_message which checks
@@ -1025,6 +1033,16 @@ class TestHandleDataMessageGroup:
         await ch._handle_receive_notification(params)
         assert len(handled) == 1
 
+    @pytest.mark.asyncio
+    async def test_group_allowlist_wildcard_preserves_content(self):
+        ch, handled = self._make_group_channel(
+            policy="allowlist", allow_from=["*"], require_mention=False
+        )
+        params = _group_envelope(group_id="grp==", source_name="Alice", message="wildcard group")
+        await ch._handle_receive_notification(params)
+        assert len(handled) == 1
+        assert "[Alice]: wildcard group" in handled[0]["content"]
+
     @pytest.mark.asyncio
     async def test_group_allowlist_rejected(self):
         ch, handled = self._make_group_channel(policy="allowlist", allow_from=["other=="])

nanobot/nanobot.py

@@ -302,7 +302,7 @@ class Nanobot:
                 ))
                 raise
             finally:
-                emitter.close()
+                await emitter.close()
 
         task = asyncio.create_task(_run())
         return RunStream(task, queue)

nanobot/providers/base.py

@@ -943,6 +943,63 @@ class LLMProvider(ABC):
         """
         pass
 
+    @staticmethod
+    def _error_response_from_exception(exc: Exception) -> LLMResponse:
+        """Convert an unexpected exception while retaining retry metadata."""
+        error_names = tuple(cls.__name__.lower() for cls in type(exc).__mro__)
+        error_kind: str | None = None
+        error_should_retry: bool | None = None
+        if any("timeout" in name for name in error_names):
+            error_kind = "timeout"
+            error_should_retry = True
+        elif any(
+            token in name
+            for name in error_names
+            for token in ("connect", "connection", "network", "protocol", "transport")
+        ):
+            error_kind = "connection"
+            error_should_retry = True
+        elif any(
+            "ratelimit" in name or "throttl" in name
+            for name in error_names
+        ):
+            error_kind = "rate_limit"
+            error_should_retry = True
+        elif any(
+            "server" in name or "internal" in name
+            for name in error_names
+        ):
+            error_kind = "server_error"
+            error_should_retry = True
+        elif any(
+            token in name
+            for name in error_names
+            for token in ("auth", "credential", "permissiondenied", "unauthor")
+        ):
+            error_kind = "authentication"
+
+        response = getattr(exc, "response", None)
+        raw_status = getattr(exc, "status_code", None)
+        if raw_status is None and response is not None:
+            raw_status = getattr(response, "status_code", None)
+        try:
+            error_status_code = int(raw_status) if raw_status is not None else None
+        except (TypeError, ValueError):
+            error_status_code = None
+
+        raw_error_type = getattr(exc, "error_type", None)
+        raw_error_code = getattr(exc, "error_code", None)
+        detail = str(exc).strip() or type(exc).__name__
+        return LLMResponse(
+            content=f"Error calling LLM: {detail}",
+            finish_reason="error",
+            error_status_code=error_status_code,
+            error_kind=error_kind,
+            error_type=str(raw_error_type) if raw_error_type is not None else None,
+            error_code=str(raw_error_code) if raw_error_code is not None else None,
+            error_should_retry=error_should_retry,
+        )
+
     @classmethod
     def _is_transient_error(cls, content: str | None) -> bool:
         err = (content or "").lower()
@@ -1226,7 +1283,7 @@ class LLMProvider(ABC):
             )
             raise
         except Exception as exc:
-            response = LLMResponse(content=f"Error calling LLM: {exc}", finish_reason="error")
+            response = self._error_response_from_exception(exc)
         return self._observe_llm_call(
             response,
             kwargs,
@@ -1368,7 +1425,7 @@ class LLMProvider(ABC):
             )
             raise
         except Exception as exc:
-            response = LLMResponse(content=f"Error calling LLM: {exc}", finish_reason="error")
+            response = self._error_response_from_exception(exc)
         return self._observe_llm_call(
             _attach_stream_timing(response),
             kwargs,

nanobot/providers/fallback_provider.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import replace
@@ -59,6 +60,7 @@ _AUTHENTICATION_ERROR_TOKENS = (
     "access_denied",
     "account_deactivated",
     "organization_deactivated",
+    "not logged in",
 )
 _NON_FALLBACK_ERROR_KINDS = frozenset({
     "content_filter",
@@ -428,7 +430,14 @@ class FallbackProvider(LLMProvider):
 
         if self._primary_available():
             primary_was_attempted = True
-            response = await call(self._primary, kwargs)
+            response, primary_exception = await self._call_provider(
+                call, self._primary, kwargs
+            )
+            if primary_exception is not None:
+                logger.warning(
+                    "Primary model '{}' raised {} before responding",
+                    primary_model, type(primary_exception).__name__,
+                )
             if response.finish_reason != "error":
                 self._primary_failures = 0
                 self._primary_tripped_at = None
@@ -457,7 +466,7 @@ class FallbackProvider(LLMProvider):
 
             if not self._should_fallback(response):
                 logger.warning(
-                    "Primary model '{}' returned non-fallbackable error: {}",
+                    "Primary model '{}' failed with non-fallbackable error: {}",
                     primary_model,
                     (response.content or "")[:120],
                 )
@@ -544,7 +553,14 @@ class FallbackProvider(LLMProvider):
                 fallback_kwargs.pop("reasoning_effort", None)
             else:
                 fallback_kwargs["reasoning_effort"] = fallback.reasoning_effort
-            fallback_response = await call(fallback_provider, fallback_kwargs)
+            fallback_response, fallback_exception = await self._call_provider(
+                call, fallback_provider, fallback_kwargs
+            )
+            if fallback_exception is not None:
+                logger.warning(
+                    "Fallback '{}' raised {}",
+                    fallback_model, type(fallback_exception).__name__,
+                )
 
             if fallback_response.finish_reason != "error":
                 # Do not publish a model switch merely because a fallback was
@@ -593,6 +609,26 @@ class FallbackProvider(LLMProvider):
             error_should_retry=True,
         )
 
+    @staticmethod
+    async def _call_provider(
+        call: Callable[[LLMProvider, dict[str, Any]], Awaitable[LLMResponse]],
+        provider: LLMProvider,
+        kwargs: dict[str, Any],
+    ) -> tuple[LLMResponse, Exception | None]:
+        """Turn provider exceptions into error responses without swallowing cancellation."""
+        try:
+            return await call(provider, kwargs), None
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            response = LLMProvider._error_response_from_exception(exc)
+            if response.error_kind is None and any(
+                token in (str(exc).strip() or type(exc).__name__).lower()
+                for token in _AUTHENTICATION_ERROR_TOKENS
+            ):
+                response.error_kind = "authentication"
+            return response, exc
+
     async def _notify_fallback_model(self, model: str) -> None:
         if self._fallback_model_observer is None:
             return

nanobot/providers/openai_codex_provider.py

@@ -111,6 +111,7 @@ class OpenAICodexProvider(LLMProvider):
             model=_strip_model_prefix(model),
         )
         session_id = provider_context.session_id if provider_context is not None else None
+        session_routing_key = _prompt_cache_key(session_id) if session_id else None
 
         body: dict[str, Any] = {
             "model": _strip_model_prefix(model),
@@ -122,8 +123,8 @@ class OpenAICodexProvider(LLMProvider):
             "tool_choice": tool_choice or "auto",
             "parallel_tool_calls": True,
         }
-        if session_id:
-            body["prompt_cache_key"] = _prompt_cache_key(session_id)
+        if session_routing_key:
+            body["prompt_cache_key"] = session_routing_key
         body["include"] = ["reasoning.encrypted_content"]
         reasoning_options = _build_reasoning_options(reasoning_effort)
         if replayed and "gpt-5.6" in _strip_model_prefix(model).lower():
@@ -136,13 +137,20 @@ class OpenAICodexProvider(LLMProvider):
         if self._extra_body:
             # Apply explicit provider overrides last, matching other provider backends.
             body.update(self._extra_body)
+        effective_cache_key = body.get("prompt_cache_key")
 
         stage = "oauth_token"
         native_compaction_applied = False
         native_compaction_state: ProviderConversationState | None = None
         try:
             token = await asyncio.to_thread(get_codex_token, proxy=self.proxy)
-            headers = _build_headers(cast(str, token.account_id), token.access)
+            headers = _build_headers(
+                cast(str, token.account_id),
+                token.access,
+                session_routing_key=(
+                    effective_cache_key if isinstance(effective_cache_key, str) else None
+                ),
+            )
 
             async def _send(
                 request_body: dict[str, Any],
@@ -416,8 +424,13 @@ def _build_reasoning_options(reasoning_effort: str | None) -> dict[str, str] | N
     return options
 
 
-def _build_headers(account_id: str, token: str) -> dict[str, str]:
-    return {
+def _build_headers(
+    account_id: str,
+    token: str,
+    *,
+    session_routing_key: str | None = None,
+) -> dict[str, str]:
+    headers = {
         "Authorization": f"Bearer {token}",
         "chatgpt-account-id": account_id,
         "OpenAI-Beta": "responses=experimental",
@@ -426,6 +439,9 @@ def _build_headers(account_id: str, token: str) -> dict[str, str]:
         "accept": "text/event-stream",
         "content-type": "application/json",
     }
+    if session_routing_key:
+        headers["session-id"] = session_routing_key
+    return headers
 
 
 class _CodexHTTPError(RuntimeError):

nanobot/sdk/streaming.py

@@ -158,15 +158,11 @@ class SDKStreamEmitter:
             resuming=resuming,
         ))
 
-    def close(self) -> None:
+    async def close(self) -> None:
         if self._closed:
             return
         self._closed = True
-        if self._queue.full():
-            with suppress(asyncio.QueueEmpty):
-                self._queue.get_nowait()
-        with suppress(asyncio.QueueFull):
-            self._queue.put_nowait(_STREAM_SENTINEL)
+        await self._queue.put(_STREAM_SENTINEL)
 
 
 class SDKStreamingHook(AgentHook):

nanobot/utils/tool_hints.py

@@ -104,6 +104,10 @@ def _fmt_known(tc: ToolCallRequest, fmt: ToolFormat, max_length: int = 40) -> st
         val = abbreviate_path(val, max_len=max_length)
     elif fmt[3]:  # is_command
         val = _abbreviate_command(val, max_len=max_length)
+    elif len(val) > max_length:
+        # Plain values (grep patterns, search queries, ...) have no path or
+        # command structure to fold, so fall back to a hard truncation.
+        val = val[:max_length - 1] + "\u2026"
     return fmt[1].format(val)
 
 

tests/agent/test_runner_fallback.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import asyncio
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import pytest
 from loguru import logger
 
@@ -347,6 +349,249 @@ class TestNoFallbackWhenPrimarySucceeds:
         factory.assert_not_called()
 
 
+class _RaisingProvider(LLMProvider):
+    """Provider whose chat/chat_stream raise, like an auth/setup failure."""
+
+    def __init__(self, name: str = "raiser", exc: BaseException | None = None):
+        super().__init__(provider_name=name)
+        self.name = name
+        self._exc = exc if exc is not None else RuntimeError("GitHub Copilot is not logged in.")
+
+    def get_default_model(self) -> str:
+        return f"{self.name}/model"
+
+    async def chat(self, **kwargs: Any) -> LLMResponse:
+        raise self._exc
+
+    async def chat_stream(self, **kwargs: Any) -> LLMResponse:
+        raise self._exc
+
+
+class _StreamingThenRaisingProvider(_RaisingProvider):
+    async def chat_stream(self, **kwargs: Any) -> LLMResponse:
+        on_content_delta = kwargs.get("on_content_delta")
+        if on_content_delta:
+            await on_content_delta("partial")
+        raise self._exc
+
+
+class TestFallbackWhenPrimaryRaises:
+    @pytest.mark.parametrize(
+        "exc",
+        [TimeoutError(), httpx.ReadTimeout(""), httpx.ConnectError(""), httpx.ReadError("")],
+        ids=["asyncio-timeout", "httpx-timeout", "connection", "httpx-network-error"],
+    )
+    @pytest.mark.asyncio
+    async def test_transient_exception_triggers_fallback(self, exc: Exception) -> None:
+        """Transient exceptions remain eligible even when their messages are empty."""
+        primary = _RaisingProvider("primary", exc)
+        fallback = _FakeProvider("fallback", _make_response("fallback ok"))
+        factory = MagicMock(return_value=fallback)
+
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        result = await fb.chat(messages=[{"role": "user", "content": "hi"}], model="primary-model")
+
+        assert result.content == "fallback ok"
+        assert result.finish_reason == "stop"
+        factory.assert_called_once_with(_fallback("fallback-a"))
+
+    @pytest.mark.asyncio
+    async def test_primary_exception_triggers_fallback(self) -> None:
+        """A primary whose chat() raises must not abort failover.
+
+        GitHubCopilotProvider.chat refreshes its token before the request and
+        raises when not logged in; the exception must be treated as a
+        fallbackable primary error, not swallow the whole fallback chain.
+        """
+        primary = _RaisingProvider("primary")
+        fallback = _FakeProvider("fallback", _make_response("fallback ok"))
+        factory = MagicMock(return_value=fallback)
+
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        result = await fb.chat(messages=[{"role": "user", "content": "hi"}], model="primary-model")
+        assert result.content == "fallback ok"
+        assert result.finish_reason == "stop"
+        factory.assert_called_once_with(_fallback("fallback-a"))
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("stream", [False, True], ids=["chat", "stream"])
+    async def test_retry_entry_point_preserves_transient_exception_metadata(
+        self,
+        stream: bool,
+    ) -> None:
+        """A retry wrapper must not hide an empty transient exception from failover."""
+        primary = _RaisingProvider("primary", TimeoutError())
+        fallback = _FakeProvider("fallback", _make_response("fallback ok"))
+        factory = MagicMock(return_value=fallback)
+
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        retry = fb.chat_stream_with_retry if stream else fb.chat_with_retry
+        result = await retry(messages=[{"role": "user", "content": "hi"}], model="primary-model")
+
+        assert result.content == "fallback ok"
+        assert result.finish_reason == "stop"
+        factory.assert_called_once_with(_fallback("fallback-a"))
+
+    @pytest.mark.asyncio
+    async def test_authentication_exception_message_is_classified(self) -> None:
+        primary = _RaisingProvider("primary")
+
+        response, exception = await FallbackProvider._call_provider(
+            lambda provider, kwargs: provider.chat(**kwargs),
+            primary,
+            {},
+        )
+
+        assert exception is primary._exc
+        assert response.error_kind == "authentication"
+
+    @pytest.mark.asyncio
+    async def test_non_string_exception_metadata_is_normalized_before_fallback(self) -> None:
+        """Provider exception metadata must be string-like before fallback consumes it."""
+
+        class NumericMetadataError(Exception):
+            error_type = 429
+            error_code = 429
+            status_code = 429
+
+        primary = _RaisingProvider("primary", NumericMetadataError("rate limited"))
+
+        response, exception = await FallbackProvider._call_provider(
+            lambda provider, kwargs: provider.chat(**kwargs),
+            primary,
+            {},
+        )
+
+        assert exception is primary._exc
+        assert response.error_type == "429"
+        assert response.error_code == "429"
+        assert FallbackProvider._should_fallback(response) is True
+
+    @pytest.mark.parametrize(
+        "exc",
+        [
+            ValueError("unexpected response shape"),
+            PermissionError("local provider file access failed"),
+        ],
+        ids=["value-error", "local-permission-error"],
+    )
+    @pytest.mark.asyncio
+    async def test_non_fallbackable_primary_exception_does_not_trigger_fallback(
+        self, exc: Exception,
+    ) -> None:
+        """An unrelated provider bug must not silently switch models."""
+        primary = _RaisingProvider("primary", exc)
+        factory = MagicMock(
+            return_value=_FakeProvider("fallback", _make_response("fallback"))
+        )
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        result = await fb.chat(
+            messages=[{"role": "user", "content": "hi"}],
+            model="primary-model",
+        )
+
+        assert result.finish_reason == "error"
+        assert result.content is not None
+        assert str(exc) in result.content
+        factory.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_fallback_exception_advances_to_next_fallback(self) -> None:
+        """A fallback whose chat() raises must advance to the next fallback."""
+        primary = _FakeProvider("primary", _error_response())
+        raising_fb = _RaisingProvider("fb1")
+        ok_fb = _FakeProvider("fb2", _make_response("second fallback ok"))
+        factory = MagicMock(side_effect=[raising_fb, ok_fb])
+
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a"), _fallback("fallback-b")],
+            provider_factory=factory,
+        )
+
+        result = await fb.chat(messages=[{"role": "user", "content": "hi"}], model="primary-model")
+        assert result.content == "second fallback ok"
+        assert result.finish_reason == "stop"
+
+    @pytest.mark.asyncio
+    async def test_primary_exception_stream_triggers_fallback(self) -> None:
+        """Streaming path: a raising primary still fails over when nothing streamed."""
+        primary = _RaisingProvider("primary")
+        fallback = _FakeProvider("fallback", _make_response("fallback ok"))
+        factory = MagicMock(return_value=fallback)
+
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        result = await fb.chat_stream(messages=[{"role": "user", "content": "hi"}], model="primary-model")
+        assert result.content == "fallback ok"
+        assert result.finish_reason == "stop"
+
+    @pytest.mark.asyncio
+    async def test_primary_exception_after_streaming_does_not_duplicate_output(self) -> None:
+        """Once content was emitted, an exception must not start a replacement stream."""
+        primary = _StreamingThenRaisingProvider("primary")
+        factory = MagicMock(return_value=_FakeProvider("fallback", _make_response("duplicate")))
+        deltas: list[str] = []
+
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        async def collect_delta(text: str) -> None:
+            deltas.append(text)
+
+        result = await fb.chat_stream(
+            messages=[{"role": "user", "content": "hi"}],
+            model="primary-model",
+            on_content_delta=collect_delta,
+        )
+
+        assert deltas == ["partial"]
+        assert result.finish_reason == "error"
+        factory.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_primary_cancellation_is_not_converted_to_failover(self) -> None:
+        """Task cancellation must retain asyncio cancellation semantics."""
+        primary = _RaisingProvider("primary", asyncio.CancelledError())
+        factory = MagicMock(return_value=_FakeProvider("fallback", _make_response("fallback")))
+        fb = FallbackProvider(
+            primary=primary,
+            fallback_presets=[_fallback("fallback-a")],
+            provider_factory=factory,
+        )
+
+        with pytest.raises(asyncio.CancelledError):
+            await fb.chat(messages=[{"role": "user", "content": "hi"}])
+
+        factory.assert_not_called()
+
 class TestFallbackOnPrimaryError:
     @pytest.mark.asyncio
     async def test_first_fallback_succeeds(self) -> None:

tests/agent/test_tool_hint.py

@@ -303,6 +303,43 @@ class TestToolHintMaxLength:
         long = _hint([_tc("list_dir", {"path": long_path})], max_length=120)
         assert len(long) > len(short)
 
+    def test_plain_value_tools_respect_max_length(self):
+        """Plain-value tools must truncate like the is_path/is_command branches.
+
+        grep, web_search, x_search and find_files carry no path or command
+        structure to fold, so their raw value used to reach the progress hint
+        untruncated: a 400-char search query produced a 400-char hint.
+        """
+        for name, key in (
+            ("grep", "pattern"),
+            ("web_search", "query"),
+            ("x_search", "query"),
+            ("find_files", "query"),
+        ):
+            short = _hint([_tc(name, {key: "x" * 400})], max_length=40)
+            long = _hint([_tc(name, {key: "x" * 400})], max_length=120)
+            assert len(long) > len(short), name
+            # Longest template is 'search X "{}"' — 12 chars of overhead.
+            assert len(short) <= 40 + 12, name
+            assert "\u2026" in short, name
+
+    def test_plain_value_short_value_untouched(self):
+        """Values inside the budget must not gain an ellipsis."""
+        result = _hint([_tc("grep", {"pattern": "TODO|FIXME"})], max_length=40)
+        assert result == 'grep "TODO|FIXME"'
+
+    def test_plain_value_exactly_at_max_length_untouched(self):
+        """A value exactly at max_length already fits."""
+        query = "a" * 40
+        result = _hint([_tc("web_search", {"query": query})], max_length=40)
+        assert result == f'search "{query}"'
+        assert "\u2026" not in result
+
+    def test_plain_value_one_over_max_length_truncates(self):
+        """One character over the budget truncates instead of overflowing."""
+        result = _hint([_tc("grep", {"pattern": "a" * 41})], max_length=40)
+        assert result == 'grep "' + "a" * 39 + "\u2026" + '"'
+
 
 class TestToolHintMalformedCalls:
     """Malformed tool calls must not crash hint formatting (see HKUDS/nanobot)."""

tests/providers/test_openai_codex_provider.py

@@ -265,6 +265,7 @@ async def test_codex_request_uses_configured_proxy(monkeypatch) -> None:
 @pytest.mark.asyncio
 async def test_codex_omits_prompt_cache_key_without_session_id(monkeypatch) -> None:
     bodies: list[dict[str, Any]] = []
+    headers_seen: list[dict[str, str]] = []
 
     _mock_codex_token(monkeypatch)
 
@@ -280,6 +281,7 @@ async def test_codex_omits_prompt_cache_key_without_session_id(monkeypatch) -> N
     ):
         _ = proxy, on_thinking_delta, on_tool_call_delta
         bodies.append(body)
+        headers_seen.append(headers)
         return provider_base.LLMResponse(content="ok")
 
     monkeypatch.setattr("nanobot.providers.openai_codex_provider._request_codex", fake_request)
@@ -294,16 +296,19 @@ async def test_codex_omits_prompt_cache_key_without_session_id(monkeypatch) -> N
     )
 
     assert "prompt_cache_key" not in bodies[0]
+    assert "session-id" not in headers_seen[0]
     assert "service_tier" not in bodies[0]
 
 
 @pytest.mark.asyncio
 async def test_codex_prompt_cache_key_prefers_stable_session_id(monkeypatch) -> None:
     bodies: list[dict[str, Any]] = []
+    headers_seen: list[dict[str, str]] = []
     _mock_codex_token(monkeypatch)
 
-    async def fake_request(_url, _headers, body, **_kwargs):
+    async def fake_request(_url, headers, body, **_kwargs):
         bodies.append(body)
+        headers_seen.append(headers)
         return provider_base.LLMResponse(content="ok")
 
     monkeypatch.setattr("nanobot.providers.openai_codex_provider._request_codex", fake_request)
@@ -326,15 +331,22 @@ async def test_codex_prompt_cache_key_prefers_stable_session_id(monkeypatch) ->
 
     assert bodies[0]["prompt_cache_key"] == bodies[1]["prompt_cache_key"]
     assert bodies[0]["prompt_cache_key"] != bodies[2]["prompt_cache_key"]
+    assert headers_seen[0]["session-id"] != "session-a"
+    assert headers_seen[2]["session-id"] != "session-b"
+    assert headers_seen[0]["session-id"] == bodies[0]["prompt_cache_key"]
+    assert headers_seen[1]["session-id"] == bodies[1]["prompt_cache_key"]
+    assert headers_seen[2]["session-id"] == bodies[2]["prompt_cache_key"]
 
 
 @pytest.mark.asyncio
 async def test_codex_provider_applies_extra_body_from_config(monkeypatch) -> None:
     bodies: list[dict[str, Any]] = []
+    headers_seen: list[dict[str, str]] = []
     _mock_codex_token(monkeypatch)
 
-    async def fake_request(_url, _headers, body, **_kwargs):
+    async def fake_request(_url, headers, body, **_kwargs):
         bodies.append(body)
+        headers_seen.append(headers)
         return provider_base.LLMResponse(content="ok")
 
     monkeypatch.setattr("nanobot.providers.openai_codex_provider._request_codex", fake_request)
@@ -347,7 +359,10 @@ async def test_codex_provider_applies_extra_body_from_config(monkeypatch) -> Non
         },
         "providers": {
             "openaiCodex": {
-                "extraBody": {"service_tier": "priority"},
+                "extraBody": {
+                    "service_tier": "priority",
+                    "prompt_cache_key": "explicit-cache-key",
+                },
             },
         },
     })
@@ -357,6 +372,8 @@ async def test_codex_provider_applies_extra_body_from_config(monkeypatch) -> Non
 
     assert response.content == "ok"
     assert bodies[0]["service_tier"] == "priority"
+    assert bodies[0]["prompt_cache_key"] == "explicit-cache-key"
+    assert headers_seen[0]["session-id"] == "explicit-cache-key"
 
 
 @pytest.mark.asyncio

tests/test_sdk_streaming.py

@@ -0,0 +1,24 @@
+"""Tests for SDK streaming primitives."""
+
+import asyncio
+
+import pytest
+
+from nanobot.sdk.streaming import SDKStreamEmitter
+from nanobot.sdk.types import STREAM_EVENT_TEXT_DELTA, StreamEvent
+
+
+@pytest.mark.asyncio
+async def test_close_preserves_events_when_queue_is_full():
+    queue: asyncio.Queue[StreamEvent | object] = asyncio.Queue(maxsize=1)
+    emitter = SDKStreamEmitter(queue)
+    event = StreamEvent(type=STREAM_EVENT_TEXT_DELTA, delta="kept")
+    await emitter.emit(event)
+
+    close_task = asyncio.create_task(emitter.close())
+    await asyncio.sleep(0)
+
+    assert not close_task.done()
+    assert queue.get_nowait() is event
+    await close_task
+    assert queue.qsize() == 1

tests/tools/test_session_messages_tool.py

@@ -2,6 +2,7 @@ import asyncio
 import json
 from pathlib import Path
 from typing import Callable
+from unittest.mock import patch
 
 import pytest
 
@@ -277,6 +278,45 @@ async def test_reply_timeout_injects_a_user_input_back_into_the_source(
     assert timeout.content == f"No reply from @{target.name} after 5 seconds."
 
 
+@pytest.mark.asyncio
+async def test_reply_timeout_observes_background_delivery_failure(
+    tmp_path: Path,
+) -> None:
+    sessions = SessionManager(tmp_path)
+    _persist(sessions, "websocket:source", "websocket:target")
+    bus = MessageBus()
+    scheduler = _Scheduler()
+    tool = SendSessionMessageTool(
+        sessions=sessions,
+        bus=bus,
+        schedule_later=scheduler,
+    )
+    target = _handle(sessions, "websocket:target")
+
+    await tool.enqueue(
+        source_session_key="websocket:source",
+        target_handle=target.name,
+        content="Question",
+        expect_reply=True,
+        reply_timeout_seconds=5,
+    )
+    await bus.consume_inbound()
+
+    async def fail_publish(_message) -> None:
+        raise RuntimeError("queue unavailable")
+
+    bus.publish_inbound = fail_publish
+    with patch("nanobot.agent.tools.session_messages.logger") as logger:
+        scheduler.calls[0][1].fire()
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+    assert tool._expiry_tasks == set()
+    logger.exception.assert_called_once_with(
+        "Session reply timeout delivery failed",
+    )
+
+
 @pytest.mark.asyncio
 async def test_reverse_message_cancels_the_pending_reply_timeout(
     tmp_path: Path,

webui/src/hooks/useNanobotStream.ts

@@ -643,6 +643,28 @@ export function useNanobotStream(
     return () => document.removeEventListener("visibilitychange", flushOnReturn);
   }, [flushPendingStreamEvents]);
 
+  useEffect(() => {
+    if (!chatId) return;
+    return client.onRunStatus((runChatId, startedAt) => {
+      if (runChatId !== chatId) return;
+      if (startedAt !== null) {
+        setRunStartedAt(startedAt);
+        setIsStreaming(true);
+        return;
+      }
+      flushPendingStreamEvents();
+      buffer.current = null;
+      activeAssistantRef.current = null;
+      closedAssistantStreamIdsRef.current.clear();
+      clearActivitySegment();
+      setMessages((prev) => prev.map((message) => (
+        message.isStreaming ? { ...message, isStreaming: false } : message
+      )));
+      setRunStartedAt(null);
+      setIsStreaming(false);
+    });
+  }, [chatId, client, clearActivitySegment, flushPendingStreamEvents]);
+
   // Reset local state when switching chats. Do not reset on every
   // ``initialMessages`` update: a brand-new chat can receive an empty/404
   // history response after the optimistic first message has already rendered.

webui/src/tests/thread-shell.test.tsx

@@ -21,6 +21,7 @@ function makeClient() {
     (modelName: string | null, modelPreset?: string | null) => void
   >();
   const sessionUpdateHandlers = new Set<(chatId: string, scope?: string) => void>();
+  const runStatusHandlers = new Set<(chatId: string, startedAt: number | null) => void>();
   const runStartedAtByChatId = new Map<string, number>();
   const runGenerationByChatId = new Map<string, number>();
   const latestRunTurnIdByChatId = new Map<string, string>();
@@ -98,6 +99,13 @@ function makeClient() {
         statusHandlers.delete(handler);
       };
     },
+    onRunStatus: (handler: (chatId: string, startedAt: number | null) => void) => {
+      runStatusHandlers.add(handler);
+      for (const [chatId, startedAt] of runStartedAtByChatId) handler(chatId, startedAt);
+      return () => {
+        runStatusHandlers.delete(handler);
+      };
+    },
     onRuntimeModelUpdate: (
       handler: (modelName: string | null, modelPreset?: string | null) => void,
     ) => {
@@ -157,11 +165,13 @@ function makeClient() {
       ) {
         advanceRunGeneration(chatId, ev.turn_id);
         runStartedAtByChatId.set(chatId, ev.started_at);
+        for (const h of runStatusHandlers) h(chatId, ev.started_at);
       } else if (
         (ev.event === "goal_status" && ev.status === "idle")
         || ev.event === "turn_end"
       ) {
         runStartedAtByChatId.delete(chatId);
+        for (const h of runStatusHandlers) h(chatId, null);
       }
       if (ev.event === "goal_state") {
         goalStateByChatId.set(chatId, ev.goal_state);

webui/src/tests/useNanobotStream.test.tsx

@@ -70,6 +70,7 @@ function normalizeProjection(messages: UIMessage[]): Array<Record<string, unknow
 function fakeClient() {
   const handlers = new Map<string, Set<(ev: InboundEvent) => void>>();
   const statusHandlers = new Set<(status: ConnectionStatus) => void>();
+  const runStatusHandlers = new Set<(chatId: string, startedAt: number | null) => void>();
   const errorHandlers = new Set<(error: StreamError) => void>();
   const runStartedAtByChatId = new Map<string, number>();
   const unsettledRunByChatId = new Map<string, boolean>();
@@ -111,6 +112,11 @@ function fakeClient() {
         handler(status);
         return () => statusHandlers.delete(handler);
       },
+      onRunStatus(handler: (chatId: string, startedAt: number | null) => void) {
+        runStatusHandlers.add(handler);
+        for (const [chatId, startedAt] of runStartedAtByChatId) handler(chatId, startedAt);
+        return () => runStatusHandlers.delete(handler);
+      },
       onError(handler: (error: StreamError) => void) {
         errorHandlers.add(handler);
         return () => errorHandlers.delete(handler);
@@ -154,6 +160,11 @@ function fakeClient() {
       status = nextStatus;
       statusHandlers.forEach((handler) => handler(status));
     },
+    emitRunStatus(chatId: string, startedAt: number | null) {
+      if (startedAt === null) runStartedAtByChatId.delete(chatId);
+      else runStartedAtByChatId.set(chatId, startedAt);
+      runStatusHandlers.forEach((handler) => handler(chatId, startedAt));
+    },
     emitError(error: StreamError) {
       errorHandlers.forEach((handler) => handler(error));
     },
@@ -327,6 +338,39 @@ describe("useNanobotStream", () => {
     });
   });
 
+  it("clears stale stream state when the transport resets a run", async () => {
+    const fake = fakeClient();
+    const { result } = renderHook(
+      () => useNanobotStream("chat-reconnect-reset", EMPTY_MESSAGES),
+      { wrapper: wrap(fake.client) },
+    );
+
+    act(() => {
+      fake.emit("chat-reconnect-reset", {
+        event: "goal_status",
+        chat_id: "chat-reconnect-reset",
+        status: "running",
+        started_at: 1_700,
+      });
+      fake.emit("chat-reconnect-reset", {
+        event: "delta",
+        chat_id: "chat-reconnect-reset",
+        text: "partial",
+      });
+    });
+    await flushStreamFrame();
+    expect(result.current.isStreaming).toBe(true);
+
+    act(() => fake.emitRunStatus("chat-reconnect-reset", null));
+
+    expect(result.current.runStartedAt).toBeNull();
+    expect(result.current.isStreaming).toBe(false);
+    expect(result.current.messages[0]).toMatchObject({
+      content: "partial",
+      isStreaming: false,
+    });
+  });
+
   it("flushes pending delta text before turn_end finalizes the turn", () => {
     const fake = fakeClient();
     const { result } = renderHook(() => useNanobotStream("chat-flush", EMPTY_MESSAGES), {