fix(teams): order clarify card after its preamble text via bounded stream-drain barrier - #16
Merged
Merged
Conversation
Add a _DrainBarrier sentinel (carrying a threading.Event) that can be enqueued onto the stream consumer's queue.Queue. The async run() loop pops the barrier, forces a flush of any pending accumulated text to the adapter, then sets the barrier's event. A new public async drain(timeout) method enqueues a barrier and waits for its event. Because run() and drain() are coroutines on the SAME event loop, drain() must not block the loop or run() could never pop the barrier (deadlock). The blocking threading.Event.wait is therefore offloaded to a thread executor via run_in_executor, freeing the loop to keep draining. drain() is bounded (~5s default) and best-effort: on timeout it logs and returns, never hanging the caller. This is an ordering barrier for out-of-band sends (e.g. the Teams clarify Adaptive Card) that must land strictly AFTER the streamed assistant commentary that introduces them. No task_done() audit needed: the queue has no queue.join() usage, and drain() relies on the barrier event rather than join(), so queue accounting is unchanged. Adds TestDrainBarrier with 3 tests (resolves-after-text-sent, returns- when-idle, bounded-on-stuck-consumer). 95 passing in the file.
…ps ahead of preamble text In Teams the clarify Adaptive Card could render ABOVE the assistant text that introduces it. Two delivery paths raced with no ordering barrier: streamed assistant commentary is ENQUEUED on the GatewayStreamConsumer queue and drained asynchronously (enqueue returns immediately), while the clarify card is sent synchronously from _clarify_callback_sync on the worker thread and its HTTP POST completes right away — beating the still- queued text. Before scheduling send_clarify(), the clarify callback now reads the live consumer from stream_consumer_holder[0] and, if it exposes drain(), schedules consumer.drain(timeout=5.0) onto the step loop via safe_schedule_threadsafe and blocks on the returned future (result timeout=6, slightly above drain's own bound). This flushes any queued preamble text to the adapter before the card is posted. Best-effort and bounded: the worker thread already blocks on fut.result() for the card send, so a bounded blocking drain in front is consistent with the existing synchronous contract. On any failure (no consumer, no drain attr, schedule/timeout error) it falls through and still schedules the card, preserving pre-fix behavior.
…er holder Three review findings on the clarify ordering-barrier path, each causing a flat ~5s worker-thread stall in an unhappy case (plus one deprecation): FINDING 1 (BLOCKER) — orphaned drain barrier in GatewayStreamConsumer.run(). If the outer loop body raised (caught by `except Exception`/`except asyncio.CancelledError`) AFTER a _DrainBarrier was popped but BEFORE its event was set, the event stayed unset. drain() then waited the full timeout against a dead consumer task. Fix: wrap the per-iteration loop body in try/finally so `if pending_barrier is not None: pending_barrier.event.set()` runs on EVERY exit path — normal fall-through, the two early `return`s, the `continue`, and any exception propagating to the outer handlers (the per-iteration finally runs before control leaves the loop body). The two previous explicit `.set()` calls (continue branch and end-of-iteration) are now subsumed by the finally and removed; setting an already-set threading.Event is a harmless no-op. The large diff is mostly the 4-space re-indent from introducing the try block; the only logic change is the finally plus the two removed redundant sets. FINDING 2 (MAJOR) — stale/finished consumer left in stream_consumer_holder (gateway/run.py). The holder set at the agent-runtime consumer (~16756) was never reset, so a clarify firing AFTER stream completion enqueued a barrier onto a queue nobody drains -> full ~5s stall. Chosen fix: reset the holder to None in a finally around the awaited `run()` inside `_start_stream_consumer` — this is the clean lifecycle-completion point for the SAME consumer that populated the holder, covering both normal completion and cancellation. The clarify callback already guards on `_sc is not None`, so it now skips the pointless pre-flush drain once the stream is done. Chose the holder-reset over a call-site done-flag guard because a clean completion point exists and resetting fixes the root cause for all callers, not just the clarify path. The SSE-proxy consumer (~15778) is untouched. FINDING 3 (MINOR) — drain() used asyncio.get_event_loop() (deprecated on 3.10+ outside a running loop). drain() always runs inside a running coroutine scheduled via run_coroutine_threadsafe, so switched to get_running_loop(). Adds a deterministic regression test (test_barrier_event_set_when_loop_body_raises_after_pop) that injects a flush failure after the barrier is popped and asserts drain() resolves promptly (via the finally) rather than timing out, and that run() exits cleanly. Tests (run in isolation): test_stream_consumer.py 96 passed, test_teams.py 46 passed, test_adapter_clarify.py 26 passed.
🔎 Lint report:
|
| Rule | Count |
|---|---|
unresolved-import |
1 |
invalid-assignment |
1 |
First entries
tests/agent/test_clarify_unicode_encoding.py:15: [unresolved-import] unresolved-import: Cannot resolve imported module `pytest`
tests/gateway/test_stream_consumer.py:2031: [invalid-assignment] invalid-assignment: Object of type `def _boom(...) -> CoroutineType[Any, Any, Unknown]` is not assignable to attribute `_send_or_edit` of type `def _send_or_edit(self, text: str, *, finalize: bool = False) -> CoroutineType[Any, Any, bool]`
✅ Fixed issues: none
Unchanged: 5093 pre-existing issues carried over.
Diagnostics are surfaced as warnings — this check never fails the build.
…ding) Bedrock Converse returns tool-call input as a native dict. Four json.dumps sites serialized it back to the arguments string with the default ensure_ascii=True, escaping em dash (U+2014) to the literal string \u2014, which surfaced verbatim on the Teams clarify card. Add ensure_ascii=False at all four sites (matches the repo-wide convention, 362 existing occurrences): - agent/bedrock_adapter.py:666 (non-streaming toolUse.input) - agent/bedrock_adapter.py:818 (streaming accumulated input) - agent/conversation_loop.py:3646 (in-place tool_calls validation) - agent/conversation_loop.py:1040 (API-message canonicalizer) Add tests/agent/test_clarify_unicode_encoding.py pinning all four sites.
Author
|
Layered on a second fix: em-dash double-encoding in clarify cards (commit b57ca25) Separate from the ordering barrier, the clarify card was rendering Root cause: Bedrock's Converse API returns tool-call Fix: added
Added |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Fixes a Teams bug where the clarify Adaptive Card could render above the assistant text that introduces it. The streamed assistant commentary is enqueued on the
GatewayStreamConsumerqueue and drained asynchronously, while the clarify card is posted synchronously from the worker thread — so the card's HTTP POST beat the still-queued preamble text, landing the card before the words describing it.How
Adds a bounded, best-effort ordering barrier so out-of-band sends (the clarify card) land strictly after the streamed text that introduces them:
gateway/stream_consumer.py— newasync drain(timeout=5.0)plus a_DrainBarriersentinel (carrying athreading.Event) enqueued on the consumer'squeue.Queue.run()pops the barrier, flushes any pending accumulated text to the adapter, then sets the event. The blocking wait is offloaded to a thread executor so the shared event loop is never blocked (no same-loop deadlock).gateway/run.py—_clarify_callback_synccallsdrain()(bounded, ~6s cap) on the live consumer before schedulingsend_clarify. On any failure it falls through and still sends the card (pre-fix behavior preserved).Review hardening (3rd commit)
A code review flagged two bounded-but-real ~5s worker-thread stalls in unhappy paths, fixed in
eb30495:finally, so a popped barrier always resolves beforerun()returns or re-raises.stream_consumer_holder[0]is reset toNonewhen the consumer'srun()completes, so a clarify after stream completion no longer enqueues a barrier nobody drains.asyncio.get_event_loop()→get_running_loop()indrain().Scope
Touches only
gateway/stream_consumer.py,gateway/run.py, andtests/gateway/test_stream_consumer.py. No card-build, adapter, approval, or file-consent changes.Tests (run in isolation — repo has known cross-file pollution in the full suite)
tests/gateway/test_stream_consumer.py— 96 passed (incl. 4 new drain/barrier tests)tests/gateway/test_teams.py— 46 passedtests/plugins/platforms/teams/test_adapter_clarify.py— 26 passedVerification still pending
Live vision check on the pod after rollout (card lands below preamble) — to be run post-merge.