"""Read-only session browsing, latest-descendant lookup and owner maintenance. """ import logging import asyncio import time from contextlib import contextmanager from pathlib import Path from typing import Dict, Optional from hermes_state_common import _RESET_CHILD_SQL, _sql_json_extract # Same logger the code used before extraction (record parity). _log = logging.getLogger("hermes_cli.web_server") _DESCENDANTS_SQL = f""" WITH RECURSIVE descendants(id, parent_session_id, started_at) AS ( SELECT id, parent_session_id, started_at FROM sessions WHERE id = ? UNION SELECT s.id, s.parent_session_id, s.started_at FROM sessions s JOIN descendants d ON s.parent_session_id = d.id -- Continuation edges only (same predicate as the session list's chain CTE): a subagent run, -- a /branch fork, a /new reset child or a tool-owned row is its own conversation, and resuming -- INTO one parks the user's chat in a row the sidebar never lists (#115092). WHERE {_sql_json_extract('s.model_config', '$._delegate_from')} IS NULL AND {_sql_json_extract('s.model_config', '$._branched_from')} IS NULL AND NOT ({_RESET_CHILD_SQL.format(a='s')}) AND COALESCE(s.source, '') != 'tool' ) SELECT id, parent_session_id, started_at FROM descendants """ # Desktop 0.22.2's ``readableError`` keeps the HTTP ``detail`` and shows its "app and backend are # on different versions" toast (with an Update action) for "out of sync (different versions)". CLIENT_UPGRADE_REQUIRED = ('client_upgrade_required: this app is out of sync (different versions) ' 'with the Hermes backend; update the app to edit sessions') def _session_latest_descendant(session_id: str, db): """Resolve a session id to the newest child leaf session. /model may create child sessions; a dashboard refresh should continue the newest child instead of reopening the old parent. Returns ``(leaf, path)``. """ sid = db.resolve_session_id(session_id) if not sid or not db.get_session(sid): return None, [] conn = getattr(db, "_conn", None) if conn is not None: keys = ("id", "parent_session_id", "started_at") rows = [dict(zip(keys, row)) for row in conn.execute(_DESCENDANTS_SQL, (sid,)).fetchall()] else: rows = db.list_sessions_rich(limit=10000, offset=0, compact_rows=True) children = {} for row in rows: rid = row.get("id") parent = row.get("parent_session_id") if rid and parent: children.setdefault(parent, []).append(row) def started(row): try: return float(row.get("started_at") or 0) except Exception: return 0.0 current = sid path = [sid] seen = {sid} while children.get(current): candidates = [r for r in children[current] if r.get("id") not in seen] if not candidates: break candidates.sort(key=started, reverse=True) current = candidates[0]["id"] path.append(current) seen.add(current) return current, path def _session_mutation_context(request, profile): """Use the same principal and profile boundary for prepare and apply.""" from fastapi import HTTPException from gateway.session_contract import SessionActor from hermes_cli.web_server import _has_valid_session_token from hermes_cli.web_server_cron import _cron_profile_home from hermes_state import _default_db_path native = getattr(request.state, 'native_http_principal', None) session = getattr(request.state, 'session', None) if native is not None: subject = native['subject'] elif session is not None: from gateway.owner_identity import principal_subject subject = principal_subject({"user_id": session.user_id, "provider": session.provider, "issuer": session.issuer}) elif not getattr(request.app.state, 'auth_required', False) and _has_valid_session_token(request): from gateway.owner_identity import host_owner_subject subject = host_owner_subject() else: raise HTTPException(status_code=401, detail='Unauthorized') authority = getattr(request.app.state, 'session_authority', None) home = Path(_cron_profile_home(profile)[1]) if profile else Path(_default_db_path()).parent capabilities = frozenset({'session:read', 'session:control', 'session:create', 'session:operator'}) if authority is None: if getattr(request.app.state, 'gateway_runner', None) is not None: raise HTTPException(status_code=503, detail='session_authority_unavailable') # Standalone ``serve``/``dashboard`` (no composed gateway; its owner cutover is a # follow-up): the store is the owner. Same principal id form as a gateway authority's # (``str(home)``), so import bindings written offline stay valid once one starts. return None, SessionActor(subject, str(home.resolve()), capabilities, 'http') # ``?profile=`` selects a served home; its own authority (not the launch one) owns the mutation. from gateway.session_authorities import authority_for_home runner = getattr(request.app.state, 'gateway_runner', None) owner = authority_for_home(runner, home) if runner is not None else None if owner is None and home.resolve() == Path(authority.db.db_path).parent.resolve(): owner = authority if owner is None: raise HTTPException(status_code=403, detail='profile_mismatch') authority = owner if native is not None and native['profile_id'] != authority.profile_id: raise HTTPException(status_code=403, detail='profile_mismatch') if native is not None: # One transport per verified native request, so the grant's own profile can claim its old # local history exactly as over the native socket (``require_history_claim``); a shared # 'http' id never can. import uuid return authority, SessionActor(subject, authority.profile_id, capabilities, 'native-http:' + uuid.uuid4().hex) return authority, SessionActor(subject, authority.profile_id, capabilities, 'http') @contextmanager def _native_history_transport(request, authority, actor): """Bind a verified native HTTP actor as a native legacy transport for one mutation.""" native = None if authority is not None and str(getattr(actor, 'transport_id', '')).startswith('native-http:'): native = getattr(request.state, 'native_http_principal', None) if native is None: yield return from gateway.session_local_migration import bind_native_transport, unbind_native_transport bind_native_transport(authority, actor, {'provider': 'local', 'profile_id': authority.profile_id, 'instance_id': native['instance_id']}) try: yield finally: unbind_native_transport(authority, actor) def _mutate_offline_store(profile, actor, params): """Standalone edit: the receipt transaction on this process's single shared writer. Runs under the exact owner lock (409 while any gateway owns the store, never a second writer beside it); revision/generation/idle guards are the gateway's own. """ from fastapi import HTTPException from hermes_state_runtime import begin_runtime_epoch, mutate_runtime_session def run(): db = _open_session_db_for_profile(profile, read_only=False) try: write_guard = None if params['operation'] == 'import': normalized, errors = db._validate_import_payload(params['payload']['sessions']) if errors: raise HTTPException(status_code=400, detail={'errors': errors}) from functools import partial from hermes_state_mutation_binding import import_history_control write_guard = partial(import_history_control, actor=actor, session_ids=tuple(item['session']['id'] for item in normalized)) with db._read_ctx() as conn: row = conn.execute('SELECT epoch FROM runtime_epoch WHERE singleton=1').fetchone() # The reservation held here is the ownership lock begin_runtime_epoch requires. epoch = row[0] if row else begin_runtime_epoch(db, instance_id='dashboard-offline') return mutate_runtime_session(db, epoch=epoch, principal_id=actor.subject, expected_generation=params.get('expected_generation'), _authorize_write=write_guard, **{key: params[key] for key in ('session_id', 'request_id', 'expected_revision', 'operation', 'payload')}) finally: db.close() return _with_session_maintenance(profile, run) async def _mutate_via_gateway_owner(profile, actor, params): """Forward a standalone dashboard mutation to the live gateway owner. While a gateway serves the same home, a separately launched ``hermes dashboard`` is not the store owner, so the offline path's exclusive-maintenance reservation can never win and every write action dies on a bare 409 (#138915). The owner exposes the same revision-fenced verb over the control socket. With no endpoint the offline store's reservation decides: no endpoint does not prove a live owner (a stale bad-mode socket, a parked secondary read ``inaccessible`` with the lock free). When the reservation loses and discovery read anything but ``absent`` (a control timeout, a profile mismatch, one starting or draining), the edit answers 503 with discovery's reason rather than the bare 409 (#144141). A dial, mint or RPC that dies on an endpoint discovery just identified is not owner-less (a slow first dial, or an owner restart between discovery and mint): the offline reservation would only lose to that same owner and answer with the bare 409 this forward exists to remove, so every transport failure surfaces a retryable 503. Only the owner's refusal frame (``GatewayRPCError``) is passed on as a verdict. The ticket is minted for the dashboard's authenticated principal, not the socket UID: ``authorize_history`` compares the actor subject against the row's stored owner, so a cold row owned by a dashboard account only takes the mutation when the forward acts as that account. """ from fastapi import HTTPException from gateway.runtime_ownership import OwnershipConflict from hermes_cli.gateway_client import GatewayClientError, GatewayRPCError, connect_gateway from hermes_cli.gateway_runtime import discover_gateway_endpoint from hermes_cli.gateway_runtime_discovery import DiscoveryError from hermes_cli.web_server_cron import _cron_profile_home from hermes_state import _default_db_path home = Path(_cron_profile_home(profile)[1]) if profile else Path(_default_db_path()).parent # Discovery is read-only and never starts a gateway. discovery = await asyncio.to_thread(discover_gateway_endpoint, home, timeout=1.5) endpoint = discovery.endpoint if endpoint is None: try: return await asyncio.to_thread(_mutate_offline_store, profile, actor, params) except HTTPException as exc: if discovery.state == 'absent' or not isinstance(exc.__cause__, OwnershipConflict): raise known = ''.join(f', {value}' for value in (discovery.reason_code, discovery.detail) if value) _log.warning('session write for %s: gateway owner not reachable (%s%s)', home, discovery.state, known) reason = discovery.reason_code or discovery.state raise HTTPException(status_code=503, detail=f'owner_unavailable: {reason}: {exc.detail}') from exc if not getattr(endpoint, 'control_home', True): # The selected home answered on its own socket (a multiplexer-served profile already # names the multiplexer root). Pin it, so the ticket is minted on the socket that just # identified as this profile's owner, never on the dashboard launch home's. from dataclasses import replace endpoint = replace(endpoint, control_home=str(home.expanduser().resolve())) call = {key: params[key] for key in ('session_id', 'request_id', 'expected_revision', 'operation', 'payload') if key in params} if params.get('expected_generation') is not None: call['expected_generation'] = params['expected_generation'] # A sign-in is the machine owner (owner_identity), which is this same-user socket's own peer # subject: let the socket mint it natively. The verifier lets a peer NAME only a structured # login subject, so forwarding the owner's OS subject explicitly would be refused. from gateway.owner_identity import host_owner_subject subject = None if actor.subject == host_owner_subject() else actor.subject try: async with connect_gateway(endpoint=endpoint, subject=subject) as client: return await client.rpc('session.mutate', **call) except GatewayRPCError: # The owner's own refusal frame: a verdict the caller maps to the direct path's status. raise except (GatewayClientError, DiscoveryError, OSError, TimeoutError) as exc: # Transport, not a verdict: a refused WebSocket dial (wrapped as GatewayClientError), a # control socket that refuses the mint after a restart (DiscoveryError), a raw socket error # or a timeout. Retrying is safe because the request reuses its request_id. raise HTTPException(status_code=503, detail='owner_unreachable') from exc # One reason -> HTTP status map for a session mutation refusal, shared by the direct path # (RuntimeStoreError) and the owner forward (GatewayClientError): a storage failure is a # server fault (503), never a session conflict. _MUTATION_REFUSAL_STATUS = {'permission_denied': 403, 'profile_mismatch': 403, 'not_found': 404, 'invalid_params': 400, 'runtime_draining': 503, 'storage_unavailable': 503} async def _mutate_session_request(request, profile, session_id, *, request_id, expected_revision, operation, payload, expected_generation=None): """Resolve the authenticated principal and owner; never open a second writer.""" import sqlite3 from fastapi import HTTPException from gateway.session_contract import SessionRef from gateway.session_mutations import mutate_session from hermes_cli.gateway_client import GatewayClientError from hermes_state_mutation_guards import MUTATION_GUARD_REFUSALS from hermes_state_runtime import RuntimeStoreError authority, actor = _session_mutation_context(request, profile) if operation == 'import' and authority is not None: _, errors = authority.db._validate_import_payload(payload['sessions']) if errors: raise HTTPException(status_code=400, detail={'errors': errors}) if request_id is None and expected_revision is None and expected_generation is None: # A pre-fence client (Desktop <= 0.22.2) sends none of the identity. The owner never # commits an unfenced write, so name the upgrade; the wording is what those builds' # toast summariser maps to "different versions, update" (#139742). raise HTTPException(status_code=409, detail=CLIENT_UPGRADE_REQUIRED) if request_id is None or expected_revision is None: raise HTTPException(status_code=409, detail='mutation_identity_required') params = dict(session_id=session_id, request_id=request_id, expected_revision=expected_revision, operation=operation, payload=payload) if expected_generation is not None: params['expected_generation'] = expected_generation try: if authority is None: # A live gateway owner answers first; a home without an endpoint takes # the offline store (whose reservation loses to any owner holding the lock). result = await _mutate_via_gateway_owner(profile, actor, params) else: with _native_history_transport(request, authority, actor): result = await mutate_session(authority, actor, SessionRef(authority.profile_id, session_id), params) return {'ok': True, **result} except RuntimeStoreError as exc: raise HTTPException(status_code=_MUTATION_REFUSAL_STATUS.get(exc.reason, 409), detail=exc.reason) from exc except GatewayClientError as exc: # The owner's own refusal verdicts (transport failures already became 503 above), # wire-shaped like RuntimeStoreError's reasons, so a forwarded write answers as the direct path. raise HTTPException(status_code=_MUTATION_REFUSAL_STATUS.get(str(exc), 409), detail=str(exc)) from exc except MUTATION_GUARD_REFUSALS as exc: raise HTTPException(status_code=409, detail='session_busy') from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except sqlite3.Error as exc: raise HTTPException(status_code=503, detail='storage_unavailable') from exc def _with_session_maintenance(profile, operation, *args): """Offline bulk actions reserve the exact owner lock through final DB close.""" from fastapi import HTTPException from gateway.runtime_ownership import OwnershipConflict, exclusive_maintenance from hermes_cli.web_server_cron import _cron_profile_home from hermes_state import _default_db_path home = Path(_cron_profile_home(profile)[1]) if profile else Path(_default_db_path()).parent try: with exclusive_maintenance([home]): return operation(*args) except OwnershipConflict as exc: # uvicorn runs at log_level=warning, so a handled 4xx leaves no server-side trace and the # refusal reads as a UI bug (#138915). Bulk store maintenance has no owner verb: name it. _log.warning('session write refused for %s: %s', home, exc) raise HTTPException(status_code=409, detail=str(exc)) from exc def _session_db_read_probe_statements() -> tuple: """Probe the declared schema without reconciling it on a browsing request.""" from hermes_state_schema import schema_read_probe_statements return schema_read_probe_statements() def _open_session_db_at_path(db_path: Path, *, read_only: bool): """Browsing never initializes or repairs the owner's session store. Owner startup reconciles schema. The one write a browse may do: a store that lags SCHEMA_SQL (missing table/column, not corruption) with NO gateway owning it gets the missing declared tables/columns once, in one short locked transaction (#140315); an owned store is refused by name instead. Explicit mutation APIs retain their writable acquisition until their separate owner-RPC migration. """ import sqlite3 from fastapi import HTTPException from hermes_state import SessionDB from hermes_state_errors import is_transient_sqlite_error from hermes_state_registry import acquire if not read_only: return acquire(db_path) try: initialized = db_path.stat().st_size > 0 except FileNotFoundError: initialized = False if not initialized: raise HTTPException( status_code=503, detail="Session store is not initialized. Start this profile's gateway to initialize it.") from hermes_state_reconcile import ( DeclarativeDriftError, is_declarative_drift_error, reconcile_drift_for_browse, ) def _open_probed(): db = SessionDB(db_path=db_path, read_only=True) try: conn = getattr(db, "_conn", None) if conn is not None: for statement in _session_db_read_probe_statements(): conn.execute(statement).fetchone() return db except BaseException: db.close() raise try: try: return _open_probed() except sqlite3.OperationalError as exc: # A missing declared table/column is upgrade drift, not corruption (#140315): catch # an unowned store up once under its runtime lock, refuse by name when a gateway # owns it. Never destructive: additive DDL only, no repair from browse. if not is_declarative_drift_error(exc): raise try: reconcile_drift_for_browse(db_path) except DeclarativeDriftError as drift: _log.warning("session store at %s is behind SCHEMA_SQL: %s", db_path, drift) raise HTTPException(status_code=503, detail=str(drift)) from exc return _open_probed() except (sqlite3.DatabaseError, UnicodeDecodeError) as exc: from hermes_state_errors import is_malformed_db_error if isinstance(exc, sqlite3.DatabaseError) and is_malformed_db_error(exc): # Same structured 503 the analytics routers publish (``corrupt_store_as_status``): a # corrupt image names the repair (`hermes doctor`), never a generic "schema unavailable". from hermes_cli.web_routers._common import corrupt_store_status raise corrupt_store_status(db_path, exc) from exc if isinstance(exc, sqlite3.OperationalError) and is_transient_sqlite_error(exc): detail = "Session store is busy (disk I/O or lock). Retry; the list was not cleared." else: detail = ( "Session store schema is unavailable or corrupt. Start or restart this profile's " "gateway to reconcile it; if it persists, run `hermes doctor` for diagnosis. " "Browsing does not repair the store.") raise HTTPException(status_code=503, detail=detail) from exc def _session_db_path_for_profile(profile: Optional[str]) -> Path: """state.db path for ``profile`` (None/empty = this process's own).""" from hermes_cli.web_server_cron import _cron_profile_home from hermes_state import _default_db_path if profile: _name, home = _cron_profile_home(profile) return Path(home) / "state.db" return Path(_default_db_path()) def _open_session_db_for_profile(profile: Optional[str], *, read_only: bool): """Open a SessionDB for ``profile`` (None/empty = this process's own state.db). Access-mode semantics: see :func:`_open_session_db_at_path`. """ return _open_session_db_at_path(_session_db_path_for_profile(profile), read_only=read_only) # In-process throttle for the opportunistic auto-archive trigger, keyed by # profile: bounds the config.yaml read to once per window; the sweep itself is # throttled far more coarsely by state_meta (sessions.min_interval_hours). _AUTO_ARCHIVE_CHECK_INTERVAL_S = 300.0 _last_auto_archive_check: dict[str, float] = {} def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None: """Config-gated owner-lifetime maintenance, never invoked by browsing. The standalone serve ticker maintains its own profile; a composed HTTP listener relies on its gateway owner's startup and housekeeping hooks. """ try: key = profile or "" now = time.monotonic() last = _last_auto_archive_check.get(key) if last is not None and now - last < _AUTO_ARCHIVE_CHECK_INTERVAL_S: return _last_auto_archive_check[key] = now from hermes_cli.config import load_config as _load_full_config from hermes_constants import reset_hermes_home_override, set_hermes_home_override # The config that governs a store is the one in that store's OWN home. A zero-arg # load_config() resolves through the PROCESS HERMES_HOME, so the dashboard swept every # profile's sessions with the launch profile's sessions.auto_archive/auto_archive_days — # one profile's retention silently decided another's. profile_home = _session_db_path_for_profile(profile).parent _home_token = set_hermes_home_override(str(profile_home)) try: cfg = (_load_full_config().get("sessions") or {}) finally: reset_hermes_home_override(_home_token) if not cfg.get("auto_archive", False): return from hermes_cli.profiles import _check_gateway_running # A live gateway owns this profile's store and runs the same sweep on its own # housekeeping tick ("state.db maintenance tick" in gateway/run.py, profile-scoped so a # multiplexed secondary's store is swept too). Opening it WRITABLE from `hermes # serve` adds a second writer to a database another process is already archiving, # for zero extra coverage (#110405). `_check_gateway_running` is the canonical # per-profile predicate (`_maybe_run_skill_maintenance` below uses it): its # multiplexer rung catches a served secondary, which owns no gateway.pid or lock # of its own and a bare lock-file probe would report stopped. if _check_gateway_running(profile_home): return db = _open_session_db_for_profile(profile, read_only=False) try: db.maybe_auto_archive( idle_days=float(cfg.get("auto_archive_days", 3)), min_interval_hours=int(cfg.get("min_interval_hours", 24))) finally: db.close() except Exception as exc: _log.debug("opportunistic auto-archive skipped: %s", exc) def _skill_maintenance_idle_for(started_at: float) -> Optional[float]: """Measure chat inactivity, not socket inactivity (Desktop stays connected).""" import tui_gateway.server as gateway from hermes_constants import get_hermes_home home = get_hermes_home().resolve() with gateway._sessions_lock: sessions = [session for session in gateway._sessions.values() if Path(session.get("profile_home") or home).resolve() == home] if any(session.get("running") for session in sessions): return None last_active = max( [started_at, gateway._closed_session_activity.get(str(home), 0)] + [float(session.get("last_active") or started_at) for session in sessions]) return max(0.0, time.time() - last_active) def _maybe_run_skill_maintenance(started_at: float) -> None: from hermes_constants import get_hermes_home from hermes_cli.profiles import _check_gateway_running # A live messaging gateway already owns these chores for this profile. if _check_gateway_running(get_hermes_home()): return from agent.curator import maybe_run_curator try: idle_for = _skill_maintenance_idle_for(started_at) if idle_for is not None: maybe_run_curator(idle_for_seconds=idle_for) except Exception as exc: _log.debug("serve curator tick skipped: %s", exc) async def _auto_archive_ticker_loop( interval_s: float = 3600.0, initial_delay_s: float = 90.0) -> None: """Poll maintenance for this serve profile, including Desktop-only installs. Individual chores own their config/interval gates. Curator additionally uses real chat inactivity; merely keeping a Desktop WebSocket open is not activity. """ started_at = time.time() def _sweep() -> None: _maybe_auto_archive_for_profile(None) _maybe_run_skill_maintenance(started_at) await asyncio.sleep(initial_delay_s) while True: try: await asyncio.to_thread(_sweep) except Exception as exc: _log.debug("auto-archive tick skipped: %s", exc) await asyncio.sleep(interval_s)