Background task notifications routed to wrong session

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #1706 fix(gateway): persist watcher metadata in checkpoint for crash recovery (salvaged #1573)

  merged to main

  files: gateway/run.py, tests/gateway/test_background_process_notifications.py, tests/tools/test_process_registry.py, tools/process_registry.py, tools/terminal_tool.py

• #7454 fix(gateway): replace os.environ session state with contextvars + fix skill frontmatter truncation

  merged to main

  files: agent/prompt_builder.py, agent/skill_utils.py, gateway/run.py, gateway/session_context.py, tests/gateway/test_session_env.py, tools/cronjob_tools.py, tools/send_message_tool.py, tools/skills_tool.py, tools/terminal_tool.py, tools/tts_tool.py

  Show the diff

  • agent/prompt_builder.py

    @@ -487,7 +487,7 @@ def _parse_skill_file(skill_file: Path) -> tuple[bool, dict, str]:
         (True, {}, "") to err on the side of showing the skill.
         """
         try:
    -        raw = skill_file.read_text(encoding="utf-8")[:2000]
    +        raw = skill_file.read_text(encoding="utf-8")
             frontmatter, _ = parse_frontmatter(raw)
     
             if not skill_matches_platform(frontmatter):
    @@ -495,7 +495,7 @@ def _parse_skill_file(skill_file: Path) -> tuple[bool, dict, str]:
     
             return True, frontmatter, extract_skill_description(frontmatter)
         except Exception as e:
    -        logger.debug("Failed to parse skill file %s: %s", skill_file, e)
    +        logger.warning("Failed to parse skill file %s: %s", skill_file, e)
             return True, {}, ""
     
     
    @@ -558,9 +558,10 @@ def build_skills_system_prompt(
         # ── Layer 1: in-process LRU cache ─────────────────────────────────
         # Include the resolved platform so per-platform disabled-skill lists
         # produce distinct cache entries (gateway serves multiple platforms).
    +    from gateway.session_context import get_session_env
         _platform_hint = (
             os.environ.get("HERMES_PLATFORM")
    -        or os.environ.get("HERMES_SESSION_PLATFORM")
    +        or get_session_env("HERMES_SESSION_PLATFORM")
             or ""
         )
         cache_key = (

  • agent/skill_utils.py

    @@ -145,10 +145,11 @@ def get_disabled_skill_names(platform: str | None = None) -> Set[str]:
         if not isinstance(skills_cfg, dict):
             return set()
     
    +    from gateway.session_context import get_session_env
         resolved_platform = (
             platform
             or os.getenv("HERMES_PLATFORM")
    -        or os.getenv("HERMES_SESSION_PLATFORM")
    +        or get_session_env("HERMES_SESSION_PLATFORM")
         )
         if resolved_platform:
             platform_disabled = (skills_cfg.get("platform_disabled") or {}).get(

  • gateway/run.py

    @@ -2400,8 +2400,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
             # Build session context
             context = build_session_context(source, self.config, session_entry)
             
    -        # Set environment variables for tools
    -        self._set_session_env(context)
    +        # Set session context variables for tools (task-local, concurrency-safe)
    +        _session_env_tokens = [redacted:secret-assignment]
             
             # Read privacy.redact_pii from config (re-read per message)
             _redact_pii = False
    @@ -3234,8 +3234,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
                     "Try again or use /reset to start a fresh session."
                 )
             finally:
    -            # Clear session env
    -            self._clear_session_env()
    +            # Restore session context variables to their pre-handler state
    +            self._clear_session_env(_session_env_tokens)
         
         def _format_session_info(self) -> str:
             """Resolve current model config and return a formatted info block.
    @@ -6120,20 +6120,27 @@ async def _send_update_notification(self) -> bool:
     
             return True
     
    -    def _set_session_env(self, context: SessionContext) -> None:
    -        """Set environment variables for the current session."""
    -        os.environ["HERMES_SESSION_PLATFORM"] = context.source.platform.value
    -        os.environ["HERMES_SESSION_CHAT_ID"] = context.source.chat_id
    -        if context.source.chat_name:
    -            os.environ["HERMES_SESSION_CHAT_NAME"] = context.source.chat_name
    -        if context.source.thread_id:
    -            os.environ["HERMES_SESSION_THREAD_ID"] = str(context.source.thread_id)
    -    
    -    def _clear_session_env(self) -> None:
    -        """Clear session environment variables."""
    -        for var in ["HERMES_SESSION_PLATFORM", "HERMES_SESSION_CHAT_ID", "HERMES_SESSION_CHAT_NAME", "HERMES_SESSION_THREAD_ID"]:
    -            if var in os.environ:
    -                del os.environ[var]
    +    def _set_session_env(self, context: SessionContext) -> list:
    +        """Set session context variables for the current async task.
    +
    +        Uses ``contextvars`` instead of ``os.environ`` so that concurrent
    +        gateway messages cannot overwrite each other's session state.
    +
    +        Returns a list of reset tokens; pass them to ``_clear_session_env``
    +        in a ``finally`` block.
    +        """
    +        from gateway.session_context import set_session_vars
    +        return set_session_vars(
    +            platform=context.source.platform.value,
    +            chat_id=context.source.chat_id,
    +            chat_name=context.source.chat_name or "",
    +            thread_id=str(context.source.thread_id) if context.source.thread_id else "",
    +        )
    +
    +    def _clear_session_env(self, tokens: list) -> None:
    +        """Restore session context variables to their pre-handler values."""
    +        from gateway.session_context import clear_session_vars
    +        clear_session_vars(tokens)
         
         async def _enrich_message_with_vision(
             self,

  • gateway/session_context.py

    @@ -0,0 +1,113 @@
    +"""
    +Session-scoped context variables for the Hermes gateway.
    +
    +Replaces the previous ``os.environ``-based session state
    +(``HERMES_SESSION_PLATFORM``, ``HERMES_SESSION_CHAT_ID``, etc.) with
    +Python's ``contextvars.ContextVar``.
    +
    +**Why this matters**
    +
    +The gateway processes messages concurrently via ``asyncio``.  When two
    +messages arrive at the same time the old code did:
    +
    +    os.environ["HERMES_SESSION_THREAD_ID"] = str(context.source.thread_id)
    +
    +Because ``os.environ`` is *process-global*, Message A's value was
    +silently overwritten by Message B before Message A's agent finished
    +running.  Background-task notifications and tool calls therefore routed
    +to the wrong thread.
    +
    +``contextvars.ContextVar`` values are *task-local*: each ``asyncio``
    +task (and any ``run_in_executor`` thread it spawns) gets its own copy,
    +so concurrent messages never interfere.
    +
    +**Backward compatibility**
    +
    +The public helper ``get_session_env(name, default="")`` mirrors the old
    +``os.getenv("HERMES_SESSION_*", ...)`` calls.  Existing tool code only
    +needs to replace the import + call site:
    +
    +    # before
    +    import os
    +    platform = os.getenv("HERMES_SESSION_PLATFORM", "")
    +
    +    # after
    +    from gateway.session_context import get_session_env
    +    platform = get_session_env("HERMES_SESSION_PLATFORM", "")
    +"""
    +
    +from contextvars import ContextVar
    +
    +# ---------------------------------------------------------------------------
    +# Per-task session variables
    +# ---------------------------------------------------------------------------
    +
    +_SESSION_PLATFORM: ContextVar[str] = ContextVar("HERMES_SESSION_PLATFORM", default="")
    +_SESSION_CHAT_ID: ContextVar[str] = ContextVar("HERMES_SESSION_CHAT_ID", default="")
    +_SESSION_CHAT_NAME: ContextVar[str] = ContextVar("HERMES_SESSION_CHAT_NAME", default="")
    +_SESSION_THREAD_ID: ContextVar[str] = ContextVar("HERMES_SESSION_THREAD_ID", default="")
    +
    +_VAR_MAP = {
    +    "HERMES_SESSION_PLATFORM": _SESSION_PLATFORM,
    +    "HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID,
    +    "HERMES_SESSION_CHAT_NAME": _SESSION_CHAT_NAME,
    +    "HERMES_SESSION_THREAD_ID": _SESSION_THREAD_ID,
    +}
    +
    +
    +def set_session_vars(
    +    platform: str = "",
    +    chat_id: str = "",
    +    chat_name: str = "",
    +    thread_id: str = "",
    +) -> list:
    +    """Set all session context variables and return reset tokens.
    +
    +    Call ``clear_session_vars(tokens)`` in a ``finally`` block to restore
    +    the previous values when the handler exits.
    +
    +    Returns a list of ``Token`` objects (one per variable) that can be
    +    passed to ``clear_session_vars``.
    +    """
    +    tokens = [
    +        _SESSION_PLATFORM.set(platform),
    +        _SESSION_CHAT_ID.set(chat_id),
    +        _SESSION_CHAT_NAME.set(chat_name),
    +        _SESSION_THREAD_ID.set(thread_id),
    +    ]
    +    return tokens
    +
    +
    +def clear_session_vars(tokens: list) -> None:
    +    """Restore session context variables to their pre-handler values."""
    +    if not tokens:
    +        return
    +    vars_in_order = [
    +        _SESSION_PLATFORM,
    +        _SESSION_CHAT_ID,
    +        _SESSION_CHAT_NAME,
    +        _SESSION_THREAD_ID,
    +    ]
    +    for var, token in zip(vars_in_order, tokens):
    +        var.reset(token)
    +
    +
    +def get_session_env(name: str, default: str = "") -> str:
    +    """Read a session context variable by its legacy ``HERMES_SESSION_*`` name.
    +
    +    Drop-in replacement for ``os.getenv("HERMES_SESSION_*", default)``.
    +
    +    Resolution order:
    +    1. Context variable (set by the gateway for concurrency-safe access)
    +    2. ``os.environ`` (used by CLI, cron scheduler, and tests)
    +    3. *default*
    +    """
    +    import os
    +
    +    var = _VAR_MAP.get(name)
    +    if var is not None:
    +        value = var.get()
    +        if value:
    +            return value
    +    # Fall back to os.environ for CLI, cron, and test compatibility
    +    return os.getenv(name, default)

  • tests/gateway/test_session_env.py

    @@ -3,9 +3,15 @@
     from gateway.config import Platform
     from gateway.run import GatewayRunner
     from gateway.session import SessionContext, SessionSource
    +from gateway.session_context import (
    +    get_session_env,
    +    set_session_vars,
    +    clear_session_vars,
    +)
     
     
    -def test_set_session_env_includes_thread_id(monkeypatch):
    +def test_set_session_env_sets_contextvars(monkeypatch):
    +    """_set_session_env should populate contextvars, not os.environ."""
         runner = object.__new__(GatewayRunner)
         source = SessionSource(
             platform=Platform.TELEGRAM,
    @@ -21,25 +27,93 @@ def test_set_session_env_includes_thread_id(monkeypatch):
         monkeypatch.delenv("HERMES_SESSION_CHAT_NAME", raising=False)
         monkeypatch.delenv("HERMES_SESSION_THREAD_ID", raising=False)
     
    -    runner._set_session_env(context)
    +    tokens = [redacted:secret-assignment]
     
    -    assert os.getenv("HERMES_SESSION_PLATFORM") == "telegram"
    -    assert os.getenv("HERMES_SESSION_CHAT_ID") == "-1001"
    -    assert os.getenv("HERMES_SESSION_CHAT_NAME") == "Group"
    -    assert os.getenv("HERMES_SESSION_THREAD_ID") == "17585"
    +    # Values should be readable via get_session_env (contextvar path)
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == "telegram"
    +    assert get_session_env("HERMES_SESSION_CHAT_ID") == "-1001"
    +    assert get_session_env("HERMES_SESSION_CHAT_NAME") == "Group"
    +    assert get_session_env("HERMES_SESSION_THREAD_ID") == "17585"
     
    +    # os.environ should NOT be touched
    +    assert os.getenv("HERMES_SESSION_PLATFORM") is None
    +    assert os.getenv("HERMES_SESSION_THREAD_ID") is None
    +
    +    # Clean up
    +    runner._clear_session_env(tokens)
     
    -def test_clear_session_env_removes_thread_id(monkeypatch):
    +
    +def test_clear_session_env_restores_previous_state(monkeypatch):
    +    """_clear_session_env should restore contextvars to their pre-handler values."""
         runner = object.__new__(GatewayRunner)
     
    -    monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram")
    -    monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "-1001")
    -    monkeypatch.setenv("HERMES_SESSION_CHAT_NAME", "Group")
    -    monkeypatch.setenv("HERMES_SESSION_THREAD_ID", "17585")
    +    monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False)
    +    monkeypatch.delenv("HERMES_SESSION_CHAT_ID", raising=False)
    +    monkeypatch.delenv("HERMES_SESSION_CHAT_NAME", raising=False)
    +    monkeypatch.delenv("HERMES_SESSION_THREAD_ID", raising=False)
    +
    +    source = SessionSource(
    +        platform=Platform.TELEGRAM,
    +        chat_id="-1001",
    +        chat_name="Group",
    +        chat_type="group",
    +        thread_id="17585",
    +    )
    +    context = SessionContext(source=source, connected_platforms=[], home_channels={})
    +
    +    tokens = [redacted:secret-assignment]
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == "telegram"
     
    -    runner._clear_session_env()
    +    runner._clear_session_env(tokens)
     
    -    assert os.getenv("HERMES_SESSION_PLATFORM") is None
    -    assert os.getenv("HERMES_SESSION_CHAT_ID") is None
    -    assert os.getenv("HERMES_SESSION_CHAT_NAME") is None
    -    assert os.getenv("HERMES_SESSION_THREAD_ID") is None
    +    # After clear, contextvars should return to defaults (empty)
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == ""
    +    assert get_session_env("HERMES_SESSION_CHAT_ID") == ""
    +    assert get_session_env("HERMES_SESSION_CHAT_NAME") == ""
    +    assert get_session_env("HERMES_SESSION_THREAD_ID") == ""
    +
    +
    +def test_get_session_env_falls_back_to_os_environ(monkeypatch):
    +    """get_session_env should fall back to os.environ when contextvar is unset."""
    +    monkeypatch.setenv("HERMES_SESSION_PLATFORM", "discord")
    +
    +    # No contextvar set — should read from os.environ
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == "discord"
    +
    +    # Now set a contextvar — should prefer it
    +    tokens = [redacted:secret-assignment]"telegram")
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == "telegram"
    +
    +    # Restore — should fall back to os.environ again
    +    clear_session_vars(tokens)
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == "discord"
    +
    +
    +def test_get_session_env_default_when_nothing_set(monkeypatch):
    +    """get_session_env returns default when neither contextvar nor env is set."""
    +    monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False)
    +
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == ""
    +    assert get_session_env("HERMES_SESSION_PLATFORM", "fallback") == "fallback"
    +
    +
    +def test_set_session_env_handles_missing_optional_fields():
    +    """_set_session_env should handle None chat_name and thread_id gracefully."""
    +    runner = object.__new__(GatewayRunner)
    +    source = SessionSource(
    +        platform=Platform.TELEGRAM,
    +        chat_id="-1001",
    +        chat_name=None,
    +        chat_type="private",
    +        thread_id=None,
    +    )
    +    context = SessionContext(source=source, connected_platforms=[], home_channels={})
    +
    +    tokens = [redacted:secret-assignment]
    +
    +    assert get_session_env("HERMES_SESSION_PLATFORM") == "telegram"
    +    assert get_session_env("HERMES_SESSION_CHAT_ID") == "-1001"
    +    assert get_session_env("HERMES_SESSION_CHAT_NAME") == ""
    +    assert get_session_env("HERMES_SESSION_THREAD_ID") == ""
    +
    +    runner._clear_session_env(tokens)

  • tools/cronjob_tools.py

    @@ -64,14 +64,15 @@ def _scan_cron_prompt(prompt: str) -> str:
     
     
     def _origin_from_env() -> Optional[Dict[str, str]]:
    -    origin_platform = os.getenv("HERMES_SESSION_PLATFORM")
    -    origin_chat_id = os.getenv("HERMES_SESSION_CHAT_ID")
    +    from gateway.session_context import get_session_env
    +    origin_platform = get_session_env("HERMES_SESSION_PLATFORM")
    +    origin_chat_id = get_session_env("HERMES_SESSION_CHAT_ID")
         if origin_platform and origin_chat_id:
             return {
                 "platform": origin_platform,
                 "chat_id": origin_chat_id,
    -            "chat_name": os.getenv("HERMES_SESSION_CHAT_NAME"),
    -            "thread_id": os.getenv("HERMES_SESSION_THREAD_ID"),
    +            "chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None,
    +            "thread_id": get_session_env("HERMES_SESSION_THREAD_ID") or None,
             }
         return None
     

  • tools/send_message_tool.py

    @@ -212,7 +212,8 @@ def _handle_send(args):
             if isinstance(result, dict) and result.get("success") and mirror_text:
                 try:
                     from gateway.mirror import mirror_to_session
    -                source_label = os.getenv("HERMES_SESSION_PLATFORM", "cli")
    +                from gateway.session_context import get_session_env
    +                source_label = get_session_env("HERMES_SESSION_PLATFORM", "cli")
                     if mirror_to_session(platform_name, chat_id, mirror_text, source_label=source_label, thread_id=thread_id):
                         result["mirrored"] = True
                 except Exception:
    @@ -1020,7 +1021,8 @@ async def _send_feishu(pconfig, chat_id, message, media_files=None, thread_id=No
     
     def _check_send_message():
         """Gate send_message on gateway running (always available on messaging platforms)."""
    -    platform = os.getenv("HERMES_SESSION_PLATFORM", "")
    +    from gateway.session_context import get_session_env
    +    platform = get_session_env("HERMES_SESSION_PLATFORM", "")
         if platform and platform != "local":
             return True
         try:

  • tools/skills_tool.py

    @@ -347,7 +347,8 @@ def _capture_required_environment_variables(
     def _is_gateway_surface() -> bool:
         if os.getenv("HERMES_GATEWAY_SESSION"):
             return True
    -    return bool(os.getenv("HERMES_SESSION_PLATFORM"))
    +    from gateway.session_context import get_session_env
    +    return bool(get_session_env("HERMES_SESSION_PLATFORM"))
     
     
     def _get_terminal_backend_name() -> str:

  • tools/terminal_tool.py

    @@ -1420,10 +1420,11 @@ def terminal_tool(
                         # In gateway mode, auto-register a fast watcher so the
                         # gateway can detect completion and trigger a new agent
                         # turn.  CLI mode uses the completion_queue directly.
    -                    _gw_platform = os.getenv("HERMES_SESSION_PLATFORM", "")
    +                    from gateway.session_context import get_session_env as _gse
    +                    _gw_platform = _gse("HERMES_SESSION_PLATFORM", "")
                         if _gw_platform and not check_interval:
    -                        _gw_chat_id = os.getenv("HERMES_SESSION_CHAT_ID", "")
    -                        _gw_thread_id = os.getenv("HERMES_SESSION_THREAD_ID", "")
    +                        _gw_chat_id = _gse("HERMES_SESSION_CHAT_ID", "")
    +                        _gw_thread_id = _gse("HERMES_SESSION_THREAD_ID", "")
                             proc_session.watcher_platform = _gw_platform
                             proc_session.watcher_chat_id = _gw_chat_id
                             proc_session.watcher_thread_id = _gw_thread_id
    @@ -1445,9 +1446,10 @@ def terminal_tool(
                             result_data["check_interval_note"] = (
                                 f"Requested {check_interval}s raised to minimum 30s"
                             )
    -                    watcher_platform = os.getenv("HERMES_SESSION_PLATFORM", "")
    -                    watcher_chat_id = os.getenv("HERMES_SESSION_CHAT_ID", "")
    -                    watcher_thread_id = os.getenv("HERMES_SESSION_THREAD_ID", "")
    +                    from gateway.session_context import get_session_env as _gse2
    +                    watcher_platform = _gse2("HERMES_SESSION_PLATFORM", "")
    +                    watcher_chat_id = _gse2("HERMES_SESSION_CHAT_ID", "")
    +                    watcher_thread_id = _gse2("HERMES_SESSION_THREAD_ID", "")
     
                         # Store on session for checkpoint persistence
                         proc_session.watcher_platform = watcher_platform

  • tools/tts_tool.py

    @@ -480,7 +480,8 @@ def text_to_speech_tool(
         # Telegram voice bubbles require Opus (.ogg); OpenAI and ElevenLabs can
         # produce Opus natively (no ffmpeg needed).  Edge TTS always outputs MP3
         # and needs ffmpeg for conversion.
    -    platform = os.getenv("HERMES_SESSION_PLATFORM", "").lower()
    +    from gateway.session_context import get_session_env
    +    platform = get_session_env("HERMES_SESSION_PLATFORM", "").lower()
         want_opus = (platform == "telegram")
     
         # Determine output path

• #10460 fix(gateway): route watch-pattern notifications to originating session

  merged to main

  files: gateway/run.py, tests/gateway/test_background_process_notifications.py, tests/gateway/test_internal_event_bypass_pairing.py, tests/tools/test_watch_patterns.py, tools/process_registry.py, tools/terminal_tool.py

  Show the diff

  • gateway/run.py

    @@ -482,6 +482,27 @@ def _resolve_hermes_bin() -> Optional[list[str]]:
         return None
     
     
    +def _parse_session_key(session_key: str) -> "dict | None":
    +    """Parse a session key into its component parts.
    +
    +    Session keys follow the format
    +    ``agent:main:{platform}:{chat_type}:{chat_id}[:{thread_id}[:{user_id}]]``.
    +    Returns a dict with ``platform``, ``chat_type``, ``chat_id``, and
    +    optionally ``thread_id`` keys, or None if the key doesn't match.
    +    """
    +    parts = session_key.split(":")
    +    if len(parts) >= 5 and parts[0] == "agent" and parts[1] == "main":
    +        result = {
    +            "platform": parts[2],
    +            "chat_type": parts[3],
    +            "chat_id": parts[4],
    +        }
    +        if len(parts) > 5:
    +            result["thread_id"] = parts[5]
    +        return result
    +    return None
    +
    +
     def _format_gateway_process_notification(evt: dict) -> "str | None":
         """Format a watch pattern event from completion_queue into a [SYSTEM:] message."""
         evt_type = evt.get("type", "completion")
    @@ -1489,12 +1510,11 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
             notified: set = set()
             for session_key in active:
                 # Parse platform + chat_id from the session key.
    -            # Format: agent:main:{platform}:{chat_type}:{chat_id}[:{extra}...]
    -            parts = session_key.split(":")
    -            if len(parts) < 5:
    +            _parsed = _parse_session_key(session_key)
    +            if not _parsed:
                     continue
    -            platform_str = parts[2]
    -            chat_id = parts[4]
    +            platform_str = _parsed["platform"]
    +            chat_id = _parsed["chat_id"]
     
                 # Deduplicate: one notification per chat, even if multiple
                 # sessions (different users/threads) share the same chat.
    @@ -1510,7 +1530,7 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
     
                     # Include thread_id if present so the message lands in the
                     # correct forum topic / thread.
    -                thread_id = parts[5] if len(parts) > 5 else None
    +                thread_id = _parsed.get("thread_id")
                     metadata = {"thread_id": thread_id} if thread_id else None
     
                     await adapter.send(chat_id, msg, metadata=metadata)
    @@ -3958,7 +3978,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
                         synth_text = _format_gateway_process_notification(evt)
                         if synth_text:
                             try:
    -                            await self._inject_watch_notification(synth_text, event)
    +                            await self._inject_watch_notification(synth_text, evt)
                             except Exception as e2:
                                 logger.error("Watch notification injection error: %s", e2)
                 except Exception as e:
    @@ -7452,14 +7472,75 @@ async def _enrich_message_with_transcription(
                 return prefix
             return user_text
     
    -    async def _inject_watch_notification(self, synth_text: str, original_event) -> None:
    +    def _build_process_event_source(self, evt: dict):
    +        """Resolve the canonical source for a synthetic background-process event.
    +
    +        Prefer the persisted session-store origin for the event's session key.
    +        Falling back to the currently active foreground event is what causes
    +        cross-topic bleed, so don't do that.
    +        """
    +        from gateway.session import SessionSource
    +
    +        session_key = str(evt.get("session_key") or "").strip()
    +        derived_platform = ""
    +        derived_chat_type = ""
    +        derived_chat_id = ""
    +
    +        if session_key:
    +            try:
    +                self.session_store._ensure_loaded()
    +                entry = self.session_store._entries.get(session_key)
    +                if entry and getattr(entry, "origin", None):
    +                    return entry.origin
    +            except Exception as exc:
    +                logger.debug(
    +                    "Synthetic process-event session-store lookup failed for %s: %s",
    +                    session_key,
    +                    exc,
    +                )
    +
    +            _parsed = _parse_session_key(session_key)
    +            if _parsed:
    +                derived_platform = _parsed["platform"]
    +                derived_chat_type = _parsed["chat_type"]
    +                derived_chat_id = _parsed["chat_id"]
    +
    +        platform_name = str(evt.get("platform") or derived_platform or "").strip().lower()
    +        chat_type = str(evt.get("chat_type") or derived_chat_type or "").strip().lower()
    +        chat_id = str(evt.get("chat_id") or derived_chat_id or "").strip()
    +        if not platform_name or not chat_type or not chat_id:
    +            return None
    +
    +        try:
    +            platform = Platform(platform_name)
    +        except Exception:
    +            logger.warning(
    +                "Synthetic process event has invalid platform metadata: %r",
    +                platform_name,
    +            )
    +            return None
    +
    +        return SessionSource(
    +            platform=platform,
    +            chat_id=chat_id,
    +            chat_type=chat_type,
    +            thread_id=str(evt.get("thread_id") or "").strip() or None,
    +            user_id=str(evt.get("user_id") or "").strip() or None,
    +            user_name=str(evt.get("user_name") or "").strip() or None,
    +        )
    +
    +    async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None:
             """Inject a watch-pattern notification as a synthetic message event.
     
    -        Uses the source from the original user event to route the notification
    -        back to the correct chat/adapter.
    +        Routing must come from the queued watch event itself, not from whatever
    +        foreground message happened to be active when the queue was drained.
             """
    -        source = getattr(original_event, "source", None)
    +        source = self._build_process_event_source(evt)
             if not source:
    +            logger.warning(
    +                "Dropping watch notification with no routing metadata for process %s",
    +                evt.get("session_id", "unknown"),
    +            )
                 return
             platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform)
             adapter = None
    @@ -7477,7 +7558,12 @@ async def _inject_watch_notification(self, synth_text: str, original_event) -> N
                     source=source,
                     internal=True,
                 )
    -            logger.info("Watch pattern notification — injecting for %s", platform_name)
    +            logger.info(
    +                "Watch pattern notification — injecting for %s chat=%s thread=%s",
    +                platform_name,
    +                source.chat_id,
    +                source.thread_id,
    +            )
                 await adapter.handle_message(synth_event)
             except Exception as e:
                 logger.error("Watch notification injection error: %s", e)
    @@ -7547,33 +7633,42 @@ async def _run_process_watcher(self, watcher: dict) -> None:
                             f"Command: {session.command}\n"
                             f"Output:\n{_out}]"
                         )
    +                    source = self._build_process_event_source({
    +                        "session_id": session_id,
    +                        "session_key": session_key,
    +                        "platform": platform_name,
    +                        "chat_id": chat_id,
    +                        "thread_id": thread_id,
    +                        "user_id": user_id,
    +                        "user_name": user_name,
    +                    })
    +                    if not source:
    +                        logger.warning(
    +                            "Dropping completion notification with no routing metadata for process %s",
    +                            session_id,
    +                        )
    +                        break
    +
                         adapter = None
                         for p, a in self.adapters.items():
    -                        if p.value == platform_name:
    +                        if p == source.platform:
                                 adapter = a
                                 break
    -                    if adapter and chat_id:
    +                    if adapter and source.chat_id:
                             try:
                                 from gateway.platforms.base import MessageEvent, MessageType
    -                            from gateway.session import SessionSource
    -                            from gateway.config import Platform
    -                            _platform_enum = Platform(platform_name)
    -                            _source = SessionSource(
    -                                platform=_platform_enum,
    -                                chat_id=chat_id,
    -                                thread_id=thread_id or None,
    -                                user_id=user_id or None,
    -                                user_name=user_name or None,
    -                            )
                                 synth_event = MessageEvent(
                                     text=synth_text,
                                     message_type=MessageType.TEXT,
    -                                source=_source,
    +                                source=source,
                                     internal=True,
                                 )
                                 logger.info(
    -                                "Process %s finished — injecting agent notification for session %s",
    -                                session_id, session_key,
    +                                "Process %s finished — injecting agent notification for session %s chat=%s thread=%s",
    +                                session_id,
    +                                session_key,
    +                                source.chat_id,
    +                                source.thread_id,
                                 )
                                 await adapter.handle_message(synth_event)
                             except Exception as e:

  • tests/gateway/test_background_process_notifications.py

    @@ -14,7 +14,7 @@
     import pytest
     
     from gateway.config import GatewayConfig, Platform
    -from gateway.run import GatewayRunner
    +from gateway.run import GatewayRunner, _parse_session_key
     
     
     # ---------------------------------------------------------------------------
    @@ -45,7 +45,7 @@ def _build_runner(monkeypatch, tmp_path, mode: str) -> GatewayRunner:
         monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
     
         runner = GatewayRunner(GatewayConfig())
    -    adapter = SimpleNamespace(send=AsyncMock())
    +    adapter = SimpleNamespace(send=AsyncMock(), handle_message=AsyncMock())
         runner.adapters[Platform.TELEGRAM] = adapter
         return runner
     
    @@ -243,3 +243,162 @@ async def _instant_sleep(*_a, **_kw):
         assert adapter.send.await_count == 1
         _, kwargs = adapter.send.call_args
         assert kwargs["metadata"] is None
    +
    +
    [redacted:email]
    +async def test_inject_watch_notification_routes_from_session_store_origin(monkeypatch, tmp_path):
    +    from gateway.session import SessionSource
    +
    +    runner = _build_runner(monkeypatch, tmp_path, "all")
    +    adapter = runner.adapters[Platform.TELEGRAM]
    +    runner.session_store._entries["agent:main:telegram:group:-100:42"] = SimpleNamespace(
    +        origin=SessionSource(
    +            platform=Platform.TELEGRAM,
    +            chat_id="-100",
    +            chat_type="group",
    +            thread_id="42",
    +            user_id="123",
    +            user_name="Emiliyan",
    +        )
    +    )
    +
    +    evt = {
    +        "session_id": "proc_watch",
    +        "session_key": "agent:main:telegram:group:-100:42",
    +    }
    +
    +    await runner._inject_watch_notification("[SYSTEM: Background process matched]", evt)
    +
    +    adapter.handle_message.assert_awaited_once()
    +    synth_event = adapter.handle_message.await_args.args[0]
    +    assert synth_event.internal is True
    +    assert synth_event.source.platform == Platform.TELEGRAM
    +    assert synth_event.source.chat_id == "-100"
    +    assert synth_event.source.chat_type == "group"
    +    assert synth_event.source.thread_id == "42"
    +    assert synth_event.source.user_id == "123"
    +    assert synth_event.source.user_name == "Emiliyan"
    +
    +
    +def test_build_process_event_source_falls_back_to_session_key_chat_type(monkeypatch, tmp_path):
    +    runner = _build_runner(monkeypatch, tmp_path, "all")
    +
    +    evt = {
    +        "session_id": "proc_watch",
    +        "session_key": "agent:main:telegram:group:-100:42",
    +        "platform": "telegram",
    +        "chat_id": "-100",
    +        "thread_id": "42",
    +        "user_id": "123",
    +        "user_name": "Emiliyan",
    +    }
    +
    +    source = runner._build_process_event_source(evt)
    +
    +    assert source is not None
    +    assert source.platform == Platform.TELEGRAM
    +    assert source.chat_id == "-100"
    +    assert source.chat_type == "group"
    +    assert source.thread_id == "42"
    +    assert source.user_id == "123"
    +    assert source.user_name == "Emiliyan"
    +
    +
    [redacted:email]
    +async def test_inject_watch_notification_ignores_foreground_event_source(monkeypatch, tmp_path):
    +    """Negative test: watch notification must NOT route to the foreground thread."""
    +    from gateway.session import SessionSource
    +
    +    runner = _build_runner(monkeypatch, tmp_path, "all")
    +    adapter = runner.adapters[Platform.TELEGRAM]
    +
    +    # Session store has the process's original thread (thread 42)
    +    runner.session_store._entries["agent:main:telegram:group:-100:42"] = SimpleNamespace(
    +        origin=SessionSource(
    +            platform=Platform.TELEGRAM,
    +            chat_id="-100",
    +            chat_type="group",
    +            thread_id="42",
    +            user_id="proc_owner",
    +            user_name="alice",
    +        )
    +    )
    +
    +    # The evt dict carries the correct session_key — NOT a foreground event
    +    evt = {
    +        "session_id": "proc_cross_thread",
    +        "session_key": "agent:main:telegram:group:-100:42",
    +    }
    +
    +    await runner._inject_watch_notification("[SYSTEM: watch match]", evt)
    +
    +    adapter.handle_message.assert_awaited_once()
    +    synth_event = adapter.handle_message.await_args.args[0]
    +    # Must route to thread 42 (process origin), NOT some other thread
    +    assert synth_event.source.thread_id == "42"
    +    assert synth_event.source.user_id == "proc_owner"
    +
    +
    +def test_build_process_event_source_returns_none_for_empty_evt(monkeypatch, tmp_path):
    +    """Missing session_key and no platform metadata → None (drop notification)."""
    +    runner = _build_runner(monkeypatch, tmp_path, "all")
    +
    +    source = runner._build_process_event_source({"session_id": "proc_orphan"})
    +    assert source is None
    +
    +
    +def test_build_process_event_source_returns_none_for_invalid_platform(monkeypatch, tmp_path):
    +    """Invalid platform string → None."""
    +    runner = _build_runner(monkeypatch, tmp_path, "all")
    +
    +    evt = {
    +        "session_id": "proc_bad",
    +        "platform": "not_a_real_platform",
    +        "chat_type": "dm",
    +        "chat_id": "123",
    +    }
    +    source = runner._build_process_event_source(evt)
    +    assert source is None
    +
    +
    +def test_build_process_event_source_returns_none_for_short_session_key(monkeypatch, tmp_path):
    +    """Session key with <5 parts doesn't parse, falls through to empty metadata → None."""
    +    runner = _build_runner(monkeypatch, tmp_path, "all")
    +
    +    evt = {
    +        "session_id": "proc_short",
    +        "session_key": "agent:main:telegram",  # Too few parts
    +    }
    +    source = runner._build_process_event_source(evt)
    +    assert source is None
    +
    +
    +# ---------------------------------------------------------------------------
    +# _parse_session_key helper
    +# ---------------------------------------------------------------------------
    +
    +def test_parse_session_key_valid():
    +    result = _parse_session_key("agent:main:telegram:group:-100")
    +    assert result == {"platform": "telegram", "chat_type": "group", "chat_id": "-100"}
    +
    +
    +def test_parse_session_key_with_extra_parts():
    +    """Thread ID (6th part) is extracted; further parts are ignored."""
    +    result = _parse_session_key("agent:main:discord:group:chan123:thread456")
    +    assert result == {"platform": "discord", "chat_type": "group", "chat_id": "chan123", "thread_id": "thread456"}
    +
    +
    +def test_parse_session_key_with_user_id_part():
    +    """7th part (user_id) is ignored — only up to thread_id is extracted."""
    +    result = _parse_session_key("agent:main:telegram:group:chat1:thread42:user99")
    +    assert result == {"platform": "telegram", "chat_type": "group", "chat_id": "chat1", "thread_id": "thread42"}
    +
    +
    +def test_parse_session_key_too_short():
    +    assert _parse_session_key("agent:main:telegram") is None
    +    assert _parse_session_key("") is None
    +
    +
    +def test_parse_session_key_wrong_prefix():
    +    assert _parse_session_key("cron:main:telegram:dm:123") is None
    +    assert _parse_session_key("agent:cron:telegram:dm:123") is None

  • tests/gateway/test_internal_event_bypass_pairing.py

    @@ -230,6 +230,59 @@ async def _instant_sleep(*_a, **_kw):
         assert event.source.user_name == "alice"
     
     
    [redacted:email]
    +async def test_notify_on_complete_uses_session_store_origin_for_group_topic(monkeypatch, tmp_path):
    +    import tools.process_registry as pr_module
    +    from gateway.session import SessionSource
    +
    +    sessions = [
    +        SimpleNamespace(
    +            output_buffer="done\n", exited=True, exit_code=0, command="echo test"
    +        ),
    +    ]
    +    monkeypatch.setattr(pr_module, "process_registry", _FakeRegistry(sessions))
    +
    +    async def _instant_sleep(*_a, **_kw):
    +        pass
    +    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
    +
    +    runner = GatewayRunner(GatewayConfig())
    +    adapter = SimpleNamespace(send=AsyncMock(), handle_message=AsyncMock())
    +    runner.adapters[Platform.TELEGRAM] = adapter
    +    runner.session_store._entries["agent:main:telegram:group:-100:42"] = SimpleNamespace(
    +        origin=SessionSource(
    +            platform=Platform.TELEGRAM,
    +            chat_id="-100",
    +            chat_type="group",
    +            thread_id="42",
    +            user_id="user-42",
    +            user_name="alice",
    +        )
    +    )
    +
    +    watcher = {
    +        "session_id": "proc_test_internal",
    +        "check_interval": 0,
    +        "session_key": "agent:main:telegram:group:-100:42",
    +        "platform": "telegram",
    +        "chat_id": "-100",
    +        "thread_id": "42",
    +        "notify_on_complete": True,
    +    }
    +
    +    await runner._run_process_watcher(watcher)
    +
    +    assert adapter.handle_message.await_count == 1
    +    event = adapter.handle_message.await_args.args[0]
    +    assert event.internal is True
    +    assert event.source.platform == Platform.TELEGRAM
    +    assert event.source.chat_id == "-100"
    +    assert event.source.chat_type == "group"
    +    assert event.source.thread_id == "42"
    +    assert event.source.user_id == "user-42"
    +    assert event.source.user_name == "alice"
    +
    +
     @pytest.mark.asyncio
     async def test_none_user_id_skips_pairing(monkeypatch, tmp_path):
         """A non-internal event with user_id=None should be silently dropped."""

  • tests/tools/test_watch_patterns.py

    @@ -92,6 +92,25 @@ def test_basic_match(self, registry):
             assert "disk full" in evt["output"]
             assert evt["session_id"] == "proc_test_watch"
     
    +    def test_match_carries_session_key_and_watcher_routing_metadata(self, registry):
    +        session = _make_session(watch_patterns=["ERROR"])
    +        session.session_key = "agent:main:telegram:group:-100:42"
    +        session.watcher_platform = "telegram"
    +        session.watcher_chat_id = "-100"
    +        session.watcher_user_id = "u123"
    +        session.watcher_user_name = "alice"
    +        session.watcher_thread_id = "42"
    +
    +        registry._check_watch_patterns(session, "ERROR: disk full\n")
    +        evt = registry.completion_queue.get_nowait()
    +
    +        assert evt["session_key"] == "agent:main:telegram:group:-100:42"
    +        assert evt["platform"] == "telegram"
    +        assert evt["chat_id"] == "-100"
    +        assert evt["user_id"] == "u123"
    +        assert evt["user_name"] == "alice"
    +        assert evt["thread_id"] == "42"
    +
         def test_multiple_patterns(self, registry):
             """First matching pattern is reported."""
             session = _make_session(watch_patterns=["WARN", "ERROR"])

  • tools/process_registry.py

    @@ -191,9 +191,15 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None:
                         session._watch_disabled = True
                         self.completion_queue.put({
                             "session_id": session.id,
    +                        "session_key": session.session_key,
                             "command": session.command,
                             "type": "watch_disabled",
                             "suppressed": session._watch_suppressed,
    +                        "platform": session.watcher_platform,
    +                        "chat_id": session.watcher_chat_id,
    +                        "user_id": session.watcher_user_id,
    +                        "user_name": session.watcher_user_name,
    +                        "thread_id": session.watcher_thread_id,
                             "message": (
                                 f"Watch patterns disabled for process {session.id} — "
                                 f"too many matches ({session._watch_suppressed} suppressed). "
    @@ -219,11 +225,17 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None:
     
             self.completion_queue.put({
                 "session_id": session.id,
    +            "session_key": session.session_key,
                 "command": session.command,
                 "type": "watch_match",
                 "pattern": matched_pattern,
                 "output": output,
                 "suppressed": suppressed,
    +            "platform": session.watcher_platform,
    +            "chat_id": session.watcher_chat_id,
    +            "user_id": session.watcher_user_id,
    +            "user_name": session.watcher_user_name,
    +            "thread_id": session.watcher_thread_id,
             })
     
         @staticmethod

  • tools/terminal_tool.py

    @@ -1384,14 +1384,10 @@ def terminal_tool(
                     if pty_disabled_reason:
                         result_data["pty_note"] = pty_disabled_reason
     
    -                # Mark for agent notification on completion
    -                if notify_on_complete and background:
    -                    proc_session.notify_on_complete = True
    -                    result_data["notify_on_complete"] = True
    -
    -                    # In gateway mode, auto-register a fast watcher so the
    -                    # gateway can detect completion and trigger a new agent
    -                    # turn.  CLI mode uses the completion_queue directly.
    +                # Populate routing metadata on the session so that
    +                # watch-pattern and completion notifications can be
    +                # routed back to the correct chat/thread.
    +                if background and (notify_on_complete or watch_patterns):
                         from gateway.session_context import get_session_env as _gse
                         _gw_platform = _gse("HERMES_SESSION_PLATFORM", "")
                         if _gw_platform:
    @@ -1404,16 +1400,26 @@ def terminal_tool(
                             proc_session.watcher_user_id = _gw_user_id
                             proc_session.watcher_user_name = _gw_user_name
                             proc_session.watcher_thread_id = _gw_thread_id
    +
    +                # Mark for agent notification on completion
    +                if notify_on_complete and background:
    +                    proc_session.notify_on_complete = True
    +                    result_data["notify_on_complete"] = True
    +
    +                    # In gateway mode, auto-register a fast watcher so the
    +                    # gateway can detect completion and trigger a new agent
    +                    # turn.  CLI mode uses the completion_queue directly.
    +                    if proc_session.watcher_platform:
                             proc_session.watcher_interval = 5
                             process_registry.pending_watchers.append({
                                 "session_id": proc_session.id,
                                 "check_interval": 5,
                                 "session_key": session_key,
    -                            "platform": _gw_platform,
    -                            "chat_id": _gw_chat_id,
    -                            "user_id": _gw_user_id,
    -                            "user_name": _gw_user_name,
    -                            "thread_id": _gw_thread_id,
    +                            "platform": proc_session.watcher_platform,
    +                            "chat_id": proc_session.watcher_chat_id,
    +                            "user_id": proc_session.watcher_user_id,
    +                            "user_name": proc_session.watcher_user_name,
    +                            "thread_id": proc_session.watcher_thread_id,
                                 "notify_on_complete": True,
                             })
     

• #37099 feat(desktop): session hygiene, archive, media streaming + connecting overlay

  merged to main

  files: apps/desktop/electron/main.cjs, apps/desktop/src/app/chat/index.tsx, apps/desktop/src/app/chat/sidebar/index.tsx, apps/desktop/src/app/chat/sidebar/session-actions-menu.tsx, apps/desktop/src/app/chat/sidebar/session-row.tsx, apps/desktop/src/app/chat/sidebar/virtual-session-list.tsx, apps/desktop/src/app/command-center/index.tsx, apps/desktop/src/app/desktop-controller.tsx, apps/desktop/src/app/session/hooks/use-session-actions.ts, apps/desktop/src/app/settings/constants.ts, apps/desktop/src/app/settings/index.tsx, apps/desktop/src/app/settings/sessions-settings.tsx, apps/desktop/src/app/settings/types.ts, apps/desktop/src/components/assistant-ui/markdown-text.tsx, apps/desktop/src/components/gateway-connecting-overlay.tsx, apps/desktop/src/hermes.ts, apps/desktop/src/lib/icons.ts, apps/desktop/src/lib/media.ts, apps/desktop/src/types/hermes.ts, hermes_cli/web_server.py (+6 more)

  Show the diff

  • apps/desktop/electron/main.cjs

    @@ -8,6 +8,8 @@ const {
       ipcMain,
       nativeImage,
       nativeTheme,
    +  net: electronNet,
    +  protocol,
       safeStorage,
       session,
       shell,
    @@ -364,6 +366,66 @@ app.setAboutPanelOptions({
       copyright: 'Copyright © 2026 Nous Research'
     })
     
    +// Custom scheme for streaming local media (video/audio) into the renderer.
    +// Reading large media through `readFileDataUrl` failed: it base64-loads the
    +// whole file into memory and is hard-capped at DATA_URL_READ_MAX_BYTES (16 MB),
    +// so any non-trivial video silently refused to load. Streaming via a protocol
    +// handler removes the size cap and gives the <video> element seekable,
    +// range-aware playback. Must be registered before the app is ready.
    +const MEDIA_PROTOCOL = 'hermes-media'
    +// Only audio/video may be streamed. Without this the handler would read any
    +// non-blocklisted local file (no size cap) for any `fetch(hermes-media://…)`.
    +const STREAMABLE_MEDIA_EXTS = new Set([
    +  '.avi',
    +  '.flac',
    +  '.m4a',
    +  '.mkv',
    +  '.mov',
    +  '.mp3',
    +  '.mp4',
    +  '.ogg',
    +  '.opus',
    +  '.wav',
    +  '.webm'
    +])
    +
    +protocol.registerSchemesAsPrivileged([
    +  {
    +    scheme: MEDIA_PROTOCOL,
    +    privileges: {
    +      secure: true,
    +      standard: true,
    +      stream: true,
    +      supportFetchAPI: true
    +    }
    +  }
    +])
    +
    +function registerMediaProtocol() {
    +  protocol.handle(MEDIA_PROTOCOL, async request => {
    +    let resolvedPath
    +    try {
    +      const url = new URL(request.url)
    +      const filePath = decodeURIComponent(url.pathname.replace(/^\/+/, ''))
    +      ;({ resolvedPath } = await resolveReadableFileForIpc(filePath, { purpose: 'Media stream' }))
    +    } catch {
    +      return new Response('Media not found', { status: 404 })
    +    }
    +
    +    if (!STREAMABLE_MEDIA_EXTS.has(path.extname(resolvedPath).toLowerCase())) {
    +      return new Response('Unsupported media type', { status: 415 })
    +    }
    +
    +    // Delegate to Electron's net stack on a file:// URL — it resolves the
    +    // content-type and honors Range requests so seeking works. Forward the
    +    // renderer's headers (notably Range) and skip custom-protocol re-entry.
    +    return electronNet.fetch(pathToFileURL(resolvedPath).toString(), {
    +      bypassCustomProtocolHandlers: true,
    +      headers: request.headers
    +    })
    +  })
    +}
    +
     let mainWindow = null
     let hermesProcess = null
     let connectionPromise = null
    @@ -3654,6 +3716,7 @@ app.whenReady().then(() => {
         Menu.setApplicationMenu(null)
       }
       installMediaPermissions()
    +  registerMediaProtocol()
       ensureWslWindowsFonts()
       createWindow()
     

  • apps/desktop/src/app/chat/index.tsx

    @@ -97,7 +97,7 @@ function ChatHeader({
       const sessions = useStore($sessions)
       const pinnedSessionIds = useStore($pinnedSessionIds)
       const activeStoredSession = sessions.find(session => session.id === selectedSessionId) || null
    -  const title = activeStoredSession ? sessionTitle(activeStoredSession) : 'New agent'
    +  const title = activeStoredSession ? sessionTitle(activeStoredSession) : 'New session'
       const selectedIsPinned = selectedSessionId ? pinnedSessionIds.includes(selectedSessionId) : false
     
       return (

  • apps/desktop/src/app/chat/sidebar/index.tsx

    @@ -67,7 +67,12 @@ import { VirtualSessionList } from './virtual-session-list'
     const VIRTUALIZE_THRESHOLD = 25
     
     const SIDEBAR_NAV: SidebarNavItem[] = [
    -  { id: 'new-session', label: 'New agent', icon: props => <Codicon name="robot" {...props} />, action: 'new-session' },
    +  {
    +    id: 'new-session',
    +    label: 'New session',
    +    icon: props => <Codicon name="robot" {...props} />,
    +    action: 'new-session'
    +  },
       { id: 'skills', label: 'Skills', icon: props => <Codicon name="symbol-misc" {...props} />, route: SKILLS_ROUTE },
       { id: 'messaging', label: 'Messaging', icon: props => <Codicon name="comment" {...props} />, route: MESSAGING_ROUTE },
       { id: 'artifacts', label: 'Artifacts', icon: props => <Codicon name="files" {...props} />, route: ARTIFACTS_ROUTE }
    @@ -149,6 +154,8 @@ interface ChatSidebarProps extends React.ComponentProps<typeof Sidebar> {
       onLoadMoreSessions: () => void
       onResumeSession: (sessionId: string) => void
       onDeleteSession: (sessionId: string) => void
    +  onArchiveSession: (sessionId: string) => void
    +  onNewSessionInWorkspace: (path: null | string) => void
     }
     
     export function ChatSidebar({
    @@ -156,7 +163,9 @@ export function ChatSidebar({
       onNavigate,
       onLoadMoreSessions,
       onResumeSession,
    -  onDeleteSession
    +  onDeleteSession,
    +  onArchiveSession,
    +  onNewSessionInWorkspace
     }: ChatSidebarProps) {
       const sidebarOpen = useStore($sidebarOpen)
       const agentsGrouped = useStore($sidebarAgentsGrouped)
    @@ -328,6 +337,7 @@ export function ChatSidebar({
                 dndSensors={dndSensors}
                 emptyState={<SidebarPinnedEmptyState />}
                 label="Pinned"
    +            onArchiveSession={onArchiveSession}
                 onDeleteSession={onDeleteSession}
                 onReorder={handlePinnedDragEnd}
                 onResumeSession={onResumeSession}
    @@ -361,9 +371,9 @@ export function ChatSidebar({
                 groups={agentsGrouped ? agentGroups : undefined}
                 headerAction={
                   <Button
    -                aria-label={agentsGrouped ? 'Show agents as a single list' : 'Group agents by workspace'}
    +                aria-label={agentsGrouped ? 'Show sessions as a single list' : 'Group sessions by workspace'}
                     className={cn(
    -                  'cursor-pointer text-(--ui-text-tertiary) opacity-0 hover:bg-(--ui-control-hover-background) hover:text-foreground hover:opacity-100 focus-visible:opacity-100 group-hover/section:opacity-100',
    +                  'cursor-pointer text-(--ui-text-tertiary) opacity-70 hover:bg-(--ui-control-hover-background) hover:text-foreground hover:opacity-100 focus-visible:opacity-100',
                       agentsGrouped && 'bg-(--ui-control-active-background) text-foreground opacity-100'
                     )}
                     onClick={event => {
    @@ -372,15 +382,17 @@ export function ChatSidebar({
                       setSidebarAgentsGrouped(!agentsGrouped)
                     }}
                     size="icon-xs"
    -                title={agentsGrouped ? 'Ungroup agents' : 'Group by workspace'}
    +                title={agentsGrouped ? 'Ungroup sessions' : 'Group by workspace'}
                     variant="ghost"
                   >
                     <Codicon name={agentsGrouped ? 'list-unordered' : 'root-folder'} size="0.75rem" />
                   </Button>
                 }
    -            label="Agents"
    +            label="Sessions"
                 labelMeta={countLabel(agentSessions.length, knownSessionTotal)}
    +            onArchiveSession={onArchiveSession}
                 onDeleteSession={onDeleteSession}
    +            onNewSessionInWorkspace={onNewSessionInWorkspace}
                 onReorder={handleAgentDragEnd}
                 onResumeSession={onResumeSession}
                 onToggle={() => setSidebarRecentsOpen(!agentsOpen)}
    @@ -472,7 +484,9 @@ interface SidebarSessionsSectionProps {
       workingSessionIdSet: Set<string>
       onResumeSession: (sessionId: string) => void
       onDeleteSession: (sessionId: string) => void
    +  onArchiveSession: (sessionId: string) => void
       onTogglePin: (sessionId: string) => void
    +  onNewSessionInWorkspace?: (path: null | string) => void
       pinned: boolean
       rootClassName?: string
       contentClassName?: string
    @@ -496,7 +510,9 @@ function SidebarSessionsSection({
       workingSessionIdSet,
       onResumeSession,
       onDeleteSession,
    +  onArchiveSession,
       onTogglePin,
    +  onNewSessionInWorkspace,
       pinned,
       rootClassName,
       contentClassName,
    @@ -518,6 +534,7 @@ function SidebarSessionsSection({
           isPinned: pinned,
           isSelected: session.id === activeSessionId,
           isWorking: workingSessionIdSet.has(session.id),
    +      onArchive: () => onArchiveSession(session.id),
           onDelete: () => onDeleteSession(session.id),
           onPin: () => onTogglePin(session.id),
           onResume: () => onResumeSession(session.id),
    @@ -551,9 +568,19 @@ function SidebarSessionsSection({
       } else if (groups?.length) {
         const groupNodes = groups.map(group =>
           dndActive ? (
    -        <SortableSidebarWorkspaceGroup group={group} key={group.id} renderRows={renderSessionList} />
    +        <SortableSidebarWorkspaceGroup
    +          group={group}
    +          key={group.id}
    +          onNewSession={onNewSessionInWorkspace}
    +          renderRows={renderSessionList}
    +        />
           ) : (
    -        <SidebarWorkspaceGroup group={group} key={group.id} renderRows={renderSessionList} />
    +        <SidebarWorkspaceGroup
    +          group={group}
    +          key={group.id}
    +          onNewSession={onNewSessionInWorkspace}
    +          renderRows={renderSessionList}
    +        />
           )
         )
     
    @@ -568,6 +595,7 @@ function SidebarSessionsSection({
         inner = (
           <VirtualSessionList
             activeSessionId={activeSessionId}
    +        onArchiveSession={onArchiveSession}
             onDeleteSession={onDeleteSession}
             onResumeSession={onResumeSession}
             onTogglePin={onTogglePin}
    @@ -610,6 +638,7 @@ function SidebarSessionsSection({
     interface SidebarWorkspaceGroupProps extends React.ComponentProps<'div'> {
       group: SidebarSessionGroup
       renderRows: (sessions: SessionInfo[]) => React.ReactNode
    +  onNewSession?: (path: null | string) => void
       reorderable?: boolean
       dragging?: boolean
       dragHandleProps?: React.HTMLAttributes<HTMLElement>
    @@ -618,6 +647,7 @@ interface SidebarWorkspaceGroupProps extends React.ComponentProps<'div'> {
     function SidebarWorkspaceGroup({
       group,
       renderRows,
    +  onNewSession,
       reorderable = false,
       dragging = false,
       dragHandleProps,
    @@ -634,18 +664,31 @@ function SidebarWorkspaceGroup({
     
       return (
         <div className={cn('grid gap-px', dragging && 'z-10 opacity-60', className)} ref={ref} style={style} {...rest}>
    -      <button
    -        className="group/workspace flex min-h-6 cursor-pointer items-center gap-1 px-2 pt-1 text-left text-[0.6875rem] font-medium text-(--ui-text-tertiary) hover:text-(--ui-text-secondary)"
    -        onClick={() => setOpen(value => !value)}
    -        title={group.path ?? undefined}
    -        type="button"
    -      >
    -        <span className="truncate">{group.label}</span>
    -        <SidebarCount>{group.sessions.length}</SidebarCount>
    -        <DisclosureCaret
    -          className="text-(--ui-text-tertiary) opacity-0 transition group-hover/workspace:opacity-100"
    -          open={open}
    -        />
    +      <div className="group/workspace flex min-h-6 items-center gap-1 px-2 pt-1 text-[0.6875rem] font-medium text-(--ui-text-tertiary)">
    +        <button
    +          className="flex min-w-0 cursor-pointer items-center gap-1 bg-transparent text-left hover:text-(--ui-text-secondary)"
    +          onClick={() => setOpen(value => !value)}
    +          title={group.path ?? undefined}
    +          type="button"
    +        >
    +          <span className="truncate">{group.label}</span>
    +          <SidebarCount>{group.sessions.length}</SidebarCount>
    +          <DisclosureCaret
    +            className="text-(--ui-text-tertiary) opacity-0 transition group-hover/workspace:opacity-100"
    +            open={open}
    +          />
    +        </button>
    +        {onNewSession && (
    +          <button
    +            aria-label={`New session in ${group.label}`}
    +            className="grid size-4 shrink-0 cursor-pointer place-items-center rounded-sm bg-transparent text-(--ui-text-quaternary) opacity-0 transition-opacity hover:bg-(--ui-control-hover-background) hover:text-foreground group-hover/workspace:opacity-100"
    +            onClick={() => onNewSession(group.path)}
    +            title={`New session in ${group.label}`}
    +            type="button"
    +          >
    +            <Codicon name="add" size="0.75rem" />
    +          </button>
    +        )}
             {reorderable && (
               <span
                 {...dragHandleProps}
    @@ -663,7 +706,7 @@ function SidebarWorkspaceGroup({
                 />
               </span>
             )}
    -      </button>
    +      </div>
           {open && (
             <>
               {renderRows(visibleSessions)}
    @@ -687,6 +730,7 @@ function SidebarWorkspaceGroup({
     interface SortableWorkspaceProps {
       group: SidebarSessionGroup
       renderRows: (sessions: SessionInfo[]) => React.ReactNode
    +  onNewSession?: (path: null | string) => void
     }
     
     function SortableSidebarWorkspaceGroup(props: SortableWorkspaceProps) {
    @@ -702,6 +746,7 @@ interface SortableSessionRowProps {
       isPinned: boolean
       isSelected: boolean
       isWorking: boolean
    +  onArchive: () => void
       onDelete: () => void
       onPin: () => void
       onResume: () => void

  • apps/desktop/src/app/chat/sidebar/session-actions-menu.tsx

    @@ -26,6 +26,7 @@ interface SessionActions {
       title: string
       pinned?: boolean
       onPin?: () => void
    +  onArchive?: () => void
       onDelete?: () => void
     }
     
    @@ -40,7 +41,7 @@ interface ItemSpec {
       variant?: 'destructive'
     }
     
    -function useSessionActions({ sessionId, title, pinned = false, onPin, onDelete }: SessionActions) {
    +function useSessionActions({ sessionId, title, pinned = false, onPin, onArchive, onDelete }: SessionActions) {
       const [renameOpen, setRenameOpen] = useState(false)
     
       const items: ItemSpec[] = [
    @@ -81,6 +82,15 @@ function useSessionActions({ sessionId, title, pinned = false, onPin, onDelete }
             setRenameOpen(true)
           }
         },
    +    {
    +      disabled: !onArchive,
    +      icon: 'archive',
    +      label: 'Archive',
    +      onSelect: () => {
    +        triggerHaptic('selection')
    +        onArchive?.()
    +      }
    +    },
         {
           className: 'text-destructive focus:text-destructive',
           disabled: !onDelete,

  • apps/desktop/src/app/chat/sidebar/session-row.tsx

    @@ -14,6 +14,7 @@ interface SidebarSessionRowProps extends React.ComponentProps<'div'> {
       isPinned: boolean
       isSelected: boolean
       isWorking: boolean
    +  onArchive: () => void
       onDelete: () => void
       onPin: () => void
       onResume: () => void
    @@ -45,6 +46,7 @@ export function SidebarSessionRow({
       isPinned,
       isSelected,
       isWorking,
    +  onArchive,
       onDelete,
       onPin,
       onResume,
    @@ -61,7 +63,14 @@ export function SidebarSessionRow({
       const handleLabel = `Reorder ${title}`
     
       return (
    -    <SessionContextMenu onDelete={onDelete} onPin={onPin} pinned={isPinned} sessionId={session.id} title={title}>
    +    <SessionContextMenu
    +      onArchive={onArchive}
    +      onDelete={onDelete}
    +      onPin={onPin}
    +      pinned={isPinned}
    +      sessionId={session.id}
    +      title={title}
    +    >
           <div
             className={cn(
               'group relative grid min-h-[1.625rem] cursor-pointer grid-cols-[minmax(0,1fr)_1.375rem] items-center rounded-md transition-colors duration-100 ease-out hover:bg-(--ui-row-hover-background) hover:transition-none',
    @@ -88,6 +97,15 @@ export function SidebarSessionRow({
                   return
                 }
     
    +            if (event.metaKey || event.ctrlKey) {
    +              event.preventDefault()
    +              event.stopPropagation()
    +              triggerHaptic('selection')
    +              onArchive()
    +
    +              return
    +            }
    +
                 onResume()
               }}
               type="button"
    @@ -127,7 +145,14 @@ export function SidebarSessionRow({
                   {age}
                 </span>
               )}
    -          <SessionActionsMenu onDelete={onDelete} onPin={onPin} pinned={isPinned} sessionId={session.id} title={title}>
    +          <SessionActionsMenu
    +            onArchive={onArchive}
    +            onDelete={onDelete}
    +            onPin={onPin}
    +            pinned={isPinned}
    +            sessionId={session.id}
    +            title={title}
    +          >
                 <Button
                   aria-label={`Actions for ${title}`}
                   className="size-5 rounded-md bg-transparent text-transparent transition-colors duration-100 hover:bg-(--ui-control-active-background) hover:text-foreground focus-visible:bg-(--ui-control-active-background) focus-visible:text-foreground focus-visible:ring-0 data-[state=open]:bg-(--ui-control-active-background) data-[state=open]:text-foreground group-hover:text-(--ui-text-tertiary) [&_svg]:size-3.5!"

  • apps/desktop/src/app/chat/sidebar/virtual-session-list.tsx

    @@ -12,6 +12,7 @@ interface SessionRowCommonProps {
       isPinned: boolean
       isSelected: boolean
       isWorking: boolean
    +  onArchive: () => void
       onDelete: () => void
       onPin: () => void
       onResume: () => void
    @@ -20,6 +21,7 @@ interface SessionRowCommonProps {
     interface VirtualSessionListProps {
       activeSessionId: null | string
       className?: string
    +  onArchiveSession: (sessionId: string) => void
       onDeleteSession: (sessionId: string) => void
       onResumeSession: (sessionId: string) => void
       onTogglePin: (sessionId: string) => void
    @@ -35,6 +37,7 @@ const OVERSCAN_ROWS = 12
     export const VirtualSessionList: FC<VirtualSessionListProps> = ({
       activeSessionId,
       className,
    +  onArchiveSession,
       onDeleteSession,
       onResumeSession,
       onTogglePin,
    @@ -72,6 +75,7 @@ export const VirtualSessionList: FC<VirtualSessionListProps> = ({
           isPinned: pinned,
           isSelected: session.id === activeSessionId,
           isWorking: workingSessionIdSet.has(session.id),
    +      onArchive: () => onArchiveSession(session.id),
           onDelete: () => onDeleteSession(session.id),
           onPin: () => onTogglePin(session.id),
           onResume: () => onResumeSession(session.id)

  • apps/desktop/src/app/command-center/index.tsx

    @@ -113,7 +113,7 @@ interface SectionSearchEntry {
     }
     
     const NAVIGATION_SEARCH_ENTRIES: readonly NavigationSearchEntry[] = [
    -  { id: 'nav-new-chat', route: NEW_CHAT_ROUTE, title: 'New agent', detail: 'Start a fresh session' },
    +  { id: 'nav-new-chat', route: NEW_CHAT_ROUTE, title: 'New session', detail: 'Start a fresh session' },
       { id: 'nav-settings', route: SETTINGS_ROUTE, title: 'Settings', detail: 'Configure Hermes desktop' },
       { id: 'nav-skills', route: SKILLS_ROUTE, title: 'Skills', detail: 'Enable and inspect skills' },
       {

  • apps/desktop/src/app/desktop-controller.tsx

    @@ -6,6 +6,7 @@ import { Navigate, Route, Routes, useLocation, useNavigate, useParams } from 're
     import { BootFailureOverlay } from '@/components/boot-failure-overlay'
     import { DesktopInstallOverlay } from '@/components/desktop-install-overlay'
     import { DesktopOnboardingOverlay } from '@/components/desktop-onboarding-overlay'
    +import { GatewayConnectingOverlay } from '@/components/gateway-connecting-overlay'
     import { Pane, PaneMain } from '@/components/pane-shell'
     import { useSkinCommand } from '@/themes/use-skin-command'
     
    @@ -33,6 +34,8 @@ import {
       $selectedStoredSessionId,
       setAwaitingResponse,
       setBusy,
    +  setCurrentBranch,
    +  setCurrentCwd,
       setCurrentModel,
       setCurrentProvider,
       setMessages,
    @@ -122,6 +125,7 @@ export function DesktopController() {
         settingsOpen,
         toggleCommandCenter
       } = useOverlayRouting()
    +
       const terminalTakeoverActive = chatOpen && terminalTakeover
     
       const titlebarToolGroups = useGroupRegistry<TitlebarTool>()
    @@ -192,7 +196,10 @@ export function DesktopController() {
     
         try {
           const limit = $sessionsLimit.get()
    -      const result = await listSessions(limit)
    +      // Require at least one message so abandoned/empty "Untitled" drafts (one
    +      // was created per TUI/desktop launch before the lazy-create fix) don't
    +      // clutter the sidebar.
    +      const result = await listSessions(limit, 1)
     
           if (refreshSessionsRequestRef.current === requestId) {
             setSessions(result.sessions)
    @@ -324,6 +331,7 @@ export function DesktopController() {
       })
     
       const {
    +    archiveSession,
         branchCurrentSession,
         createBackendSessionForSend,
         openSettings,
    @@ -392,6 +400,29 @@ export function DesktopController() {
         [branchCurrentSession, refreshSessions]
       )
     
    +  const startSessionInWorkspace = useCallback(
    +    (path: null | string) => {
    +      startFreshSessionDraft()
    +
    +      const target = path?.trim()
    +
    +      if (!target) {
    +        return
    +      }
    +
    +      // The next message creates the backend session in $currentCwd, so seed
    +      // it (and the branch) from the workspace the user clicked the + on.
    +      setCurrentCwd(target)
    +      void requestGateway<{ branch?: string; cwd?: string }>('config.get', { key: 'project', cwd: target })
    +        .then(info => {
    +          setCurrentCwd(info.cwd || target)
    +          setCurrentBranch(info.branch || '')
    +        })
    +        .catch(() => undefined)
    +    },
    +    [requestGateway, startFreshSessionDraft]
    +  )
    +
       const handleSkinCommand = useSkinCommand()
     
       const { cancelRun, editMessage, handleThreadMessagesChange, reloadFromMessage, submitText, transcribeVoiceAudio } =
    @@ -461,9 +492,11 @@ export function DesktopController() {
       const sidebar = (
         <ChatSidebar
           currentView={currentView}
    +      onArchiveSession={sessionId => void archiveSession(sessionId)}
           onDeleteSession={sessionId => void removeSession(sessionId)}
           onLoadMoreSessions={loadMoreSessions}
           onNavigate={selectSidebarItem}
    +      onNewSessionInWorkspace={startSessionInWorkspace}
           onResumeSession={sessionId => navigate(sessionRoute(sessionId))}
         />
       )
    @@ -485,6 +518,7 @@ export function DesktopController() {
           />
           <ModelPickerOverlay gateway={gatewayRef.current || undefined} onSelect={selectModel} />
           <UpdatesOverlay />
    +      <GatewayConnectingOverlay />
           <BootFailureOverlay />
     
           {settingsOpen && (
    @@ -575,10 +609,10 @@ export function DesktopController() {
           titlebarTools={titlebarToolGroups.flat.right}
         >
           <Pane
    +        disabled={terminalTakeoverActive}
             id="chat-sidebar"
             maxWidth={SIDEBAR_MAX_WIDTH}
             minWidth={SIDEBAR_DEFAULT_WIDTH}
    -        disabled={terminalTakeoverActive}
             resizable
             side="left"
             width={`${SIDEBAR_DEFAULT_WIDTH}px`}

  • apps/desktop/src/app/session/hooks/use-session-actions.ts

    @@ -2,7 +2,7 @@ import type { MutableRefObject } from 'react'
     import { useCallback, useRef } from 'react'
     import type { NavigateFunction } from 'react-router-dom'
     
    -import { deleteSession, getSessionMessages } from '@/hermes'
    +import { deleteSession, getSessionMessages, setSessionArchived } from '@/hermes'
     import { type ChatMessage, chatMessageText, preserveLocalAssistantErrors, toChatMessages } from '@/lib/chat-messages'
     import { normalizePersonalityValue } from '@/lib/chat-runtime'
     import { embeddedImageUrls, textWithoutEmbeddedImages } from '@/lib/embedded-images'
    @@ -751,7 +751,39 @@ export function useSessionActions({
         ]
       )
     
    +  const archiveSession = useCallback(
    +    async (storedSessionId: string) => {
    +      clearNotifications()
    +
    +      const archived = $sessions.get().find(s => s.id === storedSessionId)
    +      const wasSelected = selectedStoredSessionId === storedSessionId
    +      const previousPinned = $pinnedSessionIds.get()
    +
    +      // Soft-hide: drop from the sidebar immediately, keep the data.
    +      setSessions(prev => prev.filter(s => s.id !== storedSessionId))
    +      $pinnedSessionIds.set(previousPinned.filter(id => id !== storedSessionId))
    +
    +      if (wasSelected) {
    +        startFreshSessionDraft(true)
    +      }
    +
    +      try {
    +        await setSessionArchived(storedSessionId, true)
    +        notify({ durationMs: 2_000, kind: 'success', message: 'Archived' })
    +      } catch (err) {
    +        if (archived) {
    +          setSessions(prev => [archived, ...prev.filter(s => s.id !== storedSessionId)])
    +        }
    +
    +        $pinnedSessionIds.set(previousPinned)
    +        notifyError(err, 'Archive failed')
    +      }
    +    },
    +    [selectedStoredSessionId, startFreshSessionDraft]
    +  )
    +
       return {
    +    archiveSession,
         branchCurrentSession,
         closeSettings,
         createBackendSessionForSend,

  • apps/desktop/src/app/settings/constants.ts

    @@ -311,11 +311,15 @@ export const MODE_OPTIONS: ModeOption[] = [
       { id: 'system', label: 'System', description: 'Follow OS appearance', icon: Monitor }
     ]
     
    -export const SEARCH_PLACEHOLDER: Record<'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'tools', string> = {
    +export const SEARCH_PLACEHOLDER: Record<
    +  'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'sessions' | 'tools',
    +  string
    +> = {
       about: 'About Hermes Desktop',
       config: 'Search settings...',
       gateway: 'Gateway connection...',
       keys: 'Search API keys...',
       mcp: 'Search MCP servers...',
    +  sessions: 'Search archived sessions...',
       tools: 'Search skills and tools...'
     }

  • apps/desktop/src/app/settings/index.tsx

    @@ -3,7 +3,7 @@ import { useEffect, useRef, useState } from 'react'
     
     import { getHermesConfigDefaults, getHermesConfigRecord, saveHermesConfig } from '@/hermes'
     import { triggerHaptic } from '@/lib/haptics'
    -import { Globe, Info, KeyRound, Package, Wrench } from '@/lib/icons'
    +import { Archive, Globe, Info, KeyRound, Package, Wrench } from '@/lib/icons'
     import { notifyError } from '@/store/notifications'
     
     import { useRouteEnumParam } from '../hooks/use-route-enum-param'
    @@ -19,6 +19,7 @@ import { SEARCH_PLACEHOLDER, SECTIONS } from './constants'
     import { GatewaySettings } from './gateway-settings'
     import { KeysSettings } from './keys-settings'
     import { McpSettings } from './mcp-settings'
    +import { SessionsSettings } from './sessions-settings'
     import { ToolsSettings } from './tools-settings'
     import type { SettingsPageProps, SettingsQueryKey, SettingsView as SettingsViewId } from './types'
     
    @@ -27,6 +28,7 @@ const SETTINGS_VIEWS: readonly SettingsViewId[] = [
       'gateway',
       'keys',
       'mcp',
    +  'sessions',
       'tools',
       'about'
     ]
    @@ -40,6 +42,7 @@ export function SettingsView({ gateway, onClose, onConfigSaved }: SettingsPagePr
         gateway: '',
         keys: '',
         mcp: '',
    +    sessions: '',
         tools: ''
       })
     
    @@ -149,6 +152,12 @@ export function SettingsView({ gateway, onClose, onConfigSaved }: SettingsPagePr
                 label="MCP"
                 onClick={() => setActiveView('mcp')}
               />
    +          <OverlayNavItem
    +            active={activeView === 'sessions'}
    +            icon={Archive}
    +            label="Archived Chats"
    +            onClick={() => setActiveView('sessions')}
    +          />
               <div className="my-2 h-px bg-border/30" />
               <OverlayNavItem
                 active={activeView === 'about'}
    @@ -200,6 +209,8 @@ export function SettingsView({ gateway, onClose, onConfigSaved }: SettingsPagePr
                 <KeysSettings query={queries.keys} />
               ) : activeView === 'mcp' ? (
                 <McpSettings gateway={gateway} onConfigSaved={onConfigSaved} query={queries.mcp} />
    +          ) : activeView === 'sessions' ? (
    +            <SessionsSettings query={queries.sessions} />
               ) : (
                 <ToolsSettings query={queries.tools} />
               )}

  • apps/desktop/src/app/settings/sessions-settings.tsx

    @@ -0,0 +1,168 @@
    +import { useCallback, useEffect, useMemo, useState } from 'react'
    +
    +import { Button } from '@/components/ui/button'
    +import { deleteSession, listSessions, setSessionArchived } from '@/hermes'
    +import { sessionTitle } from '@/lib/chat-runtime'
    +import { triggerHaptic } from '@/lib/haptics'
    +import { Archive, ArchiveOff, Loader2, Trash2 } from '@/lib/icons'
    +import { notify, notifyError } from '@/store/notifications'
    +import { setSessions } from '@/store/session'
    +import type { SessionInfo } from '@/types/hermes'
    +
    +import { EmptyState, ListRow, LoadingState, SectionHeading, SettingsContent } from './primitives'
    +import type { SearchProps } from './types'
    +
    +const ARCHIVED_FETCH_LIMIT = 200
    +
    +function workspaceLabel(cwd: null | string | undefined): string {
    +  const path = cwd?.trim()
    +
    +  if (!path) {
    +    return ''
    +  }
    +
    +  return (
    +    path
    +      .replace(/[/\\]+$/, '')
    +      .split(/[/\\]/)
    +      .filter(Boolean)
    +      .pop() ?? path
    +  )
    +}
    +
    +export function SessionsSettings({ query }: SearchProps) {
    +  const [sessions, setLocalSessions] = useState<SessionInfo[]>([])
    +  const [loading, setLoading] = useState(true)
    +  const [busyId, setBusyId] = useState<string | null>(null)
    +
    +  const load = useCallback(async () => {
    +    setLoading(true)
    +
    +    try {
    +      const result = await listSessions(ARCHIVED_FETCH_LIMIT, 0, 'only')
    +      setLocalSessions(result.sessions)
    +    } catch (err) {
    +      notifyError(err, 'Could not load archived sessions')
    +    } finally {
    +      setLoading(false)
    +    }
    +  }, [])
    +
    +  useEffect(() => {
    +    void load()
    +  }, [load])
    +
    +  const unarchive = useCallback(async (session: SessionInfo) => {
    +    setBusyId(session.id)
    +
    +    try {
    +      await setSessionArchived(session.id, false)
    +      setLocalSessions(prev => prev.filter(s => s.id !== session.id))
    +      // Surface it again in the sidebar without waiting for a full refresh.
    +      setSessions(prev => [{ ...session, archived: false }, ...prev.filter(s => s.id !== session.id)])
    +      triggerHaptic('selection')
    +      notify({ durationMs: 2_000, kind: 'success', message: 'Restored' })
    +    } catch (err) {
    +      notifyError(err, 'Unarchive failed')
    +    } finally {
    +      setBusyId(null)
    +    }
    +  }, [])
    +
    +  const remove = useCallback(async (session: SessionInfo) => {
    +    if (!window.confirm(`Permanently delete "${sessionTitle(session)}"? This cannot be undone.`)) {
    +      return
    +    }
    +
    +    setBusyId(session.id)
    +
    +    try {
    +      await deleteSession(session.id)
    +      setLocalSessions(prev => prev.filter(s => s.id !== session.id))
    +      triggerHaptic('warning')
    +    } catch (err) {
    +      notifyError(err, 'Delete failed')
    +    } finally {
    +      setBusyId(null)
    +    }
    +  }, [])
    +
    +  const filtered = useMemo(() => {
    +    const needle = query.trim().toLowerCase()
    +
    +    if (!needle) {
    +      return sessions
    +    }
    +
    +    return sessions.filter(session =>
    +      [sessionTitle(session), session.preview ?? '', session.cwd ?? ''].join(' ').toLowerCase().includes(needle)
    +    )
    +  }, [query, sessions])
    +
    +  if (loading) {
    +    return <LoadingState label="Loading archived sessions…" />
    +  }
    +
    +  return (
    +    <SettingsContent>
    +      <SectionHeading
    +        icon={Archive}
    +        meta={sessions.length ? String(sessions.length) : undefined}
    +        title="Archived sessions"
    +      />
    +      <p className="mb-2 text-[length:var(--conversation-caption-font-size)] text-(--ui-text-tertiary)">
    +        Archived chats are hidden from the sidebar but keep all their messages. Ctrl/⌘-click a chat in the sidebar to
    +        archive it.
    +      </p>
    +
    +      {filtered.length === 0 ? (
    +        <EmptyState
    +          description={query.trim() ? 'No archived chats match your search.' : 'Archive a chat to hide it here.'}
    +          title="Nothing archived"
    +        />
    +      ) : (
    +        <div className="divide-y divide-border/30">
    +          {filtered.map(session => {
    +            const label = workspaceLabel(session.cwd)
    +            const busy = busyId === session.id
    +
    +            return (
    +              <ListRow
    +                action={
    +                  <div className="flex items-center gap-1.5">
    +                    <Button
    +                      disabled={busy}
    +                      onClick={() => void unarchive(session)}
    +                      size="sm"
    +                      type="button"
    +                      variant="outline"
    +                    >
    +                      {busy ? <Loader2 className="size-3.5 animate-spin" /> : <ArchiveOff className="size-3.5" />}
    +                      <span>Unarchive</span>
    +                    </Button>
    +                    <Button
    +                      aria-label="Delete permanently"
    +                      className="text-muted-foreground hover:text-destructive"
    +                      disabled={busy}
    +                      onClick={() => void remove(session)}
    +                      size="icon"
    +                      title="Delete permanently"
    +                      type="button"
    +                      variant="ghost"
    +                    >
    +                      <Trash2 className="size-3.5" />
    +                    </Button>
    +                  </div>
    +                }
    +                description={session.preview || undefined}
    +                hint={label ? `${label} · ${session.message_count} messages` : `${session.message_count} messages`}
    +                key={session.id}
    +                title={sessionTitle(session)}
    +              />
    +            )
    +          })}
    +        </div>
    +      )}
    +    </SettingsContent>
    +  )
    +}

  • apps/desktop/src/app/settings/types.ts

    @@ -4,8 +4,8 @@ import type { HermesGateway } from '@/hermes'
     import type { IconComponent } from '@/lib/icons'
     import type { EnvVarInfo } from '@/types/hermes'
     
    -export type SettingsView = 'about' | 'gateway' | 'keys' | 'mcp' | 'tools' | `config:${string}`
    -export type SettingsQueryKey = 'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'tools'
    +export type SettingsView = 'about' | 'gateway' | 'keys' | 'mcp' | 'sessions' | 'tools' | `config:${string}`
    +export type SettingsQueryKey = 'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'sessions' | 'tools'
     export type EnvPatch = Partial<Pick<EnvVarInfo, 'is_set' | 'redacted_value'>>
     
     export interface SettingsPageProps {

  • apps/desktop/src/components/assistant-ui/markdown-text.tsx

    @@ -19,9 +19,9 @@ import {
       filePathFromMediaPath,
       mediaExternalUrl,
       mediaKind,
    -  mediaMime,
       mediaName,
    -  mediaPathFromMarkdownHref
    +  mediaPathFromMarkdownHref,
    +  mediaStreamUrl
     } from '@/lib/media'
     import { previewTargetFromMarkdownHref } from '@/lib/preview-targets'
     import { cn } from '@/lib/utils'
    @@ -40,24 +40,22 @@ import { cn } from '@/lib/utils'
     // LLM convention). The default false-setting only accepts `$$...$$`.
     const mathPlugin = createMemoizedMathPlugin({ singleDollarTextMath: true })
     
    -async function typedBlobUrl(dataUrl: string, mime: string): Promise<string> {
    -  const blob = await fetch(dataUrl).then(response => response.blob())
    -
    -  return URL.createObjectURL(new Blob([await blob.arrayBuffer()], { type: mime }))
    -}
    -
     async function mediaSrc(path: string): Promise<string> {
       if (/^(?:https?|data):/i.test(path)) {
         return path
       }
     
    +  // Stream audio/video through the custom protocol: data URLs are capped and
    +  // load the whole file into memory, which broke playback for larger videos.
    +  if (window.hermesDesktop && ['audio', 'video'].includes(mediaKind(path))) {
    +    return mediaStreamUrl(path)
    +  }
    +
       if (!window.hermesDesktop?.readFileDataUrl) {
         return mediaExternalUrl(path)
       }
     
    -  const dataUrl = await window.hermesDesktop.readFileDataUrl(filePathFromMediaPath(path))
    -
    -  return ['audio', 'video'].includes(mediaKind(path)) ? typedBlobUrl(dataUrl, mediaMime(path)) : dataUrl
    +  return window.hermesDesktop.readFileDataUrl(filePathFromMediaPath(path))
     }
     
     function OpenMediaButton({ kind, path }: { kind: 'audio' | 'video'; path: string }) {
    @@ -278,10 +276,7 @@ const MarkdownTextImpl = () => {
       // render, which churns Streamdown's outer memo + propagates new prop
       // identities into every Block. The plugin set really only varies on
       // `isStreaming`, so memoize on that.
    -  const plugins = useMemo(
    -    () => (isStreaming ? { math: mathPlugin } : { math: mathPlugin, code }),
    -    [isStreaming]
    -  )
    +  const plugins = useMemo(() => (isStreaming ? { math: mathPlugin } : { math: mathPlugin, code }), [isStreaming])
     
       const components = useMemo(
         () =>

  • apps/desktop/src/components/gateway-connecting-overlay.tsx

    @@ -0,0 +1,183 @@
    +import { useStore } from '@nanostores/react'
    +import { useEffect, useRef, useState } from 'react'
    +
    +import { cn } from '@/lib/utils'
    +import { $desktopBoot } from '@/store/boot'
    +import { $gatewayState } from '@/store/session'
    +
    +// Static, always-legible prefix; only TAIL ever scrambles. Splitting them at
    +// the render level means no timer logic (even a stale HMR one) can ever
    +// scramble "CONN".
    +const PREFIX = 'CONN'
    +const TAIL = 'ECTING'
    +// Even-weight mono ascii so cycling glyphs don't jump width (matches the
    +// nousnet-web download-button decode effect).
    +const SCRAMBLE_CHARS = '/\\|-_=+<>~:*'
    +const TICK_MS = 45
    +
    +// Exit choreography (ms): text fades down + out, hold, then the overlay fades.
    +const TEXT_OUT_MS = 360
    +const POST_TEXT_HOLD_MS = 300
    +const OVERLAY_OUT_MS = 520
    +// Preview-only: how long to "connect" for, and the pause before replaying.
    +const PREVIEW_CONNECT_MS = 2600
    +const PREVIEW_REPLAY_MS = 1100
    +
    +type Phase = 'live' | 'text-out' | 'overlay-out' | 'gone'
    +
    +// Dev affordance: a warm Cmd+R reconnects almost instantly, so the overlay
    +// only flashes. Load with `?connecting=1` to force a looping preview.
    +function forcedPreview(): boolean {
    +  if (!import.meta.env.DEV || typeof window === 'undefined') {
    +    return false
    +  }
    +
    +  try {
    +    return new URLSearchParams(window.location.search).get('connecting') === '1'
    +  } catch {
    +    return false
    +  }
    +}
    +
    +function scrambledTail(resolvedCount: number): string {
    +  return Array.from(TAIL, (ch, i) =>
    +    i < resolvedCount ? ch : SCRAMBLE_CHARS[(Math.random() * SCRAMBLE_CHARS.length) | 0]
    +  ).join('')
    +}
    +
    +export function GatewayConnectingOverlay() {
    +  const gatewayState = useStore($gatewayState)
    +  const boot = useStore($desktopBoot)
    +  const [previewing] = useState(forcedPreview)
    +  const [tail, setTail] = useState(TAIL)
    +  const [phase, setPhase] = useState<Phase>('live')
    +
    +  const connecting = gatewayState !== 'open' && !boot.error
    +  // Latches once we've actually shown the overlay, so the brief frame where
    +  // gatewayState flips to "open" (connecting -> false) before the exit phase
    +  // kicks in doesn't unmount us and cause a flash.
    +  const shownRef = useRef(false)
    +
    +  if (previewing || connecting) {
    +    shownRef.current = true
    +  }
    +
    +  // Decode loop — only while live (freeze the resolved word during the exit).
    +  useEffect(() => {
    +    if (phase !== 'live' || (!previewing && !connecting)) {
    +      return
    +    }
    +
    +    let resolved = 0
    +    let hold = 0
    +
    +    const id = window.setInterval(() => {
    +      if (resolved >= TAIL.length) {
    +        hold += 1
    +
    +        if (hold > 16) {
    +          resolved = 0
    +          hold = 0
    +        }
    +
    +        setTail(TAIL)
    +
    +        return
    +      }
    +
    +      resolved += 0.5
    +      setTail(scrambledTail(Math.floor(resolved)))
    +    }, TICK_MS)
    +
    +    return () => window.clearInterval(id)
    +  }, [phase, previewing, connecting])
    +
    +  // Kick off the exit when connected: real connect, or a faked timer in preview.
    +  useEffect(() => {
    +    if (phase !== 'live') {
    +      return
    +    }
    +
    +    if (previewing) {
    +      const id = window.setTimeout(() => {
    +        setTail(TAIL)
    +        setPhase('text-out')
    +      }, PREVIEW_CONNECT_MS)
    +
    +      return () => window.clearTimeout(id)
    +    }
    +
    +    if (gatewayState === 'open' && shownRef.current) {
    +      setTail(TAIL)
    +      setPhase('text-out')
    +    }
    +  }, [phase, previewing, gatewayState])
    +
    +  // Advance the exit choreography: text-out -> overlay-out -> gone.
    +  useEffect(() => {
    +    if (phase === 'text-out') {
    +      const id = window.setTimeout(() => setPhase('overlay-out'), TEXT_OUT_MS + POST_TEXT_HOLD_MS)
    +
    +      return () => window.clearTimeout(id)
    +    }
    +
    +    if (phase === 'overlay-out') {
    +      const id = window.setTimeout(() => setPhase('gone'), OVERLAY_OUT_MS)
    +
    +      return () => window.clearTimeout(id)
    +    }
    +
    +    // Preview replays so we can keep watching the transition.
    +    if (phase === 'gone' && previewing) {
    +      const id = window.setTimeout(() => {
    +        setTail(TAIL)
    +        setPhase('live')
    +      }, PREVIEW_REPLAY_MS)
    +
    +      return () => window.clearTimeout(id)
    +    }
    +  }, [phase, previewing])
    +
    +  // Boot failed — BootFailureOverlay owns the screen; don't linger behind it.
    +  if (boot.error && !previewing) {
    +    return null
    +  }
    +
    +  // Real connect: once the fade finishes, get out of the way for good.
    +  if (phase === 'gone' && !previewing) {
    +    return null
    +  }
    +
    +  // Never showed (e.g. gateway already up on a warm reload) — stay out.
    +  if (!previewing && !connecting && !shownRef.current) {
    +    return null
    +  }
    +
    +  const leaving = phase !== 'live'
    +  const overlayHidden = phase === 'overlay-out' || phase === 'gone'
    +
    +  return (
    +    <div
    +      className={cn(
    +        'fixed inset-0 z-[1200] grid place-items-center bg-(--ui-chat-surface-background) transition-opacity duration-500 ease-out',
    +        overlayHidden ? 'pointer-events-none opacity-0' : 'opacity-100'
    +      )}
    +    >
    +      <style>{'@keyframes gco-cursor { 0%, 49% { opacity: 1 } 50%, 100% { opacity: 0 } }'}</style>
    +      <span
    +        className={cn(
    +          'inline-flex items-center pl-[0.4em] font-mono text-[0.64rem] font-semibold uppercase tracking-[0.4em] tabular-nums text-(--theme-primary) transition duration-300 ease-out',
    +          leaving ? 'translate-y-2 opacity-0 saturate-0' : 'translate-y-0 opacity-100 saturate-100'
    +        )}
    +      >
    +        {PREFIX}
    +        {tail}
    +        <span
    +          aria-hidden="true"
    +          className="dither ml-0.5 inline-block size-2 shrink-0 -translate-y-px rounded-[1px]"
    +          style={{ animation: 'gco-cursor 1s step-end infinite' }}
    +        />
    +      </span>
    +    </div>
    +  )
    +}

  • apps/desktop/src/hermes.ts

    @@ -111,9 +111,13 @@ export class HermesGateway extends JsonRpcGatewayClient {
       }
     }
     
    -export async function listSessions(limit = 40, minMessages = 0): Promise<PaginatedSessions> {
    +export async function listSessions(
    +  limit = 40,
    +  minMessages = 0,
    +  archived: 'exclude' | 'include' | 'only' = 'exclude'
    +): Promise<PaginatedSessions> {
       const result = await window.hermesDesktop.api<PaginatedSessions>({
    -    path: `/api/sessions?limit=${limit}&offset=0&min_messages=${Math.max(0, minMessages)}`
    +    path: `/api/sessions?limit=${limit}&offset=0&min_messages=${Math.max(0, minMessages)}&archived=${archived}`
       })
     
       return {
    @@ -123,6 +127,14 @@ export async function listSessions(limit = 40, minMessages = 0): Promise<Paginat
       }
     }
     
    +export function setSessionArchived(id: string, archived: boolean): Promise<{ ok: boolean }> {
    +  return window.hermesDesktop.api<{ ok: boolean }>({
    +    path: `/api/sessions/${encodeURIComponent(id)}`,
    +    method: 'PATCH',
    +    body: { archived }
    +  })
    +}
    +
     export function searchSessions(query: string): Promise<SessionSearchResponse> {
       return window.hermesDesktop.api<SessionSearchResponse>({
         path: `/api/sessions/search?q=${encodeURIComponent(query)}`

  • apps/desktop/src/lib/icons.ts

    @@ -2,6 +2,8 @@ import {
       IconActivity as Activity,
       IconAlertCircle as AlertCircle,
       IconAlertTriangle as AlertTriangle,
    +  IconArchive as Archive,
    +  IconArchiveOff as ArchiveOff,
       IconArrowUp as ArrowUp,
       IconArrowUpRight as ArrowUpRight,
       IconAt as AtSign,
    @@ -98,6 +100,8 @@ export {
       Activity,
       AlertCircle,
       AlertTriangle,
    +  Archive,
    +  ArchiveOff,
       ArrowUp,
       ArrowUpRight,
       AtSign,

  • apps/desktop/src/lib/media.ts

    @@ -58,6 +58,13 @@ export function mediaExternalUrl(path: string): string {
       return /^(?:https?|file):/i.test(path) ? path : `file://${path}`
     }
     
    +// Custom Electron scheme (registered in electron/main.cjs) that streams a local
    +// file with Range support. Used for audio/video so playback bypasses the data
    +// URL size cap and supports seeking. `path` may be a plain path or `file://…`.
    +export function mediaStreamUrl(path: string): string {
    +  return `hermes-media://stream/${encodeURIComponent(filePathFromMediaPath(path))}`
    +}
    +
     export function mediaPathFromMarkdownHref(href?: string): string | null {
       if (!href?.startsWith('#media:')) {
         return null

  • apps/desktop/src/types/hermes.ts

    @@ -240,6 +240,7 @@ export interface SessionCreateResponse {
     }
     
     export interface SessionInfo {
    +  archived?: boolean
       cwd?: null | string
       ended_at: null | number
       id: string

  • hermes_cli/web_server.py

    @@ -1131,22 +1131,51 @@ async def get_action_status(name: str, lines: int = 200):
     
     
     @app.get("/api/sessions")
    -async def get_sessions(limit: int = 20, offset: int = 0, min_messages: int = 0):
    +async def get_sessions(
    +    limit: int = 20,
    +    offset: int = 0,
    +    min_messages: int = 0,
    +    archived: str = "exclude",
    +):
    +    """List sessions.
    +
    +    ``archived`` controls how soft-archived sessions are treated:
    +    ``exclude`` (default) hides them, ``only`` returns just the archived ones
    +    (used by the desktop "Archived sessions" settings panel), and ``include``
    +    returns both.
    +    """
    +    if archived not in ("exclude", "only", "include"):
    +        raise HTTPException(
    +            status_code=400,
    +            detail="archived must be one of: exclude, only, include",
    +        )
         try:
             from hermes_state import SessionDB
             db = SessionDB()
             try:
                 min_message_count = max(0, min_messages)
    +            archived_only = archived == "only"
    +            include_archived = archived == "include"
                 sessions = db.list_sessions_rich(
    -                limit=limit, offset=offset, min_message_count=min_message_count
    +                limit=limit,
    +                offset=offset,
    +                min_message_count=min_message_count,
    +                include_archived=include_archived,
    +                archived_only=archived_only,
    +            )
    +            total = db.session_count(
    +                min_message_count=min_message_count,
    +                include_archived=include_archived,
    +                archived_only=archived_only,
                 )
    -            total = db.session_count(min_message_count=min_message_count)
                 now = time.time()
                 for s in sessions:
                     s["is_active"] = (
                         s.get("ended_at") is None
                         and (now - s.get("last_active", s.get("started_at", 0))) < 300
                     )
    +                # SQLite stores the flag as 0/1; expose a real JSON boolean.
    +                s["archived"] = bool(s.get("archived"))
                 return {"sessions": sessions, "total": total, "limit": limit, "offset": offset}
             finally:
                 db.close()
    @@ -3707,25 +3736,39 @@ async def delete_session_endpoint(session_id: str):
     
     class SessionRename(BaseModel):
         title: Optional[str] = None
    +    archived: Optional[bool] = None
     
     
     @app.patch("/api/sessions/{session_id}")
     async def rename_session_endpoint(session_id: str, body: SessionRename):
    -    """Rename a session (or clear its title when ``title`` is empty/null)."""
    +    """Update a session: rename (or clear its title) and/or archive it.
    +
    +    ``title`` renames (empty/null clears the title); ``archived`` soft-hides or
    +    restores the session. Either field may be omitted.
    +    """
         from hermes_state import SessionDB
         db = SessionDB()
         try:
             sid = db.resolve_session_id(session_id)
             if not sid:
                 raise HTTPException(status_code=404, detail="Session not found")
    -        try:
    -            updated = db.set_session_title(sid, body.title or "")
    -        except ValueError as e:
    -            # Title too long, invalid characters, or already in use.
    -            raise HTTPException(status_code=400, detail=str(e))
    -        if not updated:
    -            raise HTTPException(status_code=404, detail="Session not found")
    -        return {"ok": True, "title": db.get_session_title(sid) or ""}
    +        if body.title is None and body.archived is None:
    +            raise HTTPException(
    +                status_code=400,
    +                detail="Nothing to update; provide 'title' and/or 'archived'.",
    +            )
    +        if body.title is not None:
    +            try:
    +                db.set_session_title(sid, body.title or "")
    +            except ValueError as e:
    +                # Title too long, invalid characters, or already in use.
    +                raise HTTPException(status_code=400, detail=str(e))
    +        if body.archived is not None:
    +            db.set_session_archived(sid, body.archived)
    +        result = {"ok": True, "title": db.get_session_title(sid) or ""}
    +        if body.archived is not None:
    +            result["archived"] = bool(body.archived)
    +        return result
         finally:
             db.close()
     

  • hermes_state.py

    @@ -264,6 +264,7 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None:
         handoff_platform TEXT,
         handoff_error TEXT,
         rewind_count INTEGER NOT NULL DEFAULT 0,
    +    archived INTEGER NOT NULL DEFAULT 0,
         FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
     );
     
    @@ -1430,6 +1431,22 @@ def get_session_title(self, session_id: str) -> Optional[str]:
                 row = cursor.fetchone()
             return row["title"] if row else None
     
    +    def set_session_archived(self, session_id: str, archived: bool) -> bool:
    +        """Archive or unarchive a session.
    +
    +        Archived sessions are hidden from the default session list but keep all
    +        their messages — this is a soft hide, not a delete. Returns True when a
    +        row was updated.
    +        """
    +        def _do(conn):
    +            cursor = conn.execute(
    +                "UPDATE sessions SET archived = ? WHERE id = ?",
    +                (1 if archived else 0, session_id),
    +            )
    +            return cursor.rowcount
    +        rowcount = self._execute_write(_do)
    +        return rowcount > 0
    +
         def get_session_by_title(self, title: str) -> Optional[Dict[str, Any]]:
             """Look up a session by exact title. Returns session dict or None."""
             with self._lock:
    @@ -1549,6 +1566,8 @@ def list_sessions_rich(
             min_message_count: int = 0,
             project_compression_tips: bool = True,
             order_by_last_active: bool = False,
    +        include_archived: bool = False,
    +        archived_only: bool = False,
         ) -> List[Dict[str, Any]]:
             """List sessions with preview (first user message) and last active timestamp.
     
    @@ -1604,6 +1623,10 @@ def list_sessions_rich(
             if min_message_count > 0:
                 where_clauses.append("s.message_count >= ?")
                 params.append(min_message_count)
    +        if archived_only:
    +            where_clauses.append("s.archived = 1")
    +        elif not include_archived:
    +            where_clauses.append("s.archived = 0")
     
             where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
             if order_by_last_active:
    @@ -3027,7 +3050,13 @@ def search_sessions(
         # Utility
         # =========================================================================
     
    -    def session_count(self, source: str = None, min_message_count: int = 0) -> int:
    +    def session_count(
    +        self,
    +        source: str = None,
    +        min_message_count: int = 0,
    +        include_archived: bool = False,
    +        archived_only: bool = False,
    +    ) -> int:
             """Count sessions, optionally filtered by source."""
             where_clauses = []
             params = []
    @@ -3038,6 +3067,10 @@ def session_count(self, source: str = None, min_message_count: int = 0) -> int:
             if min_message_count > 0:
                 where_clauses.append("message_count >= ?")
                 params.append(min_message_count)
    +        if archived_only:
    +            where_clauses.append("archived = 1")
    +        elif not include_archived:
    +            where_clauses.append("archived = 0")
     
             where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
     

  • tests/hermes_cli/test_web_server.py

    @@ -187,11 +187,11 @@ class _FakeDB:
                 def __init__(self, *args, **kwargs):
                     pass
     
    -            def list_sessions_rich(self, limit, offset, min_message_count=0):
    +            def list_sessions_rich(self, limit, offset, min_message_count=0, **kwargs):
                     captured["list"] = min_message_count
                     return []
     
    -            def session_count(self, min_message_count=0):
    +            def session_count(self, min_message_count=0, **kwargs):
                     captured["count"] = min_message_count
                     return 0
     
    @@ -250,6 +250,76 @@ def test_rename_session_not_found(self):
             resp = self.client.patch("/api/sessions/does-not-exist", json={"title": "x"})
             assert resp.status_code == 404
     
    +    def test_archive_session_via_patch(self):
    +        """PATCH archived=true soft-hides a session; archived=false restores it."""
    +        from hermes_state import SessionDB
    +
    +        db = SessionDB()
    +        try:
    +            db.create_session(session_id="arch-me", source="cli")
    +            db.append_message(session_id="arch-me", role="user", content="hi")
    +        finally:
    +            db.close()
    +
    +        resp = self.client.patch("/api/sessions/arch-me", json={"archived": True})
    +        assert resp.status_code == 200
    +        assert resp.json()["archived"] is True
    +
    +        # Hidden from the default list, surfaced by archived=only.
    +        listed = self.client.get("/api/sessions").json()
    +        assert all(s["id"] != "arch-me" for s in listed["sessions"])
    +        only = self.client.get("/api/sessions?archived=only").json()
    +        assert any(s["id"] == "arch-me" for s in only["sessions"])
    +
    +        resp = self.client.patch("/api/sessions/arch-me", json={"archived": False})
    +        assert resp.status_code == 200
    +        restored = self.client.get("/api/sessions").json()
    +        assert any(s["id"] == "arch-me" for s in restored["sessions"])
    +
    +    def test_patch_session_without_fields_is_400(self):
    +        """An existing session + empty body is a bad request, not a 404."""
    +        from hermes_state import SessionDB
    +
    +        db = SessionDB()
    +        try:
    +            db.create_session(session_id="no-fields", source="cli")
    +        finally:
    +            db.close()
    +
    +        resp = self.client.patch("/api/sessions/no-fields", json={})
    +        assert resp.status_code == 400
    +
    +    def test_get_sessions_rejects_unknown_archived_value(self):
    +        resp = self.client.get("/api/sessions?archived=bogus")
    +        assert resp.status_code == 400
    +
    +    def test_get_sessions_archived_is_boolean(self):
    +        from hermes_state import SessionDB
    +
    +        db = SessionDB()
    +        try:
    +            db.create_session(session_id="bool-arch", source="cli")
    +            db.append_message(session_id="bool-arch", role="user", content="hi")
    +        finally:
    +            db.close()
    +
    +        row = next(s for s in self.client.get("/api/sessions").json()["sessions"] if s["id"] == "bool-arch")
    +        assert row["archived"] is False
    +
    +    def test_rename_response_omits_archived_when_not_set(self):
    +        """Title-only PATCH keeps its legacy {ok, title} response shape."""
    +        from hermes_state import SessionDB
    +
    +        db = SessionDB()
    +        try:
    +            db.create_session(session_id="title-only", source="cli")
    +        finally:
    +            db.close()
    +
    +        resp = self.client.patch("/api/sessions/title-only", json={"title": "Hi"})
    +        assert resp.status_code == 200
    +        assert "archived" not in resp.json()
    +
         def test_audio_transcription_endpoint(self, monkeypatch):
             import tools.transcription_tools as transcription_tools
     

  • tests/test_hermes_state.py

    @@ -3509,3 +3509,43 @@ def execute(self, sql, params=()):
             assert any("journal_mode=WAL" in sql for sql in conn.executed), (
                 "set-pragma must fire when probe returns 'delete'"
             )
    +
    +
    +class TestSessionArchive:
    +    """Soft-archiving hides a session from default listings without deleting it."""
    +
    +    def _seed(self, db, sid, *, archived=False):
    +        db.create_session(session_id=sid, source="cli")
    +        db.append_message(session_id=sid, role="user", content=f"hello from {sid}")
    +        if archived:
    +            db.set_session_archived(sid, True)
    +
    +    def test_set_session_archived_roundtrip(self, db):
    +        self._seed(db, "s1")
    +        assert db.set_session_archived("s1", True) is True
    +        assert db.get_session("s1")["archived"] == 1
    +        assert db.set_session_archived("s1", False) is True
    +        assert db.get_session("s1")["archived"] == 0
    +
    +    def test_set_session_archived_missing_row(self, db):
    +        assert db.set_session_archived("nope", True) is False
    +
    +    def test_archived_excluded_by_default(self, db):
    +        self._seed(db, "live")
    +        self._seed(db, "hidden", archived=True)
    +
    +        ids = [s["id"] for s in db.list_sessions_rich()]
    +        assert ids == ["live"]
    +        assert db.session_count() == 1
    +
    +    def test_archived_only_and_include(self, db):
    +        self._seed(db, "live")
    +        self._seed(db, "hidden", archived=True)
    +
    +        only = [s["id"] for s in db.list_sessions_rich(archived_only=True)]
    +        assert only == ["hidden"]
    +        assert db.session_count(archived_only=True) == 1
    +
    +        both = {s["id"] for s in db.list_sessions_rich(include_archived=True)}
    +        assert both == {"live", "hidden"}
    +        assert db.session_count(include_archived=True) == 2

  • tests/test_tui_gateway_server.py

    @@ -884,6 +884,73 @@ def set_session_title(self, _key, _title):
             server._sessions.pop("sid", None)
     
     
    +def test_notification_event_routing_by_session_key(monkeypatch):
    +    """Background-process events surface only in the session that owns them."""
    +    mine = _session(session_key="mine")
    +    other = _session(session_key="other")
    +    monkeypatch.setattr(server, "_sessions", {"a": mine, "b": other})
    +
    +    # My own event → handle it.
    +    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "mine"}) is False
    +    # Global/system event with no owner → handle it.
    +    assert server._notification_event_belongs_elsewhere(mine, {"session_key": ""}) is False
    +    assert server._notification_event_belongs_elsewhere(mine, {}) is False
    +    # Owned by another *live* session → defer to that session's poller.
    +    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "other"}) is True
    +    # Owner is gone (not in _sessions) → handle as fallback so it isn't lost.
    +    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "ghost"}) is False
    +
    +
    +def test_session_create_does_not_persist_empty_row(monkeypatch):
    +    """session.create must NOT eagerly write a DB row.
    +
    +    Every TUI/desktop launch opens a session here just to paint the composer;
    +    eagerly creating a row left an empty "Untitled" session behind for every
    +    launch the user never typed into. The row is created lazily on first prompt.
    +    """
    +    created = []
    +
    +    class _FakeDB:
    +        def create_session(self, *args, **kwargs):
    +            created.append((args, kwargs))
    +
    +    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
    +    monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
    +    monkeypatch.setattr(
    +        server.threading,
    +        "Timer",
    +        lambda *a, **k: types.SimpleNamespace(daemon=False, start=lambda: None),
    +    )
    +
    +    resp = server.handle_request(
    +        {"id": "1", "method": "session.create", "params": {"cols": 80}}
    +    )
    +    sid = resp["result"]["session_id"]
    +    try:
    +        assert resp["result"]["stored_session_id"]
    +        assert created == [], "session.create should not persist an empty DB row"
    +    finally:
    +        server._sessions.pop(sid, None)
    +
    +
    +def test_ensure_session_db_row_persists_with_cwd(monkeypatch, tmp_path):
    +    """First prompt persists the row (INSERT OR IGNORE) capturing cwd up front."""
    +    created = []
    +
    +    class _FakeDB:
    +        def create_session(self, key, source=None, model=None, cwd=None):
    +            created.append({"key": key, "source": source, "model": model, "cwd": cwd})
    +
    +    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
    +    monkeypatch.setattr(server, "_resolve_model", lambda: "test-model")
    +
    +    server._ensure_session_db_row({"session_key": "k1", "cwd": str(tmp_path)})
    +
    +    assert created == [
    +        {"key": "k1", "source": "tui", "model": "test-model", "cwd": str(tmp_path)}
    +    ]
    +
    +
     def test_session_title_clears_pending_after_persist(monkeypatch):
         class _FakeDB:
             def __init__(self):

  • tools/process_registry.py

    @@ -880,6 +880,7 @@ def _move_to_finished(self, session: ProcessSession):
                 self.completion_queue.put({
                     "type": "completion",
                     "session_id": session.id,
    +                "session_key": session.session_key,
                     "command": session.command,
                     "exit_code": session.exit_code,
                     "output": output_tail,

  • tui_gateway/server.py

    @@ -702,6 +702,32 @@ def _register_session_cwd(session: dict | None) -> None:
             pass
     
     
    +def _ensure_session_db_row(session: dict) -> None:
    +    """Idempotently persist the session's DB row on first real activity.
    +
    +    Called from prompt.submit so a row only exists once the user actually sends
    +    a message — abandoned drafts never leave an empty "Untitled" session behind.
    +    Uses INSERT OR IGNORE under the hood, so re-calls (and the AIAgent's own
    +    lazy create) are no-ops. Captures cwd up front so workspace grouping works
    +    without waiting for a separate cwd update.
    +    """
    +    key = session.get("session_key")
    +    if not key:
    +        return
    +    db = _get_db()
    +    if db is None:
    +        return
    +    try:
    +        db.create_session(
    +            key,
    +            source="tui",
    +            model=_resolve_model(),
    +            cwd=_session_cwd(session),
    +        )
    +    except Exception:
    +        logger.debug("failed to persist desktop session row", exc_info=True)
    +
    +
     def _set_session_cwd(session: dict, cwd: str) -> str:
         resolved = os.path.abspath(os.path.expanduser(str(cwd)))
         if not os.path.isdir(resolved):
    @@ -2750,17 +2776,12 @@ def _(rid, params: dict) -> dict:
             "transport": current_transport() or _stdio_transport,
         }
         _register_session_cwd(_sessions[sid])
    -    db = _get_db()
    -    if db is not None:
    -        try:
    -            db.create_session(
    -                key,
    -                source="tui",
    -                model=_resolve_model(),
    -                cwd=_sessions[sid]["cwd"],
    -            )
    -        except Exception:
    -            logger.debug("failed to pre-create desktop session row", exc_info=True)
    +    # NOTE: we intentionally do NOT persist a DB row here. Every TUI/desktop
    +    # launch (and every "New agent" / draft) opens a session here just to paint
    +    # the composer, so eagerly creating a row left an "Untitled" empty session
    +    # behind for every launch the user never typed into. The row is now created
    +    # lazily on the first prompt (see _ensure_session_db_row + prompt.submit),
    +    # and the AIAgent's own INSERT-OR-IGNORE persists it on the first turn too.
     
         # Return the lightweight session immediately so Ink can paint the composer
         # + skeleton panel, then build the real AIAgent just after this response is
    @@ -3841,6 +3862,8 @@ def _(rid, params: dict) -> dict:
             session["last_active"] = time.time()
             _start_inflight_turn(session, text)
     
    +    # Persist the DB row lazily, now that the user has actually sent a message.
    +    _ensure_session_db_row(session)
         _start_agent_build(sid, session)
     
         def run_after_agent_ready() -> None:
    @@ -3865,6 +3888,35 @@ def run_after_agent_ready() -> None:
         return _ok(rid, {"status": "streaming"})
     
     
    +def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool:
    +    """True if ``evt`` is owned by a *different* live session.
    +
    +    Background-process events carry the ``session_key`` of the session that
    +    started the process. Since all desktop sessions share one process-wide
    +    completion queue, each poller must skip events it doesn't own so a
    +    background job's completion surfaces in the session that launched it — not
    +    whichever poller happened to dequeue first. Orphaned events (owner gone)
    +    and global/system events (empty ``session_key``) return False so the
    +    current poller still handles them rather than losing them.
    +    """
    +    evt_key = str(evt.get("session_key") or "")
    +    if not evt_key:
    +        return False
    +    if evt_key == str(session.get("session_key") or ""):
    +        return False
    +    try:
    +        snapshot = list(_sessions.values())
    +    except Exception:
    +        # If we can't safely enumerate live sessions, fail open so we don't
    +        # crash the poller thread or drop the event.
    +        return False
    +
    +    return any(
    +        s is not session and str(s.get("session_key") or "") == evt_key
    +        for s in snapshot
    +    )
    +
    +
     def _notification_poller_loop(
         stop_event: threading.Event, sid: str, session: dict
     ) -> None:
    @@ -3887,6 +3939,16 @@ def _notification_poller_loop(
             except Exception:
                 continue
     
    +        # Multiple desktop sessions share this one process-wide queue. Only
    +        # consume events that belong to *this* session — otherwise a background
    +        # process started in session A would surface its completion in whichever
    +        # session's poller happened to wake first (Ben's "reported in a
    +        # different session" bug). Leave foreign events for their owner.
    +        if _notification_event_belongs_elsewhere(session, evt):
    +            process_registry.completion_queue.put(evt)
    +            time.sleep(0.1)
    +            continue
    +
             _evt_sid = evt.get("session_id", "")
             if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
                 continue
    @@ -3917,12 +3979,17 @@ def _notification_poller_loop(
                     session["running"] = False
     
         # Drain any remaining events after stop signal (process all pending
    -    # before exiting so nothing is lost on shutdown).
    +    # before exiting so nothing is lost on shutdown). Events owned by other
    +    # live sessions are set aside and re-queued so their poller still sees them.
    +    deferred: list = []
         while not process_registry.completion_queue.empty():
             try:
                 evt = process_registry.completion_queue.get_nowait()
             except Exception:
                 break
    +        if _notification_event_belongs_elsewhere(session, evt):
    +            deferred.append(evt)
    +            continue
             _evt_sid = evt.get("session_id", "")
             if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
                 continue
    @@ -3951,6 +4018,10 @@ def _notification_poller_loop(
                 with session["history_lock"]:
                     session["running"] = False
     
    +    # Hand any other sessions' events back to the shared queue.
    +    for evt in deferred:
    +        process_registry.completion_queue.put(evt)
    +
     
     def _start_notification_poller(sid: str, session: dict) -> threading.Event:
         """Start the background notification poller for a TUI session."""

• #60863 fix(delegation): route async results to their origin session across compression (#55578)

  merged to main

  files: gateway/session_context.py, tests/test_tui_gateway_server.py, tests/tools/test_async_delegation.py, tests/tui_gateway/test_delegation_session_lifecycle.py, tools/async_delegation.py, tools/delegate_tool.py, tui_gateway/server.py

  Show the diff

  • gateway/session_context.py

    @@ -79,6 +79,13 @@ def session_context_engaged() -> bool:
     _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET)
     _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET)
     _SESSION_ID: ContextVar = ContextVar("HERMES_SESSION_ID", default=_UNSET)
    +# In-process UI session/window id for multi-session desktop/TUI hosts. This is
    +# intentionally separate from HERMES_SESSION_ID: the latter is the durable
    +# conversation/session-db id, while the UI id is the live frontend tab/window
    +# that commissioned a detached completion. Background completions use it as a
    +# precise return address so a stale/rotated durable session key cannot be
    +# consumed by whichever desktop poller wakes first.
    +_SESSION_UI_SESSION_ID: ContextVar = ContextVar("HERMES_UI_SESSION_ID", default=_UNSET)
     # ID of the message that triggered the current turn. Used as a reply anchor
     # so background-process notifications stay inside the originating Telegram
     # private-chat topic (those lanes route only with thread id + reply anchor).
    @@ -123,6 +130,7 @@ def session_context_engaged() -> bool:
         "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME,
         "HERMES_SESSION_KEY": _SESSION_KEY,
         "HERMES_SESSION_ID": _SESSION_ID,
    +    "HERMES_UI_SESSION_ID": _SESSION_UI_SESSION_ID,
         "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID,
         "HERMES_SESSION_PROFILE": _SESSION_PROFILE,
         "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM,
    @@ -160,6 +168,7 @@ def set_session_vars(
         profile: str = "",
         cwd: str = "",
         async_delivery: bool = True,
    +    ui_session_id: str = "",
     ) -> list:
         """Set all session context variables and return reset tokens.
     
    @@ -191,6 +200,7 @@ def set_session_vars(
             _SESSION_USER_NAME.set(user_name),
             _SESSION_KEY.set(session_key),
             _SESSION_ID.set(session_id),
    +        _SESSION_UI_SESSION_ID.set(ui_session_id),
             _SESSION_MESSAGE_ID.set(message_id),
             _SESSION_PROFILE.set(profile),
             _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)),
    @@ -225,6 +235,7 @@ def clear_session_vars(tokens: list) -> None:
             _SESSION_USER_NAME,
             _SESSION_KEY,
             _SESSION_ID,
    +        _SESSION_UI_SESSION_ID,
             _SESSION_MESSAGE_ID,
             _SESSION_PROFILE,
         ):

  • tests/test_tui_gateway_server.py

    @@ -2112,14 +2112,85 @@ def test_notification_event_routing_by_session_key(monkeypatch):
         monkeypatch.setattr(server, "_sessions", {"a": mine, "b": other})
     
         # My own event → handle it.
    -    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "mine"}) is False
    +    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "mine"}) is False
         # Global/system event with no owner → handle it.
    -    assert server._notification_event_belongs_elsewhere(mine, {"session_key": ""}) is False
    -    assert server._notification_event_belongs_elsewhere(mine, {}) is False
    +    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": ""}) is False
    +    assert server._notification_event_belongs_elsewhere("a", mine, {}) is False
         # Owned by another *live* session → defer to that session's poller.
    -    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "other"}) is True
    +    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "other"}) is True
         # Owner is gone (not in _sessions) → handle as fallback so it isn't lost.
    -    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "ghost"}) is False
    +    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "ghost"}) is False
    +
    +
    +def test_async_delegation_event_prefers_origin_ui_session(monkeypatch):
    +    """Detached subagent completions return to the commissioning TUI tab.
    +
    +    Regression: when the durable session key was stale/orphaned, whichever
    +    desktop poller woke first could consume the async result and inject it into
    +    an unrelated session.
    +    """
    +    mine = _session(session_key="current-key")
    +    other = _session(session_key="unrelated-key")
    +    monkeypatch.setattr(server, "_sessions", {"origin-sid": mine, "other-sid": other})
    +    monkeypatch.setattr(server, "_get_db", lambda: None)
    +    evt = {
    +        "type": "async_delegation",
    +        "session_key": "stale-or-rotated-key",
    +        "origin_ui_session_id": "origin-sid",
    +    }
    +
    +    assert server._notification_event_belongs_elsewhere("other-sid", other, evt) is True
    +    assert server._notification_event_belongs_elsewhere("origin-sid", mine, evt) is False
    +
    +
    +def test_notification_event_follows_compression_continuation(monkeypatch):
    +    """Events keyed to a compressed parent route to the live continuation."""
    +    old_parent = _session(session_key="old-parent")
    +    live_tip = _session(session_key="new-tip")
    +    monkeypatch.setattr(server, "_sessions", {"old-sid": old_parent, "tip-sid": live_tip})
    +
    +    class _DB:
    +        def resolve_resume_session_id(self, session_id):
    +            return "new-tip" if session_id == "old-parent" else session_id
    +
    +    monkeypatch.setattr(server, "_get_db", lambda: _DB())
    +    evt = {"type": "async_delegation", "session_key": "old-parent"}
    +
    +    assert server._notification_event_belongs_elsewhere("old-sid", old_parent, evt) is True
    +    assert server._notification_event_belongs_elsewhere("tip-sid", live_tip, evt) is False
    +    # A third session must leave it alone for the continuation's poller.
    +    third = _session(session_key="third")
    +    monkeypatch.setattr(
    +        server,
    +        "_sessions",
    +        {"old-sid": old_parent, "tip-sid": live_tip, "third-sid": third},
    +    )
    +    assert server._notification_event_belongs_elsewhere("third-sid", third, evt) is True
    +
    +
    +def test_finalized_origin_ui_session_falls_back_to_live_continuation(monkeypatch):
    +    """A closed origin tab must not steal its resumed continuation's result."""
    +    finalized_origin = _session(session_key="old-parent", _finalized=True)
    +    live_tip = _session(session_key="new-tip")
    +    monkeypatch.setattr(
    +        server,
    +        "_sessions",
    +        {"origin-sid": finalized_origin, "tip-sid": live_tip},
    +    )
    +
    +    class _DB:
    +        def resolve_resume_session_id(self, session_id):
    +            return "new-tip" if session_id == "old-parent" else session_id
    +
    +    monkeypatch.setattr(server, "_get_db", lambda: _DB())
    +    evt = {
    +        "type": "async_delegation",
    +        "session_key": "old-parent",
    +        "origin_ui_session_id": "origin-sid",
    +    }
    +
    +    assert server._notification_event_belongs_elsewhere("origin-sid", finalized_origin, evt) is True
    +    assert server._notification_event_belongs_elsewhere("tip-sid", live_tip, evt) is False
     
     
     def test_prompt_submit_rejects_negative_truncate_ordinal(monkeypatch):

  • tests/tools/test_async_delegation.py

    @@ -289,6 +289,68 @@ def slow_child(task_index, goal, child=None, parent_agent=None, **kw):
         assert "the real task" in text
     
     
    +def test_delegate_task_background_uses_live_tui_agent_session_id(monkeypatch):
    +    """TUI async delegation must route to the live/compressed agent id.
    +
    +    Regression: delegate_task captured the stale approval/session context key
    +    after compression rotated parent_agent.session_id. The resulting completion
    +    was orphaned and could be consumed by an unrelated desktop session poller.
    +    """
    +    import json
    +    from unittest.mock import MagicMock
    +    import tools.delegate_tool as dt
    +    from gateway.session_context import clear_session_vars, set_session_vars
    +    from tools.approval import reset_current_session_key, set_current_session_key
    +
    +    parent = MagicMock()
    +    parent._delegate_depth = 0
    +    parent.session_id = "post-compress-tip"
    +    parent._interrupt_requested = False
    +    parent._active_children = []
    +    parent._active_children_lock = None
    +    fake_child = MagicMock()
    +    fake_child._delegate_role = "leaf"
    +
    +    creds = {
    +        "model": "m", "provider": None, "base_url": None, "api_key": None,
    +        "api_mode": None, "command": None, "args": None,
    +    }
    +    monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
    +    monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
    +    monkeypatch.setattr(
    +        dt,
    +        "_run_single_child",
    +        lambda *a, **k: {
    +            "task_index": 0,
    +            "status": "completed",
    +            "summary": "done",
    +            "api_calls": 1,
    +            "duration_seconds": 0.1,
    +            "model": "m",
    +            "exit_reason": "completed",
    +        },
    +    )
    +
    +    approval_token = [redacted:secret-assignment]"pre-compress-parent")
    +    session_tokens = [redacted:secret-assignment]
    +        source="tui",
    +        session_key="pre-compress-parent",
    +        ui_session_id="origin-tab",
    +    )
    +    try:
    +        out = dt.delegate_task(goal="bg task", background=True, parent_agent=parent)
    +        assert json.loads(out)["status"] == "dispatched"
    +        evt = _drain_one()
    +    finally:
    +        reset_current_session_key(approval_token)
    +        clear_session_vars(session_tokens)
    +
    +    assert evt is not None
    +    assert evt["type"] == "async_delegation"
    +    assert evt["session_key"] == "post-compress-tip"
    +    assert evt["origin_ui_session_id"] == "origin-tab"
    +
    +
     def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch):
         """A multi-item batch with background=True dispatches the WHOLE fan-out as
         ONE background unit (one handle, one async slot). The children run in

  • tests/tui_gateway/test_delegation_session_lifecycle.py

    @@ -0,0 +1,158 @@
    +"""Fail-closed ownership + session-scoped delegation lifecycle (#55578).
    +
    +Covers the two hardening rules layered on top of the origin-routing salvage:
    +
    +1. ``_session_owns_notification_event`` — positive-proof ownership. An
    +   async-delegation completion may only be injected into a session that
    +   PROVABLY commissioned it (origin UI id, or session-key/lineage match).
    +   Orphans are never adopted by a foreign chat.
    +
    +2. ``interrupt_for_session`` — a session's in-flight async delegations end
    +   with the session. ``_finalize_session`` interrupts delegations owned by
    +   the closing session (by origin UI id always; by durable key only when the
    +   TUI owns the lifecycle).
    +"""
    +
    +import threading
    +from unittest.mock import MagicMock, patch
    +
    +import pytest
    +
    +import tools.async_delegation as ad
    +from tui_gateway.server import (
    +    _finalize_session,
    +    _session_owns_notification_event,
    +)
    +
    +
    [redacted:email](autouse=True)
    +def _reset_async_delegation():
    +    ad._reset_for_tests()
    +    yield
    +    ad._reset_for_tests()
    +
    +
    +class TestSessionOwnsNotificationEvent:
    +    def _session(self, key="sess_key_1"):
    +        return {"session_key": key, "_finalized": False}
    +
    +    def test_origin_ui_match_owns(self):
    +        evt = {"type": "async_delegation", "origin_ui_session_id": "tab1", "session_key": "other"}
    +        assert _session_owns_notification_event("tab1", self._session(), evt) is True
    +
    +    def test_session_key_match_owns(self):
    +        evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": "sess_key_1"}
    +        assert _session_owns_notification_event("tabX", self._session("sess_key_1"), evt) is True
    +
    +    def test_orphan_is_not_owned(self):
    +        """No origin match, no key match, owner gone → NOT ours (fail closed)."""
    +        evt = {"type": "async_delegation", "origin_ui_session_id": "dead_tab", "session_key": "gone_key"}
    +        assert _session_owns_notification_event("tab1", self._session(), evt) is False
    +
    +    def test_empty_key_and_origin_not_owned(self):
    +        """A delegation event with no return address at all is never adopted."""
    +        evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": ""}
    +        assert _session_owns_notification_event("tab1", self._session(), evt) is False
    +
    +    def test_finalized_session_owns_nothing(self):
    +        evt = {"type": "async_delegation", "origin_ui_session_id": "tab1", "session_key": "sess_key_1"}
    +        sess = self._session()
    +        sess["_finalized"] = True
    +        assert _session_owns_notification_event("tab1", sess, evt) is False
    +
    +    def test_compression_chain_resolution_owns(self):
    +        evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": "parent_key"}
    +        db = MagicMock()
    +        db.resolve_resume_session_id.return_value = "child_key"
    +        with patch("tui_gateway.server._get_db", return_value=db):
    +            assert _session_owns_notification_event("tabX", self._session("child_key"), evt) is True
    +
    +
    +class TestInterruptForSession:
    +    def _seed_record(self, delegation_id, session_key="", origin_ui_session_id="", status="running"):
    +        fn = MagicMock()
    +        with ad._records_lock:
    +            ad._records[delegation_id] = {
    +                "delegation_id": delegation_id,
    +                "status": status,
    +                "session_key": session_key,
    +                "origin_ui_session_id": origin_ui_session_id,
    +                "interrupt_fn": fn,
    +            }
    +        return fn
    +
    +    def test_interrupts_only_matching_session(self):
    +        mine = self._seed_record("d1", session_key="sess_A")
    +        other = self._seed_record("d2", session_key="sess_B")
    +        n = ad.interrupt_for_session(session_key="sess_A")
    +        assert n == 1
    +        mine.assert_called_once()
    +        other.assert_not_called()
    +
    +    def test_matches_by_origin_ui_session_id(self):
    +        mine = self._seed_record("d1", origin_ui_session_id="tab1")
    +        other = self._seed_record("d2", origin_ui_session_id="tab2")
    +        n = ad.interrupt_for_session(origin_ui_session_id="tab1")
    +        assert n == 1
    +        mine.assert_called_once()
    +        other.assert_not_called()
    +
    +    def test_no_selector_is_noop(self):
    +        fn = self._seed_record("d1", session_key="sess_A")
    +        assert ad.interrupt_for_session() == 0
    +        fn.assert_not_called()
    +
    +    def test_completed_records_untouched(self):
    +        fn = self._seed_record("d1", session_key="sess_A", status="completed")
    +        assert ad.interrupt_for_session(session_key="sess_A") == 0
    +        fn.assert_not_called()
    +
    +
    +class TestFinalizeInterruptsOwnDelegations:
    +    def _make_session(self, session_key="sess_A", sid="tab1"):
    +        agent = MagicMock()
    +        agent.session_id = session_key
    +        agent._session_messages = None
    +        agent.model = "m"
    +        agent.platform = "tui"
    +        return {
    +            "agent": agent,
    +            "history": [{"role": "user", "content": "x"}],
    +            "history_lock": threading.Lock(),
    +            "session_key": session_key,
    +            "_finalized": False,
    +            "_sid": sid,
    +        }
    +
    +    @patch("tui_gateway.server._get_db")
    +    def test_finalize_interrupts_sessions_delegations(self, mock_get_db):
    +        mock_db = MagicMock()
    +        mock_db.get_session.return_value = {"source": "tui"}
    +        mock_get_db.return_value = mock_db
    +
    +        with patch("tools.async_delegation.interrupt_for_session") as mock_int:
    +            _finalize_session(self._make_session(), end_reason="tui_close")
    +
    +        mock_int.assert_called_once()
    +        kwargs = mock_int.call_args.kwargs
    +        assert kwargs["session_key"] == "sess_A"
    +        assert kwargs["origin_ui_session_id"] == "tab1"
    +
    +    @patch("tui_gateway.server._get_db")
    +    def test_viewer_of_gateway_session_only_interrupts_by_origin(self, mock_get_db):
    +        """Closing a TUI viewer tab on a live gateway session must not kill
    +        the gateway's own background work — key-based interrupt is skipped,
    +        origin-id interrupt (this tab's own dispatches) still applies."""
    +        mock_db = MagicMock()
    +        mock_db.get_session.return_value = {"source": "telegram"}
    +        mock_get_db.return_value = mock_db
    +
    +        with patch("tools.async_delegation.interrupt_for_session") as mock_int:
    +            _finalize_session(
    +                self._make_session(session_key="agent:main:telegram:dm:123", sid="tab9"),
    +                end_reason="ws_orphan_reap",
    +            )
    +
    +        kwargs = mock_int.call_args.kwargs
    +        assert kwargs["session_key"] == ""
    +        assert kwargs["origin_ui_session_id"] == "tab9"

  • tools/async_delegation.py

    @@ -130,6 +130,7 @@ def dispatch_async_delegation(
         model: Optional[str],
         session_key: str,
         runner: Callable[[], Dict[str, Any]],
    +    origin_ui_session_id: str = "",
         interrupt_fn: Optional[Callable[[], None]] = None,
         max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
     ) -> Dict[str, Any]:
    @@ -172,6 +173,7 @@ def dispatch_async_delegation(
             "role": role,
             "model": model,
             "session_key": session_key,
    +        "origin_ui_session_id": origin_ui_session_id,
             "status": "running",
             "dispatched_at": dispatched_at,
             "completed_at": None,
    @@ -282,6 +284,7 @@ def _push_completion_event(
             # session_key routes the completion back to the originating gateway
             # session; empty string => CLI (single-session) path.
             "session_key": record.get("session_key", ""),
    +        "origin_ui_session_id": record.get("origin_ui_session_id", ""),
             "goal": record.get("goal", ""),
             "context": record.get("context"),
             "toolsets": record.get("toolsets"),
    @@ -317,6 +320,7 @@ def dispatch_async_delegation_batch(
         model: Optional[str],
         session_key: str,
         runner: Callable[[], Dict[str, Any]],
    +    origin_ui_session_id: str = "",
         interrupt_fn: Optional[Callable[[], None]] = None,
         max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
     ) -> Dict[str, Any]:
    @@ -356,6 +360,7 @@ def dispatch_async_delegation_batch(
             "role": role,
             "model": model,
             "session_key": session_key,
    +        "origin_ui_session_id": origin_ui_session_id,
             "status": "running",
             "dispatched_at": dispatched_at,
             "completed_at": None,
    @@ -453,6 +458,7 @@ def _finalize_batch(
             "type": "async_delegation",
             "delegation_id": delegation_id,
             "session_key": event_record.get("session_key", ""),
    +        "origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
             "goal": event_record.get("goal", ""),
             "goals": event_record.get("goals"),
             "context": event_record.get("context"),
    @@ -519,6 +525,54 @@ def interrupt_all(reason: str = "shutdown") -> int:
         return count
     
     
    +def interrupt_for_session(
    +    session_key: str = "",
    +    origin_ui_session_id: str = "",
    +    reason: str = "session_end",
    +) -> int:
    +    """Signal running async delegations owned by ONE session to stop.
    +
    +    A delegation's lifecycle is bound to the session that spawned it: when
    +    that session ends, its in-flight background subagents must end with it —
    +    a completed orphan would otherwise sit on the shared completion queue
    +    with no live owner, either leaking into another chat or burning tokens
    +    with no one listening (#55578).
    +
    +    Matches on ``origin_ui_session_id`` (the live UI session that
    +    commissioned the work) and/or the durable ``session_key``; either
    +    matching field claims the record. Returns how many were interrupted.
    +    """
    +    if not session_key and not origin_ui_session_id:
    +        return 0
    +    count = 0
    +    with _records_lock:
    +        targets = [
    +            r for r in _records.values()
    +            if r.get("status") == "running"
    +            and (
    +                (origin_ui_session_id and str(r.get("origin_ui_session_id") or "") == origin_ui_session_id)
    +                or (session_key and str(r.get("session_key") or "") == session_key)
    +            )
    +        ]
    +    for r in targets:
    +        fn = r.get("interrupt_fn")
    +        if callable(fn):
    +            try:
    +                fn()
    +                count += 1
    +            except Exception as exc:
    +                logger.debug(
    +                    "interrupt_for_session: %s interrupt failed: %s",
    +                    r.get("delegation_id"), exc,
    +                )
    +    if count:
    +        logger.info(
    +            "Interrupted %d async delegation(s) for ending session (%s)",
    +            count, reason,
    +        )
    +    return count
    +
    +
     def _reset_for_tests() -> None:
         """Test-only: clear all state and tear down the executor."""
         global _executor, _executor_max_workers

  • tools/delegate_tool.py

    @@ -2796,6 +2796,25 @@ def _execute_and_aggregate() -> dict:
                 return json.dumps(_sync_result, ensure_ascii=False)
     
             _session_key = get_current_session_key(default="")
    +        _origin_ui_session_id = ""
    +        try:
    +            from gateway.session_context import get_session_env
    +
    +            _source = get_session_env("HERMES_SESSION_SOURCE", "")
    +            _origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "")
    +            # In desktop/TUI, the routable session key is the durable
    +            # AIAgent.session_id. Context compression can rotate that id during
    +            # the same turn before the TUI-side session dict is re-anchored;
    +            # if we capture the stale approval/session context key here, the
    +            # async completion becomes an orphan and any desktop poller may
    +            # consume it. Gateway chats are different: their session_key is the
    +            # platform conversation key (agent:main:...), so keep it there.
    +            if _source == "tui":
    +                _agent_session_id = str(getattr(parent_agent, "session_id", "") or "")
    +                if _agent_session_id:
    +                    _session_key = _agent_session_id
    +        except Exception:
    +            _origin_ui_session_id = ""
             _child_agents = [c for (_, _, c) in children]
     
             # Detach every child from the parent's interrupt-propagation list — the
    @@ -2836,6 +2855,7 @@ def _batch_interrupt():
                 role=top_role,
                 model=creds["model"],
                 session_key=_session_key,
    +            origin_ui_session_id=_origin_ui_session_id,
                 runner=_batch_runner,
                 interrupt_fn=_batch_interrupt,
                 max_async_children=_get_max_async_children(),

  • tui_gateway/server.py

    @@ -625,6 +625,7 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
         # Use session_id (from agent.session_id) not session_key — after compression,
         # session_key may be stale (the ended parent) while session_id is the live
         # continuation. Fix for #20001.
    +    _tui_owns_lifecycle = True
         if session_id:
             try:
                 db = _get_db()
    @@ -638,11 +639,40 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
                     # repeats on every inbound message.  (#60609)
                     row = db.get_session(session_id)
                     source = (row or {}).get("source", "")
    -                if not _is_gateway_owned_source(source):
    +                _tui_owns_lifecycle = not _is_gateway_owned_source(source)
    +                if _tui_owns_lifecycle:
                         db.end_session(session_id, end_reason)
             except Exception:
                 pass
     
    +    # A session's in-flight async delegations end WITH the session (#55578):
    +    # once nobody owns the return address, a still-running background subagent
    +    # can only burn tokens and park an orphaned completion on the shared
    +    # queue. Always interrupt delegations commissioned by THIS live UI session
    +    # (its sid); additionally interrupt by durable session_key, but only when
    +    # the TUI owns the lifecycle — closing a viewer tab on a live gateway
    +    # session must not kill the gateway's own background work.
    +    try:
    +        from tools.async_delegation import interrupt_for_session
    +
    +        _own_sid = str(session.get("_sid") or "")
    +        if not _own_sid:
    +            try:
    +                with _sessions_lock:
    +                    for _cand_sid, _cand in _sessions.items():
    +                        if _cand is session:
    +                            _own_sid = _cand_sid
    +                            break
    +            except Exception:
    +                _own_sid = ""
    +        interrupt_for_session(
    +            session_key=str(session_key or "") if _tui_owns_lifecycle else "",
    +            origin_ui_session_id=_own_sid,
    +            reason=end_reason,
    +        )
    +    except Exception:
    +        pass
    +
         # Close the slash-worker subprocess as part of finalize itself, not just
         # in the callers. Defense-in-depth: every session-end path goes through
         # _finalize_session (it's the single ``_finalized``-guarded chokepoint), so
    @@ -712,6 +742,10 @@ def _close_session_by_id(sid: str, *, end_reason: str = "tui_close") -> bool:
             session = _sessions.pop(sid, None)
         if session is None:
             return False
    +    # The session is already out of _sessions here, so downstream teardown
    +    # (e.g. _finalize_session's per-session async-delegation interrupt) can't
    +    # recover its live id by scanning the dict — stamp it on the record.
    +    session["_sid"] = sid
         _teardown_session(session, end_reason=end_reason)
         return True
     
    @@ -1946,7 +1980,12 @@ def _cwd_for_session_key(session_key: str) -> str:
         return ""
     
     
    -def _set_session_context(session_key: str, cwd: str | None = None) -> list:
    +def _set_session_context(
    +    session_key: str,
    +    cwd: str | None = None,
    +    *,
    +    ui_session_id: str = "",
    +) -> list:
         try:
             from gateway.session_context import set_session_vars
     
    @@ -1961,7 +2000,12 @@ def _set_session_context(session_key: str, cwd: str | None = None) -> list:
                     if sess.get("session_key") == session_key:
                         source = _session_source(sess)
                         break
    -        return set_session_vars(session_key=session_key, source=source, cwd=resolved)
    +        return set_session_vars(
    +            session_key=session_key,
    +            source=source,
    +            cwd=resolved,
    +            ui_session_id=ui_session_id,
    +        )
         except Exception:
             return []
     
    @@ -8451,22 +8495,76 @@ def run_after_agent_ready() -> None:
         return _ok(rid, {"status": "streaming"})
     
     
    -def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool:
    +def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> bool:
         """True if ``evt`` is owned by a *different* live session.
     
    -    Background-process events carry the ``session_key`` of the session that
    -    started the process. Since all desktop sessions share one process-wide
    -    completion queue, each poller must skip events it doesn't own so a
    -    background job's completion surfaces in the session that launched it — not
    -    whichever poller happened to dequeue first. Orphaned events (owner gone)
    -    and global/system events (empty ``session_key``) return False so the
    -    current poller still handles them rather than losing them.
    +    Background completions carry the ``session_key`` of the session that started
    +    the work. Async delegation completions from the desktop also carry
    +    ``origin_ui_session_id``: the live TUI tab/window that commissioned them.
    +    Since all desktop sessions share one process-wide completion queue, each
    +    poller must skip events it doesn't own so a detached result surfaces in the
    +    launching session, not whichever poller happened to dequeue first.
         """
    +    evt_ui_sid = str(evt.get("origin_ui_session_id") or "")
    +    if evt_ui_sid:
    +        if evt_ui_sid == str(sid or "") and not session.get("_finalized"):
    +            return False
    +        try:
    +            with _sessions_lock:
    +                owner_live = evt_ui_sid in _sessions and not _sessions[evt_ui_sid].get("_finalized")
    +        except Exception:
    +            owner_live = False
    +        if owner_live:
    +            return True
    +        # If the exact UI tab is gone, fall through to durable session_key
    +        # routing. That avoids wrong-session delivery while still allowing a
    +        # resumed continuation with the same durable key/lineage to claim it.
    +
         evt_key = str(evt.get("session_key") or "")
         if not evt_key:
             return False
    -    if evt_key == str(session.get("session_key") or ""):
    +
    +    current_keys = {
    +        str(session.get("session_key") or ""),
    +        _session_lookup_key(session, fallback=sid),
    +    }
    +
    +    # Compression can rotate AIAgent.session_id while the detached child is
    +    # still running. Resolve the event's original key to its continuation tip so
    +    # an event captured before or after compression still maps to the same live
    +    # desktop session instead of becoming an orphan that any poller may consume.
    +    resolved_key = evt_key
    +    try:
    +        db = _get_db()
    +        if db is not None:
    +            resolved_key = db.resolve_resume_session_id(evt_key) or evt_key
    +    except Exception:
    +        resolved_key = evt_key
    +
    +    # If the key has a live continuation, prefer that continuation over the
    +    # compressed parent. Otherwise a stale parent tab could consume the event
    +    # before the real current conversation sees it.
    +    if resolved_key != evt_key:
    +        if resolved_key in current_keys:
    +            return False
    +        try:
    +            with _sessions_lock:
    +                continuation_live = any(
    +                    not s.get("_finalized")
    +                    and (
    +                        str(s.get("session_key") or "") == resolved_key
    +                        or _session_lookup_key(s, fallback="") == resolved_key
    +                    )
    +                    for s in _sessions.values()
    +                )
    +        except Exception:
    +            continuation_live = False
    +        if continuation_live:
    +            return True
    +
    +    if evt_key in current_keys:
             return False
    +
         try:
             with _sessions_lock:
                 snapshot = list(_sessions.values())
    @@ -8476,11 +8574,50 @@ def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool:
             return False
     
         return any(
    -        s is not session and str(s.get("session_key") or "") == evt_key
    +        s is not session
    +        and not s.get("_finalized")
    +        and (
    +            str(s.get("session_key") or "") in {evt_key, resolved_key}
    +            or _session_lookup_key(s, fallback="") in {evt_key, resolved_key}
    +        )
             for s in snapshot
         )
     
     
    +def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool:
    +    """True iff *this* session PROVABLY owns ``evt``.
    +
    +    Positive ownership — the mirror of ``_notification_event_belongs_elsewhere``
    +    minus its orphan-adoption fallback. An event owns-matches when its
    +    ``origin_ui_session_id`` is this live session, or its ``session_key``
    +    (raw or resolved through the compression chain) matches this session's
    +    key/lineage. Used as a fail-closed gate for async-delegation payloads:
    +    "not provably elsewhere" is NOT good enough to inject a conversation
    +    payload into this chat (#55578).
    +    """
    +    if session.get("_finalized"):
    +        return False
    +    if str(evt.get("origin_ui_session_id") or "") == str(sid or ""):
    +        return True
    +    evt_key = str(evt.get("session_key") or "")
    +    if not evt_key:
    +        return False
    +    current_keys = {
    +        str(session.get("session_key") or ""),
    +        _session_lookup_key(session, fallback=sid),
    +    }
    +    if evt_key in current_keys:
    +        return True
    +    try:
    +        db = _get_db()
    +        resolved_key = (
    +            db.resolve_resume_session_id(evt_key) if db is not None else evt_key
    +        ) or evt_key
    +    except Exception:
    +        resolved_key = evt_key
    +    return resolved_key in current_keys
    +
    +
     def _notification_event_dedup_key(evt: dict) -> tuple:
         """Return the UI-emission identity for a process notification event.
     
    @@ -8546,11 +8683,37 @@ def _notification_poller_loop(
             # process started in session A would surface its completion in whichever
             # session's poller happened to wake first (Ben's "reported in a
             # different session" bug). Leave foreign events for their owner.
    -        if _notification_event_belongs_elsewhere(session, evt):
    +        if _notification_event_belongs_elsewhere(sid, session, evt):
                 process_registry.completion_queue.put(evt)
                 time.sleep(0.1)
                 continue
     
    +        # Fail closed for async-delegation results (#55578): these carry a
    +        # conversation payload, and injecting one into any chat other than the
    +        # one that commissioned it is a hard cross-session leak. The
    +        # belongs-elsewhere check above already re-queued events owned by
    +        # another LIVE session; what reaches here is either ours or an
    +        # orphan whose owner is gone. Orphaned delegation payloads are
    +        # DROPPED, not adopted — the subagent's summary is already persisted
    +        # in the delegation records/output store, so nothing is lost, whereas
    +        # a wrong-chat injection is unrecoverable. Non-delegation events
    +        # (background process completions etc.) keep the historical
    +        # adopt-orphans behavior.
    +        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
    +            sid, session, evt
    +        ):
    +            logger.warning(
    +                "async-delegation completion %s has no live owner "
    +                "(origin=%r key=%r); dropping from injection instead of "
    +                "delivering to session %s (#55578 fail-closed; result "
    +                "remains in the delegation records)",
    +                evt.get("delegation_id", "?"),
    +                str(evt.get("origin_ui_session_id") or ""),
    +                str(evt.get("session_key") or ""),
    +                sid,
    +            )
    +            continue
    +
             _evt_sid = evt.get("session_id", "")
             if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
                 continue
    @@ -8604,7 +8767,16 @@ def _notification_poller_loop(
                 evt = process_registry.completion_queue.get_nowait()
             except Exception:
                 break
    -        if _notification_event_belongs_elsewhere(session, evt):
    +        if _notification_event_belongs_elsewhere(sid, session, evt):
    +            deferred.append(evt)
    +            continue
    +        # Same fail-closed rule as the live loop: an orphaned async-delegation
    +        # payload is never adopted by a foreign session — defer it (a later
    +        # resume of the owner's lineage can still claim it) rather than
    +        # injecting another chat's conversation here (#55578).
    +        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
    +            sid, session, evt
    +        ):
                 deferred.append(evt)
                 continue
             _evt_sid = evt.get("session_id", "")
    @@ -8729,7 +8901,10 @@ def run():
                 )
     
                 approval_token = [redacted:secret-assignment]"session_key"])
    -            session_tokens = [redacted:secret-assignment]"session_key"])
    +            session_tokens = [redacted:secret-assignment]
    +                session["session_key"],
    +                ui_session_id=sid,
    +            )
                 _profile_home_str = session.get("profile_home")
                 if _profile_home_str:
                     home_token = [redacted:secret-assignment]

• #60869 fix(delegation): session-filter the post-turn notification drain (#58684)

  merged to main

  files: cli.py, gateway/run.py, tests/tools/test_process_registry.py, tools/process_registry.py, tui_gateway/server.py

  Show the diff

  • cli.py

    @@ -15082,7 +15082,9 @@ def process_loop():
                                 # and watch pattern matches) while agent is idle.
                                 try:
                                     from tools.process_registry import process_registry
    -                                for _evt, _synth in process_registry.drain_notifications():
    +                                from tools.approval import get_current_session_key
    +                                _drain_sk = get_current_session_key(default="")
    +                                for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk):
                                         self._pending_input.put(_synth)
                                 except Exception:
                                     pass

  • gateway/run.py

    @@ -15143,6 +15143,13 @@ def _build_process_event_source(self, evt: dict):
             chat_type = str(evt.get("chat_type") or derived_chat_type or "").strip().lower()
             chat_id = str(evt.get("chat_id") or derived_chat_id or "").strip()
             if not platform_name or not chat_type or not chat_id:
    +            logger.warning(
    +                "Synthetic event source unresolvable: "
    +                "session_key=%r platform=%r chat_type=%r chat_id=%r "
    +                "evt_type=%s",
    +                session_key, platform_name, chat_type, chat_id,
    +                evt.get("type", "?"),
    +            )
                 return None
     
             try:

  • tests/tools/test_process_registry.py

    @@ -1346,6 +1346,181 @@ def test_drain_notifications_empty_queue():
         assert results == []
     
     
    +def test_drain_notifications_filters_async_delegation_by_session_key():
    +    """Async-delegation events should only be consumed by the matching session's drain.
    +
    +    Regression test for issue #58684: background delegation results delivered
    +    to the wrong session when the user switches sessions while a subagent runs.
    +    """
    +    from tools.process_registry import process_registry
    +
    +    # Clear the queue first
    +    while not process_registry.completion_queue.empty():
    +        process_registry.completion_queue.get_nowait()
    +
    +    try:
    +        # Put events for different sessions
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_session_a",
    +            "session_key": "telegram:dm:111:user_a",
    +            "goal": "task A",
    +            "status": "completed",
    +            "summary": "done A",
    +            "api_calls": 1,
    +            "duration_seconds": 0.5,
    +        })
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_session_b",
    +            "session_key": "telegram:dm:222:user_b",
    +            "goal": "task B",
    +            "status": "completed",
    +            "summary": "done B",
    +            "api_calls": 1,
    +            "duration_seconds": 0.3,
    +        })
    +
    +        # Drain for session A — should only get deleg_session_a
    +        results_a = process_registry.drain_notifications(session_key="telegram:dm:111:user_a")
    +        assert len(results_a) == 1, (
    +            f"Expected 1 event for session A, got {len(results_a)}"
    +        )
    +        assert results_a[0][0]["delegation_id"] == "deleg_session_a"
    +        assert "done A" in results_a[0][1]
    +
    +        # Session B's event should have been re-queued — drain for session B
    +        results_b = process_registry.drain_notifications(session_key="telegram:dm:222:user_b")
    +        assert len(results_b) == 1, (
    +            f"Expected 1 event for session B, got {len(results_b)}"
    +        )
    +        assert results_b[0][0]["delegation_id"] == "deleg_session_b"
    +        assert "done B" in results_b[0][1]
    +
    +        # No more events should remain
    +        assert process_registry.completion_queue.empty()
    +    finally:
    +        while not process_registry.completion_queue.empty():
    +            process_registry.completion_queue.get_nowait()
    +
    +
    +def test_drain_notifications_no_filter_passes_all_async_delegation():
    +    """Without a session_key filter, all async-delegation events are consumed.
    +
    +    This ensures backward compatibility — the default (session_key="") permits
    +    all events, matching pre-fix behavior.
    +    """
    +    from tools.process_registry import process_registry
    +
    +    while not process_registry.completion_queue.empty():
    +        process_registry.completion_queue.get_nowait()
    +
    +    try:
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_1",
    +            "session_key": "telegram:dm:111:user_a",
    +            "goal": "task 1",
    +            "status": "completed",
    +            "summary": "done 1",
    +            "api_calls": 1,
    +            "duration_seconds": 0.5,
    +        })
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_2",
    +            "session_key": "telegram:dm:222:user_b",
    +            "goal": "task 2",
    +            "status": "completed",
    +            "summary": "done 2",
    +            "api_calls": 1,
    +            "duration_seconds": 0.3,
    +        })
    +
    +        # No filter — both should be consumed
    +        results = process_registry.drain_notifications()
    +        assert len(results) == 2, (
    +            f"Expected 2 events without filter, got {len(results)}"
    +        )
    +        ids = {r[0]["delegation_id"] for r in results}
    +        assert ids == {"deleg_1", "deleg_2"}
    +    finally:
    +        while not process_registry.completion_queue.empty():
    +            process_registry.completion_queue.get_nowait()
    +
    +
    +def test_drain_notifications_owns_event_callback_beats_key_equality():
    +    """The positive-proof ownership callback consumes ONLY approved events —
    +    including across a compression rotation where bare key equality would
    +    wrongly re-queue the session's own pre-compression dispatch (#55578)."""
    +    from tools.process_registry import process_registry
    +
    +    while not process_registry.completion_queue.empty():
    +        process_registry.completion_queue.get_nowait()
    +
    +    try:
    +        # Pre-compression dispatch: event carries the OLD key.
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_precompress",
    +            "session_key": "old_parent_key",
    +            "goal": "task", "status": "completed", "summary": "mine",
    +            "api_calls": 1, "duration_seconds": 0.1,
    +        })
    +        # Foreign event that plain key equality would also reject.
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_foreign",
    +            "session_key": "someone_else",
    +            "goal": "task", "status": "completed", "summary": "not mine",
    +            "api_calls": 1, "duration_seconds": 0.1,
    +        })
    +
    +        # Chain-aware ownership: this session's lineage includes old_parent_key.
    +        lineage = {"old_parent_key", "new_child_key"}
    +        results = process_registry.drain_notifications(
    +            session_key="new_child_key",
    +            owns_event=lambda e: e.get("session_key") in lineage,
    +        )
    +        assert [r[0]["delegation_id"] for r in results] == ["deleg_precompress"]
    +
    +        # The foreign event was re-queued, not consumed.
    +        leftover = process_registry.completion_queue.get_nowait()
    +        assert leftover["delegation_id"] == "deleg_foreign"
    +    finally:
    +        while not process_registry.completion_queue.empty():
    +            process_registry.completion_queue.get_nowait()
    +
    +
    +def test_drain_notifications_owns_event_callback_fails_closed():
    +    """A broken ownership callback must re-queue (never leak) the event."""
    +    from tools.process_registry import process_registry
    +
    +    while not process_registry.completion_queue.empty():
    +        process_registry.completion_queue.get_nowait()
    +
    +    try:
    +        process_registry.completion_queue.put({
    +            "type": "async_delegation",
    +            "delegation_id": "deleg_x",
    +            "session_key": "k",
    +            "goal": "task", "status": "completed", "summary": "s",
    +            "api_calls": 1, "duration_seconds": 0.1,
    +        })
    +
    +        def broken(_evt):
    +            raise RuntimeError("ownership check exploded")
    +
    +        results = process_registry.drain_notifications(
    +            session_key="k", owns_event=broken
    +        )
    +        assert results == []
    +        assert process_registry.completion_queue.get_nowait()["delegation_id"] == "deleg_x"
    +    finally:
    +        while not process_registry.completion_queue.empty():
    +            process_registry.completion_queue.get_nowait()
    +
    +
     # ---------------------------------------------------------------------------
     # _terminate_host_pid — cross-platform process-tree termination
     # ---------------------------------------------------------------------------

  • tools/process_registry.py

    @@ -1148,14 +1148,33 @@ def _drain_should_skip(self, session_id: str) -> bool:
             """
             return session_id in self._completion_consumed or session_id in self._poll_observed
     
    -    def drain_notifications(self) -> "list[tuple[dict, str]]":
    +    def drain_notifications(
    +        self, session_key: str = "", owns_event=None,
    +    ) -> "list[tuple[dict, str]]":
             """Pop all pending notification events and return formatted pairs.
     
             Returns a list of (raw_event, formatted_text) tuples.
             Skips completion events the agent already consumed via wait/log or
             observed inline via poll() (see ``_drain_should_skip``).
    +
    +        Async-delegation events carry a conversation payload, so draining one
    +        into the wrong session is a cross-chat leak (#58684, #55578). Two
    +        filter modes, strongest wins:
    +
    +        - ``owns_event(evt) -> bool``: positive-proof ownership callback.
    +          When provided, an async-delegation event is consumed ONLY if the
    +          callback returns True; everything else is re-queued for its owner.
    +          The TUI passes its compression-chain-aware ownership check here so
    +          a post-compression session still claims its own pre-compression
    +          dispatches.
    +        - ``session_key``: plain key equality (CLI and other single-session
    +          callers). Non-matching async-delegation events are re-queued.
    +
    +        With neither set, all events are consumed (legacy single-session
    +        behavior, backward compatible).
             """
    -        results = []
    +        results: "list[tuple[dict, str]]" = []
    +        requeue: "list[dict]" = []
             while not self.completion_queue.empty():
                 try:
                     evt = self.completion_queue.get_nowait()
    @@ -1164,9 +1183,28 @@ def drain_notifications(self) -> "list[tuple[dict, str]]":
                 _evt_sid = evt.get("session_id", "")
                 if evt.get("type") == "completion" and self._drain_should_skip(_evt_sid):
                     continue
    +            # Filter async-delegation events so they are not delivered to the
    +            # wrong session/thread (#58684). Positive-proof callback beats
    +            # bare key equality when the caller can provide one.
    +            if evt.get("type") == "async_delegation":
    +                if owns_event is not None:
    +                    try:
    +                        owned = bool(owns_event(evt))
    +                    except Exception:
    +                        owned = False  # fail closed — never leak on a broken check
    +                    if not owned:
    +                        requeue.append(evt)
    +                        continue
    +                elif session_key:
    +                    evt_session_key = evt.get("session_key", "") or ""
    +                    if evt_session_key != session_key:
    +                        requeue.append(evt)
    +                        continue
                 text = format_process_notification(evt)
                 if text:
                     results.append((evt, text))
    +        for evt in requeue:
    +            self.completion_queue.put(evt)
             return results
     
         def get(self, session_id: str) -> Optional[ProcessSession]:

  • tui_gateway/server.py

    @@ -9337,7 +9337,15 @@ def _stream(delta):
             try:
                 from tools.process_registry import process_registry
     
    -            for _evt, synth in process_registry.drain_notifications():
    +            # Positive-proof ownership (compression-chain aware) — the same
    +            # fail-closed gate the poller uses, so the post-turn drain can't
    +            # adopt another session's (or an orphan's) delegation payload,
    +            # while a post-compression session still claims its own
    +            # pre-compression dispatches (#55578).
    +            for _evt, synth in process_registry.drain_notifications(
    +                session_key=session.get("session_key", ""),
    +                owns_event=lambda e: _session_owns_notification_event(sid, session, e),
    +            ):
                     with session["history_lock"]:
                         if session.get("running"):
                             process_registry.completion_queue.put(_evt)

• #60871 fix(delegate): pin gateway async completions to the spawning session (#57498)

  merged to main

  files: gateway/run.py, gateway/slash_commands.py, tests/gateway/test_async_delegation_session_binding.py, tests/tools/test_async_delegation.py, tools/async_delegation.py, tools/delegate_tool.py

  Show the diff

  • gateway/run.py

    @@ -10633,6 +10633,46 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
     
             session_entry = self.session_store.get_or_create_session(source)
             session_key = session_entry.session_key
    +        pinned_session_id = str(
    +            (getattr(event, "metadata", None) or {}).get("gateway_session_id") or ""
    +        ).strip()
    +        if pinned_session_id and pinned_session_id != session_entry.session_id:
    +            # Fail closed (#55578): the spawning session may have ENDED since
    +            # dispatch (user /new-reset, compression rotation whose parent was
    +            # closed). switch_session() re-opens ended sessions, so pinning
    +            # blindly would RESURRECT a conversation the user explicitly
    +            # ended and inject into it — the same illicit-revival class as
    +            # the ws_orphan_reap loop (#60609). A completion whose spawning
    +            # session is dead is dropped from injection; the subagent's
    +            # output remains in the delegation records.
    +            pinned_row = None
    +            try:
    +                if self._session_db is not None:
    +                    # AsyncSessionDB already offloads to a thread.
    +                    pinned_row = await self._session_db.get_session(pinned_session_id)
    +            except Exception:
    +                pinned_row = None
    +            if pinned_row is None or pinned_row.get("ended_at"):
    +                logger.warning(
    +                    "Async-delegation completion pinned to session %s, which is "
    +                    "%s — dropping injection instead of resurrecting it "
    +                    "(#55578 fail-closed; result remains in the delegation "
    +                    "records).",
    +                    pinned_session_id,
    +                    "unknown" if pinned_row is None else "ended",
    +                )
    +                return
    +            prior_session_id = session_entry.session_id
    +            switched = self.session_store.switch_session(session_key, pinned_session_id)
    +            if switched is not None:
    +                session_entry = switched
    +                logger.info(
    +                    "Pinned async-delegation completion to spawning session %s "
    +                    "(was %s) for routing key %s (#57498)",
    +                    pinned_session_id,
    +                    prior_session_id,
    +                    session_key,
    +                )
             self._cache_session_source(session_key, source)
             if await asyncio.to_thread(self._is_telegram_topic_lane, source):
                 try:
    @@ -15202,12 +15242,17 @@ async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None:
             if not adapter:
                 return
             try:
    +            metadata = {}
    +            parent_session_id = str(evt.get("parent_session_id") or "").strip()
    +            if parent_session_id:
    +                metadata["gateway_session_id"] = parent_session_id
                 synth_event = MessageEvent(
                     text=synth_text,
                     message_type=MessageType.TEXT,
                     source=source,
                     internal=True,
                     message_id=str(evt.get("message_id") or "").strip() or None,
    +                metadata=metadata,
                 )
                 logger.info(
                     "Watch pattern notification — injecting for %s chat=%s thread=%s",

  • gateway/slash_commands.py

    @@ -167,6 +167,24 @@ async def _handle_reset_command(self, event: MessageEvent) -> Union[str, Ephemer
             if _qe is not None:
                 _qe.pop(session_key, None)
     
    +        # The old conversation's in-flight async delegations end WITH it
    +        # (#55578): after the reset rotates the session id, their completions
    +        # would have no live owner — a dangling subagent can only burn tokens
    +        # and park an orphaned payload on the shared queue. Interrupt by the
    +        # expiring durable session id (delegations dispatched from gateway
    +        # chats are pinned to it via parent_session_id) and by the routing
    +        # key as a fallback for older records.
    +        try:
    +            from tools.async_delegation import interrupt_for_session
    +
    +            interrupt_for_session(
    +                session_key=session_key,
    +                parent_session_id=str(getattr(old_entry, "session_id", "") or ""),
    +                reason="session_reset",
    +            )
    +        except Exception:
    +            pass
    +
             try:
                 from tools.env_passthrough import clear_env_passthrough
                 clear_env_passthrough()

  • tests/gateway/test_async_delegation_session_binding.py

    @@ -0,0 +1,130 @@
    +"""Gateway-side session binding for async delegations (#57498, #55578).
    +
    +Three invariants on the messaging-gateway surface, mirroring the TUI rules:
    +
    +1. Completions are pinned to the spawning session (contributor commit).
    +2. A dead/ended spawning session is never resurrected: the injection is
    +   dropped, fail-closed (never rerouted to the peer's current session).
    +3. /new interrupts the old conversation's in-flight async delegations.
    +"""
    +
    +import asyncio
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +import tools.async_delegation as ad
    +
    +
    [redacted:email](autouse=True)
    +def _reset_async_delegation():
    +    ad._reset_for_tests()
    +    yield
    +    ad._reset_for_tests()
    +
    +
    +def _seed_record(delegation_id, session_key="", parent_session_id="", status="running"):
    +    fn = MagicMock()
    +    with ad._records_lock:
    +        ad._records[delegation_id] = {
    +            "delegation_id": delegation_id,
    +            "status": status,
    +            "session_key": session_key,
    +            "parent_session_id": parent_session_id,
    +            "interrupt_fn": fn,
    +        }
    +    return fn
    +
    +
    +class TestInterruptForSessionByParentId:
    +    def test_parent_session_id_selector(self):
    +        mine = _seed_record("d1", session_key="agent:main:telegram:dm:1", parent_session_id="sess_old")
    +        other = _seed_record("d2", session_key="agent:main:telegram:dm:2", parent_session_id="sess_other")
    +        n = ad.interrupt_for_session(parent_session_id="sess_old")
    +        assert n == 1
    +        mine.assert_called_once()
    +        other.assert_not_called()
    +
    +    def test_reset_interrupts_by_key_and_parent(self):
    +        """A /new reset passes both selectors — either match claims the record."""
    +        by_key = _seed_record("d1", session_key="agent:main:telegram:dm:1", parent_session_id="")
    +        by_parent = _seed_record("d2", session_key="", parent_session_id="sess_old")
    +        unrelated = _seed_record("d3", session_key="other", parent_session_id="other")
    +        n = ad.interrupt_for_session(
    +            session_key="agent:main:telegram:dm:1",
    +            parent_session_id="sess_old",
    +            reason="session_reset",
    +        )
    +        assert n == 2
    +        by_key.assert_called_once()
    +        by_parent.assert_called_once()
    +        unrelated.assert_not_called()
    +
    +
    +class TestGatewayPinningFailsClosed:
    +    """The gateway injection path must never resurrect an ended session."""
    +
    +    def _make_runner(self, pinned_row):
    +        from gateway.run import GatewayRunner
    +
    +        runner = object.__new__(GatewayRunner)
    +        db = MagicMock()
    +        db.get_session = AsyncMock(return_value=pinned_row)
    +        runner._session_db = db
    +
    +        entry = MagicMock()
    +        entry.session_key = "agent:main:telegram:dm:1"
    +        entry.session_id = "sess_current"
    +        runner.session_store = MagicMock()
    +        runner.session_store.get_or_create_session.return_value = entry
    +        runner.session_store.switch_session.return_value = entry
    +        return runner, entry
    +
    +    def _run_pinning_prefix(self, runner, pinned_session_id):
    +        """Execute the pinning guard logic exactly as _handle_message does."""
    +
    +        async def _go():
    +            event = MagicMock()
    +            event.metadata = {"gateway_session_id": pinned_session_id}
    +            session_entry = runner.session_store.get_or_create_session(MagicMock())
    +            pinned = str((getattr(event, "metadata", None) or {}).get("gateway_session_id") or "").strip()
    +            if pinned and pinned != session_entry.session_id:
    +                pinned_row = None
    +                try:
    +                    if runner._session_db is not None:
    +                        pinned_row = await runner._session_db.get_session(pinned)
    +                except Exception:
    +                    pinned_row = None
    +                if pinned_row is None or pinned_row.get("ended_at"):
    +                    return "dropped"
    +                switched = runner.session_store.switch_session(session_entry.session_key, pinned)
    +                if switched is not None:
    +                    return "pinned"
    +            return "default"
    +
    +        return asyncio.run(_go())
    +
    +    def test_live_spawning_session_pins(self):
    +        runner, _ = self._make_runner({"id": "sess_old", "ended_at": None})
    +        assert self._run_pinning_prefix(runner, "sess_old") == "pinned"
    +
    +    def test_ended_spawning_session_drops(self):
    +        runner, _ = self._make_runner({"id": "sess_old", "ended_at": "2026-07-08T00:00:00"})
    +        assert self._run_pinning_prefix(runner, "sess_old") == "dropped"
    +        runner.session_store.switch_session.assert_not_called()
    +
    +    def test_unknown_spawning_session_drops(self):
    +        runner, _ = self._make_runner(None)
    +        assert self._run_pinning_prefix(runner, "sess_gone") == "dropped"
    +        runner.session_store.switch_session.assert_not_called()
    +
    +
    +class TestResetHandlerInterruptsDelegations:
    +    def test_reset_command_calls_interrupt_for_session(self):
    +        """The /new handler must sever the old conversation's delegations."""
    +        import inspect
    +        from gateway import slash_commands
    +
    +        src = inspect.getsource(slash_commands.GatewaySlashCommandsMixin._handle_reset_command)
    +        assert "interrupt_for_session" in src
    +        assert "session_reset" in src

  • tests/tools/test_async_delegation.py

    @@ -99,6 +99,7 @@ def runner():
         res = ad.dispatch_async_delegation(
             goal="compute X", context="some context", toolsets=["web", "file"],
             role="leaf", model="test-model", session_key="agent:main:cli:dm:local",
    +        parent_session_id="20260703_parent_sid",
             runner=runner, max_async_children=3,
         )
         assert res["status"] == "dispatched"
    @@ -108,6 +109,7 @@ def runner():
         assert evt["type"] == "async_delegation"
         assert evt["summary"] == "the result"
         assert evt["session_key"] == "agent:main:cli:dm:local"
    +    assert evt["parent_session_id"] == "20260703_parent_sid"
         assert evt["delegation_id"] == res["delegation_id"]
     
     

  • tools/async_delegation.py

    @@ -129,6 +129,7 @@ def dispatch_async_delegation(
         role: str,
         model: Optional[str],
         session_key: str,
    +    parent_session_id: Optional[str] = None,
         runner: Callable[[], Dict[str, Any]],
         origin_ui_session_id: str = "",
         interrupt_fn: Optional[Callable[[], None]] = None,
    @@ -146,6 +147,11 @@ def dispatch_async_delegation(
             captured on the parent thread BEFORE dispatch, because the daemon
             worker thread won't carry the contextvar. Used to route the
             completion back to the originating session.
    +    parent_session_id
    +        The durable ``state.db`` session id of the parent agent that spawned
    +        the delegation. Carried on the completion event so the gateway can
    +        pin routing to the spawning session instead of recovering the latest
    +        ``ended_at IS NULL`` row for the peer tuple (#57498).
         runner
             Zero-arg callable that builds + runs the child and returns the same
             result dict ``_run_single_child`` produces. Runs on the worker thread.
    @@ -174,6 +180,7 @@ def dispatch_async_delegation(
             "model": model,
             "session_key": session_key,
             "origin_ui_session_id": origin_ui_session_id,
    +        "parent_session_id": parent_session_id,
             "status": "running",
             "dispatched_at": dispatched_at,
             "completed_at": None,
    @@ -285,6 +292,7 @@ def _push_completion_event(
             # session; empty string => CLI (single-session) path.
             "session_key": record.get("session_key", ""),
             "origin_ui_session_id": record.get("origin_ui_session_id", ""),
    +        "parent_session_id": record.get("parent_session_id"),
             "goal": record.get("goal", ""),
             "context": record.get("context"),
             "toolsets": record.get("toolsets"),
    @@ -319,6 +327,7 @@ def dispatch_async_delegation_batch(
         role: str,
         model: Optional[str],
         session_key: str,
    +    parent_session_id: Optional[str] = None,
         runner: Callable[[], Dict[str, Any]],
         origin_ui_session_id: str = "",
         interrupt_fn: Optional[Callable[[], None]] = None,
    @@ -361,6 +370,7 @@ def dispatch_async_delegation_batch(
             "model": model,
             "session_key": session_key,
             "origin_ui_session_id": origin_ui_session_id,
    +        "parent_session_id": parent_session_id,
             "status": "running",
             "dispatched_at": dispatched_at,
             "completed_at": None,
    @@ -459,6 +469,7 @@ def _finalize_batch(
             "delegation_id": delegation_id,
             "session_key": event_record.get("session_key", ""),
             "origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
    +        "parent_session_id": event_record.get("parent_session_id"),
             "goal": event_record.get("goal", ""),
             "goals": event_record.get("goals"),
             "context": event_record.get("context"),
    @@ -528,6 +539,7 @@ def interrupt_all(reason: str = "shutdown") -> int:
     def interrupt_for_session(
         session_key: str = "",
         origin_ui_session_id: str = "",
    +    parent_session_id: str = "",
         reason: str = "session_end",
     ) -> int:
         """Signal running async delegations owned by ONE session to stop.
    @@ -538,11 +550,17 @@ def interrupt_for_session(
         with no live owner, either leaking into another chat or burning tokens
         with no one listening (#55578).
     
    -    Matches on ``origin_ui_session_id`` (the live UI session that
    -    commissioned the work) and/or the durable ``session_key``; either
    -    matching field claims the record. Returns how many were interrupted.
    +    Selectors (any matching field claims the record):
    +    - ``origin_ui_session_id``: the live TUI tab/window that commissioned it.
    +    - ``session_key``: the durable routing key captured at dispatch.
    +    - ``parent_session_id``: the spawning agent's durable session-db id —
    +      the right selector for gateway chats, whose ``session_key`` (the
    +      platform conversation key) SURVIVES a ``/new`` reset while the
    +      session id rotates.
    +
    +    Returns how many were interrupted.
         """
    -    if not session_key and not origin_ui_session_id:
    +    if not session_key and not origin_ui_session_id and not parent_session_id:
             return 0
         count = 0
         with _records_lock:
    @@ -552,6 +570,7 @@ def interrupt_for_session(
                 and (
                     (origin_ui_session_id and str(r.get("origin_ui_session_id") or "") == origin_ui_session_id)
                     or (session_key and str(r.get("session_key") or "") == session_key)
    +                or (parent_session_id and str(r.get("parent_session_id") or "") == parent_session_id)
                 )
             ]
         for r in targets:

  • tools/delegate_tool.py

    @@ -2815,6 +2815,7 @@ def _execute_and_aggregate() -> dict:
                         _session_key = _agent_session_id
             except Exception:
                 _origin_ui_session_id = ""
    +        _parent_session_id = getattr(parent_agent, "session_id", None)
             _child_agents = [c for (_, _, c) in children]
     
             # Detach every child from the parent's interrupt-propagation list — the
    @@ -2856,6 +2857,7 @@ def _batch_interrupt():
                 model=creds["model"],
                 session_key=_session_key,
                 origin_ui_session_id=_origin_ui_session_id,
    +            parent_session_id=_parent_session_id,
                 runner=_batch_runner,
                 interrupt_fn=_batch_interrupt,
                 max_async_children=_get_max_async_children(),

• #64593 fix(cli): stop new sessions adopting dead sessions' delegation completions (#64484)

  merged to main

  files: cli.py, tests/cli/test_cli_async_delegation_delivery.py, tests/tools/test_restored_delegation_ownership.py, tools/async_delegation.py, tools/delegate_tool.py, tools/process_registry.py

  Show the diff

  • cli.py

    @@ -9148,6 +9148,55 @@ def _get_goal_manager(self):
     
     
     
    +    def _owns_process_notification(self, event: dict) -> bool:
    +        """Return whether this CLI session provably owns a delegation event.
    +
    +        Delegations dispatched before context compression retain the original
    +        session key, so resolve that key to its continuation before comparing.
    +        Missing or foreign keys fail closed and remain queued for their owner.
    +        """
    +        event_key = str(event.get("session_key") or "")
    +        current_key = str(getattr(self, "session_id", "") or "")
    +        if not event_key or not current_key:
    +            return False
    +        if event_key == current_key:
    +            return True
    +        try:
    +            session_db = getattr(self, "_session_db", None)
    +            resolved_key = (
    +                session_db.resolve_resume_session_id(event_key)
    +                if session_db is not None
    +                else event_key
    +            ) or event_key
    +        except Exception:
    +            resolved_key = event_key
    +        return str(resolved_key) == current_key
    +
    +    def _drain_process_notifications(self, consumer: str) -> None:
    +        """Queue background notifications owned by this visible CLI session.
    +
    +        ``process_registry`` restores durable delegation completions into every
    +        process using the same Hermes profile.  Always pass this CLI's stable
    +        session identity when draining so another window cannot claim and mark
    +        delivered a completion that belongs to this one.
    +        """
    +        from tools.process_registry import process_registry
    +        from tools.async_delegation import (
    +            claim_event_delivery,
    +            complete_event_delivery,
    +        )
    +
    +        session_key = getattr(self, "session_id", "") or ""
    +        for event, synthetic_message in process_registry.drain_notifications(
    +            session_key=session_key,
    +            owns_event=self._owns_process_notification,
    +        ):
    +            claim = claim_event_delivery(event, consumer)
    +            if claim is None:
    +                continue
    +            self._pending_input.put(synthetic_message)
    +            complete_event_delivery(event, claim)
    +
         def _drain_interrupt_queue_to_pending_input(self) -> None:
             """Move stray messages from ``_interrupt_queue`` into ``_pending_input``.
     
    @@ -15321,18 +15370,7 @@ def process_loop():
                                 # Check for background process notifications (completions
                                 # and watch pattern matches) while agent is idle.
                                 try:
    -                                from tools.process_registry import process_registry
    -                                from tools.approval import get_current_session_key
    -                                _drain_sk = get_current_session_key(default="")
    -                                for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk):
    -                                    from tools.async_delegation import (
    -                                        claim_event_delivery, complete_event_delivery,
    -                                    )
    -                                    _claim = claim_event_delivery(_evt, "cli-idle")
    -                                    if _claim is None:
    -                                        continue
    -                                    self._pending_input.put(_synth)
    -                                    complete_event_delivery(_evt, _claim)
    +                                self._drain_process_notifications("cli-idle")
                                 except Exception:
                                     pass
                             continue
    @@ -15492,16 +15530,7 @@ def _restart_recording():
                             # Drain process notifications (completions + watch matches)
                             # that arrived while the agent was running.
                             try:
    -                            from tools.process_registry import process_registry
    -                            for _evt, _synth in process_registry.drain_notifications():
    -                                from tools.async_delegation import (
    -                                    claim_event_delivery, complete_event_delivery,
    -                                )
    -                                _claim = claim_event_delivery(_evt, "cli-post-turn")
    -                                if _claim is None:
    -                                    continue
    -                                self._pending_input.put(_synth)
    -                                complete_event_delivery(_evt, _claim)
    +                            self._drain_process_notifications("cli-post-turn")
                             except Exception:
                                 pass  # Non-fatal — don't break the main loop
     

  • tests/cli/test_cli_async_delegation_delivery.py

    @@ -0,0 +1,76 @@
    +"""Regression coverage for CLI async-delegation completion ownership."""
    +
    +import queue
    +
    +from cli import HermesCLI
    +
    +
    +def test_cli_completion_drain_uses_visible_session_identity(monkeypatch):
    +    """A CLI window must not claim another window's restored completion."""
    +    cli = HermesCLI.__new__(HermesCLI)
    +    cli.session_id = "visible-session"
    +    cli._pending_input = queue.Queue()
    +
    +    event = {
    +        "type": "async_delegation",
    +        "delegation_id": "deleg_visible",
    +        "session_key": "visible-session",
    +    }
    +    calls = []
    +
    +    class FakeRegistry:
    +        def drain_notifications(self, *, session_key="", owns_event=None):
    +            calls.append((session_key, owns_event(event)))
    +            return [(event, "completion payload")]
    +
    +    claimed = []
    +    completed = []
    +
    +    monkeypatch.setattr(
    +        "tools.process_registry.process_registry",
    +        FakeRegistry(),
    +    )
    +    monkeypatch.setattr(
    +        "tools.async_delegation.claim_event_delivery",
    +        lambda evt, consumer: claimed.append((evt, consumer)) or "claim-token",
    +    )
    +    monkeypatch.setattr(
    +        "tools.async_delegation.complete_event_delivery",
    +        lambda evt, token: [redacted:secret-assignment] token)),
    +    )
    +
    +    cli._drain_process_notifications("cli-idle")
    +
    +    assert calls == [("visible-session", True)]
    +    assert cli._pending_input.get_nowait() == "completion payload"
    +    assert claimed == [(event, "cli-idle")]
    +    assert completed == [(event, "claim-token")]
    +
    +
    +def test_cli_completion_ownership_rejects_foreign_session():
    +    cli = HermesCLI.__new__(HermesCLI)
    +    cli.session_id = "visible-session"
    +    cli._session_db = None
    +
    +    assert not cli._owns_process_notification(
    +        {"type": "async_delegation", "session_key": "foreign-session"}
    +    )
    +
    +
    +def test_cli_completion_ownership_accepts_compression_lineage():
    +    cli = HermesCLI.__new__(HermesCLI)
    +    cli.session_id = "visible-session"
    +
    +    class FakeSessionDB:
    +        def resolve_resume_session_id(self, session_id):
    +            assert session_id == "pre-compression-session"
    +            return "visible-session"
    +
    +    cli._session_db = FakeSessionDB()
    +
    +    assert cli._owns_process_notification(
    +        {
    +            "type": "async_delegation",
    +            "session_key": "pre-compression-session",
    +        }
    +    )

  • tests/tools/test_restored_delegation_ownership.py

    @@ -0,0 +1,149 @@
    +"""Regression coverage for #64484 — durable-restored delegation completions
    +must never be adopted by a session that cannot positively prove ownership.
    +
    +Layers under test:
    +1. ``restore_undelivered_completions`` stamps every restored event with
    +   ``restored=True`` (in-memory only).
    +2. ``ProcessRegistry.drain_notifications`` with NO filter (legacy
    +   consume-everything CLI path) re-queues restored events instead of
    +   consuming them.
    +3. Same-process (non-restored) keyless events keep the legacy behavior.
    +4. An owner with a matching session_key still receives its restored event.
    +"""
    +
    +import json
    +import queue
    +
    +from tools.process_registry import ProcessRegistry
    +
    +
    +def _make_registry():
    +    reg = ProcessRegistry.__new__(ProcessRegistry)
    +    import threading
    +
    +    reg._running = {}
    +    reg._finished = {}
    +    reg._lock = threading.Lock()
    +    reg.completion_queue = queue.Queue()
    +    reg._completion_consumed = set()
    +    reg._poll_observed = set()
    +    return reg
    +
    +
    +def _delegation_event(session_key="", restored=False, delegation_id="d1"):
    +    evt = {
    +        "type": "async_delegation",
    +        "delegation_id": delegation_id,
    +        "session_key": session_key,
    +        "origin_ui_session_id": "",
    +        "goal": "secret goal",
    +        "status": "success",
    +        "summary": "SECRET RESULT",
    +        "api_calls": 3,
    +        "duration_seconds": 1.5,
    +        "dispatched_at": 1.0,
    +        "completed_at": 2.0,
    +    }
    +    if restored:
    +        evt["restored"] = True
    +    return evt
    +
    +
    +def test_restore_stamps_restored_flag(tmp_path, monkeypatch):
    +    """Every durable completion re-enqueued at startup carries restored=True."""
    +    import tools.async_delegation as ad
    +
    +    monkeypatch.setattr(ad, "_db_path", lambda: tmp_path / "async_delegations.db")
    +    record = {
    +        "delegation_id": "d-old",
    +        "goal": "old goal",
    +        "context": None,
    +        "toolsets": None,
    +        "role": "leaf",
    +        "model": "m",
    +        "session_key": "OLD_SESSION_A",
    +        "origin_ui_session_id": "",
    +        "parent_session_id": "OLD_SESSION_A",
    +        "status": "running",
    +        "dispatched_at": 1.0,
    +        "completed_at": None,
    +        "interrupt_fn": None,
    +    }
    +    ad._persist_dispatch(record)
    +    evt = _delegation_event(session_key="OLD_SESSION_A", delegation_id="d-old")
    +    ad._persist_completion(evt, {"summary": "SECRET RESULT"})
    +
    +    q = queue.Queue()
    +    restored = ad.restore_undelivered_completions(q)
    +    assert restored == 1
    +    got = q.get_nowait()
    +    assert got["restored"] is True
    +    assert got["session_key"] == "OLD_SESSION_A"
    +
    +    # The stamp is in-memory only — the durable payload is unchanged.
    +    with ad._connect() as conn:
    +        row = conn.execute(
    +            "SELECT event_json FROM async_delegations WHERE delegation_id='d-old'"
    +        ).fetchone()
    +    assert "restored" not in json.loads(row[0])
    +
    +
    +def test_unfiltered_drain_never_consumes_restored_events():
    +    """The legacy consume-everything branch must fail closed on restored events."""
    +    reg = _make_registry()
    +    reg.completion_queue.put(_delegation_event(session_key="DEAD_SESSION", restored=True))
    +
    +    results = reg.drain_notifications()  # no filter — legacy CLI post-turn shape
    +
    +    assert results == []
    +    # Still queued for its real owner.
    +    assert reg.completion_queue.qsize() == 1
    +    assert reg.completion_queue.get_nowait()["session_key"] == "DEAD_SESSION"
    +
    +
    +def test_unfiltered_drain_keeps_legacy_behavior_for_same_process_events():
    +    """Non-restored keyless events (created by this process) are still consumed."""
    +    reg = _make_registry()
    +    reg.completion_queue.put(_delegation_event(session_key=""))
    +
    +    results = reg.drain_notifications()
    +
    +    assert len(results) == 1
    +    assert results[0][0]["delegation_id"] == "d1"
    +    assert reg.completion_queue.empty()
    +
    +
    +def test_owner_session_key_drain_consumes_restored_event():
    +    """The owning session (key match) still receives its restored completion."""
    +    reg = _make_registry()
    +    reg.completion_queue.put(_delegation_event(session_key="OWNER", restored=True))
    +
    +    results = reg.drain_notifications(session_key="OWNER")
    +
    +    assert len(results) == 1
    +    assert results[0][0]["session_key"] == "OWNER"
    +    assert reg.completion_queue.empty()
    +
    +
    +def test_foreign_session_key_drain_requeues_restored_event():
    +    """A different session's keyed drain must not claim the restored event."""
    +    reg = _make_registry()
    +    reg.completion_queue.put(_delegation_event(session_key="OWNER", restored=True))
    +
    +    results = reg.drain_notifications(session_key="SOMEONE_ELSE")
    +
    +    assert results == []
    +    assert reg.completion_queue.qsize() == 1
    +
    +
    +def test_owns_event_callback_beats_restored_flag():
    +    """A positive-proof ownership callback consumes restored events it owns."""
    +    reg = _make_registry()
    +    reg.completion_queue.put(_delegation_event(session_key="OWNER", restored=True))
    +
    +    results = reg.drain_notifications(
    +        owns_event=lambda e: e.get("session_key") == "OWNER"
    +    )
    +
    +    assert len(results) == 1
    +    assert reg.completion_queue.empty()

  • tools/async_delegation.py

    @@ -264,7 +264,17 @@ def recover_abandoned_delegations() -> int:
     
     
     def restore_undelivered_completions(target_queue) -> int:
    -    """Enqueue durable pending completions as fresh turns after process start."""
    +    """Enqueue durable pending completions as fresh turns after process start.
    +
    +    Every restored event is stamped ``restored=True`` (in-memory only — the
    +    stamp is added after the durable payload is deserialized and is never
    +    persisted). Restored events originate from a *previous* process, so no
    +    consumer in THIS process implicitly owns them: drain paths that run
    +    without an ownership filter (the legacy single-session behavior) must
    +    leave them queued for a consumer that can positively prove ownership,
    +    otherwise a brand-new session adopts a dead session's delegation
    +    results seconds after boot (#64484).
    +    """
         recover_abandoned_delegations()
         with _DB_LOCK, _connect() as conn:
             rows = conn.execute(
    @@ -273,7 +283,10 @@ def restore_undelivered_completions(target_queue) -> int:
                    ORDER BY completed_at, delegation_id"""
             ).fetchall()
             for _delegation_id, payload in rows:
    -            target_queue.put(json.loads(payload))
    +            evt = json.loads(payload)
    +            if isinstance(evt, dict):
    +                evt["restored"] = True
    +            target_queue.put(evt)
         return len(rows)
     
     

  • tools/delegate_tool.py

    @@ -2844,6 +2844,20 @@ def _execute_and_aggregate() -> dict:
                         _session_key = _agent_session_id
             except Exception:
                 _origin_ui_session_id = ""
    +        if not _session_key:
    +            # CLI (single-process) path: the approval contextvar is only bound
    +            # during gateway/TUI turns and HERMES_SESSION_KEY is not in the CLI
    +            # environment, so the key resolves empty here. Since #64240 the CLI
    +            # drains completions through a positive-ownership filter keyed on
    +            # the durable AIAgent.session_id — an empty session_key would fail
    +            # closed and the CLI could never claim its own completions, while
    +            # a restored foreign event with an empty key could leak into any
    +            # unfiltered consumer (#64484). Stamp the parent's durable session
    +            # id instead; compression rotations are handled on the drain side
    +            # via resolve_resume_session_id lineage resolution.
    +            _agent_session_id = str(getattr(parent_agent, "session_id", "") or "")
    +            if _agent_session_id:
    +                _session_key = _agent_session_id
             _parent_session_id = getattr(parent_agent, "session_id", None)
             _child_agents = [c for (_, _, c) in children]
     

  • tools/process_registry.py

    @@ -1211,6 +1211,19 @@ def drain_notifications(
                         if evt_session_key != session_key:
                             requeue.append(evt)
                             continue
    +                elif evt.get("restored"):
    +                    # Legacy unfiltered drain (no ownership callback, no
    +                    # session key). That behavior was safe when the in-memory
    +                    # queue could only hold events created by this very
    +                    # process — but durable restore (#63494) re-enqueues
    +                    # completions from PREVIOUS processes at startup, so an
    +                    # unfiltered consumer here would adopt a dead, unrelated
    +                    # session's conversation payload (#64484). Fail closed:
    +                    # leave restored events queued (still 'pending' on disk)
    +                    # for a consumer that can positively prove ownership,
    +                    # e.g. the owning session's --resume.
    +                    requeue.append(evt)
    +                    continue
                 text = format_process_notification(evt)
                 if text:
                     results.append((evt, text))

• #65419 fix(tui): route background-process notifications to their owning session (#35652)

  merged to main

  files: scripts/release.py, tests/test_tui_gateway_server.py, tests/tools/test_process_registry.py, tools/process_registry.py, tui_gateway/server.py

  Show the diff

  • scripts/release.py

    @@ -45,6 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    +    "[redacted:email]": "2751738943",  # PR #54785 salvage (tui: post-turn completion ownership routing)
         "[redacted:email]": "Burgunthy",  # PR #20096 salvage (gateway: profile-based routing for inbound messages)
         "[redacted:email]": "webtecnica",  # PR #63360 salvage (nous: restore inference-api base_url)
         "[redacted:email]": "Epoxidex",  # PR #29820 salvage (ollama: top-level reasoning_effort=none; #25758)

  • tests/test_tui_gateway_server.py

    @@ -2284,7 +2284,611 @@ def replace_messages(self, key, messages):
             server._sessions.pop("trunc-sid", None)
     
     
    -def test_session_create_does_not_persist_empty_row(monkeypatch):
    +class _StopAfterOneNotificationPoll:
    +    def __init__(self):
    +        self._checks = 0
    +
    +    def is_set(self):
    +        self._checks += 1
    +        return self._checks > 1
    +
    +
    +def test_notification_poller_live_loop_requeues_foreign_completion_for_owner(
    +    monkeypatch,
    +):
    +    """A foreign live-loop dequeue is handed back to its proven owner."""
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    delivered = {"a": [], "b": []}
    +    emitted = []
    +    session_a = _session(session_key="session-a-live-handoff")
    +    session_b = _session(session_key="session-b-live-handoff")
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc-live-handoff",
    +        "session_key": "session-a-live-handoff",
    +        "command": "echo owner",
    +        "exit_code": 0,
    +        "output": "owner",
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    monkeypatch.setattr(server, "_get_db", lambda: None)
    +    monkeypatch.setattr(server, "_emit", lambda *args, **_kwargs: emitted.append(args))
    +
    +    def _deliver(_rid, sid, session, text):
    +        delivered["a" if sid == "sid-a-live-handoff" else "b"].append(text)
    +        session["running"] = False
    +
    +    monkeypatch.setattr(server, "_run_prompt_submit", _deliver)
    +    server._sessions.update(
    +        {
    +            "sid-a-live-handoff": session_a,
    +            "sid-b-live-handoff": session_b,
    +        }
    +    )
    +    process_registry._completion_consumed.discard(event["session_id"])
    +
    +    try:
    +        server._notification_poller_loop(
    +            _StopAfterOneNotificationPoll(), "sid-b-live-handoff", session_b
    +        )
    +
    +        assert delivered["b"] == []
    +        assert emitted == []
    +        assert isolated_queue.qsize() == 1
    +        assert isolated_queue.queue[0] is event
    +
    +        server._notification_poller_loop(
    +            _StopAfterOneNotificationPoll(), "sid-a-live-handoff", session_a
    +        )
    +
    +        assert len(delivered["a"]) == 1
    +        assert "proc-live-handoff completed normally" in delivered["a"][0]
    +        assert delivered["b"] == []
    +        assert isolated_queue.empty()
    +    finally:
    +        server._sessions.pop("sid-a-live-handoff", None)
    +        server._sessions.pop("sid-b-live-handoff", None)
    +        process_registry._completion_consumed.discard(event["session_id"])
    +        while not isolated_queue.empty():
    +            isolated_queue.get_nowait()
    +
    +
    +def test_completion_ownership_lineage_lookup_failure_fails_closed(monkeypatch):
    +    """A provenance lookup failure cannot turn an addressed event into ours."""
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    class _BrokenDB:
    +        def resolve_resume_session_id(self, _session_key):
    +            raise RuntimeError("lineage database unavailable")
    +
    +    session = _session(session_key="unrelated-live-session")
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc-unknown-lineage",
    +        "session_key": "unknown-parent",
    +        "command": "echo unknown",
    +        "exit_code": 0,
    +        "output": "unknown",
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    monkeypatch.setattr(server, "_get_db", lambda: _BrokenDB())
    +
    +    drained = process_registry.drain_notifications(
    +        session_key="unrelated-live-session",
    +        owns_event=lambda candidate: server._session_owns_notification_event(
    +            "sid-unrelated-live", session, candidate
    +        ),
    +    )
    +
    +    assert drained == []
    +    assert isolated_queue.qsize() == 1
    +    assert isolated_queue.get_nowait() is event
    +
    +
    [redacted:email](
    +    "routing",
    +    [
    +        {"session_key": "missing-owner-key"},
    +        {"origin_ui_session_id": "missing-owner-sid"},
    +    ],
    +)
    +def test_notification_poller_live_loop_drops_addressed_orphan(
    +    monkeypatch, routing
    +):
    +    """A live poll never injects an addressed event whose owner is gone."""
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    delivered = []
    +    emitted = []
    +    session = _session(session_key="unrelated-live-key")
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc-live-orphan",
    +        "command": "echo orphan",
    +        "exit_code": 0,
    +        "output": "orphan",
    +        **routing,
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    monkeypatch.setattr(server, "_get_db", lambda: None)
    +    monkeypatch.setattr(server, "_emit", lambda *args, **_kwargs: emitted.append(args))
    +    monkeypatch.setattr(
    +        server,
    +        "_run_prompt_submit",
    +        lambda _rid, _sid, _session, text: delivered.append(text),
    +    )
    +    server._sessions["sid-live-orphan"] = session
    +    process_registry._completion_consumed.discard(event["session_id"])
    +
    +    try:
    +        server._notification_poller_loop(
    +            _StopAfterOneNotificationPoll(), "sid-live-orphan", session
    +        )
    +
    +        assert delivered == []
    +        assert emitted == []
    +        assert isolated_queue.empty()
    +    finally:
    +        server._sessions.pop("sid-live-orphan", None)
    +        process_registry._completion_consumed.discard(event["session_id"])
    +        while not isolated_queue.empty():
    +            isolated_queue.get_nowait()
    +
    +
    [redacted:email](
    +    "routing",
    +    [
    +        {"session_key": "session-b"},
    +        {"origin_ui_session_id": "sid_gone"},
    +    ],
    +)
    +def test_notification_poller_drops_orphaned_events(monkeypatch, routing):
    +    """Addressed completions whose owner is gone are dropped, not hijacked."""
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    emitted = []
    +    delivered = []
    +    sess = _session(session_key="session-a")
    +    server._sessions["sid_a"] = sess
    +    monkeypatch.setattr(server, "_emit", lambda *a, **kw: emitted.append(a))
    +    monkeypatch.setattr(
    +        server,
    +        "_run_prompt_submit",
    +        lambda _rid, _sid, _session, text: delivered.append(text),
    +    )
    +    monkeypatch.setattr(server, "_get_db", lambda: None)
    +
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    process_registry._completion_consumed.discard("proc_ghost")
    +    isolated_queue.put(
    +        {
    +            "type": "completion",
    +            "session_id": "proc_ghost",
    +            "command": "echo from ghost",
    +            "exit_code": 0,
    +            "output": "ghost output",
    +            **routing,
    +        }
    +    )
    +
    +    stop = threading.Event()
    +    stop.set()
    +
    +    try:
    +        server._notification_poller_loop(stop, "sid_a", sess)
    +
    +        assert [a for a in emitted if a[0] == "status.update"] == []
    +        assert delivered == []
    +    finally:
    +        server._sessions.pop("sid_a", None)
    +        while not process_registry.completion_queue.empty():
    +            process_registry.completion_queue.get_nowait()
    +
    +
    [redacted:email](
    +    ("routing", "resolved_key"),
    +    [
    +        ({"session_key": "session-a"}, None),
    +        (
    +            {
    +                "session_key": "stale-durable-key",
    +                "origin_ui_session_id": "sid_a",
    +            },
    +            None,
    +        ),
    +        ({"session_key": "old-parent-key"}, "session-a"),
    +    ],
    +)
    +def test_notification_poller_delivers_owned_events(
    +    monkeypatch, routing, resolved_key
    +):
    +    """Direct, UI-origin, and compression-lineage owners are delivered."""
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    class _CompressionDB:
    +        def resolve_resume_session_id(self, key):
    +            return resolved_key if key == "old-parent-key" and resolved_key else key
    +
    +    delivered = []
    +    emitted = []
    +    sess = _session(session_key="session-a")
    +    server._sessions["sid_a"] = sess
    +    monkeypatch.setattr(server, "_emit", lambda *a, **kw: emitted.append(a))
    +    monkeypatch.setattr(
    +        server,
    +        "_run_prompt_submit",
    +        lambda _rid, _sid, _session, text: delivered.append(text),
    +    )
    +    monkeypatch.setattr(server, "_get_db", lambda: _CompressionDB())
    +
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    process_registry._completion_consumed.discard("proc_mine")
    +    isolated_queue.put(
    +        {
    +            "type": "completion",
    +            "session_id": "proc_mine",
    +            "command": "echo mine",
    +            "exit_code": 0,
    +            "output": "mine",
    +            **routing,
    +        }
    +    )
    +
    +    stop = threading.Event()
    +    stop.set()
    +
    +    try:
    +        server._notification_poller_loop(stop, "sid_a", sess)
    +
    +        status_calls = [a for a in emitted if a[0] == "status.update"]
    +        assert len(status_calls) == 1
    +        assert status_calls[0][2]["kind"] == "process"
    +        assert len(delivered) == 1
    +        assert "proc_mine" in delivered[0]
    +    finally:
    +        server._sessions.pop("sid_a", None)
    +        while not process_registry.completion_queue.empty():
    +            process_registry.completion_queue.get_nowait()
    +
    +
    +def _configure_immediate_prompt_run(
    +    monkeypatch, tmp_path, *, immediate_threads=True
    +):
    +    class _ImmediateThread:
    +        def __init__(self, target=None, daemon=None, **_kwargs):
    +            self._target = target
    +
    +        def start(self):
    +            if self._target is not None:
    +                self._target()
    +
    +        def is_alive(self):
    +            return False
    +
    +    if immediate_threads:
    +        monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
    +    monkeypatch.setattr(server, "_emit", lambda *_args, **_kwargs: None)
    +    monkeypatch.setattr(server, "make_stream_renderer", lambda _cols: None)
    +    monkeypatch.setattr(server, "render_message", lambda _raw, _cols: None)
    +    monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
    +    monkeypatch.setattr(server, "_sync_agent_model_with_config", lambda *_args: None)
    +    monkeypatch.setattr(server, "_session_cwd", lambda _session: str(tmp_path))
    +    monkeypatch.setattr(server, "_register_session_cwd", lambda _session: None)
    +    monkeypatch.setattr(server, "_set_session_context", lambda *_args, **_kwargs: [])
    +    monkeypatch.setattr(server, "_clear_session_context", lambda _tokens: None)
    +    monkeypatch.setattr(server, "_session_info", lambda *_args: {})
    +    monkeypatch.setattr(server, "_get_usage", lambda _agent: {})
    +    monkeypatch.setattr(
    +        server, "_sync_session_key_after_compress", lambda *_args, **_kwargs: None
    +    )
    +    monkeypatch.setattr(server, "_drain_queued_prompt", lambda *_args: False)
    +    monkeypatch.setattr(server, "_voice_tts_enabled", lambda: False)
    +    monkeypatch.setattr(server, "_get_db", lambda: None)
    +
    +
    +class _RecordingAgent:
    +    model = "test-model"
    +    provider = "test-provider"
    +
    +    def __init__(self, turns):
    +        self._turns = turns
    +
    +    def clear_interrupt(self):
    +        return None
    +
    +    def run_conversation(
    +        self, prompt, conversation_history=None, stream_callback=None
    +    ):
    +        self._turns.append(prompt)
    +        return {"final_response": "", "messages": []}
    +
    +
    [redacted:email]("exit_code", [0, 7])
    +def test_run_prompt_submit_requeues_foreign_completion(
    +    monkeypatch, tmp_path, exit_code
    +):
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    _configure_immediate_prompt_run(monkeypatch, tmp_path)
    +    turns = []
    +    session_a = _session(session_key="session-a")
    +    session_b = _session(
    +        session_key="session-b",
    +        agent=_RecordingAgent(turns),
    +        running=True,
    +    )
    +    event = {
    +        "type": "completion",
    +        "session_id": f"proc_foreign_{exit_code}",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": exit_code,
    +        "output": "foreign",
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    server._sessions["sid_a"] = session_a
    +    server._sessions["sid_b"] = session_b
    +
    +    try:
    +        server._run_prompt_submit("rid-b", "sid_b", session_b, "session-b-turn")
    +
    +        assert turns == ["session-b-turn"]
    +        assert isolated_queue.get_nowait() == event
    +        assert isolated_queue.empty()
    +    finally:
    +        server._sessions.pop("sid_a", None)
    +        server._sessions.pop("sid_b", None)
    +        process_registry._completion_consumed.discard(event["session_id"])
    +
    +
    +def test_run_prompt_submit_delivers_completion_observed_by_poll(monkeypatch, tmp_path):
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    _configure_immediate_prompt_run(monkeypatch, tmp_path)
    +    turns = []
    +    session = _session(
    +        session_key="session-a",
    +        agent=_RecordingAgent(turns),
    +        running=True,
    +    )
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_polled",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "observed but not consumed",
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    process_registry._completion_consumed.discard(event["session_id"])
    +    process_registry._poll_observed.add(event["session_id"])
    +    server._sessions["sid_a"] = session
    +
    +    try:
    +        server._run_prompt_submit("rid-a", "sid_a", session, "session-a-turn")
    +
    +        assert turns[0] == "session-a-turn"
    +        assert len(turns) == 2
    +        assert "proc_polled" in turns[1]
    +        assert isolated_queue.empty()
    +    finally:
    +        server._sessions.pop("sid_a", None)
    +        process_registry._completion_consumed.discard(event["session_id"])
    +        process_registry._poll_observed.discard(event["session_id"])
    +
    +
    +def test_run_prompt_submit_requeues_all_unstarted_notifications_with_real_threading(
    +    monkeypatch, tmp_path
    +):
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    _configure_immediate_prompt_run(
    +        monkeypatch, tmp_path, immediate_threads=False
    +    )
    +    real_thread_class = threading.Thread
    +    threads = []
    +    nested_started = threading.Event()
    +    release_nested = threading.Event()
    +    turns = []
    +
    +    def _recording_thread(*args, **kwargs):
    +        thread = real_thread_class(*args, **kwargs)
    +        threads.append(thread)
    +        return thread
    +
    +    class _BlockingNotificationAgent(_RecordingAgent):
    +        def run_conversation(
    +            self, prompt, conversation_history=None, stream_callback=None
    +        ):
    +            turns.append(prompt)
    +            if "proc_batch_1" in prompt:
    +                nested_started.set()
    +                if not release_nested.wait(timeout=5):
    +                    raise TimeoutError("notification turn was not released")
    +            return {"final_response": "", "messages": []}
    +
    +    monkeypatch.setattr(server.threading, "Thread", _recording_thread)
    +    session = _session(
    +        session_key="session-a",
    +        agent=_BlockingNotificationAgent(turns),
    +        running=True,
    +    )
    +    events = [
    +        {
    +            "type": "completion",
    +            "session_id": f"proc_batch_{index}",
    +            "session_key": "session-a",
    +            "command": "safe-test-command",
    +            "exit_code": 0,
    +            "output": f"owned-{index}",
    +        }
    +        for index in range(1, 4)
    +    ]
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    for event in events:
    +        isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    server._sessions["sid_a"] = session
    +
    +    try:
    +        server._run_prompt_submit("rid-a", "sid_a", session, "session-a-turn")
    +
    +        assert nested_started.wait(timeout=5)
    +        threads[0].join(timeout=5)
    +        assert not threads[0].is_alive()
    +        queued = []
    +        while not isolated_queue.empty():
    +            queued.append(isolated_queue.get_nowait())
    +        assert [event["session_id"] for event in queued] == [
    +            "proc_batch_2",
    +            "proc_batch_3",
    +        ]
    +    finally:
    +        release_nested.set()
    +        for thread in threads:
    +            thread.join(timeout=5)
    +        server._sessions.pop("sid_a", None)
    +        while not isolated_queue.empty():
    +            isolated_queue.get_nowait()
    +        for event in events:
    +            process_registry._completion_consumed.discard(event["session_id"])
    +            process_registry._poll_observed.discard(event["session_id"])
    +
    +
    +def test_run_prompt_submit_delivers_completion_owned_through_compression_lineage(
    +    monkeypatch, tmp_path
    +):
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    class _CompressionDB:
    +        def resolve_resume_session_id(self, key):
    +            return "new-child-key" if key == "old-parent-key" else key
    +
    +    _configure_immediate_prompt_run(monkeypatch, tmp_path)
    +    monkeypatch.setattr(server, "_get_db", lambda: _CompressionDB())
    +    ownership_checks = []
    +    original_owns_event = server._session_owns_notification_event
    +
    +    def _record_ownership_check(sid, checked_session, checked_event):
    +        ownership_checks.append(checked_event["session_id"])
    +        return original_owns_event(sid, checked_session, checked_event)
    +
    +    monkeypatch.setattr(
    +        server, "_session_owns_notification_event", _record_ownership_check
    +    )
    +    turns = []
    +    session = _session(
    +        session_key="new-child-key",
    +        agent=_RecordingAgent(turns),
    +        running=True,
    +    )
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_precompression",
    +        "session_key": "old-parent-key",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "owned",
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    server._sessions["sid_b"] = session
    +
    +    try:
    +        server._run_prompt_submit("rid-b", "sid_b", session, "session-b-turn")
    +
    +        assert turns[0] == "session-b-turn"
    +        assert len(turns) == 2
    +        assert "proc_precompression" in turns[1]
    +        assert ownership_checks == ["proc_precompression"]
    +        assert isolated_queue.empty()
    +    finally:
    +        server._sessions.pop("sid_b", None)
    +        process_registry._completion_consumed.discard(event["session_id"])
    +
    +
    +def test_run_prompt_submit_prefers_origin_ui_session_id(monkeypatch, tmp_path):
    +    import queue as _queue_mod
    +
    +    from tools.process_registry import process_registry
    +
    +    _configure_immediate_prompt_run(monkeypatch, tmp_path)
    +    ownership_checks = []
    +    original_owns_event = server._session_owns_notification_event
    +
    +    def _record_ownership_check(sid, checked_session, checked_event):
    +        ownership_checks.append(checked_event["session_id"])
    +        return original_owns_event(sid, checked_session, checked_event)
    +
    +    monkeypatch.setattr(
    +        server, "_session_owns_notification_event", _record_ownership_check
    +    )
    +    turns = []
    +    session = _session(
    +        session_key="current-key",
    +        agent=_RecordingAgent(turns),
    +        running=True,
    +    )
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_origin_owned",
    +        "session_key": "stale-durable-key",
    +        "origin_ui_session_id": "sid_b",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "owned",
    +    }
    +    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
    +    isolated_queue.put(event)
    +    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
    +    server._sessions["sid_b"] = session
    +
    +    try:
    +        server._run_prompt_submit("rid-b", "sid_b", session, "session-b-turn")
    +
    +        assert turns[0] == "session-b-turn"
    +        assert len(turns) == 2
    +        assert "proc_origin_owned" in turns[1]
    +        assert ownership_checks == ["proc_origin_owned"]
    +        assert isolated_queue.empty()
    +    finally:
    +        server._sessions.pop("sid_b", None)
    +        process_registry._completion_consumed.discard(event["session_id"])
    +
    +
    +
         """session.create must NOT eagerly write a DB row.
     
         Every TUI/desktop launch opens a session here just to paint the composer;

  • tests/tools/test_process_registry.py

    @@ -1336,6 +1336,66 @@ def test_drain_notifications_skips_consumed():
                 process_registry.completion_queue.get_nowait()
     
     
    +def test_drain_notifications_can_deliver_poll_observed_for_gateway(registry):
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_polled",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "observed but not consumed",
    +    }
    +    registry._poll_observed.add(event["session_id"])
    +    registry.completion_queue.put(event)
    +
    +    try:
    +        results = registry.drain_notifications(
    +            session_key="session-a",
    +            owns_event=lambda _event: True,
    +            skip_poll_observed=False,
    +        )
    +
    +        assert [raw for raw, _ in results] == [event]
    +    finally:
    +        registry._poll_observed.discard(event["session_id"])
    +
    +
    [redacted:email](
    +    "skip_state", ["_poll_observed", "_completion_consumed"]
    +)
    +def test_drain_notifications_routes_foreign_before_local_skip(
    +    registry, skip_state
    +):
    +    event = {
    +        "type": "completion",
    +        "session_id": f"proc_foreign_{skip_state}",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "foreign",
    +    }
    +    ownership_calls = []
    +    getattr(registry, skip_state).add(event["session_id"])
    +    registry.completion_queue.put(event)
    +
    +    def owns_event(checked_event):
    +        ownership_calls.append(checked_event)
    +        return False
    +
    +    try:
    +        results = registry.drain_notifications(
    +            session_key="session-b",
    +            owns_event=owns_event,
    +        )
    +
    +        assert results == []
    +        assert ownership_calls == [event]
    +        assert registry.completion_queue.get_nowait() == event
    +        assert registry.completion_queue.empty()
    +    finally:
    +        getattr(registry, skip_state).discard(event["session_id"])
    +
    +
     def test_drain_notifications_empty_queue():
         from tools.process_registry import process_registry
     
    @@ -1346,6 +1406,151 @@ def test_drain_notifications_empty_queue():
         assert results == []
     
     
    [redacted:email]("exit_code", [0, 7])
    +def test_drain_notifications_filters_addressed_completion_by_owns_event(
    +    registry, exit_code
    +):
    +    owned = {
    +        "type": "completion",
    +        "session_id": f"proc_owned_{exit_code}",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": exit_code,
    +        "output": "owned",
    +    }
    +    foreign = {
    +        "type": "completion",
    +        "session_id": f"proc_foreign_{exit_code}",
    +        "session_key": "session-b",
    +        "command": "safe-test-command",
    +        "exit_code": exit_code,
    +        "output": "foreign",
    +    }
    +    registry.completion_queue.put(owned)
    +    registry.completion_queue.put(foreign)
    +
    +    results = registry.drain_notifications(
    +        session_key="session-a",
    +        owns_event=lambda event: event.get("session_key") == "session-a",
    +    )
    +
    +    assert [event["session_id"] for event, _ in results] == [
    +        f"proc_owned_{exit_code}"
    +    ]
    +    assert registry.completion_queue.get_nowait() == foreign
    +    assert registry.completion_queue.empty()
    +
    +
    +def test_drain_notifications_filters_addressed_completion_by_session_key(registry):
    +    owned = {
    +        "type": "completion",
    +        "session_id": "proc_owned",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "owned",
    +    }
    +    foreign = {
    +        "type": "completion",
    +        "session_id": "proc_foreign",
    +        "session_key": "session-b",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "foreign",
    +    }
    +    registry.completion_queue.put(owned)
    +    registry.completion_queue.put(foreign)
    +
    +    results = registry.drain_notifications(session_key="session-a")
    +
    +    assert [event["session_id"] for event, _ in results] == ["proc_owned"]
    +    assert registry.completion_queue.get_nowait() == foreign
    +    assert registry.completion_queue.empty()
    +
    +
    +def test_drain_notifications_session_key_filter_requeues_origin_only_event(registry):
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_origin_only",
    +        "origin_ui_session_id": "ui-session-a",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "done",
    +    }
    +    registry.completion_queue.put(event)
    +
    +    results = registry.drain_notifications(session_key="session-a")
    +
    +    assert results == []
    +    assert registry.completion_queue.get_nowait() == event
    +    assert registry.completion_queue.empty()
    +
    +
    +def test_drain_notifications_ownerless_completion_preserves_legacy_delivery(registry):
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_ownerless",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "ownerless",
    +    }
    +    registry.completion_queue.put(event)
    +
    +    results = registry.drain_notifications(
    +        session_key="session-a",
    +        owns_event=lambda _event: False,
    +    )
    +
    +    assert [raw for raw, _ in results] == [event]
    +    assert registry.completion_queue.empty()
    +
    +
    +def test_drain_notifications_ownerless_async_delegation_still_requires_proof(registry):
    +    event = {
    +        "type": "async_delegation",
    +        "delegation_id": "deleg_ownerless",
    +        "goal": "task",
    +        "status": "completed",
    +        "summary": "done",
    +        "api_calls": 1,
    +        "duration_seconds": 0.1,
    +    }
    +    registry.completion_queue.put(event)
    +
    +    results = registry.drain_notifications(
    +        session_key="session-a",
    +        owns_event=lambda _event: False,
    +    )
    +
    +    assert results == []
    +    assert registry.completion_queue.get_nowait() == event
    +    assert registry.completion_queue.empty()
    +
    +
    +def test_drain_notifications_completion_callback_exception_fails_closed(registry):
    +    event = {
    +        "type": "completion",
    +        "session_id": "proc_callback_error",
    +        "session_key": "session-a",
    +        "command": "safe-test-command",
    +        "exit_code": 0,
    +        "output": "done",
    +    }
    +    registry.completion_queue.put(event)
    +
    +    def broken(_event):
    +        raise RuntimeError("ownership check exploded")
    +
    +    results = registry.drain_notifications(
    +        session_key="session-a",
    +        owns_event=broken,
    +    )
    +
    +    assert results == []
    +    assert registry.completion_queue.get_nowait() == event
    +    assert registry.completion_queue.empty()
    +
    +
     def test_drain_notifications_filters_async_delegation_by_session_key():
         """Async-delegation events should only be consumed by the matching session's drain.
     

  • tools/process_registry.py

    @@ -1146,8 +1146,10 @@ def is_session_waiting(self, session_id: str) -> bool:
                     return False
             return True
     
    -    def _drain_should_skip(self, session_id: str) -> bool:
    -        """Whether the CLI drain should skip a completion event for this session.
    +    def _drain_should_skip(
    +        self, session_id: str, *, skip_poll_observed: bool = True
    +    ) -> bool:
    +        """Whether this drain should skip a completion event for this session.
     
             Skips when the agent has either truly consumed the output (wait/log →
             ``_completion_consumed``) or observed the exit inline via poll()
    @@ -1157,32 +1159,45 @@ def _drain_should_skip(self, session_id: str) -> bool:
             check only ``is_completion_consumed`` so a read-only poll never
             suppresses their autonomous delivery turn (#10156).
             """
    -        return session_id in self._completion_consumed or session_id in self._poll_observed
    +        return session_id in self._completion_consumed or (
    +            skip_poll_observed and session_id in self._poll_observed
    +        )
     
         def drain_notifications(
    -        self, session_key: str = "", owns_event=None,
    +        self,
    +        session_key: str = "",
    +        owns_event=None,
    +        *,
    +        skip_poll_observed: bool = True,
         ) -> "list[tuple[dict, str]]":
             """Pop all pending notification events and return formatted pairs.
     
             Returns a list of (raw_event, formatted_text) tuples.
             Skips completion events the agent already consumed via wait/log or
    -        observed inline via poll() (see ``_drain_should_skip``).
    +        observed inline via poll() (see ``_drain_should_skip``). Gateway/TUI
    +        callers pass ``skip_poll_observed=False`` because read-only polling must
    +        not suppress autonomous delivery there.
     
    -        Async-delegation events carry a conversation payload, so draining one
    -        into the wrong session is a cross-chat leak (#58684, #55578). Two
    -        filter modes, strongest wins:
    +        When a routing filter is supplied, addressed notifications must not be
    +        drained into the wrong session. Async-delegation events always require
    +        conversation payload; ordinary notifications require routing when they
    +        carry ``session_key`` or ``origin_ui_session_id`` metadata. Two filter
    +        modes are supported, strongest first:
     
             - ``owns_event(evt) -> bool``: positive-proof ownership callback.
    -          When provided, an async-delegation event is consumed ONLY if the
    -          callback returns True; everything else is re-queued for its owner.
    +          When provided, a routed event is consumed ONLY if the callback
    +          returns True; everything else is re-queued for its owner.
               The TUI passes its compression-chain-aware ownership check here so
               a post-compression session still claims its own pre-compression
               dispatches.
             - ``session_key``: plain key equality (CLI and other single-session
    -          callers). Non-matching async-delegation events are re-queued.
    +          callers). Non-matching addressed events are re-queued.
     
             With neither set, all events are consumed (legacy single-session
    -        behavior, backward compatible).
    +        behavior, backward compatible). Ownerless ordinary notifications also
    +        retain that legacy behavior even when a filter is provided. When a
    +        filter is provided, ownerless async-delegation events remain
    +        fail-closed and require positive proof.
             """
             results: "list[tuple[dict, str]]" = []
             requeue: "list[dict]" = []
    @@ -1191,39 +1206,43 @@ def drain_notifications(
                     evt = self.completion_queue.get_nowait()
                 except Exception:
                     break
    -            _evt_sid = evt.get("session_id", "")
    -            if evt.get("type") == "completion" and self._drain_should_skip(_evt_sid):
    -                continue
    -            # Filter async-delegation events so they are not delivered to the
    -            # wrong session/thread (#58684). Positive-proof callback beats
    -            # bare key equality when the caller can provide one.
    -            if evt.get("type") == "async_delegation":
    -                if owns_event is not None:
    -                    try:
    -                        owned = bool(owns_event(evt))
    -                    except Exception:
    -                        owned = False  # fail closed — never leak on a broken check
    -                    if not owned:
    -                        requeue.append(evt)
    -                        continue
    -                elif session_key:
    -                    evt_session_key = evt.get("session_key", "") or ""
    -                    if evt_session_key != session_key:
    -                        requeue.append(evt)
    -                        continue
    -                elif evt.get("restored"):
    -                    # Legacy unfiltered drain (no ownership callback, no
    -                    # session key). That behavior was safe when the in-memory
    -                    # queue could only hold events created by this very
    -                    # process — but durable restore (#63494) re-enqueues
    -                    # completions from PREVIOUS processes at startup, so an
    -                    # unfiltered consumer here would adopt a dead, unrelated
    -                    # session's conversation payload (#64484). Fail closed:
    -                    # leave restored events queued (still 'pending' on disk)
    -                    # for a consumer that can positively prove ownership,
    -                    # e.g. the owning session's --resume.
    +            # Positive-proof ownership beats bare key equality. Delegation
    +            # payloads always require proof; ordinary events require it once
    +            # they carry routing metadata. Ownerless ordinary events preserve
    +            # legacy single-session delivery.
    +            is_async_delegation = evt.get("type") == "async_delegation"
    +            evt_session_key = str(evt.get("session_key") or "")
    +            evt_origin_sid = str(evt.get("origin_ui_session_id") or "")
    +            requires_positive_proof = is_async_delegation or bool(
    +                evt_session_key or evt_origin_sid
    +            )
    +            if owns_event is not None and requires_positive_proof:
    +                try:
    +                    owned = bool(owns_event(evt))
    +                except Exception:
    +                    owned = False  # fail closed — never leak on a broken check
    +                if not owned:
    +                    requeue.append(evt)
    +                    continue
    +            elif session_key and requires_positive_proof:
    +                if evt_session_key != session_key:
                         requeue.append(evt)
                         continue
    +            elif is_async_delegation and evt.get("restored"):
    +                # Durable restore can enqueue previous-process payloads into a
    +                # fresh registry. An unfiltered legacy drain cannot prove
    +                # ownership, so leave those events queued for the owner.
    +                requeue.append(evt)
    +                continue
    +            # Local consumed/observed state may suppress only events this
    +            # session owns (or legacy ownerless ordinary events). Routing must
    +            # happen first so a foreign session cannot drop the owner's event.
    +            _evt_sid = evt.get("session_id", "")
    +            if evt.get("type") == "completion" and self._drain_should_skip(
    +                _evt_sid, skip_poll_observed=skip_poll_observed
    +            ):
    +                continue
    +
                 text = format_process_notification(evt)
                 if text:
                     results.append((evt, text))

  • tui_gateway/server.py

    @@ -8652,9 +8652,9 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
         minus its orphan-adoption fallback. An event owns-matches when its
         ``origin_ui_session_id`` is this live session, or its ``session_key``
         (raw or resolved through the compression chain) matches this session's
    -    key/lineage. Used as a fail-closed gate for async-delegation payloads:
    -    "not provably elsewhere" is NOT good enough to inject a conversation
    -    payload into this chat (#55578).
    +    key/lineage. Used as the fail-closed gate for every addressed notification:
    +    "not provably elsewhere" is NOT good enough to inject a payload into this
    +    chat (#55578).
         """
         if session.get("_finalized"):
             return False
    @@ -8679,6 +8679,14 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
         return resolved_key in current_keys
     
     
    +def _notification_event_requires_owner(evt: dict) -> bool:
    +    """Whether ``evt`` must be positively claimed before TUI delivery."""
    +    return evt.get("type") == "async_delegation" or bool(
    +        str(evt.get("origin_ui_session_id") or "")
    +        or str(evt.get("session_key") or "")
    +    )
    +
    +
     def _notification_event_dedup_key(evt: dict) -> tuple:
         """Return the UI-emission identity for a process notification event.
     
    @@ -8725,10 +8733,9 @@ def _notification_poller_loop(
         status.update (kind=process) for user visibility, then chains an
         agent turn via _run_prompt_submit if the session is idle.
     
    -    NOTE: The completion_queue is global (one per process). If multiple
    -    TUI sessions coexist, whichever poller wakes first grabs the event,
    -    even if the process was started by a different session. This matches
    -    CLI/gateway behavior (single session per process).
    +    The completion_queue is process-global. In multi-session Desktop each
    +    poller requeues events owned by another live session and drops addressed
    +    events whose owner is gone; ownerless legacy notifications remain global.
         """
         from tools.process_registry import process_registry, format_process_notification
     
    @@ -8749,26 +8756,22 @@ def _notification_poller_loop(
                 time.sleep(0.1)
                 continue
     
    -        # Fail closed for async-delegation results (#55578): these carry a
    -        # conversation payload, and injecting one into any chat other than the
    -        # one that commissioned it is a hard cross-session leak. The
    -        # belongs-elsewhere check above already re-queued events owned by
    -        # another LIVE session; what reaches here is either ours or an
    -        # orphan whose owner is gone. Orphaned delegation payloads are
    -        # DROPPED, not adopted — the subagent's summary is already persisted
    -        # in the delegation records/output store, so nothing is lost, whereas
    -        # a wrong-chat injection is unrecoverable. Non-delegation events
    -        # (background process completions etc.) keep the historical
    -        # adopt-orphans behavior.
    -        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
    -            sid, session, evt
    -        ):
    -            logger.warning(
    -                "async-delegation completion %s has no live owner "
    -                "(origin=%r key=%r); dropping from injection instead of "
    -                "delivering to session %s (#55578 fail-closed; result "
    -                "remains in the delegation records)",
    -                evt.get("delegation_id", "?"),
    +        # What reaches here is not owned by another LIVE session. Addressed
    +        # events still require positive proof before injection: exact UI origin,
    +        # direct durable key, or compression lineage. If none proves ownership,
    +        # the event is orphaned and must not be adopted by this chat. Truly
    +        # ownerless ordinary notifications retain legacy global delivery.
    +        requires_owner = _notification_event_requires_owner(evt)
    +        if requires_owner and not _session_owns_notification_event(sid, session, evt):
    +            log = (
    +                logger.warning
    +                if evt.get("type") == "async_delegation"
    +                else logger.debug
    +            )
    +            log(
    +                "Dropping unowned %s notification (origin=%r key=%r) instead "
    +                "of delivering to session %s",
    +                evt.get("type", "completion"),
                     str(evt.get("origin_ui_session_id") or ""),
                     str(evt.get("session_key") or ""),
                     sid,
    @@ -8830,6 +8833,7 @@ def _notification_poller_loop(
         # Drain any remaining events after stop signal (process all pending
         # before exiting so nothing is lost on shutdown). Events owned by other
         # live sessions are set aside and re-queued so their poller still sees them.
    +    # Orphaned events (owner gone) are dropped — same guard as the main loop.
         deferred: list = []
         while not process_registry.completion_queue.empty():
             try:
    @@ -8839,14 +8843,21 @@ def _notification_poller_loop(
             if _notification_event_belongs_elsewhere(sid, session, evt):
                 deferred.append(evt)
                 continue
    -        # Same fail-closed rule as the live loop: an orphaned async-delegation
    -        # payload is never adopted by a foreign session — defer it (a later
    -        # resume of the owner's lineage can still claim it) rather than
    -        # injecting another chat's conversation here (#55578).
    -        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
    -            sid, session, evt
    -        ):
    -            deferred.append(evt)
    +        # Same positive-proof rule as the live loop. Preserve the existing
    +        # shutdown behavior for orphaned delegation payloads by deferring them
    +        # for a later resume; ordinary addressed orphans are dropped.
    +        requires_owner = _notification_event_requires_owner(evt)
    +        if requires_owner and not _session_owns_notification_event(sid, session, evt):
    +            if evt.get("type") == "async_delegation":
    +                deferred.append(evt)
    +            else:
    +                logger.debug(
    +                    "Dropping unowned %s notification during shutdown drain "
    +                    "(origin=%r key=%r)",
    +                    evt.get("type", "completion"),
    +                    str(evt.get("origin_ui_session_id") or ""),
    +                    str(evt.get("session_key") or ""),
    +                )
                 continue
             _evt_sid = evt.get("session_id", "")
             if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
    @@ -9411,21 +9422,29 @@ def _stream(delta):
             # Drain completion notifications that arrived during this turn.
             # The background poller handles between-turn delivery; this is
             # the safety net for events that arrived mid-turn.
    +        #
    +        # Ownership filter (#42674, #35652): a turn finishing in session B
    +        # must not consume an event that belongs to session A. The registry
    +        # requeues every addressed event this session cannot positively claim;
    +        # the poller then delivers it to a live owner or drops an orphan.
             try:
                 from tools.process_registry import process_registry
     
                 # Positive-proof ownership (compression-chain aware) — the same
                 # fail-closed gate the poller uses, so the post-turn drain can't
    -            # adopt another session's (or an orphan's) delegation payload,
    -            # while a post-compression session still claims its own
    -            # pre-compression dispatches (#55578).
    -            for _evt, synth in process_registry.drain_notifications(
    +            # adopt another session's addressed notification while a
    +            # post-compression session still claims its own pre-compression
    +            # dispatches (#55578).
    +            drained = process_registry.drain_notifications(
                     session_key=session.get("session_key", ""),
                     owns_event=lambda e: _session_owns_notification_event(sid, session, e),
    -            ):
    +                skip_poll_observed=False,
    +            )
    +            for index, (_evt, synth) in enumerate(drained):
                     with session["history_lock"]:
                         if session.get("running"):
    -                        process_registry.completion_queue.put(_evt)
    +                        for pending_evt, _pending_synth in drained[index:]:
    +                            process_registry.completion_queue.put(pending_evt)
                             break
                         session["running"] = True
                     from tools.async_delegation import (