fix(backends,markdown): stream turn events live, no blank lines for tool blocks

tests/test_claude_sdk_backend.py

@@ -1,3 +1,4 @@
+import asyncio
 import tempfile
 from pathlib import Path
 from typing import Any
@@ -419,3 +420,41 @@ async def test_close_disconnects(cwd: Path) -> None:
         )
     assert FakeClient.instances[0].connected is False
     assert backend.sessions == {}
+
+
+class GatedClient(FakeClient):
+    """Streams one delta, then waits for ``gate`` before finishing the turn."""
+
+    gate: asyncio.Event
+
+    async def receive_response(self):
+        yield StreamEvent(
+            uuid="u", session_id="s", event={"type": "message_start", "message": {}}
+        )
+        for e in _stream(0, "first"):
+            yield e
+        await GatedClient.gate.wait()
+        yield AssistantMessage(content=[TextBlock(text="first")], model="m")
+        yield _result(self.session_id)
+
+
+async def test_deltas_reach_the_caller_before_the_turn_ends(cwd: Path) -> None:
+    """The turn is not buffered: a delta is observable while the CLI still runs."""
+    GatedClient.gate = asyncio.Event()
+    backend = _backend(cwd, InMemorySessionStore())
+    backend._factory = GatedClient
+    events = backend.complete(
+        agent=backend.agent,
+        messages=[{"role": "user", "content": "hi"}],
+        conversation_id="conv-live",
+    )
+    seen: list[Any] = []
+    async for ev in events:
+        seen.append(ev)
+        if isinstance(ev, RawContentBlockDeltaEvent):
+            break
+    assert seen[-1].delta.text == "first"
+    assert not GatedClient.gate.is_set()
+    GatedClient.gate.set()
+    rest = [e async for e in events]
+    assert isinstance(rest[-1], RawMessageStopEvent)