From 7f80efc294e420fc6fdcb5af36db693c73d7084e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 7 Sep 2026 07:53:38 -0700 Subject: [PATCH] fix(tui): post-turn completion drain resolves ownership once; barrier test uses a watch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebasing onto main surfaced three reds in tests/test_tui_gateway_server.py. Two were a real double-check: _run_post_turn_followups already drains the completion queue through _session_owns_notification_event, then handed the events to _notif_handle_ready, which re-ran the belongs-elsewhere and requires-owner gates on every event — a second compression-lineage DB lookup per notification for events the caller had just proven ours. _notif_handle_ready/_notif_handle_event take an explicit owned=True from the post-turn path and skip only the ownership gates (consumed/dedup/turn claiming still run). The poller path is unchanged (owned defaults False). The third, test_run_prompt_submit_requeues_all_unstarted_notifications_with_real_threading, asserted that three consecutive completions produce one in-flight turn plus two requeued events; under this PR's contract (#104671) consecutive completions share one turn, so the first event is now a watch_match — the documented turn barrier — and the test keeps proving that the two completions behind an in-flight notification turn are never lost. A/B: revert the tui_gateway hunks and the two lineage tests fail with ownership_checks recorded twice; restore and all 11 test_run_prompt_submit tests pass. --- tests/test_tui_gateway_server.py | 3 +++ tui_gateway/prompt_turn.py | 2 +- tui_gateway/session_notifications.py | 13 +++++++------ 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 4c758cca36..3cf2e25bd6 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -7564,6 +7564,9 @@ def test_run_prompt_submit_requeues_all_unstarted_notifications_with_real_thread } for index in range(1, 4) ] + # Consecutive completions share one turn (#104671); a watch_match is a turn + # barrier, so it is the in-flight turn behind which batch_2/batch_3 must survive. + events[0].update(type="watch_match", pattern="owned-1") isolated_queue: _queue_mod.Queue = _queue_mod.Queue() for event in events: isolated_queue.put(event) diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index 30a95ddcdb..0b82e37ef6 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -400,7 +400,7 @@ def _run_post_turn_followups( _notif_handle_ready( sid, session, [event for event, _text in drained], session.setdefault("_notification_emitted", set()), process_registry, - format_process_notification, deferred) + format_process_notification, deferred, owned=True) for event in deferred: process_registry.completion_queue.put(event) except Exception as _drain_exc: diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 3e196f7233..246880691c 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -394,21 +394,22 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None complete_event_delivery(evt, claim) -def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred, completions=None) -> bool: +def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred, completions=None, *, owned=False) -> bool: """Route one dequeued event: foreign (another live session owns it) → requeued, or onto ``deferred`` during the shutdown drain; unowned (addressed but unprovable — never adopt an orphan) → dropped, except delegation payloads deferred for a resume; ours (or ownerless legacy, kept process-global) → status.update once, then an agent turn if - idle. False = the drain must stop (session busy).""" + idle. False = the drain must stop (session busy). ``owned`` skips the ownership gates for events a caller already + drained through ``_session_owns_notification_event`` (the post-turn safety net), so lineage resolves once.""" queue = registry.completion_queue evt_type, is_delegation = evt.get("type", "completion"), evt.get("type") == "async_delegation" - if _notification_event_belongs_elsewhere(sid, session, evt): + if not owned and _notification_event_belongs_elsewhere(sid, session, evt): if deferred is not None: deferred.append(evt) else: # otherwise a process started in session A surfaces in whichever poller wakes first queue.put(evt) time.sleep(0.1) return True - if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt): + if not owned and _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt): origin, key = str(evt.get("origin_ui_session_id") or ""), str(evt.get("session_key") or "") if deferred is None: (logger.warning if is_delegation else logger.debug)( @@ -474,14 +475,14 @@ def _notif_dispatch_completions(sid, session, notifications, registry, deferred) complete_event_delivery(event, claim) -def _notif_handle_ready(sid, session, events, emitted, registry, fmt, deferred): +def _notif_handle_ready(sid, session, events, emitted, registry, fmt, deferred, *, owned=False): """One ready snapshot: ownership and UI emission per event, one turn per completion run.""" completions = [] for index, event in enumerate(events): if event.get("type", "completion") != "completion": _notif_dispatch_completions(sid, session, completions, registry, deferred) completions = [] - if not _notif_handle_event(sid, session, event, emitted, registry, fmt, deferred, completions): + if not _notif_handle_event(sid, session, event, emitted, registry, fmt, deferred, completions, owned=owned): for remaining in events[index + 1:]: (deferred.append if deferred is not None else registry.completion_queue.put)(remaining) break