diff --git a/nanobot/channels/matrix/runtime.py b/nanobot/channels/matrix/runtime.py index 11b7303ed..47cad8b69 100644 --- a/nanobot/channels/matrix/runtime.py +++ b/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: diff --git a/nanobot/channels/matrix/tests/test_matrix_channel.py b/nanobot/channels/matrix/tests/test_matrix_channel.py index 372505857..e2950df15 100644 --- a/nanobot/channels/matrix/tests/test_matrix_channel.py +++ b/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: