mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
kshitijk4poor
86cc60932f
Review follow-ups on the off-loop RPC scope and ancillary projection: - session_ancillary: the projection re-indexes authority.sessions (_require_current_route, physical_target, _owner_objects). Running in a worker, it can now see a session that retire_live_sessions popped on the loop while the read waited for a worker or the event lock. That surfaced as an uncaught KeyError, i.e. -32603 plus a "ws dispatch crash" traceback, instead of not_found. A KeyError for a session that is no longer the one the read started on now maps to not_found. A KeyError from a live session still raises. - session_authorities._prepare_owner_scope now enters the real _profile_runtime_scope(home, hydrate_secrets=False) in the worker and reads back the bound secrets and terminal scope. The scope entry order (home override, then secrets, then terminal policy) has one definition instead of a hand-kept mirror. - terminal_scope.prepare_profile_terminal_scope docstring: the build reads under the caller's home override and secret scope. - tests: the ancillary test uses the real projection with a recording event lock rather than patching the projection out. Added a retired-mid-read not_found test and a live-session KeyError test. The scope parity snapshot is keyed by name.
347 lines
17 KiB
Python
347 lines
17 KiB
Python
"""One SessionAuthority per reserved profile home; resolved by the active runtime scope.
|
|
|
|
A multiplexing gateway owns every profile under ``profiles/`` plus the default home. Each
|
|
home has its own ``state.db``, admission ledger, runtime epoch and control identity, so each
|
|
gets its own authority. Every consumer already runs inside the routed profile's
|
|
``_profile_runtime_scope`` (that is what the multiplex fixes on main established), so the
|
|
authority for "the profile this code is acting for" is the one keyed by the active home.
|
|
|
|
A single-profile gateway is a map of one: the launch home's authority answers every lookup,
|
|
scoped or not, exactly as before.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from hermes_constants import get_hermes_home, get_hermes_home_override, hermes_home_key
|
|
from hermes_state_runtime import RuntimeStoreError
|
|
|
|
|
|
class SessionAuthorities:
|
|
"""Registry of authorities keyed by ``hermes_home_key(home)``."""
|
|
|
|
def __init__(self, launch_home, *, multiplexed=False):
|
|
self.launch_key = hermes_home_key(launch_home)
|
|
self._by_key: dict[str, object] = {}
|
|
self._names: dict[str, str | None] = {}
|
|
# One event per slot reserved by ``add(home, None)`` while its authority is being built: set
|
|
# when ``replace`` fills the slot or ``remove`` drops it (the build failed).
|
|
self._starting: dict[str, asyncio.Event] = {}
|
|
# A multiplexing runtime (configured, or one that ever served a second home) keeps the
|
|
# scoped-lookup contract after its secondaries are removed: an unserved home is None.
|
|
self.multiplexed = bool(multiplexed)
|
|
|
|
def add(self, home, authority, name=None) -> None:
|
|
key = hermes_home_key(home)
|
|
if key in self._by_key:
|
|
raise RuntimeError(f'duplicate session authority for {home}')
|
|
self._by_key[key] = authority
|
|
self._names[key] = name
|
|
if authority is None:
|
|
self._starting[key] = asyncio.Event()
|
|
if key != self.launch_key:
|
|
self.multiplexed = True
|
|
|
|
def replace(self, home, authority) -> None:
|
|
"""Fill the slot reserved by ``add(home, None, name=...)`` once the authority exists."""
|
|
key = hermes_home_key(home)
|
|
if key not in self._by_key or self._by_key[key] is not None:
|
|
raise RuntimeError(f'no reserved session authority slot for {home}')
|
|
self._by_key[key] = authority
|
|
self._end_start(key)
|
|
|
|
def remove(self, home):
|
|
"""Drop *home*'s slot (a parked boot, a hot-unserved profile); returns the authority or None.
|
|
The launch slot is never removed."""
|
|
key = hermes_home_key(home)
|
|
if key == self.launch_key:
|
|
raise RuntimeError('the launch session authority cannot be removed')
|
|
self._names.pop(key, None)
|
|
self._end_start(key)
|
|
return self._by_key.pop(key, None)
|
|
|
|
def _end_start(self, key) -> None:
|
|
started = self._starting.pop(key, None)
|
|
if started is not None:
|
|
started.set()
|
|
|
|
def starting(self, profile):
|
|
"""``(key, started)`` for the reserved slot *profile* names while its authority is still being
|
|
built, else None. ``started`` is set once the build ends: filled, or dropped because it failed."""
|
|
from hermes_cli.profiles import profile_matches_home
|
|
return next(((key, started) for key, started in list(self._starting.items())
|
|
if profile_matches_home(profile, Path(key))), None)
|
|
|
|
def starting_homes(self) -> list[str]:
|
|
"""Keys of every slot whose authority is still being built (a snapshot, as in ``serving``)."""
|
|
return list(self._starting)
|
|
|
|
def profile_name(self, authority):
|
|
"""Served profile name for *authority*; None for the launch profile (its adapters are
|
|
``runner.adapters``, its session keys the historical ``agent:main`` namespace)."""
|
|
key = hermes_home_key(authority.profile_id)
|
|
return None if key == self.launch_key else self._names.get(key)
|
|
|
|
def __len__(self) -> int:
|
|
return sum(authority is not None for authority in self._by_key.values())
|
|
|
|
def __iter__(self):
|
|
"""The built authorities, retiring ones included. A slot still being built is no authority
|
|
yet, so iteration and ``len`` skip it (``starting`` names it); membership (``in``) still sees
|
|
it, so ``add`` refuses a second build of that profile."""
|
|
# A snapshot, as in ``serving``: other threads walk the set while the loop hot-serves.
|
|
return (authority for authority in list(self._by_key.values()) if authority is not None)
|
|
|
|
def __contains__(self, home) -> bool:
|
|
return hermes_home_key(home) in self._by_key
|
|
|
|
@property
|
|
def launch(self):
|
|
return self._by_key[self.launch_key]
|
|
|
|
def for_home(self, home):
|
|
"""Authority owning *home*, or None when this process does not serve it."""
|
|
return self._by_key.get(hermes_home_key(home))
|
|
|
|
def require(self, home):
|
|
authority = self.for_home(home)
|
|
if authority is None:
|
|
raise RuntimeStoreError('profile_mismatch')
|
|
return authority
|
|
|
|
def active(self):
|
|
"""Authority for the active runtime scope.
|
|
|
|
Unscoped code (startup, shutdown, background watchers) belongs to the launch profile.
|
|
Scoped code that names a home this process does not serve gets ``None``: a routed
|
|
profile must never fall back to the launch profile's ledger — including once a removed
|
|
secondary leaves only the launch slot (a late callback of that profile). Only a runtime
|
|
that never multiplexed is a map of one that answers every scope with the launch authority.
|
|
"""
|
|
if not self.multiplexed or get_hermes_home_override() is None:
|
|
return self._by_key[self.launch_key]
|
|
return self._by_key.get(hermes_home_key(get_hermes_home()))
|
|
|
|
def serving(self) -> list:
|
|
"""Authorities that accept work: not a reserved-but-unbuilt slot, not one retiring (a
|
|
failed serve or an unserve whose turn outlived Stop keeps its slot but refuses admissions)."""
|
|
# A snapshot: hosted-room coordinator threads read the served set while the loop hot-serves.
|
|
return [a for a in list(self._by_key.values()) if a is not None and not getattr(a, 'retiring', False)]
|
|
|
|
def served_profiles(self) -> list[dict]:
|
|
return [{'profile_id': a.profile_id, 'home': a.profile_id} for a in self.serving()]
|
|
|
|
def profile_ids(self) -> frozenset[str]:
|
|
return frozenset(a.profile_id for a in self.serving())
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocalBot:
|
|
"""One local bot this process serves: its profile ``name``, ``home`` and owning ``authority``.
|
|
``problem`` says why a client cannot address it by that name (None when it can); ``retiring``
|
|
marks one being unserved, which still owns its rooms but takes no new work."""
|
|
name: str
|
|
home: Path
|
|
authority: Any
|
|
problem: str | None = None
|
|
retiring: bool = False
|
|
|
|
|
|
def local_bots(runner) -> list[LocalBot]:
|
|
"""The local bots this gateway serves, one per built authority, by the name the registry
|
|
reserved each under (the launch profile keeps ``served_profile_name``). The one answer to
|
|
"which bots are local here" for both the Bots roster (``profiles.list``) and group seating
|
|
(``session_hosted_members``). A name two homes answer to, or one that is not its home's own
|
|
name (a linked ``profiles/<name>`` resolves outside ``profiles/`` and reads as ``default``), is
|
|
flagged rather than bound: a client routes, and the member transport names, a bot by name."""
|
|
from collections import Counter
|
|
registry = getattr(runner, 'session_authorities', None)
|
|
if registry is None:
|
|
authority = getattr(runner, 'session_authority', None)
|
|
pairs = [(served_profile_name(authority.profile_id), authority)] if authority is not None else []
|
|
else:
|
|
# A snapshot: hosted-room coordinator threads read the served set while the loop hot-serves.
|
|
pairs = [(registry.profile_name(a) or served_profile_name(a.profile_id), a)
|
|
for a in list(registry._by_key.values()) if a is not None]
|
|
counts = Counter(name for name, _ in pairs)
|
|
bots = []
|
|
for name, authority in pairs:
|
|
problem = None
|
|
if counts[name] > 1:
|
|
problem = 'two served homes answer to that name'
|
|
elif name != served_profile_name(authority.profile_id):
|
|
problem = 'a linked profile home is not supported yet'
|
|
bots.append(LocalBot(name, Path(authority.profile_id), authority, problem,
|
|
bool(getattr(authority, 'retiring', False))))
|
|
return bots
|
|
|
|
|
|
def served_profile_name(home) -> str:
|
|
"""Canonical profile id of a served home: ``default`` for the root, the directory name for
|
|
``profiles/<name>`` (via ``hermes_constants.profile_name_for_home``, the resolver main uses).
|
|
Any other custom HERMES_HOME is the process's own single profile and keeps the ``default``
|
|
label the CLI reports for it (``hermes -p`` never targets such a home by name)."""
|
|
from hermes_constants import profile_name_for_home
|
|
return profile_name_for_home(home) or 'default'
|
|
|
|
|
|
def authority_for_home(runner, home):
|
|
"""Explicit per-home lookup for callers that already know the routed home."""
|
|
registry = getattr(runner, 'session_authorities', None)
|
|
if registry is None:
|
|
authority = getattr(runner, 'session_authority', None)
|
|
if authority is not None and Path(authority.db.db_path).resolve().parent == Path(home).resolve():
|
|
return authority
|
|
return None
|
|
return registry.for_home(home)
|
|
|
|
|
|
def authority_for_profile_id(runner, profile_id):
|
|
return authority_for_home(runner, Path(profile_id))
|
|
|
|
|
|
def active_authority(runner):
|
|
"""Authority for the active runtime scope (see ``SessionAuthorities.active``); the single
|
|
launch authority when the runner predates the registry (bare test runners)."""
|
|
registry = getattr(runner, 'session_authorities', None)
|
|
if registry is None:
|
|
return getattr(runner, 'session_authority', None)
|
|
return registry.active()
|
|
|
|
|
|
def require_active_authority(runner, store=None):
|
|
"""The active scope's authority, or a refusal: retryable (``runtime_draining``) while its
|
|
profile is parked, else ``profile_mismatch``. With *store* (a callable returning the caller's
|
|
SessionDB, called only once an authority exists), that store must be the authority's: an API
|
|
adapter bound to another profile's SessionDB never admits here."""
|
|
authority = active_authority(runner)
|
|
if authority is None:
|
|
from gateway.run_runtime import refuse_parked_scope
|
|
refuse_parked_scope(runner) # retryable while its profile is parked
|
|
if authority is None or (store is not None and store() is not authority.db):
|
|
from hermes_state_runtime import RuntimeStoreError
|
|
raise RuntimeStoreError('profile_mismatch')
|
|
return authority
|
|
|
|
|
|
def all_authorities(runner):
|
|
registry = getattr(runner, 'session_authorities', None)
|
|
if registry is not None:
|
|
return list(registry)
|
|
authority = getattr(runner, 'session_authority', None)
|
|
return [authority] if authority is not None else []
|
|
|
|
|
|
def serving_authorities(runner):
|
|
"""The authorities that accept work (``SessionAuthorities.serving``: no reserved-but-unbuilt slot,
|
|
none retiring): what startup recovery, service creation and cross-ledger reads cover. Shutdown
|
|
keeps ``all_authorities``, so a retained authority's live writer is still settled."""
|
|
registry = getattr(runner, 'session_authorities', None)
|
|
if registry is not None:
|
|
return registry.serving()
|
|
authority = getattr(runner, 'session_authority', None)
|
|
return [authority] if authority is not None and not getattr(authority, 'retiring', False) else []
|
|
|
|
|
|
def require_serving(authority):
|
|
"""Refuse a verb reached through a stale connection to a retired authority: its ownership is
|
|
released, and until a successor bumps the epoch its own epoch still passes the store fence."""
|
|
if getattr(authority, 'retired', False):
|
|
raise RuntimeStoreError('runtime_retired')
|
|
|
|
|
|
def admission_owner(authority, session_id):
|
|
"""The admission identity a physical store row answers to: its compression root, then the
|
|
local creation id a canonical reset forks under (a reset child is not a compression child, so
|
|
``logical_owner`` alone stops at the child and misses the conversation's admissions)."""
|
|
from hermes_state_local import local_lineage_owner
|
|
sid = authority.logical_owner(session_id)
|
|
if not sid:
|
|
return sid
|
|
with authority.db._read_ctx() as conn:
|
|
return local_lineage_owner(conn, sid)
|
|
|
|
|
|
def holds_admission(authority, session_id) -> bool:
|
|
"""Whether *authority*'s store holds a canonical admission for *session_id*. A parked profile's
|
|
retained (retiring) store that cannot answer counts as holding one: startup auto-resume fails
|
|
closed for that entry instead of stopping the boot or resuming a turn that FIFO may own."""
|
|
try:
|
|
return bool(authority.db._read_one(
|
|
'SELECT 1 FROM session_admissions WHERE target_session_id IN (?,?) LIMIT 1',
|
|
(session_id, admission_owner(authority, session_id))))
|
|
except Exception as exc:
|
|
if not getattr(authority, 'retiring', False):
|
|
raise
|
|
import logging
|
|
logging.getLogger('gateway.run').warning("Skipping auto-resume of %s: retained store %s cannot answer: %s",
|
|
session_id, getattr(authority, 'profile_id', '?'), exc)
|
|
return True
|
|
|
|
|
|
def _prepare_owner_scope(home: Path):
|
|
"""Enter ``owner_scope``'s profile scope in this (worker) thread and return what it bound:
|
|
``(secrets, terminal scope)``, or ``None`` when *home* is not a directory (``owner_scope`` binds
|
|
nothing then). Entering the real scope keeps one definition of how it is built."""
|
|
if not home.is_dir():
|
|
return None
|
|
from agent.secret_scope import current_secret_scope
|
|
from gateway.run import _profile_runtime_scope
|
|
from tools.terminal_scope import get_terminal_scope
|
|
with _profile_runtime_scope(home, hydrate_secrets=False):
|
|
return current_secret_scope(), get_terminal_scope()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def async_owner_scope(authority):
|
|
"""``owner_scope`` for a coroutine on the gateway loop: the profile's ``.env``, ``config.yaml`` and
|
|
terminal policy are read in a worker thread, then only the context variables are set here.
|
|
|
|
Every RPC a client sends enters its owner's scope, and the classic CLI, Ink and Desktop poll
|
|
``subagent.list`` / ``process.list`` every 1-5 s per session: built on the loop, that scope was the
|
|
largest single cost of RPC dispatch on a multiplexed gateway with many sessions open."""
|
|
runner = getattr(authority, 'runner', None)
|
|
home = Path(str(authority.profile_id))
|
|
if not getattr(getattr(runner, 'config', None), 'multiplex_profiles', False) or not home.is_absolute():
|
|
with owner_scope(authority): # the launch scope (frozen sources) or nullcontext: no profile files
|
|
yield
|
|
return
|
|
prepared = await asyncio.to_thread(_prepare_owner_scope, home)
|
|
if prepared is None:
|
|
yield
|
|
return
|
|
from gateway.run import _profile_runtime_scope
|
|
secrets, terminal = prepared
|
|
with _profile_runtime_scope(home, secrets, prepared_terminal_scope=terminal):
|
|
yield
|
|
|
|
|
|
def owner_scope(authority, *, hydrate_secrets=False):
|
|
"""Runtime scope of the profile that owns *authority* — under multiplex only.
|
|
|
|
Owner-side handlers (RPC dispatch, cron submit, admitted execution) must read config, jobs
|
|
and policy for the OWNING profile, never the launch profile's ambient scope. A
|
|
single-profile gateway keeps its ambient scope byte-for-byte: entering a scope there
|
|
would read the profile's config/.env for every handler, including bypass (--safe-mode)
|
|
launches that must never touch the profile YAML. Test peers with symbolic ids ('fixture')
|
|
likewise keep the ambient scope.
|
|
|
|
A standalone gateway that already hosts a second home (a profile created over the wire froze
|
|
the launch env and made unscoped secret reads fail closed) binds the launch profile's own scope
|
|
instead: unbound, every launch-profile session.create raised UnscopedSecretError (#141623).
|
|
"""
|
|
from contextlib import nullcontext
|
|
runner = getattr(authority, 'runner', None)
|
|
if not getattr(getattr(runner, 'config', None), 'multiplex_profiles', False):
|
|
from tui_gateway.launch_profile_policy import launch_profile_scope_if_multiplexed
|
|
return launch_profile_scope_if_multiplexed()
|
|
home = Path(str(authority.profile_id))
|
|
if not home.is_absolute() or not home.is_dir():
|
|
return nullcontext()
|
|
from gateway.run import _profile_runtime_scope
|
|
return _profile_runtime_scope(home, hydrate_secrets=hydrate_secrets)
|