diff --git a/CHANGELOG.md b/CHANGELOG.md index 77cda584..84f45a12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ ### Fixed +- **Async delegation (`delegate_task`) results now wake the exact browser tab that started them, exactly once, even across a WebUI restart.** Two problems on the async-completion delivery path are fixed together. (1) *Misroute:* a detached completion was routed by a mutable session-key index, so a result could wake the wrong session; it now routes by the immutable `origin_ui_session_id` return address captured from the commissioning turn, which is authoritative over the index on both delivery paths (the background wakeup and the next-turn drain). (2) *Duplicate on restart:* the completion is now delivered through a durable claim/complete/release lifecycle and acknowledged only after the wakeup turn is accepted, so the restart-restore sweep no longer re-enqueues an already-delivered result. Every failure path (formatter, routing, turn-start, ACK) releases the claim and keeps the result retryable rather than dropping it, and an unrouteable async event is handed to a bounded retry instead of being silently stranded. Thanks @carlotestor (#6185) and @sysophelper-droid (#6159). (#6185, #6159, #6002, #6225) + - **The Artifacts tab now shows each artifact's file name first, with the directory path secondary.** Artifact rows previously rendered the full path on one line, so the file name (the part you scan for) was truncated at the end of a long path. The name now renders prominently with the directory shown smaller and muted beneath it. Thanks @webtecnica. (#6161) - **A session row's streaming indicator now reflects that session's own activity, not a child session's.** The sidebar "streaming" state was derived from a combined flag that was also true when a *child* (subagent) session was streaming, so a row could show as busy on its child's behalf. It now keys off the session's own streaming state. Thanks @webtecnica. (#6165) diff --git a/api/background_process.py b/api/background_process.py index e3269aa7..8cf76dfd 100644 --- a/api/background_process.py +++ b/api/background_process.py @@ -48,6 +48,16 @@ import time import uuid from typing import Any, Optional +from api.process_event_utils import ( + ASYNC_DELIVERY_ROUTING_RETRY_SECONDS, + claim_async_delegation_delivery, + complete_async_delegation_delivery, + completion_delivery_id, + release_async_delegation_delivery, + requeue_async_delegation_event, + schedule_async_delegation_claim_retry, +) + logger = logging.getLogger(__name__) _DRAIN_THREAD: Optional[threading.Thread] = None @@ -541,9 +551,9 @@ def _build_payload(evt: dict, session_id: str) -> dict: ``(session_id, event_id)`` to dedupe across reconnects. """ # ProcessRegistry completion events use the field name ``session_id`` for - # the process id. Alias it locally before exposing it as payload ``task_id`` - # to avoid confusing that wire-format name with the WebUI session id. - process_id = str(evt.get("session_id") or "") + # the process id. Async delegation completions carry ``delegation_id`` + # instead; expose either stable delivery id as payload ``task_id``. + process_id = completion_delivery_id(evt) payload: dict[str, Any] = { "session_id": str(session_id), "task_id": process_id, @@ -822,7 +832,12 @@ def _mark_registry_completion_consumed(process_id: str) -> None: # registry, pre-Option-1 spawns) it is a PURE PASS-THROUGH — it never # suppresses a legitimate Option Z wakeup on uncertainty (Option Z must keep # working). -_ENV_IMMUNE_OWNER_ATTRS = ("spawn_session_id", "owner_session_id", "turn_session_id") +_ENV_IMMUNE_OWNER_ATTRS = ( + "origin_ui_session_id", # modern hermes-agent exact browser-tab return address + "spawn_session_id", + "owner_session_id", + "turn_session_id", +) def _env_immune_spawn_owner(proc_session) -> str: @@ -872,6 +887,211 @@ def _resolve_wakeup_target( return owner +def _requeue_async_delegation_event( + process_registry, + evt: dict, + *, + claim=None, + delay: float = 0.5, +) -> bool: + """Retry without enqueueing new work once drain shutdown has started.""" + completion_queue = getattr(process_registry, "completion_queue", None) + return requeue_async_delegation_event( + evt, + completion_queue, + delay=delay, + stop_event=_DRAIN_STOP, + durable=(bool(getattr(claim, "durable", False)) if claim is not None else None), + ) + + +def _retry_unmapped_async_delegation_event(process_registry, evt: dict) -> None: + """Retry durable routing, or one bounded best-effort legacy routing pass.""" + completion_queue = getattr(process_registry, "completion_queue", None) + if schedule_async_delegation_claim_retry( + evt, + completion_queue, + delay=ASYNC_DELIVERY_ROUTING_RETRY_SECONDS, + ): + return + if evt.get("_webui_routing_retry_attempted"): + return + retry_evt = dict(evt) + retry_evt["_webui_routing_retry_attempted"] = True + _requeue_async_delegation_event( + process_registry, + retry_evt, + delay=ASYNC_DELIVERY_ROUTING_RETRY_SECONDS, + ) + + +def _record_async_delegation_accepted( + evt: dict, + *, + session_id: str, + claim, +) -> None: + """ACK durable delivery and publish live-view state after turn acceptance.""" + + complete_async_delegation_delivery(evt, claim) + payload = _build_payload(evt, session_id) + try: + _emit_bg_task_complete_events_coalesced(session_id, payload) + except Exception: + logger.debug( + "async delegation live-view emit failed for session %s", + session_id, + exc_info=True, + ) + + +def _start_async_delegation_wakeup_turn( + session_id: str, + wakeup_prompt: str, + *, + delegation_id: str, + evt: dict, + claim, + process_registry, +) -> None: + """Start one autonomous delegation turn and ACK only after acceptance.""" + + def _runner() -> None: + try: + from api.routes import start_session_turn + + resp = start_session_turn( + session_id, + wakeup_prompt, + source="process_wakeup", + ) + raw_status = (resp or {}).get("_status") + if raw_status is None: + status = 200 if (resp or {}).get("stream_id") else 500 + else: + status = int(raw_status) + if 200 <= status < 300: + _record_async_delegation_accepted( + evt, + session_id=session_id, + claim=claim, + ) + logger.info( + "async delegation wakeup turn accepted for session %s " + "(stream_id=%s)", + session_id, + (resp or {}).get("stream_id"), + ) + return + + release_async_delegation_delivery(evt, claim) + _requeue_async_delegation_event(process_registry, evt, claim=claim) + if status == 409 and (resp or {}).get("error") == "process_wakeup_paused": + logger.info( + "async delegation wakeup paused for session %s; delivery remains retryable", + session_id, + ) + else: + logger.debug( + "async delegation wakeup not accepted for session %s: " + "status=%s err=%r; requeued", + session_id, + status, + (resp or {}).get("error"), + ) + except Exception: + release_async_delegation_delivery(evt, claim) + _requeue_async_delegation_event(process_registry, evt, claim=claim) + logger.warning( + "async delegation wakeup turn failed for session %s; requeued", + session_id, + exc_info=True, + ) + + threading.Thread( + target=_runner, + name=f"hermes-webui-delegation-wakeup-{str(session_id)[:8]}", + daemon=True, + ).start() + + +def _process_async_delegation_event( + evt: dict, + *, + session_id: str, + delegation_id: str, + process_registry, +) -> None: + """Claim and route one async completion without private registry markers.""" + + try: + claim = claim_async_delegation_delivery(evt, "webui-background") + except Exception: + _requeue_async_delegation_event(process_registry, evt) + return + if claim is None: + completion_queue = getattr(process_registry, "completion_queue", None) + schedule_async_delegation_claim_retry(evt, completion_queue) + return + + try: + wakeup_prompt_raw = format_wakeup_prompt(evt) + wakeup_prompt = wakeup_prompt_raw.strip() if wakeup_prompt_raw else "" + if not wakeup_prompt: + raise RuntimeError("async delegation completion could not be formatted") + + # Do not persist async results in the process-local deferred list. If a + # foreground turn owns the session, release the durable claim and retry + # from the shared queue; the core record therefore remains restart-safe. + if _session_has_active_turn(session_id): + release_async_delegation_delivery(evt, claim) + _requeue_async_delegation_event(process_registry, evt, claim=claim) + return + + _start_async_delegation_wakeup_turn( + session_id, + wakeup_prompt, + delegation_id=delegation_id, + evt=evt, + claim=claim, + process_registry=process_registry, + ) + except Exception: + release_async_delegation_delivery(evt, claim) + _requeue_async_delegation_event(process_registry, evt, claim=claim) + logger.warning( + "server-side async delegation dispatch failed for session %s", + session_id, + exc_info=True, + ) + + +def _resolve_completion_target( + *, + session_key_resolved_sid: str, + origin_ui_session_id: str, +) -> str: + """Return the WebUI session that owns a detached completion event. + + Modern Hermes Agent events carry ``origin_ui_session_id`` as an exact, + immutable return address captured from the commissioning browser turn. + It is authoritative over the mutable/legacy session-key index. Older + Agent events omit it and retain the existing session-key fallback. + """ + resolved = str(session_key_resolved_sid or "") + owner = str(origin_ui_session_id or "") + if not owner: + return resolved + if resolved and resolved != owner: + logger.error( + "cross-session completion route BLOCKED: session_key resolved to %r " + "but exact origin_ui_session_id is %r; routing to the exact owner", + resolved, + owner, + ) + return owner + + def _process_one(evt: dict) -> None: """Route a single completion_queue event to the matching WebUI session.""" from api import config as _cfg @@ -887,55 +1107,59 @@ def _process_one(evt: dict) -> None: except Exception: _process_registry = None - process_id = str(evt.get("session_id") or "") + process_id = completion_delivery_id(evt) session_key = str(evt.get("session_key") or "") + origin_ui_session_id = str(evt.get("origin_ui_session_id") or "") # Root-cause fix (t_0f447014): the notify_on_complete completion event - # enqueued by ProcessRegistry._move_to_finished() carries NO "session_key" - # field — only the watch_match enqueue includes one. Without it the old - # `evt.get("session_key") or process_id` fell back to the process id - # ("proc_xxxx"), which is never a PROCESS_SESSION_INDEX key (only - # webui_session_id -> webui_session_id is registered at chat-start), so - # every wakeup was silently dropped here and the frontend never POSTed an - # ack. Recover the spawn-time session_key from the process registry's - # ProcessSession: the terminal tool captured it synchronously at spawn - # (while the turn's env was active), so it survives the turn-end env - # restore and is the WebUI session_id for WebUI-spawned processes. - if not session_key and process_id: + # enqueued by ProcessRegistry._move_to_finished() historically carried NO + # "session_key" field — only the watch_match enqueue included one. Without + # it the old `evt.get("session_key") or process_id` fell back to the + # process id ("proc_xxxx"), which is never a PROCESS_SESSION_INDEX key. + # Recover spawn-time routing metadata from the process registry. + if process_id and (not session_key or not origin_ui_session_id): try: if _process_registry is not None: _ps = _process_registry.get(process_id) - if _ps is not None and getattr(_ps, "session_key", ""): - session_key = str(_ps.session_key) + if _ps is not None: + if not session_key and getattr(_ps, "session_key", ""): + session_key = str(_ps.session_key) + if not origin_ui_session_id: + origin_ui_session_id = ( + str(getattr(_ps, "origin_ui_session_id", "") or "") + or str(getattr(_ps, "spawn_session_id", "") or "") + ) except Exception: logger.debug( - "session_key recovery from process registry failed for %r", + "session ownership recovery from process registry failed for %r", process_id, exc_info=True, ) - if not session_key: + if not session_key and not origin_ui_session_id: logger.debug( - "process_complete drop: no recoverable session_key for process_id=%r", + "process_complete drop: no recoverable session_key or exact UI owner " + "for process_id=%r", process_id, ) + if evt.get("type") == "async_delegation": + _retry_unmapped_async_delegation_event(_process_registry, evt) return - with _cfg.PROCESS_SESSION_INDEX_LOCK: - session_id = _cfg.PROCESS_SESSION_INDEX.get(session_key) - if not session_id: + session_id = "" + if session_key: + with _cfg.PROCESS_SESSION_INDEX_LOCK: + session_id = _cfg.PROCESS_SESSION_INDEX.get(session_key) or "" + if not session_id and not origin_ui_session_id: # No mapping — could be a cron/gateway process that uses the same - # registry but a non-WebUI session_key. Ignore. + # registry but a non-WebUI session_key. Durable delegation events stay + # pending and are retried because their WebUI ownership mapping can be + # registered shortly after process restore. logger.debug("process_complete drop: no session mapping for key=%r", session_key) + if evt.get("type") == "async_delegation": + _retry_unmapped_async_delegation_event(_process_registry, evt) return # ── xsession wakeup misroute defense-in-depth (Option 3) ────────────── - # session_id above came from PROCESS_SESSION_INDEX.get(session_key), and - # session_key was captured by the terminal tool from the (historically - # racy) process-global env at spawn. Option 1 binds the per-turn identity - # to a contextvar so that capture is no longer racy — but as an INDEPENDENT - # safety net, cross-check the resolved target against the env-immune - # spawn owner (when the core ProcessSession exposes one). On a positive - # mismatch this re-routes the wakeup (and the live-view emit + dedupe - # markers below) to the TRUE owner instead of waking the wrong session. - # Pure pass-through when no env-immune owner is available (today's core, - # cron/CLI procs, pre-Option-1 spawns) — never suppresses a valid wakeup. + # First retain the process-registry spawn-owner cross-check for legacy + # terminal events. Then apply origin_ui_session_id as the final authority; + # that exact owner is shared by process and async-delegation completions. try: _ps_xs = _process_registry.get(process_id) if (_process_registry is not None and process_id) else None except Exception: @@ -945,6 +1169,35 @@ def _process_one(evt: dict) -> None: session_key_resolved_sid=session_id, proc_session=_ps_xs, ) + # origin_ui_session_id is the exact, immutable return address; it is the + # FINAL routing authority over the (mutable) session-key/Option-3 result. + session_id = _resolve_completion_target( + session_key_resolved_sid=session_id, + origin_ui_session_id=origin_ui_session_id, + ) + if not session_id: + logger.debug("process_complete drop: completion target resolved empty") + # An async delegation event that resolves empty here must NOT silently + # return: no durable claim was taken, so the core row stays pending and + # the restart-restore sweep would re-deliver it forever. Route it into + # the bounded retry instead (arms a durable retry / one best-effort + # legacy requeue), so it stays retryable without a spurious ACK. + if evt.get("type") == "async_delegation": + _retry_unmapped_async_delegation_event(_process_registry, evt) + return + # ── THE SEAM: async delegations take the durable-claim delivery path, + # routed to the origin-resolved session. The claim/complete/release + # lifecycle (keyed on the immutable delegation_id) is the sole dedupe + + # restart-safety authority for async events, so they early-return before + # the terminal-process idempotency/emit/Option-Z machinery below. ── + if evt.get("type") == "async_delegation": + _process_async_delegation_event( + evt, + session_id=session_id, + delegation_id=process_id, + process_registry=_process_registry, + ) + return # ── Idempotency vs the REAL merged upstream #2279 (shared dedupe key) ── # The real merged #2279 next-turn drain # (api/streaming._drain_webui_process_notifications) dedupes ONLY via @@ -1056,7 +1309,7 @@ def _process_one(evt: dict) -> None: ) -def record_deferred_wakeup(session_id: str, process_id: str, wakeup_prompt: str) -> None: +def record_deferred_wakeup(session_id: str, process_id: str, wakeup_prompt: str) -> bool: """Persist a deferred process-completion wakeup for later redelivery. Called from ``_process_one`` when a completion arrives while a turn is @@ -1068,10 +1321,11 @@ def record_deferred_wakeup(session_id: str, process_id: str, wakeup_prompt: str) Idempotent per process_id: if the same process_id is already queued for this session (kill_process racing the reader thread), it is not appended - twice. Best-effort — never raises into the drain loop. + twice. Returns whether the prompt is safely queued; never raises into the + drain loop. """ if not session_id or not wakeup_prompt: - return + return False from api import config as _cfg try: @@ -1080,14 +1334,16 @@ def record_deferred_wakeup(session_id: str, process_id: str, wakeup_prompt: str) if process_id and any( e.get("process_id") == process_id for e in entries ): - return + return True entries.append( {"process_id": process_id, "wakeup_prompt": wakeup_prompt} ) + return True except Exception: logger.debug( "record_deferred_wakeup failed for session %s", session_id, exc_info=True ) + return False def claim_deferred_wakeups(session_id: str) -> list[dict]: diff --git a/api/process_event_utils.py b/api/process_event_utils.py new file mode 100644 index 00000000..85cdc768 --- /dev/null +++ b/api/process_event_utils.py @@ -0,0 +1,374 @@ +"""Shared helpers for WebUI completion/delegation delivery.""" +from __future__ import annotations + +from collections import OrderedDict +from dataclasses import dataclass +import logging +import threading +import time +from typing import Any + +logger = logging.getLogger(__name__) + +# Older Hermes Agent builds do not expose durable claim/complete/release APIs. +# Keep their in-process compatibility dedupe bounded so long-lived WebUI +# processes cannot retain every delegation id forever. +LEGACY_ASYNC_DELIVERY_DEDUPE_MAX = 1024 +ASYNC_DELIVERY_CLAIM_RETRY_SECONDS = 301.0 +ASYNC_DELIVERY_ROUTING_RETRY_SECONDS = 5.0 +_LEGACY_ASYNC_DELIVERY_LOCK = threading.Lock() +_LEGACY_ASYNC_DELIVERY_IDS: OrderedDict[str, None] = OrderedDict() +_ASYNC_DELIVERY_RETRY_LOCK = threading.Lock() +_ASYNC_DELIVERY_RETRY_TIMER: threading.Timer | None = None +_ASYNC_DELIVERY_RETRY_DEADLINE = 0.0 +_ASYNC_DELIVERY_RETRY_QUEUE: Any = None +_ASYNC_DELIVERY_RETRY_GENERATION = 0 + + +@dataclass(frozen=True) +class AsyncDelegationDeliveryClaim: + """Opaque ownership token for one WebUI async-delegation consumer.""" + + delegation_id: str + claim_id: str + durable: bool + + +def completion_delivery_id(evt: Any) -> str: + """Return the stable WebUI delivery/dedupe id for a completion event. + + Terminal background-process events use ``session_id`` for the process id. + Async ``delegate_task`` completions carry ``delegation_id`` instead, so both + WebUI delivery paths must key those events by ``delegation_id``. + """ + if not isinstance(evt, dict): + return "" + if evt.get("type") == "async_delegation": + return str( + evt.get("delegation_id") + or evt.get("session_id") + or evt.get("task_id") + or "" + ).strip() + return str(evt.get("session_id") or "").strip() + + +def _claim_bounded_local(delegation_id: str) -> bool: + with _LEGACY_ASYNC_DELIVERY_LOCK: + if delegation_id in _LEGACY_ASYNC_DELIVERY_IDS: + return False + _LEGACY_ASYNC_DELIVERY_IDS[delegation_id] = None + while len(_LEGACY_ASYNC_DELIVERY_IDS) > LEGACY_ASYNC_DELIVERY_DEDUPE_MAX: + _LEGACY_ASYNC_DELIVERY_IDS.popitem(last=False) + return True + + +def _release_bounded_local(delegation_id: str) -> None: + with _LEGACY_ASYNC_DELIVERY_LOCK: + _LEGACY_ASYNC_DELIVERY_IDS.pop(delegation_id, None) + + +def _arm_async_delegation_restore_sweep(completion_queue: Any, delay: float) -> bool: + """Arm one process-wide durable restore sweep at the earliest deadline. + + The durable database is the backlog. Keeping one shared timer avoids both + one-thread-per-event growth and lossy eviction of individual retry entries. + The sweep restores every still-pending record; atomic claims suppress races + and delivered rows are excluded by the core query. + """ + global _ASYNC_DELIVERY_RETRY_TIMER + global _ASYNC_DELIVERY_RETRY_DEADLINE + global _ASYNC_DELIVERY_RETRY_QUEUE + global _ASYNC_DELIVERY_RETRY_GENERATION + + if completion_queue is None: + return False + retry_delay = max(0.0, float(delay)) + deadline = time.monotonic() + retry_delay + + with _ASYNC_DELIVERY_RETRY_LOCK: + if ( + _ASYNC_DELIVERY_RETRY_TIMER is not None + and deadline >= _ASYNC_DELIVERY_RETRY_DEADLINE + ): + return True + previous = _ASYNC_DELIVERY_RETRY_TIMER + if previous is not None: + previous.cancel() + _ASYNC_DELIVERY_RETRY_GENERATION += 1 + generation = _ASYNC_DELIVERY_RETRY_GENERATION + _ASYNC_DELIVERY_RETRY_DEADLINE = deadline + _ASYNC_DELIVERY_RETRY_QUEUE = completion_queue + + def _restore() -> None: + global _ASYNC_DELIVERY_RETRY_TIMER + global _ASYNC_DELIVERY_RETRY_DEADLINE + global _ASYNC_DELIVERY_RETRY_QUEUE + + with _ASYNC_DELIVERY_RETRY_LOCK: + if generation != _ASYNC_DELIVERY_RETRY_GENERATION: + return + target_queue = _ASYNC_DELIVERY_RETRY_QUEUE + _ASYNC_DELIVERY_RETRY_TIMER = None + _ASYNC_DELIVERY_RETRY_DEADLINE = 0.0 + _ASYNC_DELIVERY_RETRY_QUEUE = None + try: + from tools.async_delegation import restore_undelivered_completions + + restore_undelivered_completions(target_queue) + except Exception: + logger.warning( + "Failed to restore pending async delegations; retrying sweep", + exc_info=True, + ) + _arm_async_delegation_restore_sweep( + target_queue, + ASYNC_DELIVERY_ROUTING_RETRY_SECONDS, + ) + + timer = threading.Timer(retry_delay, _restore) + timer.daemon = True + _ASYNC_DELIVERY_RETRY_TIMER = timer + timer.start() + return True + + +def schedule_async_delegation_claim_retry( + evt: Any, + completion_queue: Any, + *, + delay: float | None = None, +) -> bool: + """Schedule a durable restore sweep after a claim or routing lease delay.""" + if not isinstance(evt, dict) or evt.get("type") != "async_delegation": + return False + delegation_id = str(evt.get("delegation_id") or "").strip() + if not delegation_id or completion_queue is None: + return False + try: + from tools.async_delegation import get_durable_delegation + + durable = get_durable_delegation(delegation_id) + except (ImportError, AttributeError): + return False + except Exception: + logger.debug( + "Failed to inspect durable async delegation %s for retry", + delegation_id, + exc_info=True, + ) + return False + if not isinstance(durable, dict) or durable.get("delivery_state") != "pending": + return False + + retry_delay = ( + ASYNC_DELIVERY_CLAIM_RETRY_SECONDS if delay is None else max(0.0, float(delay)) + ) + return _arm_async_delegation_restore_sweep(completion_queue, retry_delay) + + +def requeue_async_delegation_event( + evt: Any, + completion_queue: Any, + *, + delay: float = 0.0, + stop_event: threading.Event | None = None, + durable: bool | None = None, +) -> bool: + """Requeue an async event, falling back to the durable restore sweep. + + Callers pass the queue reference they already resolved so an import failure + cannot strand a released durable claim. Legacy events without a durable row + still get one best-effort direct requeue; durable events additionally arm a + restore sweep when the direct queue write fails. + """ + if not isinstance(evt, dict) or evt.get("type") != "async_delegation": + return False + if completion_queue is None: + return False + retry_delay = max(0.0, float(delay)) + if retry_delay: + if stop_event is not None: + if stop_event.wait(retry_delay): + return False + else: + time.sleep(retry_delay) + try: + completion_queue.put(dict(evt)) + return True + except Exception: + logger.warning("Failed to requeue async delegation event", exc_info=True) + if durable is True: + return _arm_async_delegation_restore_sweep( + completion_queue, + ASYNC_DELIVERY_ROUTING_RETRY_SECONDS, + ) + return schedule_async_delegation_claim_retry( + evt, + completion_queue, + delay=ASYNC_DELIVERY_ROUTING_RETRY_SECONDS, + ) + + +def _cancel_async_delegation_claim_retry(_delegation_id: str) -> None: + # Retry is a shared durable-store sweep, not a per-delegation timer. It must + # remain armed because other pending records may rely on the same sweep. + return None + + +def async_delivery_retry_timer_count() -> int: + with _ASYNC_DELIVERY_RETRY_LOCK: + return 1 if _ASYNC_DELIVERY_RETRY_TIMER is not None else 0 + + +def claim_async_delegation_delivery( + evt: Any, + consumer: str, +) -> AsyncDelegationDeliveryClaim | None: + """Atomically claim an async completion for one WebUI delivery path. + + Current Hermes Agent builds provide a durable SQLite-backed claim contract. + Older builds fall back to the bounded process-local claim above. The local + claim also serializes duplicate legacy events on current cores where no + durable row exists yet. + """ + if not isinstance(evt, dict) or evt.get("type") != "async_delegation": + return None + delegation_id = completion_delivery_id(evt) + if not delegation_id or not _claim_bounded_local(delegation_id): + return None + + try: + from tools.async_delegation import ( + claim_event_delivery, + complete_event_delivery, # noqa: F401 - capability contract probe + release_event_delivery, # noqa: F401 - capability contract probe + ) + except (ImportError, AttributeError): + return AsyncDelegationDeliveryClaim( + delegation_id=delegation_id, + claim_id="", + durable=False, + ) + + try: + claim_id = claim_event_delivery(evt, str(consumer or "webui")) + except Exception: + _release_bounded_local(delegation_id) + logger.warning( + "Failed to claim durable async delegation delivery for %s", + delegation_id, + exc_info=True, + ) + raise + if claim_id is None: + _release_bounded_local(delegation_id) + return None + return AsyncDelegationDeliveryClaim( + delegation_id=delegation_id, + claim_id=str(claim_id or ""), + durable=True, + ) + + +def _mark_legacy_async_delivery_complete(delegation_id: str) -> bool: + """Acknowledge completion through progressively older core APIs.""" + try: + from tools import async_delegation as async_delivery + except Exception: + return False + + marker = getattr(async_delivery, "mark_completion_delivered", None) + if callable(marker): + try: + if marker(delegation_id) is not False: + return True + except Exception: + logger.debug( + "mark_completion_delivered failed for %s; trying legacy marker", + delegation_id, + exc_info=True, + ) + + legacy_marker = getattr(async_delivery, "mark_async_delegation_consumed", None) + if callable(legacy_marker): + try: + legacy_marker(delegation_id) + return True + except Exception: + logger.debug( + "Legacy async delegation marker failed for %s", + delegation_id, + exc_info=True, + ) + return False + + +def complete_async_delegation_delivery( + evt: Any, + claim: AsyncDelegationDeliveryClaim, +) -> None: + """Complete a claim after WebUI has accepted the event for delivery.""" + if claim.durable: + from tools.async_delegation import complete_event_delivery + + try: + complete_event_delivery(evt, claim.claim_id) + _cancel_async_delegation_claim_retry(claim.delegation_id) + return + except Exception: + logger.warning( + "Durable async delegation completion ACK failed for %s; " + "trying compatibility marker", + claim.delegation_id, + exc_info=True, + ) + if _mark_legacy_async_delivery_complete(claim.delegation_id): + _cancel_async_delegation_claim_retry(claim.delegation_id) + return + raise + _mark_legacy_async_delivery_complete(claim.delegation_id) + + +def release_async_delegation_delivery( + evt: Any, + claim: AsyncDelegationDeliveryClaim, +) -> None: + """Release a failed claim so a later WebUI consumer can retry it.""" + try: + if claim.durable: + from tools.async_delegation import release_event_delivery + + release_event_delivery(evt, claim.claim_id) + except Exception: + logger.warning( + "Failed to release durable async delegation delivery for %s", + claim.delegation_id, + exc_info=True, + ) + finally: + _release_bounded_local(claim.delegation_id) + + +def legacy_async_delivery_dedupe_size() -> int: + """Return bounded compatibility-dedupe size for regression coverage.""" + with _LEGACY_ASYNC_DELIVERY_LOCK: + return len(_LEGACY_ASYNC_DELIVERY_IDS) + + +def _reset_legacy_async_delivery_dedupe_for_tests() -> None: + global _ASYNC_DELIVERY_RETRY_TIMER + global _ASYNC_DELIVERY_RETRY_DEADLINE + global _ASYNC_DELIVERY_RETRY_QUEUE + global _ASYNC_DELIVERY_RETRY_GENERATION + + with _LEGACY_ASYNC_DELIVERY_LOCK: + _LEGACY_ASYNC_DELIVERY_IDS.clear() + with _ASYNC_DELIVERY_RETRY_LOCK: + timer = _ASYNC_DELIVERY_RETRY_TIMER + _ASYNC_DELIVERY_RETRY_TIMER = None + _ASYNC_DELIVERY_RETRY_DEADLINE = 0.0 + _ASYNC_DELIVERY_RETRY_QUEUE = None + _ASYNC_DELIVERY_RETRY_GENERATION += 1 + if timer is not None: + timer.cancel() diff --git a/api/streaming.py b/api/streaming.py index 491f1e3e..701c81a8 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -44,6 +44,7 @@ from api.config import ( parse_reasoning_effort, coerce_reasoning_effort_for_model, _main_model_request_overrides, + PROCESS_SESSION_INDEX, PROCESS_SESSION_INDEX_LOCK, ) from api.helpers import redact_session_data, _redact_text from api.compression_anchor import is_context_compression_marker, visible_messages_for_anchor @@ -62,6 +63,14 @@ from api.models import ( reconciled_state_db_messages_for_session, ) from api.session_ops import mark_session_title_generated, session_has_manual_title +from api.process_event_utils import ( + claim_async_delegation_delivery, + complete_async_delegation_delivery, + completion_delivery_id, + release_async_delegation_delivery, + requeue_async_delegation_event, + schedule_async_delegation_claim_retry, +) def _session_payload_with_full_messages(session, *, tool_calls=None): @@ -1840,24 +1849,23 @@ def _set_turn_session_identity(session_id: str): """Bind THIS turn's session identity to the current (task/thread-local) context and return an opaque token for _reset_turn_session_identity. - Binds two context-locals so every session-key consumer is covered without - a race: + Binds three context-locals so every session-key / UI-owner consumer is + covered without a race: * ``tools.approval._approval_session_key`` — checked FIRST by ``get_current_session_key`` (the exact call terminal_tool.py makes for a notify_on_complete background spawn: the bug path). * ``gateway.session_context._SESSION_KEY`` — read by direct - ``get_session_env("HERMES_SESSION_KEY")`` consumers (e.g. the sudo - password cache scope, terminal_tool.py:272). + ``get_session_env("HERMES_SESSION_KEY")`` consumers. + * ``gateway.session_context._SESSION_UI_SESSION_ID`` — exact browser-tab + return address stamped onto ProcessSession.origin_ui_session_id and + completion events by modern hermes-agent builds. Authoritative for + wakeup routing when present (see ``_resolve_completion_target``). It deliberately does NOT call ``gateway.session_context.set_session_vars``: that blanket setter also zeroes the platform/chat_id/user contextvars, flipping ``HERMES_SESSION_PLATFORM`` from its env fallback (``'webui'``, still written to os.environ at turn-start) to an explicit ``""`` — which - would break the ``notify_on_complete`` watcher registration gate in - terminal_tool.py:~1966. Only the session-key identity is bound; every - other session var keeps its existing os.environ fallback (CLI/cron compat - preserved — when these contextvars are _UNSET, get_session_env still falls - back to os.environ). + would break the ``notify_on_complete`` watcher registration gate. """ sid = str(session_id or "") tokens: dict = {} @@ -1871,6 +1879,11 @@ def _set_turn_session_identity(session_id: str): tokens["session_key"] = _SK.set(sid) except Exception: logger.debug("per-turn _SESSION_KEY bind failed", exc_info=True) + try: + from gateway.session_context import _SESSION_UI_SESSION_ID as _UI_SID + tokens["ui_session_id"] = _UI_SID.set(sid) + except Exception: + logger.debug("per-turn _SESSION_UI_SESSION_ID bind failed", exc_info=True) return tokens @@ -1885,6 +1898,13 @@ def _reset_turn_session_identity(tokens) -> None: """ if not tokens: return + tok = tokens.get("ui_session_id") + if tok is not None: + try: + from gateway.session_context import _SESSION_UI_SESSION_ID as _UI_SID + _UI_SID.reset(tok) + except Exception: + logger.debug("per-turn _SESSION_UI_SESSION_ID reset failed", exc_info=True) tok = tokens.get("session_key") if tok is not None: try: @@ -1947,6 +1967,14 @@ def _format_process_notification(evt: dict) -> str: """Format a completed background process notification for agent input.""" if not isinstance(evt, dict): return '' + if evt.get('type') == 'async_delegation': + try: + from tools.process_registry import format_process_notification + + return format_process_notification(evt) or '' + except Exception: + logger.debug("Failed to format async delegation notification", exc_info=True) + return '' if evt.get('type') != 'completion': return '' _sid = evt.get('session_id', '') @@ -1971,7 +1999,31 @@ def _mark_process_completion_consumed(process_registry, process_id: str) -> None logger.debug("Failed to mark process completion consumed", exc_info=True) -def _drain_webui_process_notifications(session_id: str) -> list[str]: +def _completion_event_targets_webui_session(evt_session_key: str, session_id: str) -> bool: + """Return whether a completion event belongs to this WebUI session. + + WebUI normally registers ``PROCESS_SESSION_INDEX[session_id] = session_id``. + Gateway/agent session keys can differ, so match the direct WebUI case first + and otherwise resolve through the same session-key index used by the + background wakeup path. + """ + if not evt_session_key or not session_id: + return False + if evt_session_key == session_id: + return True + try: + with PROCESS_SESSION_INDEX_LOCK: + return PROCESS_SESSION_INDEX.get(evt_session_key) == session_id + except Exception: + logger.debug("Failed to resolve completion event session key", exc_info=True) + return False + + +def _drain_webui_process_notifications( + session_id: str, + *, + pending_async_acceptances: list | None = None, +) -> list[str]: """Return completion notifications that belong to this WebUI session. The agent registry completion queue is process-wide and events do not carry @@ -1987,6 +2039,7 @@ def _drain_webui_process_notifications(session_id: str) -> list[str]: notifications: list[str] = [] skipped_events: list[dict] = [] + async_retry_events: list[tuple[dict, bool]] = [] completion_queue = getattr(process_registry, 'completion_queue', None) if completion_queue is None: return [] @@ -2004,44 +2057,142 @@ def _drain_webui_process_notifications(session_id: str) -> list[str]: logger.debug("Failed to drain process completion queue", exc_info=True) break - evt_sid = str(evt.get('session_id') or '') if isinstance(evt, dict) else '' + evt_sid = completion_delivery_id(evt) if isinstance(evt, dict) else '' if not evt_sid: skipped_events.append(evt) continue + is_async_delegation = ( + isinstance(evt, dict) and evt.get('type') == 'async_delegation' + ) try: - if process_registry.is_completion_consumed(evt_sid): + if ( + not is_async_delegation + and process_registry.is_completion_consumed(evt_sid) + ): continue - proc = process_registry.get(evt_sid) + evt_session_key = str(evt.get('session_key') or '') if isinstance(evt, dict) else '' + evt_origin_ui_session_id = ( + str(evt.get('origin_ui_session_id') or '') if isinstance(evt, dict) else '' + ) + if not evt_session_key or not evt_origin_ui_session_id: + proc = process_registry.get(evt_sid) + if not evt_session_key: + evt_session_key = str(getattr(proc, 'session_key', '') or '') + if not evt_origin_ui_session_id: + evt_origin_ui_session_id = ( + str(getattr(proc, 'origin_ui_session_id', '') or '') + or str(getattr(proc, 'spawn_session_id', '') or '') + ) except Exception: - proc = None - if getattr(proc, 'session_key', None) != session_id: + evt_session_key = '' + evt_origin_ui_session_id = '' + + # origin_ui_session_id is the exact, immutable return address and is + # authoritative over the mutable session-key index (mirrors the + # background _process_one path via _resolve_completion_target). When it + # is present, this drain claims/ACKs the event ONLY for the origin + # session — otherwise the next-turn drain could win the shared-queue + # race and deliver+ACK a completion to the wrong (session-key-index) + # session, leaving the true origin empty. Fall back to the session-key + # target check only for legacy events that carry no origin address. + if evt_origin_ui_session_id: + if evt_origin_ui_session_id != session_id: + skipped_events.append(evt) + continue + elif not _completion_event_targets_webui_session(evt_session_key, session_id): skipped_events.append(evt) continue - # Age-gate stale completions: a completion that fires long after the # user moved on must not be prepended to an unrelated later turn # (nesquena/hermes-webui#4029). Drop (consume, do not requeue) any # completion whose enqueue time is older than the configured cap. # Events without a 'completed_at' (older agent builds) are never # dropped here, preserving backward-compatible behavior. + is_stale = False + stale_age = 0.0 if stale_completion_max_age > 0 and isinstance(evt, dict): completed_at = evt.get('completed_at') if isinstance(completed_at, (int, float)) and completed_at > 0: - age = time.time() - completed_at - if age > stale_completion_max_age: - logger.info( - "Dropping stale background-process completion for " - "session %s (age %.0fs > cap %.0fs)", - evt_sid, age, stale_completion_max_age, + stale_age = time.time() - completed_at + is_stale = stale_age > stale_completion_max_age + + if is_async_delegation: + try: + claim = claim_async_delegation_delivery(evt, "webui-next-turn") + except Exception: + skipped_events.append(evt) + continue + if claim is None: + schedule_async_delegation_claim_retry(evt, completion_queue) + continue + notification_added = False + try: + if is_stale: + notification = '' + else: + notification = _format_process_notification(evt) + if not notification: + raise ValueError( + "async delegation formatter returned an empty notification" + ) + if notification: + notifications.append(notification) + notification_added = True + if is_stale: + # Stale async events are an explicit terminal disposition. + complete_async_delegation_delivery(evt, claim) + elif pending_async_acceptances is not None: + pending_async_acceptances.append( + (evt, claim, notification, completion_queue) ) - _mark_process_completion_consumed(process_registry, evt_sid) - continue + else: + # Direct callers without a live agent turn retain the + # historical synchronous acceptance behavior used by + # CLI-style drains. + complete_async_delegation_delivery(evt, claim) + except Exception: + if notification_added: + notifications.pop() + release_async_delegation_delivery(evt, claim) + async_retry_events.append( + (evt, bool(getattr(claim, "durable", False))) + ) + logger.warning( + "Failed to accept async delegation completion for session %s", + session_id, + exc_info=True, + ) + continue + if is_stale: + logger.info( + "Dropping stale async-delegation completion for session %s " + "(age %.0fs > cap %.0fs)", + evt_sid, stale_age, stale_completion_max_age, + ) + continue + + if is_stale: + logger.info( + "Dropping stale background-process completion for " + "session %s (age %.0fs > cap %.0fs)", + evt_sid, stale_age, stale_completion_max_age, + ) + _mark_process_completion_consumed(process_registry, evt_sid) + continue notification = _format_process_notification(evt) if notification: notifications.append(notification) - _mark_process_completion_consumed(process_registry, evt_sid) + # Matched but unformattable process completions are consumed rather than + # replayed forever on later turns. + _mark_process_completion_consumed(process_registry, evt_sid) + for evt, durable in async_retry_events: + requeue_async_delegation_event( + evt, + completion_queue, + durable=durable, + ) for evt in skipped_events: try: completion_queue.put(evt) @@ -2051,6 +2202,32 @@ def _drain_webui_process_notifications(session_id: str) -> list[str]: return notifications +def _accept_pending_async_delegations( + pending_async_acceptances: list, + *, + session_id: str, +) -> list[str]: + """ACK turn-bound delegation claims and return rejected notifications.""" + rejected_notifications: list[str] = [] + for evt, claim, notification, completion_queue in pending_async_acceptances: + try: + complete_async_delegation_delivery(evt, claim) + except Exception: + release_async_delegation_delivery(evt, claim) + requeue_async_delegation_event( + evt, + completion_queue, + durable=bool(getattr(claim, "durable", False)), + ) + rejected_notifications.append(notification) + logger.warning( + "Async delegation was not accepted into session %s; retrying later", + session_id, + exc_info=True, + ) + return rejected_notifications + + def _attachment_name(att) -> str: if isinstance(att, dict): return str(att.get('name') or att.get('filename') or att.get('path') or '').strip() @@ -8645,7 +8822,11 @@ def _run_agent_streaming( ) _ckpt_thread.start() - _process_notifications = _drain_webui_process_notifications(session_id) + _pending_async_acceptances = [] + _process_notifications = _drain_webui_process_notifications( + session_id, + pending_async_acceptances=_pending_async_acceptances, + ) _agent_msg_text = msg_text if _process_notifications: _agent_msg_text = "\n\n".join([*_process_notifications, msg_text]).strip() @@ -8669,6 +8850,34 @@ def _run_agent_streaming( # run_conversation() predates the moa_config kwarg. if moa_config is not None: _run_conversation_kwargs["moa_config"] = moa_config + + # Finalize durable delegation claims at the current-turn acceptance + # boundary: immediately before invoking the agent with the message + # that contains their notifications. A failed ACK is removed from + # this turn and requeued so retry cannot create a duplicate prompt. + _rejected_async_notifications = _accept_pending_async_delegations( + _pending_async_acceptances, + session_id=session_id, + ) + if _rejected_async_notifications: + for _notification in _rejected_async_notifications: + try: + _process_notifications.remove(_notification) + except ValueError: + pass + _agent_msg_text = msg_text + if _process_notifications: + _agent_msg_text = "\n\n".join( + [*_process_notifications, msg_text] + ).strip() + user_message = _build_native_multimodal_message( + workspace_ctx, + _agent_msg_text, + attachments, + workspace, + cfg=_cfg, + ) + _run_conversation_kwargs["user_message"] = user_message result = agent.run_conversation(**_run_conversation_kwargs) # #4729: the run is done — flush any reasoning tail still in the coalescing # buffer (the agent never calls reasoning_callback(None), and a turn can end on diff --git a/tests/test_async_delegation_webui_bridge.py b/tests/test_async_delegation_webui_bridge.py new file mode 100644 index 00000000..e35f9629 --- /dev/null +++ b/tests/test_async_delegation_webui_bridge.py @@ -0,0 +1,1111 @@ +"""Regression coverage for async delegate_task completion routing beyond #4912. + +#4912 taught ``format_wakeup_prompt`` how to render ``async_delegation`` events. +These tests cover the remaining WebUI bridge contract: route those events by +``session_key``, dedupe by ``delegation_id``, and coordinate with both the +current durable claim API and bounded compatibility fallbacks. +""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import queue +import subprocess +import sys +import threading +import time +import types + +import pytest + +from api import background_process as bp +from api import config as cfg +from api import process_event_utils as peu +from api import streaming + + +class _NoopLock: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + +class _FakeProcessRegistry: + def __init__(self): + self.completion_queue = queue.Queue() + self._completion_consumed: set[str] = set() + self._lock = _NoopLock() + + def is_completion_consumed(self, process_id: str) -> bool: + return process_id in self._completion_consumed + + def get(self, process_id: str): + return None + + +def _install_fake_process_registry(monkeypatch): + registry = _FakeProcessRegistry() + + def _format(evt): + return f"[ASYNC DELEGATION BATCH COMPLETE — {evt['delegation_id']}]\nbackend summary" + + fake_mod = types.ModuleType("tools.process_registry") + fake_mod.process_registry = registry + fake_mod.format_process_notification = _format + fake_pkg = sys.modules.get("tools") or types.ModuleType("tools") + monkeypatch.setitem(sys.modules, "tools", fake_pkg) + monkeypatch.setitem(sys.modules, "tools.process_registry", fake_mod) + return registry + + +def _install_fake_durable_delivery_api(monkeypatch): + calls = { + "claim": [], + "complete": [], + "release": [], + "mark": [], + "legacy": [], + "delivery_state": "pending", + "pending_ids": set(), + "restore_failures": 0, + } + fake_mod = types.ModuleType("tools.async_delegation") + + def _claim(evt, consumer): + calls["claim"].append((dict(evt), consumer)) + return f"claim:{consumer}" + + def _complete(evt, claim_id): + calls["complete"].append((dict(evt), claim_id)) + calls["delivery_state"] = "delivered" + + def _release(evt, claim_id): + calls["release"].append((dict(evt), claim_id)) + calls["delivery_state"] = "pending" + + def _get_durable(delegation_id): + if calls["delivery_state"] == "pending": + calls["pending_ids"].add(delegation_id) + return {"delivery_state": calls["delivery_state"]} + + def _restore(target_queue): + if calls["restore_failures"]: + calls["restore_failures"] -= 1 + raise RuntimeError("transient durable-store read failure") + if calls["delivery_state"] != "pending": + return 0 + pending = sorted(calls["pending_ids"]) + for delegation_id in pending: + target_queue.put(_async_delegation_event(delegation_id=delegation_id)) + return len(pending) + + def _mark(delegation_id): + calls["mark"].append(delegation_id) + return True + + def _legacy(delegation_id): + calls["legacy"].append(delegation_id) + return True + + fake_mod.claim_event_delivery = _claim + fake_mod.complete_event_delivery = _complete + fake_mod.release_event_delivery = _release + fake_mod.get_durable_delegation = _get_durable + fake_mod.restore_undelivered_completions = _restore + fake_mod.mark_completion_delivered = _mark + fake_mod.mark_async_delegation_consumed = _legacy + fake_pkg = sys.modules.get("tools") or types.ModuleType("tools") + monkeypatch.setitem(sys.modules, "tools", fake_pkg) + monkeypatch.setitem(sys.modules, "tools.async_delegation", fake_mod) + return calls + + +def _wait_until(predicate, timeout=3.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return bool(predicate()) + + +def _reset_wakeup_state() -> None: + cfg.PROCESS_SESSION_INDEX.clear() + cfg.PENDING_BG_TASK_COMPLETIONS.clear() + cfg.BG_TASK_COMPLETE_EVENTS_SEEN.clear() + cfg.DEFERRED_PROCESS_WAKEUPS.clear() + reset = getattr(peu, "_reset_legacy_async_delivery_dedupe_for_tests", None) + if reset is not None: + reset() + + +def _async_delegation_event(**overrides): + evt = { + "type": "async_delegation", + "delegation_id": "deleg_test123", + "session_key": "webui-session-1", + "status": "completed", + "is_batch": True, + "results": [{"task_index": 0, "status": "completed", "summary": "backend summary"}], + "dispatched_at": time.time() - 2, + "completed_at": time.time(), + } + evt.update(overrides) + return evt + + +def test_background_wakeup_claims_and_completes_without_registry_growth(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + cfg.PROCESS_SESSION_INDEX["webui-session-1"] = "webui-session-1" + started: list[tuple[str, str, str]] = [] + emitted: list[tuple[str, dict]] = [] + + def _accept( + session_id, + prompt, + *, + delegation_id, + evt, + claim, + process_registry, + ): + started.append((session_id, prompt, delegation_id)) + bp._record_async_delegation_accepted( + evt, + session_id=session_id, + claim=claim, + ) + + monkeypatch.setattr(bp, "_session_has_active_turn", lambda session_id: False) + monkeypatch.setattr(bp, "_start_async_delegation_wakeup_turn", _accept) + monkeypatch.setattr( + bp, + "_emit_bg_task_complete_events_coalesced", + lambda session_id, payload: emitted.append((session_id, payload)) or 1, + ) + + bp._process_one(_async_delegation_event()) + + assert started == [("webui-session-1", started[0][1], "deleg_test123")] + assert "ASYNC DELEGATION BATCH COMPLETE" in started[0][1] + assert emitted[0][1]["task_id"] == "deleg_test123" + assert cfg.BG_TASK_COMPLETE_EVENTS_SEEN == {} + assert [consumer for _evt, consumer in delivery["claim"]] == ["webui-background"] + assert len(delivery["complete"]) == 1 + assert delivery["release"] == [] + assert delivery["mark"] == [] + assert delivery["legacy"] == [] + assert registry._completion_consumed == set() + + +def test_acceptance_ack_and_queue_failure_schedules_durable_recovery(monkeypatch): + _reset_wakeup_state() + delivery = _install_fake_durable_delivery_api(monkeypatch) + async_delivery = sys.modules["tools.async_delegation"] + + async_delivery.complete_event_delivery = lambda *_args: (_ for _ in ()).throw( + RuntimeError("ACK failed") + ) + async_delivery.mark_completion_delivered = lambda _delegation_id: False + async_delivery.mark_async_delegation_consumed = lambda _delegation_id: (_ for _ in ()).throw( + RuntimeError("legacy ACK failed") + ) + async_delivery.get_durable_delegation = lambda _delegation_id: (_ for _ in ()).throw( + RuntimeError("durable store temporarily unavailable") + ) + + class _FailingQueue: + def put(self, _evt): + raise RuntimeError("queue unavailable") + + evt = _async_delegation_event() + delivery["pending_ids"].add(evt["delegation_id"]) + claim = peu.claim_async_delegation_delivery(evt, "webui-next-turn") + assert claim is not None + + try: + rejected = streaming._accept_pending_async_delegations( + [(evt, claim, "delegation notification", _FailingQueue())], + session_id="webui-session-1", + ) + assert rejected == ["delegation notification"] + assert len(delivery["release"]) == 1 + assert peu.async_delivery_retry_timer_count() == 1 + finally: + _reset_wakeup_state() + + +def test_requeue_put_failure_arms_durable_restore_sweep(monkeypatch): + _reset_wakeup_state() + delivery = _install_fake_durable_delivery_api(monkeypatch) + + class _FailingQueue: + def put(self, _evt): + raise RuntimeError("queue unavailable") + + registry = _FakeProcessRegistry() + registry.completion_queue = _FailingQueue() + evt = _async_delegation_event() + delivery["pending_ids"].add(evt["delegation_id"]) + + try: + bp._requeue_async_delegation_event(registry, evt) + assert peu.async_delivery_retry_timer_count() == 1 + finally: + _reset_wakeup_state() + + +def test_requeue_stops_without_enqueuing_during_shutdown(): + _reset_wakeup_state() + registry = _FakeProcessRegistry() + bp._DRAIN_STOP.set() + try: + bp._requeue_async_delegation_event(registry, _async_delegation_event()) + assert registry.completion_queue.empty() + finally: + bp._DRAIN_STOP.clear() + + +def test_background_unmapped_legacy_event_is_requeued_best_effort(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + _install_fake_durable_delivery_api(monkeypatch) + monkeypatch.setattr(bp, "ASYNC_DELIVERY_ROUTING_RETRY_SECONDS", 0.01) + evt = _async_delegation_event( + delegation_id=None, + session_id="proc_legacy_retry", + session_key="legacy-unmapped-session", + ) + + bp._process_one(evt) + + assert _wait_until(lambda: registry.completion_queue.qsize() == 1) + retried = registry.completion_queue.get_nowait() + assert retried["session_id"] == "proc_legacy_retry" + assert retried["_webui_routing_retry_attempted"] is True + + bp._process_one(retried) + assert registry.completion_queue.empty() + + +def test_background_wakeup_releases_claim_when_dispatch_fails(monkeypatch): + _reset_wakeup_state() + _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + cfg.PROCESS_SESSION_INDEX["webui-session-1"] = "webui-session-1" + + monkeypatch.setattr(bp, "_session_has_active_turn", lambda session_id: False) + monkeypatch.setattr(bp, "_emit_bg_task_complete_events_coalesced", lambda *_args: 1) + monkeypatch.setattr( + bp, + "_start_async_delegation_wakeup_turn", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("dispatch failed")), + ) + + bp._process_one(_async_delegation_event()) + + assert len(delivery["claim"]) == 1 + assert delivery["complete"] == [] + assert len(delivery["release"]) == 1 + assert "deleg_test123" not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN.get("webui-session-1", set()) + + +def test_background_active_turn_releases_and_requeues_without_in_memory_defer(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + cfg.PROCESS_SESSION_INDEX["webui-session-1"] = "webui-session-1" + monkeypatch.setattr(bp, "_session_has_active_turn", lambda _session_id: True) + monkeypatch.setattr( + bp, + "_start_async_delegation_wakeup_turn", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must stay deferred")), + ) + + bp._process_one(_async_delegation_event()) + + assert len(delivery["claim"]) == 1 + assert delivery["complete"] == [] + assert len(delivery["release"]) == 1 + assert registry.completion_queue.qsize() == 1 + assert cfg.DEFERRED_PROCESS_WAKEUPS == {} + assert cfg.BG_TASK_COMPLETE_EVENTS_SEEN == {} + + +@pytest.mark.parametrize("status", [None, 302, 409, 500]) +def test_autonomous_wakeup_only_acks_after_turn_acceptance(monkeypatch, status): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + from api import routes + + monkeypatch.setattr( + routes, + "start_session_turn", + lambda *_args, **_kwargs: {"_status": status, "error": "busy"}, + ) + evt = _async_delegation_event() + claim = peu.claim_async_delegation_delivery(evt, "webui-background") + assert claim is not None + + bp._start_async_delegation_wakeup_turn( + "webui-session-1", + "delegation result", + delegation_id="deleg_test123", + evt=evt, + claim=claim, + process_registry=registry, + ) + + assert _wait_until(lambda: len(delivery["release"]) == 1) + assert delivery["complete"] == [] + assert _wait_until(lambda: registry.completion_queue.qsize() == 1) + assert "deleg_test123" not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN.get("webui-session-1", set()) + + +def test_autonomous_wakeup_exception_releases_and_requeues(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + from api import routes + + monkeypatch.setattr( + routes, + "start_session_turn", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("start failed")), + ) + evt = _async_delegation_event() + claim = peu.claim_async_delegation_delivery(evt, "webui-background") + assert claim is not None + + bp._start_async_delegation_wakeup_turn( + "webui-session-1", + "delegation result", + delegation_id="deleg_test123", + evt=evt, + claim=claim, + process_registry=registry, + ) + + assert _wait_until(lambda: len(delivery["release"]) == 1) + assert delivery["complete"] == [] + assert _wait_until(lambda: registry.completion_queue.qsize() == 1) + + +def test_autonomous_wakeup_acks_after_successful_turn_acceptance(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + from api import routes + + monkeypatch.setattr( + routes, + "start_session_turn", + lambda *_args, **_kwargs: {"_status": 200, "stream_id": "stream-1"}, + ) + monkeypatch.setattr(bp, "_emit_bg_task_complete_events_coalesced", lambda *_args: 1) + evt = _async_delegation_event() + claim = peu.claim_async_delegation_delivery(evt, "webui-background") + assert claim is not None + + bp._start_async_delegation_wakeup_turn( + "webui-session-1", + "delegation result", + delegation_id="deleg_test123", + evt=evt, + claim=claim, + process_registry=registry, + ) + + assert _wait_until(lambda: len(delivery["complete"]) == 1) + assert delivery["release"] == [] + assert registry.completion_queue.qsize() == 0 + assert cfg.BG_TASK_COMPLETE_EVENTS_SEEN == {} + + +def test_background_and_next_turn_consumers_share_one_atomic_claim(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + cfg.PROCESS_SESSION_INDEX["webui-session-1"] = "webui-session-1" + evt = _async_delegation_event() + registry.completion_queue.put(dict(evt)) + starts = [] + notes = [] + barrier = threading.Barrier(2) + + monkeypatch.setattr(bp, "_session_has_active_turn", lambda _session_id: False) + monkeypatch.setattr( + bp, + "_start_async_delegation_wakeup_turn", + lambda *_args, **_kwargs: starts.append("background"), + ) + + def _background(): + barrier.wait() + bp._process_one(dict(evt)) + + def _next_turn(): + barrier.wait() + notes.extend(streaming._drain_webui_process_notifications("webui-session-1")) + + workers = [threading.Thread(target=_background), threading.Thread(target=_next_turn)] + for worker in workers: + worker.start() + for worker in workers: + worker.join(timeout=3) + + assert sum((len(starts), len(notes))) == 1 + assert len(delivery["claim"]) == 1 + assert registry._completion_consumed == set() + + +def test_legacy_async_event_id_falls_back_to_session_id(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + evt = _async_delegation_event( + delegation_id=None, + session_id="proc_deleg1", + task_id="task-legacy-1", + ) + registry.completion_queue.put(evt) + + notifications = streaming._drain_webui_process_notifications("webui-session-1") + + assert peu.completion_delivery_id(evt) == "proc_deleg1" + assert len(notifications) == 1 + assert len(delivery["claim"]) == 1 + assert len(delivery["complete"]) == 1 + + +def test_streaming_next_turn_claims_and_completes_without_registry_growth(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + registry.completion_queue.put(_async_delegation_event()) + + notifications = streaming._drain_webui_process_notifications("webui-session-1") + + assert len(notifications) == 1 + assert "ASYNC DELEGATION BATCH COMPLETE" in notifications[0] + assert [consumer for _evt, consumer in delivery["claim"]] == ["webui-next-turn"] + assert len(delivery["complete"]) == 1 + assert delivery["release"] == [] + assert registry._completion_consumed == set() + + +def test_streaming_live_turn_defers_ack_until_agent_acceptance_boundary(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + registry.completion_queue.put(_async_delegation_event()) + pending = [] + + notifications = streaming._drain_webui_process_notifications( + "webui-session-1", + pending_async_acceptances=pending, + ) + + assert len(notifications) == 1 + assert len(pending) == 1 + assert delivery["complete"] == [] + evt, claim, notification, completion_queue = pending[0] + assert notification == notifications[0] + assert completion_queue is registry.completion_queue + peu.complete_async_delegation_delivery(evt, claim) + assert len(delivery["complete"]) == 1 + + +def test_streaming_formatter_failure_releases_and_requeues(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + process_registry_mod = sys.modules["tools.process_registry"] + + def _fail_formatter(_evt): + raise RuntimeError("formatter failed") + + process_registry_mod.format_process_notification = _fail_formatter + registry.completion_queue.put(_async_delegation_event()) + pending = [] + + notifications = streaming._drain_webui_process_notifications( + "webui-session-1", + pending_async_acceptances=pending, + ) + + assert notifications == [] + assert pending == [] + assert len(delivery["claim"]) == 1 + assert delivery["complete"] == [] + assert len(delivery["release"]) == 1 + assert registry.completion_queue.qsize() == 1 + + +def test_streaming_next_turn_releases_and_requeues_when_complete_fails(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + async_delivery = sys.modules["tools.async_delegation"] + + def _fail_complete(_evt, _claim_id): + raise RuntimeError("complete failed") + + async_delivery.complete_event_delivery = _fail_complete + async_delivery.mark_completion_delivered = lambda _delegation_id: False + async_delivery.mark_async_delegation_consumed = lambda _delegation_id: (_ for _ in ()).throw( + RuntimeError("legacy marker failed") + ) + registry.completion_queue.put(_async_delegation_event()) + + notifications = streaming._drain_webui_process_notifications("webui-session-1") + + assert notifications == [] + assert len(delivery["claim"]) == 1 + assert len(delivery["release"]) == 1 + assert registry.completion_queue.qsize() == 1 + assert registry._completion_consumed == set() + + +def test_streaming_synchronous_ack_and_queue_failure_arms_durable_restore(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + async_delivery = sys.modules["tools.async_delegation"] + + async_delivery.complete_event_delivery = lambda *_args: (_ for _ in ()).throw( + RuntimeError("complete failed") + ) + async_delivery.mark_completion_delivered = lambda _delegation_id: False + async_delivery.mark_async_delegation_consumed = lambda _delegation_id: (_ for _ in ()).throw( + RuntimeError("legacy marker failed") + ) + async_delivery.get_durable_delegation = lambda _delegation_id: (_ for _ in ()).throw( + RuntimeError("durable store temporarily unavailable") + ) + registry.completion_queue.put(_async_delegation_event()) + registry.completion_queue.put = lambda _evt: (_ for _ in ()).throw( + RuntimeError("queue unavailable") + ) + + try: + notifications = streaming._drain_webui_process_notifications("webui-session-1") + assert notifications == [] + assert len(delivery["release"]) == 1 + assert peu.async_delivery_retry_timer_count() == 1 + finally: + _reset_wakeup_state() + + +def test_streaming_next_turn_drain_routes_via_process_session_index_mapping(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + cfg.PROCESS_SESSION_INDEX["gateway-session-key"] = "webui-session-1" + registry.completion_queue.put(_async_delegation_event(session_key="gateway-session-key")) + + wrong_session_notifications = streaming._drain_webui_process_notifications("webui-session-2") + right_session_notifications = streaming._drain_webui_process_notifications("webui-session-1") + + assert wrong_session_notifications == [] + assert len(right_session_notifications) == 1 + assert "ASYNC DELEGATION BATCH COMPLETE" in right_session_notifications[0] + assert len(delivery["claim"]) == 1 + assert len(delivery["complete"]) == 1 + + +def test_claim_held_by_crashed_owner_schedules_retry_instead_of_dropping(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + _install_fake_durable_delivery_api(monkeypatch) + async_delivery = sys.modules["tools.async_delegation"] + async_delivery.claim_event_delivery = lambda _evt, _consumer: None + registry.completion_queue.put(_async_delegation_event()) + + try: + notifications = streaming._drain_webui_process_notifications("webui-session-1") + assert notifications == [] + assert registry.completion_queue.empty() + assert peu.async_delivery_retry_timer_count() == 1 + finally: + _reset_wakeup_state() + + +def test_background_unmapped_durable_event_schedules_retry(monkeypatch): + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + _install_fake_durable_delivery_api(monkeypatch) + evt = _async_delegation_event() + + try: + bp._process_one(evt) + assert registry.completion_queue.empty() + assert peu.async_delivery_retry_timer_count() == 1 + finally: + _reset_wakeup_state() + + +def test_pending_durable_claim_is_requeued_after_lease_retry(monkeypatch): + _reset_wakeup_state() + _install_fake_durable_delivery_api(monkeypatch) + retry_queue = queue.Queue() + evt = _async_delegation_event() + + assert peu.schedule_async_delegation_claim_retry(evt, retry_queue, delay=0.01) + assert peu.schedule_async_delegation_claim_retry(evt, retry_queue, delay=0.01) + assert peu.async_delivery_retry_timer_count() == 1 + + retried = retry_queue.get(timeout=1) + assert retried["delegation_id"] == "deleg_test123" + assert _wait_until(lambda: peu.async_delivery_retry_timer_count() == 0) + + +def test_delivered_or_legacy_event_is_not_scheduled_for_lease_retry(monkeypatch): + _reset_wakeup_state() + delivery = _install_fake_durable_delivery_api(monkeypatch) + delivery["delivery_state"] = "delivered" + retry_queue = queue.Queue() + + assert not peu.schedule_async_delegation_claim_retry( + _async_delegation_event(), retry_queue, delay=0.01 + ) + assert not peu.schedule_async_delegation_claim_retry( + _async_delegation_event(delegation_id=None), retry_queue, delay=0.01 + ) + assert retry_queue.empty() + assert peu.async_delivery_retry_timer_count() == 0 + + +def test_async_delivery_retry_uses_one_lossless_restore_sweep(monkeypatch): + _reset_wakeup_state() + delivery = _install_fake_durable_delivery_api(monkeypatch) + retry_queue = queue.Queue() + cap = peu.LEGACY_ASYNC_DELIVERY_DEDUPE_MAX + try: + for index in range(cap + 100): + assert peu.schedule_async_delegation_claim_retry( + _async_delegation_event(delegation_id=f"deleg_timer_{index}"), + retry_queue, + delay=0.2, + ) + assert peu.async_delivery_retry_timer_count() == 1 + assert _wait_until(lambda: retry_queue.qsize() == cap + 100) + restored = { + retry_queue.get_nowait()["delegation_id"] for _ in range(cap + 100) + } + assert "deleg_timer_0" in restored + assert f"deleg_timer_{cap + 99}" in restored + assert restored == delivery["pending_ids"] + assert _wait_until(lambda: peu.async_delivery_retry_timer_count() == 0) + finally: + _reset_wakeup_state() + + +def test_async_delivery_restore_sweep_retries_transient_store_failure(monkeypatch): + _reset_wakeup_state() + delivery = _install_fake_durable_delivery_api(monkeypatch) + delivery["restore_failures"] = 1 + retry_queue = queue.Queue() + monkeypatch.setattr(peu, "ASYNC_DELIVERY_ROUTING_RETRY_SECONDS", 0.01) + try: + assert peu.schedule_async_delegation_claim_retry( + _async_delegation_event(delegation_id="deleg_transient"), + retry_queue, + delay=0.01, + ) + retried = retry_queue.get(timeout=1) + assert retried["delegation_id"] == "deleg_transient" + assert delivery["restore_failures"] == 0 + assert _wait_until(lambda: peu.async_delivery_retry_timer_count() == 0) + finally: + _reset_wakeup_state() + + +def test_legacy_completion_prefers_mark_completion_delivered(monkeypatch): + _reset_wakeup_state() + calls = {"mark": [], "legacy": []} + fake_mod = types.ModuleType("tools.async_delegation") + fake_mod.mark_completion_delivered = lambda delegation_id: calls["mark"].append(delegation_id) or True + fake_mod.mark_async_delegation_consumed = lambda delegation_id: calls["legacy"].append(delegation_id) + fake_pkg = sys.modules.get("tools") or types.ModuleType("tools") + monkeypatch.setitem(sys.modules, "tools", fake_pkg) + monkeypatch.setitem(sys.modules, "tools.async_delegation", fake_mod) + monkeypatch.setattr(fake_pkg, "async_delegation", fake_mod, raising=False) + evt = _async_delegation_event() + + claim = peu.claim_async_delegation_delivery(evt, "legacy-marker-test") + assert claim is not None + peu.complete_async_delegation_delivery(evt, claim) + + assert calls == {"mark": ["deleg_test123"], "legacy": []} + + +def test_legacy_completion_falls_back_to_old_consumed_marker(monkeypatch): + _reset_wakeup_state() + calls = {"mark": [], "legacy": []} + fake_mod = types.ModuleType("tools.async_delegation") + fake_mod.mark_completion_delivered = lambda delegation_id: calls["mark"].append(delegation_id) or False + fake_mod.mark_async_delegation_consumed = lambda delegation_id: calls["legacy"].append(delegation_id) + fake_pkg = sys.modules.get("tools") or types.ModuleType("tools") + monkeypatch.setitem(sys.modules, "tools", fake_pkg) + monkeypatch.setitem(sys.modules, "tools.async_delegation", fake_mod) + monkeypatch.setattr(fake_pkg, "async_delegation", fake_mod, raising=False) + evt = _async_delegation_event() + + claim = peu.claim_async_delegation_delivery(evt, "legacy-marker-test") + assert claim is not None + peu.complete_async_delegation_delivery(evt, claim) + + assert calls == {"mark": ["deleg_test123"], "legacy": ["deleg_test123"]} + + +def test_legacy_async_delivery_dedupe_is_bounded(monkeypatch): + _reset_wakeup_state() + fake_pkg = sys.modules.get("tools") or types.ModuleType("tools") + monkeypatch.setitem(sys.modules, "tools", fake_pkg) + monkeypatch.setitem(sys.modules, "tools.async_delegation", types.ModuleType("tools.async_delegation")) + + cap = peu.LEGACY_ASYNC_DELIVERY_DEDUPE_MAX + for index in range(5000): + evt = _async_delegation_event(delegation_id=f"deleg_legacy_{index}") + claim = peu.claim_async_delegation_delivery(evt, "bounded-growth-test") + assert claim is not None + peu.complete_async_delegation_delivery(evt, claim) + + assert peu.legacy_async_delivery_dedupe_size() == cap + + +def test_real_core_restart_delivers_async_completion_exactly_once(tmp_path): + """A real durable core record must not be restored after WebUI accepts it.""" + try: + from tools import async_delegation as ad + except Exception as exc: + pytest.skip(f"hermes-agent async delegation module unavailable: {exc}") + if not all( + hasattr(ad, name) + for name in ("claim_event_delivery", "complete_event_delivery", "release_event_delivery") + ): + pytest.skip("installed hermes-agent predates durable completion claims") + + webui_repo = Path(__file__).resolve().parents[1] + agent_repo = Path(ad.__file__).resolve().parents[1] + env = dict(os.environ) + env["HERMES_HOME"] = str(tmp_path) + env["HERMES_BASE_HOME"] = str(tmp_path) + env["HERMES_CONFIG_PATH"] = str(tmp_path / "config.yaml") + env["HERMES_WEBUI_STATE_DIR"] = str(tmp_path) + env["HERMES_WEBUI_TEST_STATE_DIR"] = str(tmp_path) + env["HERMES_WEBUI_NO_DOTENV"] = "1" + env["HERMES_WEBUI_AGENT_DIR"] = str(agent_repo) + env["PYTHONPATH"] = os.pathsep.join( + part for part in (str(webui_repo), str(agent_repo), env.get("PYTHONPATH", "")) if part + ) + + python_executable = os.environ.get("HERMES_WEBUI_PYTHON") or sys.executable + consumer = """ +import json +import time +from api import streaming +from tools import async_delegation as ad +record = ad.dispatch_async_delegation( + goal="restart", context=None, toolsets=None, role="leaf", model="m", + session_key="webui-session-1", parent_session_id="durable-parent", + runner=lambda: {"status": "completed", "summary": "after restart"}, +) +deadline = time.time() + 10 +while ad.active_count() and time.time() < deadline: + time.sleep(.01) +assert not ad.active_count() +notes = streaming._drain_webui_process_notifications("webui-session-1") +row = ad.get_durable_delegation(record["delegation_id"]) +print(json.dumps({ + "delegation_id": record["delegation_id"], + "deliveries": len(notes), + "delivery_state": row["delivery_state"], +})) +""" + first = subprocess.run( + [python_executable, "-c", consumer], + cwd=webui_repo, + env=env, + text=True, + capture_output=True, + timeout=20, + check=False, + ) + assert first.returncode == 0, first.stderr + accepted = json.loads(first.stdout.strip().splitlines()[-1]) + assert accepted["deliveries"] == 1 + assert accepted["delivery_state"] == "delivered" + + restart_probe = """ +from tools.process_registry import process_registry +print(process_registry.completion_queue.qsize()) +""" + second = subprocess.run( + [python_executable, "-c", restart_probe], + cwd=webui_repo, + env=env, + text=True, + capture_output=True, + timeout=20, + check=False, + ) + assert second.returncode == 0, second.stderr + assert second.stdout.strip().splitlines()[-1] == "0" + +def test_real_core_active_claim_survives_restart_and_retries_after_lease(tmp_path): + """A restored event with a live claim is retried after lease expiry, not lost.""" + try: + from tools import async_delegation as ad + except Exception as exc: + pytest.skip(f"hermes-agent async delegation module unavailable: {exc}") + if not all( + hasattr(ad, name) + for name in ( + "claim_event_delivery", + "complete_event_delivery", + "release_event_delivery", + "get_durable_delegation", + ) + ): + pytest.skip("installed hermes-agent predates durable completion claims") + + webui_repo = Path(__file__).resolve().parents[1] + agent_repo = Path(ad.__file__).resolve().parents[1] + env = dict(os.environ) + env["HERMES_HOME"] = str(tmp_path) + env["HERMES_BASE_HOME"] = str(tmp_path) + env["HERMES_CONFIG_PATH"] = str(tmp_path / "config.yaml") + env["HERMES_WEBUI_STATE_DIR"] = str(tmp_path) + env["HERMES_WEBUI_TEST_STATE_DIR"] = str(tmp_path) + env["HERMES_WEBUI_NO_DOTENV"] = "1" + env["HERMES_WEBUI_AGENT_DIR"] = str(agent_repo) + env["PYTHONPATH"] = os.pathsep.join( + part for part in (str(webui_repo), str(agent_repo), env.get("PYTHONPATH", "")) if part + ) + python_executable = os.environ.get("HERMES_WEBUI_PYTHON") or sys.executable + + producer = """ +import time +from tools import async_delegation as ad +from tools.process_registry import process_registry +record = ad.dispatch_async_delegation( + goal="lease-restart", context=None, toolsets=None, role="leaf", model="m", + session_key="webui-session-1", parent_session_id="durable-parent", + runner=lambda: {"status": "completed", "summary": "after lease"}, +) +deadline = time.time() + 10 +while ad.active_count() and time.time() < deadline: + time.sleep(.01) +assert not ad.active_count() +evt = process_registry.completion_queue.get_nowait() +claim = ad.claim_event_delivery(evt, "crashed-owner") +assert claim +print(record["delegation_id"]) +""" + first = subprocess.run( + [python_executable, "-c", producer], + cwd=webui_repo, + env=env, + text=True, + capture_output=True, + timeout=20, + check=False, + ) + assert first.returncode == 0, first.stderr + delegation_id = next( + line.strip() + for line in reversed(first.stdout.splitlines()) + if line.strip().startswith("deleg_") + ) + + consumer = f""" +import json +import time +from tools import async_delegation as ad +from api import process_event_utils as peu, streaming +from tools.process_registry import process_registry +peu.ASYNC_DELIVERY_CLAIM_RETRY_SECONDS = 0.05 +first_notes = streaming._drain_webui_process_notifications("webui-session-1") +first_timers = peu.async_delivery_retry_timer_count() +with ad._DB_LOCK, ad._connect() as conn: + conn.execute( + "UPDATE async_delegations SET delivery_claimed_at=? WHERE delegation_id=?", + (time.time() - 301, {delegation_id!r}), + ) +deadline = time.time() + 3 +while process_registry.completion_queue.empty() and time.time() < deadline: + time.sleep(.01) +second_notes = streaming._drain_webui_process_notifications("webui-session-1") +row = ad.get_durable_delegation({delegation_id!r}) +print(json.dumps({{ + "first_deliveries": len(first_notes), + "scheduled_retries": first_timers, + "second_deliveries": len(second_notes), + "delivery_state": row["delivery_state"] if row else None, +}})) +""" + second = subprocess.run( + [python_executable, "-c", consumer], + cwd=webui_repo, + env=env, + text=True, + capture_output=True, + timeout=20, + check=False, + ) + assert second.returncode == 0, second.stderr + result = json.loads(second.stdout.strip().splitlines()[-1]) + assert result == { + "first_deliveries": 0, + "scheduled_retries": 1, + "second_deliveries": 1, + "delivery_state": "delivered", + } + + probe = subprocess.run( + [ + python_executable, + "-c", + "from tools.process_registry import process_registry; print(process_registry.completion_queue.qsize())", + ], + cwd=webui_repo, + env=env, + text=True, + capture_output=True, + timeout=20, + check=False, + ) + assert probe.returncode == 0, probe.stderr + assert probe.stdout.strip().splitlines()[-1] == "0" + + +# ── Combined-path coverage: #6185 origin-routing ⊕ #6159 durable-claims ────── +# Neither PR alone exercises the interaction where the origin return address +# overrides the mutable session-key index AND the routed delivery still flows +# through the durable claim/complete ack. These guard the combine seam. + + +def test_origin_ui_session_id_overrides_index_and_still_acks(monkeypatch): + """An async completion whose session-key index resolves to session B but + whose origin_ui_session_id is A must start the wakeup turn in A (origin + wins), and still take exactly one durable claim + one ack.""" + _reset_wakeup_state() + _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + # The session-key index would route to session "B" ... + cfg.PROCESS_SESSION_INDEX["webui-session-1"] = "session-B" + started: list[tuple[str, str, str]] = [] + + def _accept(session_id, prompt, *, delegation_id, evt, claim, process_registry): + started.append((session_id, prompt, delegation_id)) + bp._record_async_delegation_accepted(evt, session_id=session_id, claim=claim) + + monkeypatch.setattr(bp, "_session_has_active_turn", lambda session_id: False) + monkeypatch.setattr(bp, "_start_async_delegation_wakeup_turn", _accept) + monkeypatch.setattr( + bp, "_emit_bg_task_complete_events_coalesced", lambda session_id, payload: 1 + ) + + # ... but the event carries the exact origin return address "session-A". + bp._process_one(_async_delegation_event(origin_ui_session_id="session-A")) + + assert started, "wakeup turn was never started" + assert started[0][0] == "session-A", ( + f"origin_ui_session_id must win over the index; started in {started[0][0]!r}" + ) + assert [consumer for _evt, consumer in delivery["claim"]] == ["webui-background"] + assert len(delivery["complete"]) == 1 # exactly one ack, in the origin session + assert delivery["release"] == [] + + +def test_origin_only_completion_without_session_key_routes_and_acks(monkeypatch): + """A completion with no session_key at all but a valid origin_ui_session_id + survives the drop-paths (origin is a sufficient route) and delivers+acks.""" + _reset_wakeup_state() + _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + started: list[str] = [] + + def _accept(session_id, prompt, *, delegation_id, evt, claim, process_registry): + started.append(session_id) + bp._record_async_delegation_accepted(evt, session_id=session_id, claim=claim) + + monkeypatch.setattr(bp, "_session_has_active_turn", lambda session_id: False) + monkeypatch.setattr(bp, "_start_async_delegation_wakeup_turn", _accept) + monkeypatch.setattr( + bp, "_emit_bg_task_complete_events_coalesced", lambda session_id, payload: 1 + ) + + evt = _async_delegation_event(origin_ui_session_id="session-A") + evt.pop("session_key", None) + bp._process_one(evt) + + assert started == ["session-A"] + assert len(delivery["complete"]) == 1 + + +def test_async_completion_with_unresolvable_target_retries_not_silent_drop(monkeypatch): + """The combine caveat: an async event that resolves to an empty target must + route through the bounded retry (so the durable row stays retryable), NOT a + bare return that would leave it pending forever with no ack and no retry.""" + _reset_wakeup_state() + _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + retried: list[dict] = [] + monkeypatch.setattr( + bp, + "_retry_unmapped_async_delegation_event", + lambda process_registry, evt: retried.append(dict(evt)), + ) + + # No session_key index entry and no origin_ui_session_id → empty target. + evt = _async_delegation_event() + evt.pop("session_key", None) + bp._process_one(evt) + + assert len(retried) == 1, "unrouteable async event must be handed to the retry" + # No claim was taken and no ACK fired — the durable row stays pending, and + # NOTHING was completed (which would have been a spurious ack). + assert delivery["claim"] == [] + assert delivery["complete"] == [] + + +def test_next_turn_drain_respects_origin_over_session_key_index(monkeypatch): + """Codex-caught combine gap: the next-turn drain must NOT deliver/claim/ACK a + completion to the session-key-index session when the event carries an exact + origin_ui_session_id for a DIFFERENT session. Otherwise the drain wins the + shared-queue race and ACKs the completion to the wrong session, leaving the + true origin empty. Origin is authoritative in the drain, mirroring + _resolve_completion_target in the background path.""" + _reset_wakeup_state() + registry = _install_fake_process_registry(monkeypatch) + delivery = _install_fake_durable_delivery_api(monkeypatch) + # The session-key index would route this event to session "B" ... + cfg.PROCESS_SESSION_INDEX["webui-session-1"] = "session-B" + evt = _async_delegation_event(origin_ui_session_id="session-A") + registry.completion_queue.put(evt) + + # ... but a drain for session B must SKIP it (origin says A), taking no + # claim and no ack, and leaving the event on the queue for A. + notifications_b = streaming._drain_webui_process_notifications("session-B") + assert notifications_b == [] + assert delivery["claim"] == [] + assert delivery["complete"] == [] + + # A drain for the origin session A delivers + acks exactly once. + notifications_a = streaming._drain_webui_process_notifications("session-A") + assert len(notifications_a) == 1 + assert "ASYNC DELEGATION BATCH COMPLETE" in notifications_a[0] + assert [consumer for _evt, consumer in delivery["claim"]] == ["webui-next-turn"] + assert len(delivery["complete"]) == 1 + assert delivery["release"] == [] + + diff --git a/tests/test_notify_on_complete_webui.py b/tests/test_notify_on_complete_webui.py index b74f45fd..2ba2095e 100644 --- a/tests/test_notify_on_complete_webui.py +++ b/tests/test_notify_on_complete_webui.py @@ -4,10 +4,13 @@ from pathlib import Path def test_webui_drains_only_matching_background_completion_events(): src = Path("api/streaming.py").read_text(encoding="utf-8") - assert "def _drain_webui_process_notifications(session_id: str)" in src + assert "def _drain_webui_process_notifications(" in src + assert "pending_async_acceptances: list | None = None" in src assert "from tools.process_registry import process_registry" in src assert "proc = process_registry.get(evt_sid)" in src - assert "getattr(proc, 'session_key', None) != session_id" in src + assert "def _completion_event_targets_webui_session(evt_session_key: str, session_id: str)" in src + assert "PROCESS_SESSION_INDEX.get(evt_session_key) == session_id" in src + assert "not _completion_event_targets_webui_session(evt_session_key, session_id)" in src assert "skipped_events.append(evt)" in src assert "completion_queue.put(evt)" in src @@ -15,7 +18,9 @@ def test_webui_drains_only_matching_background_completion_events(): def test_webui_injects_process_notifications_without_persisting_them_as_user_text(): src = Path("api/streaming.py").read_text(encoding="utf-8") - assert "_process_notifications = _drain_webui_process_notifications(session_id)" in src + assert "_process_notifications = _drain_webui_process_notifications(" in src + assert "pending_async_acceptances=_pending_async_acceptances" in src + assert "_accept_pending_async_delegations(" in src assert "[*_process_notifications, msg_text]" in src assert "_build_native_multimodal_message(workspace_ctx, _agent_msg_text" in src assert "persist_user_message=msg_text" in src @@ -40,8 +45,11 @@ def test_webui_age_gates_stale_background_completion_events(): assert "HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS" in src # The drain reads completed_at and drops over-age events without requeueing. assert "completed_at = evt.get('completed_at')" in src - assert "age = time.time() - completed_at" in src - assert "if age > stale_completion_max_age:" in src - # Over-age events are consumed (marked), not requeued, so they vanish. + assert "stale_age = time.time() - completed_at" in src + assert "is_stale = stale_age > stale_completion_max_age" in src + assert "if is_stale:" in src + # Over-age process events use the registry marker; async delegations finish + # their durable delivery claim. Neither path is added to skipped_events. assert "_mark_process_completion_consumed(process_registry, evt_sid)" in src + assert "complete_async_delegation_delivery(evt, claim)" in src diff --git a/tests/test_xsession_wakeup_misroute.py b/tests/test_xsession_wakeup_misroute.py index 27b5a23e..e83ac1bb 100644 --- a/tests/test_xsession_wakeup_misroute.py +++ b/tests/test_xsession_wakeup_misroute.py @@ -257,3 +257,53 @@ def test_resolve_wakeup_target_passthrough_when_owner_unknown(): session_key_resolved_sid=sid, proc_session=None, ) == sid + + +# --------------------------------------------------------------------------- +# Exact origin_ui_session_id ownership (complements Option 3) +# --------------------------------------------------------------------------- + +def test_resolve_completion_target_prefers_origin_ui_session_id(): + """When session_key resolves to B but origin_ui_session_id is A, route to A.""" + bp = importlib.import_module("api.background_process") + assert hasattr(bp, "_resolve_completion_target"), ( + "missing _resolve_completion_target — exact-owner routing not implemented" + ) + SESS_A = "webui-tab-A" + SESS_B = "webui-tab-B" + assert bp._resolve_completion_target( + session_key_resolved_sid=SESS_B, + origin_ui_session_id=SESS_A, + ) == SESS_A + + +def test_resolve_completion_target_fallback_without_origin(): + bp = importlib.import_module("api.background_process") + sid = "legacy-only" + assert bp._resolve_completion_target( + session_key_resolved_sid=sid, + origin_ui_session_id="", + ) == sid + + +def test_env_immune_owner_prefers_origin_ui_session_id(): + bp = importlib.import_module("api.background_process") + ps = _fake_ps(session_key="B", spawn_session_id="spawn-B") + # Duck-typed modern field wins over legacy spawn_session_id. + ps.origin_ui_session_id = "tab-A" + assert bp._env_immune_spawn_owner(ps) == "tab-A" + + +def test_turn_identity_binder_sets_ui_session_id(): + """Option 1 must also bind HERMES_UI_SESSION_ID so terminal_tool can stamp + origin_ui_session_id on notify_on_complete spawns.""" + streaming = importlib.import_module("api.streaming") + pytest.importorskip("gateway.session_context", reason="hermes-agent not installed") + from gateway import session_context as sc + + bind = streaming._bind_turn_session_identity + assert sc._SESSION_UI_SESSION_ID.get() is sc._UNSET + with bind("webui-sid-42"): + assert sc._SESSION_UI_SESSION_ID.get() == "webui-sid-42" + assert sc.get_session_env("HERMES_UI_SESSION_ID", "") == "webui-sid-42" + assert sc._SESSION_UI_SESSION_ID.get() is sc._UNSET