fix(backends,conversations,telegram,ui): a background subagent reports back as its own turn

tests/test_conversations.py

@@ -8,6 +8,7 @@ from zoneinfo import ZoneInfo
 
 import pytest
 from claude_agent_sdk import (
+    TaskNotificationMessage,
     AssistantMessage,
     InMemorySessionStore,
     ResultMessage,
@@ -92,6 +93,8 @@ class ScriptedClient:
         self.session_id = options.resume or str(uuid.uuid4())
         self.interrupted = False
         self.connected = False
+        self.asked = asyncio.Event()
+        self.extra: list[Any] = []
         ScriptedClient.instances.append(self)
 
     async def connect(self) -> None:
@@ -106,6 +109,26 @@ class ScriptedClient:
                 b.get("text", "") for b in content if b.get("type") == "text"
             )
         self.prompts.append(prompt)
+        self.asked.set()
+
+    async def receive_messages(self):
+        """One response per prompt, forever; ``extra`` is what the CLI sends
+        on its own between prompts (a subagent reporting back)."""
+        served = 0
+        while True:
+            while len(self.prompts) <= served:
+                self.asked.clear()
+                if self.extra:
+                    for message in self.extra:
+                        if isinstance(message, asyncio.Event):
+                            await message.wait()
+                            continue
+                        yield message
+                    self.extra = []
+                await self.asked.wait()
+            served += 1
+            async for message in self.receive_response():
+                yield message
 
     async def receive_response(self):
         prompt = self.prompts[-1]
@@ -858,3 +881,80 @@ async def test_inject_into_a_closed_master_lands_in_the_open_one(world: World) -
     assert item.conversation_id == new.id
     assert await world.statuses(old) == []
     assert await world.statuses(new) == [("normal", "queued")]
+
+
+def _self_started(session_id: str, text: str) -> list[Any]:
+    """What the CLI sends by itself once a background subagent reports back."""
+    return [
+        AssistantMessage(
+            content=[ToolUseBlock(id="tu_dig", name="Grep", input={"pattern": "au"})],
+            model="m",
+            parent_tool_use_id="tu_bg",
+        ),
+        TaskNotificationMessage(
+            subtype="task_notification",
+            data={},
+            task_id="t1",
+            status="completed",
+            output_file="",
+            summary='Agent "dig" finished',
+            uuid="n1",
+            session_id=session_id,
+        ),
+        AssistantMessage(content=[TextBlock(text=text)], model="m"),
+        ResultMessage(
+            subtype="success",
+            duration_ms=1,
+            duration_api_ms=1,
+            is_error=False,
+            num_turns=1,
+            session_id=session_id,
+            stop_reason="end_turn",
+            total_cost_usd=0.0,
+            usage={"input_tokens": 1, "output_tokens": 1},
+            origin={"kind": "task-notification"},
+        ),
+    ]
+
+
+async def test_subagent_report_between_turns_is_its_own_turn(world: World) -> None:
+    conv = await world.conversations.create(kind="master", agent="a", origin="test")
+    seen: list[dict[str, Any]] = []
+
+    async def collect() -> None:
+        async for event in world.bus.stream(conversation_id=conv.external_id):
+            seen.append(event)
+
+    task = asyncio.create_task(collect())
+    await world.conversations.post(conv, "go")
+    await world.settle(conv, 1)
+    client = ScriptedClient.instances[0]
+    client.extra = _self_started(client.session_id, "the agent found gold")
+    client.asked.set()
+    await asyncio.sleep(0.3)
+    task.cancel()
+    replies = [e for e in seen if e["type"] == "reply"]
+    assert [r["text"] for r in replies] == ["ok:go", "the agent found gold"]
+    assert replies[1]["item_origin"] == "task"
+    starts = [e for e in seen if e["type"] == "turn.start"]
+    assert [s["origin"] for s in starts] == ["user", "task"]
+    assert starts[1]["text"] == 'Agent "dig" finished (completed)'
+    assert len(client.prompts) == 1
+    first_end = next(i for i, e in enumerate(seen) if e["type"] == "turn.end")
+    later = [e for e in seen[first_end + 1 :] if e["type"] in ("stream", "tool")]
+    assert later and all(e["turn_id"] == starts[0]["turn_id"] for e in later[:1])
+    assert all(e["turn_id"] == starts[1]["turn_id"] for e in later[1:])
+    assert (await world.conversations.get(conv.external_id)).running_turn is None
+
+
+async def test_subagent_result_inside_a_turn_does_not_end_it(world: World) -> None:
+    conv = await world.conversations.create(kind="master", agent="a", origin="test")
+    await world.conversations.post(conv, "one")
+    await world.settle(conv, 1)
+    client = ScriptedClient.instances[0]
+    client.extra = _self_started(client.session_id, "late report")
+    await world.conversations.post(conv, "two")
+    await world.settle(conv, 2)
+    assert client.prompts[-1].startswith("two")
+    rows = await world.conversations.queue.recent(conv.id)
+    assert [r.status for r in rows] == ["done", "done"]