From c1f3e21653fac327c5d4e066e8d7b898e1696188 Mon Sep 17 00:00:00 2001 From: Ayush Nangia Date: Tue, 28 Jul 2026 18:25:23 +0530 Subject: [PATCH 1/2] fix(gateway): route plain background live output to its spawn-time UI owner (#61719 residual) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged 54d0948d3 landed owner-routing for post-turn completions, but the sweeper's flagged gap on #61719 survived: terminal_tool captured routing metadata only when notify_on_complete/watch_patterns were set, while live agent.terminal.output chunks are emitted for EVERY background process. _owner_sid_for_process matched only by session_key, so a delegated child's process (subagent-internal key, never a live TUI session) emitted its live output with sid "" — dropped by write_json. - ProcessSession.origin_ui_session_id: spawn-time UI owner, persisted in the crash checkpoint like the watcher metadata - terminal_tool: capture HERMES_UI_SESSION_ID for every background spawn (same session-env pattern as desktop_ui.py:39 / delegate_tool.py:3191) - _owner_sid_for_process: prefer the recorded origin while that window is live; legacy session_key equality remains the fallback Salvages the remaining scope of #61719 (bot-authored, sweeper problem unaddressed). --- tests/tools/test_notify_on_complete.py | 50 +++++++++++++++++ tests/tui_gateway/test_tui_gateway_server.py | 57 +++++++++++++++++++- tools/process_registry.py | 5 +- tools/terminal_tool_background.py | 9 ++++ tui_gateway/session_notifications.py | 5 ++ 5 files changed, 123 insertions(+), 3 deletions(-) diff --git a/tests/tools/test_notify_on_complete.py b/tests/tools/test_notify_on_complete.py index 27817cc4bf5e4..af542a6a410d6 100644 --- a/tests/tools/test_notify_on_complete.py +++ b/tests/tools/test_notify_on_complete.py @@ -441,3 +441,53 @@ 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.""" + 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): + proc = 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, + ) + spawned["proc"] = proc + return proc + + 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 getattr(spawned["proc"], "origin_ui_session_id", "") == "sid_ui_1", ( + "plain background spawn (no notify/watch) must record the spawn-time " + "UI owner so live output chunks can be positively routed" + ) diff --git a/tests/tui_gateway/test_tui_gateway_server.py b/tests/tui_gateway/test_tui_gateway_server.py index cf41edb3c29ec..f464a97c0ce18 100644 --- a/tests/tui_gateway/test_tui_gateway_server.py +++ b/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: `_owner_sid_for_process` 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_agent_terminal_output() + + 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 diff --git a/tools/process_registry.py b/tools/process_registry.py index 2b0fd982092bc..2ff78e96a51b7 100644 --- a/tools/process_registry.py +++ b/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() diff --git a/tools/terminal_tool_background.py b/tools/terminal_tool_background.py index eb70a1fc6c6cd..bed8179143106 100644 --- a/tools/terminal_tool_background.py +++ b/tools/terminal_tool_background.py @@ -163,6 +163,15 @@ def spawn_background_process( effective_task_id=effective_task_id, task_id=task_id, session_key=session_key, effective_pty=effective_pty, ) + # Capture the UI owner for every background process, not only those + # requesting a completion notification. Delegated children use their + # own internal session_key, but their live output still belongs to the + # TUI/Desktop window that started the parent turn. + try: + from gateway.session_context import get_session_env + proc_session.origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") or "" + except Exception: + pass # best effort; routing retains the session_key fallback result_data = {"output": "Background process started", "session_id": proc_session.id, "pid": proc_session.pid, "exit_code": 0, "error": None} if approval_note: diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 3ba919d9fda99..1daf350992aff 100644 --- a/tui_gateway/session_notifications.py +++ b/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 "" From 76fc9c69e629c4d544c87e2c06d1a3081a936b03 Mon Sep 17 00:00:00 2001 From: Ayush Nangia Date: Thu, 30 Jul 2026 21:34:06 +0530 Subject: [PATCH 2/2] fix(gateway): set UI owner at ProcessSession construction, before the reader starts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sweeper review: the post-spawn assignment races the local reader thread — spawn_local() starts it before returning, and _emit_output() can forward the first chunks while origin_ui_session_id is still empty. - spawn_local()/spawn_via_env() accept origin_ui_session_id and set it on ProcessSession at construction, before any reader/poller exists - terminal_tool captures HERMES_UI_SESSION_ID before the spawn call and passes it in; the post-spawn assignment is removed - capture test now asserts the kwarg is handed to the spawn (red on the post-spawn variant); new real-registry-lifecycle regression proves the first emitted chunk of a fast process already carries the owner --- tests/tools/test_notify_on_complete.py | 54 +++++++++++++++++--- tests/tools/test_terminal_task_cwd.py | 2 + tests/tui_gateway/test_tui_gateway_server.py | 4 +- tools/process_registry.py | 18 +++++-- tools/terminal_tool_background.py | 21 ++++---- 5 files changed, 74 insertions(+), 25 deletions(-) diff --git a/tests/tools/test_notify_on_complete.py b/tests/tools/test_notify_on_complete.py index af542a6a410d6..165270a28f715 100644 --- a/tests/tools/test_notify_on_complete.py +++ b/tests/tools/test_notify_on_complete.py @@ -448,7 +448,11 @@ def test_background_spawn_captures_ui_origin_without_notify(monkeypatch, tmp_pat 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.""" + 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) @@ -458,7 +462,8 @@ def test_background_spawn_captures_ui_origin_without_notify(monkeypatch, tmp_pat spawned = {} def capturing_spawn_local(**kwargs): - proc = SimpleNamespace( + spawned["kwargs"] = kwargs + return SimpleNamespace( id="proc_ui_origin_test", pid=4243, notify_on_complete=False, @@ -470,8 +475,6 @@ def capturing_spawn_local(**kwargs): watcher_message_id="", watcher_interval=0, ) - spawned["proc"] = proc - return proc monkeypatch.setattr( process_registry_module.process_registry, "spawn_local", capturing_spawn_local @@ -487,7 +490,44 @@ def capturing_spawn_local(**kwargs): tt._active_environments.pop("default", None) tt._last_activity.pop("default", None) - assert getattr(spawned["proc"], "origin_ui_session_id", "") == "sid_ui_1", ( - "plain background spawn (no notify/watch) must record the spawn-time " - "UI owner so live output chunks can be positively routed" + 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}" ) diff --git a/tests/tools/test_terminal_task_cwd.py b/tests/tools/test_terminal_task_cwd.py index 95ed40f5ed0dd..cba637d45d670 100644 --- a/tests/tools/test_terminal_task_cwd.py +++ b/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": "", }] diff --git a/tests/tui_gateway/test_tui_gateway_server.py b/tests/tui_gateway/test_tui_gateway_server.py index f464a97c0ce18..b5f4c54d54afb 100644 --- a/tests/tui_gateway/test_tui_gateway_server.py +++ b/tests/tui_gateway/test_tui_gateway_server.py @@ -22619,7 +22619,7 @@ def test_load_cfg_raw_sees_replacement_with_pinned_mtime_and_size(monkeypatch, t assert server._load_cfg_raw()["model"]["default"] == "aaaa-route" def test_agent_terminal_output_routes_by_spawn_time_ui_owner(monkeypatch): - """#61719 residual: `_owner_sid_for_process` matched only by session_key. + """#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 @@ -22635,7 +22635,7 @@ def test_agent_terminal_output_routes_by_spawn_time_ui_owner(monkeypatch): monkeypatch.setattr( server, "_emit", lambda event, sid, payload=None: emitted.append((event, sid, payload)) ) - server._wire_agent_terminal_output() + server._wire_desktop_sinks() saved = dict(server._sessions) server._sessions.clear() diff --git a/tools/process_registry.py b/tools/process_registry.py index 2ff78e96a51b7..104783706965f 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -972,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.""" @@ -983,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: @@ -1070,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 diff --git a/tools/terminal_tool_background.py b/tools/terminal_tool_background.py index bed8179143106..488e101d259dc 100644 --- a/tools/terminal_tool_background.py +++ b/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,20 +159,16 @@ 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, ) - # Capture the UI owner for every background process, not only those - # requesting a completion notification. Delegated children use their - # own internal session_key, but their live output still belongs to the - # TUI/Desktop window that started the parent turn. - try: - from gateway.session_context import get_session_env - proc_session.origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") or "" - except Exception: - pass # best effort; routing retains the session_key fallback result_data = {"output": "Background process started", "session_id": proc_session.id, "pid": proc_session.pid, "exit_code": 0, "error": None} if approval_note: