"""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 (method_ctx.bind_module), so they reference server.py globals bare.
"""

from __future__ import annotations

import logging

import contextlib

from .method_ctx import bind_module


@contextlib.contextmanager
def _session_turn_admission(session: dict):
    """Hold process admission until the history-locked running claim is visible to idle probes."""
    from hermes_cli.backend_retirement import retirement

    with retirement.work() as admitted, session["history_lock"]:
        yield admitted


def _start_session_work(target, *, name: str, session: dict | None = None):
    """Reserve before spawning; release only after the worker (including cleanup) has unwound."""
    from agent.memory_provider import spawn_context_thread
    from hermes_cli.backend_retirement import retirement

    if not retirement.acquire():
        return None

    def run():
        try:
            target()
        finally:
            retirement.release()

    try:
        thread = spawn_context_thread(run, name=name)
        if session is not None:
            session["_run_thread"] = thread
        thread.start()
        return thread
    except BaseException:
        retirement.release()
        raise


def _notify_session_boundary(event_type: str, session_id: str | None, platform: str | None = None) -> None:
    """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:
            invoke_hook(event_type, session_id=session_id, platform=_resolve_agent_platform(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"})


def _lease_metadata(live_session_id: str) -> dict:
    """Writer identity for a lease: ``live_session_id`` is half of ``_is_same_writer``; the delivery flag is
    what Bot Chat gates live delivery on (session_notifications)."""
    return {"live_session_id": live_session_id, "bot_live_delivery_consumer": True}


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]:
    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(), registry_home=profile_home,
            metadata=_lease_metadata(live_session_id),
            track_liveness=str(surface or "").strip().lower() == "desktop")
    except Exception as exc:
        logger.warning("Failed to claim active session slot: %s", exc)
        # Fail CLOSED: an errored claim has NOT proven the session unowned; lease-less = silent double-writer hole.
        # Fail CLOSED regardless of surface: per-session exclusivity is a correctness guarantee (see
        # PER_SESSION_EXCLUSIVE_SUBMIT), and a claim that errors out has NOT proven the session is unowned.
        # Proceeding without a lease here is the silent double-writer hole flagged in the #94595 review
        # (blocker 2).
        return (None, _SESSION_OWNERSHIP_UNAVAILABLE)


def _install_borrowed_lease(sid: str, session: dict, frame: dict) -> None:
    """Adopt the parent's registry slot as an INERT token on a compute-host child.

    An isolated turn runs in the compute-host CHILD process. The parent dashboard already
    claimed the session's active-session lease before routing the turn here, but the child's
    freshly built session record never carried it — so ``_admit_prompt_turn`` re-claimed from
    the child's pid and was fenced out by the parent's own entry (``_is_same_writer`` requires
    same pid AND same live_session_id): every isolated turn failed with "Session ... already
    has a live owner" (#101416). The slot is real and owned upstream, so the child must
    neither claim a second one nor be able to release/transfer the parent's: the token is
    ``enabled=False`` — ``release()`` is a no-op and ``transfer_active_session`` only retargets
    the token locally, so a compression rotation A->B inside the child never reaches the
    registry (the parent re-anchors its real lease from the reported ``session_key``, see
    ``_compute_host_adopt_frame_meta``). NOT ``released=True``: a released token makes the
    transfer fall through to a real registry claim under the child pid.

    Only installed when the frame's ``active_session_lease`` vouch names THIS stored session
    id — a parent lease still keyed on a pre-rotation id must not authorize its continuation.
    Anything else keeps the legacy behaviour — the child claims for itself and any ownership
    conflict fails CLOSED with the visible refusal.
    """
    vouch = frame.get("active_session_lease")
    if not isinstance(vouch, dict) or session.get("active_session_lease") is not None:
        return
    key = str(session.get("session_key") or "")
    if not key or str(vouch.get("session_id") or "") != key:
        return
    from hermes_cli.active_sessions import ActiveSessionLease
    session["active_session_lease"] = ActiveSessionLease(
        lease_id=f"borrowed:{vouch.get('lease_id') or sid}", session_id=key,
        surface=str(frame.get("source") or "desktop"), enabled=False)


def _ensure_active_session_slot(sid: str, session: dict) -> str | None:
    """Claim this session's cap slot on its first real turn; None when ok. session.create/resume deliberately
    do NOT claim: tile paints, reconnect-resumes and abandoned drafts would hold invisible slots (no DB row)
    that starve the messaging gateway sharing the cap. Anything holding a slot must be user-visible. An
    inert borrowed token (see _install_borrowed_lease) also lands here: present = slot held upstream."""
    if session.get("active_session_lease") is not None:
        return None
    key = str(session.get("session_key") or "")
    lease, limit_message = _claim_active_session_slot(
        key, live_session_id=sid, surface=_session_source(session), profile_home=session.get("profile_home"))
    if limit_message is None:
        _attach_lease(session, lease)
        return None
    from hermes_cli.active_sessions import SESSION_NOT_OWNED
    if getattr(limit_message, "reason", None) == SESSION_NOT_OWNED and _take_over_detached_runtime_lease(sid, session, key):
        return None
    return limit_message


def _attach_lease(session: dict, lease) -> None:
    session["active_session_lease"] = lease
    session.pop("_lease_taken_over", None)  # owning again: its finalize ends the row like any owner


def _detached_lease_holder(session: dict, key: str) -> tuple[str, dict] | None:
    """A sibling runtime of ``session`` (same key, same profile) holding an unreleased lease with no client.
    Caller holds ``_sessions_lock``."""
    for other_sid, other in _sessions.items():
        if (other is session or other.get("_finalized") or str(other.get("session_key") or "") != key
                or not _live_profile_matches(other, session.get("profile_home"))):
            continue
        lease = other.get("active_session_lease")
        if lease is not None and not lease.released and _transport_is_dead(other.get("transport")):
            return other_sid, other
    return None


def _take_over_detached_runtime_lease(sid: str, session: dict, key: str) -> bool:
    """Hand a detached sibling runtime's lease for ``key`` to ``session``; True when it now owns the slot.

    Lease identity is (pid, live_session_id), so a client that reconnects under a NEW runtime — Desktop
    restoring a chat after a renderer crash, a reload during the client-gone settle window (4009 fences
    ``session.resume``) — is a "different writer" of its own chat and was refused ``SESSION_NOT_OWNED`` until
    the old runtime's reap released the lease (up to grace + activity-stale + interrupt polls). The user IS the
    owner: the old runtime is in THIS process and has no client, so the lease moves and its turn is asked to
    stop (hard interrupt, honoured at the agent's next check — the same best-effort stop the reaper relies on).
    A live foreign pid, a sibling that still has a client, or a same-id runtime of another profile keeps
    refusing — cross-process, multi-window and cross-profile (#100029) exclusivity are untouched. See #104691.
    """
    from hermes_cli.active_sessions import transfer_active_session
    with _session_resume_lock, _sessions_lock:
        if (found := _detached_lease_holder(session, key)) is None:
            return False
        other_sid, other = found
        lease = other["active_session_lease"]
        if not transfer_active_session(lease, session_id=key, metadata=_lease_metadata(sid)):
            return False
        del other["active_session_lease"]
        other["_lease_taken_over"] = True
        _attach_lease(session, lease)
    logger.info("Session %s took over lease for %s from detached runtime %s", sid, key, other_sid)
    try:
        _interrupt_session_turn(other_sid, other, request_id=f"lease-takeover-{sid}")
    except Exception:
        logger.exception("lease-takeover interrupt failed sid=%s", other_sid)
    return True


def _lease_retry(attempts: int, fn) -> Exception | None:
    """Call ``fn`` up to ``attempts`` times (50ms*n backoff: registry writes contend across processes); last
    exception when every try failed, else None."""
    for attempt in range(attempts):
        try:
            fn()
            return None
        except Exception as exc:
            last_error = exc
            if attempt + 1 < attempts:
                time.sleep(0.05 * (attempt + 1))
    return last_error


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
    if (err := _lease_retry(3 if getattr(lease, "track_liveness", False) else 1, lambda: lease.release())) is not None:
        logger.warning("Failed to release active session slot", exc_info=err)
        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


def _release_hosted_room_turn_slot(session: dict) -> None:
    """End-of-turn release for hosted room member sessions (``source=bot_room``) only.

    A hosted room turn is serialized by the room driver's lease, not by this process, so the member
    profile's ``bot_room`` slot is needed only while a turn is in flight. Holding it for the life of
    the live session (which no reaper ever ends: room sessions have no client transport) locked every
    other room worker sharing the home — messaging gateway + Desktop ``serve`` — out of that member
    with ``Refused active session … already held by pid=…`` until this process exited (#106847).
    Call under ``history_lock`` next to ``running = False`` so the next admission never observes the
    stale lease and then runs lease-less; ``_admit_prompt_turn`` re-claims on the following turn.
    """
    from tui_gateway.hosted_room_driver import ROOM_SESSION_SOURCE
    if _session_source(session) == ROOM_SESSION_SOURCE:
        _release_active_session_slot(session)


def _own_live_lease_ids(*, exclude=None) -> set[str]:
    """Snapshot leases still backed by this process's live session records (plus leases deferred past a
    close for an unsettled isolated turn — still ours until the child settles)."""
    with _sessions_lock:
        return {str(lease.lease_id) for session in _sessions.values()
                if (lease := session.get("active_session_lease")) is not None and lease is not exclude
                } | set(_deferred_active_session_leases)


@contextlib.contextmanager
def _other_runtime_lease_guard(session_id: str, session: dict):
    """Release this runtime and lock sibling ownership through the DB write. Yields True (another runtime owns
    the lifecycle -> preserve) when the guard can't be loaded/entered in 3 tries: unknown ownership never ends a row."""
    lease = session.get("active_session_lease")
    try:
        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)
        yield True
        return
    stack = contextlib.ExitStack()
    active: list = []
    own_live_lease_ids = _own_live_lease_ids(exclude=lease)

    def _enter() -> None:
        stack.close()  # drop anything a half-failed previous attempt left behind
        if lease is not None and getattr(lease, "enabled", False):
            guard = release_active_session_liveness_guard(lease, session_id, own_live_lease_ids=own_live_lease_ids)
        else:
            guard = active_session_liveness_guard(
                session_id, registry_home=session.get("profile_home"), own_live_lease_ids=own_live_lease_ids)
        active[:] = [stack.enter_context(guard)]

    if (last_error := _lease_retry(3, _enter)) is not None:
        stack.close()
        logger.warning("Failed to inspect active session leases; preserving session %s: %s", session_id, last_error)
        yield True
        return
    try:
        yield active[0]
    finally:
        stack.close()
        if lease is not None and getattr(lease, "released", False) and session.get("active_session_lease") is lease:
            session.pop("active_session_lease", None)


def _transfer_active_session_slot(sid: str, session: dict, *, new_session_id: str) -> bool:
    if not new_session_id:
        return False
    lease = session.get("active_session_lease")
    if lease is None:
        return True
    try:
        from hermes_cli.active_sessions import transfer_active_session
        if transfer_active_session(lease, session_id=new_session_id, metadata=_lease_metadata(sid)):
            return True
    except Exception:
        logger.debug("Failed to transfer active session slot", exc_info=True)
    if getattr(lease, "track_liveness", False):
        return False
    # Fallback (entry pruned / pid-check transiently failed): reserve the new slot BEFORE releasing the old one so
    # a gateway at the cap can't grab the freed slot and leave this session lease-less; on failure KEEP the old lease.
    # See #49041.
    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 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
    if (old := session.pop("active_session_lease", None)) is not None and (err := _lease_retry(1, old.release)):
        logger.debug("Failed to release stale active session slot", exc_info=err)
    session["active_session_lease"] = new_lease
    return True


# Sources this backend must never end in state.db: the messaging gateway owns those sessions and the TUI is only
# a viewer (ending one causes the Groundhog Day loop, see _finalize_session). Self-created/CLI sources are NOT gateway-owned.
# Sources the TUI backend itself creates ("tui", plus whatever a client passes as its own ``source``) and
# the CLI's own sessions are NOT gateway-owned. See #60609.
_NON_GATEWAY_SOURCES = frozenset({
    "", "tui", "cli", "webui", "desktop", "cron", "kanban", "subagent", "test",
    "local", "acp", "webhook", "api_server", "msgraph_webhook"})


def _is_gateway_owned_source(source: str) -> bool:
    """True when ``source`` resolves to a gateway ``Platform`` (enum member or plugin via ``Platform._missing_``, so
    new platforms are covered automatically); self-owned Platform members (local/webhook/api_server) are excluded."""
    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:
        return False


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:
        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


def _lock_vault_managers(session: dict) -> None:
    """A per-session unlock ends with the session that made it; siblings in the same profile keep theirs."""
    try:
        from agent.vault_backends import unlock

        if sid := session.get("_sid"):
            unlock.release_session(sid)
    except Exception:
        logging.getLogger(__name__).debug("vault manager lock on session end failed", exc_info=True)


def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> None:
    """Best-effort finalize hook + memory commit; mirrors the CLI exit path so a force-quit mid-turn (double
    Ctrl-C, terminal close, SIGHUP) loses nothing."""
    if not session or session.get("_finalized"):
        return
    session["_finalized"] = True
    _lock_vault_managers(session)
    if (history_ready := session.get("resume_history_ready")) is not None and not history_ready.is_set():
        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")
    # Automatic Desktop cleanup releases its lease inside the lifecycle guard below; other paths keep force/end semantics.
    if not _desktop_automatic_cleanup:
        _release_active_session_slot(session)
    if (stop_event := session.get("_notif_stop")) is not None:
        stop_event.set()
    agent = session.get("agent")
    with (session.get("history_lock") or contextlib.nullcontext()):
        history = list(session.get("history", []))
    # Persist via ``_persist_session``'s marker-based dedup (gateway-shutdown flush contract). Do NOT pass
    # ``conversation_history``: ``session["history"]`` 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.
    if 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). The end-of-session
    # hooks and the memory commit read the provider's config/credentials at call time; every caller of this
    # chokepoint is an unscoped reaper/Timer/atexit/pool thread, so bind the SESSION's profile here — unscoped
    # they fail closed under multiplex (tail never committed) or, on the Desktop backend serving a named
    # profile, commit a secondary's transcript to the launch profile's memory tenant (same class as #110622).
    with _session_profile_runtime_scope(session):
        if agent is not None:
            with contextlib.suppress(Exception):
                from hermes_cli.lifecycle import invoke_hook
                invoke_hook(
                    "on_session_end", completed=False, interrupted=True,
                    session_id=getattr(agent, "session_id", None) or session.get("session_key", ""),
                    model=getattr(agent, "model", "unknown"), 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)

    session_key = session.get("session_key")
    session_id = getattr(agent, "session_id", None) or session_key
    _notify_session_boundary("on_session_finalize", session_id, _session_source(session))
    # End the state.db row so it doesn't linger as a ghost in /resume. Use session_id (agent.session_id), not
    # session_key: after compression the key may be the stale ended parent while session_id is the live continuation.
    # Fix for #20001.
    if _desktop_automatic_cleanup and not session_id:
        _release_active_session_slot(session)
    # ``_lease_taken_over``: a new runtime for the same chat owns the row and its delegations now.
    _lifecycle_guard = (_other_runtime_lease_guard(session_id, session)
                        if _desktop_automatic_cleanup and session_id
                        else contextlib.nullcontext(bool(session.get("_lease_taken_over"))))
    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)
        if session_id:
            # 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 (gateway self-heals to the parent,
                    # compression splits back to the reaped child, forever).
                    if _is_gateway_owned_source((db.get_session(session_id) or {}).get("source", "")):
                        _tui_owns_lifecycle = False
                    elif _tui_owns_lifecycle and not _desktop_automatic_cleanup:
                        # Automatic Desktop cleanup (ws_orphan_reap, idle_timeout, etc.) reclaims
                        # runtime but must not end the durable row — the conversation stays open
                        # and resumable until the user explicitly closes or archives it.  #105588
                        db.end_session(session_id, end_reason)
    # ``_teardown_session`` follows with ``agent.close()``, whose ``_finalize_owned_session_row`` ends the row as
    # ``agent_close`` through the agent's own handle — every spare decision above would be undone one call later.
    if agent is not None and (_desktop_automatic_cleanup or not _tui_owns_lifecycle):
        agent._end_session_on_close = False
    # 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 — a viewer tab must not kill gateway 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)
    # Session-persistent code kernels (execute_code) share this owner key and die at the same boundary, like the
    # gateway's /stop and /new (approval.clear_session); otherwise each finished conversation keeps a live
    # interpreter until kernel_idle_timeout. Only when the TUI owns the lifecycle: a viewer tab over a
    # gateway-owned session must not kill the gateway's kernels.
    if _tui_owns_lifecycle and session_key:
        with contextlib.suppress(Exception):
            from tools.approval import clear_session
            clear_session(str(session_key))
    # Close the slash-worker in this single ``_finalized``-guarded chokepoint (a direct caller can't leak it); idempotent.
    with contextlib.suppress(Exception):
        if worker := session.get("slash_worker"):
            worker.close()


# End reasons where the BACKEND reclaimed a session the client never asked to close (else its next prompt fails
# against a forgotten id). Client-initiated reasons (``tui_close`` etc.) are deliberately absent.
_RECLAIM_END_REASONS = frozenset({"idle_timeout", "lru_evict", "ws_orphan_reap"})


def _announce_session_reclaimed(session: dict, end_reason: str) -> None:
    """Tell connected clients a session was reclaimed out from under them. Broadcast, not session-targeted: reap
    paths run on timer threads with no contextvar binding and no live transport, so ``_emit`` would hit stdio."""
    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})
    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 (``session.close`` + 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
        # One approval callback per key: after a takeover it is the new runtime's registration.
        if (key := session.get("session_key")) and not session.get("_lease_taken_over"):
            unregister_gateway_notify(key)
    # agent.close() → shutdown_memory_provider reads the provider's config/credentials at call time; same
    # scope rule as _finalize_session (every caller here is an unscoped reaper/atexit/pool thread).
    with contextlib.suppress(Exception), _session_profile_runtime_scope(session):
        if hasattr(agent := session.get("agent"), "close"):
            agent.close()


def _attach_worker(sid: str, session: dict, worker) -> None:
    """Store worker on session iff sid still maps to it, else close it (a concurrent teardown popped the session)."""
    with _sessions_lock:
        if _sessions.get(sid) is session:
            session["slash_worker"] = worker
            return
    worker.close()


# Wall-clock timestamps, like session last_active; retained after close/reap.
_closed_session_activity: dict[str, float] = {}


def _pop_session_by_id(sid: str) -> dict | None:
    """Atomically detach one live session from the registry — the ownership claim for teardown (a concurrent
    close/reaper no-ops). Separate from ``_teardown_session``: slow finalization must not run under the resume lock."""
    with _sessions_lock:
        session = _sessions.pop(sid, None)
        if session is not None:
            from hermes_constants import get_hermes_home

            home = str(Path(session.get("profile_home") or get_hermes_home()).resolve())
            last_active = time.time() if session.get("running") else float(session.get("last_active") or 0)
            _closed_session_activity[home] = max(_closed_session_activity.get(home, 0), last_active)
            session["_closing"] = True
            session["_sid"] = sid  # out of _sessions now, so teardown can't recover the live id by scanning
    return session


def _teardown_popped_session(session: dict | None, *, end_reason: str = "tui_close") -> bool:
    """Finish a close after the caller has atomically detached the session."""
    if session is None:
        return False
    run_thread = session.get("_run_thread")
    if end_reason != "tui_shutdown" and run_thread is not None and run_thread is not threading.current_thread():
        try:
            if run_thread.is_alive():
                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)
        except Exception:
            logger.debug("failed waiting for session turn thread", exc_info=True)
    if end_reason != "tui_shutdown":
        _settle_isolated_turn_before_close(session)
    _teardown_session(session, end_reason=end_reason)
    return True


# lease_id -> REAL lease of a closed session whose isolated child turn has not settled yet. Still live
# authority for the orphan sweep (``_own_live_lease_ids``); released by ``_release_deferred_active_session_lease``.
_deferred_active_session_leases: dict[str, Any] = {}


def _settle_isolated_turn_before_close(session: dict) -> None:
    """An isolated turn runs in the compute-host child, not on ``_run_thread``: interrupt it and give it the
    same close grace, and if it still has not settled keep the REAL lease out of finalize's release — the
    completion callback releases it on the correlated turn.end/turn.error (child death fails pending turns
    the same way). The RPC close is bounded; ownership ends with the child's last write, never with the
    grace timer, else a second backend acquires the stored session while the child is still writing."""
    if not session.get("_compute_host_turn_id") or not _session_uses_compute_host(session):
        return
    with contextlib.suppress(Exception):
        _interrupt_session_turn(_lifecycle_own_sid(session), session)
    deadline = time.monotonic() + _TURN_SETTLE_BEFORE_CLOSE_SECONDS
    while session.get("_compute_host_turn_id") and time.monotonic() < deadline:
        time.sleep(0.05)
    with session["history_lock"]:
        if not session.get("_compute_host_turn_id") or (lease := session.pop("active_session_lease", None)) is None:
            return
        session["_deferred_active_session_lease"] = lease
        _deferred_active_session_leases[str(lease.lease_id)] = lease
        _deferred_active_session_lease_ages[str(lease.lease_id)] = time.time()
    logger.warning("isolated turn still live after %.1fs close grace; holding lease for %s until the child settles",
                   _TURN_SETTLE_BEFORE_CLOSE_SECONDS, session.get("session_key"))


# A deferred lease is released by the compute-host turn's completion callback
# (_on_compute_host_turn_done → _release_deferred_active_session_lease). If that callback
# is lost — supervisor restart/reload, a child killed without failing its pending turns,
# a dropped completion — the lease sits in the registry FOREVER: _own_live_lease_ids
# vouches for it, so the orphan sweep never reclaims it, and the concurrency cap treats
# the dead session as active (#62823 zombie slot). A deferred lease past this generous
# ceiling (the longest legitimate isolated turn is the compression ceiling, minutes) is
# force-released by the reaper tick instead of leaking the slot until process exit.
_DEFERRED_ACTIVE_SESSION_LEASE_TTL_SECONDS = 1800.0
_deferred_active_session_lease_ages: dict[str, float] = {}


def _reap_stale_deferred_leases(now: float | None = None) -> int:
    """Force-release deferred leases whose settlement callback never arrived. Returns the count."""
    now = time.time() if now is None else now
    stale = [
        lease_id for lease_id, deferred_at in list(_deferred_active_session_lease_ages.items())
        if now - deferred_at > _DEFERRED_ACTIVE_SESSION_LEASE_TTL_SECONDS
    ]
    reaped = 0
    for lease_id in stale:
        _deferred_active_session_lease_ages.pop(lease_id, None)
        lease = _deferred_active_session_leases.pop(lease_id, None)
        if lease is None:
            continue
        if (err := _lease_retry(3, lease.release)) is not None:
            logger.warning("Failed to force-release stale deferred active session lease %s", lease_id,
                           exc_info=err)
            continue
        logger.warning("Force-released deferred active session lease %s held past %.0fs without a "
                       "compute-host settlement (zombie concurrency slot, #62823)",
                       lease_id, _DEFERRED_ACTIVE_SESSION_LEASE_TTL_SECONDS)
        reaped += 1
    return reaped


def _release_deferred_active_session_lease(session: dict) -> None:
    """Settlement half of ``_settle_isolated_turn_before_close``; a no-op for sessions that never deferred."""
    lease = session.pop("_deferred_active_session_lease", None)
    if lease is None:
        return
    _deferred_active_session_leases.pop(str(lease.lease_id), None)
    _deferred_active_session_lease_ages.pop(str(lease.lease_id), None)
    if (err := _lease_retry(3, lease.release)) is not None:
        logger.warning("Failed to release deferred active session slot", exc_info=err)


def _close_session_by_id(
    sid: str, *, end_reason: str = "tui_close", predicate: Callable[[dict], bool] | None = None) -> bool:
    """Idempotent teardown funnel for callers with no resume race (resume-sensitive callers pop under
    ``_session_resume_lock`` and call ``_teardown_popped_session`` after releasing it). Automatic reapers pass
    ``predicate`` to revalidate under ``_sessions_lock`` right before the claim, so a stale scan can't close a
    session that reattached."""
    with _sessions_lock:  # RLock: predicate + claim in one critical section
        current = _sessions.get(sid)
        if predicate is not None and (current is None or not predicate(current)):
            return False
        session = _pop_session_by_id(sid)
    return _teardown_popped_session(session, end_reason=end_reason)


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)


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), idle."""
    return bool(_ws_session_is_detached(session) and not session.get("running"))


def _interrupt_session_turn(sid: str, session: dict, *, request_id: str | None = None) -> bool:
    """Apply the shared ``session.interrupt`` contract to one claimed session; returns whether the compute-host control
    channel was used. The WS orphan reaper reuses this so a dead client gets the same partial-history/queue semantics."""
    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(), so a lazy session would spawn a child to interrupt.
        if should_interrupt or session.get("_compute_host_active"):
            _get_compute_host_supervisor().interrupt(sid, request_id=request_id)
    else:
        run_thread_alive = (rt := session.get("_run_thread")) is not None and rt.is_alive()
    with session["history_lock"]:
        session["_turn_cancel_requested"] = True
        session["queued_prompt"] = None
        session.pop("queued_prompts", None)
        session["_queued_prompt_generation"] = int(session.get("_queued_prompt_generation", 0)) + 1
    if should_interrupt:
        # Sibling of gateway/run_agent_cache.py::_interrupt_and_clear_session: a user-initiated stop of a
        # live TUI/desktop turn is the same "loop is gone" event for plugins holding per-turn external
        # resources. Observer-only; dispatch failures never break the interrupt.
        try:
            from hermes_cli.plugins import invoke_hook as _invoke_hook
            _invoke_hook(
                "agent_loop_stopped", session_key=session.get("session_key", ""), platform="tui",
                reason="user_stop", invalidation_reason="session_interrupt",
            )
        except Exception:
            logger.debug("agent_loop_stopped hook dispatch failed", exc_info=True)
    if not use_compute_host:
        if should_interrupt:
            from agent.interrupt_compat import request_hard_interrupt
            request_hard_interrupt(session.get("agent"))
        # Background delegations are detached from the turn's interrupt fan-out; a stop ends them too
        # (own UI sid + spawner id only — a viewer tab must not kill gateway work). Each returns as an
        # interrupted completion with its partial output.
        with contextlib.suppress(Exception):
            from tools.async_delegation import interrupt_for_session
            interrupt_for_session(
                origin_ui_session_id=_lifecycle_own_sid(session, sid), reason="user_stop",
                parent_session_id=str(getattr(session.get("agent"), "session_id", "") or ""))
        if not run_thread_alive:
            with session["history_lock"]:
                if session.get("running"):
                    session["running"] = False
                    _clear_inflight_turn(session)
    _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


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 session_key so a delegation from an earlier tab keeps it alive.

    See #60609.
    """
    if session is None:
        with _sessions_lock:
            session = _sessions.get(sid)
    if not session:
        return False
    own_sid = _lifecycle_own_sid(session, sid)
    owned_session_key = session_key = str(session.get("session_key") or "")
    session_id = getattr(session.get("agent"), "session_id", None) or session_key
    if session_id:
        # Only when this session may end its durable row by key — never for gateway-originated sessions (TUI is a
        # viewer there). Unknown DB state -> assume ownership.
        # The row lives in the session's OWN store (a named-profile session's row is invisible to
        # the launch handle, which would leave this guard permanently dead).
        with contextlib.suppress(Exception), _session_db(session) as db:
            if db is not None and _is_gateway_owned_source((db.get_session(session_id) or {}).get("source", "")):
                owned_session_key = ""
    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)
        return True  # a transient registry/import failure must not become destructive cleanup


# One pending WS-orphan reap Timer per live sid; guarded by _sessions_lock. Cancelled by _cancel_ws_orphan_reap from
# every resume/reuse/transport-rebind path — else a reap on a reattached session triggers a reap->broadcast->resume storm.
_pending_ws_reaps: dict[str, threading.Timer] = {}


def _cancel_ws_orphan_reap(sid: str) -> None:
    """Cancel a pending WS-orphan reap for ``sid`` (client came back). Called from every path that re-binds a live
    transport; closes the fired-but-not-run Timer race and stops dead Timers accumulating on flappy clients."""
    with _sessions_lock:
        timer = _pending_ws_reaps.pop(sid, None)
    if timer is not None:
        with contextlib.suppress(Exception):
            timer.cancel()


def _reattach_refusal(rid, sid: str, session: dict) -> dict | None:
    """Under ``_session_resume_lock``: why a reattaching RPC (resume/activate/prompt.submit) must NOT rebind
    ``session`` — it is stale, or a client-gone interrupt is still settling and the reap Timer must keep
    polling. None when the reattach may proceed."""
    if _sessions.get(sid) is not session:
        return _err(rid, 4007, "session no longer live; retry resume")
    if session.get("_client_gone_interrupt_requested"):
        return _err(rid, 4009, "session disconnect interrupt settling")
    return None


def _rebind_live_transport(sid: str, session: dict, transport: Transport) -> None:
    """Attach a live peer without displacing existing subscribers (caller holds ``history_lock``).
    Subagent control authority needs no bookkeeping here: it resolves against ``session["transport"]``
    at RPC time (``tools.delegate_tool_registry._subagent_transport_matches``)."""
    if transport is not _detached_ws_transport and _transport_is_dead(transport):
        # The rebinding socket already closed: its disconnect cleanup ran before this late RPC (a
        # resume-then-drop burst), so nothing will detach it again. The client is NOT back — re-arm the
        # reap the caller cancelled instead of leaving a detached session with no Timer (#116464).
        with _sessions_lock:
            if _ws_session_is_detached(session) and sid not in _pending_ws_reaps:
                _schedule_ws_orphan_reap(sid)
        return
    _attach_session_transport(session, transport)
    # Every transport that showed this session (pop-outs resume the same sid); on disconnect the last
    # viewer becomes the transport instead of the drop sentinel.
    session.setdefault("viewers", {})[transport] = time.time()
    # See #83716.
    if transport is not _detached_ws_transport:
        _cancel_ws_orphan_reap(sid)  # the client is back — a pending ws-orphan reap must not fire


def _ws_orphan_turn_activity_is_fresh(session: dict) -> bool:
    """Whether a detached RUNNING turn's activity clock (``_touch_activity``) is still fresh — the reaper must NOT
    interrupt healthy detached work (closed laptop). Conservative: disabled threshold, missing/opaque agent, unreadable
    summary or never-stamped clock all report NOT fresh (eligible for interrupt-at-grace) to keep the wedged-turn net.

    Reuses the agent's existing activity summary (``_touch_activity`` is stamped by API waits, stream
    tokens, and tool heartbeats — the same clock the turn-liveness watchdog samples; see
    agent/turn_liveness.py). See #100325, #98028.
    Isolated turns mirror that clock from the child under a unique dispatch token;
    their monotonic samples keep aging even if the child or its pipe stalls.
    """
    if _WS_ORPHAN_ACTIVITY_STALE_S <= 0:
        return False
    if session.get("_compute_host_turn_id"):
        with session["history_lock"]:
            stamp = session.get("_compute_host_activity_ns")
            return (session.get("running", False) and isinstance(stamp, int)
                    and 0 <= (time.perf_counter_ns() - stamp) / 1_000_000_000 < _WS_ORPHAN_ACTIVITY_STALE_S)
    if not callable(summary_fn := getattr(session.get("agent"), "get_activity_summary", None)):
        return False
    try:
        elapsed = summary_fn().get("seconds_since_activity")
        return elapsed is not None and float(elapsed) < _WS_ORPHAN_ACTIVITY_STALE_S
    except Exception:
        return False


def _schedule_ws_orphan_reap(
    sid: str, *, delay_s: float | None = None, _expected_timer: threading.Timer | None = None,
) -> None:
    """After a grace window, reap session ``sid`` iff it's still orphaned. Called from the WS-disconnect path; a
    reconnect or ``session.resume`` cancels the reap by re-binding a live transport. Disabled when grace is 0.

    The grace is measured in AWAKE (monotonic) time: ``threading.Timer``'s wait elapses in wall-clock time on
    platforms without a monotonic condvar (macOS lacks ``pthread_condattr_setclock``), so a system sleep makes
    the timer fire "early" in awake-time terms. Without the sleep check below, closing a laptop lid for longer
    than the grace reaped the parked session at the instant of wake — before the Desktop's WS reconnect or
    ``session.resume`` could re-bind a transport — so every sleep/wake cycle 404'd the open chat (#44183)."""
    if _WS_ORPHAN_REAP_GRACE_S <= 0:
        return
    grace_s = _WS_ORPHAN_REAP_GRACE_S if delay_s is None else max(0.0, delay_s)
    # Sample both clocks at arm time: time.monotonic() (mach_absolute_time / CLOCK_MONOTONIC)
    # does not advance while the host is asleep, so the divergence between the two clocks'
    # elapsed times at fire time is exactly the time the host spent asleep.
    armed_monotonic = time.monotonic()
    armed_wall = time.time()

    def _reap() -> None:
        # The timer fired. If more wall-clock than monotonic time elapsed, the host
        # slept through the wait: the grace has NOT been granted in awake time, so
        # re-arm for the remaining awake grace instead of reaping. The slack keeps
        # ordinary timer jitter and NTP slew from re-arming a legitimately-expired
        # reap, and a fired-without-elapsed timer (tests, spurious dispatch) shows
        # zero divergence and reaps normally.
        slept_s = (time.time() - armed_wall) - (time.monotonic() - armed_monotonic)
        if slept_s > _WS_ORPHAN_REAP_SLEEP_SLACK_S:
            rearm_delay = max(0.0, grace_s - (time.monotonic() - armed_monotonic))
            if rearm_delay <= 0:
                pass  # no awake grace left — fall through and reap
            else:
                # Re-arm through the public scheduler with THIS timer as the expected
                # one: the fresh closure's identity guard then matches the entry it
                # installs, so the awake-remainder fire proceeds to the real reap.
                _schedule_ws_orphan_reap(sid, delay_s=rearm_delay, _expected_timer=timer)
                return
        # 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. Order: resume_lock -> sessions_lock.
        reschedule_delay = interrupt_session = session = None
        with _session_resume_lock, _sessions_lock:
            # Keep ownership through interrupt I/O and continuation registration. A cancelled
            # callback may already be dispatched, but cannot act on a later detachment.
            if _pending_ws_reaps.get(sid) is not timer:
                return
            current = _sessions.get(sid)
            if current is None:
                _pending_ws_reaps.pop(sid, None)
                return
            if not _ws_session_is_detached(current):
                # This Timer is abandoning the interrupt claim because another
                # writer moved the live record off the detached transport.
                # Do not leave reattach RPCs fenced with 4009, or let this
                # generation's settlement polls shorten a later detachment.
                current.pop("_client_gone_interrupt_requested", None)
                current.pop("_client_gone_interrupt_polls", None)
                _pending_ws_reaps.pop(sid, None)
                return
            if _session_has_active_delegations(sid, current):
                reschedule_delay = _WS_ORPHAN_REAP_GRACE_S
            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), re-check each grace.
                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 = current["_client_gone_interrupt_polls"] = int(current.get("_client_gone_interrupt_polls") or 0) + 1
                # See #85578.
                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 reschedule_delay is None:
                _pending_ws_reaps.pop(sid, None)
        if interrupt_session is not None:
            try:
                isolated = _interrupt_session_turn(sid, interrupt_session, request_id=f"client-gone-{sid}")
                logger.info("client_gone sid=%s action=interrupt turn_isolation=%s", sid, isolated)
            except Exception:
                logger.exception("client_gone interrupt failed sid=%s", sid)
                with _sessions_lock:
                    if (_sessions.get(sid) is interrupt_session
                            and _pending_ws_reaps.get(sid) is timer):
                        interrupt_session.pop("_client_gone_interrupt_requested", None)
        if reschedule_delay is not None:
            _schedule_ws_orphan_reap(sid, delay_s=reschedule_delay, _expected_timer=timer)
            return
        if session is not None and session.get("_client_gone_interrupt_requested"):
            logger.info("client_gone sid=%s action=reap", sid)
        _teardown_popped_session(session, end_reason="ws_orphan_reap")

    with _sessions_lock:
        if _expected_timer is not None and _pending_ws_reaps.get(sid) is not _expected_timer:
            return
        timer = threading.Timer(_WS_ORPHAN_REAP_GRACE_S if delay_s is None else max(0.0, delay_s), _reap)
        timer.daemon = True
        prior = _pending_ws_reaps.pop(sid, None)
        _pending_ws_reaps[sid] = timer
    if prior is not None:
        with contextlib.suppress(Exception):
            prior.cancel()
    timer.start()


def _close_sessions_for_transport(transport, *, end_reason: str = "ws_disconnect") -> tuple[int, int]:
    """Single WS-disconnect teardown entry point: reap close_on_disconnect sessions (sidecar/dashboard) immediately;
    re-point the rest at the detached transport (later emits miss the dead socket) for the grace-windowed WS-orphan
    reaper. Returns ``(reaped, detached)`` counts.

    Multi-client fan-out: the departing transport is DETACHED from every session first. A session that still has
    another client attached keeps streaming and is neither parked nor reaped — a watcher leaving must not end the
    turn the remaining client is reading. Only the sessions left clientless take the historical
    close_on_disconnect / park-sentinel path, so a single-client disconnect behaves exactly as it always has."""
    clientless = _detach_transport_from_sessions(transport)
    reaped = detached = 0
    for sid, session in clientless:
        claimed_for_teardown = None
        should_schedule_reap = False
        # session.resume fast-path attaches under _session_resume_lock: take it so a reconnect can't attach
        # between the detach above and the claim.
        with _session_resume_lock, _sessions_lock:
            current = _sessions.get(sid)
            if current is not session:
                continue
            # Prune the departing viewer registration in every branch; it must not affect the new owner.
            (current.get("viewers") or {}).pop(transport, None)
            # Revalidate before claiming (#77129, kept under fan-out): between _detach_transport_from_sessions
            # returning this session as clientless and this claim, a concurrent session.resume can attach a NEW
            # live transport. Tearing the session down, or parking the sentinel over it, would knock an attached
            # client into detached state and arm an orphan reap against a session that has a live owner. Attach
            # and detach both serialize on _session_transport_lock, so this check is race-free against them; that
            # lock is a leaf, so taking it under _sessions_lock is safe and _session_has_live_transport does not
            # re-acquire it.
            with _session_transport_lock:
                if _session_has_live_transport(current, excluding=transport):
                    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 (pop-out viewer) still shows the session:
                # re-bind to the most recent surviving viewer instead.
                viewers = current.get("viewers") or {}
                # See #83716.
                viewers.pop(transport, None)
                live = [vt for vt, ts in sorted(viewers.items(), key=lambda kv: kv[1]) if not _transport_is_dead(vt)]
                if live:
                    _rebind_live_transport(sid, current, live[-1])
                else:
                    current["transport"] = _detached_ws_transport
                    current.pop("_client_gone_interrupt_requested", None)
                    current.pop("_client_gone_interrupt_polls", None)
                    should_schedule_reap = True
                    # Register before releasing the detachment claim: an old disconnect
                    # must not arm its first timer over a reconnect's newer detachment.
                    with contextlib.suppress(Exception):
                        _schedule_ws_orphan_reap(sid)
        if claimed_for_teardown is not None:
            reaped += _teardown_popped_session(claimed_for_teardown, end_reason=end_reason)
        elif should_schedule_reap:
            detached += 1
    return reaped, detached


def register(server) -> None:
    """Publish this module's helpers onto ``server``, rebound to its globals."""
    bind_module(globals(), server, skip=("_",))