mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
1f91fb386a
Review follow-up for #141944. - The whole child store hops off a running loop, not only child.reserve: the child is built, run and closed inside delegate_task's callback, and its constructor already reads the store (ContextCompressor .bind_session_state). Under NeMo Relay that tripped the sticky synchronous_rpc_on_event_loop refusal and every child came back status=error, api_calls=0. - A child that finishes itself (early close after a failed constructor or a refused sibling) ends its own session row; parent retirement and restart only select live children, so that row stayed open forever. - session.create on an owner-reserved child merges its constructor facts (max_iterations, reasoning_config, ...) beside the _delegate_from marker instead of losing them to the keep-existing upsert. - Descendant lookups use a primary-key range instead of substr(), so a no-child finish is one index probe, not a table scan. - Invariant tests: depth refusal, child profile_name == parent's, a forged parent claim refused on the request path, a draining refusal receipt, mid-batch refusal closes built siblings, an early store close settles the child, constructor facts persisted, and a real-daemon Relay leg.
58 lines
2.9 KiB
Python
58 lines
2.9 KiB
Python
"""A delegated child's store inside a worker: owner-reserved, never self-registered (#141793).
|
|
|
|
The parent's ``RuntimeSessionStore`` asks the owner for a child on its own receipt stream; the
|
|
owner mints the child's session row and execution (gateway/session_worker_child.py) and the
|
|
child persists through the ordinary ``worker.persist`` path under its own claim. A refusal
|
|
(capacity, depth, draining) is a ValueError so ``delegate_task`` reports it as a tool error.
|
|
"""
|
|
import secrets
|
|
from functools import partial
|
|
|
|
from agent.runtime_session_store import RuntimeSessionStore, call_off_event_loop
|
|
|
|
_REFUSALS = {
|
|
'delegation_capacity': 'the gateway owner refused another subagent: this execution already runs '
|
|
'delegation.max_concurrent_children of them',
|
|
'delegation_depth_exceeded': 'the gateway owner refused a subagent this deep (delegation.max_spawn_depth)',
|
|
}
|
|
|
|
|
|
class ChildSessionStore(RuntimeSessionStore):
|
|
binds_delegation_ledger = False
|
|
|
|
@property
|
|
def reserved_session_id(self):
|
|
return self.scope['session_id']
|
|
|
|
def create_session(self, session_id, source, **fields):
|
|
# The owner created this row at reservation with its source, lineage and profile; only
|
|
# the constructor facts are the child's to add.
|
|
from hermes_state_worker_lifecycle import _CONSTRUCTOR_CONFIG
|
|
config = {k: v for k, v in (fields.get('model_config') or {}).items() if k in _CONSTRUCTOR_CONFIG}
|
|
return super().create_session(session_id, 'subagent', model=fields.get('model'),
|
|
model_config=config or None, system_prompt=fields.get('system_prompt'))
|
|
|
|
def close(self):
|
|
# The child's agent closes once, after its last write: that is this execution's end.
|
|
try:
|
|
if not self._outbox_owner.closed and not self.failure:
|
|
self.finish()
|
|
finally:
|
|
super().close()
|
|
|
|
|
|
def reserve_child_store(parent):
|
|
"""A ``ChildSessionStore`` the owner reserved under ``parent``'s verified execution."""
|
|
secret = secrets.token_urlsafe(32)
|
|
# Reached from delegate_task's tool callback, which can run on Relay's loop (#139029).
|
|
result = call_off_event_loop(parent._apply, 'child.reserve', {'secret': secret})
|
|
if 'refused' in result:
|
|
raise ValueError(_REFUSALS.get(result['refused'], 'the gateway owner refused a subagent: ' + result['refused']))
|
|
child = result['child']
|
|
scope = dict(parent.scope, **child, secret=secret)
|
|
outbox = parent.path.parent / ('child-' + child['execution_id'].rsplit(':', 1)[1])
|
|
# Every child RPC hops off a running loop, not only this reservation: the child is built, run
|
|
# and closed inside delegate_task's callback, and its constructor already reads the store
|
|
# (ContextCompressor.bind_session_state), which WorkerRPC refuses on Relay's loop.
|
|
return ChildSessionStore(partial(call_off_event_loop, parent.rpc), scope, outbox)
|