Bubfix: guarantee event order, harden chunk buffer, SSE-subscribe controller, and break adapter on terminal human_run - #3948
Merged
Conversation
…d human_interaction requests Root cause: the worker thread writes human_interaction events synchronously via SQLAlchemy in ask_user, while model_output_thinking/parse observer messages flow through the async consumer and are flushed only on a batched threshold (32 chunks or 250ms). When the worker suspends before that flush fires, human_interaction gains a lower event_seq number than the already-buffered observer chunks, causing the SSE replay stream to show them in the wrong order. Fix: replace the plain async-for consumer loop with a manual asyncio.wait iterator using a 50ms timeout. Once the worker finishes producing model output (i.e. right before ask_user), the loop times out and flushes any buffered observer chunks to the DB first, guaranteeing they precede the subsequent human_interaction row. Empty queue idle periods are essentially zero-cost; overall DB write frequency stays on par with the original.
…event order When the agent invokes ask_user, two independent write paths caused the human_interaction row to be persisted BEFORE model_output_thinking / parse chunks, breaking the SSE replay ordering: the worker thread writes HITL events synchronously via SQLAlchemy, while observer messages flow through the async consumer which only flushes on a batched threshold. Fix: introduce a thread-safe shared chunk buffer on RuntimeInteractionPort (port.add_chunk / port.take_chunks). The async consumer pushes every processed chunk there; the worker thread calls flush_chunks_until_idle() before dispatching any HITL event — it polls the shared buffer and waits for the async loop to drain the observer queue (20ms idle window, 500ms max wait), then persists every chunk in its own transaction. This guarantees chunk event_seq < human_interaction event_seq regardless of async scheduling latency. Also fix ImportError: openai 2.50 removed the httpx2 module. OpenAIModel now falls back from httpx2 to httpx at import time.
bernard1234
requested review from
Dallas98,
WMC001,
YehongPan,
hhhhsc701 and
jeffwu-1999
as code owners
September 17, 2026 06:37
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
…ttpx2-fallback paths Add unit tests for the RuntimeInteractionPort thread-safe chunk buffer and the flush_chunks_until_idle poll loop that guarantees observer chunks precede human_interaction events in DB order. All five HITL entry points (dispatch / boundary / receipt / finish / _wait_until_ready) are verified to invoke the idle flush before opening their transaction. Also add two tests for the openai_llm httpx2 → httpx ImportError fallback introduced to support openai >= 2.50 where the httpx2 shim was removed: one covers the fallback path, one confirms httpx2 still wins when present.
TiTi-iTiT
reviewed
Sep 17, 2026
TiTi-iTiT
reviewed
Sep 17, 2026
TiTi-iTiT
reviewed
Sep 17, 2026
TiTi-iTiT
reviewed
Sep 17, 2026
… peek_chunks, try/except safety Fix 4 real issues flagged by github-code-review: 1. Non-resettable hard_deadline in flush_chunks_until_idle — previously reset on every drain, meaning a model that kept producing chunks could stall the worker forever. deadline is now computed once at entry and the sleep call clips to hard_deadline - now. 2. _emit_in_flight Event bridges the async emit path and the worker's idle poll. Without this, buffer-empty = 'persisted' was confused with buffer-empty = 'taken for emit but still in run_blocking queue'. The worker now checks both 'buffer empty for settle_ms' AND 'no emit in flight' before deciding the async side is truly idle. 3. peek_chunks() replaces the take-put-back pattern in _flush_if_due. Previously the async loop drained the buffer, decided it was not yet due, then put everything back. That transiently-empty window (16 us normally, arbitrarily long under GIL/GC/preemption) was enough for the worker's 20 ms poll to mis-fire. We now peek (read count, no drain) and only take_chunks when we actually intend to persist. 4. emit_chunks wrapped in try/except that puts drained chunks back into the shared buffer before re-raising, and finish() wraps its flush call in try/except: pass. Guarantees (a) no chunk loss on DB failure and (b) the terminal human_run row is always written even if the flush step fails. Tests added: - 12 pure-mock unit tests in test_runtime_port_chunk_buffer.py cover hard_deadline, _emit_in_flight, peek_chunks, begin_emit/end_emit, try/except path, and every HITL entry-point's flush-before-transaction. - 1 async execute_attempt integration test in new test_application_execute_attempt.py drives the full consumer loop through _flush_if_due (peek → take → begin/end_emit) and the final flush, verifying that every patch line added in application.py is hit.
…hen disconnected When isRunning=true the EventSource already pushes human_interaction and human_execution events in real time — the 1.5s polling loop duplicated that work, hitting the DB and re-rendering the frontend for every tick. Disable polling entirely while the SSE stream is alive, and drop to 5s intervals only when the stream is closed (e.g. page load before the first run, or after a run finishes) so we can still discover WAITING_HUMAN requests that were created while the client was disconnected. Add isRunning to the useEffect dependency array so the polling cadence resets immediately when the SSE connection state changes.
…ff the write path
Frontend — /conversation polling → /{run_id}/events SSE:
- Replace the 5s conversation snapshot polling with a native EventSource
subscription to the backend's /{run_id}/events SSE stream. Discovery is
now one-shot: conversationId change and the agent stream pause
(isRunning true→false), the exact moment a HITL run is most likely to
exist. The SSE stream then keeps run state live with native auto-reconnect.
- Add dual guards inside refresh() to absorb the thundering herd from
adapter.onHumanInteractionEvent (fires once per HITL SSE chunk) plus
our own SSE effect: (1) in-flight dedupe — one snapshot absorbs all
concurrent callers and returns cached state; (2) 3s minimum interval
so bursts after the in-flight resolves do not immediately re-hit DB.
- Use a runRef mirror so refresh() stays stable and downstream effects do
not re-run on every snapshot.
- Detect terminal status inside SSE onmessage and proactively es.close()
to prevent EventSource from reconnecting forever against COMPLETED runs.
Backend — snapshot off the write path:
- Add repository.read_only() context manager: plain SELECT without
WITH FOR UPDATE, no transaction, no flush, no _expire scan. Pure reads
must not contend with worker writes on the same row lock.
- Add service.light_snapshot() using read_only. Retain snapshot() as a
writer-path API for any future lock-held callers.
- Route conversation_snapshot, run snapshot endpoint, and both snapshot
calls inside stream_run() through light_snapshot.
- Move expiration to the writer path: decide() still calls _expire inline
before processing each request, and expire_waiting() remains the
periodic scheduler sweep.
Impact: conversation snapshot calls drop from 12+/min (polling) or
10+/s (burst from adapter + SSE) to at most one every 3s. Each call is
now two plain SELECTs instead of a lock-held transaction with a possible
write from _expire. Read and write paths are fully decoupled.
…sRunning flips false After a HITL run reaches FAILED/COMPLETED, Assistant-UI's isRunning stayed true — the stop button remained visible and new messages went into the queue buffer instead of being sent normally. The root cause is that isRunning is driven entirely by the ChatModelRun generator lifetime, which only returns when the backend SSE HTTP connection closes (reader.read() -> done=true). The backend stream_run loop can hang on heartbeat even after the run is terminal when the SSE was opened during WAITING_HUMAN with attempt_active=true: the break condition requires both cursor >= event_seq AND (terminal status OR WAITING_HUMAN with attempt_active=false and empty rows). If continueHitl fires mid-flight with a stale after_event, the cursor never catches up, so the SSE stays alive forever and the generator never returns. Stop depending on the backend closing first. Inside the adapter's SSE chunk loop, detect a terminal human_run event (status in COMPLETED, FAILED, STOPPED, EXPIRED), set a hitlTerminal flag, break the inner for-loop, and let the outer while-loop exit via the same flag on the next iteration. Assistant-UI sees the generator return and flips isRunning false immediately. Only affects HITL streams — the normal non-HITL agent path never emits human_run events so this branch is never taken.
…th refactor test_human_interaction_app.py still mocked service.snapshot after commit 288ae4e moved conversation_snapshot and the run snapshot endpoint to service.light_snapshot (read-only path, no lock, no _expire). The fixture return_value and the two assert_called_once_with/assert_not_called assertions all referenced the old method name, causing CI to fail because MagicMock.snapshot was never called.
bernard1234
changed the title
Codecov reported 70.43% patch coverage (target 90%) because new error and race paths in the HITL changes had no tests. Add mocked unit tests for: leftover chunk flush in execute_attempt's finally block before the failed finish, CancelledError scope and stop-event fallbacks, RunTerminated finish race, recovery-required outcome, and chunk iterator aclose failure tolerance; runtime_port in-flight emit busy detection, chunk restoration when emit_chunks raises, and terminal status persistence on flush failure; light_snapshot/read_only service behavior with signed tenant and user scoping; and the httpx fallback when openai._base_client.httpx2 is absent. Measured locally with CI-equivalent per-file pytest isolation: patch coverage 202/202 = 100%.
Merge explanatory inline comments into docstrings, keep single-line comments for inline notes, convert TypeScript block notes to JSDoc, and drop banner/separator lines. Comment-level changes only, no behavior change.
Merge explanatory inline comments into docstrings, keep single-line comments for inline notes, convert TypeScript block notes to JSDoc, and drop banner/separator lines. Comment-level changes only, no behavior change.
…lication gate SonarCloud failed the quality gate with new_duplicated_lines_density=5.2% (threshold 3%), caused solely by test_application_execute_attempt.py: the inline _Port stub in the flush test and the one in _run_execute_attempt duplicated ~69 lines (2 CPD blocks, 14.4% file density). Extract a shared _build_port_class/_make_port_factory plus a _patched_application context manager and _execute_attempt_args so both call sites reuse a single definition; drop dead code (last_flush, install/monkeypatches, unused imports) and fix the latent bare-contextmanager NameError by using contextlib.contextmanager. No behavioral change; all 8 tests pass.
jeffwu-1999
approved these changes
Sep 18, 2026
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
3 participants
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.
human_interaction row to be persisted BEFORE model_output_thinking / parse
chunks, breaking the SSE replay ordering: the worker thread writes HITL
events synchronously via SQLAlchemy, while observer messages flow through
the async consumer which only flushes on a batched threshold.
Fix: introduce a thread-safe shared chunk buffer on RuntimeInteractionPort
(port.add_chunk / port.take_chunks). The async consumer pushes every
processed chunk there; the worker thread calls flush_chunks_until_idle()
before dispatching any HITL event — it polls the shared buffer and waits
for the async loop to drain the observer queue (20ms idle window, 500ms
max wait), then persists every chunk in its own transaction. This
guarantees chunk event_seq < human_interaction event_seq regardless of
async scheduling latency.