fix(streaming): re-check _closed under lock to avoid stranding a racing delta

add() checked _closed before taking the lock, so a delta racing a Full-driven
buffer close() could pass the check, then append after the buffer was drained
and its ticker shut down — stranding the delta, never published. Re-check
_closed under the lock before appending.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: eberki-scale

src/agentex/lib/core/services/adk/streaming.py

@@ -177,6 +177,12 @@ async def add(self, update: StreamTaskMessageDelta) -> None:
         if self._closed:
             return
         async with self._lock:
+            # Re-check under the lock: a concurrent close() (e.g. from a racing
+            # Full) may have drained and shut down the ticker after the check
+            # above but before we acquired the lock. Appending now would strand
+            # the delta in a dead buffer, never published.
+            if self._closed:
+                return
             self._buf.append(update)
             self._buf_chars += _delta_char_len(update.delta)
             if not self._first_flushed or self._buf_chars >= self.MAX_BUFFERED_CHARS:

tests/lib/core/services/adk/test_streaming.py

@@ -355,6 +355,24 @@ async def on_flush(u: StreamTaskMessageDelta) -> None:
         await buf.add(_text(task_message, "after"))
         assert flushed == []
 
+    @pytest.mark.asyncio
+    async def test_add_racing_close_is_not_stranded(self, task_message: TaskMessage) -> None:
+        """TOCTOU: a delta that passes add()'s pre-lock _closed check but only
+        acquires the lock after close() set _closed must be dropped, not appended
+        to a drained, ticker-less buffer where it would never be published."""
+        buf = CoalescingBuffer(on_flush=AsyncMock())
+        buf.start()
+        # Hold the lock so add() parks *after* its pre-lock _closed check.
+        await buf._lock.acquire()
+        add_task = asyncio.create_task(buf.add(_text(task_message, "racing")))
+        await asyncio.sleep(0)  # add() passes the _closed check, blocks on the lock
+        buf._closed = True  # close() wins the race
+        buf._lock.release()
+        await add_task
+
+        assert buf._buf == [], "racing delta was stranded in the closed buffer"
+        await buf.close()  # cleanup
+
 
 class TestCoalescingBufferCloseDuringFlush:
     @pytest.mark.asyncio