diff --git a/tui_gateway/session_lifecycle.py b/tui_gateway/session_lifecycle.py
index c69455a957..c84ab68013 100644
--- a/tui_gateway/session_lifecycle.py
+++ b/tui_gateway/session_lifecycle.py
@@ -1,7 +1,8 @@
-"""Session lifecycle: active-session slot leases, finalize/teardown/close, turn interrupt, WS-orphan reap scheduling, transport-scoped close.
+"""Session lifecycle: active-session slot leases, finalize/teardown/close, turn interrupt,
+WS-orphan reap scheduling, transport-scoped close.
 
-Bodies are rebound onto server.py's globals at install time (see
-method_ctx.bind_module), so they reference server.py globals bare.
+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
@@ -17,7 +18,6 @@ def _notify_session_boundary(event_type: str, session_id: str | None, platform:
     """Fire session lifecycle hooks with CLI parity."""
     with contextlib.suppress(Exception):
         from hermes_cli.lifecycle import finalize_session, invoke_hook
-
         if event_type == "on_session_finalize":
             finalize_session(session_id=session_id, platform=_resolve_agent_platform(platform))
         else:
@@ -27,25 +27,18 @@ def _notify_session_boundary(event_type: str, session_id: str | None, platform:
 _SESSION_OWNERSHIP_UNAVAILABLE = "Hermes could not safely reserve this session. Try again."
 
 _AUTOMATIC_SESSION_END_REASONS = frozenset({
-    "ws_orphan_reap", "ws_disconnect", "idle_timeout", "lru_evict", "tui_shutdown",
-})
+    "ws_orphan_reap", "ws_disconnect", "idle_timeout", "lru_evict", "tui_shutdown"})
 
 
 def _claim_active_session_slot(
     session_key: str, *, live_session_id: str, surface: str = "tui", profile_home: str | Path | None = None
 ) -> tuple[Any, str | None]:
-    track_liveness = str(surface or "").strip().lower() == "desktop"
     try:
         from hermes_cli.active_sessions import try_acquire_active_session
-
         return try_acquire_active_session(
-            session_id=session_key,
-            surface=surface,
-            config=_load_cfg(),
-            metadata={"live_session_id": live_session_id},
-            registry_home=profile_home,
-            track_liveness=track_liveness,
-        )
+            session_id=session_key, surface=surface, config=_load_cfg(),
+            metadata={"live_session_id": live_session_id}, registry_home=profile_home,
+            track_liveness=str(surface or "").strip().lower() == "desktop")
     except Exception as exc:
         logger.warning("Failed to claim active session slot: %s", exc)
         # Fail CLOSED regardless of surface: an errored claim has NOT proven the session
@@ -63,18 +56,16 @@ def _ensure_active_session_slot(sid: str, session: dict) -> str | None:
     if session.get("active_session_lease") is not None:
         return None
     lease, limit_message = _claim_active_session_slot(
-        str(session.get("session_key") or ""),
-        live_session_id=sid,
-        surface=_session_source(session),
-        profile_home=session.get("profile_home"),
-    )
-    if limit_message is not None:
-        return limit_message
-    session["active_session_lease"] = lease
-    return None
+        str(session.get("session_key") or ""), live_session_id=sid,
+        surface=_session_source(session), profile_home=session.get("profile_home"))
+    if limit_message is None:
+        session["active_session_lease"] = lease
+    return limit_message
 
 
-def _release_active_session_lease(lease) -> bool:
+def _release_active_session_slot(session: dict | None) -> bool:
+    """Release the session's lease (liveness-tracked leases get 3 tries); True when released."""
+    lease = session.get("active_session_lease") if session else None
     if lease is None:
         return True
     attempts = 3 if getattr(lease, "track_liveness", False) else 1
@@ -87,33 +78,24 @@ def _release_active_session_lease(lease) -> bool:
                 logger.warning("Failed to release active session slot", exc_info=True)
                 return False
             time.sleep(0.05 * (attempt + 1))
-    return bool(getattr(lease, "released", True) or not getattr(lease, "enabled", True))
-
-
-def _release_active_session_slot(session: dict | None) -> bool:
-    if not session:
-        return True
-    lease = session.get("active_session_lease")
-    if _release_active_session_lease(lease):
-        if session.get("active_session_lease") is lease:
-            session.pop("active_session_lease", None)
-        return True
-    return False
+    if not (getattr(lease, "released", True) or not getattr(lease, "enabled", True)):
+        return False
+    if session.get("active_session_lease") is lease:
+        session.pop("active_session_lease", None)
+    return True
 
 
 @contextlib.contextmanager
 def _other_runtime_lease_guard(session_id: str, session: dict):
-    """Release this runtime and lock sibling ownership through the DB write."""
+    """Release this runtime and lock sibling ownership through the DB write.
+
+    Yields True (another runtime owns the lifecycle -> preserve) when the guard cannot be
+    loaded or entered after 3 tries: an unknown ownership state must never end a row."""
     lease = session.get("active_session_lease")
     try:
-        from hermes_cli.active_sessions import (
-            active_session_liveness_guard,
-            release_active_session_liveness_guard,
-        )
+        from hermes_cli.active_sessions import active_session_liveness_guard, release_active_session_liveness_guard
     except Exception as exc:
-        logger.warning(
-            "Failed to load active session ownership guard; preserving session %s: %s", session_id, exc
-        )
+        logger.warning("Failed to load active session ownership guard; preserving session %s: %s", session_id, exc)
         yield True
         return
 
@@ -134,20 +116,14 @@ def _other_runtime_lease_guard(session_id: str, session: dict):
             if attempt < 2:
                 time.sleep(0.05 * (attempt + 1))
     else:
-        logger.warning(
-            "Failed to inspect active session leases; preserving session %s: %s", session_id, last_error
-        )
+        logger.warning("Failed to inspect active session leases; preserving session %s: %s", session_id, last_error)
         yield True
         return
     try:
         yield active
     finally:
         stack.close()
-        if (
-            lease is not None
-            and getattr(lease, "released", False)
-            and session.get("active_session_lease") is lease
-        ):
+        if lease is not None and getattr(lease, "released", False) and session.get("active_session_lease") is lease:
             session.pop("active_session_lease", None)
 
 
@@ -159,12 +135,10 @@ def _transfer_active_session_slot(sid: str, session: dict, *, new_session_id: st
         return True
     try:
         from hermes_cli.active_sessions import transfer_active_session
-
         if transfer_active_session(lease, session_id=new_session_id, metadata={"live_session_id": sid}):
             return True
     except Exception:
         logger.debug("Failed to transfer active session slot", exc_info=True)
-
     if getattr(lease, "track_liveness", False):
         return False
 
@@ -172,27 +146,22 @@ def _transfer_active_session_slot(sid: str, session: dict, *, new_session_id: st
     # releasing the old one so a gateway at the cap can't grab the freed slot and leave
     # this session lease-less. On reserve failure KEEP the old lease.
     new_lease, limit_message = _claim_active_session_slot(
-        new_session_id,
-        live_session_id=sid,
-        surface=_session_source(session),
-        profile_home=session.get("profile_home"),
-    )
-    if new_lease is not None:
-        old_lease = session.pop("active_session_lease", None)
-        if old_lease is not None:
-            try:
-                old_lease.release()
-            except Exception:
-                logger.debug("Failed to release stale active session slot", exc_info=True)
-        session["active_session_lease"] = new_lease
-        return True
-    if limit_message:
-        logger.warning(
-            "Compression session lease re-anchor failed (kept old lease): "
-            "sid=%s new_session_id=%s reason=%s",
-            sid, new_session_id, limit_message,
-        )
-    return False
+        new_session_id, live_session_id=sid, surface=_session_source(session), profile_home=session.get("profile_home"))
+    if new_lease is None:
+        if limit_message:
+            logger.warning(
+                "Compression session lease re-anchor failed (kept old lease): "
+                "sid=%s new_session_id=%s reason=%s",
+                sid, new_session_id, limit_message)
+        return False
+    old_lease = session.pop("active_session_lease", None)
+    if old_lease is not None:
+        try:
+            old_lease.release()
+        except Exception:
+            logger.debug("Failed to release stale active session slot", exc_info=True)
+    session["active_session_lease"] = new_lease
+    return True
 
 
 # Sources this backend must never end in state.db: the messaging gateway owns those
@@ -200,23 +169,18 @@ def _transfer_active_session_slot(sid: str, session: dict, *, new_session_id: st
 # loop, see _finalize_session). Self-created and CLI sources are NOT gateway-owned.
 _NON_GATEWAY_SOURCES = frozenset({
     "", "tui", "cli", "webui", "desktop", "cron", "kanban", "subagent", "test",
-    "local", "acp", "webhook", "api_server", "msgraph_webhook",
-})
+    "local", "acp", "webhook", "api_server", "msgraph_webhook"})
 
 
 def _is_gateway_owned_source(source: str) -> bool:
-    """True when ``source`` is a messaging-gateway platform owning its session lifecycle.
-
-    Structural: any source resolving to a gateway ``Platform`` (enum member or plugin via
-    ``Platform._missing_``) counts, so new platforms are covered automatically. Self-owned
-    sources (``local``/``webhook``/``api_server`` are Platform members) are excluded explicitly.
-    """
+    """True when ``source`` resolves to a gateway ``Platform`` (enum member or plugin via
+    ``Platform._missing_``), so new platforms are covered automatically; self-owned sources
+    (``local``/``webhook``/``api_server`` are Platform members) are excluded explicitly."""
     src = (source or "").strip().lower()
     if src in _NON_GATEWAY_SOURCES:
         return False
     try:
         from gateway.config import Platform
-
         Platform(src)  # raises ValueError for arbitrary non-platform strings
         return True
     except Exception:
@@ -227,14 +191,8 @@ def _lifecycle_own_sid(session: dict, sid_hint: str = "") -> str:
     """Live UI sid for ``session``: hint, stamped ``_sid``, else registry scan."""
     own_sid = str(sid_hint or session.get("_sid") or "")
     if not own_sid:
-        try:
-            with _sessions_lock:
-                for _cand_sid, _cand in _sessions.items():
-                    if _cand is session:
-                        own_sid = _cand_sid
-                        break
-        except Exception:
-            own_sid = ""
+        with contextlib.suppress(Exception), _sessions_lock:
+            own_sid = next((cand_sid for cand_sid, cand in _sessions.items() if cand is session), "")
     return own_sid
 
 
@@ -249,15 +207,12 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
         session["resume_history_error"] = "session resume cancelled"
         history_ready.set()
     _desktop_automatic_cleanup = (
-        end_reason in _AUTOMATIC_SESSION_END_REASONS
-        and _session_source(session).strip().lower() == "desktop"
-    )
+        end_reason in _AUTOMATIC_SESSION_END_REASONS and _session_source(session).strip().lower() == "desktop")
     # Automatic Desktop cleanup releases its lease inside the lock-held lifecycle guard
     # below; explicit close and non-Desktop paths keep force/end semantics.
     if not _desktop_automatic_cleanup:
         _release_active_session_slot(session)
-    stop_event = session.get("_notif_stop")
-    if stop_event is not None:
+    if (stop_event := session.get("_notif_stop")) is not None:
         stop_event.set()
 
     agent = session.get("agent")
@@ -270,26 +225,20 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
     # and ``_session_messages`` alias the SAME list after a turn, so the flush would treat
     # every message as durable and skip it — data loss when finalize is the sole persist
     # path after a WS disconnect/restart.
-    if agent is not None and hasattr(agent, "_persist_session"):
-        snapshot = getattr(agent, "_session_messages", None)
-        if snapshot:
-            with contextlib.suppress(Exception):
-                agent._persist_session(snapshot)
+    if agent is not None and hasattr(agent, "_persist_session") and (snapshot := getattr(agent, "_session_messages", None)):
+        with contextlib.suppress(Exception):
+            agent._persist_session(snapshot)
 
     # interrupted=True so crash-recovery plugins can flush state (mirrors cli.py atexit).
     if agent is not None:
         with contextlib.suppress(Exception):
             from hermes_cli.lifecycle import invoke_hook
-
             invoke_hook(
                 "on_session_end",
                 session_id=getattr(agent, "session_id", None) or session.get("session_key", ""),
-                completed=False,
-                interrupted=True,
+                completed=False, interrupted=True,
                 model=getattr(agent, "model", "unknown"),
-                platform=getattr(agent, "platform", None) or "tui",
-            )
-
+                platform=getattr(agent, "platform", None) or "tui")
     if agent is not None and history and hasattr(agent, "commit_memory_session"):
         with contextlib.suppress(Exception):
             agent.commit_memory_session(history)
@@ -306,47 +255,37 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
     _lifecycle_guard = (
         _other_runtime_lease_guard(session_id, session)
         if _desktop_automatic_cleanup and session_id
-        else contextlib.nullcontext(False)
-    )
+        else contextlib.nullcontext(False))
     with _lifecycle_guard as _other_runtime_owns_lifecycle:
         _tui_owns_lifecycle = not _other_runtime_owns_lifecycle
         if _other_runtime_owns_lifecycle:
             logger.info(
-                "Preserving session %s during %s: another backend owns an active lease",
-                session_id, end_reason,
-            )
+                "Preserving session %s during %s: another backend owns an active lease", session_id, end_reason)
         if session_id:
-            with contextlib.suppress(Exception):
-                # The *session's* profile state.db (app-global remote mode), not the launch profile's.
-                with _session_db(session) as db:
-                    if db is not None:
-                        # Never end gateway-originated sessions (Groundhog Day loop: the
-                        # gateway's self-heal recovers to the parent, compression splits back
-                        # to the reaped child, repeat on every message).
-                        row = db.get_session(session_id)
-                        source = (row or {}).get("source", "")
-                        if _is_gateway_owned_source(source):
-                            _tui_owns_lifecycle = False
-                        elif _tui_owns_lifecycle:
-                            db.end_session(session_id, end_reason)
+            # The *session's* profile state.db (app-global remote mode), not the launch profile's.
+            with contextlib.suppress(Exception), _session_db(session) as db:
+                if db is not None:
+                    # Never end gateway-originated sessions (Groundhog Day loop: the
+                    # gateway's self-heal recovers to the parent, compression splits back
+                    # to the reaped child, repeat on every message).
+                    row = db.get_session(session_id)
+                    if _is_gateway_owned_source((row or {}).get("source", "")):
+                        _tui_owns_lifecycle = False
+                    elif _tui_owns_lifecycle:
+                        db.end_session(session_id, end_reason)
 
     # In-flight async delegations end WITH the session (no return address left). Always
     # interrupt by THIS live UI sid; by durable session_key only when the TUI owns the
     # lifecycle — closing a viewer tab must not kill the gateway's own background work.
     with contextlib.suppress(Exception):
         from tools.async_delegation import interrupt_for_session
-
         interrupt_for_session(
             session_key=str(session_key or "") if _tui_owns_lifecycle else "",
-            origin_ui_session_id=_lifecycle_own_sid(session),
-            reason=end_reason,
-        )
-
+            origin_ui_session_id=_lifecycle_own_sid(session), reason=end_reason)
     # Close the slash-worker in this single ``_finalized``-guarded chokepoint so a direct
     # _finalize_session caller can't leak it. Idempotent: close() is poll()-guarded.
     with contextlib.suppress(Exception):
-        worker = session.get("slash_worker")
-        if worker:
+        if worker := session.get("slash_worker"):
             worker.close()
 
 
@@ -361,36 +300,28 @@ def _announce_session_reclaimed(session: dict, end_reason: str) -> None:
 
     Broadcast, not session-targeted: reap paths run on timer threads with no contextvar
     binding and the WS-orphan case has lost its transport, so ``_emit`` would bottom out
-    on stdio. Best-effort; never breaks teardown.
-    """
+    on stdio. Best-effort; never breaks teardown."""
     if end_reason not in _RECLAIM_END_REASONS:
         return
     try:
-        _broadcast_global_event(
-            "session.reclaimed",
-            {
-                "session_id": str(session.get("_sid") or ""),
-                "stored_session_id": str(session.get("session_key") or ""),
-                "reason": end_reason,
-            },
-        )
+        _broadcast_global_event("session.reclaimed", {
+            "session_id": str(session.get("_sid") or ""),
+            "stored_session_id": str(session.get("session_key") or ""),
+            "reason": end_reason})
     except Exception:
         logger.debug("session.reclaimed broadcast failed", exc_info=True)
 
 
 def _teardown_session(session: dict | None, *, end_reason: str = "tui_close") -> None:
-    """Fully tear down a session: finalize, unregister notifier, close agent.
-
-    Shared by ``session.close`` and the orphaned-WS reaper. The slash-worker is closed in
-    ``_finalize_session`` (the single chokepoint), NOT here. Idempotent via ``_finalized``.
-    """
+    """Fully tear down a session: finalize, unregister notifier, close agent. Shared by
+    ``session.close`` and the orphaned-WS reaper. The slash-worker is closed in
+    ``_finalize_session`` (the single chokepoint), NOT here. Idempotent via ``_finalized``."""
     if not session:
         return
     _finalize_session(session, end_reason=end_reason)
     _announce_session_reclaimed(session, end_reason)
     with contextlib.suppress(Exception):
         from tools.approval import unregister_gateway_notify
-
         if key := session.get("session_key"):
             unregister_gateway_notify(key)
     with contextlib.suppress(Exception):
@@ -415,14 +346,12 @@ def _pop_session_by_id(sid: str) -> dict | None:
     because finalization does slow external work that must not run under
     ``_session_resume_lock``."""
     with _sessions_lock:
-        session = _sessions.get(sid)
+        session = _sessions.pop(sid, None)
         if session is not None:
             session["_closing"] = True
-            _sessions.pop(sid, None)
-    if session is None:
-        return None
-    # Out of _sessions now, so teardown can't recover the live id by scanning — stamp it.
-    session["_sid"] = sid
+    if session is not None:
+        # Out of _sessions now, so teardown can't recover the live id by scanning — stamp it.
+        session["_sid"] = sid
     return session
 
 
@@ -437,9 +366,7 @@ def _teardown_popped_session(session: dict | None, *, end_reason: str = "tui_clo
                 run_thread.join(timeout=_TURN_SETTLE_BEFORE_CLOSE_SECONDS)
             if run_thread.is_alive():
                 logger.warning(
-                    "session turn thread still alive after %.1fs teardown grace",
-                    _TURN_SETTLE_BEFORE_CLOSE_SECONDS,
-                )
+                    "session turn thread still alive after %.1fs teardown grace", _TURN_SETTLE_BEFORE_CLOSE_SECONDS)
         except Exception:
             logger.debug("failed waiting for session turn thread", exc_info=True)
     _teardown_session(session, end_reason=end_reason)
@@ -466,14 +393,12 @@ def _close_session_by_id(
 
 def _ws_session_is_detached(session: dict | None) -> bool:
     """True if a live session is still bound to the disconnected-WS sentinel."""
-    return bool(
-        session and not session.get("_finalized") and session.get("transport") is _detached_ws_transport
-    )
+    return bool(session and not session.get("_finalized") and session.get("transport") is _detached_ws_transport)
 
 
 def _ws_session_is_orphaned(session: dict | None) -> bool:
-    """True if a WS session sits on ``_detached_ws_transport`` (where
-    ``handle_ws`` parks disconnected clients) with no in-flight turn."""
+    """True if a WS session sits on ``_detached_ws_transport`` (where ``handle_ws`` parks
+    disconnected clients) with no in-flight turn."""
     return bool(_ws_session_is_detached(session) and not session.get("running"))
 
 
@@ -484,7 +409,6 @@ def _interrupt_session_turn(sid: str, session: dict, *, request_id: str | None =
     use_compute_host = _session_uses_compute_host(session)
     should_interrupt = bool(session.get("running"))
     run_thread_alive = False
-
     if use_compute_host:
         # The host owns the live turn (parent `running` can lag a blocked tool), so let it
         # decide. Gate on `_compute_host_active`: HostSupervisor.interrupt() calls start(),
@@ -504,7 +428,6 @@ def _interrupt_session_turn(sid: str, session: dict, *, request_id: str | None =
     if not use_compute_host:
         if should_interrupt:
             from agent.interrupt_compat import request_hard_interrupt
-
             request_hard_interrupt(session.get("agent"))
         if not run_thread_alive:
             with session["history_lock"]:
@@ -515,7 +438,6 @@ def _interrupt_session_turn(sid: str, session: dict, *, request_id: str | None =
     _clear_pending(sid)
     with contextlib.suppress(Exception):
         from tools.approval import resolve_gateway_approval
-
         resolve_gateway_approval(session["session_key"], "deny", resolve_all=True)
     return use_compute_host
 
@@ -527,26 +449,11 @@ def _session_owns_durable_lifecycle(session_id: str | None) -> bool:
         return True
     try:
         db = _get_db()
-        if db is None:
-            return True
-        row = db.get_session(session_id)
-        return not _is_gateway_owned_source((row or {}).get("source", ""))
+        return db is None or not _is_gateway_owned_source((db.get_session(session_id) or {}).get("source", ""))
     except Exception:
         return True
 
 
-def _session_async_delegation_selectors(session: dict | None, *, sid_hint: str = "") -> tuple[str, str]:
-    """Ownership selectors for async background work tied to one UI session."""
-    if not session:
-        return "", ""
-    own_sid = _lifecycle_own_sid(session, sid_hint)
-    agent = session.get("agent")
-    session_key = str(session.get("session_key") or "")
-    session_id = getattr(agent, "session_id", None) or session_key
-    owned_session_key = session_key if _session_owns_durable_lifecycle(session_id) else ""
-    return own_sid, owned_session_key
-
-
 def _session_has_active_delegations(sid: str, session: dict | None = None) -> bool:
     """True when UI session ``sid`` still owns live background work — by live UI sid AND,
     when the TUI owns the durable lifecycle (never for gateway-viewer tabs), by durable
@@ -554,12 +461,16 @@ def _session_has_active_delegations(sid: str, session: dict | None = None) -> bo
     if session is None:
         with _sessions_lock:
             session = _sessions.get(sid)
-    own_sid, owned_session_key = _session_async_delegation_selectors(session, sid_hint=sid)
+    if not session:
+        return False
+    own_sid = _lifecycle_own_sid(session, sid)
+    session_key = str(session.get("session_key") or "")
+    session_id = getattr(session.get("agent"), "session_id", None) or session_key
+    owned_session_key = session_key if _session_owns_durable_lifecycle(session_id) else ""
     if not own_sid and not owned_session_key:
         return False
     try:
         from tools.async_delegation import has_live_for_session
-
         return has_live_for_session(session_key=owned_session_key, origin_ui_session_id=own_sid)
     except Exception:
         logger.debug("Failed to query active delegations for UI session %s", sid, exc_info=True)
@@ -592,8 +503,7 @@ def _ws_orphan_turn_activity_is_fresh(session: dict) -> bool:
     report NOT fresh, i.e. eligible for interrupt-at-grace."""
     if _WS_ORPHAN_ACTIVITY_STALE_S <= 0:
         return False
-    agent = session.get("agent")
-    summary_fn = getattr(agent, "get_activity_summary", None)
+    summary_fn = getattr(session.get("agent"), "get_activity_summary", None)
     if not callable(summary_fn):
         return False
     try:
@@ -614,9 +524,7 @@ def _schedule_ws_orphan_reap(sid: str, *, delay_s: float | None = None) -> None:
         # Serialize the re-check against session.resume (rebinds under _session_resume_lock).
         # Claim teardown by popping under both locks, then release the resume lock before
         # slow finalization. Ordering is always resume_lock -> sessions_lock (RLock).
-        reschedule_delay = None
-        interrupt_session = None
-        session = None
+        reschedule_delay = interrupt_session = session = None
         with _session_resume_lock:
             # Drop this Timer's registration so a concurrent _cancel_ws_orphan_reap can't
             # cancel a dead Timer while a rescheduled one (registered below) is the owner.
@@ -627,37 +535,32 @@ def _schedule_ws_orphan_reap(sid: str, *, delay_s: float | None = None) -> None:
                 return
             if _session_has_active_delegations(sid, current):
                 reschedule_delay = _WS_ORPHAN_REAP_GRACE_S
-            elif current.get("running"):
-                if not current.get("_client_gone_interrupt_requested") and (
-                    _ws_orphan_turn_activity_is_fresh(current)
-                ):
-                    # Client-absent but producing: keep running detached (the sentinel
-                    # buffers emits) and re-check each grace interval.
-                    logger.debug(
-                        "client_gone sid=%s action=defer (turn activity fresh; stale threshold %.0fs)",
-                        sid, _WS_ORPHAN_ACTIVITY_STALE_S,
-                    )
-                    reschedule_delay = _WS_ORPHAN_REAP_GRACE_S
-                else:
-                    # Mid-turn detached sessions must never drop the single Timer: interrupt
-                    # once after grace, then poll until turn-finalization settles.
-                    polls = int(current.get("_client_gone_interrupt_polls") or 0) + 1
-                    current["_client_gone_interrupt_polls"] = polls
-                    if polls > _WS_ORPHAN_INTERRUPT_REAP_MAX_POLLS:
-                        # Never settled inside the budget — force-reap rather than park forever.
-                        logger.error(
-                            "client_gone sid=%s: turn did not settle after %d "
-                            "interrupt polls (%.0fs) — force-reaping detached session",
-                            sid, polls - 1, (polls - 1) * _WS_ORPHAN_INTERRUPT_REAP_POLL_S,
-                        )
-                        session = _pop_session_by_id(sid)
-                    else:
-                        if not current.get("_client_gone_interrupt_requested"):
-                            current["_client_gone_interrupt_requested"] = True
-                            interrupt_session = current
-                        reschedule_delay = _WS_ORPHAN_INTERRUPT_REAP_POLL_S
-            else:
+            elif not current.get("running"):
                 session = _pop_session_by_id(sid)
+            elif not current.get("_client_gone_interrupt_requested") and _ws_orphan_turn_activity_is_fresh(current):
+                # Client-absent but producing: keep running detached (the sentinel
+                # buffers emits) and re-check each grace interval.
+                logger.debug(
+                    "client_gone sid=%s action=defer (turn activity fresh; stale threshold %.0fs)",
+                    sid, _WS_ORPHAN_ACTIVITY_STALE_S)
+                reschedule_delay = _WS_ORPHAN_REAP_GRACE_S
+            else:
+                # Mid-turn detached sessions must never drop the single Timer: interrupt
+                # once after grace, then poll until turn-finalization settles.
+                polls = int(current.get("_client_gone_interrupt_polls") or 0) + 1
+                current["_client_gone_interrupt_polls"] = polls
+                if polls > _WS_ORPHAN_INTERRUPT_REAP_MAX_POLLS:
+                    # Never settled inside the budget — force-reap rather than park forever.
+                    logger.error(
+                        "client_gone sid=%s: turn did not settle after %d "
+                        "interrupt polls (%.0fs) — force-reaping detached session",
+                        sid, polls - 1, (polls - 1) * _WS_ORPHAN_INTERRUPT_REAP_POLL_S)
+                    session = _pop_session_by_id(sid)
+                else:
+                    if not current.get("_client_gone_interrupt_requested"):
+                        current["_client_gone_interrupt_requested"] = True
+                        interrupt_session = current
+                    reschedule_delay = _WS_ORPHAN_INTERRUPT_REAP_POLL_S
 
         if interrupt_session is not None:
             try:
@@ -668,7 +571,6 @@ def _schedule_ws_orphan_reap(sid: str, *, delay_s: float | None = None) -> None:
                 with _sessions_lock:
                     if _sessions.get(sid) is interrupt_session:
                         interrupt_session.pop("_client_gone_interrupt_requested", None)
-
         if reschedule_delay is not None:
             _schedule_ws_orphan_reap(sid, delay_s=reschedule_delay)
             return
@@ -694,42 +596,38 @@ def _close_sessions_for_transport(transport, *, end_reason: str = "ws_disconnect
     Returns ``(reaped, detached)`` counts."""
     with _sessions_lock:
         owned = [(sid, s) for sid, s in _sessions.items() if s.get("transport") is transport]
-    reaped = 0
-    detached = 0
+    reaped = detached = 0
     for sid, session in owned:
         claimed_for_teardown = None
         should_schedule_reap = False
         # session.resume fast-path rebinds under _session_resume_lock: take it before
         # re-checking so a reconnect can't move the transport between check and claim.
-        with _session_resume_lock:
-            with _sessions_lock:
-                current = _sessions.get(sid)
-                if current is not session:
-                    continue
-                if current.get("transport") is not transport:
-                    # The reconnect owns this session now; drop only the old viewer registration.
-                    (current.get("viewers") or {}).pop(transport, None)
-                    continue
-                if current.get("close_on_disconnect"):
-                    claimed_for_teardown = _pop_session_by_id(sid)
+        with _session_resume_lock, _sessions_lock:
+            current = _sessions.get(sid)
+            if current is not session:
+                continue
+            if current.get("transport") is not transport:
+                # The reconnect owns this session now; drop only the old viewer registration.
+                (current.get("viewers") or {}).pop(transport, None)
+                continue
+            if current.get("close_on_disconnect"):
+                claimed_for_teardown = _pop_session_by_id(sid)
+            else:
+                # Point at the drop sentinel (NOT real stdio) so _ws_session_is_orphaned
+                # recognizes it; standalone `hermes --tui` keeps real _stdio. UNLESS
+                # another window (multi-window pop-out viewer) still shows the session:
+                # re-bind to the most recent surviving viewer instead of stranding it.
+                viewers = current.get("viewers") or {}
+                viewers.pop(transport, None)
+                live = [
+                    vt for vt, ts in sorted(viewers.items(), key=lambda kv: kv[1])
+                    if vt is not transport and not _transport_is_dead(vt)]
+                if live:
+                    current["transport"] = live[-1]
                 else:
-                    # Point at the drop sentinel (NOT real stdio) so _ws_session_is_orphaned
-                    # recognizes it; standalone `hermes --tui` keeps real _stdio. UNLESS
-                    # another window (multi-window pop-out viewer) still shows the session:
-                    # re-bind to the most recent surviving viewer instead of stranding it.
-                    viewers = current.get("viewers") or {}
-                    viewers.pop(transport, None)
-                    remaining = [
-                        (ts, vt) for vt, ts in viewers.items()
-                        if vt is not transport and not _transport_is_dead(vt)
-                    ]
-                    if remaining:
-                        remaining.sort(key=lambda item: item[0])
-                        current["transport"] = remaining[-1][1]
-                    else:
-                        current["transport"] = _detached_ws_transport
-                        current.pop("_client_gone_interrupt_requested", None)
-                        should_schedule_reap = True
+                    current["transport"] = _detached_ws_transport
+                    current.pop("_client_gone_interrupt_requested", None)
+                    should_schedule_reap = True
         if claimed_for_teardown is not None:
             if _teardown_popped_session(claimed_for_teardown, end_reason=end_reason):
                 reaped += 1
diff --git a/tui_gateway/session_reaper.py b/tui_gateway/session_reaper.py
index a2ebb036fc..9dc4744da7 100644
--- a/tui_gateway/session_reaper.py
+++ b/tui_gateway/session_reaper.py
@@ -1,7 +1,9 @@
-"""Session flush / reaping / orphan sweep / cross-backend heartbeat: exit-flush signal handlers, idle + LRU eviction, orphaned session-row sweep, backend heartbeat refresher.
+"""Session flush / reaping / orphan sweep / cross-backend heartbeat: exit-flush signal handlers,
+idle + LRU eviction, orphaned session-row sweep, backend heartbeat refresher.
 
-Bodies are rebound onto server.py's globals at install time (see
-method_ctx.bind_module), so they reference server.py globals bare.
+Bodies are rebound onto server.py's globals at install time (method_ctx.bind_module), so they
+reference server.py globals bare — including the knobs _SESSION_TTL_S, _REAPER_SCAN_S,
+_EXIT_FLUSH_BUDGET_S and _INCREMENTAL_FLUSH_INTERVAL_S.
 """
 
 from __future__ import annotations
@@ -16,21 +18,16 @@ from .method_ctx import HandlerRegistry, bind_module
 _registry = HandlerRegistry()
 
 
-# Knobs (_SESSION_TTL_S, _REAPER_SCAN_S, _EXIT_FLUSH_BUDGET_S, _INCREMENTAL_FLUSH_INTERVAL_S) live in server.py.
-
 # ── Flush-on-kill + periodic incremental flush ───────────────────────────
 # (a) SIGTERM/SIGINT run a bounded flush to state.db BEFORE normal shutdown, chained to the prior
 # handler; (b) the idle-reaper scan piggybacks an incremental flush so a SIGKILL loses at most one interval.
 
 
 def _flush_session_messages(session: dict | None) -> bool:
-    """Best-effort durable flush of one session's transcript via ``agent._persist_session``
-    (same marker-deduped contract as ``_finalize_session``): repeated calls only write
-    genuinely-unflushed messages, never duplicate rows."""
+    """Best-effort durable flush of one session's transcript via ``agent._persist_session`` (same
+    marker-deduped contract as ``_finalize_session``: repeated calls never duplicate rows)."""
     agent = session.get("agent") if session else None
-    if agent is None or not hasattr(agent, "_persist_session"):
-        return False
-    snapshot = getattr(agent, "_session_messages", None)
+    snapshot = getattr(agent, "_session_messages", None) if hasattr(agent, "_persist_session") else None
     if not snapshot:
         return False
     try:
@@ -47,14 +44,13 @@ def _reaper_session_snapshot() -> list:
 
 
 def _flush_dirty_sessions(now: float | None = None) -> int:
-    """Periodic incremental flush, driven by the idle-reaper scan. Skips ``running``
-    sessions: the turn thread owns mid-turn persistence and mutates the live message list,
-    so racing it from the reaper thread is never safe. Idle sessions flush at most once per
-    ``_INCREMENTAL_FLUSH_INTERVAL_S``; ``now`` (monotonic) is injectable for tests."""
+    """Periodic incremental flush, driven by the idle-reaper scan. Skips ``running`` sessions: the
+    turn thread owns mid-turn persistence and mutates the live message list, so racing it from the
+    reaper thread is never safe. Idle sessions flush at most once per ``_INCREMENTAL_FLUSH_INTERVAL_S``;
+    ``now`` (monotonic) is injectable for tests."""
     if _INCREMENTAL_FLUSH_INTERVAL_S <= 0:
         return 0
-    if now is None:
-        now = time.monotonic()
+    now = time.monotonic() if now is None else now
     flushed = 0
     for session in _reaper_session_snapshot():
         if not isinstance(session, dict) or session.get("running"):
@@ -62,17 +58,15 @@ def _flush_dirty_sessions(now: float | None = None) -> int:
         last = float(session.get("_last_incremental_flush") or 0.0)
         if last and (now - last) < _INCREMENTAL_FLUSH_INTERVAL_S:
             continue
-        if _flush_session_messages(session):
-            flushed += 1
+        flushed += _flush_session_messages(session)
         session["_last_incremental_flush"] = now
     return flushed
 
 
 def _flush_sessions_before_exit(budget_s: float | None = None) -> int:
-    """Bounded flush of ALL in-memory sessions on the way out. Runs on a daemon worker
-    joined with the budget so a hung SQLite write can't block exit past
-    ``HERMES_TUI_EXIT_FLUSH_BUDGET_S`` (default 5s). Running sessions are included — the
-    process is dying, so a partial transcript beats guaranteed loss."""
+    """Bounded flush of ALL in-memory sessions on the way out, on a daemon worker joined with the
+    budget so a hung SQLite write can't block exit past ``HERMES_TUI_EXIT_FLUSH_BUDGET_S`` (default
+    5s). Running sessions are included — the process is dying, a partial transcript beats loss."""
     budget = _EXIT_FLUSH_BUDGET_S if budget_s is None else max(0.0, budget_s)
     if budget <= 0:
         return 0
@@ -83,8 +77,7 @@ def _flush_sessions_before_exit(budget_s: float | None = None) -> int:
         for session in _reaper_session_snapshot():
             if time.monotonic() >= deadline:
                 break
-            if _flush_session_messages(session):
-                result["flushed"] += 1
+            result["flushed"] += _flush_session_messages(session)
 
     worker = threading.Thread(target=_run, daemon=True, name="hermes-exit-flush")
     worker.start()
@@ -97,49 +90,42 @@ _exit_flush_handlers_installed = False
 
 
 def _handle_exit_flush_signal(signum, frame) -> None:
-    """Flush in-memory sessions, then hand off to the prior handler (uvicorn's graceful
-    shutdown, a supervisor's handler, or the default disposition) — this only *prepends*
-    a bounded durable flush."""
+    """Flush in-memory sessions, then hand off to the prior handler (uvicorn's graceful shutdown, a
+    supervisor's handler, or the default disposition) — this only *prepends* a bounded flush."""
     with contextlib.suppress(Exception):
         _flush_sessions_before_exit()
     import signal as _signal
-
     prev = _exit_flush_prev_handlers.get(signum)
     if callable(prev):
         prev(signum, frame)
-        return
-    if prev is _signal.SIG_IGN:
-        return
-    # Default disposition: restore it and re-raise so the process dies with the correct
-    # signal (exit status visible to supervisors).
-    try:
-        _signal.signal(signum, _signal.SIG_DFL)
-        os.kill(os.getpid(), signum)
-    except Exception:
-        raise SystemExit(128 + int(signum)) from None
+    elif prev is not _signal.SIG_IGN:
+        # Default disposition: restore it and re-raise so the process dies with the correct
+        # signal (exit status visible to supervisors).
+        try:
+            _signal.signal(signum, _signal.SIG_DFL)
+            os.kill(os.getpid(), signum)
+        except Exception:
+            raise SystemExit(128 + int(signum)) from None
 
 
 def install_exit_flush_signal_handlers() -> bool:
-    """Install chaining SIGTERM/SIGINT flush handlers (main thread only). Called before
-    uvicorn takes over signals: its ``capture_signals()`` saves these as the "original"
-    handlers and re-raises into them after graceful shutdown, so the flush also covers
-    terminations outside uvicorn's serve window. Idempotent; False off-main-thread/on failure."""
+    """Install chaining SIGTERM/SIGINT flush handlers (main thread only). Called before uvicorn takes
+    over signals: its ``capture_signals()`` saves these as the "original" handlers and re-raises into
+    them after graceful shutdown, so the flush also covers terminations outside uvicorn's serve
+    window. Idempotent; False off-main-thread/on failure."""
     global _exit_flush_handlers_installed
     if _exit_flush_handlers_installed:
         return True
     if threading.current_thread() is not threading.main_thread():
         return False
     import signal as _signal
-
     installed = False
     for signum in (_signal.SIGTERM, _signal.SIGINT):
-        try:
+        with contextlib.suppress(ValueError, OSError, RuntimeError):
             prev = _signal.getsignal(signum)
             _signal.signal(signum, _handle_exit_flush_signal)
             _exit_flush_prev_handlers[signum] = prev
             installed = True
-        except (ValueError, OSError, RuntimeError):
-            continue
     _exit_flush_handlers_installed = installed
     return installed
 
@@ -151,9 +137,9 @@ def _transport_is_dead(transport) -> bool:
 
 
 def _reaper_session_is_detached_idle(sid: str, session: dict) -> bool:
-    """Shared hard exemptions for both reapers: never evict a session mid-turn, awaiting
-    input, still building, owning live delegated work, or on a live transport. Lazy watch
-    sessions never start a build, so their unset agent_ready must not make them immortal."""
+    """Shared hard exemptions for both reapers: never evict a session mid-turn, awaiting input, still
+    building, owning live delegated work, or on a live transport. Lazy watch sessions never start a
+    build, so their unset agent_ready must not make them immortal."""
     if session.get("running") or _session_pending_kind(sid) or _session_has_active_delegations(sid, session):
         return False
     ready = session.get("agent_ready")
@@ -180,20 +166,16 @@ def _reap_idle_sessions() -> None:
         victims = [sid for sid, s in _sessions.items() if _session_is_evictable(sid, s, now)]
     for sid in victims:
         _close_session_by_id(
-            sid,
-            end_reason="idle_timeout",
-            predicate=lambda session, vs=sid: _session_is_evictable(vs, session, time.time()),
-        )
+            sid, end_reason="idle_timeout",
+            predicate=lambda session, vs=sid: _session_is_evictable(vs, session, time.time()))
     _enforce_session_cap()
     _reclaim_orphaned_leases()
     # Long-lived processes: gen2 GC rarely runs at steady state and glibc retains freed
     # pages as RSS, so trim every scan to prevent unbounded RSS growth over days/weeks.
     try:
         from hermes_cli.mem_trim import trim_memory
-
         trim_memory(reason="idle reaper periodic trim")
-    except Exception as exc:
-        # debug, not warning — a persistent failure would repeat every scan.
+    except Exception as exc:  # debug, not warning — a persistent failure would repeat every scan.
         logger.debug("idle reaper memory trim failed: %s: %s", type(exc).__name__, exc)
 
 
@@ -201,33 +183,27 @@ def _reclaim_orphaned_leases() -> None:
     """Hand the registry the lease ids we still own so it can drop the rest."""
     try:
         from hermes_cli.active_sessions import release_orphaned_leases
-
         with _sessions_lock:
             live = {
-                lease.lease_id
-                for session in _sessions.values()
-                if (lease := session.get("active_session_lease")) is not None
-            }
+                lease.lease_id for session in _sessions.values()
+                if (lease := session.get("active_session_lease")) is not None}
         if dropped := release_orphaned_leases(live):
             logger.info("Reclaimed %d orphaned active-session lease(s)", dropped)
     except Exception:
         logger.debug("orphaned lease reclaim failed", exc_info=True)
 
 
-# Soft LRU cap on in-memory sessions: the TTL reaper only frees sessions idle for hours,
-# so a heavy reconnecting user accumulates resident detached agents. The cap evicts the
-# least-recently-active DETACHED sessions sooner — never a running / pending / mid-build /
-# live-transport one (reopening re-resumes from the DB). 0/null disables.
+# Soft LRU cap on in-memory sessions: the TTL reaper only frees sessions idle for hours, so a heavy
+# reconnecting user accumulates resident detached agents. The cap evicts the least-recently-active
+# DETACHED sessions sooner — never a running / pending / mid-build / live-transport one (reopening
+# re-resumes from the DB). 0/null disables.
 def _max_live_sessions() -> int:
     try:
         from hermes_cli.active_sessions import coerce_max_concurrent_sessions
-
         cfg = _load_cfg() or {}
         raw = cfg.get("max_live_sessions")
-        if raw is None:
-            gateway_cfg = cfg.get("gateway")
-            if isinstance(gateway_cfg, dict):
-                raw = gateway_cfg.get("max_live_sessions")
+        if raw is None and isinstance(gateway_cfg := cfg.get("gateway"), dict):
+            raw = gateway_cfg.get("max_live_sessions")
         coerced = coerce_max_concurrent_sessions(raw, key="max_live_sessions")
         return int(coerced) if coerced else 0
     except Exception:
@@ -254,28 +230,26 @@ def _enforce_session_cap() -> None:
             if len(_sessions) <= cap:
                 break
         _close_session_by_id(
-            sid,
-            end_reason="lru_evict",
-            predicate=lambda session, vs=sid: _session_is_lru_evictable(vs, session),
-        )
+            sid, end_reason="lru_evict", predicate=lambda session, vs=sid: _session_is_lru_evictable(vs, session))
 
 
-def _reaper_daemon_timer(delay: float, fn) -> None:
-    timer = threading.Timer(delay, fn)
+def _reaper_daemon_timer(delay: float, fn, fail_log: str, level: str = "debug") -> None:
+    """Run ``fn`` once on a daemon Timer after ``delay``; log (never raise) on failure."""
+
+    def _run() -> None:
+        try:
+            fn()
+        except Exception:
+            getattr(logger, level)(fail_log, exc_info=True)
+
+    timer = threading.Timer(delay, _run)
     timer.daemon = True
     timer.start()
 
 
 def _schedule_session_cap_enforcement() -> None:
     """Run the LRU sweep off the response path (eviction can call agent.close)."""
-
-    def _run():
-        try:
-            _enforce_session_cap()
-        except Exception:
-            logger.debug("session cap enforcement failed", exc_info=True)
-
-    _reaper_daemon_timer(0.1, _run)
+    _reaper_daemon_timer(0.1, _enforce_session_cap, "session cap enforcement failed")
 
 
 # ── Startup sweep for orphaned session rows ──────────────────────────────
@@ -289,15 +263,15 @@ _startup_orphan_sweep_lock = threading.Lock()
 
 
 def _session_orphan_reaper_enabled() -> bool:
-    """``dashboard.startup_orphan_sweep`` (default on). Fail-open on errors and
-    on a missing key (raw yaml, no DEFAULT_CONFIG merge on this loader)."""
+    """``dashboard.startup_orphan_sweep`` (default on). Fail-open on errors and on a missing key
+    (raw yaml, no DEFAULT_CONFIG merge on this loader)."""
     try:
         dashboard_cfg = (_load_cfg() or {}).get("dashboard") or {}
         if isinstance(dashboard_cfg, dict) and "startup_orphan_sweep" in dashboard_cfg:
             return is_truthy_value(dashboard_cfg.get("startup_orphan_sweep"), default=True)
-        return True
     except Exception:
-        return True
+        pass
+    return True
 
 
 def _live_session_ids() -> list[str]:
@@ -305,38 +279,30 @@ def _live_session_ids() -> list[str]:
     ids: set[str] = set()
     with _sessions_lock:
         for sid, session in _sessions.items():
-            if sid:
-                ids.add(str(sid))
-            if not isinstance(session, dict):
-                continue
-            for candidate in (getattr(session.get("agent"), "session_id", None), session.get("session_key")):
-                if candidate:
-                    ids.add(str(candidate))
+            candidates = [sid]
+            if isinstance(session, dict):
+                candidates += [getattr(session.get("agent"), "session_id", None), session.get("session_key")]
+            ids.update(str(c) for c in candidates if c)
     return sorted(ids)
 
 
 def _sweep_orphaned_session_rows() -> list[str]:
     """End orphaned tui/desktop/subagent rows left by a dead process.
 
-    "Provably orphaned" is inferred conservatively: the row must have been created AND last
-    messaged at least the session TTL ago (a fresh row that copied an old transcript is
-    protected by its own ``started_at``). Rows held in memory (e.g. a ``session.resume`` in
-    the startup grace window) are excluded. Cross-backend: the sweep refuses to close a row
-    any live backend (heartbeat within ``2 * TTL``) could own — see
-    ``SessionDB.sweep_orphaned_sessions``.
-    """
+    "Provably orphaned" is inferred conservatively: the row must have been created AND last messaged
+    at least the session TTL ago (a fresh row that copied an old transcript is protected by its own
+    ``started_at``). Rows held in memory (e.g. a ``session.resume`` in the startup grace window) are
+    excluded. Cross-backend: the sweep refuses to close a row any live backend (heartbeat within
+    ``2 * TTL``) could own — see ``SessionDB.sweep_orphaned_sessions``."""
     db = _get_db()
-    ttl = _SESSION_TTL_S
-    if db is None or ttl <= 0:
+    if db is None or _SESSION_TTL_S <= 0:
         return []
     swept = db.sweep_orphaned_sessions(
-        max_idle_seconds=ttl, sources=_ORPHAN_SWEEP_SOURCES, exclude_ids=tuple(_live_session_ids())
-    )
+        max_idle_seconds=_SESSION_TTL_S, sources=_ORPHAN_SWEEP_SOURCES, exclude_ids=tuple(_live_session_ids()))
     if swept:
         logger.info(
             "Closed %d orphaned session row(s) from a previous gateway process (startup_orphan_reap): %s",
-            len(swept), ", ".join(swept),
-        )
+            len(swept), ", ".join(swept))
     return swept
 
 
@@ -356,18 +322,21 @@ def _reaper_hostname() -> str:
     return os.uname().nodename if hasattr(os, "uname") else "host"
 
 
-def _backend_id_for_this_process() -> str:
-    """Stable identity for this process's heartbeat row: pid (readability) AND a startup
-    nonce so a PID-reuse respawn cannot inherit the dead predecessor's heartbeat."""
-    nonce = getattr(_backend_id_for_this_process, "_nonce", None)
-    if nonce is None:
-        import secrets as _secrets
+def _reaper_memo(fn, attr: str, compute):
+    """Compute-once value memoised as an attribute on ``fn`` itself."""
+    value = getattr(fn, attr, None)
+    if value is None:
+        value = compute()
+        with contextlib.suppress(AttributeError):  # pragma: no cover - defensive
+            setattr(fn, attr, value)
+    return value
 
-        nonce = _secrets.token_hex(4)
-        try:
-            setattr(_backend_id_for_this_process, "_nonce", nonce)
-        except AttributeError:  # pragma: no cover - defensive
-            pass
+
+def _backend_id_for_this_process() -> str:
+    """Stable identity for this process's heartbeat row: pid (readability) AND a startup nonce so a
+    PID-reuse respawn cannot inherit the dead predecessor's heartbeat."""
+    import secrets as _secrets
+    nonce = _reaper_memo(_backend_id_for_this_process, "_nonce", lambda: _secrets.token_hex(4))
     return f"{_current_profile_name()}@{_reaper_hostname()}:{os.getpid()}:{nonce}"
 
 
@@ -378,27 +347,16 @@ def _refresh_backend_heartbeat() -> None:
         return
     try:
         db.register_backend_heartbeat(
-            backend_id=_backend_id_for_this_process(),
-            pid=os.getpid(),
-            started_at=_gateway_started_at(),
-            profile=_current_profile_name(),
-            host=_reaper_hostname(),
-        )
+            backend_id=_backend_id_for_this_process(), pid=os.getpid(), started_at=_gateway_started_at(),
+            profile=_current_profile_name(), host=_reaper_hostname())
     except Exception:
         logger.debug("backend heartbeat refresh failed", exc_info=True)
 
 
 def _gateway_started_at() -> float:
-    """Wall-clock time this process started (first-call time is a good-enough
-    proxy: the heartbeat refresher runs after the gateway is fully wired up)."""
-    started = getattr(_gateway_started_at, "_t", None)
-    if started is None:
-        started = time.time()
-        try:
-            setattr(_gateway_started_at, "_t", started)
-        except AttributeError:  # pragma: no cover
-            pass
-    return started
+    """Wall-clock time this process started (first-call time is a good-enough proxy: the heartbeat
+    refresher runs after the gateway is fully wired up)."""
+    return _reaper_memo(_gateway_started_at, "_t", time.time)
 
 
 def _heartbeat_refresher_loop(stop_event: threading.Event) -> None:
@@ -412,9 +370,9 @@ def _heartbeat_refresher_loop(stop_event: threading.Event) -> None:
 
 
 def _start_backend_heartbeat_refresher() -> None:
-    """Register this backend and start the refresher thread (once per process). The first
-    refresh writes the row synchronously so this process's own sweep sees itself in the
-    heartbeat table. ``_HEARTBEAT_REFRESH_S <= 0`` means "register once, never refresh"."""
+    """Register this backend and start the refresher thread (once per process). The first refresh
+    writes the row synchronously so this process's own sweep sees itself in the heartbeat table.
+    ``_HEARTBEAT_REFRESH_S <= 0`` means "register once, never refresh"."""
     global _heartbeat_refresher_started
     with _heartbeat_refresher_lock:
         if _heartbeat_refresher_started:
@@ -431,8 +389,7 @@ def _start_backend_heartbeat_refresher() -> None:
     def _atexit_clear():
         stop_event.set()
         with contextlib.suppress(Exception):
-            db = _get_db()
-            if db is not None:
+            if (db := _get_db()) is not None:
                 db.clear_backend_heartbeat(_backend_id_for_this_process())
 
     atexit.register(_atexit_clear)
@@ -442,27 +399,18 @@ def _start_backend_heartbeat_refresher() -> None:
 
 
 def _schedule_startup_orphan_sweep() -> None:
-    """Schedule the once-per-process startup orphan sweep, delayed by the WS-orphan grace
-    window so a client reconnecting right after a restart can ``session.resume`` its row
-    first. Grace 0 (park forever), TTL 0 and ``dashboard.startup_orphan_sweep: false``
-    all suppress the sweep."""
+    """Schedule the once-per-process startup orphan sweep, delayed by the WS-orphan grace window so
+    a client reconnecting right after a restart can ``session.resume`` its row first. Grace 0 (park
+    forever), TTL 0 and ``dashboard.startup_orphan_sweep: false`` all suppress the sweep."""
     global _startup_orphan_sweep_ran
     if _WS_ORPHAN_REAP_GRACE_S <= 0 or _SESSION_TTL_S <= 0 or not _session_orphan_reaper_enabled():
         return
-    if _startup_orphan_sweep_ran:
-        return
     with _startup_orphan_sweep_lock:
         if _startup_orphan_sweep_ran:
             return
         _startup_orphan_sweep_ran = True
-
-    def _run() -> None:
-        try:
-            _sweep_orphaned_session_rows()
-        except Exception:
-            logger.warning("startup orphan session sweep failed", exc_info=True)
-
-    _reaper_daemon_timer(_WS_ORPHAN_REAP_GRACE_S, _run)
+    _reaper_daemon_timer(
+        _WS_ORPHAN_REAP_GRACE_S, _sweep_orphaned_session_rows, "startup orphan session sweep failed", level="warning")
 
 
 def register(server) -> None:
diff --git a/tui_gateway/session_workdir.py b/tui_gateway/session_workdir.py
index 4bf66d4651..77272846f8 100644
--- a/tui_gateway/session_workdir.py
+++ b/tui_gateway/session_workdir.py
@@ -1,7 +1,8 @@
-"""Session working-directory + durable session row: cwd resolution/healing, session.db row ensure, branch seed, history rewind, git meta persistence.
+"""Session working-directory + durable session row: cwd resolution/healing, session.db row
+ensure, branch seed, history rewind, git meta persistence.
 
-Bodies are rebound onto server.py's globals at install time (see
-method_ctx.bind_module), so they reference server.py globals bare.
+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
@@ -35,40 +36,34 @@ def _completion_cwd(params: dict | None = None) -> str:
         # bridged TERMINAL_CWD, so a configured terminal.cwd is read directly.
         or _launch_configured_cwd()
         or os.environ.get("TERMINAL_CWD")
-        or os.getcwd()
-    )
-    try:
+        or os.getcwd())
+    with contextlib.suppress(Exception):
         resolved = os.path.abspath(os.path.expanduser(str(raw)))
         if os.path.isdir(resolved):
             return resolved
-    except Exception:
-        pass
     return os.getcwd()
 
 
 def _workdir_terminal_cfg(key: str) -> str:
     """Stripped ``terminal.<key>`` from config, or "" when unset/unreadable."""
-    try:
+    with contextlib.suppress(Exception):
         terminal_cfg = _load_cfg().get("terminal", {})
         if isinstance(terminal_cfg, dict):
             return str(terminal_cfg.get(key) or "").strip()
-    except Exception:
-        pass
     return ""
 
 
 def _terminal_task_cwd(session: dict | None) -> str:
-    """The cwd terminal_tool should use for this TUI session. Unlike
-    ``_completion_cwd`` it is NOT validated on the host: a non-local backend's
-    cwd lives inside the target environment."""
+    """The cwd terminal_tool should use for this TUI session (NOT host-validated, unlike
+    ``_completion_cwd``: a non-local backend's cwd lives inside the target environment)."""
     return _terminal_task_cwd_with_source(session)[0]
 
 
 def _terminal_task_cwd_with_source(session: dict | None) -> tuple[str, str]:
-    """Like :func:`_terminal_task_cwd` but returns ``(cwd, source)``: ``"session"`` for THIS
-    session's workspace (``explicit_cwd``/tracked dir), ``"process"`` for the process-global
-    ``TERMINAL_CWD`` / ``terminal.cwd`` fallback — under per-session docker isolation that is a
-    launch artifact of a PREVIOUS session, so terminal_tool refuses it as a bind-mount source."""
+    """``(cwd, source)``: ``"session"`` for THIS session's workspace (``explicit_cwd``/tracked
+    dir), ``"process"`` for the process-global ``TERMINAL_CWD`` / ``terminal.cwd`` fallback —
+    under per-session docker isolation that is a launch artifact of a PREVIOUS session, so
+    terminal_tool refuses it as a bind-mount source."""
     backend = _effective_terminal_backend()
     if backend != "local":
         # THIS session's explicit workspace beats the LAST session's env var.
@@ -79,7 +74,6 @@ def _terminal_task_cwd_with_source(session: dict | None) -> tuple[str, str]:
             return raw, "process"
         if backend == "ssh":
             return "~", "process"
-
     if session and session.get("cwd"):
         return str(session["cwd"]), "session"
     return _completion_cwd(), "process"
@@ -94,9 +88,7 @@ _resolve_cwd_git = git_probe.resolve
 
 
 def _session_cwd(session: dict | None) -> str:
-    if session and session.get("cwd"):
-        return str(session["cwd"])
-    return _completion_cwd()
+    return str(session["cwd"]) if session and session.get("cwd") else _completion_cwd()
 
 
 # Sources whose launch directory is an artifact of how the app was started, not
@@ -106,11 +98,7 @@ _LAUNCH_CWD_NOT_A_WORKSPACE = {"desktop"}
 
 def _context_cwd_is_launch_artifact(session: dict | None) -> bool:
     """Whether the session cwd came from app launch rather than user intent."""
-    return bool(
-        session
-        and not session.get("explicit_cwd")
-        and _session_source(session) in _LAUNCH_CWD_NOT_A_WORKSPACE
-    )
+    return bool(session and not session.get("explicit_cwd") and _session_source(session) in _LAUNCH_CWD_NOT_A_WORKSPACE)
 
 
 def _persisted_session_cwd(session: dict) -> str | None:
@@ -132,7 +120,6 @@ def _heal_dead_cwd(cwd: str) -> str:
     raw = (cwd or "").strip()
     if not raw or os.path.isdir(raw):
         return raw
-
     probe = raw
     for _ in range(64):
         parent = os.path.dirname(probe)
@@ -141,16 +128,11 @@ def _heal_dead_cwd(cwd: str) -> str:
         probe = parent
         if os.path.isdir(probe):
             break
-
     if not os.path.isdir(probe):
         return raw
-
-    try:
-        root = _git_common_repo_root_for_cwd(probe) or _git_repo_root_for_cwd(probe)
-    except Exception:
-        root = ""
-
-    return root or probe
+    with contextlib.suppress(Exception):
+        return _git_common_repo_root_for_cwd(probe) or _git_repo_root_for_cwd(probe) or probe
+    return probe
 
 
 def _is_local_terminal_backend() -> bool:
@@ -159,10 +141,9 @@ def _is_local_terminal_backend() -> bool:
 
 
 def _effective_terminal_backend() -> str:
-    """Active terminal backend name (``local``, ``docker``, ``ssh``, ...):
-    ``TERMINAL_ENV`` when set (launchers bridge ``terminal.backend`` into env),
-    else the ``terminal.backend`` config key (desktop/TUI in-process gateways
-    skip that bridge)."""
+    """Active terminal backend name (``local``, ``docker``, ``ssh``, ...): ``TERMINAL_ENV`` when
+    set (launchers bridge ``terminal.backend`` into env), else the ``terminal.backend`` config
+    key (desktop/TUI in-process gateways skip that bridge)."""
     backend = (os.environ.get("TERMINAL_ENV") or "").strip().lower()
     if not backend or backend == "local":
         cfg_backend = _workdir_terminal_cfg("backend").lower()
@@ -177,12 +158,10 @@ def _display_session_cwd(session: dict | None) -> str:
     cwd = _session_cwd(session)
     if not _is_local_terminal_backend():
         return cwd
-
     healed = _heal_dead_cwd(cwd)
     if healed and healed != cwd and session is not None:
         session["cwd"] = healed
         _persist_session_cwd_and_schedule_git_meta(session, healed)
-
     return healed
 
 
@@ -196,27 +175,21 @@ def _reconcile_session_cwd_from_terminal(session: dict | None) -> bool:
     overridden. Local backends only (a remote cwd cannot be stat'ed or git-probed here)."""
     if not session or not _is_local_terminal_backend():
         return False
-
     # An explicit choice only moves by another explicit action; a cwd adopted
     # HERE is marked `cwd_from_settle` so successive settles keep following.
     if session.get("explicit_cwd") and not session.get("cwd_from_settle"):
         return False
-
     try:
         from tools.terminal_tool import get_session_cwd
-
         recorded = get_session_cwd(session.get("session_key") or "")
     except Exception:
         return False
-
     if not recorded:
         return False
-
     resolved = os.path.abspath(os.path.expanduser(str(recorded)))
     current = os.path.abspath(os.path.expanduser(_session_cwd(session)))
     if resolved == current or not os.path.isdir(resolved):
         return False
-
     # Worktree ROOTS (folding to the common root would hide the move), both in a git
     # tree, different from each other, sharing the SAME common .git dir.
     landed = _git_repo_root_for_cwd(resolved)
@@ -224,17 +197,14 @@ def _reconcile_session_cwd_from_terminal(session: dict | None) -> bool:
     if not landed or not current_root or landed == current_root:
         return False
     landed_common = _git_common_repo_root_for_cwd(resolved)
-    current_common = _git_common_repo_root_for_cwd(current)
-    if not landed_common or landed_common != current_common:
+    if not landed_common or landed_common != _git_common_repo_root_for_cwd(current):
         return False
-
     session["cwd"] = resolved
     # This is the session's workspace now (a desktop launch-artifact cwd earns
     # a real row); the settle marker keeps it overridable by the NEXT settle.
     session["explicit_cwd"] = True
     session["cwd_from_settle"] = True
     _register_session_cwd(session)
-
     _persist_session_cwd_and_schedule_git_meta(session, resolved)
     return True
 
@@ -250,23 +220,17 @@ def _emit_settled_session_info(sid: str, session: dict, agent) -> None:
 
 
 def _session_source(session: dict | None) -> str:
-    if session:
-        source = str(session.get("source") or "").strip()
-        if source:
-            return source
-    return _resolve_session_platform()
+    source = str(session.get("source") or "").strip() if session else ""
+    return source or _resolve_session_platform()
 
 
 def _register_session_cwd(session: dict | None) -> None:
     if not session:
         return
-    try:
+    with contextlib.suppress(Exception):
         from tools.terminal_tool import register_task_env_overrides
-
         cwd, cwd_source = _terminal_task_cwd_with_source(session)
         register_task_env_overrides(session["session_key"], {"cwd": cwd, "cwd_source": cwd_source})
-    except Exception:
-        pass
 
 
 def _workdir_row_model_config(session: dict) -> tuple[str, dict]:
@@ -279,32 +243,25 @@ def _workdir_row_model_config(session: dict) -> tuple[str, dict]:
     override = session.get("model_override")
     override = override if isinstance(override, dict) else {}
     row_model = str(override.get("model") or "").strip() or _resolve_model()
-    model_config: dict = {}
-    for cfg_key in ("model", "provider", "base_url", "api_mode"):
-        if val := override.get(cfg_key):
-            model_config[cfg_key] = str(val)
+    model_config: dict = {k: str(v) for k in ("model", "provider", "base_url", "api_mode") if (v := override.get(k))}
     # A RESOLVED provider "custom" (named ``providers:``/``custom_providers:`` entry) persisted
     # bare here is the origin of "No LLM provider configured" rows (resume routes to OpenRouter
     # with no key). Recover the durable ``custom:<name>`` identity (matches _runtime_model_config).
     if str(model_config.get("provider") or "").strip().lower() == "custom":
         try:
             from hermes_cli.runtime_provider import canonical_custom_identity
-
             healed = canonical_custom_identity(
-                base_url=model_config.get("base_url") or None,
-                model=model_config.get("model") or row_model or None,
-            )
+                base_url=model_config.get("base_url") or None, model=model_config.get("model") or row_model or None)
             if healed:
                 model_config["provider"] = healed
         except Exception:
             logger.debug("custom provider identity recovery failed (db row)", exc_info=True)
     if (reasoning := session.get("create_reasoning_override")) is not None:
         model_config["reasoning_config"] = reasoning
-    create_service_tier_override = session.get("create_service_tier_override")
-    if create_service_tier_override is not None:
+    if (service_tier := session.get("create_service_tier_override")) is not None:
         # "" is the in-memory sentinel for an explicit normal tier (bypasses _make_agent's profile
         # fallback); persist a durable marker so resume can tell it from an inherited tier.
-        model_config["service_tier"] = create_service_tier_override or "normal"
+        model_config["service_tier"] = service_tier or "normal"
     # Same ``_branched_from`` marker the TUI /branch uses (list_sessions_rich + sidebar nesting).
     if parent_session_id := session.get("parent_session_id"):
         model_config["_branched_from"] = parent_session_id
@@ -351,8 +308,7 @@ def _ensure_session_db_row(session: dict) -> bool:
                 cwd=_persisted_session_cwd(session),
                 # Self-describing rows: aggregators merging several profile DBs can't rely on
                 # which file a row came from; a NULL is only repaired by the one-shot backfill.
-                profile_name=Path(profile_home).name if profile_home else _current_profile_name(),
-            )
+                profile_name=Path(profile_home).name if profile_home else _current_profile_name())
             # Born hidden (session.create hidden=true, or set_hidden before the
             # row existed): apply the deferred intent now, like pending_title.
             if session.get("pending_hidden"):
@@ -370,7 +326,6 @@ def _ensure_session_db_row(session: dict) -> bool:
 def _workdir_reraise_disk_full(exc: BaseException, log_msg: str) -> None:
     """Re-raise a disk-full write error (the caller must surface it); debug-log the rest."""
     from hermes_state import is_disk_full_error
-
     if is_disk_full_error(exc):
         raise exc
     logger.debug(log_msg, exc_info=True)
@@ -408,12 +363,8 @@ def _persist_branch_seed(session: dict) -> None:
             # mid-copy failure leaves a partial seed with _branch_seed_persisted unset.
             db.append_messages_batch(
                 key,
-                [
-                    {"role": msg.get("role", "user"), **{f: msg.get(f) for f in _WORKDIR_SEED_FIELDS}}
-                    for msg in seed
-                ],
-                chunk_rows=500,
-            )
+                [{"role": msg.get("role", "user"), **{f: msg.get(f) for f in _WORKDIR_SEED_FIELDS}} for msg in seed],
+                chunk_rows=500)
             session["_branch_seed_persisted"] = True
         except Exception as exc:
             _workdir_reraise_disk_full(exc, "branch seed persist failed")
@@ -470,24 +421,18 @@ def _rewind_active_session_history(
         history_before_user_originated_turn,
         retryable_user_text,
         split_user_originated_turn,
-        user_originated_turn_view,
-    )
+        user_originated_turn_view)
     from agent.memory_manager import sanitize_context
     from agent.tool_dispatch_helpers import _is_multimodal_tool_result, _multimodal_text_summary
-
     def _comparison_content(message: dict) -> Any:
         content = message.get("content")
         if _is_multimodal_tool_result(content):
             content = _multimodal_text_summary(content)
         elif isinstance(content, list):
-            text_parts = []
-            for part in content:
-                if not isinstance(part, dict):
-                    continue
-                if part.get("type") == "text":
-                    text_parts.append(str(part.get("text", "")))
-                elif part.get("type") in {"image", "image_url", "input_image"}:
-                    text_parts.append("[screenshot]")
+            text_parts = [
+                str(part.get("text", "")) if part.get("type") == "text" else "[screenshot]"
+                for part in content
+                if isinstance(part, dict) and part.get("type") in {"text", "image", "image_url", "input_image"}]
             content = "\n".join(text_parts) if text_parts else None
         if message.get("role") in {"user", "assistant"} and isinstance(content, str):
             return sanitize_context(content).strip()
@@ -517,9 +462,7 @@ def _rewind_active_session_history(
                 raise RuntimeError("session history changed before the rewind could be persisted")
             durable_target_index = durable_user_indices[user_ordinal]
             durable_target = durable[durable_target_index]
-            durable_prefix, durable_live_view = history_before_user_originated_turn(
-                durable, durable_target_index
-            )
+            durable_prefix, durable_live_view = history_before_user_originated_turn(durable, durable_target_index)
             if _comparison_content(durable_live_view) != _comparison_content(live_view):
                 raise RuntimeError("session history changed before the rewind could be persisted")
             target_row_id = durable_target.get("_row_id")
@@ -529,12 +472,10 @@ def _rewind_active_session_history(
                 retryable_user_text(durable_live_view.get("content"))
             scaffold, _ = split_user_originated_turn(durable_target)
             result = db.rewind_to_message(
-                session_key,
-                target_row_id,
+                session_key, target_row_id,
                 preserve_compaction_handoff=scaffold is not None,
                 expected_active_ids=expected_active_ids,
-                expected_target_content=durable_live_view.get("content"),
-            )
+                expected_target_content=durable_live_view.get("content"))
             if scaffold is not None:
                 replacement_id = result.get("replacement_message_id")
                 if not isinstance(replacement_id, int):
@@ -548,11 +489,9 @@ def _rewind_active_session_history(
                 warm.get("role") == durable_message.get("role")
                 and bool(warm.get("display_kind")) == bool(durable_message.get("display_kind"))
                 and _comparison_content(warm) == _comparison_content(durable_message)
-                for warm, durable_message in zip(installed, durable_prefix)
-            ):
+                for warm, durable_message in zip(installed, durable_prefix)):
                 for warm, durable_message in zip(installed, durable_prefix):
-                    row_id = durable_message.get("_row_id")
-                    if isinstance(row_id, int):
+                    if isinstance(row_id := durable_message.get("_row_id"), int):
                         warm["_row_id"] = row_id
             live_view = durable_live_view
             rewound_count = int(result.get("rewound_count", 0))
@@ -576,7 +515,6 @@ def _rewind_active_session_history(
 def _history_without_ephemeral_scaffolding(history: list[dict]) -> list[dict]:
     """Return the durable transcript shape without transient recovery rows."""
     from run_agent import _is_ephemeral_scaffolding
-
     return [message.copy() for message in history if not _is_ephemeral_scaffolding(message)]
 
 
@@ -614,15 +552,13 @@ def _persist_session_git_meta(session: dict, cwd: str, generation: int) -> None:
 def _persist_session_cwd_and_schedule_git_meta(session: dict, cwd: str, *, db=None) -> int | None:
     """Claim a DB-backed probe generation, then start Git enrichment."""
     try:
-        owner = contextlib.nullcontext(db) if db is not None else _session_db(session)
-        with owner as owner_db:
+        with (contextlib.nullcontext(db) if db is not None else _session_db(session)) as owner_db:
             if owner_db is None:
                 return None
             generation = owner_db.update_session_cwd(session.get("session_key", ""), cwd)
     except Exception:
         logger.debug("failed to persist session cwd", exc_info=True)
         return None
-
     if not _workdir_valid_generation(generation):
         return None
     _persist_session_git_meta(session, cwd, generation)
@@ -631,7 +567,6 @@ def _persist_session_cwd_and_schedule_git_meta(session: dict, cwd: str, *, db=No
 
 def _set_session_cwd(session: dict, cwd: str) -> str:
     from hermes_constants import translate_cwd_for_wsl_backend
-
     cwd = translate_cwd_for_wsl_backend(str(cwd))
     resolved = os.path.abspath(os.path.expanduser(cwd))
     if not os.path.isdir(resolved):
@@ -645,12 +580,9 @@ def _set_session_cwd(session: dict, cwd: str) -> str:
     # The synchronous DB write claims ordering authority; git probes may
     # publish only for that exact generation.
     _persist_session_cwd_and_schedule_git_meta(session, resolved)
-    try:
+    with contextlib.suppress(Exception):
         from tools.terminal_tool import cleanup_vm
-
         cleanup_vm(session["session_key"])
-    except Exception:
-        pass
     return resolved