diff --git a/docs/terminal_tool_plan.md b/docs/terminal_tool_plan.md
new file mode 100644
index 0000000..a6b21bc
--- /dev/null
+++ b/docs/terminal_tool_plan.md
@@ -0,0 +1,111 @@
+# План: инструмент терминала — гэпы + UI управления терминалами
+
+Контекст: см. анализ в конце обсуждения (terminal.py, terminal_manager.py, navi_code profile).
+Гэпы: TUI не обрабатывает `terminal_output`/`terminal_closed`; `TerminalClosed` не
+эмитируется; live-stream background-терминала обрывается после `open` (per-tool sink
+закрывается); нет TUI-рендерера terminal tool_call; нет способа посмотреть/закрыть терминалы из UI.
+
+Решение по live-выводу: **модалка + count** (не live-панель). TUI накапливает `terminal_output`
+по терминалам; `/terminals` модалка показывает список + last output; статус-бар — count открытых.
+
+## Этап 1 — Backend: session-broadcast sink + lifecycle events
+
+**Корень гэпов 2+3:** `event_sink` сейчас per-tool-call (`current_event_sink.set` в
+`_execute_tools_with_sink`, `_TOOL_DONE` после возврата `open`). Background-терминал живёт в
+`terminal_manager`, его reader-tasks пишут в `output_buffer` (deque 500), но в стрим уже некому.
+`TerminalClosed` не эмитируется вообще.
+
+**Что делаем:**
+- `TerminalManager`: per-session broadcast sink — `bind_session(session_id, sink)` /
+  `unbind_session(session_id)`. Хранит `dict[session_id, sink]`. Reader-tasks и lifecycle-события
+  пишут в session-sink (живёт пока сессия), не в per-tool sink.
+- `terminal_manager.open`: reader-tasks (`_read_stream`) → `TerminalOutputDelta` в session-sink
+  (если bound), не в per-tool `event_sink`. Per-tool sink остаётся только для immediate open-feedback.
+- `terminal_manager._close_one`: emit `TerminalClosed(name, reason)` в session-sink.
+- Новое событие `TerminalOpened` (`events.py`: dataclass + `to_wire` → `{"type":"terminal_opened",
+  "terminal_name","description","pid","background"}`), emit в `open` (после старта proc).
+- `orchestrator`/`container`: `bind_session(session_id, broadcast)` при старте хода/attach,
+  `unbind` + `close_all` на session end / shutdown. (Точка: где session-broadcast создаётся —
+  orchestrator, который шлёт WS.)
+
+**Риск:** средний. Меняет sink-архитектуру terminal_manager. Совместимость: webclient уже
+handle `terminal_output`/`terminal_closed`; `terminal_opened` — новый handler (мелочь).
+
+## Этап 2 — Backend: REST endpoints для list/close
+
+`/terminals` модалке нужен источник списка + закрытие. Тянуть через REST (как sessions_picker).
+
+- `GET /sessions/{id}/terminals` → `terminal_manager.list(session_id)` (массив summary).
+- `POST /sessions/{id}/terminals/{name}/close` → `terminal_manager.close(session_id, name)`.
+- `api.py` (terminal client, async): `list_terminals(session_id)`, `close_terminal(session_id, name)`.
+
+**Риск:** низкий. Прозрачные endpoint-ы над существующим terminal_manager.
+
+## Этап 3 — TUI: `terminal` tool_call renderer (action-aware)
+
+Сейчас `tool_call` с `tool="terminal"` → generic `ToolResultRenderer` (без структуры). `filesystem`
+имеет спец-рендерер; `terminal` — нет.
+
+- `renderers/terminal.py`: action-aware (по образцу filesystem):
+  - `run`: команда + exit-code (success/error цвет) + output (capped).
+  - `open`: `terminal_name` + description + PID + background-флаг.
+  - `list`: таблица активных (статус-эмодзи, name, description, PID, uptime).
+  - `status`: name/desc/command/PID/cwd/uptime + output tail.
+  - `send_input`: echo «Sent input to <name>».
+  - `close`: «Terminal <name> closed».
+- Регистрация в `renderers/__init__.py` **перед** generic `ToolResultRenderer` (как filesystem/todo).
+- `chat_panel._item_msg` для `tool_call` уже раскрывает meta — `terminal` renderer читает
+  `args.action`/`result`/`success`/`metadata`.
+
+**Риск:** низкий. Только рендер.
+
+## Этап 4 — TUI: handle terminal events + count в статусе
+
+- `ChatModel.handle_ws_event`: cases для `terminal_opened` / `terminal_output` / `terminal_closed`:
+  - накапливать состояние `terminals: dict[name → {status, output_tail, pid, background, closed}]`;
+  - **не** создавать chat-пузырь (как `model_info`/`todo_updated` — `return` без форварда в чат).
+- `tui_app.on_ws_event`: forward в chat_model + обновлять StatusPanel count.
+- `StatusPanel`: заменить `_hint` («Ctrl+P palette | /help commands») на «Terminals: N».
+  Seed на attach (Этап 2 `api.list_terminals`), live-обновление из events.
+- Seed count на `attach_session` через `api.list_terminals` (async, в worker — как `get_todos`).
+
+**Риск:** низкий-средний. StatusPanel hint → count (видимый UX change, может захотеть combos
+куда-то перенести — но пользователь сказал заменить).
+
+## Этап 5 — TUI: `/terminals` команда + модалка
+
+- `screens/terminals_picker.py` (модалка, по образцу `sessions_picker`):
+  - список: `api.list_terminals(session_id)` (REST seed) + live из ChatModel-состояния
+    (terminal_output обновляет output_tail, terminal_opened/closed — список).
+  - строка: статус-эмодзи + name + description + PID + uptime; expand → last output tail.
+  - `up/down` — навигация, `enter` — **close** выбранного (`api.close_terminal`),
+    `escape` — cancel. Без подтверждения (явное действие в модалке).
+- `commands/builtin.py`: `TerminalsCommand` (`/terminals`) → `app._open_terminals_picker()`.
+  meta.keybind — опционально (напр. `ctrl+x t` свободен? сейчас `ctrl+x t` = toggle thinking;
+  подберём свободный или без keybind).
+- `tui_app._open_terminals_picker`: push screen + callback (close → `run_worker`).
+
+**Риск:** средний. Модалка + live-обновление списка при events (refresh при terminal_output/
+opened/closed, если модалка открыта).
+
+## Порядок и контроль
+
+Этапы 1→2 (backend) → 3→4→5 (TUI). Каждый: подтверждение подхода → реализация + тесты →
+полный pytest зелёный → коммит (без Co-Authored-By).
+
+Зависимости: Этап 4/5 зависят от 1 (events) + 2 (REST). Этап 3 (renderer) независим.
+
+## Тесты (по этапам)
+
+- **1:** terminal_manager bind/unbind; `terminal_opened`/`terminal_closed` emit в session-sink;
+  background output стримится после возврата `open`.
+- **2:** REST list/close (через test-client + mock terminal_manager).
+- **3:** renderer: каждый action → структура (run exit-code, open PID, list таблица, status tail).
+- **4:** ChatModel cases (накапливает, не создаёт пузырь); StatusPanel «Terminals: N» обновляется.
+- **5:** модалка: список seed + close → `api.close_terminal`; live refresh.
+
+## Out of scope
+
+- Live-панель с реалтайм-выводом (отклонено — модалка + count).
+- Подсветка terminal-output как код (минор, можно позже).
+- Дедуп `_resolve_working_dir` (terminal/code_exec) — maintainability, не в этом плане.
\ No newline at end of file
diff --git a/navi/core/container.py b/navi/core/container.py
index fd87f46..b386645 100644
--- a/navi/core/container.py
+++ b/navi/core/container.py
@@ -222,4 +222,8 @@
     from navi.core.orchestrator import AgentSessionOrchestrator
 
     container.orchestrator = AgentSessionOrchestrator(container)
+    # Route persistent-terminal lifecycle/stream events to the session's
+    # WebSocket(s) (terminal_opened/output/closed) — the manager emits them,
+    # the orchestrator fans them out per session_id.
+    terminal_manager.set_event_callback(container.orchestrator._on_terminal_event)
     return container
diff --git a/navi/core/events.py b/navi/core/events.py
index dbfaf11..be12122 100644
--- a/navi/core/events.py
+++ b/navi/core/events.py
@@ -348,6 +348,25 @@
 
 
 @dataclass
+class TerminalOpened:
+    """Emitted when a persistent terminal session starts (background or foreground)."""
+
+    terminal_name: str
+    description: str
+    pid: int | None
+    background: bool
+
+    def to_wire(self) -> dict:
+        return {
+            "type": "terminal_opened",
+            "terminal_name": self.terminal_name,
+            "description": self.description,
+            "pid": self.pid,
+            "background": self.background,
+        }
+
+
+@dataclass
 class TerminalClosed:
     """Emitted when a persistent terminal session ends."""
 
@@ -389,5 +408,5 @@
     ToolStarted | ToolEvent | TextDelta | ThinkingDelta | ThinkingEnd
     | StreamEnd | StreamStopped | CompressionStarted | ContextCompressed | TurnThinking | ProfileSwitched
     | PlanningStatus | PlanReady | SubagentComplete | AIHelperTokensUsed | PlanningDebugData
-    | RecallUpdate | McpStatusUpdate | TerminalOutputDelta | TerminalClosed | TodoUpdated
+    | RecallUpdate | McpStatusUpdate | TerminalOutputDelta | TerminalOpened | TerminalClosed | TodoUpdated
 )
diff --git a/navi/core/orchestrator.py b/navi/core/orchestrator.py
index 0fa8b4c..fa6fdcc 100644
--- a/navi/core/orchestrator.py
+++ b/navi/core/orchestrator.py
@@ -119,6 +119,17 @@
             if payload:
                 await self._broadcast_all_sessions(payload)
 
+    async def _on_terminal_event(self, session_id: str, event: Any) -> None:
+        """Deliver a terminal lifecycle/stream event to that session's clients.
+
+        ``terminal_opened`` / ``terminal_output`` / ``terminal_closed`` — emitted
+        by the TerminalManager (background terminals outlive any single tool
+        call, so they go straight to the WS, not through the tool-call sink).
+        """
+        payload = event.to_wire() if hasattr(event, "to_wire") else None
+        if payload:
+            await self._notify_session(session_id, payload)
+
     def _get_or_create_state(self, session_id: str) -> SessionState:
         state = self._sessions.get(session_id)
         if state is None:
diff --git a/navi/tools/_internal/terminal_manager.py b/navi/tools/_internal/terminal_manager.py
index 5a89c7c..60c9893 100644
--- a/navi/tools/_internal/terminal_manager.py
+++ b/navi/tools/_internal/terminal_manager.py
@@ -10,6 +10,7 @@
 import asyncio
 import dataclasses
 from collections import deque
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
@@ -22,6 +23,13 @@
 
 log = structlog.get_logger()
 
+# Lifecycle/stream events from a persistent terminal are delivered to the
+# caller via this callback (set by the orchestrator) rather than a per-tool-call
+# queue — the per-tool ``event_sink`` closed once ``open`` returned, which
+# dropped all background output after the open. The callback lives for the
+# terminal's lifetime (and the session's).
+TerminalEventCallback = Callable[[str, "AgentEvent"], Awaitable[None]]
+
 _MAX_OUTPUT_BUFFER = 500  # lines kept per terminal for status queries
 _DEFAULT_CLEANUP_INTERVAL = 60  # seconds between idle checks
 _DEFAULT_MAX_IDLE = 1800  # 30 minutes
@@ -89,6 +97,29 @@
         self._sessions: dict[tuple[str, str], TerminalSession] = {}
         self._max_idle = max_idle_seconds
         self._cleanup_task: asyncio.Task | None = None
+        # One callback for all sessions; the session_id is carried on each
+        # TerminalSession and passed to the callback so it routes to the right
+        # WebSocket(s). Set by the orchestrator (container wiring).
+        self._event_callback: TerminalEventCallback | None = None
+
+    def set_event_callback(self, callback: TerminalEventCallback) -> None:
+        """Register the destination for terminal lifecycle/stream events."""
+        self._event_callback = callback
+
+    async def _emit(self, session_id: str, event: "AgentEvent") -> None:
+        """Deliver a terminal event to the callback, swallowing errors.
+
+        Reader/cleanup tasks run independently of any tool call; a delivery
+        failure (e.g. no callback wired, or the client gone) must not kill the
+        background terminal's output loop.
+        """
+        cb = self._event_callback
+        if cb is None:
+            return
+        try:
+            await cb(session_id, event)
+        except Exception:
+            log.exception("terminal_manager.emit_error", session_id=session_id)
 
     # ── Lifecycle ────────────────────────────────────────────────────────────
 
@@ -142,14 +173,14 @@
         cwd: Path | None = None,
         env: dict[str, str] | None = None,
         timeout: int = 20,
-        event_sink: asyncio.Queue | None = None,
         exec_tokens: list[str] | None = None,
     ) -> TerminalSession:
         """Open a new persistent terminal session.
 
         If *background* is True the process is started and the coroutine
-        returns immediately — output is streamed via *event_sink*.
-        If False, it waits for the process to finish and returns.
+        returns immediately — output is streamed via the registered event
+        callback (``set_event_callback``). If False, it waits for the process
+        to finish and returns.
 
         Pass *exec_tokens* to run via ``create_subprocess_exec`` instead of
         ``create_subprocess_shell`` (enforces allowlist restrictions).
@@ -199,13 +230,29 @@
         session.proc = proc
         session.stdin = proc.stdin
 
-        # Start background readers
+        # Notify the client a terminal has started (for the terminals panel /
+        # count). Emitted for both foreground and background; foreground ones
+        # close immediately after, which sends a paired TerminalClosed.
+        from navi.core.events import TerminalOpened
+
+        await self._emit(
+            session_id,
+            TerminalOpened(
+                terminal_name=name,
+                description=description,
+                pid=proc.pid,
+                background=background,
+            ),
+        )
+
+        # Start background readers — they stream output via the event callback
+        # (not a per-tool sink), so output keeps flowing after open returns.
         session.stdout_task = asyncio.create_task(
-            self._read_stream(proc.stdout, session, "stdout", event_sink),
+            self._read_stream(proc.stdout, session, "stdout"),
             name=f"term-{session_id}-{name}-stdout",
         )
         session.stderr_task = asyncio.create_task(
-            self._read_stream(proc.stderr, session, "stderr", event_sink),
+            self._read_stream(proc.stderr, session, "stderr"),
             name=f"term-{session_id}-{name}-stderr",
         )
 
@@ -287,7 +334,6 @@
         stream: asyncio.StreamReader | None,
         session: TerminalSession,
         stream_name: str,
-        event_sink: asyncio.Queue | None,
     ) -> None:
         if stream is None:
             return
@@ -299,19 +345,16 @@
                 text = line.decode(errors="replace")
                 session.output_buffer.append(text)
                 session.touch()
-                if event_sink is not None:
-                    try:
-                        from navi.core.events import TerminalOutputDelta
+                from navi.core.events import TerminalOutputDelta
 
-                        await event_sink.put(
-                            TerminalOutputDelta(
-                                terminal_name=session.name,
-                                stream=stream_name,
-                                delta=text,
-                            )
-                        )
-                    except Exception:
-                        log.exception("terminal_manager.sink_put_error")
+                await self._emit(
+                    session.session_id,
+                    TerminalOutputDelta(
+                        terminal_name=session.name,
+                        stream=stream_name,
+                        delta=text,
+                    ),
+                )
         except asyncio.CancelledError:
             pass
         except Exception:
@@ -357,4 +400,12 @@
             except Exception:
                 pass
 
+        # Notify the client the terminal has ended (count −1, /terminals refresh).
+        from navi.core.events import TerminalClosed
+
+        await self._emit(
+            key[0],
+            TerminalClosed(terminal_name=key[1], reason=reason),
+        )
+
         log.info("terminal_manager.closed", session_id=key[0], name=key[1], reason=reason)
diff --git a/navi/tools/terminal.py b/navi/tools/terminal.py
index 9b7972e..1582e19 100644
--- a/navi/tools/terminal.py
+++ b/navi/tools/terminal.py
@@ -35,7 +35,6 @@
     Tool,
     ToolContext,
     ToolResult,
-    current_event_sink,
     current_user_id,
     current_user_role,
     current_working_directory,
@@ -348,7 +347,6 @@
                 background=background,
                 cwd=cwd,
                 timeout=timeout,
-                event_sink=current_event_sink.get(),
                 exec_tokens=exec_tokens,
             )
         except ValueError as e:
diff --git a/tests/unit/tools/test_terminal.py b/tests/unit/tools/test_terminal.py
index 4e53acd..c916a2e 100644
--- a/tests/unit/tools/test_terminal.py
+++ b/tests/unit/tools/test_terminal.py
@@ -215,3 +215,107 @@
         # Cleanup
         for i in range(_MAX_TERMINALS_PER_SESSION):
             await tool.execute({"action": "close", "terminal_name": f"t{i}"}, ctx=ctx)
+
+
+class TestTerminalManagerEvents:
+    """Lifecycle/stream events reach the registered callback (not a per-tool
+    sink), so background output keeps flowing after open returns."""
+
+    @pytest.fixture
+    async def manager(self):
+        tm = TerminalManager(max_idle_seconds=60)
+        tm.start()
+        yield tm
+        await tm.shutdown()
+
+    async def test_open_emits_terminal_opened(self, manager):
+        from navi.core.events import TerminalOpened
+
+        events: list = []
+
+        async def cb(session_id, event):
+            events.append((session_id, event))
+
+        manager.set_event_callback(cb)
+
+        await manager.open(
+            session_id="se1",
+            name="t1",
+            description="d",
+            command="echo hello",
+            background=False,
+            timeout=5,
+        )
+        opened = [e for _, e in events if isinstance(e, TerminalOpened)]
+        assert len(opened) == 1
+        assert opened[0].terminal_name == "t1"
+        assert opened[0].description == "d"
+        assert opened[0].background is False
+        assert opened[0].pid is not None
+
+    async def test_background_streams_output_after_open_returns(self, manager):
+        """The fix for gap 3: background output is delivered via the callback
+        even after the open coroutine returned (no per-tool sink to close)."""
+        from navi.core.events import TerminalOpened, TerminalOutputDelta
+
+        events: list = []
+
+        async def cb(session_id, event):
+            events.append((session_id, event))
+
+        manager.set_event_callback(cb)
+
+        # A quick producer so the background terminal exits on its own.
+        await manager.open(
+            session_id="se2",
+            name="bg",
+            description="bg stream",
+            command="echo streamed-line",
+            background=True,
+            timeout=5,
+        )
+        # The open call has returned; give the readers time to flush output.
+        await asyncio.sleep(0.5)
+        await manager.close("se2", "bg")
+
+        deltas = [e for _, e in events if isinstance(e, TerminalOutputDelta)]
+        assert any("streamed-line" in d.delta for d in deltas)
+        assert any(isinstance(e, TerminalOpened) for _, e in events)
+
+    async def test_close_emits_terminal_closed(self, manager):
+        from navi.core.events import TerminalClosed
+
+        events: list = []
+
+        async def cb(session_id, event):
+            events.append((session_id, event))
+
+        manager.set_event_callback(cb)
+
+        await manager.open(
+            session_id="se3",
+            name="tc",
+            description="d",
+            command="sleep 30",
+            background=True,
+            timeout=5,
+        )
+        await manager.close("se3", "tc")
+        # close_all/shutdown also closes — wait a tick for the event to land.
+        await asyncio.sleep(0.1)
+        closed = [e for _, e in events if isinstance(e, TerminalClosed)]
+        assert any(c.terminal_name == "tc" for c in closed)
+
+    async def test_no_callback_does_not_break_readers(self, manager):
+        """Without a registered callback, open/read/close still work (events
+        silently dropped) — the manager must not depend on a client."""
+        session = await manager.open(
+            session_id="se4",
+            name="nc",
+            description="d",
+            command="echo ok",
+            background=False,
+            timeout=5,
+        )
+        assert session.proc is not None
+        await manager.close("se4", "nc")