6krrt/plans/tui-live-routing-panel-sse-fix-review.md

Review: re-fix for finding #5 (async SSE) + finding #9 test gap

Status: done -- review of shipped work

What it was reviewing: the uncommitted working tree on top of 943b2d5, per opencode's summary claiming to (1) make _decision_event_stream a genuine async def generator against asyncio.Queue — the fix tui-live-routing-panel-fixes-review.md said finding #5 still needed — and (2) close the test-coverage gap that same report opened around the finding #9 fix. 7 files touched (events.py, dispatcher.py, tui_screens.py, README.md, three test files), +/- not yet committed. Reviewed via /code-review high (8 parallel finder angles) plus independent verification of every finding against the actual files and, for the two claims below, against live reproductions rather than static reading alone. Full suite: 622/622 passing (tests/test_tui.py + tests/test_events.py + tests/test_metrics_endpoint.py: 54/54).

Verdict: #9 fixed and verified; #5 not fixed — the new mechanism is broken, worse than the sync-threadpool problem it replaced

#9 — fixed and verified

test_live_decision_existing_bucket_defers_count_update (tests/test_tui.py:805) does exactly what the prior report asked for: posts a live decision into a bucket that already has a row, asserts the count is unchanged immediately after, then asserts it updates after the next _on_interval (/metrics) poll. Ran it in isolation — passes, and fails as expected if the deferred-update guard is bypassed by hand. No further action needed here.

#5 — not fixed: the new async bridge silently drops every live decision in production

Confirmed by direct reproduction, not just reading. events.publish_decision (events.py:79-84) bridges to _sse_subscribers via:

try:
    loop = asyncio.get_event_loop()
    for q in _sse_subscribers:
        loop.call_soon_threadsafe(q.put_nowait, decision)
except RuntimeError:
    pass

asyncio.get_event_loop() only returns a usable loop when called from the thread that has one running or set. Every real caller of publish_decision is persist_route_decision (dispatcher.py:820), called from route_endpoint, dispatch_endpoint, and chat_completions — all plain def, not async def (confirmed at dispatcher.py:1543, 1968, 2530), so FastAPI/Starlette runs them on an anyio worker thread, not the thread serving /events/decisions. Reproduced directly on this repo's Python 3.14 venv:

$ python3 -c "
import asyncio, threading
def worker():
    try: asyncio.get_event_loop()
    except RuntimeError as e: print('RuntimeError:', e)
threading.Thread(target=worker).start()
"
RuntimeError: There is no current event loop in thread 'Thread-1 (worker)'.

And end-to-end, simulating the actual production shape (event loop on the main thread serving a subscriber queue, publish_decision called from a separate thread exactly as persist_route_decision does it):

async def main():
    q = asyncio.Queue()
    events.subscribe_sse(q, replay=False)
    threading.Thread(target=lambda: events.publish_decision({"id": 1})).start()
    ...join...
    await asyncio.wait_for(q.get(), timeout=1.0)   # -> asyncio.TimeoutError

Times out every time — the decision never arrives. The except RuntimeError: pass swallows it with no log, so this fails silently: the dashboard shows the initial replay burst and then goes permanently idle (heartbeats only) for the rest of the process's life. This is strictly worse than the finding-#5 report's original complaint (thread-pool pinning) — that version at least delivered decisions.

Why the tests don't catch it: no test calls publish_decision from a plain thread while checking delivery into an asyncio.Queue. test_concurrent_publish_subscribe_unsubscribe_raises_no_error (tests/test_events.py:90) does call publish_decision from real threading.Threads, but never subscribes anything to _sse_subscribers and never checks delivery — it only asserts no exception escaped, and the RuntimeError is already swallowed inside publish_decision before it could. test_full_subscriber_eviction_terminates_decision_event_stream (tests/test_events.py:118) is @pytest.mark.anyio and bypasses publish_decision entirely (monkeypatch.setattr(events, "subscribe_sse", fake_subscribe_sse), then await q.put(...) directly) — the one context where get_event_loop() would have worked is also the one test that never calls the function under test.

Fix direction: capture the loop once at subscribe time (e.g. asyncio.get_running_loop() inside subscribe_sse, called from the async def endpoint where it's valid, stored alongside the queue) rather than calling get_event_loop() from the publisher's thread. call_soon_threadsafe already needs a loop reference callable from any thread — it just needs to be the right loop's reference, obtained once from a context that has one.

Secondary issues found in the same code (real, but currently masked by #5)

1. Unbounded asyncio.Queue, no backpressure — dispatcher.py:1308 creates asyncio.Queue() with no maxsize, and nothing in the SSE path ever hits QueueFull. The old queue.Queue-based design this replaced bounded each subscriber at DEFAULT_QUEUE_SIZE=100 and evicted slow consumers (still true for the now-unused subscribe/_subscribers path). A stalled dashboard connection has no cap on the SSE side — unbounded per-connection memory growth once #5 is fixed and decisions actually flow.

2. _sse_subscribers iterated without the lock that guards its mutation — events.py:80, for q in _sse_subscribers:, runs outside _subscribers_lock, while subscribe_sse/unsubscribe_sse/clear all mutate the same set under that lock. A connect/disconnect racing a publish can raise RuntimeError: Set changed size during iteration — currently unreachable in production only because get_event_loop() already raises before this loop runs, so this needs fixing in the same pass as #5, not after.

3. Old thread-based subscriber path (subscribe/unsubscribe/_subscribers) is now dead in production — dispatcher.py no longer calls events.subscribe/events.unsubscribe anywhere (confirmed by grep); only tests still exercise it. Two independent fan-out implementations now have to be kept in sync by hand, and they've already diverged: the dead one has real bounding/eviction, the live one (once #5 is fixed) doesn't (see #1 above).

4. Dead sentinel write in _decision_event_stream's finally — dispatcher.py:1326-1329 does queue.put_nowait(events._EVICTED) immediately before unsubscribe_sse(queue). By the time finally runs, this generator's own while True loop has already exited, and no other code ever reads this particular queue — nothing will ever consume the sentinel. Leftover from the old design where eviction needed to wake a different consumer; harmless but confusing cruft.

5. queue (the stdlib module) is shadowed by a same-named local variable in both dispatcher.py:1308 (_decision_event_stream) and events.py:110,134 (subscribe_sse/unsubscribe_sse's parameter). In dispatcher.py, import queue (line 40) is now unused everywhere else in the file — confirmed via grep, no remaining queue.Full/queue.Empty/ queue.Queue( call sites. Low severity, but a future edit adding queue.Full-style handling inside either function would silently resolve to the local asyncio.Queue instead and produce a confusing AttributeError.

6. Docstring guarantee dropped — events_decisions()'s docstring lost the line "Unauthenticated and loopback-bound like /metrics; each event carries only the fields already on a route_decisions row — no conversation text, prompt, or session_dir" with no replacement. The guarantee still holds in the actual payload (checked persist_route_decision's dict) — this is a documentation regression, not a behavior one, but it was the one comment warning a future editor not to add such a field at the point that actually emits it.

One claim checked and refuted

The /code-review pass also flagged tui_screens.py's on_key (narrowed to escape-only, dropping "q"/"enter") as letting q fall through to the app-level ("q", "quit", "Quit") binding and quit the whole dashboard while the popup is open. Checked this directly against the installed Textual 8.2.8: App._check_bindings resolves non-priority keys (this app's bindings are all plain tuples, so none are priority=True) via Screen._modal_binding_chain, which explicitly truncates the chain at the first modal screen — it never reaches the App bindings underneath. Verified empirically with run_test(): pressing q with DecisionDetailScreen open leaves the app running and the modal open (it also doesn't close the modal, since q was dropped from on_key — you just can't close it with q anymore, only escape). No quit-while-modal-open bug. This tui_screens.py diff isn't new work from this round anyway — it's the same uncommitted change already assessed twice as fine in the two prior reports.

What's solid

• Finding #9's fix is correctly and completely closed, with a test that would fail if the deferred-update guard regressed.
• The intent behind the #5 attempt — a genuine async generator instead of a threadpool-blocking one — is the right direction; the bridging mechanism (asyncio.get_event_loop() called from the publisher's thread) is the one piece that's wrong, not the overall design.
• _sse_data() reuse and the rest of 943b2d5's prior fixes (#1-4, #7) are untouched by this round and remain correct.

Recommendation

Don't ship this round's #5 attempt as resolving #5 — it currently makes the live feed silently non-functional rather than merely thread-pool-expensive. Fix the loop-capture bug (capture the loop once, from async def events_decisions() or inside subscribe_sse called from that context, not via get_event_loop() in the publisher thread), add a maxsize + eviction/backpressure policy to the SSE queue to match what the old path had, and move the _sse_subscribers iteration in publish_decision inside _subscribers_lock. Once those three land, add a test that exercises the real shape — publish_decision called from a plain thread, delivery checked on the subscribed asyncio.Queue — so this doesn't regress silently again. The dead old subscriber path, the dead sentinel write, and the queue shadowing are all cleanup, not urgent, and can ride along with the same pass. The docstring guarantee is a one-line restore.