ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(tui): session_lifecycle/workdir/reaper — fold single-use helpers, unify daemon-timer + memo, collapse defensive layers (-12% LOC)

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