mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
ef7984f056
groups.state member_activity carried tool/progress_at (1cd4f83) only for the legacy in-process room path: the map is fed by tui_gateway.hosted_room_member_activity.project_room_member_activity, which sees tui_gateway session frames. A gateway-owned (canonical) room runs its member turns on the gateway's canonical local-session path, whose frames go SessionAuthority.publish_execution -> the member's SessionEvents stream and never reach that projection. So the room showed "scout is working... 6s" while Scout sat in a terminal tool. Live diagnosis (real Electron + multiplexed gateway, mock inference): one process (`hermes_cli.main --profile default gateway run`) answered groups.state (CanonicalHostedRoomService.status), ran the member turn (mute_notification_reply False, so tool callbacks are wired), published tool.start/tool.complete name=terminal for the hidden member session, and answered HostedRoomAuthorityRPC.info for it every active poll. Not a process split and not a mute gate: the frames just never reach the room's map. Design: the member side reports, the owner side records, over the info poll the room driver already makes during a turn (no new polling, and the member never opens the room owner's database, as session_hosted_transport requires): - SessionEvents keeps the running execution's open tool calls and last frame time, fed by publish_execution after its generation fence (late callbacks from another generation move nothing). - HostedRoomAuthorityRPC.info adds `tool` (newest open call, absent between tools) and `progress_at` for its current started hosted task. The same answer travels the private owner socket for another local profile's member. - The driver hands every info answer to a new task_progress callback beside pending_action; HostedRoomService records it in the existing bounded map only when the info names the same task, and forgets the entry when the task's terminal is published. status() merges it as before. The 1cd4f83 in-process feed stays: the legacy HostedRoomServerRPC path runs turns in the process answering groups.state, its frames reach the projection directly, and its info carries no progress fields, so the two feeds never compete. No wire contract changes: RoomMemberActivity already declares tool/progress_at and the hosted info verb is internal to the private socket.
344 lines
19 KiB
Python
344 lines
19 KiB
Python
"""Private hosted-room producer over the gateway's canonical local-session FIFO.
|
|
|
|
One instance belongs to a durable room/member/profile binding. The service supplies
|
|
its principal and current-membership check; neither is taken from RPC payloads.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import asdict, replace
|
|
from functools import partial
|
|
import hashlib
|
|
import json
|
|
|
|
from gateway.hosted_room_driver import TaskIdentity
|
|
from gateway.owner_identity import same_owner
|
|
from gateway.session_contract import SessionRef, Submission
|
|
from hermes_state_runtime import RuntimeStoreError, list_session_admissions
|
|
|
|
_RESULTLESS_OUTCOMES = frozenset({'interrupted', 'cancelled'})
|
|
# Encoded receipts per ``results`` read; under the 512 KiB private-socket response line.
|
|
_RESULTS_BUDGET_BYTES = 384 * 1024
|
|
|
|
|
|
class HostedRoomAuthorityRPC:
|
|
accepts_attachments = True # submit resolves event-bound bytes per member
|
|
|
|
def __init__(self, authority, loop, *, room_id, member_id, profile, principal,
|
|
authorize, authorize_write=None, timeout=30):
|
|
self.authority, self.loop = authority, loop
|
|
self.room_id, self.member_id, self.profile = room_id, member_id, profile
|
|
self.principal, self.authorizer, self.timeout = principal, authorize, timeout
|
|
self.authorize_write = authorize_write
|
|
self.callbacks = {}
|
|
binding = json.dumps([room_id, member_id, profile], separators=(',', ':'))
|
|
self.creation_id = 'hosted:' + hashlib.sha256(binding.encode()).hexdigest()
|
|
from gateway.session_local_recovery import local_identity
|
|
self.ref = SessionRef(authority.profile_id, local_identity(
|
|
authority.profile_id, principal.subject, self.creation_id))
|
|
|
|
def _call(self, operation, **params):
|
|
try:
|
|
running = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
running = None
|
|
if running is self.loop:
|
|
raise RuntimeStoreError('invalid_params')
|
|
if not self.loop.is_running():
|
|
raise RuntimeStoreError('runtime_draining')
|
|
future = asyncio.run_coroutine_threadsafe(self._dispatch(operation, params), self.loop)
|
|
# A timeout is ambiguous: never cancel an admission or claim not_admitted.
|
|
return future.result(self.timeout)
|
|
|
|
async def _dispatch(self, operation, params):
|
|
from gateway.session_authorities import owner_scope
|
|
with owner_scope(self.authority):
|
|
return await self._dispatch_owned(operation, params)
|
|
|
|
async def _dispatch_owned(self, operation, params):
|
|
if (params.get('profile', self.profile) != self.profile
|
|
or self.principal.profile_id != self.authority.profile_id):
|
|
raise RuntimeStoreError('profile_mismatch')
|
|
if (params.get('source', 'bot_room') != 'bot_room'
|
|
or params.get('session_id', self.ref.session_id) != self.ref.session_id
|
|
or params.get('title', f'Group: {self.room_id}') != f'Group: {self.room_id}'):
|
|
raise RuntimeStoreError('permission_denied')
|
|
task, generation = params.get('task'), params.get('execution_generation')
|
|
if operation == 'submit' and (not isinstance(task, TaskIdentity)
|
|
or task.room_id != self.room_id or type(generation) is not int or generation < 1):
|
|
raise RuntimeStoreError('invalid_params')
|
|
if self.authorizer(operation, task, generation) is not True:
|
|
raise RuntimeStoreError('permission_denied')
|
|
return await getattr(self, '_' + operation)(params)
|
|
|
|
async def _resolve_exact(self, params):
|
|
if self.authority.db.get_session(self.ref.session_id) is None:
|
|
return None
|
|
return await self._resume(params)
|
|
|
|
async def _create(self, params):
|
|
if self.authority.db.get_session(self.ref.session_id) is not None:
|
|
return await self._resume(params)
|
|
from gateway.run import _load_gateway_config, _resolve_gateway_model
|
|
from gateway.session_policy import build_policy
|
|
from gateway.session_local import stage_local_session
|
|
from gateway.session_local_recovery import restore_local_session
|
|
from gateway.session_runtime_workers import tracked_write
|
|
from hermes_state_local import commit_local_session
|
|
from agent.skill_utils import parse_config_string_list
|
|
private = {}
|
|
config = _load_gateway_config()
|
|
# A room has no clarification reply (group controls expose approvals only), so a member
|
|
# turn that called clarify would block for clarify_timeout with its question unseen. As
|
|
# for cron, freeze it out; the member asks its question in the room reply instead.
|
|
agent = dict(config.get('agent') or {})
|
|
agent['disabled_toolsets'] = sorted(
|
|
set(parse_config_string_list(agent.get('disabled_toolsets'))) | {'clarify'})
|
|
policy = build_policy({'source': 'gui'}, {**config, 'agent': agent}, private_secrets=private)
|
|
policy = replace(policy, source='bot_room', platform='bot_room',
|
|
model=policy.model or _resolve_gateway_model(policy.config()),
|
|
toolsets=tuple(sorted(set(policy.toolsets) | {'bot_room'})))
|
|
staged = stage_local_session(self.authority, self.principal, {'request_id': self.creation_id},
|
|
trusted_policy=policy, trusted_secrets=private)
|
|
# Creation commit and label run off the owner loop (N90); a concurrent duplicate converges.
|
|
if isinstance(staged, dict):
|
|
await tracked_write(self.authority, partial(
|
|
commit_local_session, self.authority.db, epoch=self.authority.epoch, receipt=staged))
|
|
restore_local_session(self.authority, staged['session_id'])
|
|
from hermes_state_runtime import _epoch
|
|
def label(conn):
|
|
_epoch(conn, self.authority.epoch)
|
|
conn.execute('UPDATE sessions SET title=?, hidden=1 WHERE id=?',
|
|
(f'Group: {self.room_id}', self.ref.session_id))
|
|
await tracked_write(self.authority, partial(self.authority.db._execute_write, label))
|
|
return await self._resume(params)
|
|
|
|
async def _resume(self, params):
|
|
await self.authority.resolve(self.principal, self.ref)
|
|
return {'session_id': self.ref.session_id, 'title': f'Group: {self.room_id}'}
|
|
|
|
def _rows(self):
|
|
self.authority.authorize(self.principal, self.ref, 'session:read')
|
|
rows = list_session_admissions(self.authority.db, session_id=self.ref.session_id, pending_only=False)
|
|
result = []
|
|
for row in rows:
|
|
if not same_owner(row['principal_id'], self.principal.subject) or not row['request_id'].startswith('hosted:'):
|
|
continue
|
|
try:
|
|
task, generation = json.loads(row['request_id'][7:])
|
|
identity = TaskIdentity(**task)
|
|
if identity.room_id != self.room_id or type(generation) is not int:
|
|
raise ValueError('invalid hosted identity')
|
|
except (TypeError, ValueError, KeyError) as exc:
|
|
raise RuntimeStoreError('storage_unavailable') from exc
|
|
result.append((row, identity, generation))
|
|
return result
|
|
|
|
def _terminal(self, row, task, generation):
|
|
from gateway.session_results import admission_result
|
|
saved = admission_result(self.authority.db, row['admission_id'])
|
|
# Unknown discard and queued cancellation never ran: no result exists to recover.
|
|
if saved is None and row['outcome'] not in _RESULTLESS_OUTCOMES:
|
|
raise RuntimeStoreError('storage_unavailable')
|
|
value = saved['result'] if saved is not None else {}
|
|
status = {'completed': 'settled', 'interrupted': 'cancelled', 'cancelled': 'cancelled'}.get(row['outcome'], 'failed')
|
|
receipt = {'status': status, 'text': value.get('final_response', ''),
|
|
'message_id': row['admission_id'], 'settlement_id': row['admission_id'],
|
|
'task_id': task.task_id, 'execution_generation': generation}
|
|
callback = self.callbacks.pop(row['admission_id'], None)
|
|
if callback is not None:
|
|
callback(receipt)
|
|
return receipt
|
|
|
|
async def _submit(self, params):
|
|
task, generation = params['task'], params['execution_generation']
|
|
request_id = 'hosted:' + json.dumps([asdict(task), generation], sort_keys=True, separators=(',', ':'))
|
|
# Refuse unknown before submit: submit itself schedules the queue on retries.
|
|
rows = self._rows()
|
|
if any(row['status'] == 'unknown' for row, _, _ in rows):
|
|
raise RuntimeStoreError('unknown_execution')
|
|
from gateway.session_hosted_attachments import prepare_before_submit, submission_payload
|
|
from gateway.session_ingress_media import capture_lease
|
|
# One lease from capture through the admission commit: documents ride in the prompt text,
|
|
# so submit's own lease does not hold them, and an older row's release could unlink a
|
|
# shared content-addressed capture in between.
|
|
with capture_lease():
|
|
payload = await asyncio.to_thread(prepare_before_submit, partial(
|
|
submission_payload, self, params['prompt'], params.get('attachments')))
|
|
authorize_write = self.authorize_write
|
|
receipt = await self.authority.submit(self.principal, Submission(
|
|
request_id, self.ref, payload, 'queue'),
|
|
_authorize_write=(lambda conn: authorize_write(conn, task, generation))
|
|
if authorize_write is not None else None)
|
|
self.callbacks[receipt.admission_id] = params['on_terminal']
|
|
if receipt.status in {'queued', 'started'}:
|
|
waiter = self.authority.waiters.get(receipt.admission_id)
|
|
if waiter is None:
|
|
waiter = self.loop.create_future()
|
|
self.authority.waiters[receipt.admission_id] = waiter
|
|
waiter.add_done_callback(lambda done: self._completed(receipt.admission_id, done))
|
|
if receipt.status == 'terminal':
|
|
for row, identity, gen in self._rows():
|
|
if row['admission_id'] == receipt.admission_id:
|
|
self._terminal(row, identity, gen)
|
|
return asdict(receipt)
|
|
|
|
def _completed(self, admission_id, future):
|
|
if future.cancelled() or admission_id not in self.callbacks:
|
|
return
|
|
if future.exception() is not None:
|
|
# A paused FIFO remains queued/unknown; polling or an explicit
|
|
# resolution will publish its eventual durable terminal outcome.
|
|
return
|
|
if self.authorizer('terminal', None, None) is not True:
|
|
self.callbacks.pop(admission_id, None)
|
|
return
|
|
for row, task, generation in self._rows():
|
|
if row['admission_id'] == admission_id and row['status'] == 'terminal':
|
|
self._terminal(row, task, generation)
|
|
|
|
async def _history(self, params):
|
|
from gateway.session_local_recovery import local_history, prepare_history
|
|
# Authorize (restoring a cold room session) before the read names its live entry, and again after
|
|
# the await (``_rows``): a delete in between answers not_found, never a KeyError.
|
|
self.authority.authorize(self.principal, self.ref, 'session:read')
|
|
prepared = await prepare_history(self.authority, self.ref)
|
|
rows = self._rows()
|
|
history = self.authority.unless_deleted(
|
|
self.ref, lambda: list(local_history(self.authority, self.ref, prepared)))
|
|
for row, task, generation in rows:
|
|
if row['status'] == 'terminal':
|
|
receipt = self._terminal(row, task, generation)
|
|
history.append({**receipt, 'role': 'assistant', 'content': receipt['text']})
|
|
return history
|
|
|
|
async def _results(self, params):
|
|
"""Terminal receipts only, never the transcript: one bounded response line.
|
|
|
|
``admission_ids`` narrows the read to the caller's pending callbacks. Each text
|
|
keeps just over the driver's persisted bound, so its own truncation (and notice)
|
|
is unchanged; newest receipts win the byte budget and the rest follow next poll.
|
|
"""
|
|
wanted = params.get('admission_ids')
|
|
if wanted is not None and (not isinstance(wanted, list) or len(wanted) > 256
|
|
or not all(isinstance(i, str) and i for i in wanted)):
|
|
raise RuntimeStoreError('invalid_params')
|
|
from tui_gateway.hosted_room_driver import MAX_TERMINAL_TEXT_BYTES
|
|
receipts, used = [], 0
|
|
for row, task, generation in reversed(self._rows()):
|
|
if row['status'] != 'terminal' or (wanted is not None and row['admission_id'] not in wanted):
|
|
continue
|
|
receipt = self._terminal(row, task, generation)
|
|
raw = receipt['text'].encode('utf-8')
|
|
if len(raw) > MAX_TERMINAL_TEXT_BYTES:
|
|
receipt['text'] = raw[:MAX_TERMINAL_TEXT_BYTES + 4].decode('utf-8', 'ignore')
|
|
used += len(json.dumps(receipt))
|
|
if receipts and used > _RESULTS_BUDGET_BYTES:
|
|
break
|
|
receipts.append(receipt)
|
|
return receipts[::-1]
|
|
|
|
async def _info(self, params):
|
|
rows = self._rows()
|
|
for row, task, generation in rows:
|
|
if row['status'] == 'terminal':
|
|
self._terminal(row, task, generation)
|
|
current = next(((row, task, generation) for row, task, generation in rows if row['status'] in {'started', 'unknown', 'queued'}), None)
|
|
result = {'active': current is not None and current[0]['status'] != 'unknown',
|
|
'task_id': current[1].task_id if current else None,
|
|
'execution_generation': current[2] if current else None,
|
|
'status': current[0]['status'] if current else 'idle'}
|
|
live = self.authority.sessions.get(self.ref.session_id)
|
|
if current is not None and current[0]['status'] == 'started' and live is not None:
|
|
# The member's current tool and last frame, for the room's "scout · terminal · 7s".
|
|
result.update(live.event_stream.progress_for(current[0]['generation']))
|
|
snapshot = await self.authority.attach(self.principal, self.ref, history=False) # prompts only
|
|
if snapshot.prompts:
|
|
prompt = next((p for p in snapshot.prompts if p.get('kind') == 'approval'), None)
|
|
if prompt:
|
|
result.update(status='waiting_for_approval', pending_approval={
|
|
**prompt, 'request_id': prompt['prompt_id'],
|
|
'choices': [c for c in prompt['choices'] if c in {'once', 'deny'}]})
|
|
return result
|
|
|
|
async def _interrupt(self, params):
|
|
rows = self._rows()
|
|
current = next(((row, task) for row, task, _ in rows if row['status'] in {'started', 'unknown', 'queued'}), None)
|
|
if current is None or current[1].task_id != params['expected_task_id']:
|
|
raise RuntimeStoreError('stale_generation')
|
|
row, _ = current
|
|
if row['status'] == 'unknown':
|
|
raise RuntimeStoreError('unknown_execution')
|
|
if row['status'] == 'queued':
|
|
row = await self.authority.cancel_or_claimed(self.principal, self.ref, row['admission_id'])
|
|
if row is not None:
|
|
await self.authority.interrupt(self.principal, self.ref, row['generation'])
|
|
return {'interrupted': True, 'status': 'interrupted'}
|
|
|
|
async def _discard(self, params):
|
|
generation = params['execution_generation']
|
|
if type(generation) is not int or generation < 1:
|
|
raise RuntimeStoreError('invalid_params')
|
|
matches = [(row, task) for row, task, hosted_generation in self._rows()
|
|
if (row['status'] == 'unknown' or (row['status'] == 'terminal' and row['outcome'] == 'interrupted'))
|
|
and task.task_id == params['expected_task_id']
|
|
and hosted_generation == generation]
|
|
if len(matches) != 1:
|
|
raise RuntimeStoreError('stale_generation')
|
|
row, task = matches[0]
|
|
# The public fence is hosted; the canonical CAS uses its own generation.
|
|
if row['status'] == 'unknown':
|
|
await self.authority.resolve_unknown(
|
|
self.principal, self.ref, row['admission_id'], row['generation'])
|
|
return {'discarded': True, 'status': 'cancelled', 'task_id': task.task_id,
|
|
'execution_generation': generation}
|
|
|
|
async def _approve(self, params):
|
|
if params['choice'] not in {'once', 'deny'}:
|
|
raise RuntimeStoreError('invalid_params')
|
|
snapshot = await self.authority.attach(self.principal, self.ref, history=False) # prompts only
|
|
prompt = next((p for p in snapshot.prompts if p.get('prompt_id') == params['request_id']
|
|
and p.get('kind') == 'approval'), None)
|
|
if prompt is None:
|
|
raise RuntimeStoreError('stale_generation')
|
|
return await self.authority.respond(self.principal, self.ref,
|
|
snapshot.handle.execution_generation, params['request_id'], {'choice': params['choice']})
|
|
|
|
def resolve_exact(self, *, profile, title, source):
|
|
return self._call('resolve_exact', profile=profile, title=title, source=source)
|
|
|
|
def create(self, *, profile, title, source):
|
|
return self._call('create', profile=profile, title=title, source=source)
|
|
|
|
def resume(self, *, profile, session_id, source):
|
|
return self._call('resume', profile=profile, session_id=session_id, source=source)
|
|
|
|
def submit(self, *, profile, session_id, prompt, source, task, execution_generation, on_terminal,
|
|
member_id=None, attachments=None):
|
|
# One producer per member binding: a task for another member reached the wrong producer.
|
|
if member_id is not None and member_id != self.member_id:
|
|
raise RuntimeStoreError('permission_denied')
|
|
return self._call('submit', profile=profile, session_id=session_id, source=source,
|
|
prompt=prompt, task=task, execution_generation=execution_generation, on_terminal=on_terminal,
|
|
attachments=attachments)
|
|
|
|
def history(self, *, profile, session_id, source):
|
|
return self._call('history', profile=profile, session_id=session_id, source=source)
|
|
|
|
def results(self, *, profile, session_id, source, admission_ids=None):
|
|
return self._call('results', profile=profile, session_id=session_id, source=source,
|
|
admission_ids=admission_ids)
|
|
|
|
def info(self, *, profile, session_id, source):
|
|
return self._call('info', profile=profile, session_id=session_id, source=source)
|
|
|
|
def interrupt(self, *, profile, session_id, source, expected_task_id):
|
|
return self._call('interrupt', profile=profile, session_id=session_id, source=source, expected_task_id=expected_task_id)
|
|
|
|
def discard(self, *, profile, session_id, source, expected_task_id, execution_generation):
|
|
return self._call('discard', profile=profile, session_id=session_id, source=source,
|
|
expected_task_id=expected_task_id, execution_generation=execution_generation)
|
|
|
|
def approve(self, *, session_id, request_id, choice):
|
|
return self._call('approve', session_id=session_id, request_id=request_id, choice=choice)
|