fix(gateway): route plain background live output to its spawn-time UI owner (#61719 residual)

tests/tools/test_notify_on_complete.py

@@ -441,3 +441,93 @@ def test_non_ci_background_command_does_not_emit_homebrew_hint(monkeypatch, tmp_
     assert "hint" not in result, (
         f"Non-CI command using awk must not be flagged as homebrew CI poller, got: {result.get('hint')!r}"
     )
+
+
+def test_background_spawn_captures_ui_origin_without_notify(monkeypatch, tmp_path):
+    """#61719 residual: live `agent.terminal.output` chunks are emitted for
+    EVERY background process, but the spawn-time UI owner was only captured
+    when notify_on_complete/watch_patterns were set — a plain background=true
+    spawn (a delegated child's, in particular) had no positive owner and its
+    live output was dropped by the desktop router.
+
+    The owner must be passed INTO spawn_local() (sweeper review): the local
+    reader thread starts inside the spawn and can emit output before it
+    returns, so a post-spawn attribute assignment races the first chunks."""
+    import gateway.session_context as session_context
+
+    tt = _silent_bg_harness(monkeypatch, tmp_path)
+    from types import SimpleNamespace
+    from tools import process_registry as process_registry_module
+
+    spawned = {}
+
+    def capturing_spawn_local(**kwargs):
+        spawned["kwargs"] = kwargs
+        return SimpleNamespace(
+            id="proc_ui_origin_test",
+            pid=4243,
+            notify_on_complete=False,
+            watcher_platform="",
+            watcher_chat_id="",
+            watcher_user_id="",
+            watcher_user_name="",
+            watcher_thread_id="",
+            watcher_message_id="",
+            watcher_interval=0,
+        )
+
+    monkeypatch.setattr(
+        process_registry_module.process_registry, "spawn_local", capturing_spawn_local
+    )
+    monkeypatch.setattr(
+        session_context,
+        "get_session_env",
+        lambda key, default="": "sid_ui_1" if key == "HERMES_UI_SESSION_ID" else default,
+    )
+    try:
+        tt.terminal_tool(command="sleep 60", background=True)
+    finally:
+        tt._active_environments.pop("default", None)
+        tt._last_activity.pop("default", None)
+
+    assert spawned["kwargs"].get("origin_ui_session_id") == "sid_ui_1", (
+        "plain background spawn (no notify/watch) must hand the spawn-time UI "
+        "owner to spawn_local() so the session owns it before the reader starts"
+    )
+
+
+def test_immediate_output_carries_ui_owner_through_real_registry(tmp_path):
+    """Sweeper ask: real registry lifecycle, not a stubbed spawn. The very
+    first chunk a fast process emits must already see the UI owner on the
+    session — spawn_local() sets it at ProcessSession construction, before
+    the reader thread exists."""
+    from tools.process_registry import process_registry
+
+    recorded = []
+    prev_sink = process_registry.on_output
+    process_registry.on_output = lambda session, chunk: recorded.append(
+        (getattr(session, "origin_ui_session_id", ""), chunk)
+    )
+    try:
+        session = process_registry.spawn_local(
+            command="echo immediate-owner-check",
+            cwd=str(tmp_path),
+            origin_ui_session_id="sid_live_1",
+        )
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline and not any(
+            "immediate-owner-check" in chunk for _, chunk in recorded
+        ):
+            time.sleep(0.05)
+    finally:
+        process_registry.on_output = prev_sink
+        try:
+            process_registry.kill_process(session.id)
+        except Exception:
+            pass
+
+    matching = [sid for sid, chunk in recorded if "immediate-owner-check" in chunk]
+    assert matching, f"no live output observed; recorded={recorded!r}"
+    assert all(sid == "sid_live_1" for sid in matching), (
+        f"immediate output emitted without its UI owner: {matching}"
+    )

tests/tools/test_terminal_task_cwd.py

@@ -173,6 +173,8 @@ def spawn_local(self, **kwargs):
         "session_key": task_id,
         "env_vars": {},
         "use_pty": False,
+        # Spawn-time UI owner (#61719) — empty outside TUI/desktop contexts.
+        "origin_ui_session_id": "",
     }]
 
 

tests/tui_gateway/test_tui_gateway_server.py

@@ -22599,8 +22599,6 @@ def _fake_db(_params, *, writer=False):
     assert captured["row_update"] == (target, str(new_cwd))
     assert live["cwd"] == str(new_cwd)
     assert live.get("explicit_cwd") is True
-
-
 def test_load_cfg_raw_sees_replacement_with_pinned_mtime_and_size(monkeypatch, tmp_path):
     """#111105: the raw-config cache must not serve (and later write back) a stale document after a
     same-size replacement that keeps the old mtime."""
@@ -22619,3 +22617,58 @@ def test_load_cfg_raw_sees_replacement_with_pinned_mtime_and_size(monkeypatch, t
     shutil.copy2(other, cfg)
     os.utime(cfg, ns=(st.st_atime_ns, st.st_mtime_ns))
     assert server._load_cfg_raw()["model"]["default"] == "aaaa-route"
+
+def test_agent_terminal_output_routes_by_spawn_time_ui_owner(monkeypatch):
+    """#61719 residual: the desktop sink matched only by session_key.
+    A delegated child's process carries the subagent's internal key, which
+    never matches a live TUI session, so its live `agent.terminal.output`
+    chunks were emitted with sid "" and dropped by write_json. The spawn-time
+    `origin_ui_session_id` (captured by terminal_tool for every background
+    spawn) must win when it names a live session; key-equality remains the
+    fallback for processes without a recorded UI origin."""
+    from types import SimpleNamespace
+    from tools.process_registry import process_registry
+
+    monkeypatch.setattr(process_registry, "on_output", None)
+    monkeypatch.setattr(process_registry, "on_close", None)
+    emitted = []
+    monkeypatch.setattr(
+        server, "_emit", lambda event, sid, payload=None: emitted.append((event, sid, payload))
+    )
+    server._wire_desktop_sinks()
+
+    saved = dict(server._sessions)
+    server._sessions.clear()
+    try:
+        server._sessions["parent_sid"] = {"session_key": "parent-key"}
+
+        # Delegated-child shape: internal session_key, recorded UI origin.
+        child_proc = SimpleNamespace(
+            id="proc_child", session_key="subagent-internal-key",
+            origin_ui_session_id="parent_sid",
+        )
+        process_registry.on_output(child_proc, "hello from child")
+        assert emitted[-1][0] == "agent.terminal.output"
+        assert emitted[-1][1] == "parent_sid", (
+            "child process live output must route to the spawn-time UI owner"
+        )
+        assert emitted[-1][2] == {"process_id": "proc_child", "chunk": "hello from child"}
+
+        # No recorded origin: legacy session_key equality still routes.
+        plain_proc = SimpleNamespace(
+            id="proc_plain", session_key="parent-key", origin_ui_session_id="",
+        )
+        process_registry.on_output(plain_proc, "plain")
+        assert emitted[-1][1] == "parent_sid"
+
+        # Stale origin (window closed) with unknown key: falls back to "".
+        stale_proc = SimpleNamespace(
+            id="proc_stale", session_key="unknown-key", origin_ui_session_id="gone_sid",
+        )
+        process_registry.on_output(stale_proc, "stale")
+        assert emitted[-1][1] == ""
+    finally:
+        server._sessions.clear()
+        server._sessions.update(saved)
+        process_registry.on_output = None
+        process_registry.on_close = None

tools/process_registry.py

@@ -478,6 +478,9 @@ class ProcessSession:
     # Session-db id of the spawning conversation; lets the gateway drop completions whose
     # session was closed at a user boundary (/new) instead of injecting into the NEW one.
     parent_session_id: str = ""
+    # Spawn-time TUI/Desktop window that commissioned the process. Unlike
+    # session_key, this remains the owning UI when a delegated child spawns it.
+    origin_ui_session_id: str = ""
     notify_on_complete: bool = False            # Queue agent notification on exit
     watch_patterns: List[str] = field(default_factory=list)
     _watch_hits: int = field(default=0, repr=False)          # total matches delivered
@@ -518,7 +521,7 @@ def mark_exited(self, exit_code, reason: str = "exited", source: str = "") -> No
     "command", "pid", "pid_scope", "host_start_time", "systemd_unit", "cwd",
     "started_at", "task_id", "owner_task_id", "session_key",
     *(f"watcher_{k}" for k in _WATCHER_ROUTE_KEYS), "watcher_interval",
-    "parent_session_id", "notify_on_complete", "watch_patterns")
+    "parent_session_id", "origin_ui_session_id", "notify_on_complete", "watch_patterns")
 _CHECKPOINT_DEFAULTS = {
     f.name: ([] if f.name == "watch_patterns" else f.default)
     for f in ProcessSession.__dataclass_fields__.values()
@@ -969,7 +972,8 @@ def _spawn_local_pty(self, session: ProcessSession, safe_command: str, env_vars:
 
     def spawn_local(
         self, command: str, cwd: str = None, task_id: str = "", session_key: str = "",
-        env_vars: dict = None, use_pty: bool = False, owner_task_id: str = "") -> ProcessSession:
+        env_vars: dict = None, use_pty: bool = False, owner_task_id: str = "",
+        origin_ui_session_id: str = "") -> ProcessSession:
         """Spawn a background process locally (TERMINAL_ENV=local; other backends use
         spawn_via_env()). ``use_pty`` requests a pseudo-terminal via ptyprocess/pywinpty
         for interactive CLIs, falling back to a plain pipe when unavailable or failing."""
@@ -980,7 +984,11 @@ def spawn_local(
         from tools.terminal_tool_sudo import _rewrite_compound_background as _rewrite_bg
 
         safe_command = _rewrite_bg(command)
-        session = self._new_session(command, task_id, owner_task_id, session_key, _resolve_safe_cwd(cwd or os.getcwd()))
+        session = self._new_session(
+            command, task_id, owner_task_id, session_key,
+            _resolve_safe_cwd(cwd or os.getcwd()),
+            origin_ui_session_id=origin_ui_session_id,
+        )
         pty_scope_attempted = False
         if use_pty:
             try:
@@ -1067,12 +1075,17 @@ def adopt_local(
 
     def spawn_via_env(
         self, env: Any, command: str, cwd: str = None, task_id: str = "", session_key: str = "",
-        timeout: int = 10, owner_task_id: str = "") -> ProcessSession:
+        timeout: int = 10, owner_task_id: str = "",
+        origin_ui_session_id: str = "") -> ProcessSession:
         """Spawn a background process inside a non-local backend's sandbox.
         The command is wrapped to capture its in-sandbox PID and redirect output to a
         log file that later execute() calls poll. No live pipe or stdin, but it runs in
         the correct sandbox context."""
-        session = self._new_session(command, task_id, owner_task_id, session_key, cwd, env_ref=env, pid_scope="sandbox")
+        session = self._new_session(
+            command, task_id, owner_task_id, session_key, cwd,
+            env_ref=env, pid_scope="sandbox",
+            origin_ui_session_id=origin_ui_session_id,
+        )
         temp_dir = self._env_temp_dir(env)
         log_path, pid_path, exit_path = (f"{temp_dir}/hermes_bg_{session.id}.{ext}" for ext in ("log", "pid", "exit"))
         q = shlex.quote

tools/terminal_tool_background.py

@@ -87,9 +87,10 @@ def _stamp_gateway_routing(proc_session, get_session_env) -> None:
 
 
 def _spawn(process_registry, *, env, env_type, command, cwd, effective_task_id, task_id,
-           session_key, effective_pty):
+           session_key, effective_pty, origin_ui_session_id):
     common = dict(command=command, cwd=cwd, task_id=effective_task_id,
-                  owner_task_id=task_id or effective_task_id, session_key=session_key)
+                  owner_task_id=task_id or effective_task_id, session_key=session_key,
+                  origin_ui_session_id=origin_ui_session_id)
     if env_type == "local":
         return process_registry.spawn_local(
             env_vars=env.env if hasattr(env, 'env') else None, use_pty=effective_pty, **common)
@@ -158,10 +159,15 @@ def spawn_background_process(
         workdir=workdir, default_cwd=cwd, session_key=session_key, env_type=env_type,
     )
     try:
+        try:
+            from gateway.session_context import get_session_env
+            origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") or ""
+        except Exception:
+            origin_ui_session_id = ""
         proc_session = _spawn(
             process_registry, env=env, env_type=env_type, command=command, cwd=effective_cwd,
             effective_task_id=effective_task_id, task_id=task_id, session_key=session_key,
-            effective_pty=effective_pty,
+            effective_pty=effective_pty, origin_ui_session_id=origin_ui_session_id,
         )
         result_data = {"output": "Background process started", "session_id": proc_session.id,
                        "pid": proc_session.pid, "exit_code": 0, "error": None}

tui_gateway/session_notifications.py

@@ -711,6 +711,11 @@ def _wire_desktop_sinks() -> None:
 
     def _owner_sid(session) -> str:
         # session may be None (process already finished/pruned) — the tab can still linger and be closed.
+        origin_sid = str(getattr(session, "origin_ui_session_id", "") or "") if session is not None else ""
+        if origin_sid:
+            with _sessions_lock:
+                if origin_sid in _sessions:
+                    return origin_sid
         session_key = str(getattr(session, "session_key", "") or "") if session is not None else ""
         if not session_key:
             return ""