mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
fix(gateway): every Bots roster row resolves its own Bot Chat on a multiplexed gateway
The Desktop asks profiles.list of the selected bot over its one host-scoped socket. Since the unified gateway, only that bot's row carried session fields: every other served profile was listed by name alone, so on a fleet every row but one showed a hollow draft dot and no age, although the same socket may already open each of those bots' sessions. Each served sibling a host-scoped connection routes to now carries the same canonical_session / last_session, read from that sibling's own authority with the connection's subject and capabilities. The sibling set is snapshotted on the owner loop, and no connection is built for it. A profile-bound ticket, a retiring sibling, a row listed under a name that routes to another profile, and a sibling whose read fails for any reason stay metadata-only; a failing sibling is warned once, not on every poll. The owner-or-operator rule applies per profile. Each row is read on one connection that never reopens a closed store (SessionDB._read_ctx(reopen=False)), so a roster poll racing an unserve cannot reopen a writer into a home being deleted. last_session skips hidden rows again, as the pre-unified roster did, and the rows are reused while a profile's store has not moved, on the signature the pre-unified roster cache used (#117257), keyed by reader. The adopted-binding predicate from the previous commit becomes one SQL predicate for both lookups. AuthorityConnection.host_scoped names the gate _sibling_for applies. Fixes #139227 Refs #134889 (RT-24), #139161 (cherry picked from commit 714da8acbcafc41a0e6ebce0e490eaf5ad0e72a0)
This commit is contained in:
1 parent
5b1d205e29
commit
9450b0ca31
7 files changed
+413
-70
No files matched your search
@@ -1,9 +1,13 @@
|
||||
"""Profile-owned room controls over the canonical hosted service."""
|
||||
import asyncio
|
||||
import logging
|
||||
from pathlib import Path
|
||||
import weakref
|
||||
|
||||
from hermes_state_runtime import RuntimeStoreError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
GROUP_METHODS = {
|
||||
'groups.capabilities': 'session:read',
|
||||
@@ -61,13 +65,14 @@ async def dispatch_group_control(connection, method, params):
|
||||
if profile and not profile_matches_home(profile, home):
|
||||
raise RuntimeStoreError('profile_mismatch')
|
||||
supplied = {key: value for key, value in params.items() if key != 'profile'}
|
||||
siblings = _roster_siblings(connection, supplied) if method == 'profiles.list' else ()
|
||||
|
||||
def invoke():
|
||||
from gateway.run import _profile_runtime_scope
|
||||
from gateway.hosted_rooms import HostedRoomError
|
||||
with _profile_runtime_scope(home):
|
||||
if method == 'profiles.list':
|
||||
return _profiles(authority, actor, home, supplied)
|
||||
return _profiles(authority, actor, home, supplied, siblings)
|
||||
try:
|
||||
return _group(authority, actor, method, supplied, host_scope=connection.host_scoped)
|
||||
except RuntimeStoreError:
|
||||
@@ -259,8 +264,113 @@ def _execution_control(service, method, params):
|
||||
return handlers[method]()
|
||||
|
||||
|
||||
def _profiles(authority, actor, home, params):
|
||||
from hermes_cli.profiles import _profile_info, read_profile_meta
|
||||
# A local chat: minted by this owner (``local-*``), or a pre-upgrade chat it adopted, which keeps its
|
||||
# historical id as chat_id under a ``gateway.local_legacy.v1:`` binding. These are the chats
|
||||
# ``resolve_local_target`` resolves by chat_id without adopting anything (its policy-receipt arm adds
|
||||
# none: a receipt-bound chat_id that is not ``local-*`` is an adopted one). A messaging or API
|
||||
# conversation with the same title is neither.
|
||||
_LOCAL_CHAT = "(chat_id LIKE 'local-%' OR EXISTS (SELECT 1 FROM state_meta WHERE key = ? || sessions.chat_id))"
|
||||
# Sibling authorities whose last roster read failed: warned once per outage, then logged at debug.
|
||||
# Weak, so an unserved profile drops out with its authority and a re-served one (a new authority)
|
||||
# is warned again.
|
||||
_DEGRADED = weakref.WeakSet()
|
||||
|
||||
|
||||
def _roster_siblings(connection, params):
|
||||
"""The other profiles this process serves, as ``[(authority, reader)]``, snapshotted on the owner
|
||||
loop, where serve and unserve mutate the registry; an unbuilt slot is skipped. ``reader`` is the
|
||||
principal that row's session fields are read as, or None for a metadata-only row: only a
|
||||
host-scoped socket routes session verbs to a sibling (``AuthorityConnection._sibling_for``). It
|
||||
carries this socket's subject and capabilities, which is what the routed connection would carry;
|
||||
none is built, so no poll (routed or not) adds connections."""
|
||||
from gateway.session_contract import Principal
|
||||
authority, actor = connection.authority, connection.actor
|
||||
registry = getattr(getattr(authority, 'runner', None), 'session_authorities', None)
|
||||
if registry is None:
|
||||
return []
|
||||
readable = connection.host_scoped and params.get('include_sessions', True) is True
|
||||
return [(sibling, Principal(actor.subject, sibling.profile_id, actor.capabilities, actor.transport_id)
|
||||
if readable else None)
|
||||
for sibling in list(registry) if sibling is not None and sibling is not authority]
|
||||
|
||||
|
||||
def _bot_sessions(authority, actor):
|
||||
"""A roster row's session fields, read from *authority*'s own store as *actor*: the exact
|
||||
"Bot Chat" title (the status dot's session and the row's age) and the newest visible local
|
||||
conversation, each only when the actor owns it or holds operator access. Reused while the
|
||||
store has not moved, as the pre-unified roster did (#117257): every connection polls every 5 s.
|
||||
Discovery is read-only: no legacy restore/unarchive or new SessionDB handle."""
|
||||
from tui_gateway.profile_roster_cache import cached_reader_session_fields
|
||||
operator = 'session:operator' in actor.capabilities
|
||||
return cached_reader_session_fields(Path(authority.db.db_path).parent, (actor.subject, operator),
|
||||
lambda: _read_bot_sessions(authority.db, actor.subject, operator))
|
||||
|
||||
|
||||
def _read_bot_sessions(db, subject, operator):
|
||||
from hermes_state_local_migration import LEGACY_PREFIX
|
||||
from hermes_state_sessions import INTERNAL_LISTING_SOURCES
|
||||
hidden_sources = ','.join('?' * len(INTERNAL_LISTING_SOURCES))
|
||||
owner_filter, owner_params = ('', ()) if operator else ('user_id=? AND ', (subject,))
|
||||
# One read connection for the whole row. A store closed meanwhile (its profile unserved
|
||||
# mid-poll) refuses rather than reopening a writer into a home that is being deleted.
|
||||
with db._read_ctx(reopen=False) as conn:
|
||||
def summary(root):
|
||||
tip = db._compression_tip_on(conn, root['id'])
|
||||
target = conn.execute('SELECT * FROM sessions WHERE id=?', (tip,)).fetchone() or root
|
||||
message = conn.execute(
|
||||
"SELECT content FROM messages WHERE session_id=? AND active=1 "
|
||||
"AND role IN ('user','assistant') AND TRIM(COALESCE(content,''))!='' "
|
||||
"ORDER BY id DESC LIMIT 1", (tip,)).fetchone()
|
||||
text = ' '.join(str(message[0] or '').split()) if message else ''
|
||||
return {'id': root['id'], 'resolved_id': tip, 'title': target['title'] or '',
|
||||
'root_title': root['title'] or '',
|
||||
'preview': text[:80] + '...' if len(text) > 80 else text,
|
||||
'started_at': target['started_at'] or 0,
|
||||
'last_active': target['last_activity_at'] or target['started_at'] or 0,
|
||||
'message_count': target['message_count'] or 0}
|
||||
|
||||
fields = {'last_session': None, 'worker_session': None, 'canonical_session': None}
|
||||
# The named registry is not a recency window and canonical chats are hidden.
|
||||
canonical = conn.execute(f'SELECT * FROM sessions WHERE title=? AND {_LOCAL_CHAT}',
|
||||
('Bot Chat', LEGACY_PREFIX)).fetchone()
|
||||
if canonical and (canonical['user_id'] == subject or operator) and not canonical['archived']:
|
||||
fields['canonical_session'] = summary(canonical)
|
||||
# The newest conversation a picker shows: a hidden row (a Bot Chat, group member plumbing)
|
||||
# never wins, as in the pre-unified roster's listing.
|
||||
latest = conn.execute(
|
||||
f"SELECT * FROM sessions WHERE {owner_filter}{_LOCAL_CHAT} AND archived=0 AND hidden=0 "
|
||||
f"AND COALESCE(source,'') NOT IN ({hidden_sources}) "
|
||||
"ORDER BY COALESCE(last_activity_at,started_at) DESC LIMIT 1",
|
||||
(*owner_params, LEGACY_PREFIX, *INTERNAL_LISTING_SOURCES)).fetchone()
|
||||
if latest and fields['canonical_session'] and latest['id'] == canonical['id']:
|
||||
fields['last_session'] = dict(fields['canonical_session'])
|
||||
elif latest:
|
||||
fields['last_session'] = summary(latest)
|
||||
return fields
|
||||
|
||||
|
||||
def _sibling_sessions(authority, reader):
|
||||
"""A sibling row's session fields. A retiring sibling is not read: unserve marks it before it
|
||||
closes the store, and this runs right before the read (a close that still lands mid-read is
|
||||
refused by ``reopen=False``). Any failure leaves only that row metadata-only, as the
|
||||
pre-unified roster degraded per profile, instead of failing the whole roster."""
|
||||
if authority.retiring or authority.retired:
|
||||
return {}
|
||||
try:
|
||||
fields = _bot_sessions(authority, reader)
|
||||
except Exception as exc: # health: allow BLE001 -- one bot's store must never fail the whole roster
|
||||
log = logger.debug if authority in _DEGRADED else logger.warning
|
||||
_DEGRADED.add(authority)
|
||||
log('Bots roster: %s is metadata-only, its session store did not answer (%s)', authority.profile_id, exc)
|
||||
return {}
|
||||
_DEGRADED.discard(authority)
|
||||
return fields
|
||||
|
||||
|
||||
def _profiles(authority, actor, home, params, siblings=()):
|
||||
"""The Bots roster: this profile's row, then every profile it can name. ``siblings`` is
|
||||
``_roster_siblings``' snapshot; a sibling with a reader carries the same session fields."""
|
||||
from hermes_cli.profiles import _profile_info, profile_matches_home
|
||||
import hermes_yaml as yaml
|
||||
include_sessions = params.get('include_sessions', True)
|
||||
if type(include_sessions) is not bool:
|
||||
@@ -282,68 +392,21 @@ def _profiles(authority, actor, home, params):
|
||||
row['ui_meta'] = meta['ui_meta']
|
||||
row['has_avatar'] = any((home / 'assets' / f'avatar.{ext}').is_file() for ext in ('png', 'jpg', 'webp'))
|
||||
if include_sessions:
|
||||
row.update(last_session=None, worker_session=None, canonical_session=None)
|
||||
# Discovery is read-only: no legacy restore/unarchive or new SessionDB handle.
|
||||
def summary(session):
|
||||
tip = authority.db.get_compression_tip(session['id']) or session['id']
|
||||
target = authority.db.get_session(tip) or session
|
||||
with authority.db._lock:
|
||||
message = authority.db._conn.execute(
|
||||
"SELECT content FROM messages WHERE session_id=? AND active=1 "
|
||||
"AND role IN ('user','assistant') AND TRIM(COALESCE(content,''))!='' "
|
||||
"ORDER BY id DESC LIMIT 1", (tip,)).fetchone()
|
||||
text = ' '.join(str(message[0] or '').split()) if message else ''
|
||||
return {'id': session['id'], 'resolved_id': tip, 'title': target.get('title') or '',
|
||||
'root_title': session.get('title') or '',
|
||||
'preview': text[:80] + '...' if len(text) > 80 else text,
|
||||
'started_at': target.get('started_at') or 0,
|
||||
'last_active': target.get('last_activity_at') or target.get('started_at') or 0,
|
||||
'message_count': target.get('message_count') or 0}
|
||||
|
||||
# Adopted local history keeps its original id as chat_id. Its durable
|
||||
# binding, not an id spelling, distinguishes it from native messaging.
|
||||
from hermes_state_local_migration import LEGACY_PREFIX
|
||||
|
||||
def local_chat(session):
|
||||
chat_id = str(session.get('chat_id') or '')
|
||||
if chat_id.startswith('local-'):
|
||||
return True
|
||||
with authority.db._lock:
|
||||
return bool(authority.db._conn.execute(
|
||||
'SELECT 1 FROM state_meta WHERE key=?', (LEGACY_PREFIX + chat_id,)).fetchone())
|
||||
|
||||
# The named registry is not a recency window and canonical chats are hidden.
|
||||
canonical = authority.db.get_session_by_title('Bot Chat')
|
||||
if (canonical and (canonical.get('user_id') == actor.subject
|
||||
or 'session:operator' in actor.capabilities)
|
||||
and local_chat(canonical) and not canonical.get('archived')):
|
||||
row['canonical_session'] = summary(canonical)
|
||||
if 'session:operator' in actor.capabilities:
|
||||
owner_filter, owner_params = '', ()
|
||||
else:
|
||||
owner_filter, owner_params = 'user_id=? AND ', (actor.subject,)
|
||||
with authority.db._lock:
|
||||
from hermes_state_sessions import INTERNAL_LISTING_SOURCES
|
||||
hidden_sources = ','.join('?' * len(INTERNAL_LISTING_SOURCES))
|
||||
latest = authority.db._conn.execute(
|
||||
f"SELECT id FROM sessions WHERE {owner_filter}(chat_id LIKE 'local-%' OR EXISTS "
|
||||
"(SELECT 1 FROM state_meta WHERE key=? || sessions.chat_id)) AND archived=0 "
|
||||
f"AND COALESCE(source,'') NOT IN ({hidden_sources}) "
|
||||
"ORDER BY COALESCE(last_activity_at,started_at) DESC LIMIT 1",
|
||||
(*owner_params, LEGACY_PREFIX, *INTERNAL_LISTING_SOURCES)).fetchone()
|
||||
if latest:
|
||||
row['last_session'] = summary(authority.db.get_session(latest[0]))
|
||||
row.update(_bot_sessions(authority, actor))
|
||||
profiles = [row]
|
||||
seen = {home.resolve()}
|
||||
|
||||
def discovered(configured, target):
|
||||
# Discovery metadata only: never a licence to read another owner's state/configuration.
|
||||
def discovered(configured, target, reach=None):
|
||||
# Discovery alone is never a licence to read another owner's state/configuration: only a
|
||||
# sibling this connection already routes to (``reach``) adds its session fields.
|
||||
if target.resolve() in seen:
|
||||
return
|
||||
seen.add(target.resolve())
|
||||
entry = {'name': configured, 'path': str(target), 'is_default': configured == 'default',
|
||||
'model': '', 'provider': '', 'description': '', 'display_name': configured,
|
||||
'skill_count': 0}
|
||||
if include_sessions and reach is not None:
|
||||
entry.update(_sibling_sessions(*reach))
|
||||
meta_path = target / 'profile.yaml'
|
||||
try:
|
||||
meta = yaml.safe_load(meta_path.read_text(encoding='utf-8-sig')) if meta_path.is_file() else {}
|
||||
@@ -357,9 +420,13 @@ def _profiles(authority, actor, home, params):
|
||||
|
||||
# Sibling profiles this process multiplexes: the roster lists them with the same
|
||||
# standing as a hosted-room destination, and the client routes their sessions by name.
|
||||
registry = getattr(getattr(authority, 'runner', None), 'session_authorities', None)
|
||||
for sibling in (registry or ()):
|
||||
discovered(served_profile_name(Path(sibling.profile_id)), Path(sibling.profile_id))
|
||||
for sibling, reader in siblings:
|
||||
configured, target = served_profile_name(Path(sibling.profile_id)), Path(sibling.profile_id)
|
||||
# A row's fields come from the authority a click on that row opens: the client routes by
|
||||
# this name. A linked ``profiles/<name>`` resolves outside ``profiles/`` and reads as
|
||||
# ``default`` here, which routes to the launch profile, so its row stays metadata-only.
|
||||
reach = (sibling, reader) if reader is not None and profile_matches_home(configured, target) else None
|
||||
discovered(configured, target, reach)
|
||||
service = getattr(authority, 'hosted_room_service', None)
|
||||
if service is not None:
|
||||
for configured, target in service.profile_homes().items():
|
||||
|
||||
+6
-2
@@ -918,11 +918,13 @@ class SessionDB(
|
||||
return self._get_read_conn()
|
||||
|
||||
@contextmanager
|
||||
def _read_ctx(self) -> Iterator[sqlite3.Connection]:
|
||||
def _read_ctx(self, *, reopen: bool = True) -> Iterator[sqlite3.Connection]:
|
||||
"""Yield a connection for read-only statements: a pooled read-only
|
||||
connection with NO lock under WAL; otherwise (non-WAL, open failure,
|
||||
ceiling reached) the writer connection under self._lock — deliberate
|
||||
degradation: slower beats EMFILE, which the supervisor cannot see."""
|
||||
degradation: slower beats EMFILE, which the supervisor cannot see.
|
||||
``reopen=False``: a read after ``close()`` raises instead of reopening the
|
||||
writer (an optional reader must not hold a home that is being deleted)."""
|
||||
conn = self._checkout_read_conn()
|
||||
if conn is not None:
|
||||
try:
|
||||
@@ -943,6 +945,8 @@ class SessionDB(
|
||||
return
|
||||
with self._lock:
|
||||
if self._conn is None: # close() raced a still-unwinding reader
|
||||
if not reopen:
|
||||
raise sqlite3.ProgrammingError(f"SessionDB for {self.db_path} is closed")
|
||||
self._reopen_after_close_locked(context="read")
|
||||
conn = cast(sqlite3.Connection, self._conn)
|
||||
if self._wal_active or self.read_only: # WAL: no SHARED-lock wait; ro: opened with the read budget
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
"""Regression for #139227: on a multiplexed gateway, every Bots roster row resolves its own profile's
|
||||
Bot Chat (the status dot's session id and the row's age), however that chat was created.
|
||||
|
||||
Each profile here holds its Bot Chat from a different creation path: a pre-upgrade CLI chat adopted
|
||||
on its first open (the reporter's fleet), a first ``bot_relay.deliver`` (``gateway/session_bot.py``),
|
||||
``hermes -p <bot> chat -c "Bot Chat" --create-if-missing``, and the Desktop's own mint. The Desktop
|
||||
asks ``profiles.list`` of its active bot over its one host-scoped socket, so the other bots' rows
|
||||
come from that same answer.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
from tests.gateway.test_session_authorities_multiplex import _reserve_homes, _runner
|
||||
from tests.gateway.test_ticket_profile_scope import _Transport
|
||||
|
||||
AUTHOR = {"id": "bot:cto", "name": "cto", "is_bot": True}
|
||||
LEGACY_ID = '20261006_105832_fa0db9' # a pre-upgrade CLI session id, kept as chat_id on adoption
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def fleet(tmp_path, monkeypatch, request):
|
||||
from gateway.run_runtime import initialize_gateway_runtime
|
||||
from gateway.runtime_ownership import process_ownership
|
||||
from gateway.session_authorities import authority_for_profile_id, owner_scope
|
||||
from gateway.session_controls import AuthorityConnection
|
||||
|
||||
executed = []
|
||||
|
||||
async def execute(authority, ref, row):
|
||||
executed.append(ref.session_id)
|
||||
return 'pong'
|
||||
|
||||
monkeypatch.setattr('gateway.session_finite.execute_finite_admission', execute)
|
||||
root, homes = _reserve_homes(tmp_path, monkeypatch, names=('alpha', 'beta', 'gamma'))
|
||||
# An extra served home outside ``profiles/`` (what a linked ``profiles/<name>`` resolves to).
|
||||
for name in getattr(request, 'param', ()):
|
||||
(tmp_path / 'linked' / name).mkdir(parents=True)
|
||||
homes.append((name, tmp_path / 'linked' / name))
|
||||
for _, home in homes:
|
||||
(home / 'config.yaml').write_text('model:\n default: fixture\nplatform_toolsets:\n cli: []\n')
|
||||
process_ownership.reserve([home for _, home in homes])
|
||||
connections, runner = [], None
|
||||
|
||||
def connect(name, subject='uid:desk', *, scope='profile', operator=True):
|
||||
# gateway/run_api.py's redeem → identity → connection binding for a native ticket.
|
||||
store = runner.session_ticket_store
|
||||
ticket = store.mint(profile_id=str(dict(homes)[name].resolve()), subject=subject,
|
||||
purpose='interactive', scope=scope)
|
||||
grant = store.redeem(ticket, profile_id=None, purpose='interactive')
|
||||
authority = authority_for_profile_id(runner, grant['profile_id'])
|
||||
identity = {'user_id': grant['subject'], 'provider': 'local', 'profile_id': grant['profile_id'],
|
||||
'instance_id': grant['instance_id'], 'capabilities': grant['capabilities'],
|
||||
'native_bootstrap': True, 'profile_scope': grant['scope']}
|
||||
with owner_scope(authority):
|
||||
connection = AuthorityConnection(authority, _Transport(), identity, operator=operator)
|
||||
connections.append(connection)
|
||||
return connection
|
||||
|
||||
async def call(connection, method, **params):
|
||||
with owner_scope(connection.authority):
|
||||
reply = await connection.dispatch({'id': 1, 'method': method, 'params': params})
|
||||
assert 'result' in reply, (method, reply)
|
||||
return reply['result']
|
||||
|
||||
try:
|
||||
runner = _runner(root, homes)
|
||||
await initialize_gateway_runtime(runner)
|
||||
yield _Fleet(runner, dict(homes), connect, call, executed)
|
||||
finally:
|
||||
for connection in connections:
|
||||
with owner_scope(connection.authority):
|
||||
await connection.close()
|
||||
for authority in list(getattr(runner, 'session_authorities', []) or []):
|
||||
authority.db.close()
|
||||
for _, home in homes:
|
||||
process_ownership.release(Path(home))
|
||||
|
||||
|
||||
class _Fleet:
|
||||
def __init__(self, runner, homes, connect, call, executed):
|
||||
self.runner, self.homes, self.connect, self.call, self.executed = runner, homes, connect, call, executed
|
||||
|
||||
def db(self, name):
|
||||
from gateway.session_authorities import authority_for_home
|
||||
return authority_for_home(self.runner, self.homes[name]).db
|
||||
|
||||
async def bot_chats(self):
|
||||
"""Mint each bot's Bot Chat the way its creator does; return ``{profile: Bot Chat id}``."""
|
||||
from tests.waiting import async_wait_until
|
||||
from tools.bot_live_delivery import read_delivery_result
|
||||
desktop = self.connect('default', scope='host')
|
||||
# default: created by the pre-upgrade CLI, adopted by the Desktop's first open of the bot.
|
||||
legacy = self.db('default')
|
||||
legacy.create_session(LEGACY_ID, source='cli', model='fixture')
|
||||
legacy.set_session_title(LEGACY_ID, 'Bot Chat')
|
||||
legacy.append_message(LEGACY_ID, 'assistant', 'kept from before the upgrade')
|
||||
found = await self.call(desktop, 'session.list', profile='default', title='Bot Chat',
|
||||
include_hidden=True, limit=50)
|
||||
assert [row['session_id'] for row in found['sessions']] == [LEGACY_ID], found
|
||||
assert legacy.get_session(LEGACY_ID)['chat_id'] == LEGACY_ID # the reporter's row shape
|
||||
# alpha: no Bot Chat until a teammate's first delivery mints it (gateway/session_bot.py).
|
||||
relay = self.connect('alpha')
|
||||
delivered = await self.call(relay, 'bot_relay.deliver', id='a' * 32, profile='alpha',
|
||||
message='status?', author=AUTHOR)
|
||||
await async_wait_until(
|
||||
lambda: (read_delivery_result(self.homes['alpha'], 'a' * 32) or {}).get('status') == 'settled',
|
||||
timeout=30, interval=.05)
|
||||
# beta: provisioned by `hermes -p beta chat -c "Bot Chat" --create-if-missing`.
|
||||
cli = self.connect('beta')
|
||||
provisioned = await self.call(cli, 'session.create', request_id='provision', title='Bot Chat',
|
||||
source='cli', cwd=str(self.homes['beta']))
|
||||
# gamma: an ordinary chat first, then the Desktop's own mint on a bot's first click
|
||||
# (canonical-chat.ts), born hidden and newer.
|
||||
visible = await self.call(desktop, 'session.create', profile='gamma', request_id='side', source='gui')
|
||||
minted = await self.call(desktop, 'session.create', profile='gamma', title='Bot Chat',
|
||||
hidden=True, follow_profile_config=True)
|
||||
self.gamma_chat = visible['stored_session_id']
|
||||
return {'default': LEGACY_ID, 'alpha': delivered['session_id'],
|
||||
'beta': provisioned['stored_session_id'], 'gamma': minted['stored_session_id']}
|
||||
|
||||
def connections(self):
|
||||
# Every AuthorityConnection registers its transport with the authority it binds.
|
||||
return sum(len(authority.events) for authority in self.runner.session_authorities.serving())
|
||||
|
||||
|
||||
def _rows(result):
|
||||
return {row['name']: row for row in result['profiles']}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_every_bot_row_resolves_its_own_bot_chat_whichever_bot_is_active(fleet):
|
||||
chats = await fleet.bot_chats()
|
||||
# "Open recent session" offers the newest chat a picker shows: never a hidden Bot Chat.
|
||||
recent = {**chats, 'gamma': fleet.gamma_chat}
|
||||
desktop = fleet.connect('default', scope='host')
|
||||
for active in chats: # route the shared socket to every bot once, as a click does
|
||||
await fleet.call(desktop, 'session.list', profile=active)
|
||||
routed = fleet.connections()
|
||||
for active in chats:
|
||||
# The Desktop's roster poll: profiles.list of the active bot over the shared socket.
|
||||
rows = _rows(await fleet.call(desktop, 'profiles.list', profile=active))
|
||||
assert set(rows) == set(chats)
|
||||
for name, chat in chats.items():
|
||||
canonical = rows[name].get('canonical_session')
|
||||
assert canonical is not None, f'{name} row has no Bot Chat while {active} is active'
|
||||
assert canonical['resolved_id'] == chat, (active, name, canonical)
|
||||
stored = fleet.db(name).get_session(chat)
|
||||
assert canonical['last_active'] == (stored['last_activity_at'] or stored['started_at']) > 0
|
||||
assert (rows[name].get('last_session') or {}).get('id') == recent[name], (active, name, rows[name])
|
||||
assert fleet.connections() == routed, 'a roster poll built connections of its own'
|
||||
assert fleet.executed == [chats['alpha']]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_row_shows_a_bot_chat_its_viewer_may_not_read(fleet):
|
||||
chats = await fleet.bot_chats()
|
||||
# A profile-bound socket (`hermes -p beta chat`) can route nowhere else: its siblings stay
|
||||
# discovery metadata, even though each of them holds a Bot Chat.
|
||||
rows = _rows(await fleet.call(fleet.connect('beta'), 'profiles.list'))
|
||||
assert rows['beta']['canonical_session']['resolved_id'] == chats['beta']
|
||||
for name in ('default', 'alpha', 'gamma'):
|
||||
assert not {'canonical_session', 'last_session'} & set(rows[name]), rows[name]
|
||||
# Another principal without operator access reads no one's Bot Chat, on any row.
|
||||
foreign = fleet.connect('default', 'uid:other', scope='host', operator=False)
|
||||
for name, row in _rows(await fleet.call(foreign, 'profiles.list')).items():
|
||||
assert row.get('canonical_session') is None and row.get('last_session') is None, (name, row)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_sibling_that_cannot_be_read_safely_leaves_only_its_own_row_bare(fleet, monkeypatch, caplog):
|
||||
from hermes_state_errors import StateDbReplacedError
|
||||
chats = await fleet.bot_chats()
|
||||
|
||||
def replaced(*args, **kwargs):
|
||||
raise StateDbReplacedError('state.db was replaced')
|
||||
|
||||
monkeypatch.setattr(fleet.db('alpha'), '_read_ctx', replaced)
|
||||
# beta's store closes as an unserve would, between the retiring check and the read.
|
||||
fleet.db('beta').close()
|
||||
# gamma is being unserved: retirement marks it before its store closes.
|
||||
fleet.runner.session_authorities.for_home(fleet.homes['gamma']).retiring = True
|
||||
desktop = fleet.connect('default', scope='host')
|
||||
for _poll in range(2):
|
||||
rows = _rows(await fleet.call(desktop, 'profiles.list'))
|
||||
assert rows['default']['canonical_session']['resolved_id'] == chats['default']
|
||||
for name in ('alpha', 'beta', 'gamma'):
|
||||
assert not {'canonical_session', 'last_session'} & set(rows[name]), rows[name]
|
||||
assert fleet.db('beta')._conn is None, 'the roster reopened a store an unserve had closed'
|
||||
warned = [r.getMessage() for r in caplog.records if r.levelname == 'WARNING' and 'metadata-only' in r.getMessage()]
|
||||
for name in ('alpha', 'beta'): # once each across both polls; a retiring sibling is no failure
|
||||
profile_id = fleet.runner.session_authorities.for_home(fleet.homes[name]).profile_id
|
||||
assert sum(profile_id in message for message in warned) == 1, warned
|
||||
assert len(warned) == 2, warned
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize('fleet', [('delta',)], indirect=True)
|
||||
async def test_a_row_whose_name_routes_elsewhere_carries_no_bot_chat(fleet):
|
||||
# Served as ``delta`` from outside ``profiles/``, it is listed as ``default``: a click on that
|
||||
# row routes to the launch profile, so the row must not paint delta's Bot Chat.
|
||||
linked = fleet.db('delta')
|
||||
linked.create_session('local-delta', source='gui', chat_id='local-delta', user_id='uid:desk')
|
||||
linked.set_session_title('local-delta', 'Bot Chat')
|
||||
result = await fleet.call(fleet.connect('default', scope='host'), 'profiles.list')
|
||||
row, = [row for row in result['profiles'] if row['path'] == str(fleet.homes['delta'])]
|
||||
assert row['name'] == 'default'
|
||||
assert not {'canonical_session', 'last_session'} & set(row), row
|
||||
@@ -7,7 +7,7 @@ from hermes_state import SessionDB
|
||||
from hermes_state_local_migration import bind_legacy_target
|
||||
|
||||
|
||||
def test_adopted_bot_chat_keeps_canonical_and_recent_discovery(tmp_path, monkeypatch):
|
||||
def test_adopted_hidden_bot_chat_is_canonical_but_not_recent(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv('HERMES_HOME', str(tmp_path))
|
||||
sid = '20261006_105832_fixture'
|
||||
actor = Principal('fixture-owner', str(tmp_path), frozenset({'session:read'}), 'fixture')
|
||||
@@ -28,7 +28,7 @@ def test_adopted_bot_chat_keeps_canonical_and_recent_discovery(tmp_path, monkeyp
|
||||
assert row['canonical_session'] is not None, 'adopted Bot Chat disappeared from roster'
|
||||
assert row['canonical_session']['resolved_id'] == sid
|
||||
assert row['canonical_session']['preview'] == 'The retained bot reply'
|
||||
assert row['last_session']['resolved_id'] == sid
|
||||
assert row['last_session'] is None, 'a hidden Bot Chat is never the newest listed conversation'
|
||||
assert db._conn.total_changes == before, 'discovery must remain read-only'
|
||||
assert db.get_session(sid)['chat_id'] == sid
|
||||
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
"""The canonical gateway's roster memo (``cached_reader_session_fields``, #139227) keys on the reader
|
||||
as well as the store: one principal's answer is never served to another, and ``invalidate`` drops it
|
||||
like the other memos."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from tui_gateway import profile_roster_cache as cache
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store(tmp_path, monkeypatch):
|
||||
(tmp_path / "state.db").write_bytes(b"x" * 16)
|
||||
# The store reads as settled at once: the memo is exercised without waiting out the racy window.
|
||||
monkeypatch.setattr(cache, "mtime_is_racy", lambda mtime_ns: False)
|
||||
cache.invalidate()
|
||||
yield tmp_path
|
||||
cache.invalidate()
|
||||
|
||||
|
||||
def test_each_reader_gets_its_own_answer_until_the_store_moves(store):
|
||||
computed = []
|
||||
|
||||
def read(reader):
|
||||
def compute():
|
||||
computed.append(reader)
|
||||
return {"canonical_session": {"reader": reader}}
|
||||
return cache.cached_reader_session_fields(store, reader, compute)["canonical_session"]
|
||||
|
||||
owner, other = ("uid:desk", True), ("uid:other", False)
|
||||
assert read(owner) == {"reader": owner}
|
||||
assert read(other) == {"reader": other}, "another reader was served the owner's answer"
|
||||
assert read(owner) == {"reader": owner}
|
||||
assert computed == [owner, other], "an unchanged store was read again"
|
||||
|
||||
(store / "state.db").write_bytes(b"y" * 32)
|
||||
assert read(owner) == {"reader": owner}
|
||||
assert computed == [owner, other, owner], "a moved store was served from the memo"
|
||||
|
||||
cache.invalidate(store)
|
||||
read(other)
|
||||
assert computed[-1] == other, "invalidate() kept the reader memo"
|
||||
@@ -36,6 +36,9 @@ from utils import mtime_is_racy
|
||||
# profile leaves one stale entry, so a memo is dropped whole once it grows past any plausible fleet.
|
||||
_SESSION_CACHE: dict[str, tuple[tuple, dict]] = {}
|
||||
_UI_META_CACHE: dict[str, tuple[tuple, dict]] = {}
|
||||
# The canonical gateway's rows also depend on who reads them (its owner filter): keyed by
|
||||
# (profile path, reader).
|
||||
_READER_SESSION_CACHE: dict[tuple, tuple[tuple, dict]] = {}
|
||||
_MAX_ENTRIES = 512
|
||||
|
||||
_STORE_FILES = ("state.db", "state.db-wal")
|
||||
@@ -90,6 +93,13 @@ def cached_session_fields(profile_path: str | Path, compute: Callable[[], dict])
|
||||
return _cached(_SESSION_CACHE, str(profile_path), store_signature(profile_path), compute, dict)
|
||||
|
||||
|
||||
def cached_reader_session_fields(profile_path: str | Path, reader: tuple,
|
||||
compute: Callable[[], dict]) -> dict[str, Any]:
|
||||
"""``cached_session_fields`` for fields that also depend on *reader*, a hashable principal key."""
|
||||
return _cached(_READER_SESSION_CACHE, (str(profile_path), reader), store_signature(profile_path),
|
||||
compute, dict)
|
||||
|
||||
|
||||
def cached_ui_meta_fields(profile_dir: str | Path, compute: Callable[[], dict]) -> dict[str, Any]:
|
||||
"""``compute()``'s fields, reused while the profile's ``profile.yaml`` has not changed.
|
||||
|
||||
@@ -106,3 +116,5 @@ def invalidate(profile_path: str | Path | None = None) -> None:
|
||||
cache.clear()
|
||||
else:
|
||||
cache.pop(str(profile_path), None)
|
||||
for key in [key for key in _READER_SESSION_CACHE if profile_path is None or key[0] == str(profile_path)]:
|
||||
_READER_SESSION_CACHE.pop(key, None)
|
||||
@@ -22,13 +22,22 @@ hosted-room driver still uses legacy TUI session handlers; the canonical surface
|
||||
therefore reports `driver: false` and does not advertise send, retry, approval,
|
||||
or peer-grant operations. It never imports that driver to execute a group turn.
|
||||
|
||||
`profiles.list` requires `session:read` and returns only the authenticated
|
||||
profile, including its display metadata and avatar presence. `include_sessions`
|
||||
defaults to true: previews are limited to that actor's canonical local bindings,
|
||||
and `canonical_session` resolves the exact hidden `Bot Chat` title, not the most
|
||||
recent conversation. Discovery does not create, restore, or unarchive sessions,
|
||||
open sibling profile databases, or run inference. Cross-profile discovery requires
|
||||
a separately authenticated connection to the relevant owner.
|
||||
`profiles.list` requires `session:read` and returns the authenticated profile
|
||||
first, with its display metadata and avatar presence, then the served siblings and
|
||||
hosted-room destinations as discovery metadata. `include_sessions` defaults to
|
||||
true: `canonical_session` resolves the exact `Bot Chat` title, not the most recent
|
||||
conversation, and `last_session` is the newest visible local conversation (a hidden
|
||||
row never wins). Both count only local chats the actor owns or reads as an operator:
|
||||
minted `local-*` chats, and pre-upgrade chats adopted under their durable binding
|
||||
(they keep their old id). A sibling's row carries the same fields only when this
|
||||
connection already routes session verbs to it (a host-scoped grant: the Desktop's
|
||||
shared socket, the dashboard), read from that sibling's own authority with this
|
||||
connection's subject and capabilities. A profile-bound ticket gets metadata alone,
|
||||
and so does a sibling that is retiring, listed under a name that routes elsewhere,
|
||||
or whose store doesn't answer; a sibling still starting is not listed yet. Rows are
|
||||
reused while a profile's store has not moved. Discovery does not create, restore,
|
||||
or unarchive sessions, open a database handle of its own or reopen one an unserve
|
||||
closed, or run inference.
|
||||
|
||||
## Canonical ancillary reads
|
||||
|
||||
|
||||
Reference in new issue
Block a user