fix(claude-code): reset thinking once-guard per block to avoid duplicates [greptile]

_saw_thinking_stream is set on a thinking block's first delta (to claim its
index in _streamed_block_indexes) but was only reset at the materialised-message
boundary. If one streaming turn produced a SECOND thinking block, the guard
stayed True, the second block's index was never claimed, and the final assistant
envelope re-emitted it — duplicate Start/Delta/Done. Reset the guard on
content_block_stop so it is scoped per thinking block. Adds a regression test
with two streamed thinking blocks in one turn.

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

Author: declan-scale

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

@@ -345,6 +345,11 @@ async def convert_claude_code_to_agentex_events(
                     _thinking_open = False
                     _thinking_buf = ""
                     _pending_thinking_block_index = None
+                    # Reset the once-guard per thinking block: a turn can stream a
+                    # second thinking block, and without this the guard stays True,
+                    # the second block's index is never claimed, and the final
+                    # assistant envelope re-emits it (duplicate Start/Delta/Done).
+                    _saw_thinking_stream = False
                     if _thinking_index is not None:
                         yield StreamTaskMessageDone(type="done", index=_thinking_index)
                     _thinking_index = None

tests/lib/adk/test_claude_code_sync.py

@@ -247,6 +247,47 @@ async def test_streamed_thinking_emits_reasoning_start_deltas_done(self):
         assert deltas[1].delta.content_delta == " step two"
         assert len(dones) == 1
 
+    async def test_two_streamed_thinking_blocks_not_re_emitted(self):
+        """A turn that streams two thinking blocks must claim both indices, so the
+        final assistant envelope does not re-emit the second one."""
+
+        def _thinking_block(idx: int, text: str) -> list:
+            return [
+                {
+                    "type": "stream_event",
+                    "event": {"type": "content_block_start", "index": idx, "content_block": {"type": "thinking"}},
+                },
+                {
+                    "type": "stream_event",
+                    "event": {
+                        "type": "content_block_delta",
+                        "index": idx,
+                        "delta": {"type": "thinking_delta", "thinking": text},
+                    },
+                },
+                {"type": "stream_event", "event": {"type": "content_block_stop", "index": idx}},
+            ]
+
+        envelopes = [
+            *_thinking_block(0, "first thought"),
+            *_thinking_block(1, "second thought"),
+            # Final assistant envelope repeats both thinking blocks — neither should re-emit.
+            {
+                "type": "assistant",
+                "message": {
+                    "content": [
+                        {"type": "thinking", "thinking": "first thought"},
+                        {"type": "thinking", "thinking": "second thought"},
+                    ]
+                },
+            },
+        ]
+        out = await _collect(convert_claude_code_to_agentex_events(_aiter(envelopes)))
+        reasoning_starts = [
+            e for e in out if isinstance(e, StreamTaskMessageStart) and isinstance(e.content, ReasoningContent)
+        ]
+        assert len(reasoning_starts) == 2, "each streamed thinking block emitted exactly once (no duplicate)"
+
     async def test_thinking_block_start_with_no_deltas_allows_assistant_to_fill(self):
         """A thinking block_start without any deltas leaves the final assistant block
         free to emit the thinking text (the block index is not claimed as streamed)."""