fix(streaming): StreamTaskMessageFull closes the coalescing buffer

[eberki-scale]

Summary

A StreamTaskMessageFull ends the stream and marks the StreamingTaskMessageContext done, but it did not close the coalescing buffer. When the context later exits, __aexit__ → close() hits its if self._is_closed: return guard and never stops the buffer's background ticker — leaving an orphaned task per such stream. Separately, any deltas still buffered when the Full arrives could be published after the terminal Full, which a consumer treating Full as the final message reads as a stale duplicate tail.

Fix

Two small changes to StreamingTaskMessageContext in agentex/lib/core/services/adk/streaming.py:

1. stream_update — when a StreamTaskMessageFull arrives, drain and close the buffer before publishing the Full, so leftover deltas land in order (deltas → Full) and the ticker is stopped.
2. close() — reap the buffer before the _is_closed short-circuit, so a context already marked done (by a Full on another path) can never leave the ticker orphaned.

No change to CoalescingBuffer itself.

Tests

New TestFullMessageClosesBuffer:

• test_full_message_stops_ticker — after a Full, the buffer is reaped and its ticker task is done() (not orphaned).
• test_full_is_terminal_publish_no_trailing_deltas — buffered deltas publish before the Full; the Full is the terminal publish.

Full streaming suite: 33 passed (tests/lib/core/services/adk/test_streaming.py).

Scope

Deliberately narrow — just the StreamTaskMessageFull orphan + ordering fix, split out from the broader buffer work in #418 for easier review.

Greptile Summary

• Updates StreamingTaskMessageContext so terminal StreamTaskMessageFull messages drain and close the coalescing buffer before publishing the full message.
• Ensures close() reaps the coalescing buffer even when the context was already marked closed.
• Adds streaming tests covering ticker cleanup, terminal publish ordering, and add/close race handling.

Confidence Score: 5/5

The streaming lifecycle changes are narrow, directly covered by targeted tests, and do not introduce broader API or persistence risk.

The changes address buffer draining, ticker cleanup, and terminal publish ordering in the affected context, with focused tests exercising the intended edge cases.

T-Rex Logs

What T-Rex did

• Before the artifact for proof 0, the run shows base published start, delta:first-, full:FINAL, followed by a stale trailing delta:buffered-tail.
• After the artifact for proof 0, head published start, delta:first-, delta:buffered-tail, then full:FINAL; buffer_after_full_is_none was true, ticker_done_after_full was true, and no coalescing-buffer tasks remained alive after close.
• Before the artifact for proof 1, trex-artifacts/coalescing-race-guard-01-before.log shows the commit ending with buf_len=1, buf_contents=['racing'], flushed=0, RESULT=STRANDED at the race observation point.
• After the artifact for proof 1, trex-artifacts/coalescing-race-guard-02-after.log shows the commit ending with buf_len=0, buf_contents=[], flushed=0, RESULT=DROPPED at the same race observation point.

_{Ran code and verified through T-Rex}

_{Reviews (3): Last reviewed commit: "fix(streaming): re-check _closed under l..." | Re-trigger Greptile}

[smoreinis]

I saw that Greptile also commented about this and looks like it resolved its own comment but it looks to me like it's still possible that we could have a race here? e.g.

1. This code block closes the buffer and sets it to none
2. stream_update (or messages.update) below is being awaited
3. another delta comes in and skips this block since buffer is already none but _is_closed is not yet set to True

In this case, consumers will still see a delta after full since it just falls through to publishing directly. If this is something we care about (and my logic checks out and you also believe there's a race), we could have a terminal-in-progress flag like _is_closing set before awaiting buffer close to reject deltas?

[smoreinis]

if i'm reading this correctly, this code block still has a latent bug similar to the one we fixed for the Full case above, right? curious about just abstracting the code above into something like _reap_buffer and calling that here as well for readability and more defensiveness going forward.

[smoreinis]

think it would be ever so slightly cleaner to just abstract this into _reap_buffer even if we're only calling it into two places and then call it here and below

[smoreinis]

technically, "beta" may or may not stay buffered in the window since the assertions only need the Full to be the terminal publish and 1 or more delta before it, right? if so, i think this test is still a solid one as written but maybe just worth updating the comment to be more precise (or adding a clarification to the assertions below)

[smoreinis]

thanks for this fix - overall the PR is solid and I'm fine with it going in as-is.
just a couple of minor nits in the comments that I think could be nice to haves, but happy to leave these at the author discretion.