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

Merge branch 'simp/r3-12-g4-session' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:59:31 -07:00
parent 30e714cdb1 03e70a58c3
commit 896a2b1494
9 changed files with 1228 additions and 2292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/session.py
+468 -806
View File
File diff suppressed because it is too large Load Diff
gateway/session_context.py
+96 -269
View File
@@ -1,32 +1,20 @@
"""Session-scoped context variables for the Hermes gateway.
Replaces the old ``os.environ``-based session state (``HERMES_SESSION_*``)
with ``contextvars.ContextVar``. The gateway processes messages concurrently
via asyncio; ``os.environ`` is process-global, so message B silently
overwrote message A's thread id before A's agent finished and notifications
routed to the wrong thread. ContextVar values are task-local (inherited by
``run_in_executor`` threads), so concurrent messages never interfere.
``get_session_env(name, default="")`` is a drop-in for
``os.getenv("HERMES_SESSION_*", default)`` at existing tool call sites.
Replaces the old ``os.environ``-based ``HERMES_SESSION_*`` state with task-local ``ContextVar``s
(inherited by ``run_in_executor`` threads), so concurrently handled messages no longer clobber each
other's routing ids. ``get_session_env`` is a drop-in for ``os.getenv``.
"""
import os
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any, Iterator
# Distinguishes "never set in this context" (fall back to os.environ for
# CLI/cron compat) from "explicitly set to empty" by clear_session_vars (no fallback).
# "Never set here" (falls back to os.environ for CLI/cron) vs "" = explicitly cleared (no fallback).
_UNSET: Any = object()
# Process-level monotonic latch: has any code in this process bound a session via
# set_session_vars()? Concurrent multi-session hosts (gateway, ACP, API server,
# TUI, cron) do; a pure single-process CLI/one-shot does not. The subprocess-env
# bridge (tools/environments/local.py) reads this to pick its leak policy: when
# engaged, the ContextVars are authoritative and an _UNSET var means "no session
# bound in THIS task", so the last-writer-wins os.environ mirror must NOT be
# inherited by a child process. When never engaged, the os.environ fallback is
# preserved (no concurrency to leak across).
# Process-level latch: has set_session_vars() ever bound a session? When engaged, the subprocess
# env bridge treats ContextVars as authoritative and an _UNSET var as "no session in THIS task".
_session_context_engaged: bool = False
@@ -35,121 +23,64 @@ def session_context_engaged() -> bool:
return _session_context_engaged
def _var(name: str) -> ContextVar:
return ContextVar(name, default=_UNSET)
# --- Per-task session variables --------------------------------------------
_SESSION_PLATFORM = _var("HERMES_SESSION_PLATFORM")
_SESSION_SOURCE = _var("HERMES_SESSION_SOURCE")
_SESSION_CHAT_ID = _var("HERMES_SESSION_CHAT_ID")
_SESSION_CHAT_TYPE = _var("HERMES_SESSION_CHAT_TYPE")
_SESSION_CHAT_NAME = _var("HERMES_SESSION_CHAT_NAME")
_SESSION_THREAD_ID = _var("HERMES_SESSION_THREAD_ID")
_SESSION_USER_ID = _var("HERMES_SESSION_USER_ID")
_SESSION_USER_ID_ALT = _var("HERMES_SESSION_USER_ID_ALT")
_SESSION_USER_NAME = _var("HERMES_SESSION_USER_NAME")
# Platform-neutral scope discriminator (Discord guild / Slack workspace / Matrix
# server). Captured at bind time so async producers (delegate_task
# background=True, terminal watchers) can persist a completion's full routing
# origin: a relay connector's fail-closed egress guard needs scope_id (or a user
# binding) to resolve the tenant for a scoped reply after a restart.
_SESSION_SCOPE_ID = _var("HERMES_SESSION_SCOPE_ID")
_SESSION_KEY = _var("HERMES_SESSION_KEY")
_SESSION_ID = _var("HERMES_SESSION_ID")
# In-process UI tab/window id for multi-session desktop/TUI hosts — deliberately
# separate from the durable HERMES_SESSION_ID. Background completions use it as
# a precise return address so a stale/rotated durable key cannot be consumed by
# whichever desktop poller wakes first.
_SESSION_UI_SESSION_ID = _var("HERMES_UI_SESSION_ID")
# Triggering message id: reply anchor so background notifications stay inside
# the originating Telegram private-chat topic (routes only with thread id + anchor).
_SESSION_MESSAGE_ID = _var("HERMES_SESSION_MESSAGE_ID")
_SESSION_PROFILE = _var("HERMES_SESSION_PROFILE")
_BROWSER_CONTROL_PRINCIPAL = _var("HERMES_BROWSER_CONTROL_PRINCIPAL")
_BROWSER_CONTROL_TRANSPORT_FAMILY = _var("HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY")
# Per-session cron marker, tri-state: _UNSET keeps the legacy env fallback for
# CLI/tests; "1" marks cron; "" explicitly marks non-cron and masks leaked env.
_CRON_SESSION = _var("HERMES_CRON_SESSION")
# Whether this session's channel can route an ASYNC completion back to the agent
# AFTER the turn ends (wake a fresh turn). True for long-lived CLI sessions and
# real gateway platforms (persistent outbound channel + watcher/drain loops);
# False for finite runtimes that may exit before a detached completion returns
# (stateless API-server requests, dispatcher-spawned Kanban workers). Tools that
# promise async delivery (terminal notify_on_complete / watch_patterns,
# delegate_task background=True) read ``async_delivery_supported()`` and refuse a
# promise the channel can't keep. Default _UNSET => supported, so the CLI (never
# sets a platform) and contextvar-unaware paths keep working; stateless adapters
# opt OUT via ``supports_async_delivery = False`` on the adapter class, which the
# gateway propagates here at session-bind time.
_SESSION_ASYNC_DELIVERY = _var("HERMES_SESSION_ASYNC_DELIVERY")
# Cron auto-delivery vars — set per-job in run_job() so concurrent jobs don't
# clobber each other's delivery targets.
_CRON_AUTO_DELIVER_PLATFORM = _var("HERMES_CRON_AUTO_DELIVER_PLATFORM")
_CRON_AUTO_DELIVER_CHAT_ID = _var("HERMES_CRON_AUTO_DELIVER_CHAT_ID")
_CRON_AUTO_DELIVER_THREAD_ID = _var("HERMES_CRON_AUTO_DELIVER_THREAD_ID")
# Vars bound by set_session_vars / cleared to "" by clear_session_vars, in order.
# --- Per-task session variables: bound by set_session_vars / cleared to "" by clear_session_vars;
# tuple ORDER is the positional order of ``values`` in set_session_vars (zipped).
# * SCOPE_ID: platform-neutral scope (guild / workspace / Matrix server) so async producers can
# persist a completion's full routing origin (relay egress guards need it).
# * UI_SESSION_ID: in-process UI tab id, separate from the durable SESSION_ID, so a stale/rotated
# durable key is not consumed by the wrong poller.
# * MESSAGE_ID: reply anchor keeping notifications inside the originating Telegram topic.
# * CRON_SESSION: tri-state — _UNSET = legacy env fallback; "1" = cron; "" = non-cron, masks env.
_SESSION_VARS = (
_SESSION_PLATFORM,
_SESSION_SOURCE,
_SESSION_CHAT_ID,
_SESSION_CHAT_TYPE,
_SESSION_CHAT_NAME,
_SESSION_THREAD_ID,
_SESSION_USER_ID,
_SESSION_USER_ID_ALT,
_SESSION_USER_NAME,
_SESSION_SCOPE_ID,
_SESSION_KEY,
_SESSION_ID,
_SESSION_UI_SESSION_ID,
_SESSION_MESSAGE_ID,
_SESSION_PROFILE,
_BROWSER_CONTROL_PRINCIPAL,
_BROWSER_CONTROL_TRANSPORT_FAMILY,
_CRON_SESSION,
)
_SESSION_PLATFORM, _SESSION_SOURCE, _SESSION_CHAT_ID, _SESSION_CHAT_TYPE,
_SESSION_CHAT_NAME, _SESSION_THREAD_ID, _SESSION_USER_ID, _SESSION_USER_ID_ALT,
_SESSION_USER_NAME, _SESSION_SCOPE_ID, _SESSION_KEY, _SESSION_ID,
_SESSION_UI_SESSION_ID, _SESSION_MESSAGE_ID, _SESSION_PROFILE,
_BROWSER_CONTROL_PRINCIPAL, _BROWSER_CONTROL_TRANSPORT_FAMILY, _CRON_SESSION,
) = tuple(ContextVar(name, default=_UNSET) for name in (
"HERMES_SESSION_PLATFORM", "HERMES_SESSION_SOURCE", "HERMES_SESSION_CHAT_ID",
"HERMES_SESSION_CHAT_TYPE", "HERMES_SESSION_CHAT_NAME", "HERMES_SESSION_THREAD_ID",
"HERMES_SESSION_USER_ID", "HERMES_SESSION_USER_ID_ALT", "HERMES_SESSION_USER_NAME",
"HERMES_SESSION_SCOPE_ID", "HERMES_SESSION_KEY", "HERMES_SESSION_ID",
"HERMES_UI_SESSION_ID", "HERMES_SESSION_MESSAGE_ID", "HERMES_SESSION_PROFILE",
"HERMES_BROWSER_CONTROL_PRINCIPAL", "HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY",
"HERMES_CRON_SESSION",
))
# Legacy env-var name -> ContextVar, for get_session_env. _SESSION_ASYNC_DELIVERY
# is deliberately absent: it is a bool capability read via async_delivery_supported.
_VAR_MAP = {
var.name: var
for var in (*_SESSION_VARS, _CRON_AUTO_DELIVER_PLATFORM, _CRON_AUTO_DELIVER_CHAT_ID, _CRON_AUTO_DELIVER_THREAD_ID)
}
# Whether this channel can route an ASYNC completion back AFTER the turn ends (see
# ``async_delivery_supported()``). _UNSET => supported (CLI, contextvar-unaware paths); stateless
# adapters (API server, Kanban workers) opt OUT via ``supports_async_delivery = False`` at bind.
_SESSION_ASYNC_DELIVERY = ContextVar("HERMES_SESSION_ASYNC_DELIVERY", default=_UNSET)
# Cron auto-delivery vars, set per-job in run_job() so concurrent jobs don't clobber.
_CRON_AUTO_DELIVER_PLATFORM = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET)
_CRON_AUTO_DELIVER_CHAT_ID = ContextVar("HERMES_CRON_AUTO_DELIVER_CHAT_ID", default=_UNSET)
_CRON_AUTO_DELIVER_THREAD_ID = ContextVar("HERMES_CRON_AUTO_DELIVER_THREAD_ID", default=_UNSET)
# Legacy env-var name -> ContextVar for get_session_env (_SESSION_ASYNC_DELIVERY deliberately
# absent: it is a bool capability, read via async_delivery_supported).
_VAR_MAP = {var.name: var for var in (
*_SESSION_VARS, _CRON_AUTO_DELIVER_PLATFORM, _CRON_AUTO_DELIVER_CHAT_ID,
_CRON_AUTO_DELIVER_THREAD_ID,
)}
def _clear_session_cwd() -> None:
def _runtime_cwd(func: str, *args: Any) -> None:
"""Best-effort call of ``agent.runtime_cwd.<func>``; import/runtime failures are ignored."""
try:
from agent.runtime_cwd import clear_session_cwd
clear_session_cwd()
from agent import runtime_cwd
getattr(runtime_cwd, func)(*args)
except Exception:
pass
def set_current_session_id(session_id: str) -> None:
"""Synchronize ``HERMES_SESSION_ID`` across ContextVar and ``os.environ``.
Long-lived single-process entrypoints (CLI) rotate sessions via /new,
/resume, /branch, or compression splits without rebuilding the agent; tools
read ``get_session_env("HERMES_SESSION_ID")`` with an os.environ fallback,
so both stores must move together.
Delegated subagent children are the exception: they are built in the parent
process inside ``delegated_child_context()`` and their ``AIAgent.__init__``
calls this helper. Writing the child's id to process-global ``os.environ``
would clobber the parent's id for the rest of the process, so only the
task-local ContextVar write happens for them. Root agents keep both paths.
"""
import os
"""Synchronize ``HERMES_SESSION_ID`` across ContextVar and ``os.environ`` (tools read it
with an os.environ fallback). Delegated subagent children (built in the parent process)
get ONLY the task-local write, or they would clobber the parent's id."""
_SESSION_ID.set(session_id)
try:
from agent.delegation_context import is_delegated_child_context
if is_delegated_child_context():
return
except Exception:
@@ -159,12 +90,8 @@ def set_current_session_id(session_id: str) -> None:
@contextmanager
def scoped_current_session_id(session_id: str | None = None) -> Iterator[None]:
"""Bind a task-local session id and restore the prior value on exit.
``session_id=None`` makes this a pure save/restore boundary around code that
may call :func:`set_current_session_id` itself (delegated ``AIAgent``
construction). Never mutates ``os.environ``.
"""
"""Bind a task-local session id and restore the prior value on exit; never touches
``os.environ``. ``session_id=None`` is a pure save/restore boundary."""
previous = _SESSION_ID.get()
if session_id is not None:
_SESSION_ID.set(session_id)
@@ -175,188 +102,88 @@ def scoped_current_session_id(session_id: str | None = None) -> Iterator[None]:
def set_session_vars(
platform: str = "",
source: str = "",
chat_id: str = "",
chat_type: str = "",
chat_name: str = "",
thread_id: str = "",
user_id: str = "",
user_id_alt: str = "",
user_name: str = "",
scope_id: str = "",
session_key: str = "",
session_id: str = "",
message_id: str = "",
profile: str = "",
browser_control_principal: str = "",
browser_control_transport_family: str = "",
cwd: str = "",
async_delivery: bool = True,
ui_session_id: str = "",
cron_session: Any = _UNSET,
platform: str = "", source: str = "", chat_id: str = "", chat_type: str = "",
chat_name: str = "", thread_id: str = "", user_id: str = "", user_id_alt: str = "",
user_name: str = "", scope_id: str = "", session_key: str = "", session_id: str = "",
message_id: str = "", profile: str = "", browser_control_principal: str = "",
browser_control_transport_family: str = "", cwd: str = "", async_delivery: bool = True,
ui_session_id: str = "", cron_session: Any = _UNSET,
) -> list:
"""Set all session context variables and return reset tokens.
Call ``clear_session_vars(tokens)`` in a ``finally`` when the handler exits.
These helpers are not nestable: clearing resets every var to ``""`` rather
than restoring prior values, and the tokens are accepted only for API compat.
``cwd`` pins the logical working directory. ``async_delivery`` declares
whether the channel can route a background completion back after the turn
(stateless adapters such as the API server pass ``False``). ``cron_session``
is tri-state; see ``_CRON_SESSION``.
"""
# Latch the process as engaged — see _session_context_engaged.
"""Set all session context variables and return reset tokens. Call
``clear_session_vars(tokens)`` in a ``finally``; not nestable, clearing resets every var
to ``""`` rather than restoring prior values (tokens are accepted only for API compat)."""
global _session_context_engaged
_session_context_engaged = True
values = (
platform, source, chat_id, chat_type, chat_name, thread_id, user_id,
user_id_alt, user_name, scope_id, session_key, session_id, ui_session_id,
message_id, profile, browser_control_principal,
browser_control_transport_family, cron_session,
platform, source, chat_id, chat_type, chat_name, thread_id, user_id, user_id_alt,
user_name, scope_id, session_key, session_id, ui_session_id, message_id, profile,
browser_control_principal, browser_control_transport_family, cron_session,
)
tokens = [var.set(value) for var, value in zip(_SESSION_VARS, values)]
tokens.append(_SESSION_ASYNC_DELIVERY.set(bool(async_delivery)))
try:
from agent.runtime_cwd import set_session_cwd
set_session_cwd(cwd)
except Exception:
pass
_runtime_cwd("set_session_cwd", cwd)
return tokens
def clear_session_vars(tokens: list) -> None:
"""Mark session context variables as explicitly cleared.
Sets every var to ``""`` (not ``var.reset(token)``) so ``get_session_env``
returns empty instead of falling back to stale ``os.environ`` values while
staying distinguishable from "never set" (``_UNSET``). Async-delivery is
reset to ``_UNSET`` rather than a falsy value: a cleared context must fall
back to default-supported, not look like an opted-out stateless adapter.
"""
"""Mark session context variables as explicitly cleared (``""``, not ``_UNSET``), so
``get_session_env`` returns empty instead of stale ``os.environ`` values. Async-delivery
goes back to ``_UNSET``: a cleared context is default-supported, not opted-out."""
for var in _SESSION_VARS:
var.set("")
_SESSION_ASYNC_DELIVERY.set(_UNSET)
_clear_session_cwd()
_runtime_cwd("clear_session_cwd")
def reset_session_vars() -> None:
"""Reset every session context variable to ``_UNSET`` for THIS context.
Unlike :func:`clear_session_vars` (``""`` = "explicitly cleared", used when a
handler *finishes*), this restores "never bound here" — what a freshly
spawned task should look like *before* binding its own session.
Why: ``create_task`` snapshots the current context, so message B's task can
inherit message A's already-**set** vars. Until B binds its own session,
any subprocess it spawns reads A's identity through the subprocess-env
bridge — whose _UNSET-strip guard cannot help because the vars are set-to-A.
Calling this at the top of the per-message handler makes that window strip
safe (no session) instead of leaking the foreign one. See
tests/tools/test_local_env_session_leak.py and
tests/gateway/test_session_context_inheritance.py.
``_SESSION_ASYNC_DELIVERY`` is reset explicitly (it lives outside
``_VAR_MAP``): otherwise a task spawned from a context where a sibling
adapter bound ``async_delivery=False`` inherits that ``False`` through the
pre-bind window and misreports the new channel as unable to deliver.
"""
"""Reset every session var to ``_UNSET`` ("never bound here") for THIS context. Call at
the top of a fresh task *before* it binds: ``create_task`` snapshots the context, so B's
task inherits A's already-set vars and a subprocess spawned before B binds would read A's
identity. ``_SESSION_ASYNC_DELIVERY`` (outside ``_VAR_MAP``) is reset explicitly too."""
for var in _VAR_MAP.values():
var.set(_UNSET)
_SESSION_ASYNC_DELIVERY.set(_UNSET)
_clear_session_cwd()
_runtime_cwd("clear_session_cwd")
def get_session_env(name: str, default: str = "") -> str:
"""Read a session context variable by its legacy ``HERMES_SESSION_*`` name.
Drop-in for ``os.getenv(name, default)``. Resolution: the ContextVar if it
was ever set in this context (even to ``""`` — no fallback); else
``os.environ`` (CLI, cron scheduler, tests that never bind); else *default*.
"""
import os
"""Read a session var by legacy ``HERMES_SESSION_*`` name; drop-in for os.getenv. The
ContextVar wins if ever set here (even to ``""``); else ``os.environ``; else *default*."""
var = _VAR_MAP.get(name)
if var is not None and (value := var.get()) is not _UNSET:
return value
return os.getenv(name, default)
# Surfaces that are not a human chat channel. The gateway binds a platform value
# (``telegram``) to HERMES_SESSION_PLATFORM while the CLI/TUI/desktop bind
# HERMES_SESSION_SOURCE and leave platform empty, so both are consulted.
# ``local``, ``api_server``, ``webhook``, ``msgraph_webhook`` are real Platform
# values with no attachment channel behind them. Default-deny: an unrecognized
# identity counts as messaging so a new chat platform is never treated as a
# private surface before this set is updated. Mirrors LOCAL_SESSION_SOURCE_IDS
# in apps/desktop/src/lib/session-source.ts; keep roughly in sync.
NON_MESSAGING_SESSION_SURFACES = frozenset(
{
"",
"api_server",
"cli",
"codex",
"desktop",
"gateway",
"kanban",
"local",
"msgraph_webhook",
"tool",
"tui",
"webhook",
}
)
# Surfaces that are not a human chat channel (gateway binds HERMES_SESSION_PLATFORM, CLI/TUI/
# desktop bind HERMES_SESSION_SOURCE, so both are consulted). Default-deny: an unrecognized
# identity counts as messaging. Mirrors LOCAL_SESSION_SOURCE_IDS in apps/desktop session-source.ts.
NON_MESSAGING_SESSION_SURFACES = frozenset({
"", "api_server", "cli", "codex", "desktop", "gateway", "kanban", "local",
"msgraph_webhook", "tool", "tui", "webhook",
})
def session_is_messaging_surface() -> bool:
"""Whether this turn is delivered over a human messaging channel.
Decides "user is reading a chat message" vs "user is at a machine they own":
delivery tags, whether a file must land somewhere the gateway can send from,
whether narration reads as chat noise. Checks ``HERMES_PLATFORM``, then the
session platform, then the session source against
:data:`NON_MESSAGING_SESSION_SURFACES`.
"""
import os
"""Whether this turn is delivered over a human messaging channel (checks
``HERMES_PLATFORM``, then the session platform, then the session source)."""
platform = os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM", "")
source = get_session_env("HERMES_SESSION_SOURCE", "")
return any(
(ident := str(identity or "").strip().lower()) and ident not in NON_MESSAGING_SESSION_SURFACES
for identity in (platform, source)
)
idents = (platform, get_session_env("HERMES_SESSION_SOURCE", ""))
idents = (str(v or "").strip().lower() for v in idents)
return any(ident and ident not in NON_MESSAGING_SESSION_SURFACES for ident in idents)
def declare_stateless_channel() -> None:
"""Declare that this session cannot receive an async background completion.
Binds only the delivery capability. Use this instead of
``set_session_vars(async_delivery=False)`` on a pure single-process runner:
``set_session_vars`` also latches ``_session_context_engaged``, which flips
the subprocess env bridge to ContextVar-authoritative — a one-shot CLI must
not flip that latch as a side effect of declaring a capability. Callers that
build a full context (cron's ``run_job``) pass ``async_delivery=False``.
``delegate_task`` then falls through to its inline path so results return
within the turn instead of going to a channel that never delivers.
"""
"""Declare that this session cannot receive an async background completion. Unlike
``set_session_vars(async_delivery=False)`` this does NOT latch ``_session_context_engaged``
(flipping the subprocess env bridge), which a one-shot CLI must not do as a side effect."""
_SESSION_ASYNC_DELIVERY.set(False)
def async_delivery_supported() -> bool:
"""Whether the current session can deliver a background completion later.
False for finite runtimes: sessions bound by a stateless channel (API
server, ``hermes -z``, cron — see :func:`declare_stateless_channel`) and
dispatcher-spawned Kanban workers (``HERMES_KANBAN_TASK``), which are
one-shot ``chat -q`` subprocesses whose parent disappears after the quiet
turn, so a later completion has no durable consumer. Gateway platforms,
the interactive CLI, and any path that never bound the var return True.
"""
import os
# Kanban worker: force tools onto their synchronous/polling fallbacks.
"""Whether the current session can deliver a background completion later. False for
stateless channels (:func:`declare_stateless_channel`) and Kanban workers
(``HERMES_KANBAN_TASK``: one-shot subprocesses whose parent disappears after the turn)."""
if os.environ.get("HERMES_KANBAN_TASK"):
return False
value = _SESSION_ASYNC_DELIVERY.get()
gateway/session_db_recovery.py
+15 -34
View File
@@ -2,14 +2,13 @@
from __future__ import annotations
import contextlib
import threading
import time
import weakref
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
import contextlib
_INITIAL_RETRY_DELAY_SECONDS = 1.0
_MAX_RETRY_DELAY_SECONDS = 60.0
@@ -27,9 +26,8 @@ class _HealthSource:
_health_lock = threading.Lock()
_health_states: weakref.WeakKeyDictionary[_HealthSource, dict[Path, str]] = (
weakref.WeakKeyDictionary()
)
_health_states: weakref.WeakKeyDictionary[_HealthSource, dict[Path, str]]
_health_states = weakref.WeakKeyDictionary()
def _publish_health(source: _HealthSource, path: Path, state: str) -> None:
@@ -40,7 +38,6 @@ def _publish_health(source: _HealthSource, path: Path, state: str) -> None:
aggregate = next((s for s in ("retrying", "unavailable") if s in all_states), "ok")
try:
from gateway.status import write_runtime_status
write_runtime_status(session_store={"status": aggregate})
except Exception:
pass # Runtime health is diagnostic only; persistence must not depend on it.
@@ -49,16 +46,12 @@ def _publish_health(source: _HealthSource, path: Path, state: str) -> None:
class RecoverableHandleCache:
"""Cache handles by path while allowing failed opens to heal in-process.
Opens run OUTSIDE ``lock`` (single-flight per path via ``in_flight``); a
``close_all`` bumps ``_generation`` so any open that completes afterwards is
treated as stale and rejected rather than resurrecting a drained cache.
Opens run OUTSIDE ``lock`` (single-flight per path via ``in_flight``); ``close_all`` bumps
``_generation`` so a later-completing open is rejected instead of resurrecting the cache.
"""
def __init__(
self,
*,
handles: dict[Path, Any] | None = None,
lock: threading.Lock | None = None,
self, *, handles: dict[Path, Any] | None = None, lock: threading.Lock | None = None,
clock: Callable[[], float] = time.monotonic,
initial_retry_delay: float = _INITIAL_RETRY_DELAY_SECONDS,
max_retry_delay: float = _MAX_RETRY_DELAY_SECONDS,
@@ -78,20 +71,13 @@ class RecoverableHandleCache:
return generation != self._generation or self._unavailable.get(path) is not unavailable
def get(
self,
path: Path,
opener: Callable[[], Any],
*,
raise_on_error: bool = False,
self, path: Path, opener: Callable[[], Any], *, raise_on_error: bool = False,
on_recovered: Callable[[], None] | None = None,
non_cacheable: Callable[[Exception], bool] | None = None,
) -> Any:
"""Return a cached handle or make one bounded, single-flight open attempt.
Returns None while a retry is in flight or backing off (callers fall back).
``non_cacheable`` exceptions (e.g. a live-system guard) are re-raised
without recording a failure so the next call retries immediately.
"""
"""Return a cached handle or make one bounded, single-flight open attempt; None while
a retry is in flight or backing off. ``non_cacheable`` exceptions (e.g. a live-system
guard) are re-raised without recording a failure so the next call retries at once."""
path = Path(path)
with self.lock:
if path in self.handles:
@@ -102,7 +88,6 @@ class RecoverableHandleCache:
unavailable.in_flight = True
was_unavailable = unavailable.failures > 0
generation = self._generation
if was_unavailable:
_publish_health(self._health_source, path, "retrying")
@@ -118,11 +103,8 @@ class RecoverableHandleCache:
raise
if not stale:
unavailable.failures += 1
delay = min(
self._initial_retry_delay * (2 ** min(unavailable.failures - 1, 30)),
self._max_retry_delay,
)
unavailable.next_retry_at = self._clock() + delay
backoff = self._initial_retry_delay * (2 ** min(unavailable.failures - 1, 30))
unavailable.next_retry_at = self._clock() + min(backoff, self._max_retry_delay)
unavailable.in_flight = False
if not stale:
_publish_health(self._health_source, path, "unavailable")
@@ -159,7 +141,6 @@ class RecoverableHandleCache:
with contextlib.suppress(Exception):
close(handle)
with _health_lock:
states = _health_states.get(self._health_source)
if states is not None:
for path in paths:
states.pop(path, None)
states = _health_states.get(self._health_source, {})
for path in paths:
states.pop(path, None)
gateway/session_lifecycle.py
+175 -225
View File
@@ -1,13 +1,11 @@
"""SessionStore reset/expiry policy and crash-recovery markers: idle/daily reset
evaluation, expiry finalization, active-turn tokens, resume_pending,
suspension and pruning.
Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.
"""
"""SessionStore reset/expiry policy and crash-recovery markers (idle/daily reset, expiry
finalization, active-turn tokens, resume_pending, suspension, pruning) plus the shared clock/id
helpers. Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO."""
from __future__ import annotations
import logging
import os
import uuid
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Optional
@@ -19,59 +17,85 @@ if TYPE_CHECKING:
logger = logging.getLogger("gateway.session")
def _now() -> datetime:
"""Return the current local time."""
return datetime.now()
def _new_session_id(now: datetime) -> str:
return f"{now.strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}"
def _iso(dt: Optional[datetime]) -> Optional[str]:
return dt.isoformat() if dt else None
def _parse_iso(value) -> Optional[datetime]:
"""``datetime.fromisoformat`` that returns None for empty/malformed input."""
if not value:
return None
try:
return datetime.fromisoformat(value)
except (TypeError, ValueError):
return None
# Auto-continue freshness window (1 hour) after the ``resume_pending`` mark; ``gateway/run.py``
# bridges config.yaml ``agent.gateway_auto_continue_freshness`` into the env var at startup.
_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60
def auto_continue_freshness_window() -> float:
"""Auto-continue freshness window in seconds (one source of truth for the resume scheduler and
the routing-time zombie gate); env var, default when unset/malformed; non-positive disables."""
raw = os.environ.get("HERMES_AUTO_CONTINUE_FRESHNESS")
try:
return float(raw) if raw else float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
except (TypeError, ValueError):
return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
class SessionLifecycleMixin:
"""SessionStore reset/expiry policy and crash-recovery markers: idle/daily
reset evaluation, expiry finalization, active-turn tokens, resume_pending,
suspension and pruning.
"""
"""SessionStore reset/expiry policy and crash-recovery markers."""
def set_expiry_finalized(
self, entry: SessionEntry, *, clear_model_override: bool = True
) -> None:
"""Mark a session entry expiry-finalized in memory, sessions.json, AND state.db.
Single write-path for the expiry watcher so the durable flag survives
sessions.json loss. ``clear_model_override=False`` = flag only.
"""
"""Mark a session entry expiry-finalized in memory, sessions.json, AND state.db (single
write-path for the expiry watcher). ``clear_model_override=False`` = flag only."""
with self._lock:
entry.expiry_finalized = True
if clear_model_override:
# Finalization is a conversation boundary: drop the persisted
# /model override so a later message cannot rehydrate it.
# Finalization is a conversation boundary: a later message must not rehydrate it.
entry.model_override = None
self._save()
# Background caller never entered ``_profile_runtime_scope``: resolve
# the store from the key, not the ambient scope.
# Background caller never entered ``_profile_runtime_scope``: resolve the store by key.
_db = self._db_for_key(entry.session_key)
if _db:
setter = getattr(_db, "set_expiry_finalized", None)
if callable(setter):
try:
setter(entry.session_id, True)
except Exception as exc:
logger.debug("Session DB expiry_finalized write failed for %s: %s", entry.session_id, exc)
if not _db:
return
setter = getattr(_db, "set_expiry_finalized", None)
if callable(setter):
try:
# Without a durable ``session_reset`` end_reason, later agent
# cleanup ends the row as ``agent_close``, which stale-route
# recovery treats as resumable. Promotion only upgrades live/
# agent_close rows; explicit boundaries are preserved.
_db.promote_to_session_reset(entry.session_id)
setter(entry.session_id, True)
except Exception as exc:
logger.debug("Session DB promote_to_session_reset failed for %s: %s", entry.session_id, exc)
logger.debug("Session DB expiry_finalized write failed for %s: %s", entry.session_id, exc)
try:
# Without a durable ``session_reset`` end_reason, later agent cleanup ends the row as
# ``agent_close``, which stale-route recovery treats as resumable.
_db.promote_to_session_reset(entry.session_id)
except Exception as exc:
logger.debug("Session DB promote_to_session_reset failed for %s: %s", entry.session_id, exc)
@staticmethod
def _policy_reset_reason(policy, updated_at: datetime) -> Optional[str]:
"""Return "idle"/"daily" when *updated_at* is overdue under *policy*, else None."""
from gateway.session import _now
if policy.mode == "none":
return None
now = _now()
if policy.mode in {"idle", "both"} and now > updated_at + timedelta(minutes=policy.idle_minutes):
return "idle"
if policy.mode in {"daily", "both"}:
today_reset = now.replace(
hour=policy.at_hour, minute=0, second=0, microsecond=0,
)
today_reset = now.replace(hour=policy.at_hour, minute=0, second=0, microsecond=0)
if now.hour < policy.at_hour:
today_reset -= timedelta(days=1)
if updated_at < today_reset:
@@ -79,46 +103,27 @@ class SessionLifecycleMixin:
return None
def _is_session_expired(self, entry: SessionEntry) -> bool:
"""Whether the entry's reset policy has expired it (entry alone, no source).
Used by the background expiry watcher. Sessions with active
background processes are never considered expired.
"""
"""Whether the reset policy has expired *entry* (expiry watcher); sessions with active
background processes never expire."""
if self._has_active_processes_safe(entry.session_key, context="expiry"):
logger.debug("Session %s not expired — active background processes", entry.session_key)
return False
policy = self.config.get_reset_policy(
platform=entry.platform,
session_type=entry.chat_type,
)
policy = self.config.get_reset_policy(platform=entry.platform, session_type=entry.chat_type)
return self._policy_reset_reason(policy, entry.updated_at) is not None
def is_session_finalizable(self, entry: SessionEntry) -> bool:
"""True if the expiry watcher will *ever* finalize this session.
A ``mode == "none"`` session never expires, so the agent-cache idle
sweep must reap its agent itself instead of deferring to the watcher
(deferring would pin the agent for the gateway's lifetime). Policy
resolution errors count as "not finalizable" (sweep reaps — safe).
"""
"""True if the expiry watcher will *ever* finalize this session; ``mode == "none"`` never
expires, so the agent-cache sweep must reap its agent itself. Policy errors -> False."""
try:
policy = self.config.get_reset_policy(
platform=entry.platform,
session_type=entry.chat_type,
)
policy = self.config.get_reset_policy(platform=entry.platform, session_type=entry.chat_type)
return policy.mode != "none"
except Exception:
return False
def _is_session_ended_in_db(self, session_id: str) -> bool:
"""True iff state.db has this session with a non-null end_reason.
Same staleness test as ``_prune_stale_sessions_locked`` (no DB, no
row, or DB error -> False, keep). Used by ``get_or_create_session``
to self-heal at routing time, since the startup prune cannot see a
session ended while the gateway stays alive. Store resolved from the
row's owning profile, not the ambient scope.
"""
"""True iff state.db has this session with a non-null end_reason (same staleness test as
``_prune_stale_sessions_locked``; no DB/row or DB error -> False). Lets routing self-heal a
session ended while the gateway stays alive. Store resolved from the owning profile."""
db = self._db_for_session_id(session_id)
if not db or not session_id:
return False
@@ -129,40 +134,29 @@ class SessionLifecycleMixin:
return bool(row is not None and row.get("end_reason") is not None)
def _should_reset(self, entry: SessionEntry, source: SessionSource) -> Optional[str]:
"""Return the reset reason ("idle"/"daily") if policy says reset, else None.
Sessions with active background processes are never reset.
"""
"""Reset reason ("idle"/"daily") if policy says reset, else None; sessions with active
background processes are never reset."""
session_key = self._generate_session_key(source)
if self._has_active_processes_safe(session_key, context="reset"):
logger.debug("Session reset skipped for %s — active background processes", session_key)
return None
policy = self.config.get_reset_policy(
platform=source.platform,
session_type=source.chat_type
)
policy = self.config.get_reset_policy(platform=source.platform, session_type=source.chat_type)
return self._policy_reset_reason(policy, entry.updated_at)
def _route_reset_reason(
self, entry: SessionEntry, source: SessionSource, now: datetime
) -> Optional[str]:
"""Reset decision for an existing route (no lock; DB/config I/O).
``suspended`` always resets. Otherwise the reset policy decides; a
still-pending resume marker is additionally freshness-gated — but
``session_reset.mode: none`` (user opted out of ALL automatic resets)
makes an expired marker fall through to a normal resume, never a
silent fresh session.
"""
from gateway.session import auto_continue_freshness_window
"""Reset decision for an existing route (no lock; DB/config I/O). ``suspended`` always
resets; otherwise the reset policy decides, and a still-pending resume marker is also
freshness-gated — but ``session_reset.mode: none`` (user opted out of ALL automatic
resets) makes an expired marker fall through to a normal resume, never a silent fresh
session."""
if entry.suspended:
return "suspended"
reason = self._should_reset(entry, source)
if reason or not entry.resume_pending:
return reason
policy = self.config.get_reset_policy(
platform=source.platform, session_type=source.chat_type,
)
policy = self.config.get_reset_policy(platform=source.platform, session_type=source.chat_type)
if policy.mode == "none":
return None
window = auto_continue_freshness_window()
@@ -172,8 +166,8 @@ class SessionLifecycleMixin:
return None
def _update_entry(self, session_key: str, mutate) -> bool:
"""Apply ``mutate(entry)`` under ``_lock`` and full-save; False when the
entry is missing or *mutate* returned False (nothing to persist)."""
"""Apply ``mutate(entry)`` under ``_lock`` and full-save; False when the entry is missing
or *mutate* returned False (nothing to persist)."""
with self._lock:
entry = self._entry_locked(session_key)
if entry is None or mutate(entry) is False:
@@ -181,205 +175,161 @@ class SessionLifecycleMixin:
self._save()
return True
def _update_all_entries_locked(self, mutate) -> int:
"""Apply ``mutate(entry) -> bool`` to every entry under ``_lock``; save once if any
returned True. Returns the count that did."""
with self._lock:
self._ensure_loaded_locked()
changed = sum(1 for entry in self._entries.values() if mutate(entry))
if changed:
self._save()
return changed
def suspend_session(self, session_key: str) -> bool:
"""Mark a session suspended so it auto-resets on next access (/stop).
Returns True if the session existed."""
"""Mark a session suspended so it auto-resets on next access (/stop). True if it existed."""
return self._update_entry(session_key, lambda e: setattr(e, "suspended", True))
def mark_turn_active(self, session_key: str) -> Optional[str]:
"""Persist exact ownership of the agent turn running for *session_key*.
def _set_turn_marker_locked(self, session_key: str, entry: SessionEntry, token, started_at) -> None:
"""Persist the active-turn pair BEFORE publishing it in memory, so a failed write can
neither leak an unowned token nor drop a live one. Lock held."""
candidate = entry.to_dict()
candidate["active_turn_token"] = token
candidate["active_turn_started_at"] = _iso(started_at)
if started_at is not None:
# Keeps the legacy 120s startup heuristic working for an older binary during a rolling
# downgrade/upgrade window.
candidate["updated_at"] = started_at.isoformat()
self._save_entry(session_key, entry_data=candidate, lock_held=True)
entry.active_turn_token = token
entry.active_turn_started_at = started_at
if started_at is not None:
entry.updated_at = started_at
The opaque token is returned to the caller and must be supplied to
:meth:`clear_turn_active`. Re-marking replaces the previous token so
a stale asynchronous unwind cannot clear a newer turn.
"""
from gateway.session import _now
def mark_turn_active(self, session_key: str) -> Optional[str]:
"""Persist exact ownership of the running agent turn; returns the opaque token for
:meth:`clear_turn_active`. Re-marking replaces the previous token so a stale asynchronous
unwind cannot clear a newer turn."""
token = uuid.uuid4().hex
with self._lock:
entry = self._entry_locked(session_key)
if entry is None:
return None
now = _now()
candidate = entry.to_dict()
candidate["active_turn_token"] = token
candidate["active_turn_started_at"] = now.isoformat()
# Keeps the legacy 120s startup heuristic working for an older
# binary during a rolling downgrade/upgrade window.
candidate["updated_at"] = now.isoformat()
# Persist before publishing in memory so a failed write cannot
# leak an unowned token through a later unrelated save.
self._save_entry(session_key, entry_data=candidate, lock_held=True)
entry.active_turn_token = token
entry.active_turn_started_at = now
entry.updated_at = now
self._set_turn_marker_locked(session_key, entry, token, _now())
return token
def clear_turn_active(self, session_key: str, token: str) -> bool:
"""Compare-and-swap clear an active-turn marker.
Returns ``False`` when the entry disappeared or a newer turn owns it.
"""
"""Compare-and-swap clear an active-turn marker; ``False`` when the entry disappeared or a
newer turn owns it."""
with self._lock:
entry = self._entry_locked(session_key)
if entry is None or entry.active_turn_token != token:
return False
candidate = entry.to_dict()
candidate["active_turn_token"] = None
candidate["active_turn_started_at"] = None
# Keep the live token until the clear is durable (retryable).
self._save_entry(session_key, entry_data=candidate, lock_held=True)
entry.active_turn_token = None
entry.active_turn_started_at = None
self._set_turn_marker_locked(session_key, entry, None, None)
return True
def recover_interrupted_turns(
self,
max_age_seconds: int = 60 * 60,
) -> int:
def recover_interrupted_turns(self, max_age_seconds: int = 60 * 60) -> int:
"""Promote crash-left turn markers into ``resume_pending`` (unclean startup only).
Old/invalid markers are cleared without resuming; suspended sessions
are never re-armed. Returns the number of newly promoted sessions.
"""
from gateway.session import _now
Old/invalid markers are cleared without resuming; suspended sessions are never re-armed.
Returns the number of newly promoted sessions."""
now = _now()
max_age = timedelta(seconds=max(0, max_age_seconds))
promoted = 0
changed = False
with self._lock:
self._ensure_loaded_locked()
for entry in self._entries.values():
if not entry.active_turn_token:
continue
started_at = entry.active_turn_started_at
try:
marker_is_stale = (
started_at is None
or (max_age_seconds > 0 and now - started_at > max_age)
)
except TypeError:
# Mixed aware/naive timestamps: clear rather than risk an
# unsafe old resume.
marker_is_stale = True
if not marker_is_stale and not entry.suspended:
if entry.resume_pending:
# A drain-timeout marker is more specific; keep it.
if entry.last_resume_marked_at is None:
entry.last_resume_marked_at = now
else:
entry.resume_pending = True
entry.resume_reason = "restart_interrupted"
# Freshness starts at discovery, not turn start.
def _promote(entry: SessionEntry) -> bool:
nonlocal promoted
if not entry.active_turn_token:
return False
started_at = entry.active_turn_started_at
try:
marker_is_stale = started_at is None or (
max_age_seconds > 0 and now - started_at > max_age
)
except TypeError:
# Mixed aware/naive timestamps: clear rather than risk an unsafe old resume.
marker_is_stale = True
if not marker_is_stale and not entry.suspended:
if entry.resume_pending:
# A drain-timeout marker is more specific; keep it.
if entry.last_resume_marked_at is None:
entry.last_resume_marked_at = now
promoted += 1
entry.active_turn_token = None
entry.active_turn_started_at = None
changed = True
if changed:
self._save()
else:
entry.resume_pending = True
entry.resume_reason = "restart_interrupted"
entry.last_resume_marked_at = now # freshness starts at discovery
promoted += 1
entry.active_turn_token = None
entry.active_turn_started_at = None
return True
self._update_all_entries_locked(_promote)
return promoted
def discard_active_turn_markers(self) -> int:
"""Clear orphan turn markers after a verified clean shutdown."""
cleared = 0
with self._lock:
self._ensure_loaded_locked()
for entry in self._entries.values():
if not entry.active_turn_token and entry.active_turn_started_at is None:
continue
entry.active_turn_token = None
entry.active_turn_started_at = None
cleared += 1
if cleared:
self._save()
return cleared
def _discard(entry: SessionEntry) -> bool:
if not entry.active_turn_token and entry.active_turn_started_at is None:
return False
entry.active_turn_token = None
entry.active_turn_started_at = None
return True
return self._update_all_entries_locked(_discard)
def mark_resume_pending(self, session_key: str, reason: str = "restart_timeout") -> bool:
"""Mark a session resumable after a restart interruption (keeps the
session_id/transcript, unlike ``suspend_session``). True if marked."""
from gateway.session import _now
"""Mark a session resumable after a restart interruption (keeps the session_id/transcript,
unlike ``suspend_session``). True if marked."""
def _apply(entry: SessionEntry):
# Never override an explicit ``suspended`` (hard forced-wipe).
if entry.suspended:
if entry.suspended: # never override an explicit ``suspended`` (hard forced-wipe)
return False
entry.resume_pending = True
entry.resume_reason = reason
entry.last_resume_marked_at = _now()
return self._update_entry(session_key, _apply)
def clear_resume_pending(self, session_key: str) -> bool:
"""Clear the resume-pending flag after a successful resumed turn.
Returns True if a flag was cleared."""
"""Clear the resume-pending flag after a successful resumed turn; True if cleared."""
def _apply(entry: SessionEntry):
if not entry.resume_pending:
return False
entry.resume_pending = False
entry.resume_reason = None
entry.last_resume_marked_at = None
return self._update_entry(session_key, _apply)
def prune_old_entries(self, max_age_days: int) -> int:
"""Drop routing entries idle (by ``updated_at``) for more than max_age_days.
Suspended entries and entries with active background processes are
kept. The SQLite transcript stays; only the key -> session_id mapping
is dropped. ``max_age_days <= 0`` disables. Returns the count removed.
"""
from gateway.session import _now
"""Drop routing entries idle (by ``updated_at``) for more than max_age_days; suspended
entries and entries with active background processes are kept. Only the key -> session_id
mapping is dropped (the transcript stays). ``max_age_days <= 0`` disables. Returns count."""
if max_age_days is None or max_age_days <= 0:
return 0
cutoff = _now() - timedelta(days=max_age_days)
removed_keys: list[str] = []
with self._lock:
self._ensure_loaded_locked()
for key, entry in list(self._entries.items()):
if entry.suspended:
continue
removed_keys = [
key for key, entry in list(self._entries.items())
if not entry.suspended
# The callback is keyed by session_key, NOT session_id.
if self._has_active_processes_safe(entry.session_key, context="prune"):
continue
if entry.updated_at < cutoff:
removed_keys.append(key)
and not self._has_active_processes_safe(entry.session_key, context="prune")
and entry.updated_at < cutoff
]
for key in removed_keys:
self._entries.pop(key, None)
if removed_keys:
self._save()
if removed_keys:
logger.info(
"SessionStore pruned %d entries older than %d days",
len(removed_keys), max_age_days,
)
logger.info("SessionStore pruned %d entries older than %d days",
len(removed_keys), max_age_days)
return len(removed_keys)
def suspend_recently_active(self, max_age_seconds: int = 120) -> int:
"""Mark sessions active within *max_age_seconds* as ``resume_pending``
after a crash/fast restart (already-pending and suspended entries are
skipped). Returns the number marked."""
from gateway.session import _now
"""Mark sessions active within *max_age_seconds* as ``resume_pending`` after a crash/fast
restart (already-pending and suspended entries are skipped). Returns the number marked."""
cutoff = _now() - timedelta(seconds=max_age_seconds)
count = 0
with self._lock:
self._ensure_loaded_locked()
for entry in self._entries.values():
if entry.resume_pending:
continue
if not entry.suspended and entry.updated_at >= cutoff:
entry.resume_pending = True
entry.resume_reason = "restart_interrupted"
entry.last_resume_marked_at = _now()
count += 1
if count:
self._save()
return count
def _mark(entry: SessionEntry) -> bool:
if entry.resume_pending or entry.suspended or entry.updated_at < cutoff:
return False
entry.resume_pending = True
entry.resume_reason = "restart_interrupted"
entry.last_resume_marked_at = _now()
return True
return self._update_all_entries_locked(_mark)
gateway/session_persistence.py
+174 -304
View File
@@ -1,15 +1,13 @@
"""SessionStore storage plumbing: per-profile SessionDB handle resolution and the
routing-index load/save paths (state.db gateway_routing primary, sessions.json
legacy mirror).
Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.
"""
"""SessionStore storage plumbing: per-profile SessionDB handle resolution and the routing-index
load/save paths (state.db gateway_routing primary, sessions.json legacy mirror). Mixin split out of
``gateway/session.py``; bound onto ``SessionStore`` via the MRO."""
from __future__ import annotations
import logging
import json
import os
import tempfile
import threading
from pathlib import Path
from typing import Any, Dict, Optional
@@ -22,60 +20,57 @@ if TYPE_CHECKING:
# Log-record parity with the origin module.
logger = logging.getLogger("gateway.session")
# "No SessionDB pinned" sentinel: lets ``_db`` distinguish "resolve from the active scope" from a
# deliberate ``store._db = None`` (JSONL fallback).
_DB_UNPINNED = object()
# Self-documenting sentinel written first into sessions.json; "_" keys are skipped on load.
_SESSIONS_JSON_README = (
"LEGACY MIRROR of the gateway routing index (the primary copy lives in the gateway_routing "
"table in ~/.hermes/state.db). Maps messaging session keys (agent:main:<platform>:...) to "
"active session IDs. This is NOT the session list. ALL sessions (CLI, TUI, and gateway) live "
"in ~/.hermes/state.db and are shown by `hermes sessions list` and `/sessions`. Disable this "
"file with `gateway.write_sessions_json: false` in config.yaml."
)
def _is_live_system_guard(exc: BaseException) -> bool:
"""Test-isolation guard: must stay a loud failure and is never cached."""
return isinstance(exc, RuntimeError) and "live-system guard" in str(exc)
class SessionPersistenceMixin:
"""SessionStore storage plumbing: per-profile SessionDB handle resolution and
the routing-index load/save paths (state.db gateway_routing primary,
sessions.json legacy mirror).
"""
"""SessionStore storage plumbing: SessionDB handle resolution and routing-index load/save."""
def _open_session_db_for_active_scope(self, db_path: Optional[Path] = None):
"""SessionDB for the profile scope active on this task.
``db_path`` pins the store; otherwise ``_default_db_path()`` follows the
context-local HERMES_HOME from ``_profile_runtime_scope`` (resolved per
call so multiplexed profiles reach their own store). Handles are cached
per path; failed opens enter a bounded backoff during which callers keep
using the JSONL fallback.
"""
"""SessionDB for the profile scope active on this task. ``db_path`` pins the store;
otherwise ``_default_db_path()`` follows the context-local HERMES_HOME (resolved per call so
multiplexed profiles reach their own store). Handles are cached per path; failed opens enter
a bounded backoff during which callers keep using the JSONL fallback."""
from hermes_state import _default_db_path, get_shared_session_db
path = Path(db_path) if db_path is not None else Path(_default_db_path())
def _open():
try:
# Process-wide shared registry: one writer connection per path.
return get_shared_session_db(path)
return get_shared_session_db(path) # process-wide registry: one writer per path
except Exception as e:
if isinstance(e, RuntimeError) and "live-system guard" in str(e):
# Test-isolation guard: must stay a loud failure and is
# deliberately not cached so it fires again next attempt.
raise
print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}")
if not _is_live_system_guard(e):
print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}")
raise
return self._db_handle_cache.get(
path,
_open,
non_cacheable=lambda exc: (
isinstance(exc, RuntimeError) and "live-system guard" in str(exc)
),
)
return self._db_handle_cache.get(path, _open, non_cacheable=_is_live_system_guard)
def _pinned_db(self):
"""Return the explicitly pinned DB (``store._db = x``), else ``_DB_UNPINNED``."""
from gateway.session import _DB_UNPINNED
return getattr(self, "_db_pinned", _DB_UNPINNED)
@property
def _db(self):
"""The SessionDB for the active profile scope, or a pinned override.
Assigning ``store._db`` pins that value for every subsequent read
(tests install a fake or disable the DB with ``store._db = None``).
Unpinned, each read resolves the scope so a multiplexed profile's
writes reach its own store.
"""
from gateway.session import _DB_UNPINNED
"""The SessionDB for the active profile scope, or a pinned override. Assigning ``store._db``
pins that value for every subsequent read (tests install a fake or disable the DB with
``store._db = None``); unpinned, each read resolves the scope so a multiplexed profile's
writes reach its own store."""
pinned = self._pinned_db()
if pinned is not _DB_UNPINNED:
return pinned
@@ -87,16 +82,11 @@ class SessionPersistenceMixin:
@property
def _routing_db(self):
"""The one store that owns the routing index, whatever scope is active.
``_entries`` is one flat dict holding every profile's keys, so it must
persist to ONE file (``_routing_home``), not whichever profile is
scoped — otherwise a mid-turn rewrite and the unscoped startup load see
different copies and crash markers under a secondary profile go
unrecovered. A pinned handle still wins; bare test instances lacking
the handle cache report no DB.
"""
from gateway.session import _DB_UNPINNED
"""The one store that owns the routing index, whatever scope is active. ``_entries`` is one
flat dict holding every profile's keys, so it must persist to ONE file (``_routing_home``),
not whichever profile is scoped — otherwise a mid-turn rewrite and the unscoped startup
load see different copies and crash markers under a secondary profile go unrecovered. A
pinned handle still wins; bare test instances lacking the handle cache report no DB."""
pinned = self._pinned_db()
if pinned is not _DB_UNPINNED:
return pinned
@@ -109,13 +99,9 @@ class SessionPersistenceMixin:
return None
def _named_profile_for_key(self, session_key: Optional[str]) -> Optional[str]:
"""The non-default profile that owns *session_key*, or None.
None means the ambient store is authoritative (multiplexing off, or
legacy ``agent:main`` namespace). It deliberately does NOT cover "that
profile has no directory" — ownership and resolvability are separate
questions that ``_db_for_key`` answers separately.
"""
"""The non-default profile that owns *session_key*, or None (ambient store is authoritative:
multiplexing off, or legacy ``agent:main``). Deliberately does NOT cover "that profile has
no directory" — ownership and resolvability are separate questions for ``_db_for_key``."""
if not getattr(self.config, "multiplex_profiles", False):
return None
profile = self._profile_from_session_key(session_key)
@@ -124,11 +110,8 @@ class SessionPersistenceMixin:
return profile
def _profile_home_for_key(self, session_key: Optional[str]) -> Optional[Path]:
"""HERMES_HOME of the profile that owns *session_key*, or None.
None means only "no live home to point at" — no named owner, or the
owner's directory could not be resolved.
"""
"""HERMES_HOME of the profile owning *session_key*, or None (no named owner or
unresolvable)."""
profile = self._named_profile_for_key(session_key)
if profile is None:
return None
@@ -144,22 +127,18 @@ class SessionPersistenceMixin:
except Exception as exc:
logger.debug("Could not resolve profile home for %r: %s", session_key, exc)
home = None
# Only hits are memoized: a profile directory can be provisioned
# *after* startup (enrollment bridge), and a cached miss would pin
# that profile's rows to the ambient store for the process lifetime.
# Only hits are memoized: a profile directory can be provisioned *after* startup (enrollment
# bridge), and a cached miss would pin that profile's rows to the ambient store for life.
if home is not None:
cache[profile] = home
return home
def _db_for_key(self, session_key: Optional[str]):
"""The SessionDB holding *session_key*'s rows, whatever scope is active.
``_db`` follows the ambient HERMES_HOME that only the inbound message
path installs; background work (expiry watcher) runs unscoped and would
write profile rows into the ROOT store until the stale-route self-heal
drops a live conversation. The owning profile is encoded in the key.
"""
from gateway.session import _DB_UNPINNED
"""The SessionDB holding *session_key*'s rows, whatever scope is active. ``_db`` follows the
ambient HERMES_HOME that only the inbound message path installs; background work (expiry
watcher) runs unscoped and would write profile rows into the ROOT store until the
stale-route self-heal drops a live conversation. The owning profile is encoded in the
key."""
pinned = self._pinned_db()
if pinned is not _DB_UNPINNED:
return pinned
@@ -168,29 +147,21 @@ class SessionPersistenceMixin:
return self._db
home = self._profile_home_for_key(session_key)
if home is None:
# Named owner we cannot resolve (not provisioned yet, or lookup
# failed). Falling back to the ambient store would split ONE
# session identity across two physical stores — fail closed;
# callers already handle a missing DB.
# Falling back to the ambient store would split ONE session identity across two
# physical stores — fail closed; callers already handle a missing DB.
logger.warning(
"gateway.session: profile %r has no resolvable home (key %r); "
"refusing to fall back to the ambient store",
profile, session_key,
)
"gateway.session: profile %r has no resolvable home (key %r); refusing to fall "
"back to the ambient store", profile, session_key)
return None
try:
return self._open_session_db_for_active_scope(db_path=home / "state.db")
except Exception:
# Same contract as ``_db``: a failed open degrades to JSONL fallback.
return None
return None # same contract as ``_db``: a failed open degrades to JSONL fallback
def _owner_key_for_session_id(self, session_id: Optional[str]) -> Optional[str]:
"""The routing key that owns *session_id*, or None.
The published index is authoritative; ``_session_owner_hints`` covers
the window where ownership is proven but routing not yet published.
Deliberately lock-free: several callers already hold ``_lock``.
"""
"""The routing key that owns *session_id*, or None. The published index is authoritative;
``_session_owner_hints`` covers the window where ownership is proven but routing not yet
published. Deliberately lock-free: several callers already hold ``_lock``."""
if not session_id:
return None
try:
@@ -202,22 +173,18 @@ class SessionPersistenceMixin:
return (getattr(self, "_session_owner_hints", None) or {}).get(session_id)
def _db_for_session_id(self, session_id: Optional[str]):
"""The SessionDB holding *session_id*'s row (owner recovered from the
index or a pre-published hint; unknown ids fall back to the ambient store)."""
"""The SessionDB holding *session_id*'s row (owner from the index or a pre-published hint;
unknown ids fall back to the ambient store)."""
if not session_id:
return self._db
return self._db_for_key(self._owner_key_for_session_id(session_id))
def close_all_db_handles(self) -> None:
"""Close every SessionDB handle this store opened (one per path).
Closing only ``store._db`` would strand secondary profiles' handles with
their WAL lock held ('database is locked' on restart). Drained under the
lock, closed outside it; a pinned handle is the pinner's to close.
"""
"""Close every SessionDB handle this store opened (one per path). Closing only ``store._db``
would strand secondary profiles' handles with their WAL lock held ('database is locked' on
restart). Drained under the lock, closed outside it; a pinned handle is the pinner's."""
def _close(db) -> None:
# Shared instances no-op on close(); release the refcount instead.
from hermes_state import release_or_close
from hermes_state import release_or_close # shared instances no-op on close()
try:
release_or_close(db)
except Exception as exc:
@@ -236,8 +203,8 @@ class SessionPersistenceMixin:
return self._entries.get(session_key)
def _routing_scope(self) -> str:
"""Namespace for this store's gateway_routing rows: the resolved
sessions_dir, so stores with different dirs never share entries."""
"""Namespace for this store's gateway_routing rows: the resolved sessions_dir, so stores
with different dirs never share entries."""
try:
return str(Path(self.sessions_dir).resolve())
except Exception:
@@ -249,6 +216,22 @@ class SessionPersistenceMixin:
method = getattr(db, name, None) if db else None
return method if callable(method) else None
def _load_routing_rows_locked(self) -> bool:
"""Load state.db routing entries into ``_entries``; False when there is no loader or the
load failed (warned). Lock held."""
loader = self._routing_db_method("load_gateway_routing_entries")
if loader is None:
return False
try:
for key, entry_json in loader(scope=self._routing_scope()).items():
entry = self._routing_entry_from_json(key, entry_json)
if entry is not None:
self._entries[key] = entry
return True
except Exception as e:
logger.warning("gateway.session: state.db routing load failed: %s", e)
return False
@staticmethod
def _routing_entry_from_json(key: str, entry_json: str) -> Optional[SessionEntry]:
"""Parse one gateway_routing row; None (with a warning) when invalid."""
@@ -262,43 +245,19 @@ class SessionPersistenceMixin:
return None
def _ensure_loaded_locked(self) -> None:
"""Load the routing index. Must be called with self._lock held.
state.db ``gateway_routing`` is primary; sessions.json is the legacy
import for keys the DB lacks (persisted to the DB on the next _save).
"""
"""Load the routing index (lock held). state.db ``gateway_routing`` is primary;
sessions.json is the legacy import for keys the DB lacks (persisted on the next _save)."""
if self._loaded:
self._reconcile_recovered_routing_locked()
return
self.sessions_dir.mkdir(parents=True, exist_ok=True)
db_had_entries = False
db_load_succeeded = False
loader = self._routing_db_method("load_gateway_routing_entries")
if loader is not None:
try:
for key, entry_json in loader(scope=self._routing_scope()).items():
entry = self._routing_entry_from_json(key, entry_json)
if entry is not None:
self._entries[key] = entry
db_had_entries = bool(self._entries)
db_load_succeeded = True
except Exception as e:
logger.warning("gateway.session: state.db routing load failed: %s", e)
db_load_succeeded = self._load_routing_rows_locked()
db_had_entries = db_load_succeeded and bool(self._entries)
self._import_legacy_sessions_json(db_had_entries)
self._loaded = True
self._routing_db_loaded = db_load_succeeded
self._routing_fallback_baseline = (
None
if db_load_succeeded
else {key: entry.to_dict() for key, entry in self._entries.items()}
)
# A hard crash skips graceful shutdown and leaves sessions.json
# pointing at ended sessions; self-heal before the first message.
self._routing_fallback_baseline = None if db_load_succeeded else self._entries_as_dicts()
# A hard crash skips graceful shutdown and leaves sessions.json pointing at ended sessions.
self._prune_stale_sessions_locked()
def _import_legacy_sessions_json(self, db_had_entries: bool) -> None:
@@ -315,13 +274,10 @@ class SessionPersistenceMixin:
# "_"-prefixed keys are sentinels (e.g. "_README"), not entries.
if key.startswith("_") or key in self._entries:
continue
# A non-dict entry (corrupt file) must not abort the whole load.
if not isinstance(entry_data, dict):
if not isinstance(entry_data, dict): # corrupt file must not abort the whole load
logger.warning(
"Skipping invalid session entry %r: "
"expected dict, got %s",
key, type(entry_data).__name__,
)
"Skipping invalid session entry %r: expected dict, got %s", key,
type(entry_data).__name__)
continue
try:
self._entries[key] = SessionEntry.from_dict(entry_data)
@@ -330,28 +286,23 @@ class SessionPersistenceMixin:
logger.warning("Skipping invalid session entry %r: %s", key, e)
if imported and db_had_entries:
logger.info(
"gateway.session: imported %d legacy sessions.json "
"entr%s missing from state.db routing table",
imported, "y" if imported == 1 else "ies",
)
"gateway.session: imported %d legacy sessions.json entr%s missing from "
"state.db routing table", imported, "y" if imported == 1 else "ies")
except Exception as e:
print(f"[gateway] Warning: Failed to load sessions: {e}")
def _prune_stale_sessions_locked(self) -> None:
"""Remove routing entries whose session has ended in state.db (startup, lock held).
Stale == ``end_reason IS NOT NULL``. Rows absent from the DB are kept;
a ``None`` DB handle is a no-op; DB errors are non-fatal.
"""
"""Remove routing entries whose session has ended in state.db (startup, lock held). Stale ==
``end_reason IS NOT NULL``; rows absent from the DB are kept; a ``None`` DB handle is a
no-op; DB errors are non-fatal."""
if not self._entries:
return
stale_keys: list = []
recovered_keys = 0
try:
for key, entry in self._entries.items():
# Ask the store that owns the key, not the ambient handle, or a
# live secondary-profile session gets pruned on the root copy.
# Ask the store that owns the key, not the ambient handle, or a live
# secondary-profile session gets pruned on the root copy.
db = self._db_for_key(key)
if db is None:
continue
@@ -365,86 +316,61 @@ class SessionPersistenceMixin:
self._entries[key] = verdict
recovered_keys += 1
except Exception as exc:
logger.warning(
"gateway.session: stale-entry pruning skipped due to DB error: %s",
exc,
)
logger.warning("gateway.session: stale-entry pruning skipped due to DB error: %s", exc)
return
for key in stale_keys:
del self._entries[key]
if stale_keys or recovered_keys:
self._save()
def _stale_entry_verdict(self, key: str, entry, row):
"""For a routing entry whose row has ended: ``"prune"``, a replacement
entry (repoint), or None (keep as-is)."""
"""For a routing entry whose row has ended: ``"prune"``, a replacement entry (repoint), or
None (keep as-is)."""
from gateway.session import _now
recovered_entry = None
if entry.origin is not None:
try:
recovered_entry = self._recover_session_from_db(
session_key=key,
source=entry.origin,
now=_now(),
raise_on_lookup_error=True,
)
session_key=key, source=entry.origin, now=_now(), raise_on_lookup_error=True)
except Exception as exc:
# Indeterminate: keep the only routing handle.
logger.debug(
"gateway.session: recovery lookup failed for stale "
"sessions.json entry %r -> %s: %s",
key,
entry.session_id,
exc,
)
"gateway.session: recovery lookup failed for stale sessions.json entry %r -> "
"%s: %s", key, entry.session_id, exc)
return None
# Compression-ended parent with a newer live child for the same peer:
# repoint instead of dropping, or queued/resume-pending work vanishes
# until the next message.
# Compression-ended parent with a newer live child for the same peer: repoint instead of
# dropping, or queued/resume-pending work vanishes until the next message.
if recovered_entry is not None and recovered_entry.session_id != entry.session_id:
logger.warning(
"gateway.session: repointing stale sessions.json entry "
"%r from ended %s (end_reason=%r) to recovered %s",
key,
entry.session_id,
row["end_reason"],
recovered_entry.session_id,
)
"gateway.session: repointing stale sessions.json entry %r from ended %s "
"(end_reason=%r) to recovered %s", key, entry.session_id, row["end_reason"],
recovered_entry.session_id)
return recovered_entry
# Same-id recovery == successful resume: keep the ORIGINAL entry object
# (the recovered one is rebuilt minimal and would drop counters,
# model_override, resume markers, metadata). Nothing changes, no save.
# Same-id recovery == successful resume: keep the ORIGINAL entry object (the recovered one
# is rebuilt minimal and would drop counters, model_override, resume markers, metadata).
if recovered_entry is not None:
logger.info(
"gateway.session: reopened ended session %s for "
"sessions.json entry %r (end_reason=%r); keeping route",
entry.session_id, key, row["end_reason"],
)
"gateway.session: reopened ended session %s for sessions.json entry %r "
"(end_reason=%r); keeping route", entry.session_id, key, row["end_reason"])
return None
logger.warning(
"gateway.session: pruning stale sessions.json entry "
"%r -> %s (end_reason=%r); left by a crashed gateway",
key, entry.session_id, row["end_reason"],
)
"gateway.session: pruning stale sessions.json entry %r -> %s (end_reason=%r); left by "
"a crashed gateway", key, entry.session_id, row["end_reason"])
return "prune"
def _entries_as_dicts(self) -> Dict[str, Any]:
"""Serializable snapshot of ``_entries``. Lock held."""
return {key: entry.to_dict() for key, entry in self._entries.items()}
def _save(self) -> None:
"""Persist the routing index while the caller holds ``_lock``."""
data, generation = self._snapshot_routing_locked()
self._persist_routing_data(data, generation)
def _next_routing_generation_locked(self) -> int:
"""Bump and return the shared routing counter. Caller holds ``_lock``.
Full snapshots AND single-entry fast saves MUST allocate from this one
counter: the stale-write protection is a total order over
serialization times and silently breaks otherwise.
"""
"""Bump and return the shared routing counter (lock held). Full snapshots AND single-entry
fast saves MUST allocate from this one counter: the stale-write protection is a total order
over serialization times and silently breaks otherwise."""
self._routing_generation = getattr(self, "_routing_generation", 0) + 1
return self._routing_generation
@@ -453,7 +379,6 @@ class SessionPersistenceMixin:
baseline = getattr(self, "_routing_fallback_baseline", None)
if getattr(self, "_routing_db_loaded", False) or baseline is None:
return
loader = self._routing_db_method("load_gateway_routing_entries")
if loader is None:
return
@@ -462,42 +387,34 @@ class SessionPersistenceMixin:
except Exception as exc:
logger.warning("gateway.session: recovered state.db routing load failed: %s", exc)
return
current = {key: entry.to_dict() for key, entry in self._entries.items()}
current = self._entries_as_dicts()
for key, entry_json in durable.items():
durable_entry = self._routing_entry_from_json(key, entry_json)
if durable_entry is None:
continue
if key not in baseline:
# A key created while on fallback wins over a DB-only key;
# otherwise restore the authoritative row that fallback never saw.
# A key created while on fallback wins over a DB-only key; otherwise restore the
# authoritative row that fallback never saw.
self._entries.setdefault(key, durable_entry)
elif key not in current:
# The key was loaded from fallback and deliberately removed.
continue
continue # loaded from fallback and deliberately removed
elif current[key] == baseline[key]:
# Unchanged fallback data yields to the authoritative DB copy.
self._entries[key] = durable_entry
self._entries[key] = durable_entry # unchanged fallback data yields to the DB copy
self._routing_db_loaded = True
self._routing_fallback_baseline = None
def _snapshot_routing_locked(self) -> tuple[Dict[str, Any], int]:
"""Capture immutable routing data and a monotonic generation."""
self._reconcile_recovered_routing_locked()
return (
{key: entry.to_dict() for key, entry in self._entries.items()},
self._next_routing_generation_locked(),
)
return self._entries_as_dicts(), self._next_routing_generation_locked()
def _persist_routing_data(self, data: Dict[str, Any], generation: int) -> None:
"""Serialize all whole-index writers through one durable write lock."""
with self._lazy("_save_lock", threading.Lock):
if generation <= getattr(self, "_persisted_routing_generation", 0):
return
# Fold in fast upserts numbered above this snapshot: they were
# serialized after us and a delayed full rewrite must not regress them.
# Fold in fast upserts numbered above this snapshot: they were serialized after us and
# a delayed full rewrite must not regress them.
fast_persisted = getattr(self, "_fast_persisted_entries", None)
if fast_persisted:
for key, (revision, entry_json) in fast_persisted.items():
@@ -507,10 +424,7 @@ class SessionPersistenceMixin:
replacer = self._routing_db_method("replace_gateway_routing_entries")
if replacer is not None:
try:
replacer(
{k: json.dumps(v) for k, v in data.items()},
scope=self._routing_scope(),
)
replacer({k: json.dumps(v) for k, v in data.items()}, scope=self._routing_scope())
db_saved = True
except Exception as exc:
logger.warning("gateway.session: state.db routing save failed: %s", exc)
@@ -520,47 +434,24 @@ class SessionPersistenceMixin:
except Exception as exc:
if not db_saved:
raise
# state.db is authoritative. A failed legacy mirror must not
# report the already-committed primary write as failed.
# state.db is authoritative: a failed legacy mirror must not report the
# already-committed primary write as failed.
logger.warning(
"gateway.session: sessions.json mirror save failed "
"after state.db commit: %s",
exc,
)
"gateway.session: sessions.json mirror save failed after state.db commit: "
"%s", exc)
self._persisted_routing_generation = generation
# This rewrite supersedes fast records at or below its
# generation; newer ones stay for the next delayed full writer.
# This rewrite supersedes fast records at or below its generation; newer ones stay for
# the next delayed full writer.
if fast_persisted:
for key in [
k for k, (rev, _) in fast_persisted.items()
if rev <= generation
]:
for key in [k for k, (rev, _) in fast_persisted.items() if rev <= generation]:
del fast_persisted[key]
def _save_sessions_json(self, data: Dict[str, Any]) -> None:
"""Write the legacy sessions.json mirror of the routing index."""
import tempfile
"""Write the legacy sessions.json mirror of the routing index (atomic + fsync)."""
self.sessions_dir.mkdir(parents=True, exist_ok=True)
sessions_file = self.sessions_dir / "sessions.json"
# Self-documenting sentinel; "_" keys are skipped on load. Ordered
# first so it renders at the top of the file.
data = {
"_README": (
"LEGACY MIRROR of the gateway routing index (the primary copy "
"lives in the gateway_routing table in ~/.hermes/state.db). "
"Maps messaging session keys (agent:main:<platform>:...) to "
"active session IDs. This is NOT the session list. ALL "
"sessions (CLI, TUI, and gateway) live in ~/.hermes/state.db "
"and are shown by `hermes sessions list` and `/sessions`. "
"Disable this file with `gateway.write_sessions_json: false` "
"in config.yaml."
),
**data,
}
fd, tmp_path = tempfile.mkstemp(
dir=str(self.sessions_dir), suffix=".tmp", prefix=".sessions_"
)
data = {"_README": _SESSIONS_JSON_README, **data}
fd, tmp_path = tempfile.mkstemp(dir=str(self.sessions_dir), suffix=".tmp", prefix=".sessions_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
@@ -581,47 +472,34 @@ class SessionPersistenceMixin:
self._persist_routing_data(data, generation)
def _save_entry(
self,
session_key: str,
*,
entry_data: Optional[Dict[str, Any]] = None,
lock_held: bool = False,
) -> None:
"""Persist ONE routing entry via UPSERT — the per-turn fast path
(a full rewrite fsyncs a multi-MB sessions.json, ~50ms at ~1100 keys).
Invariants: the key -> session_id mapping never changes here —
structural transitions (create/recover/reset/switch/prune/heal) use the
full rewrite, which also refreshes the sessions.json mirror (it may lag
in metadata only; state.db stays primary). The entry is serialized under
``_lock`` with a revision from the shared routing generation counter
(higher == same-or-newer); under ``_save_lock`` the upsert is skipped if
a full snapshot or a newer fast save of this key already persisted (the
reverse case lives in ``_persist_routing_data``). No DB or a failed
upsert falls back to the full rewrite so DB-less installs stay durable.
``entry_data`` persists a candidate BEFORE it is published to the live
entry (failure-atomic transitions); the fallback carries the same candidate.
"""
def _capture() -> Optional[tuple[str, int, Optional[Dict[str, Any]]]]:
entry = self._entries.get(session_key)
if entry is None:
return None
serialized_entry = (
dict(entry_data) if entry_data is not None else entry.to_dict()
)
entry_json = json.dumps(serialized_entry)
revision = self._next_routing_generation_locked()
# The O(n) full snapshot is deferred to the fallback branch.
return entry_json, revision, serialized_entry if entry_data is not None else None
if lock_held:
captured = _capture()
else:
self, session_key: str, *, entry_data: Optional[Dict[str, Any]] = None,
lock_held: bool = False) -> None:
"""Persist ONE routing entry via UPSERT — the per-turn fast path (a full rewrite fsyncs a
multi-MB sessions.json). The key -> session_id mapping never changes here: structural
transitions use the full rewrite (which also refreshes the sessions.json mirror; it may lag
in metadata only). The revision comes from the shared routing generation counter; under
``_save_lock`` the upsert is skipped if a full snapshot or a newer fast save of this key
already persisted (the reverse case lives in ``_persist_routing_data``). No DB or a failed
upsert falls back to the full rewrite. ``entry_data`` persists a candidate BEFORE it is
published to the live entry (failure-atomic transitions); the fallback carries it too."""
def _locked(fn):
if lock_held:
return fn()
with self._lock:
captured = _capture()
return fn()
def _capture():
if session_key not in self._entries:
return None
entry = self._entries[session_key]
serialized = dict(entry_data) if entry_data is not None else entry.to_dict()
# The O(n) full snapshot is deferred to the fallback branch.
return json.dumps(serialized), self._next_routing_generation_locked()
captured = _locked(_capture)
if captured is None:
return
entry_json, revision, candidate_entry = captured
entry_json, revision = captured
saver = self._routing_db_method("save_gateway_routing_entry")
if saver is not None:
try:
@@ -637,20 +515,12 @@ class SessionPersistenceMixin:
return
except Exception as exc:
logger.warning(
"gateway.session: single-entry routing save failed for %r "
"(%s); falling back to full index rewrite",
session_key, exc,
)
if candidate_entry is not None:
"gateway.session: single-entry routing save failed for %r (%s); falling back "
"to full index rewrite", session_key, exc)
if entry_data is not None:
# Full-snapshot fallback carrying the candidate transition.
def _snapshot() -> Dict[str, Any]:
return {key: current.to_dict() for key, current in self._entries.items()}
if lock_held:
fallback_data = _snapshot()
else:
with self._lock:
fallback_data = _snapshot()
fallback_data[session_key] = candidate_entry
fallback_data = _locked(self._entries_as_dicts)
fallback_data[session_key] = dict(entry_data)
self._persist_routing_data(fallback_data, revision)
else:
self._save_entries()
gateway/session_recovery.py
+98 -238
View File
@@ -1,9 +1,6 @@
"""SessionStore durable-row recovery: session-key generation, legacy Slack key
migration, rebuilding a routing entry from state.db, and the SQLite side of
routing transitions (promote/reopen/create/peer).
Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.
"""
"""SessionStore durable-row recovery: session-key generation, legacy Slack key migration, rebuilding
a routing entry from state.db, and the SQLite side of routing transitions (promote/reopen/create/
peer). Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO."""
from __future__ import annotations
@@ -33,14 +30,11 @@ def _origin_json(source) -> Optional[str]:
class SessionRecoveryMixin:
"""SessionStore durable-row recovery: session-key generation, legacy Slack
key migration, rebuilding a routing entry from state.db, and the SQLite
side of routing transitions (promote/reopen/create/peer).
"""
"""SessionStore durable-row recovery and the SQLite side of routing transitions."""
def _resolve_profile_for_key(self, source: Optional[SessionSource] = None) -> Optional[str]:
"""Profile namespace for session keys: None when multiplexing is off
(legacy ``agent:main``), else ``source.profile`` or the active profile."""
"""Profile namespace for session keys: None when multiplexing is off (legacy
``agent:main``), else ``source.profile`` or the active profile."""
if not getattr(self.config, "multiplex_profiles", False):
return None
if source is not None and source.profile:
@@ -71,53 +65,37 @@ class SessionRecoveryMixin:
return "default"
def _recovered_row_allowed_for_active_profile(
self,
*,
requested_session_key: str,
recovered: Dict[str, Any],
self, *, requested_session_key: str, recovered: Dict[str, Any]
) -> bool:
"""Prevent a gateway from reviving another profile's row.
Single-profile: the row's namespace must match the ACTIVE profile.
Multiplexed: it must match the namespace of the requested key (the
active profile is meaningless there). Keyless rows stay adoptable.
"""
"""Prevent a gateway from reviving another profile's row. Single-profile: the row's
namespace must match the ACTIVE profile. Multiplexed: it must match the requested key's
namespace (the active profile is meaningless there). Keyless rows stay adoptable."""
recovered_key = str(recovered.get("session_key") or "")
if not recovered_key or recovered_key == requested_session_key:
return True
recovered_profile = self._profile_from_session_key(recovered_key)
if recovered_profile is None:
return True
if getattr(self.config, "multiplex_profiles", False):
requested_profile = self._profile_from_session_key(requested_session_key)
return requested_profile is None or recovered_profile == requested_profile
return recovered_profile == self._active_profile_name()
def _generate_session_key(self, source: SessionSource, key_source: Optional[SessionSource] = None) -> str:
"""Session key for *source* (profile resolved from *source*, key built
from *key_source* when given)."""
"""Session key for *source* (profile from *source*; key from *key_source* if given)."""
from gateway.session import build_session_key
return build_session_key(
key_source if key_source is not None else source,
group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True),
thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False),
profile=self._resolve_profile_for_key(source),
)
profile=self._resolve_profile_for_key(source))
def _legacy_slack_session_key(self, source: SessionSource) -> Optional[str]:
"""Pre-workspace Slack key for an explicitly scoped source.
Deliberately Slack-only; an unscoped Slack session may be claimed by
only one workspace because its old key cannot distinguish teams.
"""
"""Pre-workspace Slack key for an explicitly scoped source. Deliberately Slack-only: an
unscoped Slack session may be claimed by only one workspace (old key cannot tell teams)."""
if source.platform != Platform.SLACK or not source.scope_id:
return None
return self._generate_session_key(
source, replace(source, scope_id=None, guild_id=None)
)
return self._generate_session_key(source, replace(source, scope_id=None, guild_id=None))
def _claim_legacy_slack_key(self, legacy_key: Optional[str]) -> bool:
"""Atomically reserve one ambiguous legacy Slack key for migration."""
@@ -134,17 +112,10 @@ class SessionRecoveryMixin:
def _recovered_row_matches_source_scope(
recovered: Dict[str, Any], source: SessionSource
) -> bool:
"""Reject recovered rows whose recorded origin belongs to another workspace.
A workspace-scoped Slack lookup adopts a row only if its origin_json
names the same scope_id; rows without a parseable origin are rejected
(an unattributable transcript is exactly the ambiguity to avoid).
"""
if (
source.platform != Platform.SLACK
or source.chat_type == "dm"
or not source.scope_id
):
"""Reject recovered rows whose origin belongs to another workspace: a workspace-scoped Slack
lookup adopts a row only if its origin_json names the same scope_id; rows without a
parseable origin are rejected (an unattributable transcript is exactly the ambiguity)."""
if source.platform != Platform.SLACK or source.chat_type == "dm" or not source.scope_id:
return True
try:
origin = json.loads(recovered.get("origin_json") or "")
@@ -155,14 +126,10 @@ class SessionRecoveryMixin:
return origin.get("scope_id", origin.get("guild_id")) == source.scope_id
def _create_entry_from_recovered_row(
self,
*,
row: Dict[str, Any],
session_key: str,
source: SessionSource,
now: datetime,
self, *, row: Dict[str, Any], session_key: str, source: SessionSource, now: datetime,
) -> SessionEntry:
from gateway.session import SessionEntry
def _ts(value, default: datetime) -> datetime:
try:
return datetime.fromtimestamp(float(value))
@@ -176,49 +143,29 @@ class SessionRecoveryMixin:
updated_at = _ts(last_activity, created_at) if last_activity is not None else created_at
had_activity = row.get("_has_messages")
if had_activity is None:
had_activity = bool(row.get("message_count") or 0) or (
last_activity is not None
)
had_activity = bool(row.get("message_count") or 0) or last_activity is not None
return SessionEntry(
session_key=session_key,
session_id=str(row["id"]),
created_at=created_at,
updated_at=updated_at,
origin=source,
display_name=source.chat_name,
platform=source.platform,
chat_type=source.chat_type,
reset_had_activity=bool(had_activity),
)
session_key=session_key, session_id=str(row["id"]), created_at=created_at,
updated_at=updated_at, origin=source, display_name=source.chat_name,
platform=source.platform, chat_type=source.chat_type,
reset_had_activity=bool(had_activity))
def _find_gateway_session_row(
self,
*,
session_key: str,
source: SessionSource,
allow_peer_fallback: bool,
raise_on_lookup_error: bool = False,
) -> Optional[Dict[str, Any]]:
"""Query one durable gateway session row.
Scoped Slack lookups disable SessionDB's platform/chat/user fallback:
that tuple does not contain a workspace id and could therefore revive
another team's session. The caller performs one explicit exact lookup
of the old unscoped key instead.
"""
self, *, session_key: str, source: SessionSource, allow_peer_fallback: bool,
raise_on_lookup_error: bool = False) -> Optional[Dict[str, Any]]:
"""Query one durable gateway session row. Scoped Slack lookups disable SessionDB's
platform/chat/user fallback: that tuple has no workspace id and could revive another team's
session; the caller performs one explicit exact lookup of the old unscoped key instead."""
db = self._db_for_key(session_key)
finder = getattr(db, "find_latest_gateway_session_for_peer", None) if db else None
if not callable(finder):
return None
try:
return finder(
source=source.platform.value,
user_id=source.user_id,
session_key=session_key,
source=source.platform.value, user_id=source.user_id, session_key=session_key,
chat_id=source.chat_id if allow_peer_fallback else None,
chat_type=source.chat_type if allow_peer_fallback else None,
thread_id=source.thread_id,
)
thread_id=source.thread_id)
except Exception as exc:
logger.debug("Gateway session DB recovery failed for %s: %s", session_key, exc)
if raise_on_lookup_error:
@@ -226,25 +173,14 @@ class SessionRecoveryMixin:
return None
def _recover_session_from_db(
self,
*,
session_key: str,
source: SessionSource,
now: datetime,
raise_on_lookup_error: bool = False,
) -> Optional[SessionEntry]:
"""Rebuild a missing session-key mapping from durable state.db data.
Returns ``None`` when no row is recoverable, or when the recovered
session is already overdue under the reset policy — the row is then
durably promoted to a reset boundary instead of resurrected.
"""
self, *, session_key: str, source: SessionSource, now: datetime,
raise_on_lookup_error: bool = False) -> Optional[SessionEntry]:
"""Rebuild a missing session-key mapping from durable state.db data. ``None`` when no row is
recoverable, or when the recovered session is already overdue under the reset policy — the
row is then durably promoted to a reset boundary instead of resurrected."""
entry, migrated_legacy = self._query_recoverable_row(
session_key=session_key,
source=source,
now=now,
raise_on_lookup_error=raise_on_lookup_error,
)
session_key=session_key, source=source, now=now,
raise_on_lookup_error=raise_on_lookup_error)
if entry is None:
return None
reset_reason = self._should_reset(entry, source)
@@ -252,93 +188,62 @@ class SessionRecoveryMixin:
self._promote_session_reset(
session_key, entry.session_id, reset_reason,
log=lambda exc: logger.debug(
"Gateway recovered-session reset promotion failed for %s: %s",
session_key, exc,
"Gateway recovered-session reset promotion failed for %s: %s", session_key, exc,
),
)
return None
self._reopen_session_row(session_key, entry.session_id)
if migrated_legacy:
self._record_gateway_session_peer(
entry.session_id, session_key, source, display_name=entry.display_name,
)
entry.session_id, session_key, source, display_name=entry.display_name)
return entry
def _query_recoverable_session(self, *, session_key, source, now):
"""DB-only half of _recover_session_from_db (no lock needed).
Returns a SessionEntry or None. Caller assigns _entries[key] under
lock. The row is NOT reopened here: the caller evaluates reset policy
first (an agent_close/ws_orphan row may need promotion to a real reset
boundary instead).
"""
"""DB-only half of _recover_session_from_db (no lock needed): a SessionEntry or None; the
caller assigns _entries[key] under lock. The row is NOT reopened here: the caller evaluates
reset policy first (an agent_close/ws_orphan row may need promotion to a real reset)."""
entry, migrated_legacy = self._query_recoverable_row(
session_key=session_key, source=source, now=now,
)
session_key=session_key, source=source, now=now)
if entry is not None and migrated_legacy:
self._record_gateway_session_peer(
entry.session_id, session_key, source, display_name=entry.display_name,
)
entry.session_id, session_key, source, display_name=entry.display_name)
return entry
def _query_recoverable_row(
self, *, session_key, source, now, raise_on_lookup_error=False,
) -> tuple[Optional[SessionEntry], bool]:
"""Find and gate a recoverable row -> (entry or None, migrated_legacy).
The legacy (pre-workspace) Slack key fallback lives here: exact-key
lookup, claimed once per process; ``migrated_legacy`` tells the caller
to rewrite the peer row to the scoped key.
"""
"""Find and gate a recoverable row -> (entry or None, migrated_legacy). The legacy
(pre-workspace) Slack key fallback lives here: exact-key lookup, claimed once per process;
``migrated_legacy`` tells the caller to rewrite the peer row to the scoped key."""
legacy_key = self._legacy_slack_session_key(source)
recovered = self._find_gateway_session_row(
session_key=session_key,
source=source,
allow_peer_fallback=legacy_key is None,
raise_on_lookup_error=raise_on_lookup_error,
)
session_key=session_key, source=source, allow_peer_fallback=legacy_key is None,
raise_on_lookup_error=raise_on_lookup_error)
migrated_legacy = False
if (
not recovered
and legacy_key
and self._claim_legacy_slack_key(legacy_key)
):
if not recovered and legacy_key and self._claim_legacy_slack_key(legacy_key):
recovered = self._find_gateway_session_row(
session_key=legacy_key,
source=source,
allow_peer_fallback=False,
raise_on_lookup_error=raise_on_lookup_error,
)
session_key=legacy_key, source=source, allow_peer_fallback=False,
raise_on_lookup_error=raise_on_lookup_error)
migrated_legacy = bool(recovered)
if not isinstance(recovered, dict):
return None, False
if not self._recovered_row_matches_source_scope(recovered, source):
return None, False
if not self._recovered_row_allowed_for_active_profile(
requested_session_key=session_key,
recovered=recovered,
):
requested_session_key=session_key, recovered=recovered):
logger.warning(
"Gateway session DB recovery ignored %s for %s because "
"the row belongs to a different profile",
recovered.get("session_key"),
session_key,
)
"Gateway session DB recovery ignored %s for %s because the row belongs to a "
"different profile", recovered.get("session_key"), session_key)
return None, False
entry = self._create_entry_from_recovered_row(
row=recovered, session_key=session_key, source=source, now=now,
)
row=recovered, session_key=session_key, source=source, now=now)
return entry, migrated_legacy
def _promote_session_reset(self, session_key: str, session_id: str, reason: str, *, log) -> None:
"""End *session_id* with *reason* via ``promote_to_session_reset``.
Promote (not plain ``end_session``): a row already ended with a
recoverable accidental reason (agent_close / ws_orphan_reap) must be
upgraded to the explicit boundary, or stale-route recovery resurrects
it over the reset. Falls back to ``end_session`` on old SessionDBs.
``log(exc)`` reports failures (each caller has its own message).
"""
"""End *session_id* with *reason* via ``promote_to_session_reset`` (``end_session`` on old
SessionDBs). Promote, not plain end: a row already ended with a recoverable accidental
reason (agent_close / ws_orphan_reap) must be upgraded to the explicit boundary, or
stale-route recovery resurrects it over the reset. ``log(exc)`` reports failures."""
try:
db = self._db_for_key(session_key)
promote = getattr(db, "promote_to_session_reset", None)
@@ -360,13 +265,8 @@ class SessionRecoveryMixin:
logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc)
def _record_gateway_session_peer(
self,
session_id: str,
session_key: str,
source: Optional[SessionSource],
display_name: Optional[str] = None,
include_compression_ancestors: bool = False,
) -> None:
self, session_id: str, session_key: str, source: Optional[SessionSource],
display_name: Optional[str] = None, include_compression_ancestors: bool = False) -> None:
"""Persist the routing peer for an existing gateway session row."""
db = self._db_for_key(session_key)
if not db or not source:
@@ -375,25 +275,15 @@ class SessionRecoveryMixin:
if not callable(recorder):
return
peer = dict(
source=source.platform.value,
user_id=source.user_id,
session_key=session_key,
chat_id=source.chat_id,
chat_type=source.chat_type,
thread_id=source.thread_id,
)
source=source.platform.value, user_id=source.user_id, session_key=session_key,
chat_id=source.chat_id, chat_type=source.chat_type, thread_id=source.thread_id)
try:
origin_json = _origin_json(source)
recorder(
session_id,
**peer,
display_name=display_name or source.chat_name,
origin_json=origin_json,
include_compression_ancestors=include_compression_ancestors,
)
session_id, **peer, display_name=display_name or source.chat_name,
origin_json=_origin_json(source),
include_compression_ancestors=include_compression_ancestors)
except TypeError:
# Older SessionDB without display_name/origin_json kwargs.
try:
try: # older SessionDB without display_name/origin_json kwargs
recorder(session_id, **peer)
except Exception as exc:
logger.debug("Gateway session peer record failed for %s: %s", session_key, exc)
@@ -401,14 +291,10 @@ class SessionRecoveryMixin:
logger.debug("Gateway session peer record failed for %s: %s", session_key, exc)
def _adopt_legacy_slack_entry(self, source: SessionSource, session_key: str) -> None:
"""One-time migration of pre-workspace-scope Slack keys.
MOVE (not copy) the legacy entry so a second workspace with identical
Slack ids cannot attach to the same transcript. Adopt when the legacy
origin names the same workspace; a scope-less DM is claimed once by
the first workspace; a scope-less channel/group is refused (channel
ids collide across workspaces).
"""
"""One-time migration of pre-workspace-scope Slack keys: MOVE (not copy) the legacy entry so
a second workspace with identical Slack ids cannot attach to the same transcript. Adopt when
the legacy origin names the same workspace; a scope-less DM is claimed once by the first
workspace; a scope-less channel/group is refused (channel ids collide across workspaces)."""
legacy_key = self._legacy_slack_session_key(source)
if not legacy_key:
return
@@ -432,58 +318,40 @@ class SessionRecoveryMixin:
if migrated is not None:
self._save_entries()
self._record_gateway_session_peer(
migrated.session_id, session_key, source, display_name=migrated.display_name,
)
migrated.session_id, session_key, source, display_name=migrated.display_name)
def _finish_route_transition(
self,
session_key: str,
*,
end_session_id: Optional[str],
end_reason: str,
create_kwargs: Optional[Dict[str, Any]],
origin: Optional[SessionSource],
display_name: Optional[str],
during: str = "",
) -> None:
"""SQLite side of a routing transition, outside ``_lock``.
Promotes the predecessor row to an explicit reset boundary (with the
specific reason so state.db is auditable, e.g. ``resume_pending_expired``
vs a plain ``session_reset``), then INSERTs the new row + routing peer.
Both are best-effort: failures are warned and self-healed by the next
per-turn peer refresh.
"""
self, session_key: str, *, end_session_id: Optional[str], end_reason: str,
create_kwargs: Optional[Dict[str, Any]], origin: Optional[SessionSource],
display_name: Optional[str], during: str = "") -> None:
"""SQLite side of a routing transition, outside ``_lock``: promote the predecessor row to an
explicit reset boundary (with the specific reason so state.db is auditable, e.g.
``resume_pending_expired`` vs plain ``session_reset``), then INSERT the new row + routing
peer. Both best-effort: failures are warned and self-healed by the next peer refresh."""
if self._db_for_key(session_key) and end_session_id:
self._promote_session_reset(
session_key, end_session_id, end_reason,
log=lambda e: logger.warning(
"Failed to end predecessor session row %s for %s%s: %s — "
"the old row remains open and may win restart recovery "
"until the next successful peer refresh",
end_session_id, session_key, during, e,
),
"Failed to end predecessor session row %s for %s%s: %s — the old row remains "
"open and may win restart recovery until the next successful peer refresh",
end_session_id, session_key, during, e),
)
if self._db_for_key(session_key) and create_kwargs:
self._create_session_row(
session_key, create_kwargs, origin, display_name,
log=lambda e: logger.warning(
"Failed to create session row %s for %s%s: %s — deferring "
"to the self-healing peer refresh on the next turn",
create_kwargs.get("session_id"), session_key, during, e,
),
"Failed to create session row %s for %s%s: %s — deferring to the "
"self-healing peer refresh on the next turn",
create_kwargs.get("session_id"), session_key, during, e),
)
@staticmethod
def _session_create_kwargs(
*, session_id, session_key, origin, source_value, display_name, parent_session_id,
) -> Dict[str, Any]:
"""kwargs for ``SessionDB.create_session``.
Identity (origin_json) and lineage (parent/_reset_from) land atomically
in the INSERT so a crash right after cannot strand the row unroutable.
"""
origin_json = _origin_json(origin)
"""kwargs for ``SessionDB.create_session``. Identity (origin_json) and lineage
(parent/_reset_from) land atomically in the INSERT so a crash right after cannot strand the
row unroutable."""
return {
"session_id": session_id,
"source": source_value,
@@ -493,27 +361,19 @@ class SessionRecoveryMixin:
"chat_type": origin.chat_type if origin else None,
"thread_id": origin.thread_id if origin else None,
"profile_name": origin.profile if origin else None,
"origin_json": origin_json,
"origin_json": _origin_json(origin),
"display_name": display_name,
"parent_session_id": parent_session_id,
"model_config": (
{"_reset_from": parent_session_id} if parent_session_id else None
),
"model_config": {"_reset_from": parent_session_id} if parent_session_id else None,
}
def _create_session_row(self, session_key, db_create_kwargs, origin, display_name, *, log) -> None:
"""INSERT a session row and record its routing peer; ``log(exc)`` on failure.
A failed create is a routing hazard (visible warning), but the row is
self-healed with full identity by the next per-turn peer refresh.
"""
"""INSERT a session row and record its routing peer; ``log(exc)`` on failure. A failed
create is a routing hazard (visible warning), but the row is self-healed with full identity
by the next per-turn peer refresh."""
try:
self._db_for_key(session_key).create_session(**db_create_kwargs)
self._record_gateway_session_peer(
db_create_kwargs["session_id"],
session_key,
origin,
display_name=display_name,
)
db_create_kwargs["session_id"], session_key, origin, display_name=display_name)
except Exception as e:
log(e)
gateway/session_stall.py
+15 -41
View File
@@ -1,50 +1,36 @@
"""Gateway session stall notification policy.
Consumes the shared activity observation contract from ``agent.session_activity``
/ ``AIAgent.get_activity_summary()`` as the **single progress source**. This
module owns only the notify-once policy for "pending inbound + stale progress";
it never derives a parallel progress clock from turn-start or inbound timestamps.
Boundaries (keep separate): ``gateway/shutdown_watchdog.py`` is process /
event-loop liveness; ``gateway/delivery_ledger.py`` is outbound delivery
obligations. Pending inbound here is a stall *policy gate* (a queued follow-up
exists), not an obligation and not a progress timestamp. Timeout / kill / retry
policy stay in their own components.
Consumes ``AIAgent.get_activity_summary()`` as the **single progress source** and owns
only the notify-once policy for "pending inbound + stale progress"; it never derives a
parallel progress clock from turn-start or inbound timestamps. Pending inbound is a
stall *policy gate*, not a delivery obligation and not process liveness.
"""
from __future__ import annotations
import math
import time
from typing import Any, Mapping, Optional
def should_emit_session_stall_notification(
*,
timeout_seconds: float,
idle_seconds: Optional[float],
has_pending_inbound: bool,
*, timeout_seconds: float, idle_seconds: Optional[float], has_pending_inbound: bool,
already_notified: bool,
) -> bool:
"""Return True when a stall warning should be sent for this session."""
return (
timeout_seconds > 0
and has_pending_inbound
and not already_notified
and idle_seconds is not None
and idle_seconds >= timeout_seconds
timeout_seconds > 0 and has_pending_inbound and not already_notified
and idle_seconds is not None and idle_seconds >= timeout_seconds
)
def should_clear_session_stall_notification(
*,
timeout_seconds: float,
idle_seconds: Optional[float],
has_pending_inbound: bool,
*, timeout_seconds: float, idle_seconds: Optional[float], has_pending_inbound: bool,
) -> bool:
"""Return True when a prior stall notice may be cleared (episode ended)."""
if not has_pending_inbound or timeout_seconds <= 0:
return True
# Unknown progress: hold the latch. Do not treat observation gaps as recovery.
# Unknown progress holds the latch: observation gaps are not recovery.
return idle_seconds is not None and idle_seconds < timeout_seconds
@@ -66,30 +52,18 @@ def _finite_float(value: Any) -> Optional[float]:
def resolve_session_idle_seconds_from_activity(
activity: Optional[Mapping[str, Any]],
*,
now: Optional[float] = None,
activity: Optional[Mapping[str, Any]], *, now: Optional[float] = None,
) -> Optional[float]:
"""Idle seconds from a shared activity snapshot only.
Prefers ``seconds_since_activity`` when present and finite; otherwise derives
from ``last_activity_at`` / ``last_activity_ts``. Returns None when there is
no usable progress timestamp — callers must not fall back to turn-start or
pending-inbound clocks.
"""
"""Idle seconds from a shared activity snapshot: a finite ``seconds_since_activity``, else
derived from ``last_activity_at`` / ``last_activity_ts``. None when there is no usable
progress timestamp — callers must not fall back to turn-start or inbound clocks."""
if not activity:
return None
idle = _finite_float(activity.get("seconds_since_activity"))
if idle is not None:
return max(0.0, idle)
ts = activity.get("last_activity_at")
when = _finite_float(activity.get("last_activity_ts") if ts is None else ts)
if when is None:
return None
if now is None:
import time as _time
now = _time.time()
return max(0.0, float(now) - when)
return max(0.0, float(time.time() if now is None else now) - when)
gateway/session_state.py
+55 -126
View File
@@ -1,26 +1,7 @@
"""Per-session gateway state consolidated into one container.
GatewayRunner historically carried ~19 separate ``Dict[str, ...]`` attributes
keyed by session_key, each with an ad-hoc lifecycle. That shape bred three
bug classes, all structurally closed here:
1. Boundary drift — hand-copied pop-lists at conversation boundaries went
stale when a dict was added. Now one ``ConversationState.clear()``.
2. Turn-release drift — ad-hoc ``del self._running_agents[key]`` sites popped
different subsets of the turn dicts. Now ``TurnState.clear()``.
3. Wholesale-reset races — lazy-init ``self._x = {}`` replaced the ENTIRE dict,
discarding concurrent sessions' entries. Resets now touch one field of
one ``SessionState``.
Scopes follow where each dict was CLEARED: ``turn`` resets at the end of every
running turn; ``conversation`` at conversation boundaries (/new, /resume,
auto-reset, expiry, compression-exhausted reset); ``persistent`` fields have
their own lifecycles and ``run_generation`` is monotonic and NEVER reset.
Entries in ``GatewayRunner._sessions`` are never evicted (matching the old
dicts, which also leaked empty/stale entries for dead sessions). Eviction of
fully-default SessionStates is possible follow-up work.
"""
"""Per-session gateway state in one container (replaces ~19 session_key-keyed dicts on
GatewayRunner that bred boundary drift and wholesale-reset races). Scopes follow where each dict
was CLEARED: ``turn`` at the end of every turn; ``conversation`` at conversation boundaries (/new,
/resume, auto-reset, expiry); ``persistent`` fields have their own lifecycles."""
from __future__ import annotations
@@ -28,91 +9,63 @@ from collections.abc import MutableMapping
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Iterator, List, NamedTuple, Optional, Tuple
# Presence-sensitive sentinel: /fast stores "priority" or None (explicit
# normal), so key PRESENCE — not value truthiness — decides whether the
# override applies. ``_UNSET_TIER`` means "no override recorded".
# /fast stores "priority" or None (explicit normal), so key PRESENCE decides, not truthiness.
_UNSET_TIER = object()
SERVICE_TIER_UNSET = _UNSET_TIER # public alias
@dataclass
class TurnState:
"""State scoped to one running gateway turn.
"""State scoped to one running gateway turn. ``lease_token`` / ``lease_generation``
are NOT touched by ``clear()``: ``_release_turn_lease`` owns them (release exactly once)."""
``clear()`` runs at every site that ends a running turn. ``lease_token`` /
``lease_generation`` are deliberately NOT cleared by it — they are owned by
``_release_turn_lease``, which must release the registry lease exactly once
per acquiring turn.
"""
# Running AIAgent instance (or _AGENT_PENDING_SENTINEL); None = idle.
agent: Any = None
agent: Any = None # running AIAgent (or _AGENT_PENDING_SENTINEL); None = idle
started_ts: float = 0.0 # 0.0 = not running
lease: Any = None # cross-process active-session slot lease
busy_ack_ts: float = 0.0 # debounce; 0.0 = never acked
# Held turn-lease token + the run generation that acquired it. The pair
# replaces the old (session_key, generation)-keyed dict so a stale unwind
# can never free a newer turn's lease: release/rebind only match when the
# generation is current.
# Held turn-lease token + acquiring generation: release/rebind match only when the
# generation is current, so a stale unwind can never free a newer turn's lease.
lease_token: Any = None
lease_generation: Optional[int] = None
def clear(self) -> None:
"""Reset the per-turn slot (agent / start ts / lease / busy-ack).
The caller pops ``lease`` first so it can call ``lease.release()``.
"""
self.agent = None
self.started_ts = 0.0
self.lease = None
self.busy_ack_ts = 0.0
"""Reset the per-turn slot. The caller pops ``lease`` first to release it."""
self.agent = self.lease = None
self.started_ts = self.busy_ack_ts = 0.0
@dataclass
class ConversationState:
"""State scoped to one conversation (survives turns, not boundaries)."""
# /model per-session override (model/provider/api_key/base_url/api_mode).
model_override: Optional[Dict[str, Any]] = None
model_override: Optional[Dict[str, Any]] = None # /model per-session override
one_turn_restore: Optional[Dict[str, Any]] = None # /model --once snapshot
reasoning_override: Optional[Dict[str, Any]] = None # /reasoning override
# /fast per-session override: "priority" or None; _UNSET_TIER = absent.
service_tier_override: Any = _UNSET_TIER
service_tier_override: Any = _UNSET_TIER # /fast: "priority" or None; _UNSET_TIER = absent
last_resolved_model: str = "" # last successfully-resolved non-empty model
queued_events: List[Any] = field(default_factory=list) # /queue overflow FIFO (adapter slot holds the head)
queued_events: List[Any] = field(default_factory=list) # /queue overflow FIFO (head in adapter)
sidecar_notes: List[str] = field(default_factory=list) # one-shot must-deliver notes
ephemeral_pin: Optional[Tuple[Any, ...]] = None # pinned session-context bytes: (change_key, text)
ephemeral_pin: Optional[Tuple[Any, ...]] = None # pinned session-context (change_key, text)
vc_last: Optional[str] = None # last voice-channel context delivered
def clear(self) -> None:
"""Reset every field to its default — adding a field here means every
conversation boundary clears it automatically."""
"""Reset every field to its default, so new fields are cleared automatically."""
self.__dict__.update(ConversationState().__dict__)
@dataclass
class PersistentState:
"""State with its own lifecycle — NOT cleared wholesale by turn or boundary
resets (approvals/update prompts ARE cleared by the boundary *security*
funnel, but individually)."""
"""State with its own lifecycle — NOT cleared wholesale by turn or boundary resets
(approvals/update prompts ARE cleared, individually, by the boundary security funnel)."""
approvals: Optional[Dict[str, Any]] = None # {"command": ..., "pattern_key": ...}
update_prompt_pending: bool = False # /update prompt awaiting a reply
native_image_paths: List[str] = field(default_factory=list) # consumed one-shot
# Legacy runner-level pending message text (write-mostly; flushed to disk on
# shutdown). Distinct from the adapter-level ``_pending_messages``
# (Dict[str, MessageEvent]) in gateway/base.py, which shares the old name.
# Legacy runner-level pending text (flushed on shutdown); not the adapter-level one.
pending_command_text: Optional[str] = None
# Monotonic run-generation counter. NEVER reset: stale-run detection depends on it.
run_generation: int = 0
# Consecutive session-hygiene compression failures. The in-agent compressor's
# own timeout ladder is unreachable from the gateway (hygiene builds a FRESH
# AIAgent per run and bind_session_state() zeroes that counter), so the streak
# lives here and lets hygiene escalate its cooldown. Reset on a successful
# compression only. PROCESS-LOCAL by design: no disk flush, so a restart drops
# escalation to rung 1 while the DB-backed deadline survives; gateway.run
# mirrors it to the DB keyed by session_key (not session_id) so it also holds
# across compaction ROTATION, where the sid changes but the chat does not.
run_generation: int = 0 # monotonic; NEVER reset (stale-run detection depends on it)
# Consecutive hygiene compression failures (the in-agent ladder is unreachable: hygiene builds
# a FRESH AIAgent per run). Reset on success; process-local, mirrored to the DB by run.py.
hygiene_failure_streak: int = 0
@@ -125,20 +78,12 @@ class SessionState:
persistent: PersistentState = field(default_factory=PersistentState)
# ---------------------------------------------------------------------------
# Legacy dict-view adapters.
#
# Dozens of tests construct bare runners (object.__new__) and read/write the
# old dict attributes directly (``runner._running_agents = {}``, ``assert key
# in runner._pending_approvals``). Each view is a LIVE MutableMapping over one
# SessionState field across all sessions. Production code in gateway/run.py
# uses ``self._session_state(key).<scope>.<field>``.
# ---------------------------------------------------------------------------
# --- Legacy dict-view adapters: tests read/write the old dict attributes directly
# (``runner._running_agents = {}``); each view is a LIVE MutableMapping over one field.
class _FieldSpec(NamedTuple):
"""One legacy dict: scope attr, field name, default factory, presence test."""
scope: str
name: str
default: Callable[[], Any]
@@ -146,8 +91,7 @@ class _FieldSpec(NamedTuple):
def _spec(scope: str, name: str, default: Any) -> _FieldSpec:
"""``default`` is either a type (presence = truthiness) or a sentinel value
such as ``None`` / ``_UNSET_TIER`` (presence = ``is not`` sentinel)."""
"""``default`` is a type (presence = truthiness) or a sentinel (presence = ``is not``)."""
if isinstance(default, type):
return _FieldSpec(scope, name, default, bool)
return _FieldSpec(scope, name, lambda: default, lambda v: v is not default)
@@ -167,16 +111,11 @@ class _RunnerView(MutableMapping):
def __len__(self) -> int:
return sum(1 for _ in self)
# Mapping doesn't provide __eq__; tests compare against plain dicts.
def __eq__(self, other: object) -> bool:
def __eq__(self, other: object) -> bool: # Mapping has no __eq__; tests compare to dicts
if isinstance(other, (dict, MutableMapping)):
return dict(self.items()) == dict(other)
return NotImplemented
def __ne__(self, other: object) -> bool:
result = self.__eq__(other)
return NotImplemented if result is NotImplemented else not result
class SessionFieldView(_RunnerView):
"""Live dict-like view of one SessionState field across sessions."""
@@ -193,29 +132,31 @@ class SessionFieldView(_RunnerView):
def _set(self, state: SessionState, value: Any) -> None:
setattr(getattr(state, self._spec.scope), self._spec.name, value)
def __getitem__(self, key: str) -> Any:
def _present(self, key: Any) -> Optional[SessionState]:
"""The session state for ``key`` if its field is present, else None."""
state = self._sessions().get(key)
if state is None or not self._spec.is_present(value := self._value(state)):
return state if state is not None and self._spec.is_present(self._value(state)) else None
def _held(self, key: str) -> SessionState:
state = self._present(key)
if state is None:
raise KeyError(key)
return value
return state
def __getitem__(self, key: str) -> Any:
return self._value(self._held(key))
def __setitem__(self, key: str, value: Any) -> None:
self._set(self._runner._session_state(key), value)
def __delitem__(self, key: str) -> None:
state = self._sessions().get(key)
if state is None or not self._spec.is_present(self._value(state)):
raise KeyError(key)
self._set(state, self._spec.default())
self._set(self._held(key), self._spec.default())
def __iter__(self) -> Iterator[str]:
for key, state in list(self._sessions().items()):
if self._spec.is_present(self._value(state)):
yield key
return (k for k in list(self._sessions()) if self._present(k) is not None)
def __contains__(self, key: object) -> bool:
state = self._sessions().get(key) # type: ignore[arg-type]
return state is not None and self._spec.is_present(self._value(state))
return self._present(key) is not None
def clear(self) -> None: # avoid MutableMapping's popitem loop
for state in list(self._sessions().values()):
@@ -226,26 +167,22 @@ class SessionFieldView(_RunnerView):
class TurnLeaseTokenView(_RunnerView):
"""Legacy view of ``_turn_lease_tokens``: keyed by (session_key, generation).
The pair lives on ``TurnState.lease_token`` / ``lease_generation``; the lease
registry serializes acquisition per session, so at most one held token
exists per session key and the single slot equals the old tuple-keyed dict.
"""
"""Legacy (session_key, generation)-keyed view of ``_turn_lease_tokens``; the lease registry
serializes acquisition per session, so one ``TurnState`` slot per key equals the old dict."""
__slots__ = ()
def _held(self, key: Any) -> Tuple[Any, SessionState]:
"""Return (session_key, state) for a currently-held (key, gen) or raise KeyError."""
def _held(self, key: Any) -> TurnState:
"""TurnState for a currently-held (session_key, generation) or raise KeyError."""
if not isinstance(key, tuple) or len(key) != 2:
raise KeyError(key)
state = self._sessions().get(key[0])
if state is None or state.turn.lease_token is None or state.turn.lease_generation != key[1]:
raise KeyError(key)
return key[0], state
return state.turn
def __getitem__(self, key: Any) -> Any:
return self._held(key)[1].turn.lease_token
return self._held(key).lease_token
def __setitem__(self, key: Any, value: Any) -> None:
if not isinstance(key, tuple) or len(key) != 2:
@@ -254,13 +191,12 @@ class TurnLeaseTokenView(_RunnerView):
turn.lease_token, turn.lease_generation = value, key[1]
def __delitem__(self, key: Any) -> None:
turn = self._held(key)[1].turn
turn = self._held(key)
turn.lease_token = turn.lease_generation = None
def __iter__(self) -> Iterator[Tuple[str, Any]]:
for key, state in list(self._sessions().items()):
if state.turn.lease_token is not None:
yield (key, state.turn.lease_generation)
return ((k, s.turn.lease_generation) for k, s in list(self._sessions().items())
if s.turn.lease_token is not None)
def clear(self) -> None: # avoid MutableMapping's popitem loop
for key in list(self):
@@ -291,19 +227,13 @@ LEGACY_FIELD_SPECS: Dict[str, _FieldSpec] = {
def _legacy_property(make_view: Callable[[Any], MutableMapping], doc: str) -> property:
"""Dict-shaped @property over a live view.
Getter returns the view; setter accepts a plain dict (the ubiquitous test
pattern ``runner._X = {...}``), resetting the field on every known session
and then applying the given entries; ``del runner._X`` (older tests
simulating a runner without the attribute) means "no entries".
"""
"""Dict-shaped @property over a live view; the setter takes a plain dict (test pattern
``runner._X = {...}``): reset the field on every session, then apply the entries."""
def fset(self: Any, mapping: Optional[Dict[Any, Any]]) -> None:
view = make_view(self)
view.clear()
for key, value in (mapping or {}).items():
view[key] = value
view.update(mapping or {})
return property(make_view, fset, lambda self: make_view(self).clear(), doc=doc)
@@ -313,8 +243,7 @@ def legacy_dict_property(attr_name: str) -> property:
spec = LEGACY_FIELD_SPECS[attr_name]
return _legacy_property(
lambda self: SessionFieldView(self, spec),
f"Legacy dict view over SessionState.{spec.scope}.{spec.name} "
"(kept for tests that access the pre-SessionState attribute).",
f"Legacy dict view over SessionState.{spec.scope}.{spec.name} (for pre-SessionState tests).",
)
gateway/session_transcript.py
+132 -249
View File
@@ -1,8 +1,6 @@
"""SessionStore transcript I/O: SQLite append with a per-session retry queue,
compression-reroute following, FTS corruption recovery, rewrite/rewind/load.
Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.
"""
"""SessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute
following, FTS corruption recovery, rewrite/rewind/load. Mixin split out of ``gateway/session.py``;
bound onto ``SessionStore`` via the MRO."""
from __future__ import annotations
@@ -18,6 +16,14 @@ if TYPE_CHECKING:
logger = logging.getLogger("gateway.session")
class TranscriptReadError(RuntimeError):
"""Raised when persisted history cannot be read safely."""
def __init__(self, session_id: str) -> None:
self.session_id = session_id
super().__init__(f"transcript read failed for session {session_id}")
def _plain_text(content) -> str:
"""Text of a message content (str or text-part list); "" for anything else."""
if isinstance(content, list):
@@ -26,15 +32,25 @@ def _plain_text(content) -> str:
return content if isinstance(content, str) else ""
def _spool_dropped(session_id: str, message: Dict[str, Any]):
"""Spool one evicted/undeliverable message to disk (same machinery as the shutdown flush, so it
is replayed after DB recovery); path or None."""
try:
from gateway.shutdown_flush import spool_dropped_transcript_message
return spool_dropped_transcript_message(session_id, message)
except Exception:
return None
class SessionTranscriptMixin:
"""SessionStore transcript I/O: SQLite append with a per-session retry queue,
compression-reroute following, FTS corruption recovery,
rewrite/rewind/load.
"""
compression-reroute following, FTS corruption recovery, rewrite/rewind/load."""
_MAX_PENDING_PER_SESSION = 200 # in-memory pending messages per session (DB broken)
def _compression_tip_for_session_id(self, session_id: Optional[str]) -> Optional[str]:
"""Latest compression continuation for *session_id* (heals a mapping
left pointing at a compressed parent by a restart or failed send)."""
"""Latest compression continuation for *session_id* (heals a mapping left pointing at a
compressed parent by a restart or failed send)."""
if not session_id:
return session_id
db = self._db_for_session_id(session_id)
@@ -47,11 +63,8 @@ class SessionTranscriptMixin:
return session_id
def _heal_compression_tip_locked(
self,
entry: "SessionEntry",
original_session_id: Optional[str],
canonical_session_id: Optional[str],
) -> bool:
self, entry: "SessionEntry", original_session_id: Optional[str],
canonical_session_id: Optional[str]) -> bool:
"""Rewrite *entry* to the compression continuation if stale. Lock held."""
if (
not original_session_id
@@ -61,28 +74,20 @@ class SessionTranscriptMixin:
):
return False
logger.info(
"SessionStore healed compressed session mapping: %s -> %s",
entry.session_id,
canonical_session_id,
)
"SessionStore healed compressed session mapping: %s -> %s", entry.session_id,
canonical_session_id)
entry.session_id = canonical_session_id
return True
def advance_compression_session(
self,
session_key: str,
expected_session_id: str,
target_session_id: str,
self, session_key: str, expected_session_id: str, target_session_id: str,
) -> Optional[SessionEntry]:
"""CAS-advance one route along an already-verified compression lineage.
Unlike ``switch_session`` this never ends/reopens SQLite rows (the
compression transaction owns that). ``None`` means the route moved
after the caller's snapshot (e.g. /new) — caller must fail closed.
"""
"""CAS-advance one route along an already-verified compression lineage. Unlike
``switch_session`` this never ends/reopens SQLite rows (the compression transaction owns
that). ``None`` means the route moved after the caller's snapshot (e.g. /new) — caller
must fail closed."""
if not session_key or not expected_session_id or not target_session_id:
return None
with self._lock:
entry = self._entry_locked(session_key)
if entry is None:
@@ -91,14 +96,9 @@ class SessionTranscriptMixin:
return entry
if entry.session_id != expected_session_id:
return None
if not self._heal_compression_tip_locked(
entry,
expected_session_id,
target_session_id,
):
if not self._heal_compression_tip_locked(entry, expected_session_id, target_session_id):
return None
# Bookkeeping, not user activity: leave ``updated_at`` alone.
self._save()
self._save() # bookkeeping, not user activity: leave ``updated_at`` alone
return entry
def _get_transcript_drain_lock(self):
@@ -110,9 +110,7 @@ class SessionTranscriptMixin:
if not self._db_for_session_id(session_id) or skip_db:
return
with self._get_transcript_drain_lock():
self._append_to_transcript_serialized(
self._follow_reroutes(session_id), message
)
self._append_to_transcript_serialized(self._follow_reroutes(session_id), message)
def _follow_reroutes(self, session_id: str) -> str:
"""Follow the compression reroute chain (cycle-guarded)."""
@@ -123,56 +121,34 @@ class SessionTranscriptMixin:
session_id = reroutes[session_id]
return session_id
def _spool_dropped(self, session_id: str, message: Dict[str, Any]):
"""Spool one evicted/undeliverable message to disk; path or None."""
try:
from gateway.shutdown_flush import spool_dropped_transcript_message
return spool_dropped_transcript_message(session_id, message)
except Exception:
return None
def _enqueue_transcript_message(self, session_id: str, message: Dict[str, Any]) -> list:
"""Queue *message* (retry lock held); evicts + spools the oldest past the cap.
Spooling uses the same machinery as shutdown flush so the message is
replayed after DB recovery instead of being lost.
"""
"""Queue *message* (retry lock held); evicts + spools the oldest past the cap."""
pending = self._dirty_transcripts.setdefault(session_id, [])
pending.append(dict(message))
if len(pending) > self._MAX_PENDING_PER_SESSION:
spool_path = self._spool_dropped(session_id, pending.pop(0))
spool_path = _spool_dropped(session_id, pending.pop(0))
if spool_path is not None:
self._lazy("_spooled_drop_sessions", set).add(session_id)
logger.warning(
"Session DB transcript pending queue full for %s "
"(cap=%d); spooled oldest message to %s for replay "
"after DB recovery",
session_id, self._MAX_PENDING_PER_SESSION, spool_path,
)
"Session DB transcript pending queue full for %s (cap=%d); spooled oldest "
"message to %s for replay after DB recovery", session_id,
self._MAX_PENDING_PER_SESSION, spool_path)
else:
logger.warning(
"Session DB transcript pending queue full for %s "
"(cap=%d); dropping oldest message to make room "
"(on-disk spool unavailable)",
session_id, self._MAX_PENDING_PER_SESSION,
)
"Session DB transcript pending queue full for %s (cap=%d); dropping oldest "
"message to make room (on-disk spool unavailable)", session_id,
self._MAX_PENDING_PER_SESSION)
return pending
def _divert_transcript_after_db_replaced(
self, session_id: str, queue_session_id: str, exc: Exception
) -> None:
"""Stop SQLite writes on a replaced/quarantined handle and divert the backlog.
Retrying cannot succeed and the FTS rebuild must never run on this
handle; the pending queue goes to the on-disk spool + JSONL fallback.
"""
"""Stop SQLite writes on a replaced/quarantined handle and divert the backlog to the on-disk
spool + JSONL fallback: retrying cannot succeed and the FTS rebuild must never run here."""
logger.error(
"Session DB refused further writes on this handle for "
"%s (%s); stopping SQLite writes and diverting pending "
"transcripts to the on-disk fallback: %s",
session_id, type(exc).__name__, exc,
)
"Session DB refused further writes on this handle for %s (%s); stopping SQLite writes "
"and diverting pending transcripts to the on-disk fallback: %s", session_id,
type(exc).__name__, exc)
with self._transcript_retry_lock:
remaining = list(self._dirty_transcripts.get(queue_session_id, []))
self._dirty_transcripts.pop(queue_session_id, None)
@@ -180,34 +156,24 @@ class SessionTranscriptMixin:
for dropped in remaining:
try:
from gateway.shutdown_flush import spool_dropped_transcript_message
spool_dropped_transcript_message(session_id, dropped)
except Exception:
logger.warning(
"pending fallback failed for replaced "
"state.db transcript on %s",
session_id,
exc_info=True,
)
"pending fallback failed for replaced state.db transcript on %s", session_id,
exc_info=True)
try:
from hermes_state import divert_session_transcript_jsonl
divert_session_transcript_jsonl(session_id, remaining)
except Exception:
logger.warning(
"JSONL divert failed for replaced state.db "
"transcript on %s",
session_id,
exc_info=True,
)
"JSONL divert failed for replaced state.db transcript on %s", session_id,
exc_info=True)
def _live_compression_child(self, session_id: str) -> str:
"""Transitive compression tip of *session_id* if it is a different, still-live
row, else "" (a depth-1 lookup misses multi-hop lineages).
Uses the PARENT's proven owner handle: the child's id is not published
until after its write succeeds, so a by-id lookup would fall back to
the ambient store.
"""
"""Transitive compression tip of *session_id* if it is a different, still-live row, else ""
(a depth-1 lookup misses multi-hop lineages). Uses the PARENT's proven owner handle: the
child's id is unpublished until its write succeeds, so a by-id lookup would hit the ambient
store."""
owner_db = self._db_for_session_id(session_id)
if owner_db is None:
return ""
@@ -221,13 +187,10 @@ class SessionTranscriptMixin:
def _migrate_transcript_queue_to_child(
self, session_id: str, queue_session_id: str, child_id: str, pending: list, msg
) -> list:
"""Move the retry queue + failure counter from parent to child and publish
the reroute (retry lock held). Returns the child's pending list.
Older parent backlog must precede messages already queued directly on
the child. Routing is published only AFTER the queue moved (caller), so
new child writes cannot bypass older parent backlog.
"""
"""Move the retry queue + failure counter from parent to child and record the reroute (retry
lock held); returns the child's pending list. Older parent backlog must precede messages
already queued directly on the child; routing is published only AFTER the queue moved
(caller), so new child writes cannot bypass older parent backlog."""
if pending and pending[0] is msg:
pending.pop(0)
existing_child_pending = self._dirty_transcripts.get(child_id, [])
@@ -240,9 +203,7 @@ class SessionTranscriptMixin:
previous_failures = self._transcript_append_failures.pop(queue_session_id, 0)
if previous_failures:
self._transcript_append_failures[child_id] = max(
previous_failures,
self._transcript_append_failures.get(child_id, 0),
)
previous_failures, self._transcript_append_failures.get(child_id, 0))
self._transcript_reroutes[session_id] = child_id
return pending
@@ -257,11 +218,9 @@ class SessionTranscriptMixin:
if _hints:
_hints.pop(child_id, None)
def _append_to_transcript_serialized(
self, session_id: str, message: Dict[str, Any]
) -> None:
"""Append a message to a session's transcript (SQLite), draining the
per-session retry queue."""
def _append_to_transcript_serialized(self, session_id: str, message: Dict[str, Any]) -> None:
"""Append a message to a session's transcript (SQLite), draining the per-session retry
queue."""
with self._transcript_retry_lock:
pending = self._enqueue_transcript_message(session_id, message)
msg = pending[0]
@@ -283,24 +242,18 @@ class SessionTranscriptMixin:
self._append_transcript_message(session_id, msg)
except Exception as exc:
from hermes_state import (
CompressionSessionClosedError,
StateDbCorruptError,
StateDbReplacedError,
CompressionSessionClosedError, StateDbCorruptError, StateDbReplacedError,
)
if isinstance(exc, (StateDbReplacedError, StateDbCorruptError)):
self._divert_transcript_after_db_replaced(session_id, queue_session_id, exc)
return
if isinstance(exc, CompressionSessionClosedError):
# Adopt only a different, still-live compression tip, else
# fail closed.
# Adopt only a different, still-live compression tip, else fail closed.
_owner_key = self._owner_key_for_session_id(session_id)
child_id = self._live_compression_child(session_id)
if child_id:
# Record the child's owner BEFORE writing to it (the
# reroute is published only after the write succeeds
# — load-bearing for backlog order).
# Record the child's owner BEFORE writing to it (the reroute is published
# only after the write succeeds — load-bearing for backlog order).
if _owner_key:
self._lazy("_session_owner_hints", dict)[child_id] = _owner_key
try:
@@ -310,8 +263,7 @@ class SessionTranscriptMixin:
else:
with self._transcript_retry_lock:
pending = self._migrate_transcript_queue_to_child(
session_id, queue_session_id, child_id, pending, msg
)
session_id, queue_session_id, child_id, pending, msg)
queue_session_id = child_id
self._publish_transcript_reroute(session_id, child_id)
if not pending:
@@ -320,15 +272,13 @@ class SessionTranscriptMixin:
session_id = child_id
continue
else:
# Permanent routing invariant failure, not a transient
# outage: drop it so it cannot poison later writes.
# Permanent routing invariant failure, not a transient outage: drop it so it
# cannot poison later writes.
with self._transcript_retry_lock:
_ack_head()
logger.error(
"Session DB transcript append rejected for compression-ended "
"%s with no unique live child; not retrying",
session_id,
)
"Session DB transcript append rejected for compression-ended %s with "
"no unique live child; not retrying", session_id)
return
if self._is_fts_corruption_error(exc) and self._rebuild_fts_once():
try:
@@ -343,10 +293,8 @@ class SessionTranscriptMixin:
failures = self._transcript_append_failures.get(session_id, 0) + 1
self._transcript_append_failures[session_id] = failures
logger.warning(
"Session DB transcript append failed for %s "
"(failure_count=%d, pending=%d); will retry: %s",
session_id, failures, len(pending), exc,
)
"Session DB transcript append failed for %s (failure_count=%d, pending=%d); "
"will retry: %s", session_id, failures, len(pending), exc)
return
else:
with self._transcript_retry_lock:
@@ -360,22 +308,15 @@ class SessionTranscriptMixin:
continue
def _drain_spooled_drops(self, session_id: str) -> None:
"""Replay cap-dropped spooled transcript messages after DB recovery.
Best-effort: replay failures keep the spool files for the next
successful flush; nothing here may raise into the caller.
"""
"""Replay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay
failures keep the spool files for the next successful flush; nothing here may raise."""
spooled_sessions = getattr(self, "_spooled_drop_sessions", None)
if not spooled_sessions or session_id not in spooled_sessions:
return
try:
from gateway.shutdown_flush import drain_transcript_spool
_replayed, remaining = drain_transcript_spool(
session_id,
lambda message: self._append_transcript_message(
session_id, message
),
session_id, lambda message: self._append_transcript_message(session_id, message),
)
if not remaining:
spooled_sessions.discard(session_id)
@@ -386,11 +327,10 @@ class SessionTranscriptMixin:
"""Write one transcript row. Caller handles retry queuing."""
_db = self._db_for_session_id(session_id)
if _db is None:
# Named profile with no resolvable home yet: defer (caller queues)
# instead of writing into the ambient store.
# Named profile with no resolvable home yet: defer (caller queues) instead of writing
# into the ambient store.
raise RuntimeError(
f"no owning session store for {session_id}; deferring transcript write"
)
f"no owning session store for {session_id}; deferring transcript write")
is_assistant = message.get("role") == "assistant"
_db.append_message(
session_id=session_id,
@@ -407,35 +347,25 @@ class SessionTranscriptMixin:
platform_message_id=(message.get("platform_message_id") or message.get("message_id")),
observed=bool(message.get("observed")),
timestamp=message.get("timestamp"),
# Exact bytes sent to the API (prompt-cache-stable replay); must
# survive every persistence path or the next replay diverges.
# Exact bytes sent to the API (prompt-cache-stable replay); must survive every
# persistence path or the next replay diverges.
api_content=extract_api_content_sidecar(message),
# Presentation typing (e.g. "internal_notification"); DB-only.
display_kind=message.get("display_kind"),
display_metadata=message.get("display_metadata"),
)
_MAX_PENDING_PER_SESSION = 200
@staticmethod
def _is_fts_corruption_error(exc: Exception) -> bool:
"""True only when the failure is provably scoped to the FTS index.
A bare SQLITE_CORRUPT can mean structural B-tree damage; only errors
naming ``messages_fts`` or carrying FTS provenance (per
``SessionDB._is_fts_write_corruption_error``) may authorize the
one-shot rebuild-and-retry. Everything else takes the retry path.
"""
text = str(exc).lower()
if "messages_fts" in text:
"""True only when the failure is provably scoped to the FTS index. A bare SQLITE_CORRUPT
can mean structural B-tree damage; only errors naming ``messages_fts`` or carrying FTS
provenance (``SessionDB._is_fts_write_corruption_error``) may authorize the one-shot
rebuild-and-retry; everything else takes the retry path."""
if "messages_fts" in str(exc).lower():
return True
import sqlite3
from hermes_state import SessionDB
if isinstance(exc, sqlite3.DatabaseError):
return SessionDB._is_fts_write_corruption_error(exc)
return False
return isinstance(exc, sqlite3.DatabaseError) and SessionDB._is_fts_write_corruption_error(exc)
def _rebuild_fts_once(self) -> bool:
"""Attempt FTS5 ``rebuild`` once per store lifetime; True if any index was rebuilt."""
@@ -450,11 +380,9 @@ class SessionTranscriptMixin:
foreign_holders = db._foreign_state_db_holders()
if foreign_holders:
logger.warning(
"Skipping Session DB FTS rebuild while foreign processes "
"hold the database or WAL sidecars (%s); canonical "
"transcript writes remain available.",
foreign_holders,
)
"Skipping Session DB FTS rebuild while foreign processes hold the database or "
"WAL sidecars (%s); canonical transcript writes remain available.",
foreign_holders)
return False
try:
rebuilt = db.rebuild_fts()
@@ -462,10 +390,7 @@ class SessionTranscriptMixin:
logger.warning("Session DB FTS rebuild failed: %s", exc)
return False
if rebuilt:
logger.warning(
"Rebuilt %d Session DB FTS index(es) after append corruption",
rebuilt,
)
logger.warning("Rebuilt %d Session DB FTS index(es) after append corruption", rebuilt)
return rebuilt > 0
def _clear_dirty_transcript(self, session_id: str) -> None:
@@ -474,48 +399,34 @@ class SessionTranscriptMixin:
self._dirty_transcripts.pop(session_id, None)
self._transcript_append_failures.pop(session_id, None)
def has_platform_message_id(
self, session_id: str, platform_message_id: str
) -> bool:
def has_platform_message_id(self, session_id: str, platform_message_id: str) -> bool:
"""Whether a message with this platform_message_id is persisted (False without a DB)."""
db = self._db_for_session_id(session_id)
if not db:
return False
try:
return db.has_platform_message_id(
session_id, platform_message_id
)
return db.has_platform_message_id(session_id, platform_message_id)
except Exception:
logger.debug("has_platform_message_id lookup failed", exc_info=True)
return False
def rewrite_transcript(
self,
session_id: str,
messages: List[Dict[str, Any]],
active_only: bool = False,
reject_active_turn_lease: bool = False,
) -> bool:
"""Replace a session's transcript (/retry, /compress).
DESTRUCTIVE by default: ``active_only=False`` DELETEs every row incl.
soft-archived compaction history (pass ``active_only=True`` for sessions
that may carry archived rows). True when the write lands or there is no
DB, False on failure — callers committing a destructive change on top
(/compress repointing) must check it. ``reject_active_turn_lease`` is
for user-initiated rewrites that do not own the cross-process turn lease.
"""
self, session_id: str, messages: List[Dict[str, Any]], active_only: bool = False,
reject_active_turn_lease: bool = False) -> bool:
"""Replace a session's transcript (/retry, /compress). DESTRUCTIVE by default:
``active_only=False`` DELETEs every row incl. soft-archived compaction history (pass
``active_only=True`` for sessions that may carry archived rows). True when the write lands
or there is no DB, False on failure — callers committing a destructive change on top
(/compress repointing) must check it. ``reject_active_turn_lease`` is for user-initiated
rewrites that do not own the cross-process turn lease."""
db = self._db_for_session_id(session_id)
if not db:
return True
with self._get_transcript_drain_lock():
try:
db.replace_messages(
session_id,
messages,
active_only=active_only,
reject_active_turn_lease=reject_active_turn_lease,
)
session_id, messages, active_only=active_only,
reject_active_turn_lease=reject_active_turn_lease)
except Exception as e:
logger.debug("Failed to rewrite transcript in DB: %s", e)
return False
@@ -523,13 +434,9 @@ class SessionTranscriptMixin:
return True
def load_transcript(self, session_id: str) -> List[Dict[str, Any]]:
"""Load all messages from a session's transcript (state.db is canonical).
Reads follow the same routing writes use: the in-memory reroute map
(compression rotation), then the durable compression tip — otherwise
the transcript "vanishes" while every message sits under the child.
"""
from gateway.session import TranscriptReadError
"""Load all messages from a session's transcript (state.db is canonical). Reads follow the
same routing writes use — the in-memory reroute map, then the durable compression tip —
otherwise the transcript "vanishes" while every message sits under the child."""
if not self._db_for_session_id(session_id):
return []
session_id = self._follow_reroutes(session_id)
@@ -541,59 +448,38 @@ class SessionTranscriptMixin:
except Exception:
pass
try:
# repair_alternation: this feeds LIVE REPLAY; heal a durable
# user;user wedge once here instead of on every request.
# repair_alternation: this feeds LIVE REPLAY; heal a durable user;user wedge once here.
return self._db_for_session_id(session_id).get_messages_as_conversation(
session_id, repair_alternation=True
)
session_id, repair_alternation=True)
except Exception as e:
# Empty history is valid data; a failed canonical read is not —
# live-replay callers must fail closed, not start from [].
# Empty history is valid data; a failed canonical read is not — live-replay callers
# must fail closed, not start from [].
logger.error(
"Transcript read failed for session %s; refusing to treat the "
"conversation as empty: %s",
session_id,
e,
exc_info=True,
)
"Transcript read failed for session %s; refusing to treat the conversation as "
"empty: %s", session_id, e, exc_info=True)
raise TranscriptReadError(session_id) from e
def rewind_session(
self,
session_id: str,
n: int = 1,
*,
require_retryable_composite: bool = False,
self, session_id: str, n: int = 1, *, require_retryable_composite: bool = False,
) -> Optional[Dict[str, Any]]:
"""Back up ``n`` user turns via soft-delete (``active=0``), mirroring CLI ``/undo [N]``.
Returns ``{"rewound_count", "turns_undone", "target_text"}`` or ``None``
(no DB / no user turn). ``n`` clamps to the oldest user turn.
``require_retryable_composite`` is the gateway ``/retry`` guard: the
selected turn must be a composite carrier whose live payload is
losslessly replayable as text before anything changes.
"""
Returns ``{"rewound_count", "turns_undone", "target_text"}`` or ``None`` (no DB / no user
turn); ``n`` clamps to the oldest user turn. ``require_retryable_composite`` is the gateway
``/retry`` guard: the selected turn must be a composite carrier whose live payload is
losslessly replayable as text before anything changes."""
db = self._db_for_session_id(session_id)
if not db:
return None
with self._get_transcript_drain_lock():
if n < 1:
n = 1
n = max(n, 1)
from agent.context_compressor import (
retryable_user_text,
split_user_originated_turn,
user_originated_turn_view,
retryable_user_text, split_user_originated_turn, user_originated_turn_view,
)
try:
expected_active_ids = db.get_active_message_ids(session_id)
durable = db.get_messages_as_conversation(
session_id,
include_row_ids=True,
)
durable = db.get_messages_as_conversation(session_id, include_row_ids=True)
user_indices = [
index
for index, message in enumerate(durable)
index for index, message in enumerate(durable)
if user_originated_turn_view(message) is not None
]
if not user_indices:
@@ -612,17 +498,14 @@ class SessionTranscriptMixin:
logger.debug("rewind_session: failed to resolve canonical target: %s", e)
return None
if require_retryable_composite:
# Keep replay-policy failures distinct from persistence errors
# so /retry can explain why the selected carrier is unsafe.
# Keep replay-policy failures distinct from persistence errors so /retry can explain
# why the selected carrier is unsafe.
target_text = retryable_user_text(target_view.get("content"))
try:
result = db.rewind_to_message(
session_id,
target_id,
preserve_compaction_handoff=handoff is not None,
session_id, target_id, preserve_compaction_handoff=handoff is not None,
expected_active_ids=expected_active_ids,
expected_target_content=target_view.get("content"),
)
expected_target_content=target_view.get("content"))
except ValueError as e:
logger.debug("rewind_session: %s", e)
return None
@@ -630,8 +513,8 @@ class SessionTranscriptMixin:
logger.debug("rewind_session: rewind_to_message failed: %s", e)
return None
self._clear_dirty_transcript(session_id)
# ``target_view`` is the live projection; a composite carrier's raw
# row holds the summary wrapper and must not be echoed as prompt.
# ``target_view`` is the live projection; a composite carrier's raw row holds the
# summary wrapper and must not be echoed as prompt.
if not require_retryable_composite:
target_text = _plain_text(target_view.get("content") or "")
return {