tui_gateway: session.resume abandons the profile SessionDB it opens
This commit is contained in:
Yishova
kshitij
parent
fecba5afcc
commit
79625e3c0b
|
|
@ -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=<profile>/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
|
||||
|
|
@ -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
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in New Issue