fix(streaming): StreamTaskMessageFull closes the coalescing buffer

A StreamTaskMessageFull ends the stream and marks the context done, but it did
not close the coalescing buffer. __aexit__'s close() then short-circuits on
_is_closed and never stops the buffer's ticker, leaving an orphaned background
task. Buffered deltas could also publish after the terminal Full, which a
consumer treating Full as final reads as a stale duplicate tail.

Drain and stop the buffer before publishing the Full (deltas -> Full ordering),
and reap the buffer in close() before the _is_closed short-circuit so it can't
be orphaned on any path.

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

Author: eberki-scale

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

@@ -420,15 +420,17 @@ async def close(self) -> TaskMessage:
         if not self.task_message:
             raise ValueError("Context not properly initialized - no task message")
 
-        if self._is_closed:
-            return self.task_message  # Already done
-
-        # Drain any buffered deltas before announcing DONE so consumers see the
-        # full sequence in order.
+        # Reap the buffer (stopping its ticker) before the _is_closed
+        # short-circuit, so a context already marked done by a Full update can't
+        # leave the ticker orphaned. Draining here also lets consumers see the
+        # full delta sequence in order before DONE.
         if self._buffer is not None:
             await self._buffer.close()
             self._buffer = None
 
+        if self._is_closed:
+            return self.task_message  # Already done (buffer reaped above)
+
         # Send the DONE event
         done_event = StreamTaskMessageDone(
             parent_task_message=self.task_message,
@@ -486,6 +488,15 @@ async def stream_update(self, update: TaskMessageUpdate) -> TaskMessageUpdate |
                 await self._buffer.add(update)
                 return update
 
+        # A Full ends the stream and supersedes buffered deltas. Drain and stop
+        # the buffer BEFORE publishing the Full, so leftover deltas land in order
+        # (deltas -> Full) instead of trailing the terminal Full as a stale
+        # duplicate tail. This also stops the ticker, which would otherwise be
+        # orphaned when __aexit__'s close() short-circuits on _is_closed.
+        if isinstance(update, StreamTaskMessageFull) and self._buffer is not None:
+            await self._buffer.close()
+            self._buffer = None
+
         result = await self._streaming_service.stream_update(update)
 
         if isinstance(update, StreamTaskMessageDone):

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

@@ -22,7 +22,10 @@
     ToolResponseDelta,
     ReasoningSummaryDelta,
 )
-from agentex.types.task_message_update import StreamTaskMessageDelta
+from agentex.types.task_message_update import (
+    StreamTaskMessageFull,
+    StreamTaskMessageDelta,
+)
 from agentex.lib.core.services.adk.streaming import (
     CoalescingBuffer,
     StreamingTaskMessageContext,
@@ -520,3 +523,57 @@ async def test_open_without_created_at_passes_omit(self) -> None:
 
         kwargs = client.messages.create.call_args.kwargs
         assert kwargs["created_at"] is omit
+
+
+class TestFullMessageClosesBuffer:
+    """A StreamTaskMessageFull must stop the buffer ticker and drain its deltas
+    before the terminal Full. Marking the context done without closing the
+    buffer leaves close()'s _is_closed short-circuit to orphan the ticker, and
+    publishing buffered deltas after the Full reads as a stale duplicate tail."""
+
+    @pytest.mark.asyncio
+    async def test_full_message_stops_ticker(self) -> None:
+        ctx, _svc, tm = await _make_context("coalesced")
+        # A delta makes the buffer and its ticker live.
+        await ctx.stream_update(_text(tm, "hello"))
+        buf = ctx._buffer
+        assert buf is not None
+        task = buf._task
+        assert task is not None and not task.done()
+
+        await ctx.stream_update(
+            StreamTaskMessageFull(
+                parent_task_message=tm,
+                content=TextContent(author="agent", content="final", format="markdown"),
+                type="full",
+            )
+        )
+
+        assert ctx._buffer is None, "Full message left the buffer un-closed"
+        assert task.done(), "coalescing-buffer ticker still running after Full (orphaned)"
+
+    @pytest.mark.asyncio
+    async def test_full_is_terminal_publish_no_trailing_deltas(self) -> None:
+        # Buffered deltas must publish BEFORE the Full, never after (a trailing
+        # delta after the terminal Full reads as a stale duplicate tail).
+        ctx, svc, tm = await _make_context("coalesced")
+        # "alpha" flushes immediately; "beta" stays buffered in the window.
+        await ctx.stream_update(_text(tm, "alpha"))
+        await ctx.stream_update(_text(tm, "beta"))
+
+        full = StreamTaskMessageFull(
+            parent_task_message=tm,
+            content=TextContent(author="agent", content="alphabeta", format="markdown"),
+            type="full",
+        )
+        await ctx.stream_update(full)
+
+        published = [c.args[0] for c in svc.stream_update.await_args_list]
+        assert published, "nothing was published"
+        assert published[-1] is full, (
+            f"Full must be the terminal publish; saw trailing "
+            f"{type(published[-1]).__name__} after it (stale duplicate tail)"
+        )
+        assert any(isinstance(u, StreamTaskMessageDelta) for u in published[:-1]), (
+            "expected the buffered deltas to be published before the Full"
+        )