fix(channels): harden message split boundaries

Signed-off-by: chengyongru <2755839590@qq.com>

docs/chat-apps.md

@@ -382,7 +382,7 @@ nanobot plugins enable matrix
 | `groupAllowFrom` | Room allowlist (used when policy is `allowlist`). |
 | `allowRoomMentions` | Accept `@room` mentions in mention mode. |
 | `e2eeEnabled` | E2EE support (default `true`). Set `false` for plaintext-only. |
-| `sasVerification` | Auto-complete SAS device verification requests from allowed users (default `false`). Useful for Element X, which does not expose manual trust for third-party devices. |
+| `sasVerification` | Complete Element-initiated SAS device verification for allowed users (default `false`). This does not add cross-signing, clear Element's cross-signing trust warning, or let the bot initiate verification. |
 | `maxMediaBytes` | Max attachment size (default `20MB`). Set `0` to block all media. |
 
 

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/matrix/runtime.py

@@ -47,6 +47,8 @@ try:
         SyncError,
         SyncResponse,
         ToDeviceError,
+        ToDeviceMessage,
+        UnknownToDeviceEvent,
         UploadError,
     )
     from nio.crypto.attachments import decrypt_attachment
@@ -75,6 +77,10 @@ _ATTACH_FAILED = "[attachment: {} - download failed]"
 _ATTACH_UPLOAD_FAILED = "[attachment: {} - upload failed]"
 _DEFAULT_ATTACH_NAME = "attachment"
 _MSGTYPE_MAP = {"m.image": "image", "m.audio": "audio", "m.video": "video", "m.file": "file"}
+_SAS_METHOD = "m.sas.v1"
+_SAS_REQUEST_MAX_AGE_MS = 10 * 60 * 1000
+_SAS_REQUEST_MAX_FUTURE_MS = 5 * 60 * 1000
+_SAS_REQUEST_MAX_PENDING = 256
 
 MATRIX_MEDIA_EVENT_FILTER = (RoomMessageMedia, RoomEncryptedMedia)
 MatrixMediaEvent: TypeAlias = RoomMessageMedia | RoomEncryptedMedia
@@ -200,6 +206,15 @@ class _StreamBuf:
     event_id: str | None = None
     last_edit: float = 0.0
 
+
+@dataclass(frozen=True)
+class _SasVerificationRequest:
+    """An allowed Element verification request awaiting SAS completion."""
+
+    sender: str
+    device_id: str
+    timestamp_ms: int
+
 def _render_markdown_html(text: str) -> str | None:
     """Render markdown to sanitized HTML; returns None for plain text."""
     try:
@@ -321,6 +336,7 @@ class MatrixChannel(BaseChannel):
         self._server_upload_limit_bytes: int | None = None
         self._server_upload_limit_checked = False
         self._stream_bufs: dict[str, _StreamBuf] = {}
+        self._sas_verification_requests: dict[str, _SasVerificationRequest] = {}
         self._started_at_ms: int = 0
         self._media_download_semaphore = asyncio.Semaphore(
             max(1, int(self.config.max_concurrent_media_downloads))
@@ -642,7 +658,7 @@ class MatrixChannel(BaseChannel):
             stream_end = False
         if stream_end:
             stream_key = _matrix_stream_key(chat_id, stream_id)
-            buf = self._stream_bufs.pop(stream_key, None)
+            buf = self._stream_bufs.get(stream_key)
             if not buf or not buf.event_id or not buf.text:
                 return
 
@@ -653,14 +669,19 @@ class MatrixChannel(BaseChannel):
                 buf.event_id,
                 thread_relates_to=relates_to,
             )
-            await self._send_room_content(chat_id, content)
+            response = await self._send_room_content(chat_id, content)
+            if isinstance(response, RoomSendError):
+                raise RuntimeError(f"Matrix stream was not delivered: {response}")
+            self._stream_bufs.pop(stream_key, None)
             return
 
         stream_key = _matrix_stream_key(chat_id, stream_id)
         buf = self._stream_bufs.get(stream_key)
+        created_buf = buf is None
         if buf is None:
             buf = _StreamBuf()
             self._stream_bufs[stream_key] = buf
+        previous_text = buf.text
         buf.text += delta
 
         if not buf.text.strip():
@@ -676,13 +697,19 @@ class MatrixChannel(BaseChannel):
                     thread_relates_to=relates_to,
                 )
                 response = await self._send_room_content(chat_id, content)
+                if isinstance(response, RoomSendError):
+                    raise RuntimeError(f"Matrix stream was not delivered: {response}")
                 buf.last_edit = now
                 if not buf.event_id:
                     # we are editing the same message all the time, so only the first time the event id needs to be set
                     buf.event_id = cast(RoomSendResponse, response).event_id
             except Exception:
+                buf.text = previous_text
+                if created_buf:
+                    self._stream_bufs.pop(stream_key, None)
                 self.logger.error("Stream send/edit failed for chat_id={}", chat_id, exc_info=True)
                 await self._stop_typing_keepalive(chat_id, clear_typing=True)
+                raise
 
 
     def _register_event_callbacks(self) -> None:
@@ -696,7 +723,7 @@ class MatrixChannel(BaseChannel):
             client = self._callback_registrar()
             client.add_to_device_callback(
                 self._on_key_verification_event,
-                (KeyVerificationEvent,),
+                (KeyVerificationEvent, UnknownToDeviceEvent),
             )
 
     def _register_response_callbacks(self) -> None:
@@ -709,7 +736,10 @@ class MatrixChannel(BaseChannel):
     def _is_sas_sender_allowed(self, sender: str) -> bool:
         return bool(sender and self.is_allowed(sender))
 
-    async def _on_key_verification_event(self, event: KeyVerificationEvent) -> None:
+    async def _on_key_verification_event(
+        self,
+        event: KeyVerificationEvent | UnknownToDeviceEvent,
+    ) -> None:
         try:
             await self._handle_key_verification_event(event)
         except asyncio.CancelledError:
@@ -717,18 +747,173 @@ class MatrixChannel(BaseChannel):
         except Exception:
             self.logger.exception("Matrix SAS verification handling failed")
 
-    async def _handle_key_verification_event(self, event: KeyVerificationEvent) -> None:
+    @staticmethod
+    def _unknown_verification_content(
+        event: UnknownToDeviceEvent,
+    ) -> tuple[str, dict[str, object]] | None:
+        event_type = event.type
+        source = event.source
+        content = source.get("content")
+        if not isinstance(content, dict):
+            return None
+        return event_type, cast(dict[str, object], content)
+
+    @staticmethod
+    def _content_string(content: dict[str, object], key: str) -> str:
+        value = content.get(key)
+        return value if isinstance(value, str) else ""
+
+    def _prune_sas_verification_requests(self, now_ms: int) -> None:
+        oldest_allowed = now_ms - _SAS_REQUEST_MAX_AGE_MS
+        self._sas_verification_requests = {
+            transaction_id: request
+            for transaction_id, request in self._sas_verification_requests.items()
+            if request.timestamp_ms >= oldest_allowed
+        }
+
+    def _remember_sas_verification_request(
+        self,
+        transaction_id: str,
+        request: _SasVerificationRequest,
+    ) -> None:
+        self._sas_verification_requests[transaction_id] = request
+        while len(self._sas_verification_requests) > _SAS_REQUEST_MAX_PENDING:
+            oldest_transaction_id = next(iter(self._sas_verification_requests))
+            self._sas_verification_requests.pop(oldest_transaction_id)
+
+    async def _send_sas_control_message(
+        self,
+        *,
+        event_type: str,
+        sender: str,
+        device_id: str,
+        content: dict[str, object],
+    ) -> bool:
+        if not self.client:
+            return False
+        response = await self.client.to_device(
+            ToDeviceMessage(
+                type=event_type,
+                recipient=sender,
+                recipient_device=device_id,
+                content=content,
+            )
+        )
+        if isinstance(response, ToDeviceError):
+            self.logger.warning("Matrix SAS {} failed for {}: {}", event_type, sender, response)
+            return False
+        return True
+
+    async def _handle_unknown_verification_event(
+        self,
+        event: UnknownToDeviceEvent,
+        sender: str,
+    ) -> None:
+        parsed = self._unknown_verification_content(event)
+        if parsed is None:
+            return
+        event_type, content = parsed
+        if event_type not in {
+            "m.key.verification.request",
+            "m.key.verification.ready",
+            "m.key.verification.done",
+        }:
+            return
+
+        transaction_id = self._content_string(content, "transaction_id")
+        if not transaction_id:
+            return
+
+        if event_type == "m.key.verification.request":
+            from_device = self._content_string(content, "from_device")
+            methods = content.get("methods")
+            timestamp = content.get("timestamp")
+            if (
+                not from_device
+                or not isinstance(methods, list)
+                or _SAS_METHOD not in methods
+                or isinstance(timestamp, bool)
+                or not isinstance(timestamp, int)
+            ):
+                return
+
+            now_ms = int(time.time() * 1000)
+            if not (
+                now_ms - _SAS_REQUEST_MAX_AGE_MS
+                <= timestamp
+                <= now_ms + _SAS_REQUEST_MAX_FUTURE_MS
+            ):
+                self.logger.info("Ignoring expired Matrix SAS request from {}", sender)
+                return
+
+            self._prune_sas_verification_requests(now_ms)
+            request = _SasVerificationRequest(sender, from_device, timestamp)
+            existing = self._sas_verification_requests.get(transaction_id)
+            if existing is not None and existing != request:
+                self.logger.warning(
+                    "Ignoring conflicting Matrix SAS transaction {} from {}",
+                    transaction_id,
+                    sender,
+                )
+                return
+
+            own_device = str(self.client.device_id or "") if self.client else ""
+            if not own_device:
+                return
+            sent = await self._send_sas_control_message(
+                event_type="m.key.verification.ready",
+                sender=sender,
+                device_id=from_device,
+                content={
+                    "from_device": own_device,
+                    "methods": [_SAS_METHOD],
+                    "transaction_id": transaction_id,
+                },
+            )
+            if sent:
+                self._remember_sas_verification_request(transaction_id, request)
+            return
+
+        if event_type == "m.key.verification.done":
+            request = self._sas_verification_requests.get(transaction_id)
+            if request is not None and request.sender == sender:
+                self._sas_verification_requests.pop(transaction_id, None)
+                self.logger.info("Matrix SAS verification finished with {}", sender)
+
+        # Ready is deliberately ignored: this channel does not initiate verification.
+
+    async def _handle_key_verification_event(
+        self,
+        event: KeyVerificationEvent | UnknownToDeviceEvent,
+    ) -> None:
         if not (self.config.e2ee_enabled and self.config.sas_verification):
             return
         if not self.client:
             return
 
         sender = str(getattr(event, "sender", "") or "")
+        if not self._is_sas_sender_allowed(sender):
+            return
+
+        if isinstance(event, UnknownToDeviceEvent):
+            await self._handle_unknown_verification_event(event, sender)
+            return
+
         transaction_id = str(getattr(event, "transaction_id", "") or "")
-        if not transaction_id or not self._is_sas_sender_allowed(sender):
+        if not transaction_id:
             return
 
         if isinstance(event, KeyVerificationStart):
+            request = self._sas_verification_requests.get(transaction_id)
+            from_device = str(getattr(event, "from_device", "") or "")
+            if request is not None and (
+                request.sender != sender or request.device_id != from_device
+            ):
+                self.logger.warning(
+                    "Ignoring Matrix SAS start for transaction {} from unexpected device",
+                    transaction_id,
+                )
+                return
             if "emoji" not in (getattr(event, "short_authentication_string", None) or []):
                 self.logger.info(
                     "Ignoring Matrix SAS verification from {} without emoji support",
@@ -756,9 +941,27 @@ class MatrixChannel(BaseChannel):
             sas = getattr(self.client, "key_verifications", {}).get(transaction_id)
             if sas is not None and getattr(sas, "verified", False):
                 self.logger.info("Matrix SAS verification completed for {}", sender)
+                request = self._sas_verification_requests.get(transaction_id)
+                other_device = str(getattr(getattr(sas, "other_olm_device", None), "id", ""))
+                if (
+                    request is not None
+                    and request.sender == sender
+                    and request.device_id == other_device
+                ):
+                    sent = await self._send_sas_control_message(
+                        event_type="m.key.verification.done",
+                        sender=sender,
+                        device_id=request.device_id,
+                        content={"transaction_id": transaction_id},
+                    )
+                    if sent:
+                        self._sas_verification_requests.pop(transaction_id, None)
             return
 
         if isinstance(event, KeyVerificationCancel):
+            request = self._sas_verification_requests.get(transaction_id)
+            if request is not None and request.sender == sender:
+                self._sas_verification_requests.pop(transaction_id, None)
             self.logger.info(
                 "Matrix SAS verification cancelled by {}: {}",
                 sender,

nanobot/channels/matrix/tests/test_matrix_channel.py

@@ -220,10 +220,11 @@ class _FakeAsyncClient:
 
 
 class _FakeSas:
-    def __init__(self, *, verified: bool = False) -> None:
+    def __init__(self, *, verified: bool = False, device_id: str = "ALICEDEVICE") -> None:
         self.share_key_called = False
         self.get_mac_called = False
         self.verified = verified
+        self.other_olm_device = SimpleNamespace(id=device_id)
 
     def share_key(self):
         self.share_key_called = True
@@ -240,10 +241,12 @@ class _FakeKeyVerificationStart:
         *,
         sender: str = "@alice:matrix.org",
         transaction_id: str = "tx1",
+        from_device: str = "ALICEDEVICE",
         short_authentication_string: list[str] | None = None,
     ) -> None:
         self.sender = sender
         self.transaction_id = transaction_id
+        self.from_device = from_device
         self.short_authentication_string = short_authentication_string or ["emoji"]
 
 
@@ -275,6 +278,18 @@ def _patch_key_verification_events(monkeypatch) -> None:
     monkeypatch.setattr(matrix_module, "KeyVerificationMac", _FakeKeyVerificationMac)
 
 
+def _unknown_verification_event(
+    event_type: str,
+    *,
+    sender: str = "@alice:matrix.org",
+    transaction_id: str = "tx1",
+    **content: object,
+):
+    event_content = {"transaction_id": transaction_id, **content}
+    source = {"type": event_type, "sender": sender, "content": event_content}
+    return matrix_module.UnknownToDeviceEvent(source, sender, event_type)
+
+
 def _make_config(**kwargs) -> MatrixConfig:
     kwargs.setdefault("allow_from", ["*"])
     return MatrixConfig(
@@ -345,7 +360,10 @@ def test_register_to_device_callbacks_when_sas_verification_enabled() -> None:
     channel._register_to_device_callbacks()
 
     assert client.to_device_callbacks == [
-        (channel._on_key_verification_event, (matrix_module.KeyVerificationEvent,))
+        (
+            channel._on_key_verification_event,
+            (matrix_module.KeyVerificationEvent, matrix_module.UnknownToDeviceEvent),
+        )
     ]
 
 
@@ -425,6 +443,163 @@ async def test_sas_verification_ignores_when_disabled(monkeypatch) -> None:
     assert client.to_device_calls == []
 
 
+@pytest.mark.asyncio
+async def test_sas_verification_request_sends_ready_to_allowed_device(monkeypatch) -> None:
+    monkeypatch.setattr(matrix_module.time, "time", lambda: 1_000.0)
+    channel = MatrixChannel(
+        _make_config(
+            allow_from=["@alice:matrix.org"],
+            e2ee_enabled=True,
+            sas_verification=True,
+        ),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    client.device_id = "BOTDEVICE"
+    channel.client = client
+
+    event = _unknown_verification_event(
+        "m.key.verification.request",
+        from_device="ALICEDEVICE",
+        methods=["m.sas.v1"],
+        timestamp=1_000_000,
+    )
+    await channel._handle_key_verification_event(event)
+
+    assert len(client.to_device_calls) == 1
+    ready = client.to_device_calls[0]
+    assert ready.type == "m.key.verification.ready"
+    assert ready.recipient == "@alice:matrix.org"
+    assert ready.recipient_device == "ALICEDEVICE"
+    assert ready.content == {
+        "from_device": "BOTDEVICE",
+        "methods": ["m.sas.v1"],
+        "transaction_id": "tx1",
+    }
+    assert channel._sas_verification_requests["tx1"].device_id == "ALICEDEVICE"
+
+
+@pytest.mark.asyncio
+async def test_sas_verification_requests_are_bounded(monkeypatch) -> None:
+    monkeypatch.setattr(matrix_module.time, "time", lambda: 1_000.0)
+    channel = MatrixChannel(
+        _make_config(
+            allow_from=["@alice:matrix.org"],
+            e2ee_enabled=True,
+            sas_verification=True,
+        ),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    client.device_id = "BOTDEVICE"
+    channel.client = client
+
+    for index in range(matrix_module._SAS_REQUEST_MAX_PENDING + 1):
+        await channel._handle_key_verification_event(_unknown_verification_event(
+            "m.key.verification.request",
+            transaction_id=f"tx-{index}",
+            from_device="ALICEDEVICE",
+            methods=["m.sas.v1"],
+            timestamp=1_000_000,
+        ))
+
+    assert len(channel._sas_verification_requests) == matrix_module._SAS_REQUEST_MAX_PENDING
+    assert "tx-0" not in channel._sas_verification_requests
+    assert f"tx-{matrix_module._SAS_REQUEST_MAX_PENDING}" in channel._sas_verification_requests
+
+
+@pytest.mark.asyncio
+async def test_sas_verification_start_must_match_requested_device(monkeypatch) -> None:
+    _patch_key_verification_events(monkeypatch)
+    monkeypatch.setattr(matrix_module.time, "time", lambda: 1_000.0)
+    channel = MatrixChannel(
+        _make_config(
+            allow_from=["@alice:matrix.org"],
+            e2ee_enabled=True,
+            sas_verification=True,
+        ),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    client.device_id = "BOTDEVICE"
+    channel.client = client
+
+    await channel._handle_key_verification_event(_unknown_verification_event(
+        "m.key.verification.request",
+        from_device="ALICEDEVICE",
+        methods=["m.sas.v1"],
+        timestamp=1_000_000,
+    ))
+    await channel._handle_key_verification_event(
+        _FakeKeyVerificationStart(from_device="OTHERDEVICE")
+    )
+
+    assert client.accept_key_verification_calls == []
+    assert channel._sas_verification_requests["tx1"].device_id == "ALICEDEVICE"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("sender", "methods", "timestamp"),
+    [
+        ("@mallory:matrix.org", ["m.sas.v1"], 1_000_000),
+        ("@alice:matrix.org", ["m.qr_code.scan.v1"], 1_000_000),
+        ("@alice:matrix.org", ["m.sas.v1"], 1),
+        ("@alice:matrix.org", ["m.sas.v1"], 2_000_000),
+    ],
+)
+async def test_sas_verification_request_rejects_untrusted_or_invalid_input(
+    monkeypatch,
+    sender: str,
+    methods: list[str],
+    timestamp: int,
+) -> None:
+    monkeypatch.setattr(matrix_module.time, "time", lambda: 1_000.0)
+    channel = MatrixChannel(
+        _make_config(
+            allow_from=["@alice:matrix.org"],
+            e2ee_enabled=True,
+            sas_verification=True,
+        ),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    client.device_id = "BOTDEVICE"
+    channel.client = client
+
+    event = _unknown_verification_event(
+        "m.key.verification.request",
+        sender=sender,
+        from_device="ALICEDEVICE",
+        methods=methods,
+        timestamp=timestamp,
+    )
+    await channel._handle_key_verification_event(event)
+
+    assert client.to_device_calls == []
+    assert channel._sas_verification_requests == {}
+
+
+@pytest.mark.asyncio
+async def test_sas_verification_ready_is_ignored_without_bot_initiated_flow() -> None:
+    channel = MatrixChannel(
+        _make_config(allow_from=["@alice:matrix.org"], sas_verification=True),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    channel.client = client
+
+    event = _unknown_verification_event(
+        "m.key.verification.ready",
+        from_device="ALICEDEVICE",
+        methods=["m.sas.v1"],
+    )
+    await channel._handle_key_verification_event(event)
+
+    assert client.to_device_calls == []
+    assert channel._sas_verification_requests == {}
+
+
 @pytest.mark.asyncio
 async def test_sas_verification_key_confirms_allowed_sender(monkeypatch) -> None:
     _patch_key_verification_events(monkeypatch)
@@ -464,6 +639,74 @@ async def test_sas_verification_mac_does_not_resend_mac(monkeypatch) -> None:
     assert client.to_device_calls == []
 
 
+@pytest.mark.asyncio
+async def test_sas_verification_mac_sends_done_for_element_request(monkeypatch) -> None:
+    _patch_key_verification_events(monkeypatch)
+    monkeypatch.setattr(matrix_module.time, "time", lambda: 1_000.0)
+    channel = MatrixChannel(
+        _make_config(
+            allow_from=["@alice:matrix.org"],
+            e2ee_enabled=True,
+            sas_verification=True,
+        ),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    client.device_id = "BOTDEVICE"
+    channel.client = client
+
+    request = _unknown_verification_event(
+        "m.key.verification.request",
+        from_device="ALICEDEVICE",
+        methods=["m.sas.v1"],
+        timestamp=1_000_000,
+    )
+    await channel._handle_key_verification_event(request)
+    client.key_verifications["tx1"] = _FakeSas(verified=True)
+
+    await channel._handle_key_verification_event(_FakeKeyVerificationMac())
+
+    assert [message.type for message in client.to_device_calls] == [
+        "m.key.verification.ready",
+        "m.key.verification.done",
+    ]
+    done = client.to_device_calls[1]
+    assert done.recipient == "@alice:matrix.org"
+    assert done.recipient_device == "ALICEDEVICE"
+    assert done.content == {"transaction_id": "tx1"}
+    assert channel._sas_verification_requests == {}
+
+
+@pytest.mark.asyncio
+async def test_sas_verification_done_clears_matching_request(monkeypatch) -> None:
+    monkeypatch.setattr(matrix_module.time, "time", lambda: 1_000.0)
+    channel = MatrixChannel(
+        _make_config(
+            allow_from=["@alice:matrix.org"],
+            e2ee_enabled=True,
+            sas_verification=True,
+        ),
+        MessageBus(),
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    client.device_id = "BOTDEVICE"
+    channel.client = client
+
+    request = _unknown_verification_event(
+        "m.key.verification.request",
+        from_device="ALICEDEVICE",
+        methods=["m.sas.v1"],
+        timestamp=1_000_000,
+    )
+    await channel._handle_key_verification_event(request)
+
+    done = _unknown_verification_event("m.key.verification.done")
+    await channel._handle_key_verification_event(done)
+
+    assert channel._sas_verification_requests == {}
+    assert len(client.to_device_calls) == 1
+
+
 def test_media_event_filter_does_not_match_text_events() -> None:
     assert not issubclass(matrix_module.RoomMessageText, matrix_module.MATRIX_MEDIA_EVENT_FILTER)
 
@@ -2240,7 +2483,7 @@ async def test_send_delta_stream_end_noop_when_buffer_missing() -> None:
 
 
 @pytest.mark.asyncio
-async def test_send_delta_on_error_stops_typing(monkeypatch) -> None:
+async def test_send_delta_on_error_restores_buffer_and_raises(monkeypatch) -> None:
     channel = MatrixChannel(_make_config(), MessageBus())
     channel.logger = MagicMock()
     client = _FakeAsyncClient("", "", "", None)
@@ -2250,10 +2493,14 @@ async def test_send_delta_on_error_stops_typing(monkeypatch) -> None:
     now = 100.0
     monkeypatch.setattr(channel, "monotonic_time", lambda: now)
 
-    await channel.send_delta("!room:matrix.org", "Hello", {"room_id": "!room:matrix.org"})
+    with pytest.raises(RuntimeError, match="send failed"):
+        await channel.send_delta(
+            "!room:matrix.org",
+            "Hello",
+            {"room_id": "!room:matrix.org"},
+        )
 
-    assert "!room:matrix.org" in channel._stream_bufs
+    assert "!room:matrix.org" not in channel._stream_bufs
-    assert channel._stream_bufs["!room:matrix.org"].text == "Hello"
     assert len(client.room_send_calls) == 1
 
     assert len(client.typing_calls) == 1
@@ -2261,6 +2508,52 @@ async def test_send_delta_on_error_stops_typing(monkeypatch) -> None:
         "Stream send/edit failed for chat_id={}", "!room:matrix.org", exc_info=True
     )
 
+    client.raise_on_send = False
+    await channel.send_delta("!room:matrix.org", "Hello")
+
+    assert channel._stream_bufs["!room:matrix.org"].text == "Hello"
+
+
+@pytest.mark.asyncio
+async def test_send_delta_raises_when_room_send_returns_error(monkeypatch) -> None:
+    class _FakeRoomSendError:
+        def __str__(self) -> str:
+            return "temporary homeserver failure"
+
+    channel = MatrixChannel(_make_config(), MessageBus())
+    client = _FakeAsyncClient("", "", "", None)
+    client.room_send_response = _FakeRoomSendError()
+    channel.client = client
+    monkeypatch.setattr(matrix_module, "RoomSendError", _FakeRoomSendError)
+
+    with pytest.raises(RuntimeError, match="temporary homeserver failure"):
+        await channel.send_delta("!room:matrix.org", "Hello")
+
+    assert "!room:matrix.org" not in channel._stream_bufs
+
+
+@pytest.mark.asyncio
+async def test_send_delta_stream_end_keeps_buffer_when_send_returns_error(monkeypatch) -> None:
+    class _FakeRoomSendError:
+        def __str__(self) -> str:
+            return "temporary homeserver failure"
+
+    channel = MatrixChannel(_make_config(), MessageBus())
+    client = _FakeAsyncClient("", "", "", None)
+    client.room_send_response = _FakeRoomSendError()
+    channel.client = client
+    monkeypatch.setattr(matrix_module, "RoomSendError", _FakeRoomSendError)
+    channel._stream_bufs["!room:matrix.org"] = matrix_module._StreamBuf(
+        text="Final text",
+        event_id="event-1",
+        last_edit=100.0,
+    )
+
+    with pytest.raises(RuntimeError, match="temporary homeserver failure"):
+        await channel.send_delta("!room:matrix.org", "", stream_end=True)
+
+    assert channel._stream_bufs["!room:matrix.org"].text == "Final text"
+
 
 @pytest.mark.asyncio
 async def test_send_delta_ignores_whitespace_only_delta(monkeypatch) -> None:

nanobot/channels/signal/runtime.py

@@ -269,19 +269,18 @@ def _partition_styles(
     if not text_styles:
         return [[] for _ in chunks]
 
-    # Locate each chunk's UTF-16 start in plain_text. split_message lstrips at
+    # Locate each chunk in the original text. This accounts for delimiters
-    # boundaries (but not before the first chunk), so we skip whitespace
+    # removed at split points while retaining indentation inside a chunk.
-    # between chunks to mirror that.
     chunk_ranges: list[tuple[int, int]] = []
     cursor = 0  # Python codepoint cursor in plain_text
-    for i, chunk in enumerate(chunks):
+    for chunk in chunks:
-        if i > 0:
+        chunk_start = plain_text.find(chunk, cursor)
-            while cursor < len(plain_text) and plain_text[cursor].isspace():
+        if chunk_start < 0:
-                cursor += 1
+            chunk_start = cursor
-        utf16_start = _utf16_len(plain_text[:cursor])
+        utf16_start = _utf16_len(plain_text[:chunk_start])
         utf16_end = utf16_start + _utf16_len(chunk)
         chunk_ranges.append((utf16_start, utf16_end))
-        cursor += len(chunk)
+        cursor = chunk_start + len(chunk)
 
     result: list[list[str]] = [[] for _ in chunks]
     for entry in text_styles:
@@ -862,6 +861,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 +1061,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 +1073,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/channels/signal/tests/test_signal_markdown.py

@@ -384,6 +384,23 @@ def test_partition_styles_drops_styles_outside_chunks():
     assert parts == [[], []]
 
 
+def test_partition_styles_keeps_offset_in_indented_chunk():
+    """Styles after preserved indentation remain relative to the chunk."""
+    plain = "header\n    code"
+    chunks = split_message(plain, 10)
+
+    assert chunks == ["header", "    code"]
+    assert _partition_styles(plain, chunks, ["11:4:BOLD"]) == [[], ["4:4:BOLD"]]
+
+
+def test_partition_styles_keeps_offset_after_crlf_boundary():
+    plain = "header\r\n    code"
+    chunks = split_message(plain, 10)
+
+    assert chunks == ["header", "    code"]
+    assert _partition_styles(plain, chunks, ["12:4:BOLD"]) == [[], ["4:4:BOLD"]]
+
+
 def test_partition_styles_long_message_preserves_chunk_one_styles():
     """A bold span deep in the message must follow the message into chunk 1."""
     # Two ~30-char paragraphs separated by a blank line, then **tail**.

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:
+        if session_routing_key:
-            body["prompt_cache_key"] = _prompt_cache_key(session_id)
+            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]:
+def _build_headers(
-    return {
+    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():
+        await self._queue.put(_STREAM_SENTINEL)
-            with suppress(asyncio.QueueEmpty):
-                self._queue.get_nowait()
-        with suppress(asyncio.QueueFull):
-            self._queue.put_nowait(_STREAM_SENTINEL)
 
 
 class SDKStreamingHook(AgentHook):

nanobot/utils/helpers.py

@@ -633,21 +633,69 @@ def split_message(content: str, max_len: int = 2000) -> list[str]:
         return [content]
     if len(content) <= max_len:
         return [content]
+    original_content = content
     chunks: list[str] = []
     while content:
         if len(content) <= max_len:
-            chunks.append(content)
+            if content.strip():
+                chunks.append(content)
             break
         cut = content[:max_len]
-        # Try to break at newline first, then space, then hard break
+        # Consume only the newline itself so indentation starts the next chunk.
-        pos = cut.rfind("\n")
+        newline_pos = cut.rfind("\n")
-        if pos <= 0:
+        if newline_pos >= 0:
-            pos = cut.rfind(" ")
+            # Exclude both bytes of a CRLF boundary from the emitted chunk.
-        if pos <= 0:
+            line_end = newline_pos
-            pos = max_len
+            if line_end > 0 and content[line_end - 1] == "\r":
-        chunks.append(content[:pos])
+                line_end -= 1
-        content = content[pos:].lstrip()
+            chunk = content[:line_end]
-    return chunks
+            if chunk.strip():
+                chunks.append(chunk)
+            content = content[newline_pos + 1 :]
+            continue
+
+        # Keep the existing word-boundary behavior, but avoid emitting a
+        # whitespace-only chunk when an indented line exceeds max_len.
+        space_pos = cut.rfind(" ")
+        if space_pos > 0 and cut[:space_pos].strip():
+            chunks.append(content[:space_pos])
+            content = content[space_pos:].lstrip(" \t")
+            # A space boundary may sit immediately before a line break. Drop
+            # that delimiter too, without stripping the next line's indent.
+            if content.startswith("\r\n"):
+                content = content[2:]
+            elif content.startswith("\n"):
+                content = content[1:]
+            continue
+
+        # Do not split between the two code points of a CRLF delimiter.
+        if cut.endswith("\r") and content[max_len : max_len + 1] == "\n":
+            chunk = cut[:-1]
+            if chunk.strip():
+                chunks.append(chunk)
+            content = content[max_len + 1 :]
+            continue
+
+        chunk = content[:max_len]
+        if chunk.strip():
+            chunks.append(chunk)
+        content = content[max_len:]
+        if not chunk.strip():
+            # Keep any remaining indentation so the final non-blank chunk can
+            # retain as much of it as the channel limit permits.
+            continue
+        # A delimiter can sit immediately after the hard-break boundary. Keep
+        # ordinary space trimming, but consume only the newline so indentation
+        # on the following line is preserved.
+        content = content.lstrip(" \t")
+        if content.startswith("\r\n"):
+            content = content[2:]
+        elif content.startswith("\n"):
+            content = content[1:]
+    # Preserve the historical non-empty-input contract for callers that take
+    # the first chunk directly. This fallback is only reachable for content
+    # made entirely of whitespace.
+    return chunks or [original_content[:max_len]]
 
 
 def build_assistant_message(

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,

tests/utils/test_helpers.py

@@ -17,6 +17,77 @@ def test_split_message_no_code_blocks_unchanged():
     assert split_message(content, max_len=12) == ["alpha beta", "gamma delta"]
 
 
+def test_split_message_preserves_indentation_after_newline():
+    content = "header\n    indented code"
+
+    assert split_message(content, max_len=18) == ["header", "    indented code"]
+
+
+def test_split_message_preserves_indentation_across_hard_break():
+    content = "head\n    abcdefghij"
+
+    assert split_message(content, max_len=8) == ["head", "    abcd", "efghij"]
+
+
+def test_split_message_preserves_indentation_when_newline_is_at_hard_break():
+    content = "abcdefgh\n    code"
+
+    assert split_message(content, max_len=8) == ["abcdefgh", "    code"]
+    assert split_message(content.replace("\n", "\r\n"), max_len=8) == [
+        "abcdefgh",
+        "    code",
+    ]
+
+
+def test_split_message_handles_crlf_before_hard_break():
+    content = "header\r\n    indented code"
+
+    assert split_message(content, max_len=18) == ["header", "    indented code"]
+    assert split_message("abcdefg\r\n    code", max_len=8) == [
+        "abcdefg",
+        "    code",
+    ]
+
+
+def test_split_message_preserves_indent_after_space_then_newline_boundary():
+    content = "abcdef \n    code"
+
+    assert split_message(content, max_len=7) == ["abcdef", "    cod", "e"]
+    assert split_message(content.replace("\n", "\r\n"), max_len=7) == [
+        "abcdef",
+        "    cod",
+        "e",
+    ]
+
+
+def test_split_message_drops_blank_chunks_from_long_indentation():
+    content = "head\n" + " " * 20 + "x"
+
+    chunks = split_message(content, max_len=8)
+
+    assert chunks == ["head", "    x"]
+    assert all(chunk.strip() for chunk in chunks)
+
+
+def test_split_message_drops_whitespace_only_line_at_boundary():
+    content = "    \nhello world"
+
+    assert split_message(content, max_len=8) == ["hello", "world"]
+
+
+def test_split_message_drops_whitespace_only_tail_after_hard_break():
+    prefix = "abcdefgh"
+
+    assert split_message(prefix + "\n", max_len=8) == [prefix]
+    assert split_message(prefix + " ", max_len=8) == [prefix]
+
+
+def test_split_message_keeps_one_chunk_for_all_whitespace_input():
+    content = " " * 10
+
+    assert split_message(content, max_len=4) == [" " * 4]
+
+
 def test_split_message_nonpositive_maxlen_returns_unsplit():
     content = "alpha beta gamma delta"
 

webui/src/components/LanguageSwitcher.tsx

@@ -50,14 +50,7 @@ export function LanguageSwitcher() {
         >
           {supportedLocales.map((option) => (
             <DropdownMenuRadioItem key={option.code} value={option.code}>
-              <span className="flex min-w-0 items-center gap-2">
+              {option.nativeLabel}
-                <span>{option.nativeLabel}</span>
-                {option.nativeLabel !== option.label ? (
-                  <span className="truncate text-xs text-muted-foreground">
-                    {option.label}
-                  </span>
-                ) : null}
-              </span>
             </DropdownMenuRadioItem>
           ))}
         </DropdownMenuRadioGroup>

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/i18n/config.ts

@@ -1,16 +1,16 @@
 export const LOCALE_STORAGE_KEY = "nanobot.locale";
 
 export const supportedLocales = [
-  { code: "en", label: "English", nativeLabel: "English" },
+  { code: "en", nativeLabel: "English" },
-  { code: "zh-CN", label: "Chinese (Simplified)", nativeLabel: "简体中文" },
+  { code: "zh-CN", nativeLabel: "简体中文" },
-  { code: "zh-TW", label: "Chinese (Traditional)", nativeLabel: "繁體中文" },
+  { code: "zh-TW", nativeLabel: "繁體中文" },
-  { code: "fr", label: "French", nativeLabel: "Français" },
+  { code: "fr", nativeLabel: "Français" },
-  { code: "ja", label: "Japanese", nativeLabel: "日本語" },
+  { code: "ja", nativeLabel: "日本語" },
-  { code: "ko", label: "Korean", nativeLabel: "한국어" },
+  { code: "ko", nativeLabel: "한국어" },
-  { code: "es", label: "Spanish", nativeLabel: "Español" },
+  { code: "es", nativeLabel: "Español" },
-  { code: "pt-BR", label: "Portuguese (Brazil)", nativeLabel: "Português (Brasil)" },
+  { code: "pt-BR", nativeLabel: "Português (Brasil)" },
-  { code: "vi", label: "Vietnamese", nativeLabel: "Tiếng Việt" },
+  { code: "vi", nativeLabel: "Tiếng Việt" },
-  { code: "id", label: "Indonesian", nativeLabel: "Bahasa Indonesia" },
+  { code: "id", nativeLabel: "Bahasa Indonesia" },
 ] as const;
 
 export type SupportedLocale = (typeof supportedLocales)[number]["code"];

webui/src/tests/i18n.test.tsx

@@ -452,6 +452,17 @@ describe("webui i18n", () => {
     expect(resolveInitialLocale()).toBe("zh-CN");
   });
 
+  it("lists each language by its native name", async () => {
+    const user = userEvent.setup();
+
+    render(<LanguageSwitcher />);
+    await user.click(screen.getByRole("button", { name: "Change language" }));
+
+    for (const { nativeLabel } of supportedLocales) {
+      expect(screen.getByRole("menuitemradio", { name: nativeLabel })).toBeInTheDocument();
+    }
+  });
+
   it("switches UI copy and document locale through the language switcher", async () => {
     const user = userEvent.setup();
 

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), {