mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
kshitijk4poor
c91bd232c0
The salvaged commit raised the managed worker's owner-link open timeout from 5 s to 60 s (with a HERMES_WORKER_RPC_OPEN_TIMEOUT override) and retried a timed-out handshake three times. Those retries cannot succeed against a real owner: WorkerRPC._connection mints ONE worker-adoption ticket and every attempt resent it, while the gateway's bootstrap tickets are single-use and expire after 30 s (gateway/runtime_bootstrap.py TTL_SECONDS, redeem() deletes the entry). An attempt past 30 s, or any retry, is refused, and the same path serves every idle redial during a turn, so a failing redial now took about 3 minutes instead of 5 s while holding WorkerRPC.lock. The env knob also went against the repo's no-new-HERMES_*-config rule. The lost-worker disposition (worker_lost event, settled-admission reclaim) does not depend on the handshake budget: a worker that dies in worker.adopt is now booked within one tick either way. Sizing the handshake needs a fresh ticket per attempt and a budget within the ticket TTL, which is its own change.
498 lines
23 KiB
Python
498 lines
23 KiB
Python
"""Explicit worker persistence facade; never opens the canonical SQLite store.
|
|
|
|
This first operation family is deliberately NOT a complete SessionDB substitute.
|
|
No existing cron/child/compute consumer is switched until its inventory is covered.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import contextvars
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import queue
|
|
import sys
|
|
import threading
|
|
from hermes_dir_acl import make_user_dir
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_worker_process = False
|
|
|
|
|
|
def is_worker_process():
|
|
"""A successful self-registration makes this interpreter compute-only.
|
|
|
|
Process-scoped (not ContextVar): import-time recovery and fresh helper threads
|
|
must not regain owner-ledger access. This is cooperative runtime ownership,
|
|
not OS confinement or an authorization credential.
|
|
"""
|
|
return _worker_process
|
|
|
|
|
|
|
|
def apply_row_annotations(messages, annotations):
|
|
"""Mirror the owner's in-place row stamps onto the worker's dicts (None = the owner popped it)."""
|
|
for message, annotation in zip(messages, annotations, strict=True):
|
|
for key, value in annotation.items():
|
|
if value is None:
|
|
message.pop(key, None)
|
|
else:
|
|
message[key] = value
|
|
|
|
class WorkerPersistenceError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _caller_summary(limit=8):
|
|
"""module:function:line of the frames above this store (its own frames skipped), never
|
|
locals or argument values: best-effort callers swallow the guard's raise (#139029)."""
|
|
frame, frames = sys._getframe(2), []
|
|
while frame is not None and len(frames) < limit:
|
|
module = frame.f_globals.get('__name__', '?')
|
|
if not module.startswith('agent.runtime_session_'):
|
|
frames.append(f'{module}:{frame.f_code.co_name}:{frame.f_lineno}')
|
|
frame = frame.f_back
|
|
return ' <- '.join(frames)
|
|
|
|
|
|
def call_off_event_loop(fn, *args, **kwargs):
|
|
"""Call ``fn`` here, or on a worker thread when this thread is running an event loop.
|
|
|
|
For synchronous best-effort writers that can run inside a callback Relay drives on its private
|
|
loop (provider/tool callbacks): there ``WorkerRPC`` refuses the call (``synchronous_rpc_on_event_loop``,
|
|
sticky for the rest of the turn; #139029). The callback already blocks that loop, so waiting for the
|
|
thread costs it nothing; the context is copied so session-scoped state survives the hop, and errors
|
|
re-raise here. Async code awaits ``asyncio.to_thread`` instead.
|
|
"""
|
|
try:
|
|
asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
return fn(*args, **kwargs)
|
|
context = contextvars.copy_context()
|
|
with ThreadPoolExecutor(max_workers=1, thread_name_prefix='worker-persist-off-loop') as pool:
|
|
return pool.submit(context.run, fn, *args, **kwargs).result()
|
|
|
|
|
|
class _OwnerLink:
|
|
"""One owner WebSocket and the only thread that reads it. Events the owner broadcasts to every
|
|
transport are discarded as they arrive, idle or not: left queued, websockets' bounded receive
|
|
buffer pauses socket reads, which also stalls pongs and the close handshake. The reply to the
|
|
one outstanding request id goes to its caller; a closed connection goes to it as the exception."""
|
|
def __init__(self, ws):
|
|
self.ws = ws
|
|
self.awaiting = None
|
|
self.replies = queue.SimpleQueue()
|
|
self.reader = threading.Thread(target=self._read, name='worker-rpc-reader', daemon=True)
|
|
self.reader.start()
|
|
|
|
def _read(self):
|
|
try:
|
|
while True:
|
|
try:
|
|
frame = json.loads(self.ws.recv())
|
|
except ValueError:
|
|
continue
|
|
if isinstance(frame, dict) and self.awaiting is not None and frame.get('id') == self.awaiting:
|
|
self.replies.put(frame)
|
|
except Exception as exc: # health: allow BLE001 -- handed to the caller, which raises it (ConnectionClosed included)
|
|
self.replies.put(exc)
|
|
|
|
def send(self, request_id, message):
|
|
self.awaiting = request_id
|
|
self.ws.send(message)
|
|
|
|
def receive(self, timeout):
|
|
try:
|
|
reply = self.replies.get(timeout=timeout)
|
|
except queue.Empty:
|
|
raise TimeoutError('worker_receipt_timeout') from None
|
|
if isinstance(reply, BaseException):
|
|
raise reply
|
|
return reply
|
|
|
|
|
|
class WorkerRPC:
|
|
"""Bounded authenticated calls to an existing owner; never starts a daemon.
|
|
|
|
No owner-loop synchronous self-RPC or SQLite fallback is permitted.
|
|
One connection serves every call (serialized by ``lock``, correlated by id); it is
|
|
discovered, identified and ticketed on connect exactly as each call once was.
|
|
"""
|
|
def __init__(self, home):
|
|
self.home = Path(home).resolve()
|
|
self.lock = threading.Lock()
|
|
self.endpoint = None
|
|
self.link = None
|
|
self.request_id = 0
|
|
|
|
def _connection(self):
|
|
if self.link is not None:
|
|
if self.link.reader.is_alive():
|
|
return self.link
|
|
# The owner closed it while idle, before any send: redial with a fresh ticket.
|
|
self._drop()
|
|
from hermes_cli.gateway_client import _session_ticket, gateway_ws_target
|
|
from hermes_cli.gateway_runtime_discovery import query_identify
|
|
from websockets.sync.client import connect
|
|
# A served secondary has no socket; its multiplexer's descriptor names it.
|
|
from hermes_cli.gateway_runtime import discover_gateway_endpoint
|
|
discovery = discover_gateway_endpoint(self.home, timeout=5)
|
|
if discovery.state == 'ready' and discovery.endpoint is not None:
|
|
self.endpoint = discovery.endpoint
|
|
elif discovery.state != 'draining' or self.endpoint is None:
|
|
raise WorkerPersistenceError('owner_unavailable')
|
|
from hermes_cli.gateway_runtime import control_home_for
|
|
endpoint = self.endpoint
|
|
descriptor = query_identify(control_home_for(self.home, endpoint), timeout=5)
|
|
if descriptor.get('pid') == os.getpid():
|
|
raise WorkerPersistenceError('synchronous_self_rpc')
|
|
ticket = _session_ticket(self.home, endpoint, purpose='worker-adoption')
|
|
url, protocols = gateway_ws_target(endpoint, ticket)
|
|
# The owner is loopback, like connect_gateway's peer: an inherited HTTPS_PROXY must not carry it.
|
|
# No library keepalive: its pong deadline would run independently of (and could fail) a call
|
|
# inside its receipt budget; that budget and the observed close are the liveness checks.
|
|
self.link = _OwnerLink(connect(url, subprotocols=protocols, open_timeout=5, close_timeout=1,
|
|
max_size=8 * 1024 * 1024, proxy=None, ping_interval=None))
|
|
return self.link
|
|
|
|
def _drop(self):
|
|
link, self.link = self.link, None
|
|
if link is not None:
|
|
with contextlib.suppress(OSError): # a broken socket is already gone
|
|
link.ws.close()
|
|
|
|
def close(self):
|
|
with self.lock:
|
|
self._drop()
|
|
|
|
def __call__(self, method, **params):
|
|
try:
|
|
asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
pass
|
|
else:
|
|
logger.warning('synchronous_rpc_on_event_loop: %s on thread %s; caller: %s',
|
|
method, threading.current_thread().name, _caller_summary())
|
|
raise WorkerPersistenceError('synchronous_rpc_on_event_loop')
|
|
from websockets.exceptions import ConnectionClosed
|
|
with self.lock:
|
|
self.request_id += 1
|
|
request_id = self.request_id
|
|
message = json.dumps({'jsonrpc': '2.0', 'id': request_id, 'method': method, 'params': params})
|
|
try:
|
|
link = self._connection()
|
|
try:
|
|
link.send(request_id, message)
|
|
except ConnectionClosed:
|
|
# websockets raises this from send only before a complete frame is written (closed
|
|
# on entry, or the socket write failed): the owner closed the link after the idle
|
|
# check and never saw this request, so redial and send it once more.
|
|
self._drop()
|
|
link = self._connection()
|
|
link.send(request_id, message)
|
|
response = link.receive(20)
|
|
except BaseException:
|
|
# Outcome unknown once sent: surface the error and never resend; the next call redials.
|
|
self._drop()
|
|
raise
|
|
if 'error' in response:
|
|
raise WorkerPersistenceError(response['error'].get('message', 'persistence_failed'))
|
|
if method in ('worker.register', 'worker.adopt') and params.get('pid') == os.getpid():
|
|
global _worker_process
|
|
_worker_process = True
|
|
return response['result']
|
|
|
|
|
|
from agent.runtime_session_compression import RuntimeSessionCompressionMixin
|
|
|
|
|
|
from agent.runtime_session_lifecycle import RuntimeSessionLifecycleMixin
|
|
|
|
|
|
class RuntimeSessionStore(RuntimeSessionCompressionMixin, RuntimeSessionLifecycleMixin):
|
|
"""Synchronous durable results with a private byte-bounded retry journal.
|
|
|
|
A failed write remains pending, and failure is sticky until explicit retry or
|
|
adoption. Local queue acceptance is never returned as canonical commit.
|
|
Pending journals live outside age-pruned cache trees.
|
|
"""
|
|
# The process's async-delegation ledger belongs to the adopted execution, never a reserved child.
|
|
binds_delegation_ledger = True
|
|
|
|
def __init__(self, rpc, scope, outbox_dir, adopted=None, *, max_bytes=4 * 1024 * 1024):
|
|
self.rpc = rpc
|
|
self.scope = dict(scope)
|
|
self.max_bytes = max_bytes
|
|
self.lock = threading.RLock()
|
|
self.failure = None
|
|
self.path = Path(outbox_dir) / 'pending.json'
|
|
make_user_dir(self.path.parent, mode=0o700, parents=True, exist_ok=True)
|
|
if self.path.parent.is_symlink() or self.path.is_symlink():
|
|
raise WorkerPersistenceError('unsafe_outbox')
|
|
os.chmod(self.path.parent, 0o700)
|
|
from gateway.status import _try_acquire_file_lock
|
|
lock_path = self.path.parent / 'owner.lock'
|
|
if lock_path.is_symlink():
|
|
raise WorkerPersistenceError('unsafe_outbox')
|
|
self._outbox_owner = lock_path.open('a+', encoding='utf-8')
|
|
os.chmod(lock_path, 0o600)
|
|
if not _try_acquire_file_lock(self._outbox_owner):
|
|
self._outbox_owner.close()
|
|
raise WorkerPersistenceError('outbox_in_use')
|
|
try:
|
|
if self.path.exists():
|
|
if self.path.stat().st_size > max_bytes:
|
|
raise WorkerPersistenceError('outbox_full')
|
|
self.journal = json.loads(self.path.read_text(encoding="utf-8-sig"))
|
|
reconciled = self.journal['scope'] != self.scope
|
|
if reconciled:
|
|
self._reconcile_adopted(adopted)
|
|
else:
|
|
reconciled = False
|
|
self.journal = {'scope': self.scope, 'next_sequence': 1, 'pending': []}
|
|
self._save(self.journal)
|
|
if self.journal['pending']:
|
|
self.retry_pending()
|
|
if reconciled and self.scope != dict(scope):
|
|
# Replay must land the journal exactly on the adopted assignment (a pending
|
|
# publication moves it to the child); anything else is not this successor's outbox.
|
|
raise WorkerPersistenceError('outbox_scope_mismatch')
|
|
if is_worker_process() and self.binds_delegation_ledger:
|
|
from tools.async_delegation_worker import bind_worker_delegation_store
|
|
bind_worker_delegation_store(self)
|
|
except Exception:
|
|
self._outbox_owner.close()
|
|
raise
|
|
|
|
def _reconcile_adopted(self, adopted):
|
|
"""Carry a retained outbox across verified owner replacement.
|
|
|
|
``adopted`` is the separately authenticated ``worker.adopt`` result for the requested
|
|
scope; never self-adopt. Only the owner epoch (forward) and the physical target may
|
|
differ from the journal (a same-epoch restart can still find the target rotated).
|
|
Pending entries replay under the journal's own target with the adopted epoch, so an
|
|
exact lost-ACK retry resolves its original receipt instead of creating a second one;
|
|
the constructor then requires the replayed scope to equal the adopted one.
|
|
"""
|
|
old, new = self.journal['scope'], self.scope
|
|
fixed = set(new) - {'epoch', 'session_id'}
|
|
if (not isinstance(adopted, dict) or set(old) != set(new)
|
|
or any(old[key] != new[key] for key in fixed)
|
|
or adopted.get('execution_id') != new.get('execution_id')
|
|
or adopted.get('generation') != new.get('generation')
|
|
or adopted.get('session_id') != new.get('session_id')
|
|
or adopted.get('owner_epoch') != new.get('epoch') or adopted.get('status') != 'running'
|
|
or type(old.get('epoch')) is not int or type(new.get('epoch')) is not int
|
|
or old['epoch'] > new['epoch']
|
|
# Only a pending publication can still move the target; replay must prove it.
|
|
or (old['session_id'] != new['session_id'] and not any(
|
|
entry['operation'] == 'compression.publish' for entry in self.journal['pending']))):
|
|
raise WorkerPersistenceError('outbox_scope_mismatch')
|
|
candidate = dict(self.journal, scope=dict(old, epoch=new['epoch']))
|
|
self._save(candidate)
|
|
self.scope = candidate['scope']
|
|
self.journal = candidate
|
|
|
|
def _save(self, journal):
|
|
if self._outbox_owner.closed:
|
|
raise WorkerPersistenceError('outbox_closed')
|
|
encoded = json.dumps(journal, ensure_ascii=True, allow_nan=False, separators=(',', ':')).encode()
|
|
if len(encoded) > self.max_bytes:
|
|
self.failure = 'outbox_full'
|
|
raise WorkerPersistenceError(self.failure)
|
|
from utils import atomic_write_bytes
|
|
# Owner-only journal (the dir is 0700 too); the rename and its directory entry are durable.
|
|
atomic_write_bytes(self.path, encoded, tmp_prefix='.pending-', mode=0o600, fsync_dir=True)
|
|
|
|
def _session(self, session_id):
|
|
if session_id != self.scope['session_id']:
|
|
raise WorkerPersistenceError('permission_denied')
|
|
|
|
def _apply(self, operation, payload):
|
|
with self.lock:
|
|
if self.failure:
|
|
raise WorkerPersistenceError(self.failure)
|
|
if self.journal['pending']:
|
|
raise WorkerPersistenceError('pending_receipt')
|
|
entry = {'sequence': self.journal['next_sequence'], 'operation': operation, 'payload': payload}
|
|
candidate = json.loads(json.dumps(self.journal))
|
|
candidate['pending'].append(json.loads(json.dumps(entry, allow_nan=False)))
|
|
candidate['next_sequence'] += 1
|
|
try:
|
|
self._save(candidate)
|
|
except Exception as exc:
|
|
self.failure = str(exc)
|
|
raise
|
|
self.journal = candidate
|
|
return self.retry_pending()[0]
|
|
|
|
def retry_pending(self):
|
|
with self.lock:
|
|
results = []
|
|
try:
|
|
while self.journal['pending']:
|
|
entry = self.journal['pending'][0]
|
|
result = self.rpc('worker.persist', **self.scope, **entry)
|
|
candidate = dict(self.journal, pending=self.journal['pending'][1:])
|
|
candidate = self._compression_receipt_journal(candidate, result)
|
|
self._save(candidate)
|
|
self.scope = candidate['scope']
|
|
self.journal = candidate
|
|
results.append(result)
|
|
except Exception as exc:
|
|
self.failure = str(exc)
|
|
raise
|
|
self.failure = None
|
|
return results
|
|
|
|
def adopt(self, epoch):
|
|
"""Install only a separately verified worker.adopt result; never self-adopt."""
|
|
with self.lock:
|
|
candidate = dict(self.journal, scope=dict(self.scope, epoch=epoch))
|
|
self._save(candidate)
|
|
self.scope = candidate['scope']
|
|
self.journal = candidate
|
|
self.failure = None
|
|
|
|
def append_messages_batch(self, session_id, messages, compression_lock_holder=None,
|
|
turn_lease_holder=None, chunk_rows=None, turn_lease_ttl_seconds=300.0):
|
|
self._session(session_id)
|
|
if chunk_rows is not None:
|
|
raise WorkerPersistenceError('unsupported_operation')
|
|
if compression_lock_holder is not None or turn_lease_ttl_seconds != 300.0:
|
|
return self._append_compression_messages(session_id, messages, compression_lock_holder,
|
|
turn_lease_holder, turn_lease_ttl_seconds)
|
|
result = self._apply('transcript.append', {'messages': messages, 'turn_lease_holder': turn_lease_holder})
|
|
apply_row_annotations(messages, result['annotations'])
|
|
return result['count']
|
|
|
|
def try_acquire_session_turn_lease(self, session_id, holder, *, ttl_seconds=300.0, patience_s=None):
|
|
self._session(session_id)
|
|
return self._apply('turn.acquire', {'holder': holder, 'ttl_seconds': ttl_seconds})['value']
|
|
|
|
def acquire_session_turn_lease(self, session_id, holder, *, ttl_seconds=300.0,
|
|
wait_seconds=1800.0, poll_interval_seconds=1.0, on_wait=None,
|
|
wait_notice_interval_seconds=15.0, should_abort=None, acquire_patience_s=0.5,
|
|
on_contended=None):
|
|
# Reuse only the local polling orchestrator, not the SQLite mixin surface. Same signature
|
|
# as the owner's (``on_contended``: a busy-database notice; the RPC try never raises it).
|
|
from hermes_state_compression import SessionCompressionMixin
|
|
return SessionCompressionMixin.acquire_session_turn_lease(self, session_id, holder,
|
|
ttl_seconds=ttl_seconds, wait_seconds=wait_seconds,
|
|
poll_interval_seconds=poll_interval_seconds, on_wait=on_wait,
|
|
wait_notice_interval_seconds=wait_notice_interval_seconds,
|
|
should_abort=should_abort, acquire_patience_s=acquire_patience_s,
|
|
on_contended=on_contended)
|
|
|
|
def refresh_session_turn_lease(self, session_id, holder, *, ttl_seconds=300.0, patience_s=None):
|
|
self._session(session_id)
|
|
return self._apply('turn.renew', {'holder': holder, 'ttl_seconds': ttl_seconds})['value']
|
|
|
|
def release_session_turn_lease(self, session_id, holder):
|
|
# DurableTurnLease retains its admission target across physical rotation.
|
|
self._apply('turn.cleanup', {'target': session_id, 'holder': holder})
|
|
|
|
def queue_token_counts(self, session_id, **usage):
|
|
self._session(session_id)
|
|
self._apply('usage.main', usage)
|
|
|
|
def record_auxiliary_usage(self, session_id, task, **usage):
|
|
self._session(session_id)
|
|
self._apply('usage.auxiliary', dict(usage, task=task))
|
|
|
|
def flush_token_counts(self, timeout=5.0):
|
|
with self.lock:
|
|
if self.failure:
|
|
raise WorkerPersistenceError(self.failure)
|
|
if self.journal['pending']:
|
|
raise WorkerPersistenceError('pending_receipt')
|
|
return True
|
|
|
|
def get_session_title(self, session_id):
|
|
return self.get_session(session_id).get('title')
|
|
|
|
# Transcript-tail reads the agent core makes on its session_db (failed-turn boundary, in-place
|
|
# compaction coverage); a missing one raised AttributeError inside a best-effort guard.
|
|
def latest_conversation_role(self, session_id):
|
|
self._session(session_id)
|
|
return self._apply('session.tail_role', {})['value']
|
|
|
|
def get_message_role(self, session_id, row_id):
|
|
self._session(session_id)
|
|
return self._apply('session.row_role', {'row_id': int(row_id)})['value']
|
|
|
|
def get_compression_failure_cooldown_row(self, session_id):
|
|
row = self.get_session(session_id)
|
|
return {'session_exists': True, 'cooldown_until': row.get('compression_failure_cooldown_until'),
|
|
'error': row.get('compression_failure_error')}
|
|
|
|
def get_compression_failure_cooldown(self, session_id):
|
|
import time
|
|
now = time.time()
|
|
row = self.get_compression_failure_cooldown_row(session_id)
|
|
deadline = row['cooldown_until']
|
|
if deadline is None or float(deadline) <= now:
|
|
return None
|
|
return {'cooldown_until': float(deadline), 'remaining_seconds': float(deadline) - now,
|
|
'error': row['error']}
|
|
|
|
def _session_number(self, session_id, column, cast, zero):
|
|
row = self.get_session(session_id)
|
|
try:
|
|
return max(zero, cast(row.get(column) or zero))
|
|
except (TypeError, ValueError):
|
|
return zero
|
|
|
|
def get_compression_fallback_streak(self, session_id):
|
|
return self._session_number(session_id, 'compression_fallback_streak', int, 0)
|
|
|
|
def get_compression_ineffective_count(self, session_id):
|
|
return self._session_number(session_id, 'compression_ineffective_count', int, 0)
|
|
|
|
def get_compression_recovery_deadline(self, session_id):
|
|
return self._session_number(session_id, 'compression_recovery_deadline', float, 0.0)
|
|
|
|
def get_session_model_config_value(self, session_id, key, default=None):
|
|
from hermes_state_sessions import _parse_model_config
|
|
return _parse_model_config(self.get_session(session_id).get('model_config')).get(key, default)
|
|
|
|
def update_conversation_model_config_value(self, session_id, key, update):
|
|
# A worker owns exactly its session row, so the read-modify-write is that row's; the owner applies
|
|
# the patch in its own transaction (``None`` removes the key, as on SessionDB).
|
|
self._session(session_id)
|
|
self.patch_session_model_config(session_id, {key: update(self.get_session_model_config_value(session_id, key))})
|
|
return session_id
|
|
|
|
def session_ids_with_model_config_key(self, key):
|
|
# The scan a worker can answer: its own session (it may read no other row).
|
|
sid = self.scope['session_id']
|
|
return [sid] if self.get_session_model_config_value(sid, key) is not None else []
|
|
|
|
def update_system_prompt(self, session_id, system_prompt):
|
|
self._session(session_id)
|
|
self._apply('session.prompt', {'system_prompt': system_prompt})
|
|
|
|
def patch_session_model_config(self, session_id, patch):
|
|
self._session(session_id)
|
|
self._apply('session.sidecars', {'patch': patch})
|
|
|
|
def update_session_tool_names(self, session_id, pin):
|
|
# The tools[] freeze pin ({"version", "tools"[, "delegation_backend"]}; None clears): without it every
|
|
# fresh worker re-probes check_fns and a config flip between turns silently forks the cached prefix.
|
|
self._session(session_id)
|
|
self._apply('session.tools', {'tool_names': pin})
|
|
|
|
def finish(self):
|
|
return self._apply('execution.finish', {})
|
|
|
|
def close(self):
|
|
with self.lock:
|
|
try:
|
|
self.flush_token_counts()
|
|
finally:
|
|
self._outbox_owner.close()
|