fix(matrix): propagate stream delivery failures

Signed-off-by: Lanre Shittu <136805224+Shizoqua@users.noreply.github.com>

nanobot/channels/matrix/runtime.py

@@ -658,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
 
@@ -669,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():
@@ -692,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:

nanobot/channels/matrix/tests/test_matrix_channel.py

@@ -2483,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)
@@ -2493,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 channel._stream_bufs["!room:matrix.org"].text == "Hello"
+    assert "!room:matrix.org" not in channel._stream_bufs
     assert len(client.room_send_calls) == 1
 
     assert len(client.typing_calls) == 1
@@ -2504,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: