mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-22 16:25:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(delegation): harden durable completion delivery

Browse source
This commit is contained in:
teknium1 2026-07-12 17:53:10 -07:00 • committed by Teknium
parent 67f4e1b4a9
commit d0e9a42cec
9 changed files with 696 additions and 153 deletions
Download patch file Download diff file Expand all files Collapse all files

20
cli.py
View file

@ -15189,10 +15189,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
from tools.approval import get_current_session_key from tools.approval import get_current_session_key
_drain_sk = get_current_session_key(default="") _drain_sk = get_current_session_key(default="")
for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk): for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk):
from tools.async_delegation import (
claim_event_delivery, complete_event_delivery,
)
_claim = claim_event_delivery(_evt, "cli-idle")
if _claim is None:
continue
self._pending_input.put(_synth) self._pending_input.put(_synth)
if _evt.get("type") == "async_delegation": complete_event_delivery(_evt, _claim)
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
except Exception: except Exception:
pass pass
continue continue
@ -15354,10 +15358,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
try: try:
from tools.process_registry import process_registry from tools.process_registry import process_registry
for _evt, _synth in process_registry.drain_notifications(): for _evt, _synth in process_registry.drain_notifications():
from tools.async_delegation import (
claim_event_delivery, complete_event_delivery,
)
_claim = claim_event_delivery(_evt, "cli-post-turn")
if _claim is None:
continue
self._pending_input.put(_synth) self._pending_input.put(_synth)
if _evt.get("type") == "async_delegation": complete_event_delivery(_evt, _claim)
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
except Exception: except Exception:
pass # Non-fatal — don't break the main loop pass # Non-fatal — don't break the main loop

424
gateway/run.py
View file

@ -1772,6 +1772,7 @@ from gateway.config import (
load_gateway_config, load_gateway_config,
) )
from gateway.session import ( from gateway.session import (
AsyncSessionStore,
SessionStore, SessionStore,
SessionSource, SessionSource,
SessionContext, SessionContext,
@ -2881,6 +2882,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
key, max_active_age=_bg_max_age_seconds, key, max_active_age=_bg_max_age_seconds,
), ),
) )
# One enforced loop-side boundary for the synchronous SessionStore.
# Sync helpers keep using ``session_store`` directly; async gateway
# handlers call this facade and await every operation.
self._async_session_store = AsyncSessionStore(self.session_store)
self.delivery_router = DeliveryRouter(self.config) self.delivery_router = DeliveryRouter(self.config)
self._running = False self._running = False
self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None
@ -2974,6 +2979,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# cannot grow unbounded over a long-running gateway lifetime. # cannot grow unbounded over a long-running gateway lifetime.
self._session_sources: "OrderedDict[str, SessionSource]" = OrderedDict() self._session_sources: "OrderedDict[str, SessionSource]" = OrderedDict()
self._session_sources_max = 512 self._session_sources_max = 512
# Completion delivery is intentionally lifecycle-scoped. This closes
# duplicate queue/watcher races inside one gateway without pretending
# the adapter call and a persistence write can be exactly-once across
# a process crash. Any durable async-delegation replay state remains
# owned by tools.async_delegation, not a parallel gateway ledger.
self._completion_delivery_lock = threading.Lock()
self._completion_deliveries_inflight: set[tuple[str, str, object]] = set()
self._completion_deliveries_delivered: "OrderedDict[tuple[str, str, object], None]" = OrderedDict()
self._completion_delivery_retention = 2048
# Cache AIAgent instances per session to preserve prompt caching. # Cache AIAgent instances per session to preserve prompt caching.
# Without this, a new AIAgent is created per message, rebuilding the # Without this, a new AIAgent is created per message, rebuilding the
@ -4815,7 +4829,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"""Load reasoning effort from config.yaml. """Load reasoning effort from config.yaml.
Reads agent.reasoning_effort from config.yaml. Valid: "none", Reads agent.reasoning_effort from config.yaml. Valid: "none",
"minimal", "low", "medium", "high", "xhigh". Returns None to use "minimal", "low", "medium", "high", "xhigh", "max", "ultra". Returns None to use
default (medium). default (medium).
""" """
from hermes_constants import parse_reasoning_effort from hermes_constants import parse_reasoning_effort
@ -5202,7 +5216,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception: except Exception:
return False return False
def _session_has_compression_in_flight(self, session_key: str) -> bool: async def _session_has_compression_in_flight(self, session_key: str) -> bool:
"""Return True when a compression lock is held for this session's id. """Return True when a compression lock is held for this session's id.
Context compression is interrupt-protected (#23975) but gateway Context compression is interrupt-protected (#23975) but gateway
@ -5210,28 +5224,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
the pre-rotation parent while compression is mid-flight, producing the pre-rotation parent while compression is mid-flight, producing
orphaned compression siblings (#56391). Callers demote interrupt to orphaned compression siblings (#56391). Callers demote interrupt to
queue when this returns True. queue when this returns True.
Both blocking sources — the ``session_store`` lock + JSON load, and the
SQLite ``get_compression_lock_holder`` SELECT — are offloaded to a
worker thread so a large state.db never freezes the event loop (#5).
""" """
session_store = getattr(self, "session_store", None) session_store = getattr(self, "session_store", None)
if not session_key or session_store is None: if not session_key or session_store is None:
return False return False
try: try:
with session_store._lock: # noqa: SLF001 — snapshot entry under lock session_id = await asyncio.to_thread(
session_store._ensure_loaded_locked() # noqa: SLF001 self._lookup_session_id_under_store_lock, session_store, session_key
entry = session_store._entries.get(session_key) # noqa: SLF001 )
session_id = getattr(entry, "session_id", None) if entry is not None else None
if not session_id:
return False
except Exception: except Exception:
return False return False
if not session_id:
return False
session_db = getattr(self, "_session_db", None) session_db = getattr(self, "_session_db", None)
if session_db is None: if session_db is None:
return False return False
db = getattr(session_db, "_db", session_db) raw_db = getattr(session_db, "_db", session_db)
try: try:
return bool(db.get_compression_lock_holder(str(session_id))) holder = await asyncio.to_thread(
raw_db.get_compression_lock_holder, str(session_id)
)
return bool(holder)
except Exception: except Exception:
return False return False
@staticmethod
def _lookup_session_id_under_store_lock(session_store, session_key: str):
"""Sync helper run in the thread pool: read session_id under the store lock."""
# noqa: SLF001 — intentional private access; runs off the event loop.
with session_store._lock: # noqa: SLF001
session_store._ensure_loaded_locked() # noqa: SLF001
entry = session_store._entries.get(session_key) # noqa: SLF001
return getattr(entry, "session_id", None) if entry is not None else None
# Hard cap on per-session pending follow-ups for busy_input_mode=queue # Hard cap on per-session pending follow-ups for busy_input_mode=queue
# (and the draining/steer-fallback/subagent-demotion paths that share # (and the draining/steer-fallback/subagent-demotion paths that share
# this entry point). Without a cap, a stuck agent + a rapid-fire user # this entry point). Without a cap, a stuck agent + a rapid-fire user
@ -5454,7 +5483,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
effective_mode = "queue" effective_mode = "queue"
demoted_for_compression = ( demoted_for_compression = (
effective_mode == "interrupt" effective_mode == "interrupt"
and self._session_has_compression_in_flight(session_key) and await self._session_has_compression_in_flight(session_key)
) )
if demoted_for_compression: if demoted_for_compression:
logger.info( logger.info(
@ -5743,7 +5772,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
source = None source = None
try: try:
if getattr(self, "session_store", None) is not None: if getattr(self, "session_store", None) is not None:
self.session_store._ensure_loaded() await self.async_session_store._ensure_loaded()
entry = self.session_store._entries.get(session_key) entry = self.session_store._entries.get(session_key)
source = getattr(entry, "origin", None) if entry else None source = getattr(entry, "origin", None) if entry else None
except Exception as e: except Exception as e:
@ -7013,7 +7042,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
pass pass
else: else:
try: try:
suspended = self.session_store.suspend_recently_active() suspended = await self.async_session_store.suspend_recently_active()
if suspended: if suspended:
logger.info("Marked %d in-flight session(s) as resumable from previous run", suspended) logger.info("Marked %d in-flight session(s) as resumable from previous run", suspended)
except Exception as e: except Exception as e:
@ -7574,13 +7603,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# Make sure there's an entry in the session_store for this key. If # Make sure there's an entry in the session_store for this key. If
# the home channel has never been used, get_or_create_session # the home channel has never been used, get_or_create_session
# creates one; switch_session then re-points it. # creates one; switch_session then re-points it.
self.session_store.get_or_create_session(dest_source) await self.async_session_store.get_or_create_session(dest_source)
# Re-bind the destination key to the CLI session_id. switch_session # Re-bind the destination key to the CLI session_id. switch_session
# ends the prior session in SQLite and reopens the CLI session under # ends the prior session in SQLite and reopens the CLI session under
# the new key. The CLI's transcript becomes the active one for the # the new key. The CLI's transcript becomes the active one for the
# gateway from this moment on. # gateway from this moment on.
switched = self.session_store.switch_session(session_key, cli_session_id) switched = await self.async_session_store.switch_session(session_key, cli_session_id)
if switched is None: if switched is None:
raise RuntimeError( raise RuntimeError(
f"could not switch session key {session_key} → {cli_session_id}" f"could not switch session key {session_key} → {cli_session_id}"
@ -7657,13 +7686,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_MAX_FINALIZE_RETRIES = 3 _MAX_FINALIZE_RETRIES = 3
while self._running: while self._running:
try: try:
self.session_store._ensure_loaded() await self.async_session_store._ensure_loaded()
# Collect expired sessions first, then log a single summary. # Collect expired sessions first, then log a single summary.
_expired_entries = [] _expired_entries = []
for key, entry in list(self.session_store._entries.items()): for key, entry in list(self.session_store._entries.items()):
if entry.expiry_finalized: if entry.expiry_finalized:
continue continue
if not self.session_store._is_session_expired(entry): if not await self.async_session_store._is_session_expired(entry):
continue continue
_expired_entries.append((key, entry)) _expired_entries.append((key, entry))
@ -7747,7 +7776,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# state.db (single write-path, #9006) — also drops # state.db (single write-path, #9006) — also drops
# the persisted /model override, since finalization # the persisted /model override, since finalization
# is a conversation boundary. # is a conversation boundary.
self.session_store.set_expiry_finalized(entry) await self.async_session_store.set_expiry_finalized(entry)
logger.debug( logger.debug(
"Session expiry finalized for %s", "Session expiry finalized for %s",
entry.session_id, entry.session_id,
@ -7762,7 +7791,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"Marking as finalized to prevent infinite retry loop.", "Marking as finalized to prevent infinite retry loop.",
failures, entry.session_id, e, failures, entry.session_id, e,
) )
self.session_store.set_expiry_finalized( await self.async_session_store.set_expiry_finalized(
entry, clear_model_override=False entry, clear_model_override=False
) )
_finalize_failures.pop(entry.session_id, None) _finalize_failures.pop(entry.session_id, None)
@ -7815,7 +7844,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
getattr(self.config, "session_store_max_age_days", 0) or 0 getattr(self.config, "session_store_max_age_days", 0) or 0
) )
if _max_age > 0: if _max_age > 0:
_pruned = self.session_store.prune_old_entries(_max_age) _pruned = await self.async_session_store.prune_old_entries(_max_age)
if _pruned: if _pruned:
logger.info( logger.info(
"SessionStore prune: dropped %d stale entries", "SessionStore prune: dropped %d stale entries",
@ -8149,7 +8178,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if _agent is _AGENT_PENDING_SENTINEL: if _agent is _AGENT_PENDING_SENTINEL:
continue continue
try: try:
self.session_store.mark_resume_pending( await self.async_session_store.mark_resume_pending(
_sk, _sk,
"restart_timeout" if self._restart_requested else "shutdown_timeout", "restart_timeout" if self._restart_requested else "shutdown_timeout",
) )
@ -8180,7 +8209,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
for _sk in _pre_drain_keys: for _sk in _pre_drain_keys:
if _sk not in self._running_agents: if _sk not in self._running_agents:
try: try:
self.session_store.clear_resume_pending(_sk) await self.async_session_store.clear_resume_pending(_sk)
except Exception as _e: except Exception as _e:
logger.debug( logger.debug(
"clear_resume_pending after drain failed for %s: %s", "clear_resume_pending after drain failed for %s: %s",
@ -8223,7 +8252,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if _agent is _AGENT_PENDING_SENTINEL: if _agent is _AGENT_PENDING_SENTINEL:
continue continue
try: try:
self.session_store.mark_resume_pending(_sk, _resume_reason) await self.async_session_store.mark_resume_pending(_sk, _resume_reason)
except Exception as _e: except Exception as _e:
logger.debug( logger.debug(
"mark_resume_pending failed for %s: %s", "mark_resume_pending failed for %s: %s",
@ -9596,7 +9625,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# the id out from under it, forking orphaned compression # the id out from under it, forking orphaned compression
# siblings. Demote to queue semantics so the follow-up waits # siblings. Demote to queue semantics so the follow-up waits
# for the in-flight compression + rotation to land. # for the in-flight compression + rotation to land.
if self._session_has_compression_in_flight(_quick_key): if await self._session_has_compression_in_flight(_quick_key):
logger.info( logger.info(
"PRIORITY interrupt demoted to queue for session %s " "PRIORITY interrupt demoted to queue for session %s "
"because context compression is in flight (#56391)", "because context compression is in flight (#56391)",
@ -9637,7 +9666,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if isinstance(quick_commands, dict) and command in quick_commands: if isinstance(quick_commands, dict) and command in quick_commands:
qcmd = quick_commands[command] qcmd = quick_commands[command]
if qcmd.get("type") == "alias": if qcmd.get("type") == "alias":
target = qcmd.get("target", "").strip() target = (qcmd.get("target") or "").strip()
if target: if target:
target = target if target.startswith("/") else f"/{target}" target = target if target.startswith("/") else f"/{target}"
target_command = target.lstrip("/") target_command = target.lstrip("/")
@ -10049,7 +10078,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
else: else:
return f"Quick command '/{command}' has no command defined." return f"Quick command '/{command}' has no command defined."
elif qcmd.get("type") == "alias": elif qcmd.get("type") == "alias":
target = qcmd.get("target", "").strip() target = (qcmd.get("target") or "").strip()
if target: if target:
target = target if target.startswith("/") else f"/{target}" target = target if target.startswith("/") else f"/{target}"
target_command = target.lstrip("/") target_command = target.lstrip("/")
@ -10303,7 +10332,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# on error. Let the user drive the next turn. # on error. Let the user drive the next turn.
if _final_text.strip(): if _final_text.strip():
try: try:
session_entry = self.session_store.get_or_create_session(source) session_entry = await self.async_session_store.get_or_create_session(source)
except Exception: except Exception:
session_entry = None session_entry = None
if session_entry is not None: if session_entry is not None:
@ -10619,8 +10648,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
from agent.model_metadata import get_model_context_length_async from agent.model_metadata import get_model_context_length_async
_msg_cwd = os.environ.get("TERMINAL_CWD", os.path.expanduser("~")) _msg_cwd = os.environ.get("TERMINAL_CWD", os.path.expanduser("~"))
_msg_runtime = _resolve_runtime_agent_kwargs()
_msg_config_ctx = None _msg_config_ctx = None
_msg_cfg = None
_msg_model_cfg = {}
_msg_custom_providers = []
try: try:
_msg_cfg = _load_gateway_config() _msg_cfg = _load_gateway_config()
_msg_model_cfg = _msg_cfg.get("model", {}) _msg_model_cfg = _msg_cfg.get("model", {})
@ -10628,13 +10659,57 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_msg_raw_ctx = _msg_model_cfg.get("context_length") _msg_raw_ctx = _msg_model_cfg.get("context_length")
if _msg_raw_ctx is not None: if _msg_raw_ctx is not None:
_msg_config_ctx = int(_msg_raw_ctx) _msg_config_ctx = int(_msg_raw_ctx)
try:
from hermes_cli.config import get_compatible_custom_providers
_msg_custom_providers = get_compatible_custom_providers(_msg_cfg)
except Exception:
_msg_custom_providers = _msg_cfg.get("custom_providers") or []
except Exception: except Exception:
pass pass
# Resolve the session's actual model/provider/base_url the
# same way the hygiene compression block does (~11080).
# GatewayRunner has no self._model/self._base_url attrs
# (that was copy-pasted from HermesCLI, which does carry
# self.model/self.base_url), so using them here always raised
# AttributeError, silently caught below, meaning this feature
# never ran.
_msg_model, _msg_runtime = self._resolve_session_agent_runtime(
source=source,
session_key=session_key,
user_config=_msg_cfg,
)
_msg_base_url = _msg_runtime.get("base_url") or ""
# A global model.context_length belongs to the configured
# model, not a session /model or channel override. Prefer a
# matching per-custom-provider model limit when available.
_msg_configured_model = (
_msg_model_cfg.get("default") or _msg_model_cfg.get("model")
if isinstance(_msg_model_cfg, dict)
else _msg_model_cfg
)
if _msg_model != _msg_configured_model:
_msg_config_ctx = None
if _msg_custom_providers and _msg_base_url:
try:
from hermes_cli.config import get_custom_provider_context_length
_msg_custom_ctx = get_custom_provider_context_length(
model=_msg_model,
base_url=_msg_base_url,
custom_providers=_msg_custom_providers,
)
if _msg_custom_ctx:
_msg_config_ctx = _msg_custom_ctx
except Exception:
pass
_msg_ctx_len = await get_model_context_length_async( _msg_ctx_len = await get_model_context_length_async(
self._model, _msg_model,
base_url=self._base_url or _msg_runtime.get("base_url") or "", base_url=_msg_base_url,
api_key=_msg_runtime.get("api_key") or "", api_key=_msg_runtime.get("api_key") or "",
config_context_length=_msg_config_ctx, config_context_length=_msg_config_ctx,
provider=_msg_runtime.get("provider") or "",
custom_providers=_msg_custom_providers,
) )
_ctx_result = await preprocess_context_references_async( _ctx_result = await preprocess_context_references_async(
message_text, message_text,
@ -10653,10 +10728,35 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if _ctx_result.expanded: if _ctx_result.expanded:
message_text = _ctx_result.message message_text = _ctx_result.message
except Exception as exc: except Exception as exc:
logger.debug("@ context reference expansion failed: %s", exc) logger.warning("@ context reference expansion failed: %s", exc)
logger.debug("@ context reference expansion failure detail", exc_info=True)
return message_text return message_text
async def _prepare_profile_scoped_inbound_message_text(
self,
*,
event: MessageEvent,
source: SessionSource,
history: List[Dict[str, Any]],
session_key: Optional[str] = None,
) -> Optional[str]:
"""Run inbound preprocessing under the routed profile when multiplexed."""
if getattr(getattr(self, "config", None), "multiplex_profiles", False):
with _profile_runtime_scope(self._resolve_profile_home_for_source(source)):
return await self._prepare_inbound_message_text(
event=event,
source=source,
history=history,
session_key=session_key,
)
return await self._prepare_inbound_message_text(
event=event,
source=source,
history=history,
session_key=session_key,
)
def _consume_pending_native_image_paths(self, session_key: str) -> List[str]: def _consume_pending_native_image_paths(self, session_key: str) -> List[str]:
pending_native = getattr(self, "_pending_native_image_paths_by_session", None) pending_native = getattr(self, "_pending_native_image_paths_by_session", None)
if not pending_native: if not pending_native:
@ -10684,6 +10784,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception: except Exception:
pass pass
@property
def async_session_store(self) -> AsyncSessionStore:
"""Return the single async facade for this runner's SessionStore."""
facade = getattr(self, "_async_session_store", None)
if facade is None or facade._store is not self.session_store:
facade = AsyncSessionStore(self.session_store)
self._async_session_store = facade
return facade
def _get_cached_session_source(self, session_key: str): def _get_cached_session_source(self, session_key: str):
if not session_key: if not session_key:
return None return None
@ -10727,7 +10836,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception: except Exception:
pass pass
session_entry = self.session_store.get_or_create_session(source) session_entry = await self.async_session_store.get_or_create_session(source)
session_key = session_entry.session_key session_key = session_entry.session_key
pinned_session_id = str( pinned_session_id = str(
(getattr(event, "metadata", None) or {}).get("gateway_session_id") or "" (getattr(event, "metadata", None) or {}).get("gateway_session_id") or ""
@ -10759,7 +10868,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
) )
return return
prior_session_id = session_entry.session_id prior_session_id = session_entry.session_id
switched = self.session_store.switch_session(session_key, pinned_session_id) switched = await self.async_session_store.switch_session(session_key, pinned_session_id)
if switched is not None: if switched is not None:
session_entry = switched session_entry = switched
logger.info( logger.info(
@ -10809,7 +10918,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# lane session is ended cleanly. Mutating session_entry in # lane session is ended cleanly. Mutating session_entry in
# place here created a split-brain state where the JSON # place here created a split-brain state where the JSON
# index pointed at one id but code downstream used another. # index pointed at one id but code downstream used another.
switched = self.session_store.switch_session(session_key, bound_session_id) switched = await self.async_session_store.switch_session(session_key, bound_session_id)
if switched is not None: if switched is not None:
session_entry = switched session_entry = switched
# If the stored binding pointed at a parent, rewrite it to the # If the stored binding pointed at a parent, rewrite it to the
@ -10997,7 +11106,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
logger.warning("[Gateway] Failed to auto-load skill(s) %s: %s", _skill_names, e) logger.warning("[Gateway] Failed to auto-load skill(s) %s: %s", _skill_names, e)
# Load conversation history from transcript # Load conversation history from transcript
history = self.session_store.load_transcript(session_entry.session_id) history = await self.async_session_store.load_transcript(session_entry.session_id)
# ----------------------------------------------------------------- # -----------------------------------------------------------------
# Session hygiene: auto-compress pathologically large transcripts # Session hygiene: auto-compress pathologically large transcripts
@ -11260,7 +11369,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
) )
if _hyg_rotated: if _hyg_rotated:
session_entry.session_id = _hyg_new_sid session_entry.session_id = _hyg_new_sid
self.session_store._save() await self.async_session_store._save()
await asyncio.to_thread( await asyncio.to_thread(
self._sync_telegram_topic_binding, self._sync_telegram_topic_binding,
source, session_entry, source, session_entry,
@ -11288,7 +11397,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# messages and replace them with only the compressed # messages and replace them with only the compressed
# summary (permanent data loss, #21301). # summary (permanent data loss, #21301).
if _hyg_rotated: if _hyg_rotated:
self.session_store.rewrite_transcript( await self.async_session_store.rewrite_transcript(
session_entry.session_id, _compressed session_entry.session_id, _compressed
) )
# Reset stored token count — transcript rewritten # Reset stored token count — transcript rewritten
@ -11405,7 +11514,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
) )
# First-message onboarding -- only on the very first interaction ever # First-message onboarding -- only on the very first interaction ever
if not history and not self.session_store.has_any_sessions(): if not history and not await self.async_session_store.has_any_sessions():
# Default first-contact note: a brief self-introduction. # Default first-contact note: a brief self-introduction.
_intro_note = ( _intro_note = (
"\n\n[System note: This is the user's very first message ever. " "\n\n[System note: This is the user's very first message ever. "
@ -11490,7 +11599,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# attachments (documents, audio, etc.) are not sent to the vision # attachments (documents, audio, etc.) are not sent to the vision
# tool even when they appear in the same message. # tool even when they appear in the same message.
# ----------------------------------------------------------------- # -----------------------------------------------------------------
message_text = await self._prepare_inbound_message_text( message_text = await self._prepare_profile_scoped_inbound_message_text(
event=event, event=event,
source=source, source=source,
history=history, history=history,
@ -11649,7 +11758,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if session_key and _should_clear_resume_pending_after_turn(agent_result): if session_key and _should_clear_resume_pending_after_turn(agent_result):
self._clear_restart_failure_count(session_key) self._clear_restart_failure_count(session_key)
try: try:
self.session_store.clear_resume_pending(session_key) await self.async_session_store.clear_resume_pending(session_key)
except Exception as _e: except Exception as _e:
logger.debug( logger.debug(
"clear_resume_pending failed for %s: %s", "clear_resume_pending failed for %s: %s",
@ -11671,8 +11780,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if agent_result.get("session_id") and agent_result["session_id"] != session_entry.session_id: if agent_result.get("session_id") and agent_result["session_id"] != session_entry.session_id:
if session_entry.session_id == _run_start_session_id: if session_entry.session_id == _run_start_session_id:
session_entry.session_id = agent_result["session_id"] session_entry.session_id = agent_result["session_id"]
self.session_store._save() await self.async_session_store._save()
self.session_store._record_gateway_session_peer( await self.async_session_store._record_gateway_session_peer(
session_entry.session_id, session_entry.session_id,
session_key, session_key,
source, source,
@ -11870,7 +11979,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"Auto-resetting session %s after compression exhaustion.", "Auto-resetting session %s after compression exhaustion.",
session_entry.session_id, session_entry.session_id,
) )
new_entry = self.session_store.reset_session(session_key) new_entry = await self.async_session_store.reset_session(session_key)
self._evict_cached_agent(session_key) self._evict_cached_agent(session_key)
self._session_model_overrides.pop(session_key, None) self._session_model_overrides.pop(session_key, None)
self._set_session_reasoning_override(session_key, None) self._set_session_reasoning_override(session_key, None)
@ -11914,7 +12023,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
pass # Skip all transcript writes — don't grow a broken session pass # Skip all transcript writes — don't grow a broken session
elif not history: elif not history:
tool_defs = agent_result.get("tools", []) tool_defs = agent_result.get("tools", [])
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, session_entry.session_id,
{ {
"role": "session_meta", "role": "session_meta",
@ -11968,7 +12077,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# after transient failures). #47237 # after transient failures). #47237
_skip_persist = ( _skip_persist = (
event.message_id event.message_id
and self.session_store.has_platform_message_id( and await self.async_session_store.has_platform_message_id(
session_entry.session_id, str(event.message_id) session_entry.session_id, str(event.message_id)
) )
) )
@ -11979,7 +12088,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
event.message_id, session_entry.session_id, event.message_id, session_entry.session_id,
) )
else: else:
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, session_entry.session_id,
_user_entry, _user_entry,
skip_db=agent_persisted, skip_db=agent_persisted,
@ -12005,13 +12114,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
} }
if event.message_id: if event.message_id:
_user_entry["message_id"] = str(event.message_id) _user_entry["message_id"] = str(event.message_id)
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, session_entry.session_id,
_user_entry, _user_entry,
skip_db=agent_persisted, skip_db=agent_persisted,
) )
if response: if response:
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, session_entry.session_id,
{"role": "assistant", "content": response, "timestamp": ts}, {"role": "assistant", "content": response, "timestamp": ts},
skip_db=agent_persisted, skip_db=agent_persisted,
@ -12036,7 +12145,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
): ):
entry["message_id"] = str(event.message_id) entry["message_id"] = str(event.message_id)
_user_msg_id_attached = True _user_msg_id_attached = True
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, entry, session_entry.session_id, entry,
skip_db=agent_persisted, skip_db=agent_persisted,
) )
@ -12044,7 +12153,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# Token counts and model are now persisted by the agent directly. # Token counts and model are now persisted by the agent directly.
# Keep only last_prompt_tokens here for context-window tracking and # Keep only last_prompt_tokens here for context-window tracking and
# compression decisions. # compression decisions.
self.session_store.update_session( await self.async_session_store.update_session(
session_entry.session_key, session_entry.session_key,
last_prompt_tokens=agent_result.get("last_prompt_tokens", 0), last_prompt_tokens=agent_result.get("last_prompt_tokens", 0),
) )
@ -12144,7 +12253,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if 'message_text' in locals() and message_text is not None and session_entry is not None: if 'message_text' in locals() and message_text is not None and session_entry is not None:
_already_persisted = False _already_persisted = False
try: try:
_recent_transcript = self.session_store.load_transcript(session_entry.session_id) _recent_transcript = await self.async_session_store.load_transcript(session_entry.session_id)
except Exception: except Exception:
_recent_transcript = [] _recent_transcript = []
for _msg in reversed(_recent_transcript[-10:]): for _msg in reversed(_recent_transcript[-10:]):
@ -12172,7 +12281,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
} }
if getattr(event, "message_id", None): if getattr(event, "message_id", None):
_user_entry["message_id"] = str(event.message_id) _user_entry["message_id"] = str(event.message_id)
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, session_entry.session_id,
_user_entry, _user_entry,
) )
@ -12627,7 +12736,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception: except Exception:
return 20 return 20
def _get_goal_manager_for_event(self, event: "MessageEvent"): async def _get_goal_manager_for_event(self, event: "MessageEvent"):
"""Return a GoalManager bound to the session for this gateway event. """Return a GoalManager bound to the session for this gateway event.
Returns ``(manager, session_entry)`` or ``(None, None)`` if the Returns ``(manager, session_entry)`` or ``(None, None)`` if the
@ -12639,7 +12748,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
logger.debug("goal manager unavailable: %s", exc) logger.debug("goal manager unavailable: %s", exc)
return None, None return None, None
try: try:
session_entry = self.session_store.get_or_create_session(event.source) session_entry = await self.async_session_store.get_or_create_session(event.source)
except Exception as exc: except Exception as exc:
logger.debug("goal manager: session lookup failed: %s", exc) logger.debug("goal manager: session lookup failed: %s", exc)
return None, None return None, None
@ -14082,8 +14191,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"content": f"[IMPORTANT: MCP servers have been reloaded. {change_detail}{tool_summary}. The tool list for this conversation has been updated accordingly.]", "content": f"[IMPORTANT: MCP servers have been reloaded. {change_detail}{tool_summary}. The tool list for this conversation has been updated accordingly.]",
} }
try: try:
session_entry = self.session_store.get_or_create_session(event.source) session_entry = await self.async_session_store.get_or_create_session(event.source)
self.session_store.append_to_transcript( await self.async_session_store.append_to_transcript(
session_entry.session_id, reload_msg session_entry.session_id, reload_msg
) )
except Exception: except Exception:
@ -15354,11 +15463,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
user_name=str(evt.get("user_name") or "").strip() or None, user_name=str(evt.get("user_name") or "").strip() or None,
) )
async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None: async def _inject_watch_notification(
"""Inject a watch-pattern notification as a synthetic message event. self, synth_text: str, evt: dict,
) -> Optional[bool]:
"""Inject a watch/completion notification as a synthetic message event.
Routing must come from the queued watch event itself, not from whatever Routing must come from the queued event itself, not from whatever
foreground message happened to be active when the queue was drained. foreground message happened to be active when the queue was drained.
Returns ``True`` after adapter acceptance, ``False`` after a retryable
adapter failure, and ``None`` when the event has no gateway route. This
is not a transactional boundary: a process crash after adapter
acceptance can still cause durable at-least-once replay.
""" """
source = self._build_process_event_source(evt) source = self._build_process_event_source(evt)
if not source: if not source:
@ -15366,7 +15481,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"Dropping watch notification with no routing metadata for process %s", "Dropping watch notification with no routing metadata for process %s",
evt.get("session_id", "unknown"), evt.get("session_id", "unknown"),
) )
return return None
platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform) platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform)
adapter = None adapter = None
for p, a in self.adapters.items(): for p, a in self.adapters.items():
@ -15374,7 +15489,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
adapter = a adapter = a
break break
if not adapter: if not adapter:
return return None
try: try:
metadata = {} metadata = {}
parent_session_id = str(evt.get("parent_session_id") or "").strip() parent_session_id = str(evt.get("parent_session_id") or "").strip()
@ -15395,8 +15510,118 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
source.thread_id, source.thread_id,
) )
await adapter.handle_message(synth_event) await adapter.handle_message(synth_event)
return True
except Exception as e: except Exception as e:
logger.error("Watch notification injection error: %s", e) logger.error("Watch notification injection error: %s", e)
return False
@staticmethod
def _completion_delivery_identity(evt: dict) -> Optional[tuple[str, str, object]]:
"""Return a producer-stable identity when one is available.
Delegation UUIDs identify one producer completion. Process session IDs
are normally unique too, but include the persisted spawn epoch so an
explicitly reused ID represents a distinct process incarnation. Legacy
process events without ``started_at`` are delivered without deduplication
rather than risking suppression of a real completion.
"""
evt_type = str(evt.get("type") or "")
if evt_type == "async_delegation":
producer_id = str(evt.get("delegation_id") or "")
return (evt_type, producer_id, "") if producer_id else None
if evt_type == "completion":
producer_id = str(evt.get("session_id") or "")
started_at = evt.get("started_at")
if producer_id and started_at is not None:
return (evt_type, producer_id, started_at)
return None
async def _deliver_completion_notification(
self, synth_text: str, evt: dict,
) -> Optional[bool]:
"""Deliver once per live gateway, or return False for a retry.
``True`` means this caller reached adapter acceptance, ``False`` means
injection failed and the claim was released for retry, and ``None``
means either another same-lifecycle caller owns/delivered the producer
event or the event has no gateway route. No cross-process exactly-once
guarantee is claimed.
"""
identity = self._completion_delivery_identity(evt)
durable_claim_id = ""
durable_delegation_id = ""
if evt.get("type") == "async_delegation":
durable_delegation_id = str(evt.get("delegation_id") or "")
if durable_delegation_id:
try:
from tools.async_delegation import claim_completion_delivery
durable_claim_id = f"gateway:{id(self)}:{__import__('uuid').uuid4().hex}"
if not claim_completion_delivery(
durable_delegation_id, durable_claim_id,
):
return None
except Exception as exc:
logger.warning(
"Could not claim durable async completion %s: %s",
durable_delegation_id, exc,
)
return False
if identity is not None:
with self._completion_delivery_lock:
if (
identity in self._completion_deliveries_inflight
or identity in self._completion_deliveries_delivered
):
return None
self._completion_deliveries_inflight.add(identity)
accepted = False
try:
injection_result = await self._inject_watch_notification(synth_text, evt)
if injection_result is not True:
return injection_result
accepted = True
if identity is not None:
with self._completion_delivery_lock:
self._completion_deliveries_inflight.discard(identity)
self._completion_deliveries_delivered[identity] = None
while (
len(self._completion_deliveries_delivered)
> self._completion_delivery_retention
):
self._completion_deliveries_delivered.popitem(last=False)
# If the durable async-delegation producer branch is present, its
# SQLite row remains the authoritative replay state. Acknowledge it
# after adapter acceptance; this gateway keeps no parallel ledger.
if durable_claim_id:
try:
from tools.async_delegation import complete_completion_delivery
complete_completion_delivery(
durable_delegation_id, durable_claim_id,
)
except Exception as exc:
logger.warning(
"Could not acknowledge durable async completion %s: %s",
durable_delegation_id, exc,
)
return True
finally:
if identity is not None and not accepted:
with self._completion_delivery_lock:
self._completion_deliveries_inflight.discard(identity)
if durable_claim_id and not accepted:
try:
from tools.async_delegation import release_completion_delivery
release_completion_delivery(
durable_delegation_id, durable_claim_id,
)
except Exception:
logger.debug("Could not release durable completion claim", exc_info=True)
def _enrich_async_delegation_routing(self, evt: dict) -> None: def _enrich_async_delegation_routing(self, evt: dict) -> None:
"""Fill platform/chat_id/thread_id/chat_type on an async-delegation event. """Fill platform/chat_id/thread_id/chat_type on an async-delegation event.
@ -15459,10 +15684,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if not synth_text: if not synth_text:
continue continue
try: try:
await self._inject_watch_notification(synth_text, evt) delivered = await self._deliver_completion_notification(synth_text, evt)
from tools.async_delegation import mark_completion_delivered if delivered is False:
mark_completion_delivered(str(evt.get("delegation_id") or "")) _pr.completion_queue.put(evt)
except Exception as e: except Exception as e:
_pr.completion_queue.put(evt)
logger.error("Async delegation injection error: %s", e) logger.error("Async delegation injection error: %s", e)
except Exception as e: except Exception as e:
logger.debug("Async delegation watcher error: %s", e) logger.debug("Async delegation watcher error: %s", e)
@ -15528,8 +15754,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# (#10156) — a status check must not suppress this delivery turn. # (#10156) — a status check must not suppress this delivery turn.
from tools.process_registry import format_process_notification, process_registry as _pr_check from tools.process_registry import format_process_notification, process_registry as _pr_check
if agent_notify and not _pr_check.is_completion_consumed(session_id): if agent_notify and not _pr_check.is_completion_consumed(session_id):
from agent.redact import redact_terminal_output
from tools.ansi_strip import strip_ansi from tools.ansi_strip import strip_ansi
_command = getattr(session, "command", "") or ""
_raw = strip_ansi(session.output_buffer) if session.output_buffer else "" _raw = strip_ansi(session.output_buffer) if session.output_buffer else ""
_raw = redact_terminal_output(_raw, _command)
_command = _redact_gateway_user_facing_secrets(_command)
# Truncate at line boundaries so notifications never start # Truncate at line boundaries so notifications never start
# mid-line (fixes #23284). Keep the last ~2000 chars but # mid-line (fixes #23284). Keep the last ~2000 chars but
# snap to the nearest preceding newline, then prepend a # snap to the nearest preceding newline, then prepend a
@ -15542,57 +15772,34 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_out = f"[… output truncated — showing last {len(_tail)} chars]\n{_tail}" _out = f"[… output truncated — showing last {len(_tail)} chars]\n{_tail}"
else: else:
_out = _raw _out = _raw
synth_text = format_process_notification({ completion_evt = {
"type": "completion", "type": "completion",
"session_id": session_id,
"command": session.command,
"exit_code": session.exit_code,
"completion_reason": getattr(session, "completion_reason", "exited"),
"termination_source": getattr(session, "termination_source", ""),
"output": _out,
})
if not synth_text:
break
source = self._build_process_event_source({
"session_id": session_id, "session_id": session_id,
"session_key": session_key, "session_key": session_key,
"platform": platform_name, "platform": platform_name,
"chat_type": watcher.get("chat_type", ""),
"chat_id": chat_id, "chat_id": chat_id,
"thread_id": thread_id, "thread_id": thread_id,
"user_id": user_id, "user_id": user_id,
"user_name": user_name, "user_name": user_name,
}) "message_id": message_id,
if not source: "started_at": getattr(session, "started_at", None),
logger.warning( "command": _command,
"Dropping completion notification with no routing metadata for process %s", "exit_code": session.exit_code,
session_id, "completion_reason": getattr(session, "completion_reason", "exited"),
) "termination_source": getattr(session, "termination_source", ""),
"output": _out,
}
synth_text = format_process_notification(completion_evt)
if not synth_text:
break break
delivered = await self._deliver_completion_notification(
adapter = None synth_text, completion_evt,
for p, a in self.adapters.items(): )
if p == source.platform: if delivered is False:
adapter = a # The process remains terminal; retry after failed
break # adapter injection instead of suppressing the result.
if adapter and source.chat_id: continue
try:
synth_event = MessageEvent(
text=synth_text,
message_type=MessageType.TEXT,
source=source,
internal=True,
message_id=message_id,
)
logger.info(
"Process %s finished — injecting agent notification for session %s chat=%s thread=%s",
session_id,
session_key,
source.chat_id,
source.thread_id,
)
await adapter.handle_message(synth_event)
except Exception as e:
logger.error("Agent notify injection error: %s", e)
break break
# --- Normal text-only notification --- # --- Normal text-only notification ---
@ -16553,7 +16760,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if url: if url:
return url.rstrip("/") return url.rstrip("/")
cfg = _load_gateway_config() cfg = _load_gateway_config()
url = (cfg.get("gateway") or {}).get("proxy_url", "").strip() url = (cfg.get("gateway") or {}).get("proxy_url")
url = (url or "").strip()
if url: if url:
return url.rstrip("/") return url.rstrip("/")
return None return None
@ -19840,7 +20048,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
session_key or "?", session_key or "?",
exc_info=True, exc_info=True,
) )
next_message = await self._prepare_inbound_message_text( next_message = await self._prepare_profile_scoped_inbound_message_text(
event=pending_event, event=pending_event,
source=next_source, source=next_source,
history=updated_history, history=updated_history,

25
hermes_state.py
View file

@ -140,7 +140,7 @@ T = TypeVar("T")
DEFAULT_DB_PATH = get_hermes_home() / "state.db" DEFAULT_DB_PATH = get_hermes_home() / "state.db"
SCHEMA_VERSION = 20 SCHEMA_VERSION = 21
# Cap on user-controlled FTS5 query input before regex/sanitizer processing. # Cap on user-controlled FTS5 query input before regex/sanitizer processing.
# Search queries do not need to be arbitrarily large, and bounding them keeps # Search queries do not need to be arbitrarily large, and bounding them keeps
@ -827,6 +827,27 @@ CREATE TABLE IF NOT EXISTS compression_locks (
expires_at REAL NOT NULL expires_at REAL NOT NULL
); );
CREATE TABLE IF NOT EXISTS async_delegations (
delegation_id TEXT PRIMARY KEY,
origin_session TEXT NOT NULL,
origin_ui_session_id TEXT NOT NULL DEFAULT '',
parent_session_id TEXT,
state TEXT NOT NULL,
dispatched_at REAL NOT NULL,
completed_at REAL,
updated_at REAL NOT NULL,
event_json TEXT,
result_json TEXT,
delivery_state TEXT NOT NULL DEFAULT 'pending',
delivery_attempts INTEGER NOT NULL DEFAULT 0,
delivered_at REAL,
owner_pid INTEGER,
owner_started_at INTEGER,
task_json TEXT,
delivery_claim TEXT,
delivery_claimed_at REAL
);
CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source); CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
CREATE INDEX IF NOT EXISTS idx_sessions_source_id ON sessions(source, id); CREATE INDEX IF NOT EXISTS idx_sessions_source_id ON sessions(source, id);
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id); CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
@ -835,6 +856,8 @@ CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestam
CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at); CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at);
CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id); CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id);
CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model); CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model);
CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery
ON async_delegations(delivery_state, completed_at);
""" """
# Indexes that reference columns added in later schema versions must be # Indexes that reference columns added in later schema versions must be

4
tests/gateway/test_completion_delivery.py
View file

@ -175,8 +175,8 @@ def test_failed_async_injection_is_retried_and_only_success_is_acked(
acknowledgements = [] acknowledgements = []
monkeypatch.setattr( monkeypatch.setattr(
async_delegation, async_delegation,
"mark_completion_delivered", "complete_completion_delivery",
lambda delegation_id: acknowledgements.append(delegation_id) or True, lambda delegation_id, _claim_id: acknowledgements.append(delegation_id) or True,
raising=False, raising=False,
) )

99
tests/tools/test_async_delegation.py
View file

@ -244,7 +244,8 @@ def test_completion_is_persisted_and_delivery_can_be_acknowledged(tmp_path, monk
assert row["state"] == "completed" assert row["state"] == "completed"
assert row["result"]["summary"] == "survived" assert row["result"]["summary"] == "survived"
assert row["delivery_state"] == "pending" assert row["delivery_state"] == "pending"
assert row["delivery_attempts"] >= 2 # Queue publication/restoration is not a destination delivery attempt.
assert row["delivery_attempts"] == 0
assert ad.mark_completion_delivered(dispatched["delegation_id"]) assert ad.mark_completion_delivered(dispatched["delegation_id"])
assert ad.restore_undelivered_completions(queue.Queue()) == 0 assert ad.restore_undelivered_completions(queue.Queue()) == 0
@ -305,6 +306,102 @@ assert ad.mark_completion_delivered({delegation_id!r})
assert probe.stdout.strip().splitlines()[-1] == "0" assert probe.stdout.strip().splitlines()[-1] == "0"
def test_submit_failure_removes_durable_running_record(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
class _BrokenExecutor:
def submit(self, *_args, **_kwargs):
raise RuntimeError("submit failed")
monkeypatch.setattr(ad, "_get_executor", lambda _max_workers: _BrokenExecutor())
result = ad.dispatch_async_delegation(
goal="never ran", context=None, toolsets=None, role="leaf", model="m",
session_key="owner", runner=lambda: {},
)
assert result["status"] == "rejected"
with ad._DB_LOCK, ad._connect() as conn:
assert conn.execute("SELECT COUNT(*) FROM async_delegations").fetchone()[0] == 0
def test_pending_retention_prunes_delivered_before_undelivered(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setattr(ad, "_MAX_RETAINED_COMPLETED", 2)
for index, delivery_state in enumerate(("pending", "delivered", "pending")):
delegation_id = f"deleg_{index}"
record = {
"delegation_id": delegation_id,
"session_key": "owner",
"origin_ui_session_id": "",
"parent_session_id": None,
"dispatched_at": float(index + 1),
}
ad._persist_dispatch(record)
ad._persist_completion(
{
"delegation_id": delegation_id,
"status": "completed",
"completed_at": float(index + 1),
},
{"status": "completed", "summary": delegation_id},
)
if delivery_state == "delivered":
ad.mark_completion_delivered(delegation_id)
ad._prune_durable_records()
assert ad.get_durable_delegation("deleg_0") is not None
assert ad.get_durable_delegation("deleg_1") is None
assert ad.get_durable_delegation("deleg_2") is not None
def test_recover_marks_abandoned_running_record_unknown(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
record = {
"delegation_id": "deleg_abandoned",
"session_key": "owner",
"origin_ui_session_id": "",
"parent_session_id": None,
"dispatched_at": 1.0,
}
ad._persist_dispatch(record)
with ad._DB_LOCK, ad._connect() as conn:
conn.execute(
"UPDATE async_delegations SET owner_pid=?, owner_started_at=NULL WHERE delegation_id=?",
(99999999, "deleg_abandoned"),
)
assert ad.recover_abandoned_delegations() == 1
durable = ad.get_durable_delegation("deleg_abandoned")
assert durable["state"] == "unknown"
assert durable["delivery_state"] == "pending"
restored = queue.Queue()
assert ad.restore_undelivered_completions(restored) == 1
assert restored.get_nowait()["status"] == "unknown"
def test_durable_delivery_claim_is_exclusive_and_retryable(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
record = {
"delegation_id": "deleg_claim", "session_key": "owner",
"origin_ui_session_id": "", "parent_session_id": None,
"dispatched_at": 1.0,
}
ad._persist_dispatch(record)
ad._persist_completion(
{"delegation_id": "deleg_claim", "status": "completed", "completed_at": 2.0},
{"status": "completed", "summary": "done"},
)
assert ad.claim_completion_delivery("deleg_claim", "consumer-a")
assert not ad.claim_completion_delivery("deleg_claim", "consumer-b")
assert ad.release_completion_delivery("deleg_claim", "consumer-a")
assert ad.claim_completion_delivery("deleg_claim", "consumer-b")
assert ad.complete_completion_delivery("deleg_claim", "consumer-b")
assert not ad.claim_completion_delivery("deleg_claim", "consumer-c")
assert ad.get_durable_delegation("deleg_claim")["delivery_state"] == "delivered"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Integration: delegate_task(background=True) routing # Integration: delegate_task(background=True) routing
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

221
tools/async_delegation.py
View file

@ -76,11 +76,12 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3
# How many completed records to retain for status queries before pruning. # How many completed records to retain for status queries before pruning.
_MAX_RETAINED_COMPLETED = 50 _MAX_RETAINED_COMPLETED = 50
_DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60 _DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60
_MAX_DURABLE_PENDING = 1000
_DB_LOCK = threading.Lock() _DB_LOCK = threading.Lock()
def _db_path(): def _db_path():
return get_hermes_home() / "delegations.db" return get_hermes_home() / "state.db"
def _connect() -> sqlite3.Connection: def _connect() -> sqlite3.Connection:
@ -102,40 +103,97 @@ def _connect() -> sqlite3.Connection:
result_json TEXT, result_json TEXT,
delivery_state TEXT NOT NULL DEFAULT 'pending', delivery_state TEXT NOT NULL DEFAULT 'pending',
delivery_attempts INTEGER NOT NULL DEFAULT 0, delivery_attempts INTEGER NOT NULL DEFAULT 0,
delivered_at REAL delivered_at REAL,
owner_pid INTEGER,
owner_started_at INTEGER,
task_json TEXT,
delivery_claim TEXT,
delivery_claimed_at REAL
)""" )"""
) )
columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")}
for name, sql_type in (
("owner_pid", "INTEGER"),
("owner_started_at", "INTEGER"),
("task_json", "TEXT"),
("delivery_claim", "TEXT"),
("delivery_claimed_at", "REAL"),
):
if name not in columns:
conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}")
return conn return conn
def _persist_dispatch(record: Dict[str, Any]) -> None: def _persist_dispatch(record: Dict[str, Any]) -> None:
now = time.time() now = time.time()
try:
from gateway.status import get_process_start_time
owner_started_at = get_process_start_time(__import__("os").getpid())
except Exception:
owner_started_at = None
task_payload = {
key: record.get(key)
for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch")
if key in record
}
with _DB_LOCK, _connect() as conn: with _DB_LOCK, _connect() as conn:
conn.execute( conn.execute(
"""INSERT OR REPLACE INTO async_delegations """INSERT OR REPLACE INTO async_delegations
(delegation_id, origin_session, origin_ui_session_id, (delegation_id, origin_session, origin_ui_session_id,
parent_session_id, state, dispatched_at, updated_at, parent_session_id, state, dispatched_at, updated_at,
delivery_state, delivery_attempts) delivery_state, delivery_attempts, owner_pid,
VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0)""", owner_started_at, task_json)
VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""",
(record["delegation_id"], record.get("session_key", ""), (record["delegation_id"], record.get("session_key", ""),
record.get("origin_ui_session_id", ""), record.get("parent_session_id"), record.get("origin_ui_session_id", ""), record.get("parent_session_id"),
record["dispatched_at"], now), record["dispatched_at"], now, __import__("os").getpid(),
owner_started_at, json.dumps(task_payload)),
) )
cutoff = now - _DURABLE_RETENTION_SECONDS _prune_durable_records()
def _delete_durable_delegation(delegation_id: str) -> None:
with _DB_LOCK, _connect() as conn:
conn.execute("DELETE FROM async_delegations WHERE delegation_id=?", (delegation_id,))
def _prune_durable_records() -> None:
"""Bound terminal history, preferring delivered records for deletion."""
now = time.time()
cutoff = now - _DURABLE_RETENTION_SECONDS
with _DB_LOCK, _connect() as conn:
conn.execute( conn.execute(
"""DELETE FROM async_delegations WHERE delegation_id IN ( "DELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?",
SELECT delegation_id FROM async_delegations (cutoff,),
WHERE (delivery_state = 'delivered' AND updated_at < ?)
OR (completed_at IS NOT NULL AND completed_at < ?)
ORDER BY updated_at ASC
)""", (cutoff, cutoff),
)
conn.execute(
"""DELETE FROM async_delegations WHERE delegation_id IN (
SELECT delegation_id FROM async_delegations
WHERE state != 'running' ORDER BY updated_at DESC LIMIT -1 OFFSET ?
)""", (_MAX_RETAINED_COMPLETED,),
) )
terminal_count = conn.execute(
"SELECT COUNT(*) FROM async_delegations WHERE state NOT IN ('running','finalizing')"
).fetchone()[0]
excess = max(0, terminal_count - _MAX_RETAINED_COMPLETED)
if excess:
conn.execute(
"""DELETE FROM async_delegations WHERE delegation_id IN (
SELECT delegation_id FROM async_delegations
WHERE state NOT IN ('running','finalizing')
ORDER BY CASE delivery_state WHEN 'delivered' THEN 0 ELSE 1 END,
updated_at ASC LIMIT ?
)""",
(excess,),
)
pending_count = conn.execute(
"""SELECT COUNT(*) FROM async_delegations
WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'"""
).fetchone()[0]
overflow = max(0, pending_count - _MAX_DURABLE_PENDING)
if overflow:
conn.execute(
"""DELETE FROM async_delegations WHERE delegation_id IN (
SELECT delegation_id FROM async_delegations
WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'
ORDER BY updated_at ASC LIMIT ?
)""",
(overflow,),
)
def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None: def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None:
@ -158,20 +216,64 @@ def _note_delivery_attempt(delegation_id: str) -> None:
) )
def recover_abandoned_delegations() -> int:
"""Classify records whose owning process disappeared as outcome unknown."""
try:
from gateway.status import _pid_exists, get_process_start_time
except Exception:
return 0
now = time.time()
recovered = 0
with _DB_LOCK, _connect() as conn:
rows = conn.execute(
"""SELECT delegation_id, origin_session, origin_ui_session_id,
parent_session_id, dispatched_at, owner_pid,
owner_started_at, task_json
FROM async_delegations WHERE state IN ('running','finalizing')"""
).fetchall()
for row in rows:
delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json = row
live = False
if pid:
live = _pid_exists(int(pid))
if live and started is not None:
live = get_process_start_time(int(pid)) == int(started)
if live:
continue
task = json.loads(task_json or "{}")
event = {
"type": "async_delegation", "delegation_id": delegation_id,
"session_key": session_key, "origin_ui_session_id": origin_ui,
"parent_session_id": parent_id, "goal": task.get("goal", ""),
"goals": task.get("goals"), "context": task.get("context"),
"toolsets": task.get("toolsets"), "role": task.get("role"),
"model": task.get("model"), "is_batch": bool(task.get("is_batch")),
"status": "unknown", "summary": None,
"error": "Delegation owner exited before recording a terminal result; outcome unknown.",
"dispatched_at": dispatched_at, "completed_at": now,
}
result = {"status": "unknown", "summary": None, "error": event["error"]}
conn.execute(
"""UPDATE async_delegations SET state='unknown', completed_at=?,
updated_at=?, event_json=?, result_json=?, delivery_state='pending'
WHERE delegation_id=?""",
(now, now, json.dumps(event), json.dumps(result), delegation_id),
)
recovered += 1
return recovered
def restore_undelivered_completions(target_queue) -> int: def restore_undelivered_completions(target_queue) -> int:
"""Enqueue durable pending completions as fresh turns after process start.""" """Enqueue durable pending completions as fresh turns after process start."""
recover_abandoned_delegations()
with _DB_LOCK, _connect() as conn: with _DB_LOCK, _connect() as conn:
rows = conn.execute( rows = conn.execute(
"""SELECT delegation_id, event_json FROM async_delegations """SELECT delegation_id, event_json FROM async_delegations
WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL
ORDER BY completed_at, delegation_id""" ORDER BY completed_at, delegation_id"""
).fetchall() ).fetchall()
for delegation_id, payload in rows: for _delegation_id, payload in rows:
target_queue.put(json.loads(payload)) target_queue.put(json.loads(payload))
conn.execute(
"UPDATE async_delegations SET delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=?",
(time.time(), delegation_id),
)
return len(rows) return len(rows)
@ -187,6 +289,75 @@ def mark_completion_delivered(delegation_id: str) -> bool:
return cur.rowcount == 1 return cur.rowcount == 1
def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool:
"""Claim one pending completion across competing consumers/processes."""
now = time.time()
with _DB_LOCK, _connect() as conn:
row = conn.execute(
"SELECT delivery_state FROM async_delegations WHERE delegation_id=?",
(delegation_id,),
).fetchone()
if row is None:
return True # legacy event created before durable dispatch
cur = conn.execute(
"""UPDATE async_delegations SET delivery_claim=?, delivery_claimed_at=?,
delivery_attempts=delivery_attempts+1, updated_at=?
WHERE delegation_id=? AND delivery_state='pending'
AND (delivery_claim IS NULL OR delivery_claimed_at < ?)""",
(claim_id, now, now, delegation_id, now - 300),
)
return cur.rowcount == 1
def claim_event_delivery(evt: Dict[str, Any], consumer: str) -> Optional[str]:
"""Claim a durable delegation event; non-durable events need no token."""
if evt.get("type") != "async_delegation":
return ""
delegation_id = str(evt.get("delegation_id") or "")
if not delegation_id:
return ""
claim_id = f"{consumer}:{__import__('os').getpid()}:{uuid.uuid4().hex}"
return claim_id if claim_completion_delivery(delegation_id, claim_id) else None
def release_completion_delivery(delegation_id: str, claim_id: str) -> bool:
"""Release a failed delivery claim so another consumer may retry."""
with _DB_LOCK, _connect() as conn:
cur = conn.execute(
"""UPDATE async_delegations SET delivery_claim=NULL,
delivery_claimed_at=NULL, updated_at=?
WHERE delegation_id=? AND delivery_state='pending'
AND delivery_claim=?""",
(time.time(), delegation_id, claim_id),
)
return cur.rowcount == 1
def complete_completion_delivery(delegation_id: str, claim_id: str) -> bool:
"""Acknowledge acceptance for the consumer holding this claim."""
now = time.time()
with _DB_LOCK, _connect() as conn:
cur = conn.execute(
"""UPDATE async_delegations SET delivery_state='delivered',
delivered_at=?, updated_at=?, delivery_claim=NULL,
delivery_claimed_at=NULL
WHERE delegation_id=? AND delivery_state='pending'
AND delivery_claim=?""",
(now, now, delegation_id, claim_id),
)
return cur.rowcount == 1
def complete_event_delivery(evt: Dict[str, Any], claim_id: str) -> None:
if claim_id and evt.get("type") == "async_delegation":
complete_completion_delivery(str(evt.get("delegation_id") or ""), claim_id)
def release_event_delivery(evt: Dict[str, Any], claim_id: str) -> None:
if claim_id and evt.get("type") == "async_delegation":
release_completion_delivery(str(evt.get("delegation_id") or ""), claim_id)
def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
with _DB_LOCK, _connect() as conn: with _DB_LOCK, _connect() as conn:
row = conn.execute( row = conn.execute(
@ -365,6 +536,7 @@ def dispatch_async_delegation(
except Exception as exc: # pragma: no cover — pool submit failure is rare except Exception as exc: # pragma: no cover — pool submit failure is rare
with _records_lock: with _records_lock:
_records.pop(delegation_id, None) _records.pop(delegation_id, None)
_delete_durable_delegation(delegation_id)
return { return {
"status": "rejected", "status": "rejected",
"error": f"Failed to schedule async delegation: {exc}", "error": f"Failed to schedule async delegation: {exc}",
@ -449,7 +621,6 @@ def _push_completion_event(
_persist_completion(evt, result) _persist_completion(evt, result)
try: try:
process_registry.completion_queue.put(evt) process_registry.completion_queue.put(evt)
_note_delivery_attempt(str(record.get("delegation_id") or ""))
except Exception as exc: # pragma: no cover except Exception as exc: # pragma: no cover
logger.error( logger.error(
"Async delegation %s: failed to enqueue completion event; " "Async delegation %s: failed to enqueue completion event; "
@ -566,6 +737,7 @@ def dispatch_async_delegation_batch(
except Exception as exc: # pragma: no cover except Exception as exc: # pragma: no cover
with _records_lock: with _records_lock:
_records.pop(delegation_id, None) _records.pop(delegation_id, None)
_delete_durable_delegation(delegation_id)
return { return {
"status": "rejected", "status": "rejected",
"error": f"Failed to schedule async delegation batch: {exc}", "error": f"Failed to schedule async delegation batch: {exc}",
@ -628,7 +800,6 @@ def _finalize_batch(
_persist_completion(evt, combined) _persist_completion(evt, combined)
try: try:
process_registry.completion_queue.put(evt) process_registry.completion_queue.put(evt)
_note_delivery_attempt(delegation_id)
except Exception as exc: # pragma: no cover except Exception as exc: # pragma: no cover
logger.error( logger.error(
"Async delegation batch %s: failed to enqueue completion event; " "Async delegation batch %s: failed to enqueue completion event; "

33
tui_gateway/server.py
View file

@ -8793,13 +8793,18 @@ def _notification_poller_loop(
continue continue
rid = f"__notif__{int(time.time() * 1000)}" rid = f"__notif__{int(time.time() * 1000)}"
from tools.async_delegation import (
claim_event_delivery, complete_event_delivery, release_event_delivery,
)
_claim = claim_event_delivery(evt, "tui-poller")
if _claim is None:
continue
try: try:
_emit("message.start", sid) _emit("message.start", sid)
_run_prompt_submit(rid, sid, session, text) _run_prompt_submit(rid, sid, session, text)
if evt.get("type") == "async_delegation": complete_event_delivery(evt, _claim)
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(evt.get("delegation_id") or ""))
except Exception as exc: except Exception as exc:
release_event_delivery(evt, _claim)
print( print(
f"[tui_gateway] notification poller dispatch failed: " f"[tui_gateway] notification poller dispatch failed: "
f"{type(exc).__name__}: {exc}", f"{type(exc).__name__}: {exc}",
@ -8848,13 +8853,18 @@ def _notification_poller_loop(
session["running"] = True session["running"] = True
rid = f"__notif__{int(time.time() * 1000)}" rid = f"__notif__{int(time.time() * 1000)}"
from tools.async_delegation import (
claim_event_delivery, complete_event_delivery, release_event_delivery,
)
_claim = claim_event_delivery(evt, "tui-poller")
if _claim is None:
continue
try: try:
_emit("message.start", sid) _emit("message.start", sid)
_run_prompt_submit(rid, sid, session, text) _run_prompt_submit(rid, sid, session, text)
if evt.get("type") == "async_delegation": complete_event_delivery(evt, _claim)
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(evt.get("delegation_id") or ""))
except Exception as exc: except Exception as exc:
release_event_delivery(evt, _claim)
print( print(
f"[tui_gateway] notification poller dispatch failed: " f"[tui_gateway] notification poller dispatch failed: "
f"{type(exc).__name__}: {exc}", f"{type(exc).__name__}: {exc}",
@ -9404,13 +9414,18 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
process_registry.completion_queue.put(_evt) process_registry.completion_queue.put(_evt)
break break
session["running"] = True session["running"] = True
from tools.async_delegation import (
claim_event_delivery, complete_event_delivery, release_event_delivery,
)
_claim = claim_event_delivery(_evt, "tui-post-turn")
if _claim is None:
continue
try: try:
_emit("message.start", sid) _emit("message.start", sid)
_run_prompt_submit(rid, sid, session, synth) _run_prompt_submit(rid, sid, session, synth)
if _evt.get("type") == "async_delegation": complete_event_delivery(_evt, _claim)
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
except Exception as _n_exc: except Exception as _n_exc:
release_event_delivery(_evt, _claim)
print( print(
f"[tui_gateway] completion notification dispatch failed: " f"[tui_gateway] completion notification dispatch failed: "
f"{type(_n_exc).__name__}: {_n_exc}", f"{type(_n_exc).__name__}: {_n_exc}",

15
website/docs/user-guide/features/delegation.md
View file

@ -129,6 +129,21 @@ When you provide a `tasks` array, subagents run in **parallel** using a thread p
Single-task delegation runs directly without thread pool overhead. Single-task delegation runs directly without thread pool overhead.
### Durable background completions
When a background delegation finishes, Hermes stores its completion event in
the active profile's `state.db` before publishing it to the normal fresh-turn
queue. If Hermes restarts after completion but before delivery, the pending
event is restored and routed through the same ownership checks. Competing
consumers use a durable claim, so only the consumer that successfully accepts
the synthetic turn acknowledges delivery; failed attempts release the claim for
retry.
This does not resume child execution after a crash. A delegation whose owner
process disappears while it is still running is recorded as `unknown`, because
Hermes cannot prove whether its external side effects happened. Pending and
delivered records are bounded and profile-local.
## Model Override ## Model Override
You can configure a different model for subagents via `config.yaml` — useful for delegating simple tasks to cheaper/faster models: You can configure a different model for subagents via `config.yaml` — useful for delegating simple tasks to cheaper/faster models:

8
website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/delegation.md
View file

@ -127,7 +127,13 @@ delegate_task(
- **结果排序：** 结果按任务索引排序，与输入顺序一致，不受完成顺序影响 - **结果排序：** 结果按任务索引排序，与输入顺序一致，不受完成顺序影响
- **中断传播：** 中断父智能体（例如发送新消息）会中断所有活跃的子智能体 - **中断传播：** 中断父智能体（例如发送新消息）会中断所有活跃的子智能体
单任务委派直接运行，无线程池开销。 单任务委派直接运行，不会产生线程池开销。
### 持久化后台完成事件
后台委派完成后，Hermes 会先把完成事件写入当前 profile 的 `state.db`，再发布到正常的新轮次队列。如果 Hermes 在完成后、交付前重启，待处理事件会被恢复，并继续经过相同的所有权检查。多个消费者通过持久化 claim 竞争；只有成功接收合成轮次的消费者会确认交付，失败尝试会释放 claim 以便重试。
这不会在崩溃后恢复子智能体执行。如果委派仍在运行时其所有者进程消失，Hermes 会将其记录为 `unknown`，因为无法证明外部副作用是否已经发生。待处理和已交付记录都有界，并按 profile 隔离。
## 模型覆盖 ## 模型覆盖