From 9875f08016e0c024d72bfbb608d0404185e4f96b Mon Sep 17 00:00:00 2001
From: h <h@kotikot.com>
Date: Tue, 13 Oct 2026 01:22:40 +0200
Subject: [PATCH] fix(events): a conversation's own stream carries its updates

---
 src/beaver_gateway/events/bus.py |  8 +++++++-
 tests/test_conversations.py      | 20 ++++++++++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/src/beaver_gateway/events/bus.py b/src/beaver_gateway/events/bus.py
index 587547d..f60f5a4 100644
--- a/src/beaver_gateway/events/bus.py
+++ b/src/beaver_gateway/events/bus.py
@@ -57,10 +57,16 @@ class EventBus:
                 event = await queue.get()
                 if (
                     conversation_id is None
-                    or event.get("conversation_id") == conversation_id
+                    or _conversation_of(event) == conversation_id
                 ):
                     yield event
 
     @property
     def subscribers(self) -> int:
         return len(self._subscribers)
+
+
+def _conversation_of(event: Event) -> Any:
+    if event["type"].startswith("conversation."):
+        return event.get("id")
+    return event.get("conversation_id")
diff --git a/tests/test_conversations.py b/tests/test_conversations.py
index 859bcba..30e4bca 100644
--- a/tests/test_conversations.py
+++ b/tests/test_conversations.py
@@ -1052,3 +1052,23 @@ def test_reply_text_keeps_progress_updates_and_drops_hidden_reasoning() -> None:
 def test_reply_text_on_a_plain_answer_is_unchanged() -> None:
     blocks = [AnthropicTextBlock(type="text", text="  just an answer  ")]
     assert reply_text(blocks, thinking_is_progress=True) == "just an answer"
+
+
+async def test_a_conversation_stream_carries_updates_about_that_conversation() -> None:
+    bus = EventBus()
+    seen: list[tuple[str, str | None]] = []
+
+    async def collect() -> None:
+        async for event in bus.stream(conversation_id="c1"):
+            seen.append((event["type"], event.get("id")))
+            if len(seen) == 2:
+                return
+
+    task = asyncio.create_task(collect())
+    await asyncio.sleep(0)
+    bus.publish("conversation.updated", id="c2", running_turn=None)
+    bus.publish("conversation.updated", id="c1", running_turn=None)
+    bus.publish("stream", conversation_id="c2")
+    bus.publish("stream", conversation_id="c1")
+    await asyncio.wait_for(task, 1)
+    assert seen == [("conversation.updated", "c1"), ("stream", None)]