mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-20 15:33:54 +00:00
fix(delegation): harden durable completion delivery
This commit is contained in:
teknium1
Teknium
parent
67f4e1b4a9
commit
d0e9a42cec
9 changed files with 696 additions and 153 deletions
20
cli.py
20
cli.py
|
|
@ -15189,10 +15189,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
|
|||
from tools.approval import get_current_session_key
|
||||
_drain_sk = get_current_session_key(default="")
|
||||
for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk):
|
||||
from tools.async_delegation import (
|
||||
claim_event_delivery, complete_event_delivery,
|
||||
)
|
||||
_claim = claim_event_delivery(_evt, "cli-idle")
|
||||
if _claim is None:
|
||||
continue
|
||||
self._pending_input.put(_synth)
|
||||
if _evt.get("type") == "async_delegation":
|
||||
from tools.async_delegation import mark_completion_delivered
|
||||
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
|
||||
complete_event_delivery(_evt, _claim)
|
||||
except Exception:
|
||||
pass
|
||||
continue
|
||||
|
|
@ -15354,10 +15358,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
|
|||
try:
|
||||
from tools.process_registry import process_registry
|
||||
for _evt, _synth in process_registry.drain_notifications():
|
||||
from tools.async_delegation import (
|
||||
claim_event_delivery, complete_event_delivery,
|
||||
)
|
||||
_claim = claim_event_delivery(_evt, "cli-post-turn")
|
||||
if _claim is None:
|
||||
continue
|
||||
self._pending_input.put(_synth)
|
||||
if _evt.get("type") == "async_delegation":
|
||||
from tools.async_delegation import mark_completion_delivered
|
||||
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
|
||||
complete_event_delivery(_evt, _claim)
|
||||
except Exception:
|
||||
pass # Non-fatal — don't break the main loop
|
||||
|
||||
|
|
|
|||
424
gateway/run.py
424
gateway/run.py
|
|
@ -1772,6 +1772,7 @@ from gateway.config import (
|
|||
load_gateway_config,
|
||||
)
|
||||
from gateway.session import (
|
||||
AsyncSessionStore,
|
||||
SessionStore,
|
||||
SessionSource,
|
||||
SessionContext,
|
||||
|
|
@ -2881,6 +2882,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
key, max_active_age=_bg_max_age_seconds,
|
||||
),
|
||||
)
|
||||
# One enforced loop-side boundary for the synchronous SessionStore.
|
||||
# Sync helpers keep using ``session_store`` directly; async gateway
|
||||
# handlers call this facade and await every operation.
|
||||
self._async_session_store = AsyncSessionStore(self.session_store)
|
||||
self.delivery_router = DeliveryRouter(self.config)
|
||||
self._running = False
|
||||
self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None
|
||||
|
|
@ -2974,6 +2979,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# cannot grow unbounded over a long-running gateway lifetime.
|
||||
self._session_sources: "OrderedDict[str, SessionSource]" = OrderedDict()
|
||||
self._session_sources_max = 512
|
||||
# Completion delivery is intentionally lifecycle-scoped. This closes
|
||||
# duplicate queue/watcher races inside one gateway without pretending
|
||||
# the adapter call and a persistence write can be exactly-once across
|
||||
# a process crash. Any durable async-delegation replay state remains
|
||||
# owned by tools.async_delegation, not a parallel gateway ledger.
|
||||
self._completion_delivery_lock = threading.Lock()
|
||||
self._completion_deliveries_inflight: set[tuple[str, str, object]] = set()
|
||||
self._completion_deliveries_delivered: "OrderedDict[tuple[str, str, object], None]" = OrderedDict()
|
||||
self._completion_delivery_retention = 2048
|
||||
|
||||
# Cache AIAgent instances per session to preserve prompt caching.
|
||||
# Without this, a new AIAgent is created per message, rebuilding the
|
||||
|
|
@ -4815,7 +4829,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"""Load reasoning effort from config.yaml.
|
||||
|
||||
Reads agent.reasoning_effort from config.yaml. Valid: "none",
|
||||
"minimal", "low", "medium", "high", "xhigh". Returns None to use
|
||||
"minimal", "low", "medium", "high", "xhigh", "max", "ultra". Returns None to use
|
||||
default (medium).
|
||||
"""
|
||||
from hermes_constants import parse_reasoning_effort
|
||||
|
|
@ -5202,7 +5216,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
except Exception:
|
||||
return False
|
||||
|
||||
def _session_has_compression_in_flight(self, session_key: str) -> bool:
|
||||
async def _session_has_compression_in_flight(self, session_key: str) -> bool:
|
||||
"""Return True when a compression lock is held for this session's id.
|
||||
|
||||
Context compression is interrupt-protected (#23975) but gateway
|
||||
|
|
@ -5210,28 +5224,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
the pre-rotation parent while compression is mid-flight, producing
|
||||
orphaned compression siblings (#56391). Callers demote interrupt to
|
||||
queue when this returns True.
|
||||
|
||||
Both blocking sources — the ``session_store`` lock + JSON load, and the
|
||||
SQLite ``get_compression_lock_holder`` SELECT — are offloaded to a
|
||||
worker thread so a large state.db never freezes the event loop (#5).
|
||||
"""
|
||||
session_store = getattr(self, "session_store", None)
|
||||
if not session_key or session_store is None:
|
||||
return False
|
||||
try:
|
||||
with session_store._lock: # noqa: SLF001 — snapshot entry under lock
|
||||
session_store._ensure_loaded_locked() # noqa: SLF001
|
||||
entry = session_store._entries.get(session_key) # noqa: SLF001
|
||||
session_id = getattr(entry, "session_id", None) if entry is not None else None
|
||||
if not session_id:
|
||||
return False
|
||||
session_id = await asyncio.to_thread(
|
||||
self._lookup_session_id_under_store_lock, session_store, session_key
|
||||
)
|
||||
except Exception:
|
||||
return False
|
||||
if not session_id:
|
||||
return False
|
||||
session_db = getattr(self, "_session_db", None)
|
||||
if session_db is None:
|
||||
return False
|
||||
db = getattr(session_db, "_db", session_db)
|
||||
raw_db = getattr(session_db, "_db", session_db)
|
||||
try:
|
||||
return bool(db.get_compression_lock_holder(str(session_id)))
|
||||
holder = await asyncio.to_thread(
|
||||
raw_db.get_compression_lock_holder, str(session_id)
|
||||
)
|
||||
return bool(holder)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _lookup_session_id_under_store_lock(session_store, session_key: str):
|
||||
"""Sync helper run in the thread pool: read session_id under the store lock."""
|
||||
# noqa: SLF001 — intentional private access; runs off the event loop.
|
||||
with session_store._lock: # noqa: SLF001
|
||||
session_store._ensure_loaded_locked() # noqa: SLF001
|
||||
entry = session_store._entries.get(session_key) # noqa: SLF001
|
||||
return getattr(entry, "session_id", None) if entry is not None else None
|
||||
|
||||
# Hard cap on per-session pending follow-ups for busy_input_mode=queue
|
||||
# (and the draining/steer-fallback/subagent-demotion paths that share
|
||||
# this entry point). Without a cap, a stuck agent + a rapid-fire user
|
||||
|
|
@ -5454,7 +5483,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
effective_mode = "queue"
|
||||
demoted_for_compression = (
|
||||
effective_mode == "interrupt"
|
||||
and self._session_has_compression_in_flight(session_key)
|
||||
and await self._session_has_compression_in_flight(session_key)
|
||||
)
|
||||
if demoted_for_compression:
|
||||
logger.info(
|
||||
|
|
@ -5743,7 +5772,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
source = None
|
||||
try:
|
||||
if getattr(self, "session_store", None) is not None:
|
||||
self.session_store._ensure_loaded()
|
||||
await self.async_session_store._ensure_loaded()
|
||||
entry = self.session_store._entries.get(session_key)
|
||||
source = getattr(entry, "origin", None) if entry else None
|
||||
except Exception as e:
|
||||
|
|
@ -7013,7 +7042,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
pass
|
||||
else:
|
||||
try:
|
||||
suspended = self.session_store.suspend_recently_active()
|
||||
suspended = await self.async_session_store.suspend_recently_active()
|
||||
if suspended:
|
||||
logger.info("Marked %d in-flight session(s) as resumable from previous run", suspended)
|
||||
except Exception as e:
|
||||
|
|
@ -7574,13 +7603,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Make sure there's an entry in the session_store for this key. If
|
||||
# the home channel has never been used, get_or_create_session
|
||||
# creates one; switch_session then re-points it.
|
||||
self.session_store.get_or_create_session(dest_source)
|
||||
await self.async_session_store.get_or_create_session(dest_source)
|
||||
|
||||
# Re-bind the destination key to the CLI session_id. switch_session
|
||||
# ends the prior session in SQLite and reopens the CLI session under
|
||||
# the new key. The CLI's transcript becomes the active one for the
|
||||
# gateway from this moment on.
|
||||
switched = self.session_store.switch_session(session_key, cli_session_id)
|
||||
switched = await self.async_session_store.switch_session(session_key, cli_session_id)
|
||||
if switched is None:
|
||||
raise RuntimeError(
|
||||
f"could not switch session key {session_key} → {cli_session_id}"
|
||||
|
|
@ -7657,13 +7686,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_MAX_FINALIZE_RETRIES = 3
|
||||
while self._running:
|
||||
try:
|
||||
self.session_store._ensure_loaded()
|
||||
await self.async_session_store._ensure_loaded()
|
||||
# Collect expired sessions first, then log a single summary.
|
||||
_expired_entries = []
|
||||
for key, entry in list(self.session_store._entries.items()):
|
||||
if entry.expiry_finalized:
|
||||
continue
|
||||
if not self.session_store._is_session_expired(entry):
|
||||
if not await self.async_session_store._is_session_expired(entry):
|
||||
continue
|
||||
_expired_entries.append((key, entry))
|
||||
|
||||
|
|
@ -7747,7 +7776,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# state.db (single write-path, #9006) — also drops
|
||||
# the persisted /model override, since finalization
|
||||
# is a conversation boundary.
|
||||
self.session_store.set_expiry_finalized(entry)
|
||||
await self.async_session_store.set_expiry_finalized(entry)
|
||||
logger.debug(
|
||||
"Session expiry finalized for %s",
|
||||
entry.session_id,
|
||||
|
|
@ -7762,7 +7791,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"Marking as finalized to prevent infinite retry loop.",
|
||||
failures, entry.session_id, e,
|
||||
)
|
||||
self.session_store.set_expiry_finalized(
|
||||
await self.async_session_store.set_expiry_finalized(
|
||||
entry, clear_model_override=False
|
||||
)
|
||||
_finalize_failures.pop(entry.session_id, None)
|
||||
|
|
@ -7815,7 +7844,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
getattr(self.config, "session_store_max_age_days", 0) or 0
|
||||
)
|
||||
if _max_age > 0:
|
||||
_pruned = self.session_store.prune_old_entries(_max_age)
|
||||
_pruned = await self.async_session_store.prune_old_entries(_max_age)
|
||||
if _pruned:
|
||||
logger.info(
|
||||
"SessionStore prune: dropped %d stale entries",
|
||||
|
|
@ -8149,7 +8178,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if _agent is _AGENT_PENDING_SENTINEL:
|
||||
continue
|
||||
try:
|
||||
self.session_store.mark_resume_pending(
|
||||
await self.async_session_store.mark_resume_pending(
|
||||
_sk,
|
||||
"restart_timeout" if self._restart_requested else "shutdown_timeout",
|
||||
)
|
||||
|
|
@ -8180,7 +8209,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
for _sk in _pre_drain_keys:
|
||||
if _sk not in self._running_agents:
|
||||
try:
|
||||
self.session_store.clear_resume_pending(_sk)
|
||||
await self.async_session_store.clear_resume_pending(_sk)
|
||||
except Exception as _e:
|
||||
logger.debug(
|
||||
"clear_resume_pending after drain failed for %s: %s",
|
||||
|
|
@ -8223,7 +8252,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if _agent is _AGENT_PENDING_SENTINEL:
|
||||
continue
|
||||
try:
|
||||
self.session_store.mark_resume_pending(_sk, _resume_reason)
|
||||
await self.async_session_store.mark_resume_pending(_sk, _resume_reason)
|
||||
except Exception as _e:
|
||||
logger.debug(
|
||||
"mark_resume_pending failed for %s: %s",
|
||||
|
|
@ -9596,7 +9625,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# the id out from under it, forking orphaned compression
|
||||
# siblings. Demote to queue semantics so the follow-up waits
|
||||
# for the in-flight compression + rotation to land.
|
||||
if self._session_has_compression_in_flight(_quick_key):
|
||||
if await self._session_has_compression_in_flight(_quick_key):
|
||||
logger.info(
|
||||
"PRIORITY interrupt demoted to queue for session %s "
|
||||
"because context compression is in flight (#56391)",
|
||||
|
|
@ -9637,7 +9666,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if isinstance(quick_commands, dict) and command in quick_commands:
|
||||
qcmd = quick_commands[command]
|
||||
if qcmd.get("type") == "alias":
|
||||
target = qcmd.get("target", "").strip()
|
||||
target = (qcmd.get("target") or "").strip()
|
||||
if target:
|
||||
target = target if target.startswith("/") else f"/{target}"
|
||||
target_command = target.lstrip("/")
|
||||
|
|
@ -10049,7 +10078,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
else:
|
||||
return f"Quick command '/{command}' has no command defined."
|
||||
elif qcmd.get("type") == "alias":
|
||||
target = qcmd.get("target", "").strip()
|
||||
target = (qcmd.get("target") or "").strip()
|
||||
if target:
|
||||
target = target if target.startswith("/") else f"/{target}"
|
||||
target_command = target.lstrip("/")
|
||||
|
|
@ -10303,7 +10332,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# on error. Let the user drive the next turn.
|
||||
if _final_text.strip():
|
||||
try:
|
||||
session_entry = self.session_store.get_or_create_session(source)
|
||||
session_entry = await self.async_session_store.get_or_create_session(source)
|
||||
except Exception:
|
||||
session_entry = None
|
||||
if session_entry is not None:
|
||||
|
|
@ -10619,8 +10648,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
from agent.model_metadata import get_model_context_length_async
|
||||
|
||||
_msg_cwd = os.environ.get("TERMINAL_CWD", os.path.expanduser("~"))
|
||||
_msg_runtime = _resolve_runtime_agent_kwargs()
|
||||
_msg_config_ctx = None
|
||||
_msg_cfg = None
|
||||
_msg_model_cfg = {}
|
||||
_msg_custom_providers = []
|
||||
try:
|
||||
_msg_cfg = _load_gateway_config()
|
||||
_msg_model_cfg = _msg_cfg.get("model", {})
|
||||
|
|
@ -10628,13 +10659,57 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_msg_raw_ctx = _msg_model_cfg.get("context_length")
|
||||
if _msg_raw_ctx is not None:
|
||||
_msg_config_ctx = int(_msg_raw_ctx)
|
||||
try:
|
||||
from hermes_cli.config import get_compatible_custom_providers
|
||||
|
||||
_msg_custom_providers = get_compatible_custom_providers(_msg_cfg)
|
||||
except Exception:
|
||||
_msg_custom_providers = _msg_cfg.get("custom_providers") or []
|
||||
except Exception:
|
||||
pass
|
||||
# Resolve the session's actual model/provider/base_url the
|
||||
# same way the hygiene compression block does (~11080).
|
||||
# GatewayRunner has no self._model/self._base_url attrs
|
||||
# (that was copy-pasted from HermesCLI, which does carry
|
||||
# self.model/self.base_url), so using them here always raised
|
||||
# AttributeError, silently caught below, meaning this feature
|
||||
# never ran.
|
||||
_msg_model, _msg_runtime = self._resolve_session_agent_runtime(
|
||||
source=source,
|
||||
session_key=session_key,
|
||||
user_config=_msg_cfg,
|
||||
)
|
||||
_msg_base_url = _msg_runtime.get("base_url") or ""
|
||||
# A global model.context_length belongs to the configured
|
||||
# model, not a session /model or channel override. Prefer a
|
||||
# matching per-custom-provider model limit when available.
|
||||
_msg_configured_model = (
|
||||
_msg_model_cfg.get("default") or _msg_model_cfg.get("model")
|
||||
if isinstance(_msg_model_cfg, dict)
|
||||
else _msg_model_cfg
|
||||
)
|
||||
if _msg_model != _msg_configured_model:
|
||||
_msg_config_ctx = None
|
||||
if _msg_custom_providers and _msg_base_url:
|
||||
try:
|
||||
from hermes_cli.config import get_custom_provider_context_length
|
||||
|
||||
_msg_custom_ctx = get_custom_provider_context_length(
|
||||
model=_msg_model,
|
||||
base_url=_msg_base_url,
|
||||
custom_providers=_msg_custom_providers,
|
||||
)
|
||||
if _msg_custom_ctx:
|
||||
_msg_config_ctx = _msg_custom_ctx
|
||||
except Exception:
|
||||
pass
|
||||
_msg_ctx_len = await get_model_context_length_async(
|
||||
self._model,
|
||||
base_url=self._base_url or _msg_runtime.get("base_url") or "",
|
||||
_msg_model,
|
||||
base_url=_msg_base_url,
|
||||
api_key=_msg_runtime.get("api_key") or "",
|
||||
config_context_length=_msg_config_ctx,
|
||||
provider=_msg_runtime.get("provider") or "",
|
||||
custom_providers=_msg_custom_providers,
|
||||
)
|
||||
_ctx_result = await preprocess_context_references_async(
|
||||
message_text,
|
||||
|
|
@ -10653,10 +10728,35 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if _ctx_result.expanded:
|
||||
message_text = _ctx_result.message
|
||||
except Exception as exc:
|
||||
logger.debug("@ context reference expansion failed: %s", exc)
|
||||
logger.warning("@ context reference expansion failed: %s", exc)
|
||||
logger.debug("@ context reference expansion failure detail", exc_info=True)
|
||||
|
||||
return message_text
|
||||
|
||||
async def _prepare_profile_scoped_inbound_message_text(
|
||||
self,
|
||||
*,
|
||||
event: MessageEvent,
|
||||
source: SessionSource,
|
||||
history: List[Dict[str, Any]],
|
||||
session_key: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
"""Run inbound preprocessing under the routed profile when multiplexed."""
|
||||
if getattr(getattr(self, "config", None), "multiplex_profiles", False):
|
||||
with _profile_runtime_scope(self._resolve_profile_home_for_source(source)):
|
||||
return await self._prepare_inbound_message_text(
|
||||
event=event,
|
||||
source=source,
|
||||
history=history,
|
||||
session_key=session_key,
|
||||
)
|
||||
return await self._prepare_inbound_message_text(
|
||||
event=event,
|
||||
source=source,
|
||||
history=history,
|
||||
session_key=session_key,
|
||||
)
|
||||
|
||||
def _consume_pending_native_image_paths(self, session_key: str) -> List[str]:
|
||||
pending_native = getattr(self, "_pending_native_image_paths_by_session", None)
|
||||
if not pending_native:
|
||||
|
|
@ -10684,6 +10784,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
@property
|
||||
def async_session_store(self) -> AsyncSessionStore:
|
||||
"""Return the single async facade for this runner's SessionStore."""
|
||||
facade = getattr(self, "_async_session_store", None)
|
||||
if facade is None or facade._store is not self.session_store:
|
||||
facade = AsyncSessionStore(self.session_store)
|
||||
self._async_session_store = facade
|
||||
return facade
|
||||
|
||||
def _get_cached_session_source(self, session_key: str):
|
||||
if not session_key:
|
||||
return None
|
||||
|
|
@ -10727,7 +10836,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
session_entry = self.session_store.get_or_create_session(source)
|
||||
session_entry = await self.async_session_store.get_or_create_session(source)
|
||||
session_key = session_entry.session_key
|
||||
pinned_session_id = str(
|
||||
(getattr(event, "metadata", None) or {}).get("gateway_session_id") or ""
|
||||
|
|
@ -10759,7 +10868,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
)
|
||||
return
|
||||
prior_session_id = session_entry.session_id
|
||||
switched = self.session_store.switch_session(session_key, pinned_session_id)
|
||||
switched = await self.async_session_store.switch_session(session_key, pinned_session_id)
|
||||
if switched is not None:
|
||||
session_entry = switched
|
||||
logger.info(
|
||||
|
|
@ -10809,7 +10918,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# lane session is ended cleanly. Mutating session_entry in
|
||||
# place here created a split-brain state where the JSON
|
||||
# index pointed at one id but code downstream used another.
|
||||
switched = self.session_store.switch_session(session_key, bound_session_id)
|
||||
switched = await self.async_session_store.switch_session(session_key, bound_session_id)
|
||||
if switched is not None:
|
||||
session_entry = switched
|
||||
# If the stored binding pointed at a parent, rewrite it to the
|
||||
|
|
@ -10997,7 +11106,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
logger.warning("[Gateway] Failed to auto-load skill(s) %s: %s", _skill_names, e)
|
||||
|
||||
# Load conversation history from transcript
|
||||
history = self.session_store.load_transcript(session_entry.session_id)
|
||||
history = await self.async_session_store.load_transcript(session_entry.session_id)
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# Session hygiene: auto-compress pathologically large transcripts
|
||||
|
|
@ -11260,7 +11369,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
)
|
||||
if _hyg_rotated:
|
||||
session_entry.session_id = _hyg_new_sid
|
||||
self.session_store._save()
|
||||
await self.async_session_store._save()
|
||||
await asyncio.to_thread(
|
||||
self._sync_telegram_topic_binding,
|
||||
source, session_entry,
|
||||
|
|
@ -11288,7 +11397,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# messages and replace them with only the compressed
|
||||
# summary (permanent data loss, #21301).
|
||||
if _hyg_rotated:
|
||||
self.session_store.rewrite_transcript(
|
||||
await self.async_session_store.rewrite_transcript(
|
||||
session_entry.session_id, _compressed
|
||||
)
|
||||
# Reset stored token count — transcript rewritten
|
||||
|
|
@ -11405,7 +11514,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
)
|
||||
|
||||
# First-message onboarding -- only on the very first interaction ever
|
||||
if not history and not self.session_store.has_any_sessions():
|
||||
if not history and not await self.async_session_store.has_any_sessions():
|
||||
# Default first-contact note: a brief self-introduction.
|
||||
_intro_note = (
|
||||
"\n\n[System note: This is the user's very first message ever. "
|
||||
|
|
@ -11490,7 +11599,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# attachments (documents, audio, etc.) are not sent to the vision
|
||||
# tool even when they appear in the same message.
|
||||
# -----------------------------------------------------------------
|
||||
message_text = await self._prepare_inbound_message_text(
|
||||
message_text = await self._prepare_profile_scoped_inbound_message_text(
|
||||
event=event,
|
||||
source=source,
|
||||
history=history,
|
||||
|
|
@ -11649,7 +11758,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if session_key and _should_clear_resume_pending_after_turn(agent_result):
|
||||
self._clear_restart_failure_count(session_key)
|
||||
try:
|
||||
self.session_store.clear_resume_pending(session_key)
|
||||
await self.async_session_store.clear_resume_pending(session_key)
|
||||
except Exception as _e:
|
||||
logger.debug(
|
||||
"clear_resume_pending failed for %s: %s",
|
||||
|
|
@ -11671,8 +11780,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if agent_result.get("session_id") and agent_result["session_id"] != session_entry.session_id:
|
||||
if session_entry.session_id == _run_start_session_id:
|
||||
session_entry.session_id = agent_result["session_id"]
|
||||
self.session_store._save()
|
||||
self.session_store._record_gateway_session_peer(
|
||||
await self.async_session_store._save()
|
||||
await self.async_session_store._record_gateway_session_peer(
|
||||
session_entry.session_id,
|
||||
session_key,
|
||||
source,
|
||||
|
|
@ -11870,7 +11979,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"Auto-resetting session %s after compression exhaustion.",
|
||||
session_entry.session_id,
|
||||
)
|
||||
new_entry = self.session_store.reset_session(session_key)
|
||||
new_entry = await self.async_session_store.reset_session(session_key)
|
||||
self._evict_cached_agent(session_key)
|
||||
self._session_model_overrides.pop(session_key, None)
|
||||
self._set_session_reasoning_override(session_key, None)
|
||||
|
|
@ -11914,7 +12023,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
pass # Skip all transcript writes — don't grow a broken session
|
||||
elif not history:
|
||||
tool_defs = agent_result.get("tools", [])
|
||||
self.session_store.append_to_transcript(
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id,
|
||||
{
|
||||
"role": "session_meta",
|
||||
|
|
@ -11968,7 +12077,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# after transient failures). #47237
|
||||
_skip_persist = (
|
||||
event.message_id
|
||||
and self.session_store.has_platform_message_id(
|
||||
and await self.async_session_store.has_platform_message_id(
|
||||
session_entry.session_id, str(event.message_id)
|
||||
)
|
||||
)
|
||||
|
|
@ -11979,7 +12088,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
event.message_id, session_entry.session_id,
|
||||
)
|
||||
else:
|
||||
self.session_store.append_to_transcript(
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id,
|
||||
_user_entry,
|
||||
skip_db=agent_persisted,
|
||||
|
|
@ -12005,13 +12114,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
}
|
||||
if event.message_id:
|
||||
_user_entry["message_id"] = str(event.message_id)
|
||||
self.session_store.append_to_transcript(
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id,
|
||||
_user_entry,
|
||||
skip_db=agent_persisted,
|
||||
)
|
||||
if response:
|
||||
self.session_store.append_to_transcript(
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id,
|
||||
{"role": "assistant", "content": response, "timestamp": ts},
|
||||
skip_db=agent_persisted,
|
||||
|
|
@ -12036,7 +12145,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
):
|
||||
entry["message_id"] = str(event.message_id)
|
||||
_user_msg_id_attached = True
|
||||
self.session_store.append_to_transcript(
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id, entry,
|
||||
skip_db=agent_persisted,
|
||||
)
|
||||
|
|
@ -12044,7 +12153,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Token counts and model are now persisted by the agent directly.
|
||||
# Keep only last_prompt_tokens here for context-window tracking and
|
||||
# compression decisions.
|
||||
self.session_store.update_session(
|
||||
await self.async_session_store.update_session(
|
||||
session_entry.session_key,
|
||||
last_prompt_tokens=agent_result.get("last_prompt_tokens", 0),
|
||||
)
|
||||
|
|
@ -12144,7 +12253,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if 'message_text' in locals() and message_text is not None and session_entry is not None:
|
||||
_already_persisted = False
|
||||
try:
|
||||
_recent_transcript = self.session_store.load_transcript(session_entry.session_id)
|
||||
_recent_transcript = await self.async_session_store.load_transcript(session_entry.session_id)
|
||||
except Exception:
|
||||
_recent_transcript = []
|
||||
for _msg in reversed(_recent_transcript[-10:]):
|
||||
|
|
@ -12172,7 +12281,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
}
|
||||
if getattr(event, "message_id", None):
|
||||
_user_entry["message_id"] = str(event.message_id)
|
||||
self.session_store.append_to_transcript(
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id,
|
||||
_user_entry,
|
||||
)
|
||||
|
|
@ -12627,7 +12736,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
except Exception:
|
||||
return 20
|
||||
|
||||
def _get_goal_manager_for_event(self, event: "MessageEvent"):
|
||||
async def _get_goal_manager_for_event(self, event: "MessageEvent"):
|
||||
"""Return a GoalManager bound to the session for this gateway event.
|
||||
|
||||
Returns ``(manager, session_entry)`` or ``(None, None)`` if the
|
||||
|
|
@ -12639,7 +12748,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
logger.debug("goal manager unavailable: %s", exc)
|
||||
return None, None
|
||||
try:
|
||||
session_entry = self.session_store.get_or_create_session(event.source)
|
||||
session_entry = await self.async_session_store.get_or_create_session(event.source)
|
||||
except Exception as exc:
|
||||
logger.debug("goal manager: session lookup failed: %s", exc)
|
||||
return None, None
|
||||
|
|
@ -14082,8 +14191,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"content": f"[IMPORTANT: MCP servers have been reloaded. {change_detail}{tool_summary}. The tool list for this conversation has been updated accordingly.]",
|
||||
}
|
||||
try:
|
||||
session_entry = self.session_store.get_or_create_session(event.source)
|
||||
self.session_store.append_to_transcript(
|
||||
session_entry = await self.async_session_store.get_or_create_session(event.source)
|
||||
await self.async_session_store.append_to_transcript(
|
||||
session_entry.session_id, reload_msg
|
||||
)
|
||||
except Exception:
|
||||
|
|
@ -15354,11 +15463,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
user_name=str(evt.get("user_name") or "").strip() or None,
|
||||
)
|
||||
|
||||
async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None:
|
||||
"""Inject a watch-pattern notification as a synthetic message event.
|
||||
async def _inject_watch_notification(
|
||||
self, synth_text: str, evt: dict,
|
||||
) -> Optional[bool]:
|
||||
"""Inject a watch/completion notification as a synthetic message event.
|
||||
|
||||
Routing must come from the queued watch event itself, not from whatever
|
||||
Routing must come from the queued event itself, not from whatever
|
||||
foreground message happened to be active when the queue was drained.
|
||||
Returns ``True`` after adapter acceptance, ``False`` after a retryable
|
||||
adapter failure, and ``None`` when the event has no gateway route. This
|
||||
is not a transactional boundary: a process crash after adapter
|
||||
acceptance can still cause durable at-least-once replay.
|
||||
"""
|
||||
source = self._build_process_event_source(evt)
|
||||
if not source:
|
||||
|
|
@ -15366,7 +15481,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"Dropping watch notification with no routing metadata for process %s",
|
||||
evt.get("session_id", "unknown"),
|
||||
)
|
||||
return
|
||||
return None
|
||||
platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform)
|
||||
adapter = None
|
||||
for p, a in self.adapters.items():
|
||||
|
|
@ -15374,7 +15489,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
adapter = a
|
||||
break
|
||||
if not adapter:
|
||||
return
|
||||
return None
|
||||
try:
|
||||
metadata = {}
|
||||
parent_session_id = str(evt.get("parent_session_id") or "").strip()
|
||||
|
|
@ -15395,8 +15510,118 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
source.thread_id,
|
||||
)
|
||||
await adapter.handle_message(synth_event)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("Watch notification injection error: %s", e)
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _completion_delivery_identity(evt: dict) -> Optional[tuple[str, str, object]]:
|
||||
"""Return a producer-stable identity when one is available.
|
||||
|
||||
Delegation UUIDs identify one producer completion. Process session IDs
|
||||
are normally unique too, but include the persisted spawn epoch so an
|
||||
explicitly reused ID represents a distinct process incarnation. Legacy
|
||||
process events without ``started_at`` are delivered without deduplication
|
||||
rather than risking suppression of a real completion.
|
||||
"""
|
||||
evt_type = str(evt.get("type") or "")
|
||||
if evt_type == "async_delegation":
|
||||
producer_id = str(evt.get("delegation_id") or "")
|
||||
return (evt_type, producer_id, "") if producer_id else None
|
||||
if evt_type == "completion":
|
||||
producer_id = str(evt.get("session_id") or "")
|
||||
started_at = evt.get("started_at")
|
||||
if producer_id and started_at is not None:
|
||||
return (evt_type, producer_id, started_at)
|
||||
return None
|
||||
|
||||
async def _deliver_completion_notification(
|
||||
self, synth_text: str, evt: dict,
|
||||
) -> Optional[bool]:
|
||||
"""Deliver once per live gateway, or return False for a retry.
|
||||
|
||||
``True`` means this caller reached adapter acceptance, ``False`` means
|
||||
injection failed and the claim was released for retry, and ``None``
|
||||
means either another same-lifecycle caller owns/delivered the producer
|
||||
event or the event has no gateway route. No cross-process exactly-once
|
||||
guarantee is claimed.
|
||||
"""
|
||||
identity = self._completion_delivery_identity(evt)
|
||||
durable_claim_id = ""
|
||||
durable_delegation_id = ""
|
||||
if evt.get("type") == "async_delegation":
|
||||
durable_delegation_id = str(evt.get("delegation_id") or "")
|
||||
if durable_delegation_id:
|
||||
try:
|
||||
from tools.async_delegation import claim_completion_delivery
|
||||
|
||||
durable_claim_id = f"gateway:{id(self)}:{__import__('uuid').uuid4().hex}"
|
||||
if not claim_completion_delivery(
|
||||
durable_delegation_id, durable_claim_id,
|
||||
):
|
||||
return None
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"Could not claim durable async completion %s: %s",
|
||||
durable_delegation_id, exc,
|
||||
)
|
||||
return False
|
||||
if identity is not None:
|
||||
with self._completion_delivery_lock:
|
||||
if (
|
||||
identity in self._completion_deliveries_inflight
|
||||
or identity in self._completion_deliveries_delivered
|
||||
):
|
||||
return None
|
||||
self._completion_deliveries_inflight.add(identity)
|
||||
|
||||
accepted = False
|
||||
try:
|
||||
injection_result = await self._inject_watch_notification(synth_text, evt)
|
||||
if injection_result is not True:
|
||||
return injection_result
|
||||
accepted = True
|
||||
|
||||
if identity is not None:
|
||||
with self._completion_delivery_lock:
|
||||
self._completion_deliveries_inflight.discard(identity)
|
||||
self._completion_deliveries_delivered[identity] = None
|
||||
while (
|
||||
len(self._completion_deliveries_delivered)
|
||||
> self._completion_delivery_retention
|
||||
):
|
||||
self._completion_deliveries_delivered.popitem(last=False)
|
||||
|
||||
# If the durable async-delegation producer branch is present, its
|
||||
# SQLite row remains the authoritative replay state. Acknowledge it
|
||||
# after adapter acceptance; this gateway keeps no parallel ledger.
|
||||
if durable_claim_id:
|
||||
try:
|
||||
from tools.async_delegation import complete_completion_delivery
|
||||
|
||||
complete_completion_delivery(
|
||||
durable_delegation_id, durable_claim_id,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"Could not acknowledge durable async completion %s: %s",
|
||||
durable_delegation_id, exc,
|
||||
)
|
||||
return True
|
||||
finally:
|
||||
if identity is not None and not accepted:
|
||||
with self._completion_delivery_lock:
|
||||
self._completion_deliveries_inflight.discard(identity)
|
||||
if durable_claim_id and not accepted:
|
||||
try:
|
||||
from tools.async_delegation import release_completion_delivery
|
||||
|
||||
release_completion_delivery(
|
||||
durable_delegation_id, durable_claim_id,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Could not release durable completion claim", exc_info=True)
|
||||
|
||||
def _enrich_async_delegation_routing(self, evt: dict) -> None:
|
||||
"""Fill platform/chat_id/thread_id/chat_type on an async-delegation event.
|
||||
|
|
@ -15459,10 +15684,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if not synth_text:
|
||||
continue
|
||||
try:
|
||||
await self._inject_watch_notification(synth_text, evt)
|
||||
from tools.async_delegation import mark_completion_delivered
|
||||
mark_completion_delivered(str(evt.get("delegation_id") or ""))
|
||||
delivered = await self._deliver_completion_notification(synth_text, evt)
|
||||
if delivered is False:
|
||||
_pr.completion_queue.put(evt)
|
||||
except Exception as e:
|
||||
_pr.completion_queue.put(evt)
|
||||
logger.error("Async delegation injection error: %s", e)
|
||||
except Exception as e:
|
||||
logger.debug("Async delegation watcher error: %s", e)
|
||||
|
|
@ -15528,8 +15754,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# (#10156) — a status check must not suppress this delivery turn.
|
||||
from tools.process_registry import format_process_notification, process_registry as _pr_check
|
||||
if agent_notify and not _pr_check.is_completion_consumed(session_id):
|
||||
from agent.redact import redact_terminal_output
|
||||
from tools.ansi_strip import strip_ansi
|
||||
_command = getattr(session, "command", "") or ""
|
||||
_raw = strip_ansi(session.output_buffer) if session.output_buffer else ""
|
||||
_raw = redact_terminal_output(_raw, _command)
|
||||
_command = _redact_gateway_user_facing_secrets(_command)
|
||||
# Truncate at line boundaries so notifications never start
|
||||
# mid-line (fixes #23284). Keep the last ~2000 chars but
|
||||
# snap to the nearest preceding newline, then prepend a
|
||||
|
|
@ -15542,57 +15772,34 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_out = f"[… output truncated — showing last {len(_tail)} chars]\n{_tail}"
|
||||
else:
|
||||
_out = _raw
|
||||
synth_text = format_process_notification({
|
||||
completion_evt = {
|
||||
"type": "completion",
|
||||
"session_id": session_id,
|
||||
"command": session.command,
|
||||
"exit_code": session.exit_code,
|
||||
"completion_reason": getattr(session, "completion_reason", "exited"),
|
||||
"termination_source": getattr(session, "termination_source", ""),
|
||||
"output": _out,
|
||||
})
|
||||
if not synth_text:
|
||||
break
|
||||
source = self._build_process_event_source({
|
||||
"session_id": session_id,
|
||||
"session_key": session_key,
|
||||
"platform": platform_name,
|
||||
"chat_type": watcher.get("chat_type", ""),
|
||||
"chat_id": chat_id,
|
||||
"thread_id": thread_id,
|
||||
"user_id": user_id,
|
||||
"user_name": user_name,
|
||||
})
|
||||
if not source:
|
||||
logger.warning(
|
||||
"Dropping completion notification with no routing metadata for process %s",
|
||||
session_id,
|
||||
)
|
||||
"message_id": message_id,
|
||||
"started_at": getattr(session, "started_at", None),
|
||||
"command": _command,
|
||||
"exit_code": session.exit_code,
|
||||
"completion_reason": getattr(session, "completion_reason", "exited"),
|
||||
"termination_source": getattr(session, "termination_source", ""),
|
||||
"output": _out,
|
||||
}
|
||||
synth_text = format_process_notification(completion_evt)
|
||||
if not synth_text:
|
||||
break
|
||||
|
||||
adapter = None
|
||||
for p, a in self.adapters.items():
|
||||
if p == source.platform:
|
||||
adapter = a
|
||||
break
|
||||
if adapter and source.chat_id:
|
||||
try:
|
||||
synth_event = MessageEvent(
|
||||
text=synth_text,
|
||||
message_type=MessageType.TEXT,
|
||||
source=source,
|
||||
internal=True,
|
||||
message_id=message_id,
|
||||
)
|
||||
logger.info(
|
||||
"Process %s finished — injecting agent notification for session %s chat=%s thread=%s",
|
||||
session_id,
|
||||
session_key,
|
||||
source.chat_id,
|
||||
source.thread_id,
|
||||
)
|
||||
await adapter.handle_message(synth_event)
|
||||
except Exception as e:
|
||||
logger.error("Agent notify injection error: %s", e)
|
||||
delivered = await self._deliver_completion_notification(
|
||||
synth_text, completion_evt,
|
||||
)
|
||||
if delivered is False:
|
||||
# The process remains terminal; retry after failed
|
||||
# adapter injection instead of suppressing the result.
|
||||
continue
|
||||
break
|
||||
|
||||
# --- Normal text-only notification ---
|
||||
|
|
@ -16553,7 +16760,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if url:
|
||||
return url.rstrip("/")
|
||||
cfg = _load_gateway_config()
|
||||
url = (cfg.get("gateway") or {}).get("proxy_url", "").strip()
|
||||
url = (cfg.get("gateway") or {}).get("proxy_url")
|
||||
url = (url or "").strip()
|
||||
if url:
|
||||
return url.rstrip("/")
|
||||
return None
|
||||
|
|
@ -19840,7 +20048,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
session_key or "?",
|
||||
exc_info=True,
|
||||
)
|
||||
next_message = await self._prepare_inbound_message_text(
|
||||
next_message = await self._prepare_profile_scoped_inbound_message_text(
|
||||
event=pending_event,
|
||||
source=next_source,
|
||||
history=updated_history,
|
||||
|
|
|
|||
|
|
@ -140,7 +140,7 @@ T = TypeVar("T")
|
|||
|
||||
DEFAULT_DB_PATH = get_hermes_home() / "state.db"
|
||||
|
||||
SCHEMA_VERSION = 20
|
||||
SCHEMA_VERSION = 21
|
||||
|
||||
# Cap on user-controlled FTS5 query input before regex/sanitizer processing.
|
||||
# Search queries do not need to be arbitrarily large, and bounding them keeps
|
||||
|
|
@ -827,6 +827,27 @@ CREATE TABLE IF NOT EXISTS compression_locks (
|
|||
expires_at REAL NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS async_delegations (
|
||||
delegation_id TEXT PRIMARY KEY,
|
||||
origin_session TEXT NOT NULL,
|
||||
origin_ui_session_id TEXT NOT NULL DEFAULT '',
|
||||
parent_session_id TEXT,
|
||||
state TEXT NOT NULL,
|
||||
dispatched_at REAL NOT NULL,
|
||||
completed_at REAL,
|
||||
updated_at REAL NOT NULL,
|
||||
event_json TEXT,
|
||||
result_json TEXT,
|
||||
delivery_state TEXT NOT NULL DEFAULT 'pending',
|
||||
delivery_attempts INTEGER NOT NULL DEFAULT 0,
|
||||
delivered_at REAL,
|
||||
owner_pid INTEGER,
|
||||
owner_started_at INTEGER,
|
||||
task_json TEXT,
|
||||
delivery_claim TEXT,
|
||||
delivery_claimed_at REAL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_source_id ON sessions(source, id);
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
|
||||
|
|
@ -835,6 +856,8 @@ CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestam
|
|||
CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model);
|
||||
CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery
|
||||
ON async_delegations(delivery_state, completed_at);
|
||||
"""
|
||||
|
||||
# Indexes that reference columns added in later schema versions must be
|
||||
|
|
|
|||
|
|
@ -175,8 +175,8 @@ def test_failed_async_injection_is_retried_and_only_success_is_acked(
|
|||
acknowledgements = []
|
||||
monkeypatch.setattr(
|
||||
async_delegation,
|
||||
"mark_completion_delivered",
|
||||
lambda delegation_id: acknowledgements.append(delegation_id) or True,
|
||||
"complete_completion_delivery",
|
||||
lambda delegation_id, _claim_id: acknowledgements.append(delegation_id) or True,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -244,7 +244,8 @@ def test_completion_is_persisted_and_delivery_can_be_acknowledged(tmp_path, monk
|
|||
assert row["state"] == "completed"
|
||||
assert row["result"]["summary"] == "survived"
|
||||
assert row["delivery_state"] == "pending"
|
||||
assert row["delivery_attempts"] >= 2
|
||||
# Queue publication/restoration is not a destination delivery attempt.
|
||||
assert row["delivery_attempts"] == 0
|
||||
|
||||
assert ad.mark_completion_delivered(dispatched["delegation_id"])
|
||||
assert ad.restore_undelivered_completions(queue.Queue()) == 0
|
||||
|
|
@ -305,6 +306,102 @@ assert ad.mark_completion_delivered({delegation_id!r})
|
|||
assert probe.stdout.strip().splitlines()[-1] == "0"
|
||||
|
||||
|
||||
def test_submit_failure_removes_durable_running_record(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
|
||||
class _BrokenExecutor:
|
||||
def submit(self, *_args, **_kwargs):
|
||||
raise RuntimeError("submit failed")
|
||||
|
||||
monkeypatch.setattr(ad, "_get_executor", lambda _max_workers: _BrokenExecutor())
|
||||
result = ad.dispatch_async_delegation(
|
||||
goal="never ran", context=None, toolsets=None, role="leaf", model="m",
|
||||
session_key="owner", runner=lambda: {},
|
||||
)
|
||||
|
||||
assert result["status"] == "rejected"
|
||||
with ad._DB_LOCK, ad._connect() as conn:
|
||||
assert conn.execute("SELECT COUNT(*) FROM async_delegations").fetchone()[0] == 0
|
||||
|
||||
|
||||
def test_pending_retention_prunes_delivered_before_undelivered(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
monkeypatch.setattr(ad, "_MAX_RETAINED_COMPLETED", 2)
|
||||
for index, delivery_state in enumerate(("pending", "delivered", "pending")):
|
||||
delegation_id = f"deleg_{index}"
|
||||
record = {
|
||||
"delegation_id": delegation_id,
|
||||
"session_key": "owner",
|
||||
"origin_ui_session_id": "",
|
||||
"parent_session_id": None,
|
||||
"dispatched_at": float(index + 1),
|
||||
}
|
||||
ad._persist_dispatch(record)
|
||||
ad._persist_completion(
|
||||
{
|
||||
"delegation_id": delegation_id,
|
||||
"status": "completed",
|
||||
"completed_at": float(index + 1),
|
||||
},
|
||||
{"status": "completed", "summary": delegation_id},
|
||||
)
|
||||
if delivery_state == "delivered":
|
||||
ad.mark_completion_delivered(delegation_id)
|
||||
|
||||
ad._prune_durable_records()
|
||||
|
||||
assert ad.get_durable_delegation("deleg_0") is not None
|
||||
assert ad.get_durable_delegation("deleg_1") is None
|
||||
assert ad.get_durable_delegation("deleg_2") is not None
|
||||
|
||||
|
||||
def test_recover_marks_abandoned_running_record_unknown(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
record = {
|
||||
"delegation_id": "deleg_abandoned",
|
||||
"session_key": "owner",
|
||||
"origin_ui_session_id": "",
|
||||
"parent_session_id": None,
|
||||
"dispatched_at": 1.0,
|
||||
}
|
||||
ad._persist_dispatch(record)
|
||||
with ad._DB_LOCK, ad._connect() as conn:
|
||||
conn.execute(
|
||||
"UPDATE async_delegations SET owner_pid=?, owner_started_at=NULL WHERE delegation_id=?",
|
||||
(99999999, "deleg_abandoned"),
|
||||
)
|
||||
|
||||
assert ad.recover_abandoned_delegations() == 1
|
||||
durable = ad.get_durable_delegation("deleg_abandoned")
|
||||
assert durable["state"] == "unknown"
|
||||
assert durable["delivery_state"] == "pending"
|
||||
restored = queue.Queue()
|
||||
assert ad.restore_undelivered_completions(restored) == 1
|
||||
assert restored.get_nowait()["status"] == "unknown"
|
||||
|
||||
|
||||
def test_durable_delivery_claim_is_exclusive_and_retryable(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
record = {
|
||||
"delegation_id": "deleg_claim", "session_key": "owner",
|
||||
"origin_ui_session_id": "", "parent_session_id": None,
|
||||
"dispatched_at": 1.0,
|
||||
}
|
||||
ad._persist_dispatch(record)
|
||||
ad._persist_completion(
|
||||
{"delegation_id": "deleg_claim", "status": "completed", "completed_at": 2.0},
|
||||
{"status": "completed", "summary": "done"},
|
||||
)
|
||||
|
||||
assert ad.claim_completion_delivery("deleg_claim", "consumer-a")
|
||||
assert not ad.claim_completion_delivery("deleg_claim", "consumer-b")
|
||||
assert ad.release_completion_delivery("deleg_claim", "consumer-a")
|
||||
assert ad.claim_completion_delivery("deleg_claim", "consumer-b")
|
||||
assert ad.complete_completion_delivery("deleg_claim", "consumer-b")
|
||||
assert not ad.claim_completion_delivery("deleg_claim", "consumer-c")
|
||||
assert ad.get_durable_delegation("deleg_claim")["delivery_state"] == "delivered"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration: delegate_task(background=True) routing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -76,11 +76,12 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3
|
|||
# How many completed records to retain for status queries before pruning.
|
||||
_MAX_RETAINED_COMPLETED = 50
|
||||
_DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60
|
||||
_MAX_DURABLE_PENDING = 1000
|
||||
_DB_LOCK = threading.Lock()
|
||||
|
||||
|
||||
def _db_path():
|
||||
return get_hermes_home() / "delegations.db"
|
||||
return get_hermes_home() / "state.db"
|
||||
|
||||
|
||||
def _connect() -> sqlite3.Connection:
|
||||
|
|
@ -102,40 +103,97 @@ def _connect() -> sqlite3.Connection:
|
|||
result_json TEXT,
|
||||
delivery_state TEXT NOT NULL DEFAULT 'pending',
|
||||
delivery_attempts INTEGER NOT NULL DEFAULT 0,
|
||||
delivered_at REAL
|
||||
delivered_at REAL,
|
||||
owner_pid INTEGER,
|
||||
owner_started_at INTEGER,
|
||||
task_json TEXT,
|
||||
delivery_claim TEXT,
|
||||
delivery_claimed_at REAL
|
||||
)"""
|
||||
)
|
||||
columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")}
|
||||
for name, sql_type in (
|
||||
("owner_pid", "INTEGER"),
|
||||
("owner_started_at", "INTEGER"),
|
||||
("task_json", "TEXT"),
|
||||
("delivery_claim", "TEXT"),
|
||||
("delivery_claimed_at", "REAL"),
|
||||
):
|
||||
if name not in columns:
|
||||
conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}")
|
||||
return conn
|
||||
|
||||
|
||||
def _persist_dispatch(record: Dict[str, Any]) -> None:
|
||||
now = time.time()
|
||||
try:
|
||||
from gateway.status import get_process_start_time
|
||||
owner_started_at = get_process_start_time(__import__("os").getpid())
|
||||
except Exception:
|
||||
owner_started_at = None
|
||||
task_payload = {
|
||||
key: record.get(key)
|
||||
for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch")
|
||||
if key in record
|
||||
}
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
conn.execute(
|
||||
"""INSERT OR REPLACE INTO async_delegations
|
||||
(delegation_id, origin_session, origin_ui_session_id,
|
||||
parent_session_id, state, dispatched_at, updated_at,
|
||||
delivery_state, delivery_attempts)
|
||||
VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0)""",
|
||||
delivery_state, delivery_attempts, owner_pid,
|
||||
owner_started_at, task_json)
|
||||
VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""",
|
||||
(record["delegation_id"], record.get("session_key", ""),
|
||||
record.get("origin_ui_session_id", ""), record.get("parent_session_id"),
|
||||
record["dispatched_at"], now),
|
||||
record["dispatched_at"], now, __import__("os").getpid(),
|
||||
owner_started_at, json.dumps(task_payload)),
|
||||
)
|
||||
cutoff = now - _DURABLE_RETENTION_SECONDS
|
||||
_prune_durable_records()
|
||||
|
||||
|
||||
def _delete_durable_delegation(delegation_id: str) -> None:
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
conn.execute("DELETE FROM async_delegations WHERE delegation_id=?", (delegation_id,))
|
||||
|
||||
|
||||
def _prune_durable_records() -> None:
|
||||
"""Bound terminal history, preferring delivered records for deletion."""
|
||||
now = time.time()
|
||||
cutoff = now - _DURABLE_RETENTION_SECONDS
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
conn.execute(
|
||||
"""DELETE FROM async_delegations WHERE delegation_id IN (
|
||||
SELECT delegation_id FROM async_delegations
|
||||
WHERE (delivery_state = 'delivered' AND updated_at < ?)
|
||||
OR (completed_at IS NOT NULL AND completed_at < ?)
|
||||
ORDER BY updated_at ASC
|
||||
)""", (cutoff, cutoff),
|
||||
)
|
||||
conn.execute(
|
||||
"""DELETE FROM async_delegations WHERE delegation_id IN (
|
||||
SELECT delegation_id FROM async_delegations
|
||||
WHERE state != 'running' ORDER BY updated_at DESC LIMIT -1 OFFSET ?
|
||||
)""", (_MAX_RETAINED_COMPLETED,),
|
||||
"DELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?",
|
||||
(cutoff,),
|
||||
)
|
||||
terminal_count = conn.execute(
|
||||
"SELECT COUNT(*) FROM async_delegations WHERE state NOT IN ('running','finalizing')"
|
||||
).fetchone()[0]
|
||||
excess = max(0, terminal_count - _MAX_RETAINED_COMPLETED)
|
||||
if excess:
|
||||
conn.execute(
|
||||
"""DELETE FROM async_delegations WHERE delegation_id IN (
|
||||
SELECT delegation_id FROM async_delegations
|
||||
WHERE state NOT IN ('running','finalizing')
|
||||
ORDER BY CASE delivery_state WHEN 'delivered' THEN 0 ELSE 1 END,
|
||||
updated_at ASC LIMIT ?
|
||||
)""",
|
||||
(excess,),
|
||||
)
|
||||
pending_count = conn.execute(
|
||||
"""SELECT COUNT(*) FROM async_delegations
|
||||
WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'"""
|
||||
).fetchone()[0]
|
||||
overflow = max(0, pending_count - _MAX_DURABLE_PENDING)
|
||||
if overflow:
|
||||
conn.execute(
|
||||
"""DELETE FROM async_delegations WHERE delegation_id IN (
|
||||
SELECT delegation_id FROM async_delegations
|
||||
WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'
|
||||
ORDER BY updated_at ASC LIMIT ?
|
||||
)""",
|
||||
(overflow,),
|
||||
)
|
||||
|
||||
|
||||
def _persist_completion(event: Dict[str, Any], result: Dict[str, Any]) -> None:
|
||||
|
|
@ -158,20 +216,64 @@ def _note_delivery_attempt(delegation_id: str) -> None:
|
|||
)
|
||||
|
||||
|
||||
def recover_abandoned_delegations() -> int:
|
||||
"""Classify records whose owning process disappeared as outcome unknown."""
|
||||
try:
|
||||
from gateway.status import _pid_exists, get_process_start_time
|
||||
except Exception:
|
||||
return 0
|
||||
now = time.time()
|
||||
recovered = 0
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
rows = conn.execute(
|
||||
"""SELECT delegation_id, origin_session, origin_ui_session_id,
|
||||
parent_session_id, dispatched_at, owner_pid,
|
||||
owner_started_at, task_json
|
||||
FROM async_delegations WHERE state IN ('running','finalizing')"""
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json = row
|
||||
live = False
|
||||
if pid:
|
||||
live = _pid_exists(int(pid))
|
||||
if live and started is not None:
|
||||
live = get_process_start_time(int(pid)) == int(started)
|
||||
if live:
|
||||
continue
|
||||
task = json.loads(task_json or "{}")
|
||||
event = {
|
||||
"type": "async_delegation", "delegation_id": delegation_id,
|
||||
"session_key": session_key, "origin_ui_session_id": origin_ui,
|
||||
"parent_session_id": parent_id, "goal": task.get("goal", ""),
|
||||
"goals": task.get("goals"), "context": task.get("context"),
|
||||
"toolsets": task.get("toolsets"), "role": task.get("role"),
|
||||
"model": task.get("model"), "is_batch": bool(task.get("is_batch")),
|
||||
"status": "unknown", "summary": None,
|
||||
"error": "Delegation owner exited before recording a terminal result; outcome unknown.",
|
||||
"dispatched_at": dispatched_at, "completed_at": now,
|
||||
}
|
||||
result = {"status": "unknown", "summary": None, "error": event["error"]}
|
||||
conn.execute(
|
||||
"""UPDATE async_delegations SET state='unknown', completed_at=?,
|
||||
updated_at=?, event_json=?, result_json=?, delivery_state='pending'
|
||||
WHERE delegation_id=?""",
|
||||
(now, now, json.dumps(event), json.dumps(result), delegation_id),
|
||||
)
|
||||
recovered += 1
|
||||
return recovered
|
||||
|
||||
|
||||
def restore_undelivered_completions(target_queue) -> int:
|
||||
"""Enqueue durable pending completions as fresh turns after process start."""
|
||||
recover_abandoned_delegations()
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
rows = conn.execute(
|
||||
"""SELECT delegation_id, event_json FROM async_delegations
|
||||
WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL
|
||||
ORDER BY completed_at, delegation_id"""
|
||||
).fetchall()
|
||||
for delegation_id, payload in rows:
|
||||
for _delegation_id, payload in rows:
|
||||
target_queue.put(json.loads(payload))
|
||||
conn.execute(
|
||||
"UPDATE async_delegations SET delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=?",
|
||||
(time.time(), delegation_id),
|
||||
)
|
||||
return len(rows)
|
||||
|
||||
|
||||
|
|
@ -187,6 +289,75 @@ def mark_completion_delivered(delegation_id: str) -> bool:
|
|||
return cur.rowcount == 1
|
||||
|
||||
|
||||
def claim_completion_delivery(delegation_id: str, claim_id: str) -> bool:
|
||||
"""Claim one pending completion across competing consumers/processes."""
|
||||
now = time.time()
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT delivery_state FROM async_delegations WHERE delegation_id=?",
|
||||
(delegation_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return True # legacy event created before durable dispatch
|
||||
cur = conn.execute(
|
||||
"""UPDATE async_delegations SET delivery_claim=?, delivery_claimed_at=?,
|
||||
delivery_attempts=delivery_attempts+1, updated_at=?
|
||||
WHERE delegation_id=? AND delivery_state='pending'
|
||||
AND (delivery_claim IS NULL OR delivery_claimed_at < ?)""",
|
||||
(claim_id, now, now, delegation_id, now - 300),
|
||||
)
|
||||
return cur.rowcount == 1
|
||||
|
||||
|
||||
def claim_event_delivery(evt: Dict[str, Any], consumer: str) -> Optional[str]:
|
||||
"""Claim a durable delegation event; non-durable events need no token."""
|
||||
if evt.get("type") != "async_delegation":
|
||||
return ""
|
||||
delegation_id = str(evt.get("delegation_id") or "")
|
||||
if not delegation_id:
|
||||
return ""
|
||||
claim_id = f"{consumer}:{__import__('os').getpid()}:{uuid.uuid4().hex}"
|
||||
return claim_id if claim_completion_delivery(delegation_id, claim_id) else None
|
||||
|
||||
|
||||
def release_completion_delivery(delegation_id: str, claim_id: str) -> bool:
|
||||
"""Release a failed delivery claim so another consumer may retry."""
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
cur = conn.execute(
|
||||
"""UPDATE async_delegations SET delivery_claim=NULL,
|
||||
delivery_claimed_at=NULL, updated_at=?
|
||||
WHERE delegation_id=? AND delivery_state='pending'
|
||||
AND delivery_claim=?""",
|
||||
(time.time(), delegation_id, claim_id),
|
||||
)
|
||||
return cur.rowcount == 1
|
||||
|
||||
|
||||
def complete_completion_delivery(delegation_id: str, claim_id: str) -> bool:
|
||||
"""Acknowledge acceptance for the consumer holding this claim."""
|
||||
now = time.time()
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
cur = conn.execute(
|
||||
"""UPDATE async_delegations SET delivery_state='delivered',
|
||||
delivered_at=?, updated_at=?, delivery_claim=NULL,
|
||||
delivery_claimed_at=NULL
|
||||
WHERE delegation_id=? AND delivery_state='pending'
|
||||
AND delivery_claim=?""",
|
||||
(now, now, delegation_id, claim_id),
|
||||
)
|
||||
return cur.rowcount == 1
|
||||
|
||||
|
||||
def complete_event_delivery(evt: Dict[str, Any], claim_id: str) -> None:
|
||||
if claim_id and evt.get("type") == "async_delegation":
|
||||
complete_completion_delivery(str(evt.get("delegation_id") or ""), claim_id)
|
||||
|
||||
|
||||
def release_event_delivery(evt: Dict[str, Any], claim_id: str) -> None:
|
||||
if claim_id and evt.get("type") == "async_delegation":
|
||||
release_completion_delivery(str(evt.get("delegation_id") or ""), claim_id)
|
||||
|
||||
|
||||
def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
|
||||
with _DB_LOCK, _connect() as conn:
|
||||
row = conn.execute(
|
||||
|
|
@ -365,6 +536,7 @@ def dispatch_async_delegation(
|
|||
except Exception as exc: # pragma: no cover — pool submit failure is rare
|
||||
with _records_lock:
|
||||
_records.pop(delegation_id, None)
|
||||
_delete_durable_delegation(delegation_id)
|
||||
return {
|
||||
"status": "rejected",
|
||||
"error": f"Failed to schedule async delegation: {exc}",
|
||||
|
|
@ -449,7 +621,6 @@ def _push_completion_event(
|
|||
_persist_completion(evt, result)
|
||||
try:
|
||||
process_registry.completion_queue.put(evt)
|
||||
_note_delivery_attempt(str(record.get("delegation_id") or ""))
|
||||
except Exception as exc: # pragma: no cover
|
||||
logger.error(
|
||||
"Async delegation %s: failed to enqueue completion event; "
|
||||
|
|
@ -566,6 +737,7 @@ def dispatch_async_delegation_batch(
|
|||
except Exception as exc: # pragma: no cover
|
||||
with _records_lock:
|
||||
_records.pop(delegation_id, None)
|
||||
_delete_durable_delegation(delegation_id)
|
||||
return {
|
||||
"status": "rejected",
|
||||
"error": f"Failed to schedule async delegation batch: {exc}",
|
||||
|
|
@ -628,7 +800,6 @@ def _finalize_batch(
|
|||
_persist_completion(evt, combined)
|
||||
try:
|
||||
process_registry.completion_queue.put(evt)
|
||||
_note_delivery_attempt(delegation_id)
|
||||
except Exception as exc: # pragma: no cover
|
||||
logger.error(
|
||||
"Async delegation batch %s: failed to enqueue completion event; "
|
||||
|
|
|
|||
|
|
@ -8793,13 +8793,18 @@ def _notification_poller_loop(
|
|||
continue
|
||||
|
||||
rid = f"__notif__{int(time.time() * 1000)}"
|
||||
from tools.async_delegation import (
|
||||
claim_event_delivery, complete_event_delivery, release_event_delivery,
|
||||
)
|
||||
_claim = claim_event_delivery(evt, "tui-poller")
|
||||
if _claim is None:
|
||||
continue
|
||||
try:
|
||||
_emit("message.start", sid)
|
||||
_run_prompt_submit(rid, sid, session, text)
|
||||
if evt.get("type") == "async_delegation":
|
||||
from tools.async_delegation import mark_completion_delivered
|
||||
mark_completion_delivered(str(evt.get("delegation_id") or ""))
|
||||
complete_event_delivery(evt, _claim)
|
||||
except Exception as exc:
|
||||
release_event_delivery(evt, _claim)
|
||||
print(
|
||||
f"[tui_gateway] notification poller dispatch failed: "
|
||||
f"{type(exc).__name__}: {exc}",
|
||||
|
|
@ -8848,13 +8853,18 @@ def _notification_poller_loop(
|
|||
session["running"] = True
|
||||
|
||||
rid = f"__notif__{int(time.time() * 1000)}"
|
||||
from tools.async_delegation import (
|
||||
claim_event_delivery, complete_event_delivery, release_event_delivery,
|
||||
)
|
||||
_claim = claim_event_delivery(evt, "tui-poller")
|
||||
if _claim is None:
|
||||
continue
|
||||
try:
|
||||
_emit("message.start", sid)
|
||||
_run_prompt_submit(rid, sid, session, text)
|
||||
if evt.get("type") == "async_delegation":
|
||||
from tools.async_delegation import mark_completion_delivered
|
||||
mark_completion_delivered(str(evt.get("delegation_id") or ""))
|
||||
complete_event_delivery(evt, _claim)
|
||||
except Exception as exc:
|
||||
release_event_delivery(evt, _claim)
|
||||
print(
|
||||
f"[tui_gateway] notification poller dispatch failed: "
|
||||
f"{type(exc).__name__}: {exc}",
|
||||
|
|
@ -9404,13 +9414,18 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
|
|||
process_registry.completion_queue.put(_evt)
|
||||
break
|
||||
session["running"] = True
|
||||
from tools.async_delegation import (
|
||||
claim_event_delivery, complete_event_delivery, release_event_delivery,
|
||||
)
|
||||
_claim = claim_event_delivery(_evt, "tui-post-turn")
|
||||
if _claim is None:
|
||||
continue
|
||||
try:
|
||||
_emit("message.start", sid)
|
||||
_run_prompt_submit(rid, sid, session, synth)
|
||||
if _evt.get("type") == "async_delegation":
|
||||
from tools.async_delegation import mark_completion_delivered
|
||||
mark_completion_delivered(str(_evt.get("delegation_id") or ""))
|
||||
complete_event_delivery(_evt, _claim)
|
||||
except Exception as _n_exc:
|
||||
release_event_delivery(_evt, _claim)
|
||||
print(
|
||||
f"[tui_gateway] completion notification dispatch failed: "
|
||||
f"{type(_n_exc).__name__}: {_n_exc}",
|
||||
|
|
|
|||
|
|
@ -129,6 +129,21 @@ When you provide a `tasks` array, subagents run in **parallel** using a thread p
|
|||
|
||||
Single-task delegation runs directly without thread pool overhead.
|
||||
|
||||
### Durable background completions
|
||||
|
||||
When a background delegation finishes, Hermes stores its completion event in
|
||||
the active profile's `state.db` before publishing it to the normal fresh-turn
|
||||
queue. If Hermes restarts after completion but before delivery, the pending
|
||||
event is restored and routed through the same ownership checks. Competing
|
||||
consumers use a durable claim, so only the consumer that successfully accepts
|
||||
the synthetic turn acknowledges delivery; failed attempts release the claim for
|
||||
retry.
|
||||
|
||||
This does not resume child execution after a crash. A delegation whose owner
|
||||
process disappears while it is still running is recorded as `unknown`, because
|
||||
Hermes cannot prove whether its external side effects happened. Pending and
|
||||
delivered records are bounded and profile-local.
|
||||
|
||||
## Model Override
|
||||
|
||||
You can configure a different model for subagents via `config.yaml` — useful for delegating simple tasks to cheaper/faster models:
|
||||
|
|
|
|||
|
|
@ -127,7 +127,13 @@ delegate_task(
|
|||
- **结果排序：** 结果按任务索引排序，与输入顺序一致，不受完成顺序影响
|
||||
- **中断传播：** 中断父智能体（例如发送新消息）会中断所有活跃的子智能体
|
||||
|
||||
单任务委派直接运行，无线程池开销。
|
||||
单任务委派直接运行，不会产生线程池开销。
|
||||
|
||||
### 持久化后台完成事件
|
||||
|
||||
后台委派完成后，Hermes 会先把完成事件写入当前 profile 的 `state.db`，再发布到正常的新轮次队列。如果 Hermes 在完成后、交付前重启，待处理事件会被恢复，并继续经过相同的所有权检查。多个消费者通过持久化 claim 竞争；只有成功接收合成轮次的消费者会确认交付，失败尝试会释放 claim 以便重试。
|
||||
|
||||
这不会在崩溃后恢复子智能体执行。如果委派仍在运行时其所有者进程消失，Hermes 会将其记录为 `unknown`，因为无法证明外部副作用是否已经发生。待处理和已交付记录都有界，并按 profile 隔离。
|
||||
|
||||
## 模型覆盖
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue