fix(tui): post-turn completion drain resolves ownership once; barrier test uses a watch

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

@@ -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)

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:

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