fix(claude-code): stop dropping later-turn text/thinking blocks [greptile]

_saw_text_stream / _saw_thinking_stream were turn-wide latches: once any block
was streamed via stream_event deltas, every later assistant text/thinking block
that was NOT streamed got skipped (model output silently dropped). Use only the
precise per-block signal (idx in _streamed_block_indexes) and reset that set
(and the thinking once-guard) at each materialised-message boundary, so an
earlier turn's block index can't linger and drop a later turn's block at the
same index. Drop the now-dead _saw_text_stream latch. Add a regression test for
a non-streamed text block in a later turn.

Also (P2) replace asyncio.run() at module import in the claude-code conformance
test with a loop-free driver: the fixture conversion only iterates in-memory
envelopes, so it never suspends on real I/O. asyncio.run() at import raises when
a loop is already running (programmatic pytest, Jupyter, session-scoped loops);
the manual driver is unaffected by ambient loop state.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: declan-scale

src/agentex/lib/adk/_modules/_claude_code_sync.py

@@ -100,9 +100,11 @@ async def convert_claude_code_to_agentex_events(
     _text_index: int | None = None
     # Track which assistant-message block indices were already streamed via
     # stream_event triples. Those blocks must not be re-emitted when the full
-    # assistant message arrives.
+    # assistant message arrives. Reset at each message boundary (see below) so a
+    # later turn's block indices don't collide with an earlier turn's.
     _streamed_block_indexes: set[int] = set()
-    _saw_text_stream = False
+    # Once-guard so a thinking block's pending index is claimed on its first
+    # thinking_delta only. Reset per turn alongside _streamed_block_indexes.
     _saw_thinking_stream = False
     # For deferred ReasoningStarted: if a content_block_start(thinking) arrives
     # but no thinking_delta ever follows, the final assistant block's thinking
@@ -142,8 +144,9 @@ async def convert_claude_code_to_agentex_events(
                 block_type = block.get("type", "")
 
                 if block_type == "text":
-                    # Skip if already delivered via stream_event deltas
-                    if _saw_text_stream or idx in _streamed_block_indexes:
+                    # Skip only the specific blocks already delivered via
+                    # stream_event deltas (per-block, not a turn-wide latch).
+                    if idx in _streamed_block_indexes:
                         continue
                     text = block.get("text", "")
                     if text:
@@ -166,8 +169,9 @@ async def convert_claude_code_to_agentex_events(
                         yield StreamTaskMessageDone(type="done", index=msg_index)
 
                 elif block_type == "thinking":
-                    # Skip if already delivered via stream_event deltas
-                    if _saw_thinking_stream or idx in _streamed_block_indexes:
+                    # Skip only the specific blocks already delivered via
+                    # stream_event deltas (per-block, not a turn-wide latch).
+                    if idx in _streamed_block_indexes:
                         continue
                     thinking_text = block.get("thinking", "")
                     if thinking_text:
@@ -239,6 +243,13 @@ async def convert_claude_code_to_agentex_events(
                         ),
                     )
 
+            # End of a materialised message: reset per-turn streaming dedup state
+            # so the next turn's stream_event indices start clean. Without this,
+            # a block index streamed in an earlier turn would linger in the set
+            # and silently drop a later turn's non-streamed block at that index.
+            _streamed_block_indexes = set()
+            _saw_thinking_stream = False
+
         # -----------------------------------------------------------------------
         # stream_event — incremental streaming deltas
         # -----------------------------------------------------------------------
@@ -277,7 +288,6 @@ async def convert_claude_code_to_agentex_events(
                 elif btype == "text":
                     _text_open = True
                     _text_buf = ""
-                    _saw_text_stream = True
                     if isinstance(block_index, int):
                         _streamed_block_indexes.add(block_index)
                     msg_index = next_index

tests/lib/adk/test_claude_code_sync.py

@@ -140,6 +140,30 @@ async def test_streamed_text_not_re_emitted_by_assistant_block(self):
         text_starts = [e for e in out if isinstance(e, StreamTaskMessageStart) and isinstance(e.content, TextContent)]
         assert len(text_starts) == 1, "Text block must not be emitted twice"
 
+    async def test_later_turn_non_streamed_text_not_dropped(self):
+        """A non-streamed text block in a later turn must not be dropped because an
+        earlier turn streamed a block at the same index."""
+        envelopes = [
+            # Turn 1: streamed text at index 0 (dedup'd against the materialised msg).
+            {
+                "type": "stream_event",
+                "event": {"type": "content_block_start", "index": 0, "content_block": {"type": "text"}},
+            },
+            {
+                "type": "stream_event",
+                "event": {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "first"}},
+            },
+            {"type": "stream_event", "event": {"type": "content_block_stop", "index": 0}},
+            {"type": "assistant", "message": {"content": [{"type": "text", "text": "first"}]}},
+            # Turn 2: a NON-streamed text block, also at index 0.
+            {"type": "assistant", "message": {"content": [{"type": "text", "text": "second"}]}},
+        ]
+        out = await _collect(convert_claude_code_to_agentex_events(_aiter(envelopes)))
+        deltas = [
+            e.delta.text_delta for e in out if isinstance(e, StreamTaskMessageDelta) and isinstance(e.delta, TextDelta)
+        ]
+        assert deltas == ["first", "second"], "Later turn's non-streamed text must still be delivered"
+
 
 # ---------------------------------------------------------------------------
 # Thinking / reasoning content

tests/lib/core/harness/conformance/test_claude_code_conformance.py

@@ -29,6 +29,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import pytest
 
 from agentex.lib.adk._modules._claude_code_sync import convert_claude_code_to_agentex_events
@@ -151,16 +153,27 @@ async def _build_fixture(name: str, envelopes: list[dict]) -> Fixture:
     return Fixture(name=name, events=events)
 
 
-# We need to run the async conversion synchronously at module import time so
-# that fixtures are ready before pytest collects. Use asyncio.run() carefully —
-# this is safe in a test module since no event loop is running at import time.
-import asyncio as _asyncio
+# Fixtures must exist before pytest collects (they parametrize the test below),
+# so they are built at import time. The conversion only iterates in-memory
+# envelopes — it never suspends on a real future — so we drive the coroutine to
+# completion by hand instead of asyncio.run(). asyncio.run() at import raises
+# RuntimeError when an event loop is already running (programmatic pytest, a
+# Jupyter kernel, or session-scoped asyncio loops); the loop-free driver below
+# is unaffected by the ambient loop state.
+def _run_pure_async(coro: Any) -> Any:
+    try:
+        coro.send(None)
+    except StopIteration as stop:
+        return stop.value
+    coro.close()
+    raise RuntimeError("conformance fixture build unexpectedly suspended on real I/O")
+
 
 _FIXTURES: list[Fixture] = [
-    _asyncio.run(_build_fixture("claude-code-text-only", _TEXT_ENVELOPES)),
-    _asyncio.run(_build_fixture("claude-code-tool-call-result", _TOOL_ENVELOPES)),
-    _asyncio.run(_build_fixture("claude-code-thinking-block", _THINKING_ENVELOPES)),
-    _asyncio.run(_build_fixture("claude-code-multi-step", _MULTI_STEP_ENVELOPES)),
+    _run_pure_async(_build_fixture("claude-code-text-only", _TEXT_ENVELOPES)),
+    _run_pure_async(_build_fixture("claude-code-tool-call-result", _TOOL_ENVELOPES)),
+    _run_pure_async(_build_fixture("claude-code-thinking-block", _THINKING_ENVELOPES)),
+    _run_pure_async(_build_fixture("claude-code-multi-step", _MULTI_STEP_ENVELOPES)),
 ]
 
 # Register into the shared registry so all_fixtures() can enumerate them