fix(events): a conversation's own stream carries its updates

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")

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)]