diff --git a/cli.py b/cli.py index 37c158dabf1a..14dac5196e0d 100644 --- a/cli.py +++ b/cli.py @@ -15190,6 +15190,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): _drain_sk = get_current_session_key(default="") for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk): 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 "")) except Exception: pass continue @@ -15352,6 +15355,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): from tools.process_registry import process_registry for _evt, _synth in process_registry.drain_notifications(): 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 "")) except Exception: pass # Non-fatal — don't break the main loop diff --git a/gateway/run.py b/gateway/run.py index e908781d0fc3..5690a906ce54 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1772,7 +1772,6 @@ from gateway.config import ( load_gateway_config, ) from gateway.session import ( - AsyncSessionStore, SessionStore, SessionSource, SessionContext, @@ -2882,10 +2881,6 @@ 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 @@ -2979,15 +2974,6 @@ 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 @@ -4829,7 +4815,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", "max", "ultra". Returns None to use + "minimal", "low", "medium", "high", "xhigh". Returns None to use default (medium). """ from hermes_constants import parse_reasoning_effort @@ -5216,7 +5202,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: return False - async def _session_has_compression_in_flight(self, session_key: str) -> bool: + 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 @@ -5224,43 +5210,28 @@ 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: - session_id = await asyncio.to_thread( - self._lookup_session_id_under_store_lock, session_store, session_key - ) + 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 except Exception: return False - if not session_id: - return False session_db = getattr(self, "_session_db", None) if session_db is None: return False - raw_db = getattr(session_db, "_db", session_db) + db = getattr(session_db, "_db", session_db) try: - holder = await asyncio.to_thread( - raw_db.get_compression_lock_holder, str(session_id) - ) - return bool(holder) + return bool(db.get_compression_lock_holder(str(session_id))) 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 @@ -5483,7 +5454,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew effective_mode = "queue" demoted_for_compression = ( effective_mode == "interrupt" - and await self._session_has_compression_in_flight(session_key) + and self._session_has_compression_in_flight(session_key) ) if demoted_for_compression: logger.info( @@ -5772,7 +5743,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew source = None try: if getattr(self, "session_store", None) is not None: - await self.async_session_store._ensure_loaded() + self.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: @@ -7042,7 +7013,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew pass else: try: - suspended = await self.async_session_store.suspend_recently_active() + suspended = self.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: @@ -7603,13 +7574,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. - await self.async_session_store.get_or_create_session(dest_source) + self.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 = await self.async_session_store.switch_session(session_key, cli_session_id) + switched = self.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}" @@ -7686,13 +7657,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _MAX_FINALIZE_RETRIES = 3 while self._running: try: - await self.async_session_store._ensure_loaded() + self.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 await self.async_session_store._is_session_expired(entry): + if not self.session_store._is_session_expired(entry): continue _expired_entries.append((key, entry)) @@ -7776,7 +7747,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # state.db (single write-path, #9006) — also drops # the persisted /model override, since finalization # is a conversation boundary. - await self.async_session_store.set_expiry_finalized(entry) + self.session_store.set_expiry_finalized(entry) logger.debug( "Session expiry finalized for %s", entry.session_id, @@ -7791,7 +7762,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Marking as finalized to prevent infinite retry loop.", failures, entry.session_id, e, ) - await self.async_session_store.set_expiry_finalized( + self.session_store.set_expiry_finalized( entry, clear_model_override=False ) _finalize_failures.pop(entry.session_id, None) @@ -7844,7 +7815,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew getattr(self.config, "session_store_max_age_days", 0) or 0 ) if _max_age > 0: - _pruned = await self.async_session_store.prune_old_entries(_max_age) + _pruned = self.session_store.prune_old_entries(_max_age) if _pruned: logger.info( "SessionStore prune: dropped %d stale entries", @@ -8178,7 +8149,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _agent is _AGENT_PENDING_SENTINEL: continue try: - await self.async_session_store.mark_resume_pending( + self.session_store.mark_resume_pending( _sk, "restart_timeout" if self._restart_requested else "shutdown_timeout", ) @@ -8209,7 +8180,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew for _sk in _pre_drain_keys: if _sk not in self._running_agents: try: - await self.async_session_store.clear_resume_pending(_sk) + self.session_store.clear_resume_pending(_sk) except Exception as _e: logger.debug( "clear_resume_pending after drain failed for %s: %s", @@ -8252,7 +8223,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _agent is _AGENT_PENDING_SENTINEL: continue try: - await self.async_session_store.mark_resume_pending(_sk, _resume_reason) + self.session_store.mark_resume_pending(_sk, _resume_reason) except Exception as _e: logger.debug( "mark_resume_pending failed for %s: %s", @@ -9625,7 +9596,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 await self._session_has_compression_in_flight(_quick_key): + if 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)", @@ -9666,7 +9637,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") or "").strip() + target = qcmd.get("target", "").strip() if target: target = target if target.startswith("/") else f"/{target}" target_command = target.lstrip("/") @@ -10078,7 +10049,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") or "").strip() + target = qcmd.get("target", "").strip() if target: target = target if target.startswith("/") else f"/{target}" target_command = target.lstrip("/") @@ -10332,7 +10303,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # on error. Let the user drive the next turn. if _final_text.strip(): try: - session_entry = await self.async_session_store.get_or_create_session(source) + session_entry = self.session_store.get_or_create_session(source) except Exception: session_entry = None if session_entry is not None: @@ -10648,10 +10619,8 @@ 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", {}) @@ -10659,57 +10628,13 @@ 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( - _msg_model, - base_url=_msg_base_url, + self._model, + base_url=self._base_url or _msg_runtime.get("base_url") or "", 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, @@ -10728,35 +10653,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _ctx_result.expanded: message_text = _ctx_result.message except Exception as exc: - logger.warning("@ context reference expansion failed: %s", exc) - logger.debug("@ context reference expansion failure detail", exc_info=True) + logger.debug("@ context reference expansion failed: %s", exc) 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: @@ -10784,15 +10684,6 @@ 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 @@ -10836,7 +10727,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: pass - session_entry = await self.async_session_store.get_or_create_session(source) + session_entry = self.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 "" @@ -10868,7 +10759,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) return prior_session_id = session_entry.session_id - switched = await self.async_session_store.switch_session(session_key, pinned_session_id) + switched = self.session_store.switch_session(session_key, pinned_session_id) if switched is not None: session_entry = switched logger.info( @@ -10918,7 +10809,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 = await self.async_session_store.switch_session(session_key, bound_session_id) + switched = self.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 @@ -11106,7 +10997,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 = await self.async_session_store.load_transcript(session_entry.session_id) + history = self.session_store.load_transcript(session_entry.session_id) # ----------------------------------------------------------------- # Session hygiene: auto-compress pathologically large transcripts @@ -11369,7 +11260,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) if _hyg_rotated: session_entry.session_id = _hyg_new_sid - await self.async_session_store._save() + self.session_store._save() await asyncio.to_thread( self._sync_telegram_topic_binding, source, session_entry, @@ -11397,7 +11288,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # messages and replace them with only the compressed # summary (permanent data loss, #21301). if _hyg_rotated: - await self.async_session_store.rewrite_transcript( + self.session_store.rewrite_transcript( session_entry.session_id, _compressed ) # Reset stored token count — transcript rewritten @@ -11514,7 +11405,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) # First-message onboarding -- only on the very first interaction ever - if not history and not await self.async_session_store.has_any_sessions(): + if not history and not self.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. " @@ -11599,7 +11490,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_profile_scoped_inbound_message_text( + message_text = await self._prepare_inbound_message_text( event=event, source=source, history=history, @@ -11758,7 +11649,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: - await self.async_session_store.clear_resume_pending(session_key) + self.session_store.clear_resume_pending(session_key) except Exception as _e: logger.debug( "clear_resume_pending failed for %s: %s", @@ -11780,8 +11671,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"] - await self.async_session_store._save() - await self.async_session_store._record_gateway_session_peer( + self.session_store._save() + self.session_store._record_gateway_session_peer( session_entry.session_id, session_key, source, @@ -11979,7 +11870,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Auto-resetting session %s after compression exhaustion.", session_entry.session_id, ) - new_entry = await self.async_session_store.reset_session(session_key) + new_entry = self.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) @@ -12023,7 +11914,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", []) - await self.async_session_store.append_to_transcript( + self.session_store.append_to_transcript( session_entry.session_id, { "role": "session_meta", @@ -12077,7 +11968,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # after transient failures). #47237 _skip_persist = ( event.message_id - and await self.async_session_store.has_platform_message_id( + and self.session_store.has_platform_message_id( session_entry.session_id, str(event.message_id) ) ) @@ -12088,7 +11979,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew event.message_id, session_entry.session_id, ) else: - await self.async_session_store.append_to_transcript( + self.session_store.append_to_transcript( session_entry.session_id, _user_entry, skip_db=agent_persisted, @@ -12114,13 +12005,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew } if event.message_id: _user_entry["message_id"] = str(event.message_id) - await self.async_session_store.append_to_transcript( + self.session_store.append_to_transcript( session_entry.session_id, _user_entry, skip_db=agent_persisted, ) if response: - await self.async_session_store.append_to_transcript( + self.session_store.append_to_transcript( session_entry.session_id, {"role": "assistant", "content": response, "timestamp": ts}, skip_db=agent_persisted, @@ -12145,7 +12036,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ): entry["message_id"] = str(event.message_id) _user_msg_id_attached = True - await self.async_session_store.append_to_transcript( + self.session_store.append_to_transcript( session_entry.session_id, entry, skip_db=agent_persisted, ) @@ -12153,7 +12044,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. - await self.async_session_store.update_session( + self.session_store.update_session( session_entry.session_key, last_prompt_tokens=agent_result.get("last_prompt_tokens", 0), ) @@ -12253,7 +12144,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 = await self.async_session_store.load_transcript(session_entry.session_id) + _recent_transcript = self.session_store.load_transcript(session_entry.session_id) except Exception: _recent_transcript = [] for _msg in reversed(_recent_transcript[-10:]): @@ -12281,7 +12172,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew } if getattr(event, "message_id", None): _user_entry["message_id"] = str(event.message_id) - await self.async_session_store.append_to_transcript( + self.session_store.append_to_transcript( session_entry.session_id, _user_entry, ) @@ -12736,7 +12627,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: return 20 - async def _get_goal_manager_for_event(self, event: "MessageEvent"): + 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 @@ -12748,7 +12639,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew logger.debug("goal manager unavailable: %s", exc) return None, None try: - session_entry = await self.async_session_store.get_or_create_session(event.source) + session_entry = self.session_store.get_or_create_session(event.source) except Exception as exc: logger.debug("goal manager: session lookup failed: %s", exc) return None, None @@ -14191,8 +14082,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 = await self.async_session_store.get_or_create_session(event.source) - await self.async_session_store.append_to_transcript( + session_entry = self.session_store.get_or_create_session(event.source) + self.session_store.append_to_transcript( session_entry.session_id, reload_msg ) except Exception: @@ -15463,17 +15354,11 @@ 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, - ) -> Optional[bool]: - """Inject a watch/completion notification as a synthetic message event. + async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None: + """Inject a watch-pattern notification as a synthetic message event. - Routing must come from the queued event itself, not from whatever + Routing must come from the queued watch 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: @@ -15481,7 +15366,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Dropping watch notification with no routing metadata for process %s", evt.get("session_id", "unknown"), ) - return None + return platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform) adapter = None for p, a in self.adapters.items(): @@ -15489,7 +15374,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew adapter = a break if not adapter: - return None + return try: metadata = {} parent_session_id = str(evt.get("parent_session_id") or "").strip() @@ -15510,96 +15395,8 @@ 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) - 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 evt.get("type") == "async_delegation": - from tools import async_delegation - - mark_completion_delivered = getattr( - async_delegation, "mark_completion_delivered", None, - ) - if mark_completion_delivered is not None: - try: - mark_completion_delivered(str(evt.get("delegation_id") or "")) - except Exception as exc: - # Adapter acceptance already happened. Retrying now - # would duplicate the turn; SQLite may replay after a - # restart, which is the honest at-least-once contract. - logger.warning( - "Could not acknowledge durable async completion %s: %s", - evt.get("delegation_id", "unknown"), - exc, - ) - return True - finally: - if identity is not None and not accepted: - with self._completion_delivery_lock: - self._completion_deliveries_inflight.discard(identity) def _enrich_async_delegation_routing(self, evt: dict) -> None: """Fill platform/chat_id/thread_id/chat_type on an async-delegation event. @@ -15662,11 +15459,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not synth_text: continue try: - delivered = await self._deliver_completion_notification(synth_text, evt) - if delivered is False: - _pr.completion_queue.put(evt) + 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 "")) 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) @@ -15732,12 +15528,8 @@ 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 @@ -15750,34 +15542,57 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _out = f"[… output truncated — showing last {len(_tail)} chars]\n{_tail}" else: _out = _raw - completion_evt = { + synth_text = format_process_notification({ "type": "completion", "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, - "message_id": message_id, - "started_at": getattr(session, "started_at", None), - "command": _command, + "command": session.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 - 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 + source = self._build_process_event_source({ + "session_id": session_id, + "session_key": session_key, + "platform": platform_name, + "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, + ) + 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) break # --- Normal text-only notification --- @@ -16738,8 +16553,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if url: return url.rstrip("/") cfg = _load_gateway_config() - url = (cfg.get("gateway") or {}).get("proxy_url") - url = (url or "").strip() + url = (cfg.get("gateway") or {}).get("proxy_url", "").strip() if url: return url.rstrip("/") return None @@ -20026,7 +19840,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew session_key or "?", exc_info=True, ) - next_message = await self._prepare_profile_scoped_inbound_message_text( + next_message = await self._prepare_inbound_message_text( event=pending_event, source=next_source, history=updated_history, diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 7714d3c8c08a..bf2187e740c4 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -5,7 +5,11 @@ onto the shared process_registry.completion_queue, the rich re-injection block formatting, capacity rejection, and crash handling. """ +import json +import os import queue +import subprocess +import sys import threading import time @@ -223,6 +227,84 @@ def test_completed_records_pruned_to_cap(): assert len(ad.list_async_delegations()) <= ad._MAX_RETAINED_COMPLETED +def test_completion_is_persisted_and_delivery_can_be_acknowledged(tmp_path, monkeypatch): + """A finished child remains pending on disk until its queue consumer acks it.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + dispatched = ad.dispatch_async_delegation( + goal="durable", context="ctx", toolsets=["terminal"], role="leaf", + model="m", session_key="owner", parent_session_id="parent", + runner=lambda: {"status": "completed", "summary": "survived"}, + ) + assert _drain_one() is not None + + restored = queue.Queue() + assert ad.restore_undelivered_completions(restored) == 1 + row = ad.get_durable_delegation(dispatched["delegation_id"]) + assert row["origin_session"] == "owner" + assert row["state"] == "completed" + assert row["result"]["summary"] == "survived" + assert row["delivery_state"] == "pending" + assert row["delivery_attempts"] >= 2 + + assert ad.mark_completion_delivered(dispatched["delegation_id"]) + assert ad.restore_undelivered_completions(queue.Queue()) == 0 + assert ad.get_durable_delegation(dispatched["delegation_id"])["delivery_state"] == "delivered" + + +def test_real_process_restart_restores_owned_completion_once(tmp_path): + """Real-import E2E: a fresh interpreter restores a prior process's result.""" + repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) + env = {**os.environ, "HERMES_HOME": str(tmp_path), "PYTHONPATH": repo} + producer = r''' +import time +from tools import async_delegation as ad +r = ad.dispatch_async_delegation( + goal="restart", context=None, toolsets=None, role="leaf", model="m", + session_key="owner-session", parent_session_id="durable-parent", + runner=lambda: {"status": "completed", "summary": "after restart"}, +) +deadline = time.time() + 5 +while ad.active_count() and time.time() < deadline: + time.sleep(.01) +print(r["delegation_id"]) +''' + first = subprocess.run( + [sys.executable, "-c", producer], cwd=repo, env=env, + text=True, capture_output=True, timeout=15, check=True, + ) + delegation_id = first.stdout.strip().splitlines()[-1] + + consumer = r''' +import json +from tools.process_registry import process_registry +evt = process_registry.completion_queue.get_nowait() +print(json.dumps(evt, sort_keys=True)) +''' + second = subprocess.run( + [sys.executable, "-c", consumer], cwd=repo, env=env, + text=True, capture_output=True, timeout=15, check=True, + ) + evt = json.loads(second.stdout.strip().splitlines()[-1]) + assert evt["delegation_id"] == delegation_id + assert evt["session_key"] == "owner-session" + assert evt["parent_session_id"] == "durable-parent" + assert evt["summary"] == "after restart" + + acker = f''' +from tools import async_delegation as ad +assert ad.mark_completion_delivered({delegation_id!r}) +''' + subprocess.run( + [sys.executable, "-c", acker], cwd=repo, env=env, + text=True, capture_output=True, timeout=15, check=True, + ) + probe = subprocess.run( + [sys.executable, "-c", "from tools.process_registry import process_registry; print(process_registry.completion_queue.qsize())"], + cwd=repo, env=env, text=True, capture_output=True, timeout=15, check=True, + ) + assert probe.stdout.strip().splitlines()[-1] == "0" + + # --------------------------------------------------------------------------- # Integration: delegate_task(background=True) routing # --------------------------------------------------------------------------- diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 9b4b0e9646ac..011b398c78c6 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -36,13 +36,16 @@ logic stays in one place. from __future__ import annotations +import json import logging +import sqlite3 import threading import time import uuid from concurrent.futures import ThreadPoolExecutor from typing import Any, Callable, Dict, List, Optional +from hermes_constants import get_hermes_home from tools.daemon_pool import DaemonThreadPoolExecutor from tools.thread_context import propagate_context_to_thread @@ -72,6 +75,133 @@ _records: Dict[str, Dict[str, Any]] = {} _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 +_DB_LOCK = threading.Lock() + + +def _db_path(): + return get_hermes_home() / "delegations.db" + + +def _connect() -> sqlite3.Connection: + path = _db_path() + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path, timeout=10) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute( + """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 + )""" + ) + return conn + + +def _persist_dispatch(record: Dict[str, Any]) -> None: + now = time.time() + 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)""", + (record["delegation_id"], record.get("session_key", ""), + record.get("origin_ui_session_id", ""), record.get("parent_session_id"), + record["dispatched_at"], now), + ) + cutoff = now - _DURABLE_RETENTION_SECONDS + 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,), + ) + + +def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None: + now = time.time() + with _DB_LOCK, _connect() as conn: + conn.execute( + """UPDATE async_delegations SET state=?, completed_at=?, updated_at=?, + event_json=?, result_json=?, delivery_state='pending' + WHERE delegation_id=?""", + (event.get("status", "completed"), event.get("completed_at", now), now, + json.dumps(event), json.dumps(result), event["delegation_id"]), + ) + + +def _note_delivery_attempt(delegation_id: str) -> None: + with _DB_LOCK, _connect() as conn: + conn.execute( + "UPDATE async_delegations SET delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=?", + (time.time(), delegation_id), + ) + + +def restore_undelivered_completions(target_queue) -> int: + """Enqueue durable pending completions as fresh turns after process start.""" + 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: + 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) + + +def mark_completion_delivered(delegation_id: str) -> bool: + """Atomically acknowledge successful injection of a durable completion.""" + now = time.time() + with _DB_LOCK, _connect() as conn: + cur = conn.execute( + """UPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=? + WHERE delegation_id=? AND delivery_state!='delivered'""", + (now, now, delegation_id), + ) + return cur.rowcount == 1 + + +def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: + with _DB_LOCK, _connect() as conn: + row = conn.execute( + """SELECT origin_session, state, dispatched_at, completed_at, + result_json, delivery_state, delivery_attempts + FROM async_delegations WHERE delegation_id=?""", (delegation_id,), + ).fetchone() + if row is None: + return None + return { + "delegation_id": delegation_id, "origin_session": row[0], "state": row[1], + "dispatched_at": row[2], "completed_at": row[3], + "result": json.loads(row[4]) if row[4] else None, + "delivery_state": row[5], "delivery_attempts": row[6], + } def _get_executor(max_workers: int) -> ThreadPoolExecutor: @@ -96,7 +226,7 @@ def _get_executor(max_workers: int) -> ThreadPoolExecutor: def active_count() -> int: """Number of async delegations currently running.""" with _records_lock: - return sum(1 for r in _records.values() if r.get("status") == "running") + return sum(1 for r in _records.values() if r.get("status") in {"running", "finalizing"}) def _new_delegation_id() -> str: @@ -206,6 +336,7 @@ def dispatch_async_delegation( } _records[delegation_id] = record + _persist_dispatch(record) executor = _get_executor(max_async_children) def _worker() -> None: @@ -252,14 +383,20 @@ def _finalize(delegation_id: str, result: Dict[str, Any], status: str) -> None: record = _records.get(delegation_id) if record is None: return - record["status"] = status + # Stay active until durable persistence and queue publication finish; + # otherwise process shutdown can kill this daemon worker in the narrow + # gap after status flips but before SQLite is committed. + record["status"] = "finalizing" record["completed_at"] = time.time() record["interrupt_fn"] = None # drop the closure; child is done - # Snapshot fields needed for the event while holding the lock. event_record = dict(record) - _prune_completed_locked() _push_completion_event(event_record, result, status) + with _records_lock: + record = _records.get(delegation_id) + if record is not None: + record["status"] = status + _prune_completed_locked() def _push_completion_event( @@ -309,8 +446,10 @@ def _push_completion_event( "completed_at": completed_at, "exit_reason": result.get("exit_reason"), } + _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; " @@ -393,6 +532,7 @@ def dispatch_async_delegation_batch( } _records[delegation_id] = record + _persist_dispatch(record) executor = _get_executor(max_async_children) def _worker() -> None: @@ -446,11 +586,10 @@ def _finalize_batch( record = _records.get(delegation_id) if record is None: return - record["status"] = status + record["status"] = "finalizing" record["completed_at"] = time.time() record["interrupt_fn"] = None event_record = dict(record) - _prune_completed_locked() try: from tools.process_registry import process_registry @@ -486,14 +625,22 @@ def _finalize_batch( "dispatched_at": dispatched_at, "completed_at": completed_at, } + _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; " "result lost: %s", delegation_id, exc, ) + finally: + with _records_lock: + record = _records.get(delegation_id) + if record is not None: + record["status"] = status + _prune_completed_locked() def list_async_delegations() -> List[Dict[str, Any]]: diff --git a/tools/process_registry.py b/tools/process_registry.py index f9ce491f6dc9..f7e6e8471b56 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -171,6 +171,13 @@ class ProcessRegistry: # gateway drain this after each agent turn to auto-trigger new turns. import queue as _queue_mod self.completion_queue: _queue_mod.Queue = _queue_mod.Queue() + # Rehydrate durable delegation completions only at registry startup. + # Consumers still inject them as fresh turns through this existing rail. + try: + from tools.async_delegation import restore_undelivered_completions + restore_undelivered_completions(self.completion_queue) + except Exception as exc: + logger.warning("Could not restore async delegation completions: %s", exc) # Track sessions whose completion was already consumed by the agent # via wait/log. Drain loops AND gateway/tui watchers skip notifications diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 8f72de91691b..5dc90b9f6f4b 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -8796,6 +8796,9 @@ def _notification_poller_loop( 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 "")) except Exception as exc: print( f"[tui_gateway] notification poller dispatch failed: " @@ -8848,6 +8851,9 @@ def _notification_poller_loop( 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 "")) except Exception as exc: print( f"[tui_gateway] notification poller dispatch failed: " @@ -9401,6 +9407,9 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None: 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 "")) except Exception as _n_exc: print( f"[tui_gateway] completion notification dispatch failed: "