diff --git a/tests/tui_gateway/test_session_resume_db_ownership.py b/tests/tui_gateway/test_session_resume_db_ownership.py new file mode 100644 index 0000000000000..309d1e3998a03 --- /dev/null +++ b/tests/tui_gateway/test_session_resume_db_ownership.py @@ -0,0 +1,290 @@ +"""``session.resume`` must not abandon the profile-scoped SessionDB it opens. + +In app-global remote mode a resume for another local profile opens a DEDICATED +``SessionDB(db_path=/state.db)`` handle (the ``session.resume`` handler +in tui_gateway/methods_session.py). That handle is the caller's to close until +it is handed to the long-lived agent by ``_init_session`` — and +``_init_session`` never closes a caller-supplied ``session_db`` (its +``_init_owns_db`` stays False for that case). + +Every early return before that transfer used to drop the handle on the floor, +so its SQLite fds stayed open for as long as anything kept the instance +reachable — and a ``SessionDB`` pins ITSELF once its background token writer +starts (``atexit.register(self._drain_token_queue_at_exit)``, which only +``close()`` unregisters). + +Pinned here, in both directions: + +* the pre-transfer early returns (session-not-found, "resume failed", the + live-session fast path, the deferred cold-resume return) all close it; +* a resume that COMPLETES the transfer leaves it open — closing there would + fault every later turn with "Cannot operate on a closed database"; +* an ``_init_session`` that raises AFTER registering the session must drop that + half-built registration, otherwise the live-session fast path serves a + session whose db we just closed on every later resume of the same id; +* the shared launch-profile handle (``_get_db()``) is never closed, since it + outlives the RPC. +""" + +from __future__ import annotations + +import types + +import pytest + +from tui_gateway import server + + +class _RecordingDB: + """Stand-in for ``hermes_state.SessionDB`` that counts ``close()`` calls. + + Implements only the surface ``session.resume`` touches. + """ + + def __init__(self, db_path=None, **_kwargs): + self.db_path = db_path + self.closed = 0 + self.rows: dict = {} + self.reopen_error: Exception | None = None + + def close(self): + self.closed += 1 + + def get_session(self, target): + return self.rows.get(target) + + def get_session_by_title(self, _target): + return None + + def resolve_resume_session_id(self, target): + return target + + def reopen_session(self, _target): + if self.reopen_error is not None: + raise self.reopen_error + + def get_resume_conversations(self, _target): + return ([], []) + + def get_ancestor_display_prefix(self, _target): + return [] + + def get_messages_as_conversation(self, _target, **_kwargs): + return [] + + +@pytest.fixture() +def profile_dbs(monkeypatch, tmp_path): + """Route profile-scoped opens to _RecordingDB; yield the list of opens. + + ``params['profile']`` selects the profile scope; omitting it resolves to + the launch profile (``_profile_home`` -> None) and the shared handle. + """ + opened: list[_RecordingDB] = [] + profile_home = tmp_path / "work" + profile_home.mkdir() + + def _factory(db_path=None, **kwargs): + db = _RecordingDB(db_path=db_path, **kwargs) + opened.append(db) + return db + + monkeypatch.setattr("hermes_state.SessionDB", _factory) + monkeypatch.setattr( + server, "_profile_home", lambda profile: profile_home if profile else None + ) + monkeypatch.setattr(server, "_profile_configured_cwd", lambda _home: str(tmp_path)) + # The handler builds nothing on the paths under test; keep it hermetic and + # off the real agent/secret/HERMES_HOME machinery. + monkeypatch.setattr(server, "_enable_gateway_prompts", lambda: None) + monkeypatch.setattr(server, "_find_live_session_by_key", lambda _key: None) + monkeypatch.setattr(server, "_schedule_agent_build", lambda *a, **k: None) + monkeypatch.setattr(server, "_schedule_session_cap_enforcement", lambda *a, **k: None) + monkeypatch.setattr(server, "_maybe_schedule_auto_continue", lambda *a, **k: None) + monkeypatch.setattr(server, "_default_session_cwd", lambda *a, **k: str(tmp_path)) + known = set(server._sessions) + yield opened + with server._sessions_lock: + for sid in [s for s in server._sessions if s not in known]: + server._sessions.pop(sid, None) + + +def _resume(**params): + return server.handle_request( + {"id": "1", "method": "session.resume", "params": params} + ) + + +def test_resume_closes_profile_db_when_session_not_found(profile_dbs): + """The 'session not found' early return must not leak the handle.""" + resp = _resume(session_id="missing", profile="work") + + assert resp["error"]["code"] == 4007 + assert len(profile_dbs) == 1 + assert profile_dbs[0].closed == 1 + + +def test_resume_closes_profile_db_when_reopen_fails(profile_dbs, monkeypatch): + """The 'resume failed' early return must not leak the handle.""" + + def _factory(db_path=None, **kwargs): + db = _RecordingDB(db_path=db_path, **kwargs) + db.rows["s1"] = {"id": "s1", "cwd": ""} + db.reopen_error = RuntimeError("database is locked") + profile_dbs.append(db) + return db + + monkeypatch.setattr("hermes_state.SessionDB", _factory) + + resp = _resume(session_id="s1", profile="work") + + assert resp["error"]["code"] == 5000 + assert "resume failed" in resp["error"]["message"] + assert profile_dbs[0].closed == 1 + + +def test_resume_closes_profile_db_on_live_session_fast_path(profile_dbs, monkeypatch): + """Re-resuming an already-live session returns early — and must close. + + This is the hottest leak in practice: every reconnect/tile-paint resume of + a chat that is already live takes this path, so the fd growth tracked + reconnect count rather than anything rare. + """ + + def _factory(db_path=None, **kwargs): + db = _RecordingDB(db_path=db_path, **kwargs) + db.rows["s1"] = {"id": "s1", "cwd": ""} + profile_dbs.append(db) + return db + + monkeypatch.setattr("hermes_state.SessionDB", _factory) + monkeypatch.setattr(server, "_find_live_session_by_key", lambda _key: ("live-sid", {})) + monkeypatch.setattr( + server, + "_live_session_payload", + lambda sid, session, **_kwargs: {"session_id": sid}, + ) + monkeypatch.setattr(server, "_child_run_active", lambda _key: False) + + resp = _resume(session_id="s1", profile="work") + + assert resp["result"]["resumed"] == "s1" + assert profile_dbs[0].closed == 1 + + +def test_resume_closes_profile_db_on_deferred_cold_resume(profile_dbs, monkeypatch): + """The DEFAULT resume path returns before any transfer — and must close. + + A cold resume without ``eager_build`` registers a deferred session record + (no agent, no db reference) and builds the agent later off the response + path, so the handle opened here is never handed to anyone. + """ + + def _factory(db_path=None, **kwargs): + db = _RecordingDB(db_path=db_path, **kwargs) + db.rows["s1"] = {"id": "s1", "cwd": ""} + profile_dbs.append(db) + return db + + monkeypatch.setattr("hermes_state.SessionDB", _factory) + monkeypatch.setattr(server, "_stored_session_runtime_overrides", lambda _found: {}) + + resp = _resume(session_id="s1", profile="work") + + assert resp["result"]["session_key"] == "s1" + assert resp["result"]["status"] == "idle" + assert profile_dbs[0].closed == 1 + + +def test_resume_keeps_profile_db_open_after_ownership_transfer(profile_dbs, monkeypatch): + """A COMPLETED resume transfers the handle to the agent — do not close it. + + Guards the other direction: closing here would hand the live session a dead + connection and fault every subsequent turn. + """ + captured: dict = {} + + def _factory(db_path=None, **kwargs): + db = _RecordingDB(db_path=db_path, **kwargs) + db.rows["s1"] = {"id": "s1", "cwd": ""} + profile_dbs.append(db) + return db + + def _fake_make_agent(sid, key, session_db=None, **_kwargs): + captured["agent_db"] = session_db + return types.SimpleNamespace(model="test") + + def _fake_init_session(sid, key, agent, history, session_db=None, **_kwargs): + captured["init_db"] = session_db + + monkeypatch.setattr("hermes_state.SessionDB", _factory) + monkeypatch.setattr(server, "_make_agent", _fake_make_agent) + monkeypatch.setattr(server, "_init_session", _fake_init_session) + monkeypatch.setattr(server, "_set_session_context", lambda _target: []) + monkeypatch.setattr(server, "_clear_session_context", lambda _tokens: None) + monkeypatch.setattr(server, "_stored_session_runtime_overrides", lambda _found: {}) + monkeypatch.setattr(server, "_session_info", lambda agent, *a: {"model": "test"}) + + resp = _resume(session_id="s1", profile="work", eager_build=True) + + assert resp["result"]["session_key"] == "s1" + db = profile_dbs[0] + # The agent and the live session both took THIS handle... + assert captured["agent_db"] is db + assert captured["init_db"] is db + # ...so the handler must have released ownership instead of closing it. + assert db.closed == 0 + + +def test_resume_drops_half_built_session_when_init_session_raises( + profile_dbs, monkeypatch +): + """Closing the handle is only safe if the failed registration goes with it. + + ``_init_session`` publishes ``_sessions[sid]`` BEFORE its first read through + the handle. If that read raises, the handle is still ours (and gets closed), + so the half-built session must not stay registered — otherwise the + live-session fast path serves that dead session on every later resume of the + same id, forever, with "'NoneType' object has no attribute 'execute'". + """ + captured: dict = {} + + def _factory(db_path=None, **kwargs): + db = _RecordingDB(db_path=db_path, **kwargs) + db.rows["s1"] = {"id": "s1", "cwd": ""} + profile_dbs.append(db) + return db + + def _fake_init_session(sid, key, agent, history, session_db=None, **_kwargs): + # Same ordering as the real one: register, THEN read through the db. + captured["sid"] = sid + with server._sessions_lock: + server._sessions[sid] = {"agent": agent, "session_key": key} + raise RuntimeError("database is locked") + + monkeypatch.setattr("hermes_state.SessionDB", _factory) + monkeypatch.setattr( + server, "_make_agent", lambda *a, **k: types.SimpleNamespace(model="test") + ) + monkeypatch.setattr(server, "_init_session", _fake_init_session) + monkeypatch.setattr(server, "_set_session_context", lambda _target: []) + monkeypatch.setattr(server, "_clear_session_context", lambda _tokens: None) + monkeypatch.setattr(server, "_stored_session_runtime_overrides", lambda _found: {}) + + resp = _resume(session_id="s1", profile="work", eager_build=True) + + assert resp["error"]["code"] == 5000 + assert profile_dbs[0].closed == 1 + assert captured["sid"] not in server._sessions + + +def test_resume_never_closes_shared_launch_db(profile_dbs, monkeypatch): + """No profile scope -> the shared ``_get_db()`` handle, which we never close.""" + shared = _RecordingDB(db_path="launch") + monkeypatch.setattr(server, "_get_db", lambda: shared) + + resp = _resume(session_id="missing") + + assert resp["error"]["code"] == 4007 + assert profile_dbs == [] # no dedicated handle was opened + assert shared.closed == 0 diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py index c2b4f793deae8..42fb9bae32384 100644 --- a/tui_gateway/methods_session.py +++ b/tui_gateway/methods_session.py @@ -321,185 +321,276 @@ def _(rid, params: dict) -> dict: # the caller explicitly requests it; other clients keep upstream behavior. omit_messages = is_truthy_value(params.get("omit_messages", False)) - # In a profile scope, the agent OWNS a long-lived db handle bound to that - # profile (do NOT auto-close it here). Otherwise reuse the shared launch db. + # In a profile scope this opens a DEDICATED handle we own until the agent + # takes it (see the ownership transfer at _init_session below); every path + # that returns before that transfer must close it. Otherwise reuse the + # shared launch db, which outlives the RPC and is never closed here. + owns_db = False if profile_home is not None: from hermes_state import SessionDB db = SessionDB(db_path=profile_home / "state.db") + owns_db = True else: db = _get_db() - if db is None: - return _db_unavailable_error(rid, code=5000) + try: + if db is None: + return _db_unavailable_error(rid, code=5000) - found = db.get_session(target) - if not found: - found = db.get_session_by_title(target) - if found: - target = found["id"] - elif is_truthy_value(params.get("lazy", False)) and _child_run_active(target): - # Race: a watch window opened on a freshly-spawned subagent. The - # child relays `subagent.start` (which carries child_session_id and - # triggers the window) BEFORE its first run_conversation() flushes - # the DB row via _ensure_db_session, so db.get_session(target) is - # momentarily empty. On slower hosts (notably WSL2, where SQLite + - # process scheduling widen the gap) the window's resume consistently - # lands inside this window and used to hard-fail "session not found" - # — the frontend then 404'd on the REST messages fallback and the - # window spun forever. The child is provably live (_child_run_active), - # so proceed into the lazy branch with empty history; the live mirror - # streams the whole turn anyway and the row exists by upgrade time. - found = {} - else: - return _err(rid, 4007, "session not found") + found = db.get_session(target) + if not found: + found = db.get_session_by_title(target) + if found: + target = found["id"] + elif is_truthy_value(params.get("lazy", False)) and _child_run_active(target): + # Race: a watch window opened on a freshly-spawned subagent. The + # child relays `subagent.start` (which carries child_session_id and + # triggers the window) BEFORE its first run_conversation() flushes + # the DB row via _ensure_db_session, so db.get_session(target) is + # momentarily empty. On slower hosts (notably WSL2, where SQLite + + # process scheduling widen the gap) the window's resume consistently + # lands inside this window and used to hard-fail "session not found" + # — the frontend then 404'd on the REST messages fallback and the + # window spun forever. The child is provably live (_child_run_active), + # so proceed into the lazy branch with empty history; the live mirror + # streams the whole turn anyway and the row exists by upgrade time. + found = {} + else: + return _err(rid, 4007, "session not found") - # Follow the compression-continuation chain to the live tip so a resume on - # a rotated-out parent id binds to the descendant that actually holds the - # post-compression turns. Auto-compression ends the session and forks a - # continuation child; without this, resuming the original id (the desktop's - # routed id when the chat was opened before it rotated) reloads the parent - # transcript and the response generated after compression is missing — the - # "I came back and the reply isn't there" bug on large sessions. Resolving - # here also re-anchors the fast path below so a still-live rotated session - # is reused (by its new key) instead of rebuilding a duplicate agent on the - # stale parent. Skipped for lazy watch windows, which intentionally attach - # to the exact child branch they were opened on. - if found and not is_truthy_value(params.get("lazy", False)): - try: - tip = db.resolve_resume_session_id(target) - except Exception: - tip = target - if tip and tip != target: - target = tip - found = db.get_session(target) or found + # Follow the compression-continuation chain to the live tip so a resume on + # a rotated-out parent id binds to the descendant that actually holds the + # post-compression turns. Auto-compression ends the session and forks a + # continuation child; without this, resuming the original id (the desktop's + # routed id when the chat was opened before it rotated) reloads the parent + # transcript and the response generated after compression is missing — the + # "I came back and the reply isn't there" bug on large sessions. Resolving + # here also re-anchors the fast path below so a still-live rotated session + # is reused (by its new key) instead of rebuilding a duplicate agent on the + # stale parent. Skipped for lazy watch windows, which intentionally attach + # to the exact child branch they were opened on. + if found and not is_truthy_value(params.get("lazy", False)): + try: + tip = db.resolve_resume_session_id(target) + except Exception: + tip = target + if tip and tip != target: + target = tip + found = db.get_session(target) or found - profile_resume_cwd = str(found.get("cwd") or "").strip() or _profile_configured_cwd( - profile_home - ) - - def _reuse_live_payload(sid: str, session: dict) -> dict: - payload = _live_session_payload( - sid, - session, - cols=cols, - touch=True, - transport=current_transport() or _stdio_transport, - omit_messages=omit_messages, + profile_resume_cwd = str(found.get("cwd") or "").strip() or _profile_configured_cwd( + profile_home ) - payload["resumed"] = target - # A lazy watch session never owns a run loop, so its payload's running - # flag is always False — overlay the child-run registry so a reconnecting - # watch window keeps its busy indicator while the child is still mid-run. - if session.get("agent") is None and _child_run_active(target): - payload["running"] = True - payload["status"] = "streaming" - return payload - # Fast path: if the session is already live, reuse it under the lock. - with _session_resume_lock: - live = _find_live_session_by_key(target) - if live is not None: - return _ok(rid, _reuse_live_payload(*live)) - - # Lazy/watch resume: register the live session WITHOUT building an agent. - # Used by the desktop's subagent windows — the child runs inside the - # parent's turn, so its window only needs the stored history plus a - # transport for the child-mirror's live events. Skipping _make_agent here - # is what keeps the window cheap while the backend is busy running the - # delegation. A later prompt.submit upgrades it via _start_agent_build - # (resume_session_id keeps the upgrade on the stored conversation). - if is_truthy_value(params.get("lazy", False)): - sid = uuid.uuid4().hex[:8] - source = _resolve_session_source(str(params.get("source") or "").strip() or None) - lease = None # claimed lazily on the first turn (_ensure_active_session_slot) - try: - db.reopen_session(target) - # The child's OWN conversation only — include_ancestors would prepend - # the parent's transcript onto the subagent's branch. - # repair_alternation: this resume feeds LIVE REPLAY (the loaded - # history becomes the resumed session record's working conversation), - # so heal a durable ``user;user`` violation once here instead of - # re-firing the pre-request repair on every subsequent turn. - history = db.get_messages_as_conversation(target, repair_alternation=True) - except Exception as e: - if lease is not None: - lease.release() - return _err(rid, 5000, f"resume failed: {e}") - cwd = profile_resume_cwd or _default_session_cwd() - record = _deferred_session_record( - target, - cols=cols, - cwd=cwd, - history=history, - lease=lease, - source=source, - close_on_disconnect=is_truthy_value(params.get("close_on_disconnect", False)), - profile_home=profile_home, - lazy=True, - ) - if (live := _claim_or_reuse_live(sid, target, record, lease)) is not None: - return _ok(rid, _reuse_live_payload(*live)) - # A delegated child mid-run emits no session events of its own — report - # its liveness from the relay registry so the window shows a busy turn. - child_running = _child_run_active(target) - # User-visible messages use the VERBATIM display projection (child-only, - # no ancestors — matching the repaired read above), so model-invisible - # rows persisted by #65919 (verification candidates collapsed by - # repair_message_sequence) survive in the watch window just as they do - # on the eager resume + REST paths. The repaired ``history`` above still - # feeds live replay. Fall back to it if the display read fails. - try: - display_history = db.get_messages_as_conversation( - target, repair_alternation=False, include_row_ids=True + def _reuse_live_payload(sid: str, session: dict) -> dict: + payload = _live_session_payload( + sid, + session, + cols=cols, + touch=True, + transport=current_transport() or _stdio_transport, + omit_messages=omit_messages, ) - except Exception: - logger.debug("child-watch display projection read failed", exc_info=True) - display_history = history - messages = [] if omit_messages else _history_to_messages(display_history) - return _ok( - rid, - { + payload["resumed"] = target + # A lazy watch session never owns a run loop, so its payload's running + # flag is always False — overlay the child-run registry so a reconnecting + # watch window keeps its busy indicator while the child is still mid-run. + if session.get("agent") is None and _child_run_active(target): + payload["running"] = True + payload["status"] = "streaming" + return payload + + # Fast path: if the session is already live, reuse it under the lock. + with _session_resume_lock: + live = _find_live_session_by_key(target) + if live is not None: + return _ok(rid, _reuse_live_payload(*live)) + + # Lazy/watch resume: register the live session WITHOUT building an agent. + # Used by the desktop's subagent windows — the child runs inside the + # parent's turn, so its window only needs the stored history plus a + # transport for the child-mirror's live events. Skipping _make_agent here + # is what keeps the window cheap while the backend is busy running the + # delegation. A later prompt.submit upgrades it via _start_agent_build + # (resume_session_id keeps the upgrade on the stored conversation). + if is_truthy_value(params.get("lazy", False)): + sid = uuid.uuid4().hex[:8] + source = _resolve_session_source(str(params.get("source") or "").strip() or None) + lease = None # claimed lazily on the first turn (_ensure_active_session_slot) + try: + db.reopen_session(target) + # The child's OWN conversation only — include_ancestors would prepend + # the parent's transcript onto the subagent's branch. + # repair_alternation: this resume feeds LIVE REPLAY (the loaded + # history becomes the resumed session record's working conversation), + # so heal a durable ``user;user`` violation once here instead of + # re-firing the pre-request repair on every subsequent turn. + history = db.get_messages_as_conversation(target, repair_alternation=True) + except Exception as e: + if lease is not None: + lease.release() + return _err(rid, 5000, f"resume failed: {e}") + cwd = profile_resume_cwd or _default_session_cwd() + record = _deferred_session_record( + target, + cols=cols, + cwd=cwd, + history=history, + lease=lease, + source=source, + close_on_disconnect=is_truthy_value(params.get("close_on_disconnect", False)), + profile_home=profile_home, + lazy=True, + ) + if (live := _claim_or_reuse_live(sid, target, record, lease)) is not None: + return _ok(rid, _reuse_live_payload(*live)) + # A delegated child mid-run emits no session events of its own — report + # its liveness from the relay registry so the window shows a busy turn. + child_running = _child_run_active(target) + # User-visible messages use the VERBATIM display projection (child-only, + # no ancestors — matching the repaired read above), so model-invisible + # rows persisted by #65919 (verification candidates collapsed by + # repair_message_sequence) survive in the watch window just as they do + # on the eager resume + REST paths. The repaired ``history`` above still + # feeds live replay. Fall back to it if the display read fails. + try: + display_history = db.get_messages_as_conversation( + target, repair_alternation=False, include_row_ids=True + ) + except Exception: + logger.debug("child-watch display projection read failed", exc_info=True) + display_history = history + messages = [] if omit_messages else _history_to_messages(display_history) + return _ok( + rid, + { + "session_id": sid, + "resumed": target, + "message_count": len(display_history) if omit_messages else len(messages), + "messages": messages, + "messages_omitted": omit_messages, + "info": _lazy_resume_info(cwd, profile=profile), + "inflight": None, + "running": child_running, + "session_key": target, + "started_at": record["created_at"], + "status": "streaming" if child_running else "idle", + }, + ) + + # Cold resume default: register the live session and read its stored + # transcript, but build the agent OFF the response path. _make_agent can + # block for seconds (MCP discovery, prompt/skill build, AIAgent + # construction), and every resume caller (desktop + Ink TUI) awaits this RPC + # before it paints — so building eagerly is the bulk of the multi-second + # "switching sessions is frozen" latency. Return the full display transcript + # immediately and pre-warm the agent on a short timer (the same deferred- + # build contract session.create uses); _sess() also builds on demand if the + # first prompt beats the timer. A caller that needs the agent built + # synchronously (e.g. tests of the build race) passes ``eager_build: true`` + # to fall through to the eager path below. Distinct from the lazy/watch + # branch above: a normal resume restores the full ancestor history and the + # session's persisted runtime identity, and is a real (upgradable) session. + if not is_truthy_value(params.get("eager_build", False)): + sid = uuid.uuid4().hex[:8] + source = _resolve_session_source(str(params.get("source") or "").strip() or None) + lease = None # claimed lazily on the first turn (_ensure_active_session_slot) + # Interactive resume routes approvals/clarify through gateway prompts; + # the deferred build wires the remaining per-session callbacks. + _enable_gateway_prompts() + try: + db.reopen_session(target) + # One lineage SELECT feeds both projections (#67142-adjacent perf, + # from the desktop audit): the model-fed copy is alternation-repaired + # (raw_history → sanitize_replay_history → the resumed session's + # working conversation) and the display copy stays verbatim — + # inspection/export must show what is actually stored. + if omit_messages: + raw_history = db.get_messages_as_conversation( + target, repair_alternation=True + ) + display_history = [] + else: + raw_history, display_history = db.get_resume_conversations(target) + except Exception as e: + if lease is not None: + lease.release() + return _err(rid, 5000, f"resume failed: {e}") + # Display keeps the full transcript; the model-fed history drops a + # dangling/interrupted tool-call tail so a session killed mid-loop does + # not replay the unanswered call forever (#29086). + prefix = [] if omit_messages else db.get_ancestor_display_prefix(target) + history = sanitize_replay_history(raw_history) + # Restore the model/provider/reasoning/tier this chat last used so the + # deferred build (and the info below) match the eager path — without them + # the build drops the provider ("No LLM provider configured"). + overrides = _stored_session_runtime_overrides(found) or {} + model_override = overrides.get("model_override") or {} + cwd = profile_resume_cwd or _default_session_cwd() + record = _deferred_session_record( + target, + cols=cols, + cwd=cwd, + history=history, + lease=lease, + source=source, + close_on_disconnect=is_truthy_value(params.get("close_on_disconnect", False)), + display_history_prefix=prefix, + profile_home=profile_home, + model_override=overrides.get("model_override"), + resume_runtime_overrides=overrides or None, + ) + if (live := _claim_or_reuse_live(sid, target, record, lease)) is not None: + return _ok(rid, _reuse_live_payload(*live)) + + _schedule_agent_build(sid) + _schedule_session_cap_enforcement() # trim detached idle sessions over the cap + auto_continue = _maybe_schedule_auto_continue(sid, record, target) + + messages = [] if omit_messages else _history_to_messages(display_history) + payload = { "session_id": sid, "resumed": target, - "message_count": len(display_history) if omit_messages else len(messages), + "message_count": len(raw_history) if omit_messages else len(messages), "messages": messages, "messages_omitted": omit_messages, - "info": _lazy_resume_info(cwd, profile=profile), + "info": _lazy_resume_info( + cwd, + model=model_override.get("model") or "", + provider=overrides.get("provider_override") or "", + profile=profile, + ), "inflight": None, - "running": child_running, + "running": False, "session_key": target, "started_at": record["created_at"], - "status": "streaming" if child_running else "idle", - }, - ) + "status": "idle", + } + if auto_continue is not None: + payload["auto_continue"] = auto_continue + return _ok(rid, payload) - # Cold resume default: register the live session and read its stored - # transcript, but build the agent OFF the response path. _make_agent can - # block for seconds (MCP discovery, prompt/skill build, AIAgent - # construction), and every resume caller (desktop + Ink TUI) awaits this RPC - # before it paints — so building eagerly is the bulk of the multi-second - # "switching sessions is frozen" latency. Return the full display transcript - # immediately and pre-warm the agent on a short timer (the same deferred- - # build contract session.create uses); _sess() also builds on demand if the - # first prompt beats the timer. A caller that needs the agent built - # synchronously (e.g. tests of the build race) passes ``eager_build: true`` - # to fall through to the eager path below. Distinct from the lazy/watch - # branch above: a normal resume restores the full ancestor history and the - # session's persisted runtime identity, and is a real (upgradable) session. - if not is_truthy_value(params.get("eager_build", False)): + # Build the agent OUTSIDE the lock — _make_agent can block for seconds + # (MCP discovery, prompt/skill build, AIAgent construction). Holding + # _session_resume_lock across it would stall session.close on the main + # dispatch thread (it's not a _LONG_HANDLER), blocking fast-path RPCs. sid = uuid.uuid4().hex[:8] source = _resolve_session_source(str(params.get("source") or "").strip() or None) lease = None # claimed lazily on the first turn (_ensure_active_session_slot) - # Interactive resume routes approvals/clarify through gateway prompts; - # the deferred build wires the remaining per-session callbacks. _enable_gateway_prompts() + home_token = ( + set_hermes_home_override(str(profile_home)) if profile_home is not None else None + ) + secret_token = ( + set_secret_scope(build_profile_secret_scope(Path(str(profile_home)))) + if profile_home is not None + else None + ) try: db.reopen_session(target) - # One lineage SELECT feeds both projections (#67142-adjacent perf, - # from the desktop audit): the model-fed copy is alternation-repaired - # (raw_history → sanitize_replay_history → the resumed session's - # working conversation) and the display copy stays verbatim — - # inspection/export must show what is actually stored. + # One lineage SELECT feeds both projections (see the interactive resume + # above): the model-fed copy is alternation-repaired for LIVE REPLAY, the + # display copy stays verbatim. if omit_messages: raw_history = db.get_messages_as_conversation( target, repair_alternation=True @@ -507,200 +598,147 @@ def _(rid, params: dict) -> dict: display_history = [] else: raw_history, display_history = db.get_resume_conversations(target) - except Exception as e: - if lease is not None: - lease.release() - return _err(rid, 5000, f"resume failed: {e}") - # Display keeps the full transcript; the model-fed history drops a - # dangling/interrupted tool-call tail so a session killed mid-loop does - # not replay the unanswered call forever (#29086). - prefix = [] if omit_messages else db.get_ancestor_display_prefix(target) - history = sanitize_replay_history(raw_history) - # Restore the model/provider/reasoning/tier this chat last used so the - # deferred build (and the info below) match the eager path — without them - # the build drops the provider ("No LLM provider configured"). - overrides = _stored_session_runtime_overrides(found) or {} - model_override = overrides.get("model_override") or {} - cwd = profile_resume_cwd or _default_session_cwd() - record = _deferred_session_record( - target, - cols=cols, - cwd=cwd, - history=history, - lease=lease, - source=source, - close_on_disconnect=is_truthy_value(params.get("close_on_disconnect", False)), - display_history_prefix=prefix, - profile_home=profile_home, - model_override=overrides.get("model_override"), - resume_runtime_overrides=overrides or None, - ) - if (live := _claim_or_reuse_live(sid, target, record, lease)) is not None: - return _ok(rid, _reuse_live_payload(*live)) - - _schedule_agent_build(sid) - _schedule_session_cap_enforcement() # trim detached idle sessions over the cap - auto_continue = _maybe_schedule_auto_continue(sid, record, target) - - messages = [] if omit_messages else _history_to_messages(display_history) - payload = { - "session_id": sid, - "resumed": target, - "message_count": len(raw_history) if omit_messages else len(messages), - "messages": messages, - "messages_omitted": omit_messages, - "info": _lazy_resume_info( - cwd, - model=model_override.get("model") or "", - provider=overrides.get("provider_override") or "", - profile=profile, - ), - "inflight": None, - "running": False, - "session_key": target, - "started_at": record["created_at"], - "status": "idle", - } - if auto_continue is not None: - payload["auto_continue"] = auto_continue - return _ok(rid, payload) - - # Build the agent OUTSIDE the lock — _make_agent can block for seconds - # (MCP discovery, prompt/skill build, AIAgent construction). Holding - # _session_resume_lock across it would stall session.close on the main - # dispatch thread (it's not a _LONG_HANDLER), blocking fast-path RPCs. - sid = uuid.uuid4().hex[:8] - source = _resolve_session_source(str(params.get("source") or "").strip() or None) - lease = None # claimed lazily on the first turn (_ensure_active_session_slot) - _enable_gateway_prompts() - home_token = ( - set_hermes_home_override(str(profile_home)) if profile_home is not None else None - ) - secret_token = ( - set_secret_scope(build_profile_secret_scope(Path(str(profile_home)))) - if profile_home is not None - else None - ) - try: - db.reopen_session(target) - # One lineage SELECT feeds both projections (see the interactive resume - # above): the model-fed copy is alternation-repaired for LIVE REPLAY, the - # display copy stays verbatim. - if omit_messages: - raw_history = db.get_messages_as_conversation( - target, repair_alternation=True + # The display transcript keeps every row so the user still sees their + # full history. The model-fed history is sanitized: a session whose + # last turn died mid-tool-loop persists a dangling assistant(tool_calls) + # (or interrupted assistant→tool) tail; replaying it makes the model + # re-issue the unanswered call forever — the permanent-"thinking" stuck + # session in #29086. The messaging gateway already strips this; this is + # the WebUI/TUI resume path picking up the same cleanup. + display_history_prefix = ( + [] if omit_messages else db.get_ancestor_display_prefix(target) ) - display_history = [] - else: - raw_history, display_history = db.get_resume_conversations(target) - # The display transcript keeps every row so the user still sees their - # full history. The model-fed history is sanitized: a session whose - # last turn died mid-tool-loop persists a dangling assistant(tool_calls) - # (or interrupted assistant→tool) tail; replaying it makes the model - # re-issue the unanswered call forever — the permanent-"thinking" stuck - # session in #29086. The messaging gateway already strips this; this is - # the WebUI/TUI resume path picking up the same cleanup. - display_history_prefix = ( - [] if omit_messages else db.get_ancestor_display_prefix(target) - ) - history = sanitize_replay_history(raw_history) - messages = [] if omit_messages else _history_to_messages(display_history) - tokens = _set_session_context(target) - try: - # Pass the profile's db so the agent persists turns to the right - # state.db; home override is active here so config/skills/model - # resolve to the profile too. Runtime identity is restored from the - # stored session row so switching chats does not inherit whatever - # global model another chat last selected. - stored_runtime_overrides = _stored_session_runtime_overrides(found) - agent = _make_agent( - sid, - target, - session_id=target, - session_db=db, - platform_override=source, - **stored_runtime_overrides, - ) - finally: - _clear_session_context(tokens) - except Exception as e: - if lease is not None: - lease.release() - return _err(rid, 5000, f"resume failed: {e}") - finally: - if home_token is not None: - reset_hermes_home_override(home_token) - if secret_token is not None: - reset_secret_scope(secret_token) - - # Double-checked locking: another concurrent resume may have created the - # live session while we were building. Re-check under the lock; if it won, - # discard our just-built agent and reuse theirs (no worker/poller wired yet). - with _session_resume_lock: - live = _find_live_session_by_key(target) - if live is not None: + history = sanitize_replay_history(raw_history) + messages = [] if omit_messages else _history_to_messages(display_history) + tokens = _set_session_context(target) try: - if hasattr(agent, "close"): - agent.close() - except Exception: - pass - if lease is not None: - lease.release() - other_sid, other_session = live - payload = _live_session_payload( - other_sid, - other_session, - cols=cols, - touch=True, - transport=current_transport() or _stdio_transport, - omit_messages=omit_messages, - ) - payload["resumed"] = target - return _ok(rid, payload) - try: - init_home_token = ( - set_hermes_home_override(str(profile_home)) - if profile_home is not None - else None - ) - init_secret_token = ( - set_secret_scope(build_profile_secret_scope(Path(str(profile_home)))) - if profile_home is not None - else None - ) - try: - _init_session( + # Pass the profile's db so the agent persists turns to the right + # state.db; home override is active here so config/skills/model + # resolve to the profile too. Runtime identity is restored from the + # stored session row so switching chats does not inherit whatever + # global model another chat last selected. + stored_runtime_overrides = _stored_session_runtime_overrides(found) + agent = _make_agent( sid, target, - agent, - history, - cols=cols, - cwd=profile_resume_cwd, + session_id=target, session_db=db, - source=source, + platform_override=source, + **stored_runtime_overrides, ) finally: - if init_home_token is not None: - reset_hermes_home_override(init_home_token) - if init_secret_token is not None: - reset_secret_scope(init_secret_token) - if sid in _sessions: - if stored_runtime_overrides.get("model_override") is not None: - _sessions[sid]["model_override"] = stored_runtime_overrides[ - "model_override" - ] - _sessions[sid]["display_history_prefix"] = display_history_prefix - # Remember the profile home so each turn re-binds HERMES_HOME (the - # agent persists to its own db, but mid-turn home reads — memory, - # skills — must resolve to the resumed profile too). - if profile_home is not None: - _sessions[sid]["profile_home"] = str(profile_home) - _sessions[sid]["active_session_lease"] = lease + _clear_session_context(tokens) except Exception as e: if lease is not None: lease.release() return _err(rid, 5000, f"resume failed: {e}") - session = _sessions.get(sid) or {} + finally: + if home_token is not None: + reset_hermes_home_override(home_token) + if secret_token is not None: + reset_secret_scope(secret_token) + + # Double-checked locking: another concurrent resume may have created the + # live session while we were building. Re-check under the lock; if it won, + # discard our just-built agent and reuse theirs (no worker/poller wired yet). + with _session_resume_lock: + live = _find_live_session_by_key(target) + if live is not None: + try: + if hasattr(agent, "close"): + agent.close() + except Exception: + pass + if lease is not None: + lease.release() + other_sid, other_session = live + payload = _live_session_payload( + other_sid, + other_session, + cols=cols, + touch=True, + transport=current_transport() or _stdio_transport, + omit_messages=omit_messages, + ) + payload["resumed"] = target + return _ok(rid, payload) + try: + init_home_token = ( + set_hermes_home_override(str(profile_home)) + if profile_home is not None + else None + ) + init_secret_token = ( + set_secret_scope(build_profile_secret_scope(Path(str(profile_home)))) + if profile_home is not None + else None + ) + try: + _init_session( + sid, + target, + agent, + history, + cols=cols, + cwd=profile_resume_cwd, + session_db=db, + source=source, + ) + # Ownership TRANSFER — the registered session's agent now + # holds this handle for its whole life, and _init_session + # never closes a caller-supplied session_db (its + # _init_owns_db stays False). Closing it in the finally + # below would fault every later turn on this session with + # "Cannot operate on a closed database". + owns_db = False + finally: + if init_home_token is not None: + reset_hermes_home_override(init_home_token) + if init_secret_token is not None: + reset_secret_scope(init_secret_token) + if sid in _sessions: + if stored_runtime_overrides.get("model_override") is not None: + _sessions[sid]["model_override"] = stored_runtime_overrides[ + "model_override" + ] + _sessions[sid]["display_history_prefix"] = display_history_prefix + # Remember the profile home so each turn re-binds HERMES_HOME (the + # agent persists to its own db, but mid-turn home reads — memory, + # skills — must resolve to the resumed profile too). + if profile_home is not None: + _sessions[sid]["profile_home"] = str(profile_home) + _sessions[sid]["active_session_lease"] = lease + except Exception as e: + # _init_session registers _sessions[sid] BEFORE its first read + # through this handle. If it raised in between — "database is + # locked" is the realistic trigger — the half-built session is + # still registered while the finally below closes the handle it + # holds, and the live-session fast path above would then serve + # that dead session on every later resume of this id + # ("'NoneType' object has no attribute 'execute'", permanently). + # owns_db still True means ownership never transferred, so the + # registration is ours to undo. + if owns_db: + with _sessions_lock: + _sessions.pop(sid, None) + if lease is not None: + lease.release() + return _err(rid, 5000, f"resume failed: {e}") + session = _sessions.get(sid) or {} + finally: + # Every return that does NOT reach the transfer above abandons this + # handle — session-not-found, both "resume failed" paths, the live-session + # fast path (the hot one: reconnects re-resume live chats through it), the + # deferred cold/lazy returns, and the double-checked-locking discard. + # Dropping it merely relied on refcounting to release the sqlite fds; that + # stops being true the moment anything pins the instance — SessionDB pins + # ITSELF once its background token writer starts, via + # atexit.register(_drain_token_queue_at_exit) (hermes_state.py), which only + # close() unregisters. A pinned handle keeps its db/-wal/-shm fds and its + # writer thread for the life of the process. + if owns_db and db is not None: + with contextlib.suppress(Exception): + db.close() auto_continue = ( _maybe_schedule_auto_continue(sid, session, target) if session else None )