fix(wakeup): route async-delegation completions by origin + durable-claim delivery (#6283)

Route async-delegation completions by the immutable origin_ui_session_id (exact origin tab) and deliver them through a durable claim/complete/release lifecycle so they arrive exactly once, restart-safe, on both the background wakeup and next-turn drain paths. Combines #6185 (@carlotestor) + #6159 (@sysophelper-droid); supersedes #6002/#6225.

Co-authored-by: carlotestor <carlotestor@users.noreply.github.com>
Co-authored-by: sysophelper-droid <sysophelper-droid@users.noreply.github.com>

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)

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``
+    # the process id. Async delegation completions carry ``delegation_id``
-    # to avoid confusing that wire-format name with the WebUI session id.
+    # instead; expose either stable delivery id as payload ``task_id``.
-    process_id = str(evt.get("session_id") or "")
+    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"
+    # enqueued by ProcessRegistry._move_to_finished() historically carried NO
-    # field — only the watch_match enqueue includes one. Without it the old
+    # "session_key" field — only the watch_match enqueue included one. Without
-    # `evt.get("session_key") or process_id` fell back to the process id
+    # it the old `evt.get("session_key") or process_id` fell back to the
-    # ("proc_xxxx"), which is never a PROCESS_SESSION_INDEX key (only
+    # process id ("proc_xxxx"), which is never a PROCESS_SESSION_INDEX key.
-    # webui_session_id -> webui_session_id is registered at chat-start), so
+    # Recover spawn-time routing metadata from the process registry.
-    # every wakeup was silently dropped here and the frontend never POSTed an
+    if process_id and (not session_key or not origin_ui_session_id):
-    # 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:
         try:
             if _process_registry is not None:
                 _ps = _process_registry.get(process_id)
-                if _ps is not None and getattr(_ps, "session_key", ""):
+                if _ps is not None:
-                    session_key = str(_ps.session_key)
+                    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 = ""
-        session_id = _cfg.PROCESS_SESSION_INDEX.get(session_key)
+    if session_key:
-    if not session_id:
+        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
+    # First retain the process-registry spawn-owner cross-check for legacy
-    # session_key was captured by the terminal tool from the (historically
+    # terminal events. Then apply origin_ui_session_id as the final authority;
-    # racy) process-global env at spawn. Option 1 binds the per-turn identity
+    # that exact owner is shared by process and async-delegation completions.
-    # 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.
     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]:

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()

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
+    Binds three context-locals so every session-key / UI-owner consumer is
-    a race:
+    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
+        ``get_session_env("HERMES_SESSION_KEY")`` consumers.
-        password cache scope, terminal_tool.py:272).
+      * ``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
+    would break the ``notify_on_complete`` watcher registration gate.
-    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).
     """
     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
+            evt_session_key = ''
-        if getattr(proc, 'session_key', None) != session_id:
+            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
+                stale_age = time.time() - completed_at
-                if age > stale_completion_max_age:
+                is_stale = stale_age > stale_completion_max_age
-                    logger.info(
+
-                        "Dropping stale background-process completion for "
+        if is_async_delegation:
-                        "session %s (age %.0fs > cap %.0fs)",
+            try:
-                        evt_sid, age, stale_completion_max_age,
+                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)
+                else:
-                    continue
+                    # 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

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 "stale_age = time.time() - completed_at" in src
-    assert "if age > stale_completion_max_age:" in src
+    assert "is_stale = stale_age > stale_completion_max_age" in src
-    # Over-age events are consumed (marked), not requeued, so they vanish.
+    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
 

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