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

feat(delegation): persist background completions

Browse source
This commit is contained in:
teknium1 2026-07-09 18:56:14 -07:00 • committed by Teknium
parent 94a7705bdd
commit 67f4e1b4a9
6 changed files with 366 additions and 301 deletions
Download patch file Download diff file Expand all files Collapse all files

6
cli.py
View file

@ -15190,6 +15190,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
_drain_sk = get_current_session_key(default="") _drain_sk = get_current_session_key(default="")
for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk): for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk):
self._pending_input.put(_synth) 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: except Exception:
pass pass
continue continue
@ -15352,6 +15355,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
from tools.process_registry import process_registry from tools.process_registry import process_registry
for _evt, _synth in process_registry.drain_notifications(): for _evt, _synth in process_registry.drain_notifications():
self._pending_input.put(_synth) 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: except Exception:
pass # Non-fatal — don't break the main loop pass # Non-fatal — don't break the main loop

404
gateway/run.py
View file

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

82
tests/tools/test_async_delegation.py
View file

@ -5,7 +5,11 @@ onto the shared process_registry.completion_queue, the rich re-injection block
formatting, capacity rejection, and crash handling. formatting, capacity rejection, and crash handling.
""" """
import json
import os
import queue import queue
import subprocess
import sys
import threading import threading
import time import time
@ -223,6 +227,84 @@ def test_completed_records_pruned_to_cap():
assert len(ad.list_async_delegations()) <= ad._MAX_RETAINED_COMPLETED 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 # Integration: delegate_task(background=True) routing
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

159
tools/async_delegation.py
View file

@ -36,13 +36,16 @@ logic stays in one place.
from __future__ import annotations from __future__ import annotations
import json
import logging import logging
import sqlite3
import threading import threading
import time import time
import uuid import uuid
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
from hermes_constants import get_hermes_home
from tools.daemon_pool import DaemonThreadPoolExecutor from tools.daemon_pool import DaemonThreadPoolExecutor
from tools.thread_context import propagate_context_to_thread from tools.thread_context import propagate_context_to_thread
@ -72,6 +75,133 @@ _records: Dict[str, Dict[str, Any]] = {}
_DEFAULT_MAX_ASYNC_CHILDREN = 3 _DEFAULT_MAX_ASYNC_CHILDREN = 3
# How many completed records to retain for status queries before pruning. # How many completed records to retain for status queries before pruning.
_MAX_RETAINED_COMPLETED = 50 _MAX_RETAINED_COMPLETED = 50
_DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60
_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: def _get_executor(max_workers: int) -> ThreadPoolExecutor:
@ -96,7 +226,7 @@ def _get_executor(max_workers: int) -> ThreadPoolExecutor:
def active_count() -> int: def active_count() -> int:
"""Number of async delegations currently running.""" """Number of async delegations currently running."""
with _records_lock: 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: def _new_delegation_id() -> str:
@ -206,6 +336,7 @@ def dispatch_async_delegation(
} }
_records[delegation_id] = record _records[delegation_id] = record
_persist_dispatch(record)
executor = _get_executor(max_async_children) executor = _get_executor(max_async_children)
def _worker() -> None: def _worker() -> None:
@ -252,14 +383,20 @@ def _finalize(delegation_id: str, result: Dict[str, Any], status: str) -> None:
record = _records.get(delegation_id) record = _records.get(delegation_id)
if record is None: if record is None:
return 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["completed_at"] = time.time()
record["interrupt_fn"] = None # drop the closure; child is done record["interrupt_fn"] = None # drop the closure; child is done
# Snapshot fields needed for the event while holding the lock.
event_record = dict(record) event_record = dict(record)
_prune_completed_locked()
_push_completion_event(event_record, result, status) _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( def _push_completion_event(
@ -309,8 +446,10 @@ def _push_completion_event(
"completed_at": completed_at, "completed_at": completed_at,
"exit_reason": result.get("exit_reason"), "exit_reason": result.get("exit_reason"),
} }
_persist_completion(evt, result)
try: try:
process_registry.completion_queue.put(evt) process_registry.completion_queue.put(evt)
_note_delivery_attempt(str(record.get("delegation_id") or ""))
except Exception as exc: # pragma: no cover except Exception as exc: # pragma: no cover
logger.error( logger.error(
"Async delegation %s: failed to enqueue completion event; " "Async delegation %s: failed to enqueue completion event; "
@ -393,6 +532,7 @@ def dispatch_async_delegation_batch(
} }
_records[delegation_id] = record _records[delegation_id] = record
_persist_dispatch(record)
executor = _get_executor(max_async_children) executor = _get_executor(max_async_children)
def _worker() -> None: def _worker() -> None:
@ -446,11 +586,10 @@ def _finalize_batch(
record = _records.get(delegation_id) record = _records.get(delegation_id)
if record is None: if record is None:
return return
record["status"] = status record["status"] = "finalizing"
record["completed_at"] = time.time() record["completed_at"] = time.time()
record["interrupt_fn"] = None record["interrupt_fn"] = None
event_record = dict(record) event_record = dict(record)
_prune_completed_locked()
try: try:
from tools.process_registry import process_registry from tools.process_registry import process_registry
@ -486,14 +625,22 @@ def _finalize_batch(
"dispatched_at": dispatched_at, "dispatched_at": dispatched_at,
"completed_at": completed_at, "completed_at": completed_at,
} }
_persist_completion(evt, combined)
try: try:
process_registry.completion_queue.put(evt) process_registry.completion_queue.put(evt)
_note_delivery_attempt(delegation_id)
except Exception as exc: # pragma: no cover except Exception as exc: # pragma: no cover
logger.error( logger.error(
"Async delegation batch %s: failed to enqueue completion event; " "Async delegation batch %s: failed to enqueue completion event; "
"result lost: %s", "result lost: %s",
delegation_id, exc, 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]]: def list_async_delegations() -> List[Dict[str, Any]]:

7
tools/process_registry.py
View file

@ -171,6 +171,13 @@ class ProcessRegistry:
# gateway drain this after each agent turn to auto-trigger new turns. # gateway drain this after each agent turn to auto-trigger new turns.
import queue as _queue_mod import queue as _queue_mod
self.completion_queue: _queue_mod.Queue = _queue_mod.Queue() 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 # Track sessions whose completion was already consumed by the agent
# via wait/log. Drain loops AND gateway/tui watchers skip notifications # via wait/log. Drain loops AND gateway/tui watchers skip notifications

9
tui_gateway/server.py
View file

@ -8796,6 +8796,9 @@ def _notification_poller_loop(
try: try:
_emit("message.start", sid) _emit("message.start", sid)
_run_prompt_submit(rid, sid, session, text) _run_prompt_submit(rid, sid, session, text)
if evt.get("type") == "async_delegation":
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(evt.get("delegation_id") or ""))
except Exception as exc: except Exception as exc:
print( print(
f"[tui_gateway] notification poller dispatch failed: " f"[tui_gateway] notification poller dispatch failed: "
@ -8848,6 +8851,9 @@ def _notification_poller_loop(
try: try:
_emit("message.start", sid) _emit("message.start", sid)
_run_prompt_submit(rid, sid, session, text) _run_prompt_submit(rid, sid, session, text)
if evt.get("type") == "async_delegation":
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(evt.get("delegation_id") or ""))
except Exception as exc: except Exception as exc:
print( print(
f"[tui_gateway] notification poller dispatch failed: " f"[tui_gateway] notification poller dispatch failed: "
@ -9401,6 +9407,9 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
try: try:
_emit("message.start", sid) _emit("message.start", sid)
_run_prompt_submit(rid, sid, session, synth) _run_prompt_submit(rid, sid, session, synth)
if _evt.get("type") == "async_delegation":
from tools.async_delegation import mark_completion_delivered
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
except Exception as _n_exc: except Exception as _n_exc:
print( print(
f"[tui_gateway] completion notification dispatch failed: " f"[tui_gateway] completion notification dispatch failed: "