From 6383e2f1cd2d9299d37dc9ac9c94113ca46fd695 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:18:07 -0700
Subject: [PATCH 1/4] =?UTF-8?q?refactor(tui):=20session=5Fnotifications=20?=
 =?UTF-8?q?=E2=80=94=20kanban=20text=20dispatch=20table,=20shared=20turn-c?=
 =?UTF-8?q?laim/ownership=20helpers?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tui_gateway/session_notifications.py | 223 ++++++++++++++-------------
 1 file changed, 112 insertions(+), 111 deletions(-)

diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py
index 2cb14a14c3..f03be675f4 100644
--- a/tui_gateway/session_notifications.py
+++ b/tui_gateway/session_notifications.py
@@ -101,9 +101,7 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
     if not evt_key:
         return False
     current_keys = _notif_current_keys(sid, session)
-    if evt_key in current_keys:
-        return True
-    return _notif_resolve_event_key(evt_key) in current_keys
+    return evt_key in current_keys or _notif_resolve_event_key(evt_key) in current_keys
 
 
 def _notification_event_requires_owner(evt: dict) -> bool:
@@ -135,7 +133,6 @@ def _notification_event_dedup_key(evt: dict) -> tuple:
 # the cursor advances past them and they can't wedge a later completed/blocked
 # event behind an unclaimed row.
 _KANBAN_NOTIFY_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked")
-_KANBAN_SILENT_KINDS = frozenset({"archived", "unblocked"})
 _KANBAN_POLL_SECONDS = 5.0
 _LOOP_POLL_SECONDS = 5.0
 
@@ -145,6 +142,15 @@ def _notif_release_turn(session: dict) -> None:
         session["running"] = False
 
 
+def _notif_claim_turn(session: dict) -> bool:
+    """Claim the idle session (running=True) under history_lock; False if a turn is live."""
+    with session["history_lock"]:
+        if session.get("running"):
+            return False
+        session["running"] = True
+        return True
+
+
 def _notif_log_failure(what: str, exc: BaseException) -> None:
     print(f"[tui_gateway] {what}: {type(exc).__name__}: {exc}", file=sys.stderr)
 
@@ -164,11 +170,8 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
     mgr = LoopManager(session_id=sid_key)
     if not mgr.is_due() or goal_blocks_loop_tick(sid_key):
         return
-
-    with session["history_lock"]:
-        if session.get("running"):
-            return  # busy — stays due, next poll retries
-        session["running"] = True
+    if not _notif_claim_turn(session):
+        return  # busy — stays due, next poll retries
 
     wakeup = mgr.fire_tick()
     if not wakeup:
@@ -196,11 +199,9 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
                 if payload.get("type") == "send" and payload.get("message"):
                     # Command resolves to a prompt (skill command etc.) — run it
                     # as a normal turn; the post-turn hook completes the tick.
-                    with session["history_lock"]:
-                        if session.get("running"):
-                            mgr.abandon_tick()
-                            return
-                        session["running"] = True
+                    if not _notif_claim_turn(session):
+                        mgr.abandon_tick()
+                        return
                     _emit("message.start", sid)
                     _run_prompt_submit(rid, sid, session, payload["message"])
                     return
@@ -219,45 +220,64 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
             mgr.abandon_tick()
 
 
+def _kb_first_line(value: Any, limit: int) -> str:
+    lines = str(value).strip().splitlines()
+    return f"\n{lines[0][:limit]}" if lines else ""
+
+
+def _kb_completed(task, payload: dict, title: str) -> str:
+    if payload.get("summary"):
+        handoff = _kb_first_line(payload["summary"], 200)
+    elif getattr(task, "result", None):
+        handoff = _kb_first_line(task.result, 160)
+    else:
+        handoff = ""
+    return f" done — {title}{handoff}"
+
+
+def _kb_blocked(task, payload: dict, title: str) -> str:
+    reason = f": {str(payload.get('reason'))[:160]}" if payload.get("reason") else ""
+    return f" blocked{reason}"
+
+
+def _kb_gave_up(task, payload: dict, title: str) -> str:
+    err = f"\n{str(payload.get('error'))[:200]}" if payload.get("error") else ""
+    return f" gave up after repeated spawn failures{err}"
+
+
+def _kb_timed_out(task, payload: dict, title: str) -> str:
+    try:
+        limit = int(payload.get("limit_seconds") or 0)
+    except (TypeError, ValueError):
+        limit = 0
+    return f" timed out (max_runtime={limit}s); will retry"
+
+
+# kind -> (glyph, suffix formatter); silent kinds (archived/unblocked) are absent → None.
+_KANBAN_EVENT_FORMATTERS = {
+    "completed": ("✔", _kb_completed),
+    "blocked": ("⏸", _kb_blocked),
+    "gave_up": ("✖", _kb_gave_up),
+    "crashed": ("✖", lambda task, payload, title: " worker crashed (pid gone); dispatcher will retry"),
+    "timed_out": ("⏱", _kb_timed_out),
+    "status": ("🔄", lambda task, payload, title: f" → {payload.get('status') or ''}"),
+}
+
+
 def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[str]:
     """Single-line notification text for one kanban event; wording mirrors
     gateway/kanban_watchers.py so it reads the same as on Telegram. None for silent kinds."""
-    kind = getattr(ev, "kind", "")
-    if not kind or kind in _KANBAN_SILENT_KINDS:
+    entry = _KANBAN_EVENT_FORMATTERS.get(getattr(ev, "kind", ""))
+    if entry is None:
         return None
+    glyph, fmt = entry
     task_id = sub.get("task_id", "")
     title = (getattr(task, "title", None) or task_id)[:120]
     board_tag = f"[{board_slug}] " if board_slug else ""
     who = getattr(task, "assignee", None) or ""
     tag = f"@{who} " if who else ""
     payload = getattr(ev, "payload", None) or {}
-    if kind == "completed":
-        handoff = ""
-        summary = payload.get("summary")
-        if summary:
-            lines = str(summary).strip().splitlines()
-            handoff = f"\n{lines[0][:200]}" if lines else ""
-        elif getattr(task, "result", None):
-            lines = str(task.result).strip().splitlines()
-            handoff = f"\n{lines[0][:160]}" if lines else ""
-        return f"✔ {board_tag}{tag}Kanban {task_id} done — {title}{handoff}"
-    if kind == "blocked":
-        reason = f": {str(payload.get('reason'))[:160]}" if payload.get("reason") else ""
-        return f"⏸ {board_tag}{tag}Kanban {task_id} blocked{reason}"
-    if kind == "gave_up":
-        err = f"\n{str(payload.get('error'))[:200]}" if payload.get("error") else ""
-        return f"✖ {board_tag}{tag}Kanban {task_id} gave up after repeated spawn failures{err}"
-    if kind == "crashed":
-        return f"✖ {board_tag}{tag}Kanban {task_id} worker crashed (pid gone); dispatcher will retry"
-    if kind == "timed_out":
-        try:
-            limit = int(payload.get("limit_seconds") or 0)
-        except (TypeError, ValueError):
-            limit = 0
-        return f"⏱ {board_tag}{tag}Kanban {task_id} timed out (max_runtime={limit}s); will retry"
-    if kind == "status":
-        return f"🔄 {board_tag}{tag}Kanban {task_id} → {payload.get('status') or ''}"
-    return None
+    return f"{glyph} {board_tag}{tag}Kanban {task_id}{fmt(task, payload, title)}"
 
 
 def _collect_kanban_notifications(session: dict) -> list:
@@ -324,10 +344,7 @@ def _collect_kanban_notifications(session: dict) -> list:
                 if not events:
                     continue
                 task = _kb.get_task(conn, sub["task_id"])
-                for ev in events:
-                    text = _format_kanban_event_text(sub, task, ev, slug)
-                    if text:
-                        texts.append(text)
+                texts.extend(t for t in (_format_kanban_event_text(sub, task, ev, slug) for ev in events) if t)
                 # Unsubscribe only on archive: ``done`` is reversible in review/controller flows,
                 # so keeping the sub lets a later reopen notify the same session. The claimed
                 # cursor prevents replay.
@@ -348,9 +365,9 @@ def _notif_poll_kanban(sid: str, session: dict) -> None:
     except Exception as _kb_exc:
         _notif_log_failure("kanban notification poll failed", _kb_exc)
         _kanban_texts = []
+    for _kb_text in _kanban_texts:
+        _emit("status.update", sid, {"kind": "process", "text": _kb_text})
     if _kanban_texts:
-        for _kb_text in _kanban_texts:
-            _emit("status.update", sid, {"kind": "process", "text": _kb_text})
         session.setdefault("_kanban_pending", []).extend(_kanban_texts)
     _pending = session.get("_kanban_pending") or []
     if not _pending:
@@ -362,10 +379,9 @@ def _notif_poll_kanban(sid: str, session: dict) -> None:
             _batch = list(_pending)
             session["_kanban_pending"] = []
     if _batch:
-        rid = f"__notif__{int(time.time() * 1000)}"
         try:
             _emit("message.start", sid)
-            _run_prompt_submit(rid, sid, session, "\n".join(_batch))
+            _run_prompt_submit(f"__notif__{int(time.time() * 1000)}", sid, session, "\n".join(_batch))
         except Exception as exc:
             _notif_log_failure("kanban notification dispatch failed", exc)
             _notif_release_turn(session)
@@ -397,13 +413,11 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None
         return
     try:
         _emit("message.start", sid)
-        if evt.get("type") == "async_delegation":
-            _run_prompt_submit(
-                rid, sid, session, text, display_kind="async_delegation_complete",
-                display_metadata=_async_delegation_display_metadata(evt),
-            )
-        else:
-            _run_prompt_submit(rid, sid, session, text)
+        kwargs = (
+            {"display_kind": "async_delegation_complete", "display_metadata": _async_delegation_display_metadata(evt)}
+            if evt.get("type") == "async_delegation" else {}
+        )
+        _run_prompt_submit(rid, sid, session, text, **kwargs)
         complete_event_delivery(evt, claim)
     except Exception as exc:
         release_event_delivery(evt, claim)
@@ -411,6 +425,16 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None
         _notif_release_turn(session)
 
 
+def _notif_classify_event(sid: str, session: dict, evt: dict) -> str:
+    """'foreign' (another live session owns it), 'unowned' (addressed but unprovable — never adopt
+    an orphan), or 'mine' (ours, or an ownerless legacy notification kept process-global)."""
+    if _notification_event_belongs_elsewhere(sid, session, evt):
+        return "foreign"
+    if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
+        return "unowned"
+    return "mine"
+
+
 def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None:
     """Poll completion_queue and dispatch notifications autonomously (daemon thread started by
     _init_session()): emit a status.update (kind=process), then chain an agent turn via
@@ -420,6 +444,7 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
     ``_KANBAN_POLL_SECONDS`` — the delivery path for platform="tui" rows."""
     from tools.process_registry import process_registry, format_process_notification
 
+    queue = process_registry.completion_queue
     _emitted = set()  # dedup re-queued events so one completion isn't emitted 50 times while busy
     _last_kanban_poll = 0.0
     _last_loop_poll = 0.0
@@ -437,21 +462,18 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
             _last_kanban_poll = _now
             _notif_poll_kanban(sid, session)
         try:
-            evt = process_registry.completion_queue.get(timeout=0.5)
+            evt = queue.get(timeout=0.5)
         except Exception:
             continue
 
-        # Leave foreign events for their owner — otherwise a process started in
-        # session A surfaces its completion in whichever poller wakes first.
-        if _notification_event_belongs_elsewhere(sid, session, evt):
-            process_registry.completion_queue.put(evt)
+        owner = _notif_classify_event(sid, session, evt)
+        if owner == "foreign":
+            # Leave foreign events for their owner — otherwise a process started in
+            # session A surfaces its completion in whichever poller wakes first.
+            queue.put(evt)
             time.sleep(0.1)
             continue
-
-        # Not owned by another LIVE session, but addressed events still need positive proof
-        # (exact UI origin, direct durable key, or compression lineage) — an orphan must not be
-        # adopted by this chat. Truly ownerless ordinary notifications keep legacy global delivery.
-        if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
+        if owner == "unowned":
             log = logger.warning if evt.get("type") == "async_delegation" else logger.debug
             log(
                 "Dropping unowned %s notification (origin=%r key=%r) instead of delivering to session %s",
@@ -463,59 +485,45 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
         text = _notif_render_event(sid, evt, _emitted, process_registry, format_process_notification)
         if not text:
             continue
-
-        _requeued = False
-        with session["history_lock"]:
-            if session.get("running"):
-                process_registry.completion_queue.put(evt)
-                _requeued = True
-            else:
-                session["running"] = True
-        if _requeued:
+        if not _notif_claim_turn(session):
             # Back off: the re-queued event keeps the queue non-empty, so
             # without a sleep this loop spins at 100% CPU while busy.
+            queue.put(evt)
             time.sleep(0.25)
             continue
-
         _notif_dispatch_event(sid, session, evt, text)
 
     # Drain remaining events after the stop signal so nothing is lost on shutdown. Other live
     # sessions' events are set aside and re-queued; orphaned events (owner gone) are dropped —
     # same guard as the main loop, except orphaned delegation payloads are deferred for a resume.
     deferred: list = []
-    while not process_registry.completion_queue.empty():
+    while not queue.empty():
         try:
-            evt = process_registry.completion_queue.get_nowait()
+            evt = queue.get_nowait()
         except Exception:
             break
-        if _notification_event_belongs_elsewhere(sid, session, evt):
+        owner = _notif_classify_event(sid, session, evt)
+        if owner == "foreign" or (owner == "unowned" and evt.get("type") == "async_delegation"):
             deferred.append(evt)
             continue
-        if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
-            if evt.get("type") == "async_delegation":
-                deferred.append(evt)
-            else:
-                logger.debug(
-                    "Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
-                    evt.get("type", "completion"), str(evt.get("origin_ui_session_id") or ""),
-                    str(evt.get("session_key") or ""),
-                )
+        if owner == "unowned":
+            logger.debug(
+                "Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
+                evt.get("type", "completion"), str(evt.get("origin_ui_session_id") or ""),
+                str(evt.get("session_key") or ""),
+            )
             continue
         text = _notif_render_event(sid, evt, _emitted, process_registry, format_process_notification)
         if not text:
             continue
-
-        with session["history_lock"]:
-            if session.get("running"):
-                process_registry.completion_queue.put(evt)
-                break
-            session["running"] = True
-
+        if not _notif_claim_turn(session):
+            queue.put(evt)
+            break
         _notif_dispatch_event(sid, session, evt, text)
 
     # Hand any other sessions' events back to the shared queue.
     for evt in deferred:
-        process_registry.completion_queue.put(evt)
+        queue.put(evt)
 
 
 def _async_delegation_display_metadata(evt: dict) -> dict:
@@ -542,11 +550,6 @@ def _wire_agent_terminal_output() -> None:
     `_emit` is `_stdout_lock`-guarded, so the registry's reader threads may call it."""
     from tools.process_registry import process_registry
 
-    has_output_sink = getattr(process_registry, "on_output", None) is not None
-    has_close_sink = getattr(process_registry, "on_close", None) is not None
-    if has_output_sink and has_close_sink:
-        return
-
     def _owner_sid_for_process(session) -> str:
         session_key = str(getattr(session, "session_key", "") or "")
         if not session_key:
@@ -557,18 +560,16 @@ def _wire_agent_terminal_output() -> None:
                     return sid
         return ""
 
-    def _emit_agent_terminal_output(session, chunk):
-        _emit("agent.terminal.output", _owner_sid_for_process(session), {"process_id": session.id, "chunk": chunk})
-
     def _emit_agent_terminal_close(session, process_id):
         # session may be None (process already finished/pruned) — the tab can
         # still linger and be closed; route to the owning window when we can.
-        sid = _owner_sid_for_process(session) if session is not None else ""
-        _emit("terminal.close", sid, {"process_id": process_id})
+        _emit("terminal.close", _owner_sid_for_process(session) if session is not None else "", {"process_id": process_id})
 
-    if not has_output_sink:
-        process_registry.on_output = _emit_agent_terminal_output
-    if not has_close_sink:
+    if getattr(process_registry, "on_output", None) is None:
+        process_registry.on_output = lambda session, chunk: _emit(
+            "agent.terminal.output", _owner_sid_for_process(session), {"process_id": session.id, "chunk": chunk}
+        )
+    if getattr(process_registry, "on_close", None) is None:
         process_registry.on_close = _emit_agent_terminal_close
 
 

From 765599a1a65fcfd218a3308e539163f5a765769f Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:25:33 -0700
Subject: [PATCH 2/4] =?UTF-8?q?refactor(tui):=20session=5Fcompression=20?=
 =?UTF-8?q?=E2=80=94=20collapse=20defensive=20try/except=20layers,=20flatt?=
 =?UTF-8?q?en=20partial-compress=20split?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tui_gateway/session_compression.py   |  78 +++----
 tui_gateway/session_notifications.py | 300 +++++++++++++--------------
 2 files changed, 167 insertions(+), 211 deletions(-)

diff --git a/tui_gateway/session_compression.py b/tui_gateway/session_compression.py
index 61b933f64c..f0fb53a2f8 100644
--- a/tui_gateway/session_compression.py
+++ b/tui_gateway/session_compression.py
@@ -57,9 +57,7 @@ def _derived_default_threshold_percent(agent: Any, compression: dict) -> float:
         model = getattr(agent, "model", "") or ""
         provider = getattr(agent, "provider", "") or ""
         autoraise_enabled = str(compression.get("codex_gpt55_autoraise", True)).lower() in {"true", "1", "yes"}
-        model_cthresh = _compression_threshold_for_model(
-            model, provider, allow_codex_gpt55_autoraise=autoraise_enabled
-        )
+        model_cthresh = _compression_threshold_for_model(model, provider, allow_codex_gpt55_autoraise=autoraise_enabled)
         pct, _notice = _resolve_compression_threshold(
             pct, model_cthresh, model=model,
             is_codex_autoraise=_is_codex_gpt54_or_gpt55(model, provider) or _is_codex_spark(model, provider),
@@ -95,10 +93,7 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
     model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {}
 
     enabled_raw = compression.get("enabled", True)
-    if isinstance(enabled_raw, bool):
-        agent.compression_enabled = enabled_raw
-    else:
-        agent.compression_enabled = str(enabled_raw).lower() in {"true", "1", "yes"}
+    agent.compression_enabled = enabled_raw if isinstance(enabled_raw, bool) else str(enabled_raw).lower() in {"true", "1", "yes"}
 
     agent.codex_responses_native_compaction = is_truthy_value(compression.get("codex_responses_native", False))
     native_threshold_raw = compression.get("codex_responses_compact_threshold", 200_000)
@@ -114,11 +109,8 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
     agent.codex_responses_compact_threshold = native_threshold
 
     # Absence restores the agent_init/config default (0 = disabled).
-    idle_raw = compression.get("idle_compact_after_seconds", 0)
-    try:
-        agent.compression_idle_compact_after_seconds = max(0, int(idle_raw or 0))
-    except (TypeError, ValueError):
-        pass
+    with contextlib.suppress(TypeError, ValueError):
+        agent.compression_idle_compact_after_seconds = max(0, int(compression.get("idle_compact_after_seconds", 0) or 0))
 
     cc = getattr(agent, "context_compressor", None)
     if cc is None:
@@ -138,32 +130,25 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
             continue
         setattr(cc, key, max(min_value, value))
 
-    try:
+    with contextlib.suppress(TypeError, ValueError):
         ratio_raw = compression.get("target_ratio", _compressor_ctor_default("summary_target_ratio", 0.20))
         cc.summary_target_ratio = max(0.10, min(float(ratio_raw), 0.80))
-    except (TypeError, ValueError):
-        pass
 
+    # Absent or invalid shape (agent_init treats both as empty): stale overrides must stop steering.
     raw_thresholds = compression.get("model_thresholds")
-    if isinstance(raw_thresholds, dict):
-        cc.model_thresholds = {
-            str(k): float(v)
-            for k, v in raw_thresholds.items() if isinstance(v, (int, float)) and not isinstance(v, bool)
-        }
-    else:
-        # Absent or invalid shape (agent_init treats both as empty): stale overrides must stop steering.
-        cc.model_thresholds = {}
+    cc.model_thresholds = {
+        str(k): float(v)
+        for k, v in raw_thresholds.items() if isinstance(v, (int, float)) and not isinstance(v, bool)
+    } if isinstance(raw_thresholds, dict) else {}
 
     # threshold: present value wins; absence derives via the agent_init resolution (default + autoraise).
     pct: float | None = None
     if "threshold" in compression:
-        try:
+        with contextlib.suppress(TypeError, ValueError):
             pct = float(compression["threshold"])
-        except (TypeError, ValueError):
-            pct = None
     if pct is None:
         pct = _derived_default_threshold_percent(agent, compression)
-    try:
+    with contextlib.suppress(TypeError, ValueError):
         cc._config_threshold_percent = pct
         cc._configured_threshold_percent = pct
         base = pct
@@ -180,8 +165,6 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
                 cc.threshold_percent = pct
         else:
             cc.threshold_percent = pct
-    except (TypeError, ValueError):
-        pass
 
     raw_ctx = model_cfg.get("context_length")
     if raw_ctx is not None:
@@ -191,10 +174,8 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
             new_ctx = 0
         if new_ctx > 0:
             cc._config_context_length = new_ctx
-            try:
+            with contextlib.suppress(Exception):
                 cc.context_length = new_ctx
-            except Exception:
-                pass
     elif getattr(cc, "_config_context_length", None) is not None:
         # model.context_length removed: drop the override and force re-inference from model metadata
         # on next access (construction's deferred resolution); re-applies the small-context floor too.
@@ -222,9 +203,7 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
         compute = getattr(cc, "_compute_threshold_tokens", None)
         if callable(compute):
             cc.threshold_tokens = compute(
-                getattr(cc, "context_length", 0) or 0,
-                getattr(cc, "threshold_percent", 0.5),
-                getattr(cc, "max_tokens", None),
+                getattr(cc, "context_length", 0) or 0, getattr(cc, "threshold_percent", 0.5), getattr(cc, "max_tokens", None)
             )
         cc._apply_threshold_tokens_cap()
     else:
@@ -313,18 +292,14 @@ def _compress_session_history(
     partial, keep_last, focus_topic = parse_partial_compress_args(focus_topic or "")
     # Only the head is summarized; the last `keep_last` exchanges ride along verbatim. A degenerate
     # split (empty tail) falls back to full compression so the user still gets an action.
-    tail: list = []
-    head = history
-    if partial:
-        head, tail = split_history_for_partial_compress(history, keep_last)
-        if not tail:
-            partial = False
-            head = history
+    head, tail = (split_history_for_partial_compress(history, keep_last) if partial else (history, []))
+    if not tail:
+        partial, head = False, history
     if approx_tokens is None:
         # Include system prompt + tool schemas so the figure reflects real request pressure.
-        _sys_prompt = getattr(agent, "_cached_system_prompt", "") or ""
-        _tools = getattr(agent, "tools", None) or None
-        approx_tokens = estimate_request_tokens_rough(history, system_prompt=_sys_prompt, tools=_tools)
+        approx_tokens = estimate_request_tokens_rough(
+            history, system_prompt=getattr(agent, "_cached_system_prompt", "") or "", tools=getattr(agent, "tools", None) or None
+        )
     # system_message=None: _compress_context rebuilds the system prompt via _build_system_prompt(None);
     # passing the cached prompt (already holding the identity block) appends the identity twice.
     # force=True: every caller is a manual /compress path, which bypasses the summary-failure
@@ -346,7 +321,7 @@ def _compress_session_history(
         finalize_context_engine_compression_notification(agent, committed=False)
         raise CompressionLockHeld(_lock_skipped if isinstance(_lock_skipped, str) else None)
 
-    if partial and tail:
+    if partial:
         compressed = rejoin_compressed_head_and_tail(compressed, tail)
     with session["history_lock"]:
         if int(session.get("history_version", 0)) != history_version:
@@ -375,14 +350,15 @@ def _sync_session_key_after_compress(
     if not new_session_id or new_session_id == old_key:
         return
 
-    lease_reanchored = _transfer_active_session_slot(sid, session, new_session_id=new_session_id)
-    if not lease_reanchored:
+    if not _transfer_active_session_slot(sid, session, new_session_id=new_session_id):
         logger.warning(
             "Compression session lease did not re-anchor: sid=%s old_session_id=%s new_session_id=%s",
             sid, old_key, new_session_id,
         )
 
-    try:
+    # Even if the approval module fails to import, anchor session_key on the continuation id.
+    session["session_key"] = new_session_id
+    with contextlib.suppress(Exception):
         from tools.approval import (
             disable_session_yolo, enable_session_yolo, is_session_yolo_enabled, register_gateway_notify,
             unregister_gateway_notify,
@@ -390,7 +366,6 @@ def _sync_session_key_after_compress(
 
         with contextlib.suppress(Exception):
             unregister_gateway_notify(old_key)
-        session["session_key"] = new_session_id
         try:
             yolo_was_on = is_session_yolo_enabled(old_key)
         except Exception:
@@ -401,9 +376,6 @@ def _sync_session_key_after_compress(
                 disable_session_yolo(old_key)
         with contextlib.suppress(Exception):
             register_gateway_notify(new_session_id, lambda data: _emit_approval_request(sid, data))
-    except Exception:
-        # Even if the approval module fails to import, anchor session_key on the continuation id.
-        session["session_key"] = new_session_id
 
     # Invalidate any in-flight ``_drain_queued_prompt`` claim taken under the pre-rotation key: a raced
     # drain must not dispatch on the continuation (its envelope is restored to the queue).
diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py
index f03be675f4..b782921735 100644
--- a/tui_gateway/session_notifications.py
+++ b/tui_gateway/session_notifications.py
@@ -1,8 +1,6 @@
-"""Per-session notification poller: kanban/loop/delegation events routed to the owning session, desktop UI wiring, HUD surface note.
-
-Bodies are rebound onto server.py's globals at install time (see
-method_ctx.bind_module), so they reference server.py globals bare.
-"""
+"""Per-session notification poller: kanban/loop/delegation events routed to the owning session,
+desktop UI wiring, HUD surface note. Bodies are rebound onto server.py's globals at install time
+(method_ctx.bind_module), so they reference server.py globals bare."""
 
 from __future__ import annotations
 
@@ -15,8 +13,7 @@ _registry = HandlerRegistry()
 
 
 def _notif_locked_sessions(fn, default):
-    """Run ``fn(_sessions)`` under ``_sessions_lock``; ``default`` if that fails (the poller
-    thread must never crash on a lock/enumeration failure)."""
+    """Run ``fn(_sessions)`` under ``_sessions_lock``; ``default`` on failure (poller must never crash)."""
     try:
         with _sessions_lock:
             return fn(_sessions)
@@ -32,6 +29,13 @@ def _notif_session_matches(s: dict, keys) -> bool:
     return str(s.get("session_key") or "") in keys or _session_lookup_key(s, fallback="") in keys
 
 
+def _notif_live_session_matches(keys) -> bool:
+    """Any non-finalized live session matches ``keys`` (False if the registry can't be read)."""
+    return _notif_locked_sessions(
+        lambda ss: any(not s.get("_finalized") and _notif_session_matches(s, keys) for s in ss.values()), False
+    )
+
+
 def _notif_resolve_event_key(evt_key: str) -> str:
     """Resolve a compression-rotated session key to its continuation tip (or itself)."""
     try:
@@ -56,13 +60,10 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) ->
             return True
         # Exact UI tab gone: fall through to durable session_key routing so a
         # resumed continuation with the same key/lineage can still claim it.
-
     evt_key = str(evt.get("session_key") or "")
     if not evt_key:
         return False
-
     current_keys = _notif_current_keys(sid, session)
-
     # Compression can rotate AIAgent.session_id while the detached child is still running: map
     # the event's original key to its continuation tip so it reaches the live session instead of
     # becoming an orphan any poller may consume. A live continuation wins over the compressed
@@ -71,19 +72,13 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) ->
     if resolved_key != evt_key:
         if resolved_key in current_keys:
             return False
-        if _notif_locked_sessions(
-            lambda ss: any(not s.get("_finalized") and _notif_session_matches(s, {resolved_key}) for s in ss.values()),
-            False,
-        ):
+        if _notif_live_session_matches({resolved_key}):
             return True
-
     if evt_key in current_keys:
         return False
-
     snapshot = _notif_locked_sessions(lambda ss: list(ss.values()), None)
     if snapshot is None:
         return False  # can't enumerate: fail open rather than drop the event
-
     keys = {evt_key, resolved_key}
     return any(s is not session and not s.get("_finalized") and _notif_session_matches(s, keys) for s in snapshot)
 
@@ -129,9 +124,9 @@ def _notification_event_dedup_key(evt: dict) -> tuple:
     return (evt_sid, evt_type)
 
 
-# Mirror gateway/kanban_watchers.py TERMINAL_KINDS: claim silent kinds too so
-# the cursor advances past them and they can't wedge a later completed/blocked
-# event behind an unclaimed row.
+# Mirror gateway/kanban_watchers.py TERMINAL_KINDS: claim silent kinds (archived/unblocked) too so
+# the cursor advances past them and they can't wedge a later completed/blocked event behind an
+# unclaimed row.
 _KANBAN_NOTIFY_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked")
 _KANBAN_POLL_SECONDS = 5.0
 _LOOP_POLL_SECONDS = 5.0
@@ -155,6 +150,17 @@ def _notif_log_failure(what: str, exc: BaseException) -> None:
     print(f"[tui_gateway] {what}: {type(exc).__name__}: {exc}", file=sys.stderr)
 
 
+def _notif_submit(rid: str, sid: str, session: dict, text: str, what: str, **kwargs) -> None:
+    """message.start + _run_prompt_submit for a claimed (running=True) turn; releases on failure."""
+    try:
+        _emit("message.start", sid)
+        _run_prompt_submit(rid, sid, session, text, **kwargs)
+    except Exception as exc:
+        _notif_log_failure(what, exc)
+        _notif_release_turn(session)
+        raise
+
+
 def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
     """Fire a due /loop wakeup for an idle TUI/Desktop/dashboard session (per-session poller,
     coarse cadence). Claims the session under history_lock (running=True) before dispatching so
@@ -163,56 +169,50 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
         from hermes_cli.loops import LoopManager, goal_blocks_loop_tick
     except Exception:
         return
-
     sid_key = session.get("session_key") or ""
     if not sid_key:
         return
     mgr = LoopManager(session_id=sid_key)
-    if not mgr.is_due() or goal_blocks_loop_tick(sid_key):
-        return
-    if not _notif_claim_turn(session):
+    if not mgr.is_due() or goal_blocks_loop_tick(sid_key) or not _notif_claim_turn(session):
         return  # busy — stays due, next poll retries
-
     wakeup = mgr.fire_tick()
     if not wakeup:
         _notif_release_turn(session)
         return
-
     tick_no = mgr.state.ticks_fired if mgr.state else "?"
     rid = f"__loop__{int(time.time() * 1000)}"
     try:
         _emit("status.update", sid, {"kind": "loop", "text": f"↻ /loop wakeup #{tick_no} firing…"})
-        if wakeup.lstrip().startswith("/"):
-            # Slash-command loop: route through the slash pipeline, not the
-            # model. No model reply to evaluate — complete immediately.
-            _notif_release_turn(session)
-            try:
-                parts = wakeup.lstrip()[1:].split(None, 1)
-                resp = _methods["command.dispatch"](
-                    rid,
-                    {"name": parts[0] if parts else "", "arg": parts[1] if len(parts) > 1 else "", "session_id": sid},
-                )
-                payload = (resp or {}).get("result") or {}
-                out = str(payload.get("output") or "").strip()
-                if out:
-                    _emit("status.update", sid, {"kind": "loop", "text": out})
-                if payload.get("type") == "send" and payload.get("message"):
-                    # Command resolves to a prompt (skill command etc.) — run it
-                    # as a normal turn; the post-turn hook completes the tick.
-                    if not _notif_claim_turn(session):
-                        mgr.abandon_tick()
-                        return
-                    _emit("message.start", sid)
-                    _run_prompt_submit(rid, sid, session, payload["message"])
-                    return
-            except Exception:
-                pass
-            decision = mgr.complete_tick("")
-            if decision.get("message"):
-                _emit("status.update", sid, {"kind": "loop", "text": decision["message"]})
+        if not wakeup.lstrip().startswith("/"):
+            _emit("message.start", sid)
+            _run_prompt_submit(rid, sid, session, wakeup)
             return
-        _emit("message.start", sid)
-        _run_prompt_submit(rid, sid, session, wakeup)
+        # Slash-command loop: route through the slash pipeline, not the model. No model reply
+        # to evaluate — complete immediately.
+        _notif_release_turn(session)
+        try:
+            parts = wakeup.lstrip()[1:].split(None, 1)
+            resp = _methods["command.dispatch"](
+                rid, {"name": parts[0] if parts else "", "arg": parts[1] if len(parts) > 1 else "", "session_id": sid}
+            )
+            payload = (resp or {}).get("result") or {}
+            out = str(payload.get("output") or "").strip()
+            if out:
+                _emit("status.update", sid, {"kind": "loop", "text": out})
+            if payload.get("type") == "send" and payload.get("message"):
+                # Command resolves to a prompt (skill command etc.) — run it as a normal turn;
+                # the post-turn hook completes the tick.
+                if not _notif_claim_turn(session):
+                    mgr.abandon_tick()
+                    return
+                _emit("message.start", sid)
+                _run_prompt_submit(rid, sid, session, payload["message"])
+                return
+        except Exception:
+            pass
+        decision = mgr.complete_tick("")
+        if decision.get("message"):
+            _emit("status.update", sid, {"kind": "loop", "text": decision["message"]})
     except Exception as exc:
         _notif_log_failure("loop wakeup dispatch failed", exc)
         _notif_release_turn(session)
@@ -228,23 +228,11 @@ def _kb_first_line(value: Any, limit: int) -> str:
 def _kb_completed(task, payload: dict, title: str) -> str:
     if payload.get("summary"):
         handoff = _kb_first_line(payload["summary"], 200)
-    elif getattr(task, "result", None):
-        handoff = _kb_first_line(task.result, 160)
     else:
-        handoff = ""
+        handoff = _kb_first_line(task.result, 160) if getattr(task, "result", None) else ""
     return f" done — {title}{handoff}"
 
 
-def _kb_blocked(task, payload: dict, title: str) -> str:
-    reason = f": {str(payload.get('reason'))[:160]}" if payload.get("reason") else ""
-    return f" blocked{reason}"
-
-
-def _kb_gave_up(task, payload: dict, title: str) -> str:
-    err = f"\n{str(payload.get('error'))[:200]}" if payload.get("error") else ""
-    return f" gave up after repeated spawn failures{err}"
-
-
 def _kb_timed_out(task, payload: dict, title: str) -> str:
     try:
         limit = int(payload.get("limit_seconds") or 0)
@@ -253,20 +241,21 @@ def _kb_timed_out(task, payload: dict, title: str) -> str:
     return f" timed out (max_runtime={limit}s); will retry"
 
 
-# kind -> (glyph, suffix formatter); silent kinds (archived/unblocked) are absent → None.
+# kind -> (glyph, suffix after "Kanban <id>"); silent kinds (archived/unblocked) are absent → None.
 _KANBAN_EVENT_FORMATTERS = {
     "completed": ("✔", _kb_completed),
-    "blocked": ("⏸", _kb_blocked),
-    "gave_up": ("✖", _kb_gave_up),
-    "crashed": ("✖", lambda task, payload, title: " worker crashed (pid gone); dispatcher will retry"),
+    "blocked": ("⏸", lambda t, p, title: " blocked" + (f": {str(p.get('reason'))[:160]}" if p.get("reason") else "")),
+    "gave_up": ("✖", lambda t, p, title: " gave up after repeated spawn failures"
+                + (f"\n{str(p.get('error'))[:200]}" if p.get("error") else "")),
+    "crashed": ("✖", lambda t, p, title: " worker crashed (pid gone); dispatcher will retry"),
     "timed_out": ("⏱", _kb_timed_out),
-    "status": ("🔄", lambda task, payload, title: f" → {payload.get('status') or ''}"),
+    "status": ("🔄", lambda t, p, title: f" → {p.get('status') or ''}"),
 }
 
 
 def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[str]:
-    """Single-line notification text for one kanban event; wording mirrors
-    gateway/kanban_watchers.py so it reads the same as on Telegram. None for silent kinds."""
+    """Single-line notification text for one kanban event; wording mirrors gateway/kanban_watchers.py
+    so it reads the same as on Telegram. None for silent kinds."""
     entry = _KANBAN_EVENT_FORMATTERS.get(getattr(ev, "kind", ""))
     if entry is None:
         return None
@@ -276,8 +265,7 @@ def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[
     board_tag = f"[{board_slug}] " if board_slug else ""
     who = getattr(task, "assignee", None) or ""
     tag = f"@{who} " if who else ""
-    payload = getattr(ev, "payload", None) or {}
-    return f"{glyph} {board_tag}{tag}Kanban {task_id}{fmt(task, payload, title)}"
+    return f"{glyph} {board_tag}{tag}Kanban {task_id}{fmt(task, getattr(ev, 'payload', None) or {}, title)}"
 
 
 def _collect_kanban_notifications(session: dict) -> list:
@@ -303,8 +291,8 @@ def _collect_kanban_notifications(session: dict) -> list:
             boards = [_kb.read_board_metadata(_kb.DEFAULT_BOARD)]
         except Exception:
             return []
-    # Poll each resolved DB path once — multiple slugs can point at the same
-    # DB when HERMES_KANBAN_DB pins the board path (same guard as the gateway).
+    # Poll each resolved DB path once — multiple slugs can point at the same DB when
+    # HERMES_KANBAN_DB pins the board path (same guard as the gateway).
     seen_db_paths: set = set()
     for board_meta in boards:
         slug = (board_meta or {}).get("slug") or _kb.DEFAULT_BOARD
@@ -316,9 +304,9 @@ def _collect_kanban_notifications(session: dict) -> list:
         if resolved in seen_db_paths:
             continue
         seen_db_paths.add(resolved)
-        # One poller per live session: don't open this board writable unless it
-        # has a subscription owned by this exact session. If the read-only probe
-        # fails (locked/corrupt DB), preserve delivery and fall through.
+        # One poller per live session: don't open this board writable unless it has a subscription
+        # owned by this exact session. If the read-only probe fails (locked/corrupt DB), preserve
+        # delivery and fall through.
         try:
             if _kb.count_notify_subs(board=slug, platform="tui", chat_id=session_key) == 0:
                 continue
@@ -345,9 +333,9 @@ def _collect_kanban_notifications(session: dict) -> list:
                     continue
                 task = _kb.get_task(conn, sub["task_id"])
                 texts.extend(t for t in (_format_kanban_event_text(sub, task, ev, slug) for ev in events) if t)
-                # Unsubscribe only on archive: ``done`` is reversible in review/controller flows,
-                # so keeping the sub lets a later reopen notify the same session. The claimed
-                # cursor prevents replay.
+                # Unsubscribe only on archive: ``done`` is reversible in review/controller flows, so
+                # keeping the sub lets a later reopen notify the same session. The claimed cursor
+                # prevents replay.
                 if task and getattr(task, "status", "") == "archived":
                     with contextlib.suppress(Exception):
                         _kb.remove_notify_sub(conn, **sub_ident)
@@ -379,12 +367,11 @@ def _notif_poll_kanban(sid: str, session: dict) -> None:
             _batch = list(_pending)
             session["_kanban_pending"] = []
     if _batch:
-        try:
-            _emit("message.start", sid)
-            _run_prompt_submit(f"__notif__{int(time.time() * 1000)}", sid, session, "\n".join(_batch))
-        except Exception as exc:
-            _notif_log_failure("kanban notification dispatch failed", exc)
-            _notif_release_turn(session)
+        with contextlib.suppress(Exception):
+            _notif_submit(
+                f"__notif__{int(time.time() * 1000)}", sid, session, "\n".join(_batch),
+                "kanban notification dispatch failed",
+            )
 
 
 def _notif_render_event(sid: str, evt: dict, emitted: set, process_registry, format_process_notification) -> Optional[str]:
@@ -407,22 +394,21 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None
     """Run the claimed (running=True) agent turn for one notification event."""
     from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery
 
-    rid = f"__notif__{int(time.time() * 1000)}"
     claim = claim_event_delivery(evt, "tui-poller")
     if claim is None:
         return
+    kwargs = (
+        {"display_kind": "async_delegation_complete", "display_metadata": _async_delegation_display_metadata(evt)}
+        if evt.get("type") == "async_delegation" else {}
+    )
     try:
-        _emit("message.start", sid)
-        kwargs = (
-            {"display_kind": "async_delegation_complete", "display_metadata": _async_delegation_display_metadata(evt)}
-            if evt.get("type") == "async_delegation" else {}
+        _notif_submit(
+            f"__notif__{int(time.time() * 1000)}", sid, session, text, "notification poller dispatch failed", **kwargs
         )
-        _run_prompt_submit(rid, sid, session, text, **kwargs)
-        complete_event_delivery(evt, claim)
-    except Exception as exc:
+    except Exception:
         release_event_delivery(evt, claim)
-        _notif_log_failure("notification poller dispatch failed", exc)
-        _notif_release_turn(session)
+        return
+    complete_event_delivery(evt, claim)
 
 
 def _notif_classify_event(sid: str, session: dict, evt: dict) -> str:
@@ -435,6 +421,52 @@ def _notif_classify_event(sid: str, session: dict, evt: dict) -> str:
     return "mine"
 
 
+def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> bool:
+    """Route one dequeued event. ``deferred`` is None in the live loop (foreign events go straight
+    back on the queue) and the shutdown-drain list otherwise (foreign + orphaned delegation
+    payloads are set aside for a resume). Returns False when the drain must stop (session busy)."""
+    queue = registry.completion_queue
+    owner = _notif_classify_event(sid, session, evt)
+    if owner == "foreign":
+        if deferred is None:
+            # Leave foreign events for their owner — otherwise a process started in session A
+            # surfaces its completion in whichever poller wakes first.
+            queue.put(evt)
+            time.sleep(0.1)
+        else:
+            deferred.append(evt)
+        return True
+    if owner == "unowned":
+        is_delegation = evt.get("type") == "async_delegation"
+        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)(
+                "Dropping unowned %s notification (origin=%r key=%r) instead of delivering to session %s",
+                evt.get("type", "completion"), origin, key, sid,
+            )
+        elif is_delegation:
+            deferred.append(evt)
+        else:
+            logger.debug(
+                "Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
+                evt.get("type", "completion"), origin, key,
+            )
+        return True
+    text = _notif_render_event(sid, evt, emitted, registry, fmt)
+    if not text:
+        return True
+    if not _notif_claim_turn(session):
+        queue.put(evt)
+        if deferred is not None:
+            return False
+        # Back off: the re-queued event keeps the queue non-empty, so without a sleep this loop
+        # spins at 100% CPU while busy.
+        time.sleep(0.25)
+        return True
+    _notif_dispatch_event(sid, session, evt, text)
+    return True
+
+
 def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None:
     """Poll completion_queue and dispatch notifications autonomously (daemon thread started by
     _init_session()): emit a status.update (kind=process), then chain an agent turn via
@@ -446,8 +478,7 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
 
     queue = process_registry.completion_queue
     _emitted = set()  # dedup re-queued events so one completion isn't emitted 50 times while busy
-    _last_kanban_poll = 0.0
-    _last_loop_poll = 0.0
+    _last_kanban_poll = _last_loop_poll = 0.0
     while not stop_event.is_set() and not session.get("_finalized"):
         _now = time.monotonic()
         # /loop wakeup driver: fire a due tick for THIS session while idle (same claim-under-lock
@@ -465,63 +496,18 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
             evt = queue.get(timeout=0.5)
         except Exception:
             continue
+        _notif_handle_event(sid, session, evt, _emitted, process_registry, format_process_notification, None)
 
-        owner = _notif_classify_event(sid, session, evt)
-        if owner == "foreign":
-            # Leave foreign events for their owner — otherwise a process started in
-            # session A surfaces its completion in whichever poller wakes first.
-            queue.put(evt)
-            time.sleep(0.1)
-            continue
-        if owner == "unowned":
-            log = logger.warning if evt.get("type") == "async_delegation" else logger.debug
-            log(
-                "Dropping unowned %s notification (origin=%r key=%r) instead of delivering to session %s",
-                evt.get("type", "completion"), str(evt.get("origin_ui_session_id") or ""),
-                str(evt.get("session_key") or ""), sid,
-            )
-            continue
-
-        text = _notif_render_event(sid, evt, _emitted, process_registry, format_process_notification)
-        if not text:
-            continue
-        if not _notif_claim_turn(session):
-            # Back off: the re-queued event keeps the queue non-empty, so
-            # without a sleep this loop spins at 100% CPU while busy.
-            queue.put(evt)
-            time.sleep(0.25)
-            continue
-        _notif_dispatch_event(sid, session, evt, text)
-
-    # Drain remaining events after the stop signal so nothing is lost on shutdown. Other live
-    # sessions' events are set aside and re-queued; orphaned events (owner gone) are dropped —
-    # same guard as the main loop, except orphaned delegation payloads are deferred for a resume.
+    # Drain remaining events after the stop signal so nothing is lost on shutdown; foreign and
+    # orphaned-delegation events are handed back to the shared queue afterwards.
     deferred: list = []
     while not queue.empty():
         try:
             evt = queue.get_nowait()
         except Exception:
             break
-        owner = _notif_classify_event(sid, session, evt)
-        if owner == "foreign" or (owner == "unowned" and evt.get("type") == "async_delegation"):
-            deferred.append(evt)
-            continue
-        if owner == "unowned":
-            logger.debug(
-                "Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
-                evt.get("type", "completion"), str(evt.get("origin_ui_session_id") or ""),
-                str(evt.get("session_key") or ""),
-            )
-            continue
-        text = _notif_render_event(sid, evt, _emitted, process_registry, format_process_notification)
-        if not text:
-            continue
-        if not _notif_claim_turn(session):
-            queue.put(evt)
+        if not _notif_handle_event(sid, session, evt, _emitted, process_registry, format_process_notification, deferred):
             break
-        _notif_dispatch_event(sid, session, evt, text)
-
-    # Hand any other sessions' events back to the shared queue.
     for evt in deferred:
         queue.put(evt)
 
@@ -561,8 +547,8 @@ def _wire_agent_terminal_output() -> None:
         return ""
 
     def _emit_agent_terminal_close(session, process_id):
-        # session may be None (process already finished/pruned) — the tab can
-        # still linger and be closed; route to the owning window when we can.
+        # session may be None (process already finished/pruned) — the tab can still linger and
+        # be closed; route to the owning window when we can.
         _emit("terminal.close", _owner_sid_for_process(session) if session is not None else "", {"process_id": process_id})
 
     if getattr(process_registry, "on_output", None) is None:
@@ -587,7 +573,6 @@ def _wire_desktop_ui() -> None:
         from tools import desktop_ui
     except Exception:
         return
-
     desktop_ui.set_emitter(lambda sid, event, payload: _emit(event, sid, payload))
     _desktop_ui_wired = True
 
@@ -599,11 +584,10 @@ _notification_pollers: list = []
 
 
 def _start_notification_poller(sid: str, session: dict) -> threading.Event:
-    """Start the background notification poller for a TUI session."""
+    """Start the background notification poller for a TUI session (thread name is greppable)."""
     _wire_agent_terminal_output()
     _wire_desktop_ui()
     stop = threading.Event()
-    # Thread name is greppable for debuggers/test teardowns.
     t = threading.Thread(
         target=_notification_poller_loop, args=(stop, sid, session), daemon=True, name=f"tui-notif-poller-{sid}"
     )

From 46423af83725b7a547f99bfe86986039200db74e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:30:36 -0700
Subject: [PATCH 3/4] =?UTF-8?q?refactor(tui):=20session=5Fauto=5Fcontinue?=
 =?UTF-8?q?=20=E2=80=94=20unify=20steer/redirect=20correction=20path,=20fl?=
 =?UTF-8?q?atten=20queue=20scrub?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tui_gateway/session_auto_continue.py | 276 ++++++++++++---------------
 1 file changed, 121 insertions(+), 155 deletions(-)

diff --git a/tui_gateway/session_auto_continue.py b/tui_gateway/session_auto_continue.py
index 7c263d1557..d4313c21a3 100644
--- a/tui_gateway/session_auto_continue.py
+++ b/tui_gateway/session_auto_continue.py
@@ -12,11 +12,11 @@ from .method_ctx import HandlerRegistry, bind_module
 _registry = HandlerRegistry()
 
 
-# A concluded turn (success, handled error, interrupt) clears its durable marker
-# (turn_marker.py) in _run_prompt_submit's finally; only a process death leaves it
-# behind, so a marker at session.resume proves the turn never finished AND the
-# client never saw a terminal frame. Fresh: re-submit automatically (as the
-# messaging gateway does). Stale: clear it and let the partial transcript speak.
+# A concluded turn (success, handled error, interrupt) clears its durable marker (turn_marker.py)
+# in _run_prompt_submit's finally; only a process death leaves it behind, so a marker at
+# session.resume proves the turn never finished AND the client never saw a terminal frame. Fresh:
+# re-submit automatically (as the messaging gateway does). Stale: clear it and let the partial
+# transcript speak.
 
 _AUTO_CONTINUE_ENABLED_DEFAULT = True
 _AUTO_CONTINUE_FRESHNESS_MINUTES_DEFAULT = 15
@@ -36,9 +36,7 @@ def _auto_continue_config() -> tuple[bool, float, int]:
     return (
         is_truthy_value(cfg.get("enabled"), default=_AUTO_CONTINUE_ENABLED_DEFAULT),
         max(0.0, minutes) * 60.0,
-        _coerce_int_config_value(
-            cfg.get("max_attempts"), _AUTO_CONTINUE_MAX_ATTEMPTS_DEFAULT, min_value=0
-        ),
+        _coerce_int_config_value(cfg.get("max_attempts"), _AUTO_CONTINUE_MAX_ATTEMPTS_DEFAULT, min_value=0),
     )
 
 
@@ -51,10 +49,10 @@ def _session_home(session: dict) -> Path:
 def _retire_turn_marker(session: dict, *keys: str) -> None:
     """Drop the crash marker for a turn whose outcome is about to reach the client.
 
-    Called right before the terminal frame, not at turn-thread end: post-turn work
-    (titles, memory sync, goal hooks) outlives the client's answer, and quitting in
-    that window would leave a marker that re-runs a finished turn on next launch.
-    Extra ``keys`` cover a session_key that compression rotated mid-turn.
+    Called right before the terminal frame, not at turn-thread end: post-turn work (titles, memory
+    sync, goal hooks) outlives the client's answer, and quitting in that window would leave a marker
+    that re-runs a finished turn on next launch. Extra ``keys`` cover a session_key that compression
+    rotated mid-turn.
     """
     home = _session_home(session)
     for key in dict.fromkeys((*keys, str(session.get("session_key") or ""))):
@@ -63,8 +61,8 @@ def _retire_turn_marker(session: dict, *keys: str) -> None:
 
 
 def _auto_continue_note(prompt: str) -> str:
-    # Same opening as the gateway's recovery notes (transcript tooling recognizes
-    # both). The prompt is embedded: a hard crash persists nothing else of the turn.
+    # Same opening as the gateway's recovery notes (transcript tooling recognizes both). The prompt
+    # is embedded: a hard crash persists nothing else of the turn.
     return (
         f"{_AUTO_CONTINUE_NOTE_PREFIX} — the app or its backend process "
         "stopped before the turn could finish. Some of the work may already "
@@ -81,19 +79,22 @@ def _ac_release_turn(session: dict, *, unschedule: bool = False) -> None:
             session["_auto_continue_scheduled"] = False
 
 
+def _ac_dispatch_failed(what: str, exc: BaseException) -> None:
+    print(f"[tui_gateway] {what} dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr)
+
+
 def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) -> dict | None:
     """Kick off a continuation turn for a crash-interrupted session.
 
-    Called from session.resume's cold paths once the live record is registered.
-    Returns a descriptor for the resume payload when scheduled, else None. The turn
-    runs on a background thread after the deferred agent build via the normal
-    _run_prompt_submit path, so the client that just resumed streams it live.
+    Called from session.resume's cold paths once the live record is registered. Returns a descriptor
+    for the resume payload when scheduled, else None. The turn runs on a background thread after the
+    deferred agent build via the normal _run_prompt_submit path, so the client that just resumed
+    streams it live.
     """
-    # Hosted room turns are recovered by their durable task/lease state machine;
-    # generic auto-continue would bypass its execution generation and duplicate work.
+    # Hosted room turns are recovered by their durable task/lease state machine; generic
+    # auto-continue would bypass its execution generation and duplicate work.
     if session.get("source") == "bot_room":
         return None
-
     home = _session_home(session)
     marker = read_turn_marker(home, session_key)
     if marker is None:
@@ -129,17 +130,17 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
                 return
             session["running"] = True
             session["last_active"] = time.time()
-        # Ownership admission BEFORE message.start: a sibling backend sharing this
-        # HERMES_HOME may have written the marker and still be mid-turn. Leave the
-        # marker so a later resume retries once the owner finishes or dies.
+        # Ownership admission BEFORE message.start: a sibling backend sharing this HERMES_HOME may
+        # have written the marker and still be mid-turn. Leave the marker so a later resume retries
+        # once the owner finishes or dies.
         if _ensure_active_session_slot(sid, session) is not None:
             logger.info("auto-continue for %s refused: session has another live owner", session_key)
             _ac_release_turn(session, unschedule=True)
             return
         with session["history_lock"]:
-            # Marker inputs read back by _run_prompt_submit: count the attempt (crash
-            # breaker) and re-record the ORIGINAL prompt (no nested notes). Set here,
-            # not at schedule time, so a bail above leaves nothing for a racing user turn.
+            # Marker inputs read back by _run_prompt_submit: count the attempt (crash breaker) and
+            # re-record the ORIGINAL prompt (no nested notes). Set here, not at schedule time, so a
+            # bail above leaves nothing for a racing user turn.
             session["_auto_continue_attempt"] = attempt
             session["_auto_continue_prompt"] = marker["prompt"]
         try:
@@ -147,13 +148,12 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
             _emit("message.start", sid)
             _run_prompt_submit(rid, sid, session, text, display_kind="auto_continue")
         except Exception as exc:
-            print(f"[tui_gateway] auto-continue dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr)
+            _ac_dispatch_failed("auto-continue", exc)
             _ac_release_turn(session)
 
     threading.Thread(target=kickoff, daemon=True).start()
     logger.info(
-        "auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago)",
-        session_key, attempt, age,
+        "auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago)", session_key, attempt, age
     )
     return {"attempt": attempt, "interrupted_at": marker["started_at"]}
 
@@ -161,10 +161,10 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
 def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[str] | None = None) -> None:
     """Stash a message to run as the very next turn once the live one ends.
 
-    Text-only arrivals share a slot and merge losslessly (like the consecutive-user
-    merge in ``repair_message_sequence``); image-bearing ones stay separate envelopes
-    so attachment ownership/chronology survive. ``transport`` is pinned so the drained
-    turn streams back to its sender even if the session transport is rebound.
+    Text-only arrivals share a slot and merge losslessly (like the consecutive-user merge in
+    ``repair_message_sequence``); image-bearing ones stay separate envelopes so attachment
+    ownership/chronology survive. ``transport`` is pinned so the drained turn streams back to its
+    sender even if the session transport is rebound.
     """
     image_paths = list(image_paths or [])
     # Scrub live-turn self-duplicates first so the text merge below can't glue
@@ -173,7 +173,7 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[
     # Never queue a text-only self-copy of the live prompt: draining it would restart it.
     if not image_paths and isinstance(text, str):
         turn = session.get("inflight_turn")
-        original = (str(turn.get("user") or "").strip() if isinstance(turn, dict) else "")
+        original = str(turn.get("user") or "").strip() if isinstance(turn, dict) else ""
         if original and text.strip() == original:
             return
     queued = {"text": text, "transport": transport}
@@ -181,35 +181,28 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[
         queued["image_paths"] = image_paths
     existing = session.get("queued_prompt")
     if (
-        existing
-        and isinstance(existing.get("text"), str)
-        and isinstance(text, str)
-        and not existing.get("image_paths")
-        and not image_paths
-        and not session.get("queued_prompts")
+        existing and isinstance(existing.get("text"), str) and isinstance(text, str)
+        and not existing.get("image_paths") and not image_paths and not session.get("queued_prompts")
     ):
         prev = existing["text"]
         existing["text"] = f"{prev}\n\n{text}" if prev and text else (prev or text)
-        return
-    if existing:
+    elif existing:
         session.setdefault("queued_prompts", []).append(queued)
-        return
-    session["queued_prompt"] = queued
+    else:
+        session["queued_prompt"] = queued
 
 
 def _sanitize_queued_entry_vs_inflight_user(entry: Any, original: str) -> dict | None:
     """Drop (``None``) or rewrite a queue envelope that re-carries the live user text.
 
-    Text-only self-duplicates are dropped; a merged slot ``"{original}\\n\\n{later}"``
-    is rewritten to ``later`` so the correction survives without re-firing the
-    original. Image-bearing envelopes are left alone (chronology is load-bearing).
+    Text-only self-duplicates are dropped; a merged slot ``"{original}\\n\\n{later}"`` is rewritten
+    to ``later`` so the correction survives without re-firing the original. Image-bearing envelopes
+    are left alone (chronology is load-bearing).
     """
-    if not original or not isinstance(entry, dict):
-        return entry if isinstance(entry, dict) else None
-    if entry.get("image_paths"):
-        return entry
+    if not isinstance(entry, dict):
+        return None
     text = entry.get("text")
-    if not isinstance(text, str):
+    if not original or entry.get("image_paths") or not isinstance(text, str):
         return entry
     stripped = text.strip()
     if not stripped or stripped == original:
@@ -219,35 +212,25 @@ def _sanitize_queued_entry_vs_inflight_user(entry: Any, original: str) -> dict |
         prefix = original + sep
         if text.startswith(prefix):
             rest = text[len(prefix) :].strip()
-            if not rest or rest == original:
-                return None
-            return {**entry, "text": rest}
+            return None if not rest or rest == original else {**entry, "text": rest}
     return entry
 
 
 def _drop_queued_duplicates_of_inflight_user(session: dict) -> None:
     """Remove server-queue copies of the live turn's original user text.
 
-    A mid-turn ``prompt.submit`` of the same text can be queued while redirect is
-    unavailable (build window, tool boundary); after a later redirect it must not
-    drain and restart the original as a fresh turn. Unrelated follow-ups stay.
+    A mid-turn ``prompt.submit`` of the same text can be queued while redirect is unavailable
+    (build window, tool boundary); after a later redirect it must not drain and restart the original
+    as a fresh turn. Unrelated follow-ups stay.
     """
     turn = session.get("inflight_turn")
-    if not isinstance(turn, dict):
-        return
-    original = str(turn.get("user") or "").strip()
+    original = str(turn.get("user") or "").strip() if isinstance(turn, dict) else ""
     if not original:
         return
-
     head = session.get("queued_prompt")
-    rest = list(session.get("queued_prompts") or [])
-    kept: list[dict] = []
-    for entry in ([head] if head else []) + rest:
-        cleaned = _sanitize_queued_entry_vs_inflight_user(entry, original)
-        if cleaned is not None:
-            kept.append(cleaned)
-
-    _ac_set_queue(session, kept)
+    entries = ([head] if head else []) + list(session.get("queued_prompts") or [])
+    cleaned = (_sanitize_queued_entry_vs_inflight_user(entry, original) for entry in entries)
+    _ac_set_queue(session, [c for c in cleaned if c is not None])
 
 
 def _ac_set_queue(session: dict, entries: list) -> None:
@@ -262,15 +245,13 @@ def _ac_set_queue(session: dict, entries: list) -> None:
 def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
     """Interrupt a busy turn on a worker thread, never under ``history_lock``.
 
-    Some providers can't apply ``interrupt()`` until a blocking tool/network call
-    returns; doing it inline stalled ``session.resume`` and the queued prompt.
-    At most one interrupt worker per session so repeated steering can't leak threads.
+    Some providers can't apply ``interrupt()`` until a blocking tool/network call returns; doing it
+    inline stalled ``session.resume`` and the queued prompt. At most one interrupt worker per session
+    so repeated steering can't leak threads.
     """
     use_agent = agent is not None and hasattr(agent, "interrupt")
-    use_compute_host = not use_agent and _session_uses_compute_host(session)
-    if not use_agent and not use_compute_host:
+    if not use_agent and not _session_uses_compute_host(session):
         return
-
     with session["history_lock"]:
         if session.get("_busy_interrupt_pending"):
             return
@@ -291,27 +272,31 @@ def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
     threading.Thread(target=interrupt, daemon=True, name=f"busy-interrupt-{sid}").start()
 
 
-def _ac_record_inflight_correction(session: dict, plain_text: str) -> None:
-    """Record an accepted steer/redirect; scrub stale self-duplicates so the
-    live turn's original text is not re-fired from the queue after settle."""
+def _ac_try_correction(rid, session: dict, agent: Any, method: str, plain_text: str, status: str) -> dict | None:
+    """Apply ``agent.<method>(plain_text)`` (steer/redirect); on acceptance record the correction, scrub
+    stale self-duplicates so the live turn's original text is not re-fired after settle, and return
+    the ``status`` reply. None → caller falls through to the queue path."""
+    try:
+        if not getattr(agent, method)(plain_text):
+            return None
+    except Exception:
+        return None
     with session["history_lock"]:
         _record_inflight_correction(session, plain_text)
         _drop_queued_duplicates_of_inflight_user(session)
         session["last_active"] = time.time()
+    return _ok(rid, {"status": status})
 
 
-def _handle_busy_submit(
-    rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False
-) -> dict | None:
-    """Apply ``display.busy_input_mode`` to a prompt that lands mid-turn instead of
-    rejecting it with ``session busy`` (rejection made clients busy-retry and
-    silently drop sends when teardown outlived their deadline).
+def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False) -> dict | None:
+    """Apply ``display.busy_input_mode`` to a prompt that lands mid-turn instead of rejecting it with
+    ``session busy`` (rejection made clients busy-retry and silently drop sends when teardown
+    outlived their deadline).
 
-    Modes: ``interrupt`` (default) → redirect the live turn, falling back to hard
-    interrupt + queue for older agents; ``queue`` → queue only; ``steer`` → inject
-    after the current atomic action. ``queued=True`` (client queue drain) forces
-    queue mode: a "run after" message must NEVER become a live-turn correction,
-    even when the drain loses the settle race against a still-unwinding turn.
+    Modes: ``interrupt`` (default) → redirect the live turn, falling back to hard interrupt + queue
+    for older agents; ``queue`` → queue only; ``steer`` → inject after the current atomic action.
+    ``queued=True`` (client queue drain) forces queue mode: a "run after" message must NEVER become a
+    live-turn correction, even when the drain loses the settle race against a still-unwinding turn.
     """
     mode = "queue" if queued else _load_busy_input_mode()
     agent = session.get("agent")
@@ -325,31 +310,23 @@ def _handle_busy_submit(
             session["attached_images"] = []
     text_only = not image_paths and _is_text_only_busy_payload(text)
     plain_text = _coerce_message_text(text).strip() if text_only else ""
-    if mode == "steer" and text_only and plain_text and agent is not None and hasattr(agent, "steer"):
-        try:
-            if agent.steer(plain_text):
-                _ac_record_inflight_correction(session, plain_text)
-                return _ok(rid, {"status": "steered"})
-        except Exception:
-            pass  # fall through to queue
-    # Text-only corrections redirect in place when supported; media payloads and
-    # older agents fall through to the proven interrupt + queue path.
-    if (
-        mode == "interrupt"
-        and text_only
-        and plain_text
-        and agent is not None
-        and getattr(agent, "_supports_active_turn_redirect", False) is True
-        and hasattr(agent, "redirect")
-    ):
-        try:
-            if agent.redirect(plain_text):
-                _ac_record_inflight_correction(session, plain_text)
-                return _ok(rid, {"status": "redirected"})
-        except Exception:
-            pass  # preserve the proven interrupt + queue fallback below
-    # Queue before asking the live turn to stop. Never call a provider/compute-host
-    # method under history_lock: an interrupt can wait behind the op it cancels.
+    if plain_text and agent is not None:
+        if mode == "steer" and hasattr(agent, "steer"):
+            resp = _ac_try_correction(rid, session, agent, "steer", plain_text, "steered")
+            if resp is not None:
+                return resp
+        # Text-only corrections redirect in place when supported; media payloads and older agents
+        # fall through to the proven interrupt + queue path.
+        if (
+            mode == "interrupt"
+            and getattr(agent, "_supports_active_turn_redirect", False) is True
+            and hasattr(agent, "redirect")
+        ):
+            resp = _ac_try_correction(rid, session, agent, "redirect", plain_text, "redirected")
+            if resp is not None:
+                return resp
+    # Queue before asking the live turn to stop. Never call a provider/compute-host method under
+    # history_lock: an interrupt can wait behind the op it cancels.
     with session["history_lock"]:
         if not session.get("running"):
             if image_paths:
@@ -357,11 +334,10 @@ def _handle_busy_submit(
             return None
         _enqueue_prompt(session, text, transport, image_paths=image_paths)
         session["last_active"] = time.time()
-
-    # Attachments need their own model invocation: queue without cancelling so the
-    # user gets both results in order. ``steer`` must NEVER escalate to a hard
-    # interrupt: it would kill the live turn AND drop ``AIAgent._pending_steer``,
-    # destroying earlier accepted steers; steer fall-throughs stay FIFO-queued.
+    # Attachments need their own model invocation: queue without cancelling so the user gets both
+    # results in order. ``steer`` must NEVER escalate to a hard interrupt: it would kill the live
+    # turn AND drop ``AIAgent._pending_steer``, destroying earlier accepted steers; steer
+    # fall-throughs stay FIFO-queued.
     if mode == "interrupt" and not image_paths:
         _interrupt_busy_session(sid, session, agent)
     return _ok(rid, {"status": "queued"})
@@ -370,8 +346,8 @@ def _handle_busy_submit(
 def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
     """Fire a queued next-turn prompt if one is waiting and the session is idle.
 
-    True when dispatched: the caller skips lower-priority follow-ups this cycle
-    (the user's message wins). Claim-under-lock like the goal-continuation re-fire.
+    True when dispatched: the caller skips lower-priority follow-ups this cycle (the user's message
+    wins). Claim-under-lock like the goal-continuation re-fire.
     """
     with session["history_lock"]:
         if session.get("_closing"):
@@ -390,15 +366,11 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
     use_compute_host = _session_uses_compute_host(session)
     with session["history_lock"]:
         if int(session.get("_queued_prompt_generation", 0)) != queue_generation:
-            # Generation bump cancelled the claim (Stop, compress re-anchor, …): don't
-            # dispatch, but restore the envelope (claimed head first, then whatever
-            # advanced into the slot) so a legitimate follow-up isn't dropped.
-            rest: list = []
+            # Generation bump cancelled the claim (Stop, compress re-anchor, …): don't dispatch, but
+            # restore the envelope (claimed head first, then whatever advanced into the slot) so a
+            # legitimate follow-up isn't dropped.
             advanced = session.get("queued_prompt")
-            if advanced:
-                rest.append(advanced)
-            rest.extend(session.get("queued_prompts") or [])
-            _ac_set_queue(session, [queued, *rest])
+            _ac_set_queue(session, [queued, *([advanced] if advanced else []), *(session.get("queued_prompts") or [])])
             session["running"] = False
             return True
     kwargs: dict = {"queued_prompt_generation": queue_generation}
@@ -418,7 +390,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
         else:
             _run_prompt_submit(rid, sid, session, queued["text"], **kwargs)
     except Exception as exc:
-        print(f"[tui_gateway] queued prompt dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr)
+        _ac_dispatch_failed("queued prompt", exc)
         _ac_release_turn(session)
         dispatch_failed = True
     if dispatch_failed:
@@ -440,23 +412,20 @@ def _inflight_snapshot(session: dict) -> dict | None:
     if not user and not assistant and not streaming and not error:
         return None
     snapshot = {"assistant": assistant, "streaming": streaming, "user": user}
-    raw_corrections = turn.get("corrections") or []
     raw_offsets = turn.get("correction_offsets") or []
     correction_pairs = [
         (str(c), raw_offsets[i] if i < len(raw_offsets) else None)
-        for i, c in enumerate(raw_corrections)
-        if str(c).strip()
+        for i, c in enumerate(turn.get("corrections") or []) if str(c).strip()
     ]
     if correction_pairs:
-        # Mid-turn redirects alongside (not over) the original prompt so resume can
-        # rebuild every user bubble; offsets only when every correction has one so
-        # clients can trust the pairing.
+        # Mid-turn redirects alongside (not over) the original prompt so resume can rebuild every
+        # user bubble; offsets only when every correction has one so clients can trust the pairing.
         snapshot["corrections"] = [c for c, _ in correction_pairs]
         if all(isinstance(offset, int) and offset >= 0 for _, offset in correction_pairs):
             snapshot["correction_offsets"] = [int(offset) for _, offset in correction_pairs]  # type: ignore[arg-type]
     if error:
-        # Retained failed turn (_fail_inflight_turn): a resuming client must rebuild
-        # the failed bubble, not render the partial text as a healthy reply.
+        # Retained failed turn (_fail_inflight_turn): a resuming client must rebuild the failed
+        # bubble, not render the partial text as a healthy reply.
         snapshot["error"] = error
         snapshot["status"] = str(turn.get("status") or "error")
         snapshot["recoverable"] = bool(turn.get("recoverable"))
@@ -469,23 +438,21 @@ def _inflight_snapshot(session: dict) -> dict | None:
 def _emit_terminal_turn_error(
     sid: str, session: dict, error: Any, error_surface: Optional[dict] = None, *, retire_marker: bool = True
 ) -> None:
-    """Close a failed turn with the same ``status: "error"`` ``message.complete``
-    frame as ``_run_prompt_submit``'s returned-error path, retaining the turn via
-    ``_fail_inflight_turn`` so a client that missed the frame recovers it from
-    ``session.resume``'s ``inflight``. Callers that know the failing layer pass
-    ``error_surface``; exception callers leave it None and it is classified here.
+    """Close a failed turn with the same ``status: "error"`` ``message.complete`` frame as
+    ``_run_prompt_submit``'s returned-error path, retaining the turn via ``_fail_inflight_turn`` so a
+    client that missed the frame recovers it from ``session.resume``'s ``inflight``. Callers that
+    know the failing layer pass ``error_surface``; exception callers leave it None and it is
+    classified here.
     """
     agent = session.get("agent")
-    # {layer, code, retryable} descriptor so the desktop can say "Provider error" /
-    # "Gateway error" with matching recovery actions. Advisory: never raises.
+    # {layer, code, retryable} descriptor so the desktop can say "Provider error" / "Gateway error"
+    # with matching recovery actions. Advisory: never raises.
     if error_surface is None and isinstance(error, BaseException):
         try:
             from agent.error_surface import build_error_surface_from_exception
 
             error_surface = build_error_surface_from_exception(
-                error,
-                provider=str(getattr(agent, "provider", "") or ""),
-                model=str(getattr(agent, "model", "") or ""),
+                error, provider=str(getattr(agent, "provider", "") or ""), model=str(getattr(agent, "model", "") or "")
             )
         except Exception:
             error_surface = None
@@ -519,9 +486,8 @@ def _emit_terminal_turn_error(
 
 
 def _restore_agent_history_after_turn_error(session: dict, agent) -> bool:
-    """Keep a failed turn's working transcript: ``AIAgent`` persists its messages
-    independently, so after a raise the next prompt must see them, not the
-    pre-turn snapshot."""
+    """Keep a failed turn's working transcript: ``AIAgent`` persists its messages independently, so
+    after a raise the next prompt must see them, not the pre-turn snapshot."""
     agent_messages = getattr(agent, "_session_messages", None)
     if not isinstance(agent_messages, list):
         return False
@@ -532,8 +498,8 @@ def _restore_agent_history_after_turn_error(session: dict, agent) -> bool:
 
 
 def _queued_prompt_snapshot(session: dict) -> dict | None:
-    """The accepted next-turn prompt without its transport handle, for the
-    live-session projection (Desktop may reconnect while it is still queued)."""
+    """The accepted next-turn prompt without its transport handle, for the live-session projection
+    (Desktop may reconnect while it is still queued)."""
     queued = session.get("queued_prompt")
     if not isinstance(queued, dict):
         return None

From 4af78ae474fae786e823e56f81a6cc2cefc2278b Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:39:38 -0700
Subject: [PATCH 4/4] =?UTF-8?q?refactor(tui):=20render/slash=5Ffuzzy/turn?=
 =?UTF-8?q?=5Fmarker/synthetic=5Fturn/slash=5Fworker=20=E2=80=94=20shared?=
 =?UTF-8?q?=20import=20bridge,=20tier=20table,=20marker=20update=20helper?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tui_gateway/render.py         |  43 +++++--------
 tui_gateway/slash_fuzzy.py    |  78 +++++++++--------------
 tui_gateway/slash_worker.py   |  60 ++++++++----------
 tui_gateway/synthetic_turn.py |  91 +++++++++------------------
 tui_gateway/turn_marker.py    | 112 ++++++++++++++--------------------
 5 files changed, 144 insertions(+), 240 deletions(-)

diff --git a/tui_gateway/render.py b/tui_gateway/render.py
index c15ddef7c0..5a7adeda49 100644
--- a/tui_gateway/render.py
+++ b/tui_gateway/render.py
@@ -6,44 +6,31 @@ everything returns None and the TUI falls back to its own markdown.tsx.
 
 from __future__ import annotations
 
+import importlib
 
-def render_message(text: str, cols: int = 80) -> str | None:
+
+def _rich(name: str, *args, cols: int):
+    """Call ``agent.rich_output.<name>(*args, cols=cols)``; retry without ``cols`` for older
+    signatures; None when the module is missing or the renderer fails."""
     try:
-        from agent.rich_output import format_response
-    except ImportError:
+        fn = getattr(importlib.import_module("agent.rich_output"), name)
+    except (ImportError, AttributeError):
         return None
-
     try:
-        return format_response(text, cols=cols)
+        return fn(*args, cols=cols)
     except TypeError:
-        return format_response(text)
+        return fn(*args)
     except Exception:
         return None
 
 
+def render_message(text: str, cols: int = 80) -> str | None:
+    return _rich("format_response", text, cols=cols)
+
+
 def render_diff(text: str, cols: int = 80) -> str | None:
-    try:
-        from agent.rich_output import render_diff as _rd
-    except ImportError:
-        return None
-
-    try:
-        return _rd(text, cols=cols)
-    except TypeError:
-        return _rd(text)
-    except Exception:
-        return None
+    return _rich("render_diff", text, cols=cols)
 
 
 def make_stream_renderer(cols: int = 80):
-    try:
-        from agent.rich_output import StreamingRenderer
-    except ImportError:
-        return None
-
-    try:
-        return StreamingRenderer(cols=cols)
-    except TypeError:
-        return StreamingRenderer()
-    except Exception:
-        return None
+    return _rich("StreamingRenderer", cols=cols)
diff --git a/tui_gateway/slash_fuzzy.py b/tui_gateway/slash_fuzzy.py
index 254b811e3a..01031ef631 100644
--- a/tui_gateway/slash_fuzzy.py
+++ b/tui_gateway/slash_fuzzy.py
@@ -1,12 +1,11 @@
 """Description-aware fuzzy scoring for slash-menu completions.
 
-Ported from superagent-ai/grok-cli ``src/ui/slash-menu.ts`` (mirrored on the
-TUI client in ``ui-tui/src/app/slash/fuzzyScore.ts``): candidates are scored
-in tiers — exact match on the command token (0), prefix (1), substring (2) —
-and the DESCRIPTION text is tokenized and matched at a +3 offset (exact word
-3, word prefix 4, word substring 5). Typing ``/summary`` thus surfaces a
-command whose description mentions summaries even though no command name
-starts with it. Lower score wins; ``math.inf`` means no match.
+Ported from superagent-ai/grok-cli ``src/ui/slash-menu.ts`` (mirrored on the TUI client in
+``ui-tui/src/app/slash/fuzzyScore.ts``): candidates are scored in tiers — exact match on the
+command token (0), prefix (1), substring (2) — and the DESCRIPTION text is tokenized and matched
+at a +3 offset (exact word 3, word prefix 4, word substring 5). Typing ``/summary`` thus surfaces
+a command whose description mentions summaries even though no command name starts with it.
+Lower score wins; ``math.inf`` means no match.
 """
 
 from __future__ import annotations
@@ -17,6 +16,13 @@ from typing import Callable
 
 _TOKEN_SPLIT = re.compile(r"[^a-z0-9]+")
 
+# (tier bump, field predicate) in priority order: exact, prefix, substring.
+_TIERS = (
+    (0, lambda field, q: field == q or f"/{field}" == q),
+    (1, lambda field, q: field.startswith(q) or f"/{field}".startswith(q)),
+    (2, lambda field, q: q in field),
+)
+
 
 def tokenize_search_text(value: str) -> list[str]:
     """Lowercase ``value`` and return it alongside its alphanumeric words."""
@@ -30,31 +36,16 @@ def normalize_slash_search_query(query: str) -> str:
 
 
 def _score_fields(fields: list[str], query: str, offset: int) -> float:
-    for field in fields:
-        if field == query or f"/{field}" == query:
-            return offset
-    for field in fields:
-        if field.startswith(query) or f"/{field}".startswith(query):
-            return offset + 1
-    for field in fields:
-        if query in field:
-            return offset + 2
-    return math.inf
+    return next((offset + bump for bump, pred in _TIERS if any(pred(f, query) for f in fields)), math.inf)
 
 
 def score_slash_completion_item(item: dict, query: str) -> float:
-    """Score one completion item dict (``text`` + ``meta``) against ``query``.
-
-    ``text`` is the replacement token (may carry a leading slash or trailing
-    space); ``meta`` is the human description. Lower is better; ``math.inf``
-    means no match at all.
-    """
+    """Score one completion item dict against ``query``: ``text`` is the replacement token (may carry
+    a leading slash or trailing space); ``meta`` is the human description. Lower is better."""
     name = str(item.get("text", "")).strip().lstrip("/")
-    command_fields = tokenize_search_text(name)
-    description_fields = tokenize_search_text(str(item.get("meta", "")))
     return min(
-        _score_fields(command_fields, query, 0),
-        _score_fields(description_fields, query, 3),
+        _score_fields(tokenize_search_text(name), query, 0),
+        _score_fields(tokenize_search_text(str(item.get("meta", ""))), query, 3),
     )
 
 
@@ -63,29 +54,16 @@ def fuzzy_rank_slash_items(
 ) -> tuple[list[dict], Callable[[dict], float]]:
     """Merge description/substring matches into ``items`` and sort by score.
 
-    ``items`` are the completer's own (prefix-filtered) rows and keep their
-    identity; ``catalog`` is the full command/skill universe, from which any
-    entry the prefix filter missed but the fuzzy scorer matches is appended.
-    Returns the score-sorted rows (stable within a tier) plus a ``score_of``
-    lookup for downstream rankers to use as a leading sort key.
+    ``items`` are the completer's own (prefix-filtered) rows and keep their identity; ``catalog`` is
+    the full command/skill universe, from which any entry the prefix filter missed but the fuzzy
+    scorer matches is appended. Returns the score-sorted rows (stable within a tier) plus a
+    ``score_of`` lookup for downstream rankers to use as a leading sort key.
     """
     seen = {str(item.get("text", "")).strip() for item in items}
-    merged = list(items)
-    for item in catalog:
-        if str(item.get("text", "")).strip() in seen:
-            continue
-        if not math.isinf(score_slash_completion_item(item, query)):
-            merged.append(item)
-
-    scores: dict[int, float] = {}
-    scored: list[tuple[float, int, dict]] = []
-    for index, item in enumerate(merged):
-        score = score_slash_completion_item(item, query)
-        if math.isinf(score):
-            continue
-        scores[id(item)] = score
-        scored.append((score, index, item))
-    scored.sort(key=lambda entry: (entry[0], entry[1]))
-
-    ranked = [item for _, _, item in scored]
+    merged = list(items) + [
+        item for item in catalog
+        if str(item.get("text", "")).strip() not in seen and not math.isinf(score_slash_completion_item(item, query))
+    ]
+    scores = {id(item): score_slash_completion_item(item, query) for item in merged}
+    ranked = sorted((item for item in merged if not math.isinf(scores[id(item)])), key=lambda item: scores[id(item)])
     return ranked, lambda item: scores.get(id(item), math.inf)
diff --git a/tui_gateway/slash_worker.py b/tui_gateway/slash_worker.py
index 7e55e22949..d7b4b33162 100644
--- a/tui_gateway/slash_worker.py
+++ b/tui_gateway/slash_worker.py
@@ -3,12 +3,11 @@
 Protocol: reads JSON lines from stdin {id, command}, writes {id, ok, output|error} to stdout.
 """
 
-# Stop a ``utils/`` (or ``proxy/``, ``ui/``) package in the launch directory
-# from shadowing Hermes's own top-level modules.  This worker is spawned as
-# ``-m tui_gateway.slash_worker`` and inherits the user's CWD, so the ``import
-# cli`` below would otherwise resolve ``utils`` to a colliding local package and
-# crash the child in a retry loop.  ``hermes_bootstrap`` lives at the repo root
-# (no collision risk), so importing it before the guard runs is safe.
+# Stop a ``utils/`` (or ``proxy/``, ``ui/``) package in the launch directory from shadowing Hermes's
+# own top-level modules. This worker is spawned as ``-m tui_gateway.slash_worker`` and inherits the
+# user's CWD, so the ``import cli`` below would otherwise resolve ``utils`` to a colliding local
+# package and crash the child in a retry loop. ``hermes_bootstrap`` lives at the repo root (no
+# collision risk), so importing it before the guard runs is safe.
 import hermes_bootstrap
 
 hermes_bootstrap.harden_import_path()
@@ -42,11 +41,8 @@ def _is_orphaned(original_ppid, getppid=os.getppid) -> bool:
 
 
 def _prepare_slash_worker_runtime() -> None:
-    """Start bounded MCP discovery before HermesCLI snapshots tools.
-
-    Each slash_worker child is its own process — the parent ``hermes serve``
-    discovery thread does not populate this registry.
-    """
+    """Start bounded MCP discovery before HermesCLI snapshots tools: each slash_worker child is its
+    own process — the parent ``hermes serve`` discovery thread does not populate this registry."""
     from hermes_cli.mcp_startup import start_background_mcp_discovery, wait_for_mcp_discovery
 
     start_background_mcp_discovery(logger=logger, thread_name="slash-worker-mcp-discovery")
@@ -71,27 +67,22 @@ def _run(cli: HermesCLI, command: str) -> str:
         return ""
     if not cmd.startswith("/"):
         cmd = f"/{cmd}"
-
     buf = io.StringIO()
-
-    # Rich Console captures its file handle at construction, so redirect_stdout
-    # won't affect it; swap the console's file so self.console.print() is captured.
+    # Rich Console captures its file handle at construction, so redirect_stdout won't affect it;
+    # swap the console's file so self.console.print() is captured.
     cli.console = Console(file=buf, force_terminal=True, width=120)
-
     old = getattr(cli_mod, "_cprint", None)
     if old is not None:
         cli_mod._cprint = lambda text: print(text)
-
     try:
         with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
             cli.process_command(cmd)
     finally:
         if old is not None:
             cli_mod._cprint = old
-
-    # Desktop chat bubbles render plain text, not ANSI.  A command that emits
-    # Rich color (e.g. /journey building its own Console under the gateway's
-    # inherited COLORTERM) would leak raw escapes; strip at this single choke point.
+    # Desktop chat bubbles render plain text, not ANSI. A command that emits Rich color (e.g.
+    # /journey building its own Console under the gateway's inherited COLORTERM) would leak raw
+    # escapes; strip at this single choke point.
     from tools.ansi_strip import strip_ansi
 
     return strip_ansi(buf.getvalue().rstrip())
@@ -106,48 +97,47 @@ def main():
     os.environ["HERMES_SESSION_KEY"] = args.session_key
     os.environ["HERMES_INTERACTIVE"] = "1"
 
-    # Start before the (hundreds-of-ms) HermesCLI build — that window is itself
-    # an orphan risk if the gateway dies mid-spawn.
+    # Start before the (hundreds-of-ms) HermesCLI build — that window is itself an orphan risk if
+    # the gateway dies mid-spawn.
     _start_parent_death_watchdog(os.getppid())
     _prepare_slash_worker_runtime()
 
     with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
         cli = HermesCLI(model=args.model or None, compact=True, resume=args.session_key, verbose=False)
 
-    # Spurious stdin-EOF recovery (same shared-file-description O_NONBLOCK issue
-    # as the gateway entry point — any child inheriting fd 0 can flip the flag).
+    # Spurious stdin-EOF recovery (same shared-file-description O_NONBLOCK issue as the gateway
+    # entry point — any child inheriting fd 0 can flip the flag).
     _sw_recovery_times: list[float] = []
 
     def _sw_log(reason: str) -> None:
         print(f"[slash-worker] {reason}", file=sys.stderr, flush=True)
 
+    def _reply(**fields) -> None:
+        sys.stdout.write(json.dumps(fields) + "\n")
+        sys.stdout.flush()
+
     while True:
         raw = sys.stdin.readline()
         if not raw:
             if not handle_spurious_eof(_sw_recovery_times, _sw_log):
                 break
             continue
-
         line = raw.strip()
         if not line:
             continue
-
         _in_flight.set()
         rid = None
         try:
             req = json.loads(line)
             rid = req.get("id")
-            out = _run(cli, req.get("command", ""))
-            sys.stdout.write(json.dumps({"id": rid, "ok": True, "output": out}) + "\n")
-            sys.stdout.flush()
+            _reply(id=rid, ok=True, output=_run(cli, req.get("command", "")))
         except Exception as e:
-            sys.stdout.write(json.dumps({"id": rid, "ok": False, "error": str(e)}) + "\n")
-            sys.stdout.flush()
+            _reply(id=rid, ok=False, error=str(e))
         finally:
             _in_flight.clear()
-            # Workers persist for the TUI session: release allocator pages at the
-            # command boundary like other long-lived gateway processes
-            # (trim_memory's shared cooldown coalesces nearby activity).
+            # Workers persist for the TUI session: release allocator pages at the command boundary
+            # like other long-lived gateway processes (trim_memory's shared cooldown coalesces
+            # nearby activity).
             try:
                 from hermes_cli.mem_trim import trim_memory
 
diff --git a/tui_gateway/synthetic_turn.py b/tui_gateway/synthetic_turn.py
index a1df1965a5..6c1fd45c59 100644
--- a/tui_gateway/synthetic_turn.py
+++ b/tui_gateway/synthetic_turn.py
@@ -1,22 +1,15 @@
 """Synthetic GIL-heavy turn driver for the AC-4 isolation certify harness.
 
-The regime under test is interpreter-wide GIL starvation: concurrent heavy agent
-turns run compute in threads of the SERVING process and starve the event loop
-that flushes WebSocket frames (loop thread parked in ``take_gil`` — NOT blocked
-on I/O).  To certify the isolation fix without spending real tokens, the turn
-driver must reproduce THAT: sustained pure-Python CPU holding the GIL for the
-turn's duration.  A network/sleep stub is WRONG — it releases the GIL during I/O
-and never reproduces ``take_gil`` contention, so a green off it is fake.
+The regime under test is interpreter-wide GIL starvation: concurrent heavy agent turns run compute in
+threads of the SERVING process and starve the event loop that flushes WebSocket frames (loop thread
+parked in ``take_gil`` — NOT blocked on I/O). To certify the isolation fix without spending real
+tokens the turn driver must reproduce THAT: sustained pure-Python CPU holding the GIL. A network/
+sleep stub is WRONG — it releases the GIL during I/O, so a green off it is fake.
 
-This module is a **test seam**: dead unless ``HERMES_ISO_CERTIFY_SYNTH_TURN=1``.
-When armed, ``tui_gateway.server._make_agent`` returns a
-:class:`SyntheticHeavyAgent` instead of a real ``AIAgent``.  Both the in-process
-``_pool`` path (isolation OFF) and the compute-host child path (isolation ON)
-build through ``_make_agent``, so the isolation boundary is the only variable.
-
-Per-turn intensity (wall duration, CPU chunk size, delta cadence, token
-accounting) rides in the prompt text as a JSON object; any other prompt falls
-back to env / built-in defaults.
+Test seam: dead unless ``HERMES_ISO_CERTIFY_SYNTH_TURN=1``. When armed, ``server._make_agent`` returns
+a :class:`SyntheticHeavyAgent` on both the in-process ``_pool`` path (isolation OFF) and the compute-
+host child path (isolation ON), so the isolation boundary is the only variable. Per-turn intensity
+rides in the prompt text as a JSON object; any other prompt falls back to env / built-in defaults.
 """
 
 from __future__ import annotations
@@ -38,10 +31,10 @@ def synth_turn_armed() -> bool:
 class SyntheticHeavyAgent:
     """An AIAgent-shaped object whose turn is a GIL-holding CPU burn.
 
-    Presents only the surface ``tui_gateway.server``'s turn path and status
-    helpers read (``run_conversation``/``interrupt``/``clear_interrupt`` plus the
-    ``model``/``provider``/``session_*`` attributes consumed by ``_get_usage`` and
-    ``_session_info``).  Never opens a socket or spawns a subprocess.
+    Presents only the surface ``tui_gateway.server``'s turn path and status helpers read
+    (``run_conversation``/``interrupt``/``clear_interrupt`` plus the ``model``/``provider``/
+    ``session_*`` attributes consumed by ``_get_usage`` and ``_session_info``). Never opens a socket
+    or spawns a subprocess.
     """
 
     def __init__(self, session_id: str, *, model: str = "synthetic-heavy") -> None:
@@ -49,9 +42,7 @@ class SyntheticHeavyAgent:
         self.model = model
         self.provider = "synthetic"
         self.api_mode = "chat_completions"
-        self.base_url = ""
-        self.api_key = ""
-        self.platform = ""
+        self.base_url = self.api_key = self.platform = ""
         self.tools: list[Any] = []
         self.reasoning_config: dict | None = None
         self.service_tier: str | None = None
@@ -59,17 +50,13 @@ class SyntheticHeavyAgent:
         self._config_context_length = 200_000
         self._cached_system_prompt = ""
         # Cumulative session counters (read by _get_usage → status bar).
-        self.session_input_tokens = 0
-        self.session_output_tokens = 0
-        self.session_prompt_tokens = 0
-        self.session_completion_tokens = 0
-        self.session_reasoning_tokens = 0
-        self.session_total_tokens = 0
+        self.session_input_tokens = self.session_output_tokens = self.session_prompt_tokens = 0
+        self.session_completion_tokens = self.session_reasoning_tokens = self.session_total_tokens = 0
         self.session_api_calls = 0
         self.history: list[dict[str, str]] = []
         self._interrupt = threading.Event()
 
-    # ── interrupt contract (mirrors AIAgent) ───────────────────────────
+    # interrupt contract (mirrors AIAgent)
     def clear_interrupt(self) -> None:
         self._interrupt.clear()
 
@@ -96,15 +83,15 @@ class SyntheticHeavyAgent:
         return {
             # Wall-clock seconds of GIL-holding compute.
             "duration_s": float(spec.get("duration_s", _env_float("HERMES_ISO_CERTIFY_DURATION_S", 8.0))),
-            # Pure-Python ops per interrupt-check chunk: small enough that an
-            # interrupt lands within ms, large enough to stay hot on the GIL.
+            # Pure-Python ops per interrupt-check chunk: small enough that an interrupt lands within
+            # ms, large enough to stay hot on the GIL.
             "chunk": int(spec.get("chunk", _env_int("HERMES_ISO_CERTIFY_CHUNK", 20_000))),
             # Streamed-delta cadence: each delta is a loop wakeup marshalling a frame.
             "delta_interval_s": float(spec.get("delta_interval_s", _env_float("HERMES_ISO_CERTIFY_DELTA_S", 0.05))),
             # Notional output tokens per delta (drives the 100K+-token heavy-turn proxy).
             "tokens_per_delta": int(spec.get("tokens_per_delta", _env_int("HERMES_ISO_CERTIFY_TPD", 512))),
-            # Optional per-chunk sleep for a mixed regime (0 = pure burn).  --dry-run
-            # uses a short duration, NOT a sleep, so it still exercises the real seam.
+            # Optional per-chunk sleep for a mixed regime (0 = pure burn). --dry-run uses a short
+            # duration, NOT a sleep, so it still exercises the real seam.
             "sleep_s": float(spec.get("sleep_s", 0.0)),
         }
 
@@ -136,8 +123,8 @@ class SyntheticHeavyAgent:
             now = time.monotonic()
             if now - start >= duration:
                 break
-            # A tight integer loop never releases the GIL — the exact contention
-            # that starves the serving loop.
+            # A tight integer loop never releases the GIL — the exact contention that starves the
+            # serving loop.
             for _ in range(chunk):
                 acc = (acc * 1_000_003 + 12_345) & 0xFFFFFFFFFFFFFFFF
             if sleep_s:
@@ -152,34 +139,20 @@ class SyntheticHeavyAgent:
                 last_delta = now
 
         self.session_api_calls += 1
-        # Fold the checksum into the reply so the loop can't be eliminated and
-        # the turn is deterministic and inspectable.
+        # Fold the checksum into the reply so the loop can't be eliminated and the turn is
+        # deterministic and inspectable.
         final = (
             f"[synthetic heavy turn] deltas={deltas} "
             f"out_tokens={self.session_output_tokens} "
             f"interrupted={interrupted} checksum={acc & 0xFFFF:04x}"
         )
-        messages = [
-            *base_history,
-            {"role": "user", "content": str(message)[:200]},
-            {"role": "assistant", "content": final},
-        ]
-        self.history = messages
-        return {
-            "final_response": final,
-            "messages": messages,
-            "interrupted": interrupted,
-            "error": None,
-            "last_reasoning": None,
-        }
+        self.history = [*base_history, {"role": "user", "content": str(message)[:200]}, {"role": "assistant", "content": final}]
+        return {"final_response": final, "messages": self.history, "interrupted": interrupted, "error": None, "last_reasoning": None}
 
 
 def maybe_build_synthetic_agent(session_id: str, model_override: Any = None) -> SyntheticHeavyAgent | None:
-    """Return a :class:`SyntheticHeavyAgent` when the seam is armed, else ``None``.
-
-    ``model_override`` (dict or str) only influences the reported ``model`` label;
-    it never changes the compute.
-    """
+    """Return a :class:`SyntheticHeavyAgent` when the seam is armed, else ``None``. ``model_override``
+    (dict or str) only influences the reported ``model`` label; it never changes the compute."""
     if not synth_turn_armed():
         return None
     model = "synthetic-heavy"
@@ -190,8 +163,4 @@ def maybe_build_synthetic_agent(session_id: str, model_override: Any = None) ->
     return SyntheticHeavyAgent(session_id, model=model)
 
 
-__all__ = [
-    "SyntheticHeavyAgent",
-    "maybe_build_synthetic_agent",
-    "synth_turn_armed",
-]
+__all__ = ["SyntheticHeavyAgent", "maybe_build_synthetic_agent", "synth_turn_armed"]
diff --git a/tui_gateway/turn_marker.py b/tui_gateway/turn_marker.py
index dfb58e4f8c..ae938120a9 100644
--- a/tui_gateway/turn_marker.py
+++ b/tui_gateway/turn_marker.py
@@ -1,23 +1,18 @@
 """Durable interrupted-turn markers for the desktop/TUI auto-continue path.
 
-A running turn's progress lives only in process memory (the agent flushes to
-SQLite at turn end), so an app/backend/machine death mid-turn leaves no durable
-trace of the interrupted prompt.  A marker is written when a turn starts and
-cleared when it concludes — success, handled error, or interrupt all clear it —
-so only a process death leaves one behind.  ``session.resume`` reads it to
-decide whether to auto-continue (``_maybe_schedule_auto_continue`` in
-``tui_gateway/server.py``).
-
-Markers are stored per ``HERMES_HOME`` (profile sessions keep state in their own
-profile dir) and the file is bounded: writes prune entries older than
-``_MAX_AGE_SECS`` and cap the count, so a crash streak can't grow it unboundedly.
-
-Every function is best-effort — marker bookkeeping must never break a turn — so
-I/O errors degrade to "no marker" instead of raising.
+A running turn's progress lives only in process memory (the agent flushes to SQLite at turn end),
+so an app/backend/machine death mid-turn leaves no durable trace of the interrupted prompt. A marker
+is written when a turn starts and cleared when it concludes (success, handled error, or interrupt),
+so only a process death leaves one behind; ``session.resume`` reads it to decide whether to
+auto-continue (``_maybe_schedule_auto_continue``). Markers are stored per ``HERMES_HOME`` (profile
+sessions keep state in their own profile dir); writes prune entries older than ``_MAX_AGE_SECS`` and
+cap the count so a crash streak can't grow the file unboundedly. Every function is best-effort —
+marker bookkeeping must never break a turn — so I/O errors degrade to "no marker" instead of raising.
 """
 
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
 import os
@@ -29,19 +24,16 @@ from typing import Any
 
 logger = logging.getLogger(__name__)
 
-_MARKER_DIR = "desktop"
-_MARKER_FILE = "interrupted_turns.json"
 _MAX_AGE_SECS = 24 * 3600
 _MAX_ENTRIES = 32
-# Enough to re-submit any realistic prompt; guards against a multi-megabyte
-# paste being journaled on every turn.
+# Enough to re-submit any realistic prompt; guards against a multi-megabyte paste being journaled.
 _MAX_PROMPT_CHARS = 64_000
 
 _lock = threading.Lock()
 
 
 def _marker_path(home: Path | str) -> Path:
-    return Path(home) / _MARKER_DIR / _MARKER_FILE
+    return Path(home) / "desktop" / "interrupted_turns.json"
 
 
 def _started_at(entry: dict) -> float:
@@ -57,17 +49,14 @@ def _load(path: Path) -> dict[str, dict]:
     except Exception:
         logger.debug("unreadable turn-marker file %s; starting fresh", path, exc_info=True)
         return {}
-    if not isinstance(data, dict):
-        return {}
-    return {k: v for k, v in data.items() if isinstance(v, dict)}
+    return {k: v for k, v in data.items() if isinstance(v, dict)} if isinstance(data, dict) else {}
 
 
 def _prune(entries: dict[str, dict], now: float) -> dict[str, dict]:
     fresh = {k: e for k, e in entries.items() if now - _started_at(e) <= _MAX_AGE_SECS}
     if len(fresh) <= _MAX_ENTRIES:
         return fresh
-    newest = sorted(fresh.items(), key=lambda item: _started_at(item[1]), reverse=True)
-    return dict(newest[:_MAX_ENTRIES])
+    return dict(sorted(fresh.items(), key=lambda item: _started_at(item[1]), reverse=True)[:_MAX_ENTRIES])
 
 
 def _store(path: Path, entries: dict[str, dict]) -> None:
@@ -81,54 +70,47 @@ def _store(path: Path, entries: dict[str, dict]) -> None:
             json.dump(entries, f)
         os.replace(tmp, path)
     except Exception:
-        try:
+        with contextlib.suppress(OSError):
             os.unlink(tmp)
-        except OSError:
-            pass
         raise
 
 
-def record_turn_start(
-    home: Path | str, session_key: str, prompt: str, *, attempts: int = 0
-) -> None:
-    """Persist the marker for a turn that is about to run.
-
-    ``attempts`` counts how many auto-continues led to this run: 0 for a
-    user-initiated turn, N for the Nth automatic re-run — the crash-loop
-    breaker reads it back on the next resume.
-    """
-    if not session_key or not prompt:
-        return
-    now = time.time()
-    entry = {
-        "attempts": max(0, int(attempts)),
-        "prompt": prompt[:_MAX_PROMPT_CHARS],
-        "started_at": now,
-    }
-    try:
-        with _lock:
-            path = _marker_path(home)
-            entries = _prune(_load(path), now)
-            entries[session_key] = entry
-            _store(path, entries)
-    except Exception:
-        logger.debug("failed to record turn marker for %s", session_key, exc_info=True)
-
-
-def clear_turn_marker(home: Path | str, session_key: str) -> None:
-    """Remove the marker once its turn concluded (any outcome the client saw)."""
-    if not session_key:
-        return
+def _update(home: Path | str, session_key: str, mutate, what: str) -> None:
+    """Load → ``mutate(entries)`` → store under the lock; ``mutate`` returns False to skip the write."""
     try:
         with _lock:
             path = _marker_path(home)
             entries = _load(path)
-            if session_key not in entries:
-                return
-            del entries[session_key]
-            _store(path, entries)
+            if mutate(entries) is not False:
+                _store(path, entries)
     except Exception:
-        logger.debug("failed to clear turn marker for %s", session_key, exc_info=True)
+        logger.debug("failed to %s turn marker for %s", what, session_key, exc_info=True)
+
+
+def record_turn_start(home: Path | str, session_key: str, prompt: str, *, attempts: int = 0) -> None:
+    """Persist the marker for a turn that is about to run.
+
+    ``attempts`` counts how many auto-continues led to this run: 0 for a user-initiated turn, N for
+    the Nth automatic re-run — the crash-loop breaker reads it back on the next resume.
+    """
+    if not session_key or not prompt:
+        return
+    now = time.time()
+    entry = {"attempts": max(0, int(attempts)), "prompt": prompt[:_MAX_PROMPT_CHARS], "started_at": now}
+
+    def mutate(entries: dict) -> None:
+        pruned = _prune(entries, now)
+        entries.clear()
+        entries.update(pruned)
+        entries[session_key] = entry
+
+    _update(home, session_key, mutate, "record")
+
+
+def clear_turn_marker(home: Path | str, session_key: str) -> None:
+    """Remove the marker once its turn concluded (any outcome the client saw)."""
+    if session_key:
+        _update(home, session_key, lambda entries: entries.pop(session_key, None) is not None, "clear")
 
 
 def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | None:
@@ -146,8 +128,6 @@ def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | Non
     if not prompt.strip():
         return None
     try:
-        started_at = _started_at(entry)
-        attempts = max(0, int(entry.get("attempts") or 0))
+        return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry)}
     except (TypeError, ValueError):
         return None
-    return {"attempts": attempts, "prompt": prompt, "started_at": started_at}