feat(desktop): live agent terminals + agent-driven tab close

Make the read-only agent terminal mirrors stream in real time and give
the agent a desktop-only way to dismiss its own tabs.

- Stream background output live: the local reader used a blocking
  read(4096) that buffered small periodic output until EOF, so agent
  tabs only "filled in" at process exit. Switch to buffer.read1(4096)
  (decoded) for incremental chunks.
- Route agent.terminal.output / terminal.close to the window that owns
  the process (its gateway session) instead of an empty session id, so
  events actually reach the desktop renderer.
- Add close_terminal: a HERMES_DESKTOP-gated tool (sibling of
  read_terminal) that drops a process's read-only tab WITHOUT killing it
  via process_registry.on_close; output keeps buffering and the user can
  reopen from the status stack.
- ⌘W now closes a focused agent tab: mark the agent instance
  data-terminal and focus it on activation so isFocusWithin routes there.
- ensureTerminal() no longer spawns an extra user shell when a tab
  already exists (e.g. opening a background task from the status stack).

apps/desktop/src/app/right-sidebar/terminal/instance.tsx

@@ -84,7 +84,12 @@ export function AgentTerminalInstance({ active, procId }: AgentTerminalInstanceP
   const { hostRef } = useAgentTerminal({ active, procId })
 
   return (
-    <div className={cn(INSTANCE_CLASS, active ? 'visible' : 'invisible pointer-events-none')}>
+    <div
+      className={cn(INSTANCE_CLASS, active ? 'visible' : 'invisible pointer-events-none')}
+      // Same focus-scope marker as the user terminal so isFocusWithin('[data-terminal]')
+      // routes ⌘W here and closes the focused agent tab (not a preview).
+      data-terminal=""
+    >
       <div
         className="h-full min-h-0 overflow-hidden text-(--ui-text-secondary) [&_.xterm]:h-full [&_.xterm-screen]:bg-(--ui-editor-surface-background)! [&_.xterm-viewport]:bg-(--ui-editor-surface-background)!"
         ref={hostRef}

apps/desktop/src/app/right-sidebar/terminal/rail.tsx

@@ -123,7 +123,10 @@ function TerminalRailItem({ active, canCloseOthers, index, term, toggleHint }: T
       <ContextMenuTrigger asChild>
         <li className="relative flex w-full justify-center [-webkit-app-region:no-drag]">
           {active && (
-            <span aria-hidden="true" className="absolute inset-y-0.5 right-0 w-0.5 rounded-l-sm bg-(--ui-stroke-primary)" />
+            <span
+              aria-hidden="true"
+              className="absolute inset-y-0.5 right-0 w-0.5 rounded-l-sm bg-(--ui-stroke-primary)"
+            />
           )}
           <Tip label={hintLabel(label, toggleHint)} side="left">
             <button
@@ -152,7 +155,7 @@ function TerminalRailItem({ active, canCloseOthers, index, term, toggleHint }: T
             >
               <Codicon
                 className={cn(term.kind === 'agent' && !active && 'text-primary')}
-                name={term.kind === 'agent' ? 'sparkle' : 'terminal'}
+                name={term.kind === 'agent' ? 'agent' : 'terminal'}
                 size="0.875rem"
               />
             </button>

apps/desktop/src/app/right-sidebar/terminal/terminals.ts

@@ -87,10 +87,11 @@ export function openAgentTerminal(procId: string, title: string): void {
   setTerminalTakeover(true)
 }
 
-/** Guarantee at least one user shell exists (called when the pane opens) — agent
- *  mirror tabs don't count, so opening the pane always yields a real shell. */
+/** Guarantee at least one tab exists when the pane opens.
+ *  If a status-stack click already opened an agent tab, don't create a
+ *  second, unrelated user shell just because the pane became visible. */
 export function ensureTerminal(): void {
-  if (!$terminals.get().some(term => term.kind === 'user')) {
+  if ($terminals.get().length === 0) {
     createTerminal()
   }
 }
@@ -136,6 +137,23 @@ export function closeTerminal(id: string): void {
   }
 }
 
+/** Close the read-only agent tab mirroring a background process. The agent
+ *  drives this via `process(action="close_terminal")` → `terminal.close`. The
+ *  process is NOT killed — only the view is dropped; `surfacedProcs` keeps it
+ *  from auto-resurfacing, and the status-stack row can reopen it on demand.
+ *  No-op when no such tab exists. */
+export function closeAgentTerminalByProc(procId: string): boolean {
+  const term = $terminals.get().find(t => t.kind === 'agent' && t.procId === procId)
+
+  if (!term) {
+    return false
+  }
+
+  closeTerminal(term.id)
+
+  return true
+}
+
 export function closeActiveTerminal(): void {
   const id = $activeTerminalId.get()
 

apps/desktop/src/app/right-sidebar/terminal/use-agent-terminal.ts

@@ -117,6 +117,10 @@ export function useAgentTerminal({ active, procId }: { active: boolean; procId:
       fitRef.current?.()
       webglRef.current?.clearTextureAtlas()
       term?.refresh(0, term.rows - 1)
+      // Take focus on activation (parity with the user terminal) so the active
+      // agent tab holds focus and ⌘W's isFocusWithin('[data-terminal]') routes
+      // the close to this tab rather than to a preview.
+      term?.focus()
     })
 
     return () => cancelAnimationFrame(frame)

apps/desktop/src/app/session/hooks/use-message-stream.ts

@@ -2,6 +2,7 @@ import type { QueryClient } from '@tanstack/react-query'
 import { type MutableRefObject, useCallback, useEffect, useRef } from 'react'
 
 import { writeAgentTerminalChunk } from '@/app/right-sidebar/terminal/agent-terminal-stream'
+import { closeAgentTerminalByProc } from '@/app/right-sidebar/terminal/terminals'
 import { readActiveTerminal } from '@/app/right-sidebar/terminal/buffer'
 import { translateNow } from '@/i18n'
 import {
@@ -1169,6 +1170,10 @@ export function useMessageStream({
       } else if (event.type === 'agent.terminal.output') {
         // Live chunk from a background process → its read-only agent terminal tab.
         writeAgentTerminalChunk(payload?.process_id ?? '', payload?.chunk ?? '')
+      } else if (event.type === 'terminal.close') {
+        // Agent closed its own read-only tab via process(action="close_terminal").
+        // The process is untouched — this only drops the view.
+        closeAgentTerminalByProc(payload?.process_id ?? '')
       } else if (event.type === 'status.update') {
         if (sessionId && payload?.kind === 'compacting') {
           setSessionCompacting(sessionId, true)

tests/tools/test_process_registry.py

@@ -141,6 +141,118 @@ class TestGetAndPoll:
         assert result["exit_code"] == 0
 
 
+def test_request_close_terminal_without_sink_is_desktop_only_error(registry):
+    """No UI close sink wired (CLI/messaging) → clear desktop-only error, no raise."""
+    s = _make_session(sid="proc_close_nosink")
+    registry._running[s.id] = s
+
+    result = registry.request_close_terminal(s.id)
+
+    assert result["status"] == "error"
+    assert "desktop" in result["error"].lower()
+
+
+def test_request_close_terminal_invokes_sink_without_killing(registry):
+    """With a sink wired, close routes (session, process_id) to the UI and leaves
+    the process running — close is a view drop, not a kill."""
+    s = _make_session(sid="proc_close_live")
+    registry._running[s.id] = s
+    calls = []
+    registry.on_close = lambda session, pid: calls.append((session, pid))
+
+    result = registry.request_close_terminal(s.id)
+
+    assert result["status"] == "ok"
+    assert result["closed"] == "proc_close_live"
+    assert calls == [(s, "proc_close_live")]
+    # Still tracked as running — closing the tab must not reap the process.
+    assert s.id in registry._running
+
+
+def test_close_terminal_tool_requires_process_id():
+    """The desktop-gated close_terminal tool rejects a missing process_id."""
+    from tools.close_terminal_tool import close_terminal_tool
+
+    assert json.loads(close_terminal_tool(""))["error"]
+
+
+def test_close_terminal_tool_routes_to_registry(monkeypatch):
+    """close_terminal delegates to process_registry.request_close_terminal."""
+    import tools.close_terminal_tool as ct
+
+    seen = {}
+
+    def _fake_close(sid):
+        seen["sid"] = sid
+
+        return {"status": "ok", "closed": sid}
+
+    monkeypatch.setattr(ct.process_registry, "request_close_terminal", _fake_close)
+
+    out = ct.close_terminal_tool("proc_abc")
+
+    assert json.loads(out)["closed"] == "proc_abc"
+    assert seen["sid"] == "proc_abc"
+
+
+def test_close_terminal_tool_gated_on_desktop(monkeypatch):
+    """Hidden unless HERMES_DESKTOP is set (mirrors read_terminal gating)."""
+    from tools.close_terminal_tool import check_close_terminal_requirements
+
+    monkeypatch.delenv("HERMES_DESKTOP", raising=False)
+    assert check_close_terminal_requirements() is False
+
+    monkeypatch.setenv("HERMES_DESKTOP", "1")
+    assert check_close_terminal_requirements() is True
+
+
+def test_reader_loop_streams_incremental_chunks_from_read1(registry, monkeypatch):
+    """Local reader must emit live chunks, not one EOF burst.
+
+    Regression for desktop agent terminals: ``stdout.read(4096)`` can buffer
+    until process exit for small periodic output. ``buffer.read1(4096)`` should
+    surface each chunk as it arrives.
+    """
+
+    class _FakeBuffer:
+        def __init__(self, chunks):
+            self._chunks = list(chunks)
+
+        def read1(self, _n):
+            if self._chunks:
+                return self._chunks.pop(0)
+            return b""
+
+    class _FakeStdout:
+        def __init__(self, chunks):
+            self.buffer = _FakeBuffer(chunks)
+
+    class _FakeProcess:
+        def __init__(self, chunks):
+            self.stdout = _FakeStdout(chunks)
+            self.returncode = 0
+
+        def wait(self, timeout=None):
+            return 0
+
+    session = _make_session(sid="proc_reader_live")
+    session.process = _FakeProcess([b"tick 1\n", b"tick 2\n", b"tick 3\n", b""])
+    emitted = []
+    moved = []
+
+    monkeypatch.setattr(registry, "_check_watch_patterns", lambda _s, _c: None)
+    monkeypatch.setattr(registry, "_emit_output", lambda _s, chunk: emitted.append(chunk))
+    monkeypatch.setattr(registry, "_move_to_finished", lambda _s: moved.append(_s.id))
+
+    registry._reader_loop(session)
+
+    assert emitted == ["tick 1\n", "tick 2\n", "tick 3\n"]
+    assert session.output_buffer == "tick 1\ntick 2\ntick 3\n"
+    assert session.exited is True
+    assert session.exit_code == 0
+    assert moved == ["proc_reader_live"]
+
+
 # =========================================================================
 # Orphaned-pipe reconciliation (issue #17327)
 # =========================================================================

tools/close_terminal_tool.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""Close a read-only agent terminal tab in the Hermes desktop GUI.
+
+Each ``terminal(background=true)`` process is mirrored as a read-only tab in the
+desktop's terminal pane. This tool lets the agent drop a tab it no longer needs
+to show — WITHOUT killing the process (use ``process(action='kill')`` for that).
+The output keeps buffering and the user can reopen the tab from the status stack.
+
+It routes through the process registry's ``on_close`` sink, which the desktop
+gateway wires to emit a ``terminal.close`` event the renderer handles. Like
+``read_terminal`` it is gated on ``HERMES_DESKTOP`` so it never appears outside
+the GUI.
+"""
+
+import json
+import os
+
+from tools.process_registry import process_registry
+from tools.registry import registry, tool_error
+
+
+def close_terminal_tool(process_id: str) -> str:
+    """Ask the desktop GUI to close a background process's read-only tab."""
+    pid = (process_id or "").strip()
+    if not pid:
+        return tool_error("process_id is required (the background process whose tab to close).")
+
+    return json.dumps(process_registry.request_close_terminal(pid), ensure_ascii=False)
+
+
+def check_close_terminal_requirements() -> bool:
+    """Desktop GUI only — HERMES_DESKTOP is set on the gateway the app spawns."""
+    return (os.getenv("HERMES_DESKTOP") or "").strip().lower() in ("1", "true", "yes")
+
+
+CLOSE_TERMINAL_SCHEMA = {
+    "name": "close_terminal",
+    "description": (
+        "Close the read-only terminal tab for one of your background processes in "
+        "the Hermes desktop GUI (the tabs mirroring terminal(background=true) runs). "
+        "This does NOT kill the process — it only drops the tab/view; the output "
+        "keeps buffering and the user can reopen it from the status stack. Use it "
+        "to tidy up when a background process's live terminal is no longer worth "
+        "showing. To actually stop the process, use process(action='kill') instead."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "process_id": {
+                "type": "string",
+                "description": (
+                    "The background process's session id (from terminal(background=true) "
+                    "output or process(action='list')) whose tab should be closed."
+                ),
+            },
+        },
+        "required": ["process_id"],
+    },
+}
+
+
+registry.register(
+    name="close_terminal",
+    toolset="terminal",
+    schema=CLOSE_TERMINAL_SCHEMA,
+    handler=lambda args, **kw: close_terminal_tool(process_id=args.get("process_id", "")),
+    check_fn=check_close_terminal_requirements,
+    emoji="🖥️",
+)

tools/process_registry.py

@@ -200,6 +200,11 @@ class ProcessRegistry:
         # reader threads with (session, chunk) to stream output to a UI in
         # real time, instead of polling the output tail.
         self.on_output = None
+        # Close-view sink set by a driver (desktop gateway): called with
+        # (session_or_none, process_id) when the agent asks to close a read-only
+        # terminal tab. Distinct from kill — the process keeps running; only the
+        # UI view is dropped (the user can reopen it from the status stack).
+        self.on_close = None
 
     @staticmethod
     def _clean_shell_noise(text: str) -> str:
@@ -916,13 +921,33 @@ class ProcessRegistry:
     # ----- Reader / Poller Threads -----
 
     def _reader_loop(self, session: ProcessSession):
-        """Background thread: read stdout from a local Popen process."""
+        """Background thread: read stdout from a local Popen process.
+
+        IMPORTANT: avoid ``TextIOWrapper.read(4096)`` here. On pipes that call can
+        block until EOF (or a large buffer fills), which makes "live" output land
+        in one burst at process exit. ``buffer.read1(4096)`` yields incremental
+        chunks as bytes become available, then we decode to text.
+        """
         first_chunk = True
         try:
+            stdout = session.process.stdout
+            if stdout is None:
+                return
+
+            raw_read = getattr(getattr(stdout, "buffer", None), "read1", None)
             while True:
-                chunk = session.process.stdout.read(4096)
-                if not chunk:
-                    break
+                if raw_read is not None:
+                    raw = raw_read(4096)
+                    if not raw:
+                        break
+                    chunk = raw.decode("utf-8", errors="replace")
+                else:
+                    # Fallback for mocked/alternate streams without a buffered raw
+                    # interface. This may be less "live", but keeps compatibility.
+                    chunk = stdout.read(4096)
+                    if not chunk:
+                        break
+
                 if first_chunk:
                     chunk = self._clean_shell_noise(chunk)
                     first_chunk = False
@@ -1501,6 +1526,37 @@ class ProcessRegistry:
         """Send data + newline to a running process's stdin (like pressing Enter)."""
         return self.write_stdin(session_id, data + "\n")
 
+    def request_close_terminal(self, session_id: str) -> dict:
+        """Ask the desktop GUI to close the read-only terminal tab mirroring this
+        background process.
+
+        This does NOT kill the process — it only drops the view. Output keeps
+        streaming into the (capped) buffer and the user can reopen the tab from
+        the status stack. Desktop-only: returns an error if no UI close sink is
+        wired (e.g. CLI / messaging)."""
+        sink = self.on_close
+        if sink is None:
+            return {
+                "status": "error",
+                "error": "close_terminal is only available in the Hermes desktop app.",
+            }
+        # The session may already be finished (or pruned) — the tab can still
+        # linger and be closed, so a missing session is not an error here.
+        session = self.get(session_id)
+        try:
+            sink(session, session_id)
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+        return {
+            "status": "ok",
+            "closed": session_id,
+            "note": (
+                "Closed the read-only terminal tab. The process was not killed; "
+                "its output remains available and the user can reopen the tab "
+                "from the status stack."
+            ),
+        }
+
     def close_stdin(self, session_id: str) -> dict:
         """Close a running process's stdin / send EOF without killing the process."""
         session = self.get(session_id)

toolsets.py

@@ -33,9 +33,10 @@ _HERMES_CORE_TOOLS = [
     "web_search", "web_extract",
     # Terminal + process management
     "terminal", "process",
-    # Read the desktop GUI's embedded terminal pane (gated on HERMES_DESKTOP
-    # via check_fn in tools/read_terminal_tool.py — hidden outside the GUI).
-    "read_terminal",
+    # Read the desktop GUI's embedded terminal pane, and close an agent's
+    # read-only terminal tab (both gated on HERMES_DESKTOP via check_fn —
+    # hidden outside the GUI).
+    "read_terminal", "close_terminal",
     # File manipulation
     "read_file", "write_file", "patch", "search_files",
     # Vision + image generation
@@ -345,7 +346,7 @@ TOOLSETS = {
         "description": "Coding-focused toolset: files, terminal, search, web docs, skills, todo, delegate, vision, browser",
         "tools": [
             "web_search", "web_extract",
-            "terminal", "process", "read_terminal",
+            "terminal", "process", "read_terminal", "close_terminal",
             "read_file", "write_file", "patch", "search_files",
             "vision_analyze",
             "skills_list", "skill_view", "skill_manage",

tui_gateway/server.py

@@ -8312,19 +8312,44 @@ def _notification_poller_loop(
 
 
 def _wire_agent_terminal_output() -> None:
-    """Idempotently route background-process output chunks to the desktop as
-    `agent.terminal.output` events (keyed by process id). Read-only agent
-    terminal tabs stream these live instead of polling the output tail.
-    `_emit`/`write_json` is `_stdout_lock`-guarded, so calling it from the
-    registry's reader threads is safe."""
+    """Idempotently route background-process output (and tab-close requests) to
+    the desktop, keyed by process id. Read-only agent terminal tabs stream
+    `agent.terminal.output` chunks live instead of polling the output tail, and
+    `process_registry.request_close_terminal` emits `terminal.close` so the agent
+    can drop a tab without killing the process. Events are routed to the window
+    that owns the process (its gateway session); `_emit`/`write_json` is
+    `_stdout_lock`-guarded, so calling it from the registry's reader threads is
+    safe."""
     from tools.process_registry import process_registry
 
     if getattr(process_registry, "on_output", None) is not None:
         return
 
-    process_registry.on_output = lambda session, chunk: _emit(
-        "agent.terminal.output", "", {"process_id": session.id, "chunk": chunk}
-    )
+    def _owner_sid_for_process(session) -> str:
+        session_key = str(getattr(session, "session_key", "") or "")
+        if not session_key:
+            return ""
+        with _sessions_lock:
+            for sid, tui_session in _sessions.items():
+                if str(tui_session.get("session_key") or "") == session_key:
+                    return sid
+        return ""
+
+    def _emit_agent_terminal_output(session, chunk):
+        _emit(
+            "agent.terminal.output",
+            _owner_sid_for_process(session),
+            {"process_id": session.id, "chunk": chunk},
+        )
+
+    def _emit_agent_terminal_close(session, process_id):
+        # session may be None (process already finished/pruned) — the tab can
+        # still linger and be closed; route to the owning window when we can.
+        sid = _owner_sid_for_process(session) if session is not None else ""
+        _emit("terminal.close", sid, {"process_id": process_id})
+
+    process_registry.on_output = _emit_agent_terminal_output
+    process_registry.on_close = _emit_agent_terminal_close
 
 
 def _start_notification_poller(sid: str, session: dict) -> threading.Event: