diff --git a/nanobot/channels/matrix.py b/nanobot/channels/matrix.py index 481b2b86..8a2fc9d8 100644 --- a/nanobot/channels/matrix.py +++ b/nanobot/channels/matrix.py @@ -192,6 +192,10 @@ def _build_matrix_text_content( return content +def _matrix_stream_key(chat_id: str, stream_id: str | None) -> str: + return chat_id if stream_id is None else f"{chat_id}\0{stream_id}" + + class MatrixConfig(Base): """Matrix (Element) channel configuration.""" @@ -543,7 +547,8 @@ class MatrixChannel(BaseChannel): relates_to = self._build_thread_relates_to(metadata) if stream_end: - buf = self._stream_bufs.pop(chat_id, None) + stream_key = _matrix_stream_key(chat_id, stream_id) + buf = self._stream_bufs.pop(stream_key, None) if not buf or not buf.event_id or not buf.text: return @@ -557,10 +562,11 @@ class MatrixChannel(BaseChannel): await self._send_room_content(chat_id, content) return - buf = self._stream_bufs.get(chat_id) + stream_key = _matrix_stream_key(chat_id, stream_id) + buf = self._stream_bufs.get(stream_key) if buf is None: buf = _StreamBuf() - self._stream_bufs[chat_id] = buf + self._stream_bufs[stream_key] = buf buf.text += delta if not buf.text.strip(): diff --git a/tests/channels/test_matrix_channel.py b/tests/channels/test_matrix_channel.py index f1e20dfd..da82ba4b 100644 --- a/tests/channels/test_matrix_channel.py +++ b/tests/channels/test_matrix_channel.py @@ -1913,6 +1913,36 @@ async def test_send_delta_stream_end_replaces_existing_message() -> None: } +@pytest.mark.asyncio +async def test_send_delta_keeps_same_room_stream_ids_independent(monkeypatch) -> None: + channel = MatrixChannel(_make_config(), MessageBus()) + client = _FakeAsyncClient("", "", "", None) + channel.client = client + + event_ids = ["event-a", "event-b"] + + async def _send_room_content(room_id, content): + client.room_send_calls.append({"room_id": room_id, "content": content}) + return SimpleNamespace(event_id=event_ids.pop(0) if event_ids else "event-final") + + monkeypatch.setattr(channel, "_send_room_content", _send_room_content) + + await channel.send_delta("!room:matrix.org", "A", stream_id="stream-a") + await channel.send_delta("!room:matrix.org", "B", stream_id="stream-b") + await channel.send_delta("!room:matrix.org", "1", stream_id="stream-a") + await channel.send_delta("!room:matrix.org", "2", stream_id="stream-b") + + await channel.send_delta("!room:matrix.org", "", stream_id="stream-a", stream_end=True) + await channel.send_delta("!room:matrix.org", "", stream_id="stream-b", stream_end=True) + + final_a = client.room_send_calls[-2]["content"] + final_b = client.room_send_calls[-1]["content"] + assert final_a["body"] == "A1" + assert final_a["m.relates_to"]["event_id"] == "event-a" + assert final_b["body"] == "B2" + assert final_b["m.relates_to"]["event_id"] == "event-b" + + @pytest.mark.asyncio async def test_send_delta_starts_threaded_stream_inside_thread() -> None: channel = MatrixChannel(_make_config(), MessageBus())