From d0e9a42cecefe5dbc7a750ae43c2149f4d3264f8 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 12 Jul 2026 17:53:10 -0700 Subject: [PATCH] fix(delegation): harden durable completion delivery --- cli.py | 20 +- gateway/run.py | 424 +++++++++++++----- hermes_state.py | 25 +- tests/gateway/test_completion_delivery.py | 4 +- tests/tools/test_async_delegation.py | 99 +++- tools/async_delegation.py | 221 +++++++-- tui_gateway/server.py | 33 +- .../docs/user-guide/features/delegation.md | 15 + .../current/user-guide/features/delegation.md | 8 +- 9 files changed, 696 insertions(+), 153 deletions(-) diff --git a/cli.py b/cli.py index 14dac5196e0d..9887bb029780 100644 --- a/cli.py +++ b/cli.py @@ -15189,10 +15189,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): from tools.approval import get_current_session_key _drain_sk = get_current_session_key(default="") 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) - if _evt.get("type") == "async_delegation": - from tools.async_delegation import mark_completion_delivered - mark_completion_delivered(str(_evt.get("delegation_id") or "")) + complete_event_delivery(_evt, _claim) except Exception: pass continue @@ -15354,10 +15358,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): try: from tools.process_registry import process_registry 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) - if _evt.get("type") == "async_delegation": - from tools.async_delegation import mark_completion_delivered - mark_completion_delivered(str(_evt.get("delegation_id") or "")) + complete_event_delivery(_evt, _claim) except Exception: pass # Non-fatal — don't break the main loop diff --git a/gateway/run.py b/gateway/run.py index 5690a906ce54..94b91d885e10 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1772,6 +1772,7 @@ from gateway.config import ( load_gateway_config, ) from gateway.session import ( + AsyncSessionStore, SessionStore, SessionSource, SessionContext, @@ -2881,6 +2882,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew 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._running = False 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. self._session_sources: "OrderedDict[str, SessionSource]" = OrderedDict() 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. # 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. 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). """ from hermes_constants import parse_reasoning_effort @@ -5202,7 +5216,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: 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. 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 orphaned compression siblings (#56391). Callers demote interrupt to 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) if not session_key or session_store is None: return False try: - with session_store._lock: # noqa: SLF001 — snapshot entry under lock - session_store._ensure_loaded_locked() # noqa: SLF001 - 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 + session_id = await asyncio.to_thread( + self._lookup_session_id_under_store_lock, session_store, session_key + ) except Exception: return False + if not session_id: + return False session_db = getattr(self, "_session_db", None) if session_db is None: return False - db = getattr(session_db, "_db", session_db) + raw_db = getattr(session_db, "_db", session_db) 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: 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 # (and the draining/steer-fallback/subagent-demotion paths that share # 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" demoted_for_compression = ( 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: logger.info( @@ -5743,7 +5772,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew source = None try: 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) source = getattr(entry, "origin", None) if entry else None except Exception as e: @@ -7013,7 +7042,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew pass else: try: - suspended = self.session_store.suspend_recently_active() + suspended = await self.async_session_store.suspend_recently_active() if suspended: logger.info("Marked %d in-flight session(s) as resumable from previous run", suspended) 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 # the home channel has never been used, get_or_create_session # 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 # 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 # 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: raise RuntimeError( f"could not switch session key {session_key} → {cli_session_id}" @@ -7657,13 +7686,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _MAX_FINALIZE_RETRIES = 3 while self._running: try: - self.session_store._ensure_loaded() + await self.async_session_store._ensure_loaded() # Collect expired sessions first, then log a single summary. _expired_entries = [] for key, entry in list(self.session_store._entries.items()): if entry.expiry_finalized: continue - if not self.session_store._is_session_expired(entry): + if not await self.async_session_store._is_session_expired(entry): continue _expired_entries.append((key, entry)) @@ -7747,7 +7776,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # state.db (single write-path, #9006) — also drops # the persisted /model override, since finalization # is a conversation boundary. - self.session_store.set_expiry_finalized(entry) + await self.async_session_store.set_expiry_finalized(entry) logger.debug( "Session expiry finalized for %s", entry.session_id, @@ -7762,7 +7791,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Marking as finalized to prevent infinite retry loop.", failures, entry.session_id, e, ) - self.session_store.set_expiry_finalized( + await self.async_session_store.set_expiry_finalized( entry, clear_model_override=False ) _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 ) 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: logger.info( "SessionStore prune: dropped %d stale entries", @@ -8149,7 +8178,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _agent is _AGENT_PENDING_SENTINEL: continue try: - self.session_store.mark_resume_pending( + await self.async_session_store.mark_resume_pending( _sk, "restart_timeout" if self._restart_requested else "shutdown_timeout", ) @@ -8180,7 +8209,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew for _sk in _pre_drain_keys: if _sk not in self._running_agents: try: - self.session_store.clear_resume_pending(_sk) + await self.async_session_store.clear_resume_pending(_sk) except Exception as _e: logger.debug( "clear_resume_pending after drain failed for %s: %s", @@ -8223,7 +8252,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _agent is _AGENT_PENDING_SENTINEL: continue 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: logger.debug( "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 # siblings. Demote to queue semantics so the follow-up waits # 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( "PRIORITY interrupt demoted to queue for session %s " "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: qcmd = quick_commands[command] if qcmd.get("type") == "alias": - target = qcmd.get("target", "").strip() + target = (qcmd.get("target") or "").strip() if target: target = target if target.startswith("/") else f"/{target}" target_command = target.lstrip("/") @@ -10049,7 +10078,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew else: return f"Quick command '/{command}' has no command defined." elif qcmd.get("type") == "alias": - target = qcmd.get("target", "").strip() + target = (qcmd.get("target") or "").strip() if target: target = target if target.startswith("/") else f"/{target}" target_command = target.lstrip("/") @@ -10303,7 +10332,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # on error. Let the user drive the next turn. if _final_text.strip(): 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: session_entry = 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 _msg_cwd = os.environ.get("TERMINAL_CWD", os.path.expanduser("~")) - _msg_runtime = _resolve_runtime_agent_kwargs() _msg_config_ctx = None + _msg_cfg = None + _msg_model_cfg = {} + _msg_custom_providers = [] try: _msg_cfg = _load_gateway_config() _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") if _msg_raw_ctx is not None: _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: 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( - self._model, - base_url=self._base_url or _msg_runtime.get("base_url") or "", + _msg_model, + base_url=_msg_base_url, api_key=_msg_runtime.get("api_key") or "", config_context_length=_msg_config_ctx, + provider=_msg_runtime.get("provider") or "", + custom_providers=_msg_custom_providers, ) _ctx_result = await preprocess_context_references_async( message_text, @@ -10653,10 +10728,35 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _ctx_result.expanded: message_text = _ctx_result.message 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 + 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]: pending_native = getattr(self, "_pending_native_image_paths_by_session", None) if not pending_native: @@ -10684,6 +10784,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: 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): if not session_key: return None @@ -10727,7 +10836,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: 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 pinned_session_id = str( (getattr(event, "metadata", None) or {}).get("gateway_session_id") or "" @@ -10759,7 +10868,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) return 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: session_entry = switched logger.info( @@ -10809,7 +10918,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # lane session is ended cleanly. Mutating session_entry in # place here created a split-brain state where the JSON # 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: session_entry = switched # 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) # 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 @@ -11260,7 +11369,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) if _hyg_rotated: session_entry.session_id = _hyg_new_sid - self.session_store._save() + await self.async_session_store._save() await asyncio.to_thread( self._sync_telegram_topic_binding, source, session_entry, @@ -11288,7 +11397,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # messages and replace them with only the compressed # summary (permanent data loss, #21301). if _hyg_rotated: - self.session_store.rewrite_transcript( + await self.async_session_store.rewrite_transcript( session_entry.session_id, _compressed ) # 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 - 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. _intro_note = ( "\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 # 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, source=source, history=history, @@ -11649,7 +11758,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if session_key and _should_clear_resume_pending_after_turn(agent_result): self._clear_restart_failure_count(session_key) try: - self.session_store.clear_resume_pending(session_key) + await self.async_session_store.clear_resume_pending(session_key) except Exception as _e: logger.debug( "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 session_entry.session_id == _run_start_session_id: session_entry.session_id = agent_result["session_id"] - self.session_store._save() - self.session_store._record_gateway_session_peer( + await self.async_session_store._save() + await self.async_session_store._record_gateway_session_peer( session_entry.session_id, session_key, source, @@ -11870,7 +11979,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Auto-resetting session %s after compression exhaustion.", 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._session_model_overrides.pop(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 elif not history: tool_defs = agent_result.get("tools", []) - self.session_store.append_to_transcript( + await self.async_session_store.append_to_transcript( session_entry.session_id, { "role": "session_meta", @@ -11968,7 +12077,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # after transient failures). #47237 _skip_persist = ( 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) ) ) @@ -11979,7 +12088,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew event.message_id, session_entry.session_id, ) else: - self.session_store.append_to_transcript( + await self.async_session_store.append_to_transcript( session_entry.session_id, _user_entry, skip_db=agent_persisted, @@ -12005,13 +12114,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew } if 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, _user_entry, skip_db=agent_persisted, ) if response: - self.session_store.append_to_transcript( + await self.async_session_store.append_to_transcript( session_entry.session_id, {"role": "assistant", "content": response, "timestamp": ts}, skip_db=agent_persisted, @@ -12036,7 +12145,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ): entry["message_id"] = str(event.message_id) _user_msg_id_attached = True - self.session_store.append_to_transcript( + await self.async_session_store.append_to_transcript( session_entry.session_id, entry, skip_db=agent_persisted, ) @@ -12044,7 +12153,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Token counts and model are now persisted by the agent directly. # Keep only last_prompt_tokens here for context-window tracking and # compression decisions. - self.session_store.update_session( + await self.async_session_store.update_session( session_entry.session_key, 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: _already_persisted = False 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: _recent_transcript = [] for _msg in reversed(_recent_transcript[-10:]): @@ -12172,7 +12281,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew } if getattr(event, "message_id", None): _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, _user_entry, ) @@ -12627,7 +12736,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: 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. 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) return None, None 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: logger.debug("goal manager: session lookup failed: %s", exc) 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.]", } try: - session_entry = self.session_store.get_or_create_session(event.source) - self.session_store.append_to_transcript( + session_entry = await self.async_session_store.get_or_create_session(event.source) + await self.async_session_store.append_to_transcript( session_entry.session_id, reload_msg ) except Exception: @@ -15354,11 +15463,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew user_name=str(evt.get("user_name") or "").strip() or None, ) - async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None: - """Inject a watch-pattern notification as a synthetic message event. + async def _inject_watch_notification( + 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. + 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) if not source: @@ -15366,7 +15481,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Dropping watch notification with no routing metadata for process %s", evt.get("session_id", "unknown"), ) - return + return None platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform) adapter = None for p, a in self.adapters.items(): @@ -15374,7 +15489,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew adapter = a break if not adapter: - return + return None try: metadata = {} parent_session_id = str(evt.get("parent_session_id") or "").strip() @@ -15395,8 +15510,118 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew source.thread_id, ) await adapter.handle_message(synth_event) + return True except Exception as 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: """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: continue try: - await self._inject_watch_notification(synth_text, evt) - from tools.async_delegation import mark_completion_delivered - mark_completion_delivered(str(evt.get("delegation_id") or "")) + delivered = await self._deliver_completion_notification(synth_text, evt) + if delivered is False: + _pr.completion_queue.put(evt) except Exception as e: + _pr.completion_queue.put(evt) logger.error("Async delegation injection error: %s", e) except Exception as 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. 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): + from agent.redact import redact_terminal_output from tools.ansi_strip import strip_ansi + _command = getattr(session, "command", "") or "" _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 # mid-line (fixes #23284). Keep the last ~2000 chars but # 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}" else: _out = _raw - synth_text = format_process_notification({ + completion_evt = { "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_key": session_key, "platform": platform_name, + "chat_type": watcher.get("chat_type", ""), "chat_id": chat_id, "thread_id": thread_id, "user_id": user_id, "user_name": user_name, - }) - if not source: - logger.warning( - "Dropping completion notification with no routing metadata for process %s", - session_id, - ) + "message_id": message_id, + "started_at": getattr(session, "started_at", None), + "command": _command, + "exit_code": session.exit_code, + "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 - - adapter = None - for p, a in self.adapters.items(): - if p == source.platform: - adapter = a - break - if adapter and source.chat_id: - 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) + delivered = await self._deliver_completion_notification( + synth_text, completion_evt, + ) + if delivered is False: + # The process remains terminal; retry after failed + # adapter injection instead of suppressing the result. + continue break # --- Normal text-only notification --- @@ -16553,7 +16760,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if url: return url.rstrip("/") 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: return url.rstrip("/") return None @@ -19840,7 +20048,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew session_key or "?", exc_info=True, ) - next_message = await self._prepare_inbound_message_text( + next_message = await self._prepare_profile_scoped_inbound_message_text( event=pending_event, source=next_source, history=updated_history, diff --git a/hermes_state.py b/hermes_state.py index da2e484f7cf1..11d1507bb318 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -140,7 +140,7 @@ T = TypeVar("T") 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. # 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 ); +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_id ON sessions(source, 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_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_async_delegations_delivery + ON async_delegations(delivery_state, completed_at); """ # Indexes that reference columns added in later schema versions must be diff --git a/tests/gateway/test_completion_delivery.py b/tests/gateway/test_completion_delivery.py index b123f45886f1..57c75f755a66 100644 --- a/tests/gateway/test_completion_delivery.py +++ b/tests/gateway/test_completion_delivery.py @@ -175,8 +175,8 @@ def test_failed_async_injection_is_retried_and_only_success_is_acked( acknowledgements = [] monkeypatch.setattr( async_delegation, - "mark_completion_delivered", - lambda delegation_id: acknowledgements.append(delegation_id) or True, + "complete_completion_delivery", + lambda delegation_id, _claim_id: acknowledgements.append(delegation_id) or True, raising=False, ) diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index bf2187e740c4..77732960bcaf 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -244,7 +244,8 @@ def test_completion_is_persisted_and_delivery_can_be_acknowledged(tmp_path, monk assert row["state"] == "completed" assert row["result"]["summary"] == "survived" 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.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" +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 # --------------------------------------------------------------------------- diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 011b398c78c6..d2f2dc23b91b 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -76,11 +76,12 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3 # How many completed records to retain for status queries before pruning. _MAX_RETAINED_COMPLETED = 50 _DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60 +_MAX_DURABLE_PENDING = 1000 _DB_LOCK = threading.Lock() def _db_path(): - return get_hermes_home() / "delegations.db" + return get_hermes_home() / "state.db" def _connect() -> sqlite3.Connection: @@ -102,40 +103,97 @@ def _connect() -> sqlite3.Connection: result_json TEXT, delivery_state TEXT NOT NULL DEFAULT 'pending', 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 def _persist_dispatch(record: Dict[str, Any]) -> None: 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: conn.execute( """INSERT OR REPLACE INTO async_delegations (delegation_id, origin_session, origin_ui_session_id, parent_session_id, state, dispatched_at, updated_at, - delivery_state, delivery_attempts) - VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0)""", + delivery_state, delivery_attempts, owner_pid, + owner_started_at, task_json) + VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""", (record["delegation_id"], record.get("session_key", ""), 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( - """DELETE FROM async_delegations WHERE delegation_id IN ( - SELECT delegation_id FROM async_delegations - 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,), + "DELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?", + (cutoff,), ) + 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: @@ -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: """Enqueue durable pending completions as fresh turns after process start.""" + recover_abandoned_delegations() with _DB_LOCK, _connect() as conn: rows = conn.execute( """SELECT delegation_id, event_json FROM async_delegations WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL ORDER BY completed_at, delegation_id""" ).fetchall() - for delegation_id, payload in rows: + for _delegation_id, payload in rows: 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) @@ -187,6 +289,75 @@ def mark_completion_delivered(delegation_id: str) -> bool: 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]]: with _DB_LOCK, _connect() as conn: row = conn.execute( @@ -365,6 +536,7 @@ def dispatch_async_delegation( except Exception as exc: # pragma: no cover — pool submit failure is rare with _records_lock: _records.pop(delegation_id, None) + _delete_durable_delegation(delegation_id) return { "status": "rejected", "error": f"Failed to schedule async delegation: {exc}", @@ -449,7 +621,6 @@ def _push_completion_event( _persist_completion(evt, result) try: process_registry.completion_queue.put(evt) - _note_delivery_attempt(str(record.get("delegation_id") or "")) except Exception as exc: # pragma: no cover logger.error( "Async delegation %s: failed to enqueue completion event; " @@ -566,6 +737,7 @@ def dispatch_async_delegation_batch( except Exception as exc: # pragma: no cover with _records_lock: _records.pop(delegation_id, None) + _delete_durable_delegation(delegation_id) return { "status": "rejected", "error": f"Failed to schedule async delegation batch: {exc}", @@ -628,7 +800,6 @@ def _finalize_batch( _persist_completion(evt, combined) try: process_registry.completion_queue.put(evt) - _note_delivery_attempt(delegation_id) except Exception as exc: # pragma: no cover logger.error( "Async delegation batch %s: failed to enqueue completion event; " diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 5dc90b9f6f4b..83c79975e2ae 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -8793,13 +8793,18 @@ def _notification_poller_loop( continue 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: _emit("message.start", sid) _run_prompt_submit(rid, sid, session, text) - if evt.get("type") == "async_delegation": - from tools.async_delegation import mark_completion_delivered - mark_completion_delivered(str(evt.get("delegation_id") or "")) + complete_event_delivery(evt, _claim) except Exception as exc: + release_event_delivery(evt, _claim) print( f"[tui_gateway] notification poller dispatch failed: " f"{type(exc).__name__}: {exc}", @@ -8848,13 +8853,18 @@ def _notification_poller_loop( session["running"] = True 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: _emit("message.start", sid) _run_prompt_submit(rid, sid, session, text) - if evt.get("type") == "async_delegation": - from tools.async_delegation import mark_completion_delivered - mark_completion_delivered(str(evt.get("delegation_id") or "")) + complete_event_delivery(evt, _claim) except Exception as exc: + release_event_delivery(evt, _claim) print( f"[tui_gateway] notification poller dispatch failed: " 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) break 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: _emit("message.start", sid) _run_prompt_submit(rid, sid, session, synth) - if _evt.get("type") == "async_delegation": - from tools.async_delegation import mark_completion_delivered - mark_completion_delivered(str(_evt.get("delegation_id") or "")) + complete_event_delivery(_evt, _claim) except Exception as _n_exc: + release_event_delivery(_evt, _claim) print( f"[tui_gateway] completion notification dispatch failed: " f"{type(_n_exc).__name__}: {_n_exc}", diff --git a/website/docs/user-guide/features/delegation.md b/website/docs/user-guide/features/delegation.md index 037c2e806ae1..6a5b39faa22e 100644 --- a/website/docs/user-guide/features/delegation.md +++ b/website/docs/user-guide/features/delegation.md @@ -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. +### 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 You can configure a different model for subagents via `config.yaml` — useful for delegating simple tasks to cheaper/faster models: diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/delegation.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/delegation.md index 6458a9ec71a6..0c923ebf492f 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/delegation.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/delegation.md @@ -127,7 +127,13 @@ delegate_task( - **结果排序：** 结果按任务索引排序，与输入顺序一致，不受完成顺序影响 - **中断传播：** 中断父智能体（例如发送新消息）会中断所有活跃的子智能体 -单任务委派直接运行，无线程池开销。 +单任务委派直接运行，不会产生线程池开销。 + +### 持久化后台完成事件 + +后台委派完成后，Hermes 会先把完成事件写入当前 profile 的 `state.db`，再发布到正常的新轮次队列。如果 Hermes 在完成后、交付前重启，待处理事件会被恢复，并继续经过相同的所有权检查。多个消费者通过持久化 claim 竞争；只有成功接收合成轮次的消费者会确认交付，失败尝试会释放 claim 以便重试。 + +这不会在崩溃后恢复子智能体执行。如果委派仍在运行时其所有者进程消失，Hermes 会将其记录为 `unknown`，因为无法证明外部副作用是否已经发生。待处理和已交付记录都有界，并按 profile 隔离。 ## 模型覆盖