diff --git a/CHANGES b/CHANGES index f234a148..85ac8f4c 100644 --- a/CHANGES +++ b/CHANGES @@ -6,6 +6,13 @@ _Notes on upcoming releases will be added here_ +### What's new + +- New pane tools: `snapshot_pane`, `wait_for_content_change`, `select_pane`, `swap_pane`, `pipe_pane`, `display_message`, `enter_copy_mode`, `exit_copy_mode`, and `paste_text` (#11) +- New session tool: `select_window` — navigate by ID, index, or direction (#11) +- New window tool: `move_window` — reorder within a session or move across sessions (#11) +- New models: `PaneSnapshot` and `ContentChangeResult` (#11) + ### Documentation - Visual improvements to API docs from [gp-sphinx](https://gp-sphinx.git-pull.com)-based Sphinx packages (#10) diff --git a/README.md b/README.md index 44769b95..70930f37 100644 --- a/README.md +++ b/README.md @@ -16,9 +16,9 @@ Give your AI agent hands inside the terminal — create sessions, run commands, | Module | Tools | |--------|-------| | **Server** | `list_sessions`, `create_session`, `kill_server`, `get_server_info` | -| **Session** | `list_windows`, `create_window`, `rename_session`, `kill_session` | -| **Window** | `list_panes`, `split_window`, `rename_window`, `kill_window`, `select_layout`, `resize_window` | -| **Pane** | `send_keys`, `capture_pane`, `resize_pane`, `kill_pane`, `set_pane_title`, `get_pane_info`, `clear_pane`, `search_panes`, `wait_for_text` | +| **Session** | `list_windows`, `create_window`, `rename_session`, `select_window`, `kill_session` | +| **Window** | `list_panes`, `split_window`, `rename_window`, `select_layout`, `resize_window`, `move_window`, `kill_window` | +| **Pane** | `send_keys`, `paste_text`, `capture_pane`, `snapshot_pane`, `search_panes`, `get_pane_info`, `wait_for_text`, `wait_for_content_change`, `display_message`, `select_pane`, `swap_pane`, `resize_pane`, `set_pane_title`, `clear_pane`, `pipe_pane`, `enter_copy_mode`, `exit_copy_mode`, `kill_pane` | | **Options** | `show_option`, `set_option` | | **Environment** | `show_environment`, `set_environment` | diff --git a/docs/index.md b/docs/index.md index e83026be..850d2e57 100644 --- a/docs/index.md +++ b/docs/index.md @@ -53,13 +53,13 @@ Config blocks for Claude Desktop, Claude Code, Cursor, and others. Read tmux state without changing anything. -{toolref}`list-sessions` · {toolref}`capture-pane` · {toolref}`get-pane-info` · {toolref}`search-panes` · {toolref}`wait-for-text` +{toolref}`list-sessions` · {toolref}`capture-pane` · {toolref}`snapshot-pane` · {toolref}`get-pane-info` · {toolref}`search-panes` · {toolref}`wait-for-text` · {toolref}`wait-for-content-change` · {toolref}`display-message` ### Act (mutating) Create or modify tmux objects. -{toolref}`create-session` · {toolref}`send-keys` · {toolref}`create-window` · {toolref}`split-window` · {toolref}`resize-pane` · {toolref}`set-option` +{toolref}`create-session` · {toolref}`send-keys` · {toolref}`paste-text` · {toolref}`create-window` · {toolref}`split-window` · {toolref}`select-pane` · {toolref}`select-window` · {toolref}`move-window` · {toolref}`resize-pane` · {toolref}`pipe-pane` · {toolref}`set-option` ### Destroy (destructive) diff --git a/docs/tools/index.md b/docs/tools/index.md index c3de5bac..0a8e29f2 100644 --- a/docs/tools/index.md +++ b/docs/tools/index.md @@ -8,18 +8,36 @@ All tools accept an optional `socket_name` parameter for multi-server support. I **Reading terminal content?** - Know which pane? → {tool}`capture-pane` +- Need text + cursor + mode in one call? → {tool}`snapshot-pane` - Don't know which pane? → {tool}`search-panes` -- Need to wait for output? → {tool}`wait-for-text` +- Need to wait for specific output? → {tool}`wait-for-text` +- Need to wait for *any* change? → {tool}`wait-for-content-change` - Only need metadata (PID, path, size)? → {tool}`get-pane-info` +- Need an arbitrary tmux variable? → {tool}`display-message` **Running a command?** - {tool}`send-keys` — then {tool}`wait-for-text` + {tool}`capture-pane` +- Pasting multi-line text? → {tool}`paste-text` **Creating workspace structure?** - New session → {tool}`create-session` - New window → {tool}`create-window` - New pane → {tool}`split-window` +**Navigating?** +- Switch pane → {tool}`select-pane` (by ID or direction) +- Switch window → {tool}`select-window` (by ID, index, or direction) + +**Rearranging layout?** +- Swap two panes → {tool}`swap-pane` +- Move window → {tool}`move-window` +- Change layout → {tool}`select-layout` + +**Scrollback / copy mode?** +- Enter copy mode → {tool}`enter-copy-mode` +- Exit copy mode → {tool}`exit-copy-mode` +- Log output to file → {tool}`pipe-pane` + **Changing settings?** - tmux options → {tool}`show-option` / {tool}`set-option` - Environment vars → {tool}`show-environment` / {tool}`set-environment` @@ -91,6 +109,24 @@ Query a tmux option value. Show tmux environment variables. ::: +:::{grid-item-card} snapshot_pane +:link: snapshot-pane +:link-type: ref +Rich capture: content + cursor + mode + scroll. +::: + +:::{grid-item-card} wait_for_content_change +:link: wait-for-content-change +:link-type: ref +Wait for any screen change. +::: + +:::{grid-item-card} display_message +:link: display-message +:link-type: ref +Query arbitrary tmux format strings. +::: + :::: ## Act @@ -178,6 +214,54 @@ Set a tmux option. Set a tmux environment variable. ::: +:::{grid-item-card} select_pane +:link: select-pane +:link-type: ref +Focus a pane by ID or direction. +::: + +:::{grid-item-card} select_window +:link: select-window +:link-type: ref +Focus a window by ID, index, or direction. +::: + +:::{grid-item-card} swap_pane +:link: swap-pane +:link-type: ref +Exchange positions of two panes. +::: + +:::{grid-item-card} move_window +:link: move-window +:link-type: ref +Move window to another index or session. +::: + +:::{grid-item-card} pipe_pane +:link: pipe-pane +:link-type: ref +Stream pane output to a file. +::: + +:::{grid-item-card} enter_copy_mode +:link: enter-copy-mode +:link-type: ref +Enter copy mode for scrollback. +::: + +:::{grid-item-card} exit_copy_mode +:link: exit-copy-mode +:link-type: ref +Exit copy mode. +::: + +:::{grid-item-card} paste_text +:link: paste-text +:link-type: ref +Paste multi-line text via tmux buffer. +::: + :::: ## Destroy diff --git a/docs/tools/panes.md b/docs/tools/panes.md index 3a7b8f1b..13312580 100644 --- a/docs/tools/panes.md +++ b/docs/tools/panes.md @@ -81,7 +81,7 @@ Response: "pane_active": "1", "window_id": "@0", "session_id": "$0", - "is_caller": null + "is_caller": false } ``` @@ -130,7 +130,7 @@ Response: "FAIL: test_upload (AssertionError)", "3 tests: 2 passed, 1 failed" ], - "is_caller": null + "is_caller": false } ] ``` @@ -183,6 +183,133 @@ Response: ```{fastmcp-tool-input} pane_tools.wait_for_text ``` +--- + +```{fastmcp-tool} pane_tools.snapshot_pane +``` + +**Use when** you need a complete picture of a pane in a single call — visible +text plus cursor position, whether the pane is in copy mode, scroll offset, +and scrollback history size. Replaces separate `capture_pane` + +`get_pane_info` calls when you need to reason about cursor location or +terminal mode. + +**Avoid when** you only need raw text — {tooliconl}`capture-pane` is lighter. + +**Side effects:** None. Readonly. + +**Example:** + +```json +{ + "tool": "snapshot_pane", + "arguments": { + "pane_id": "%0" + } +} +``` + +Response: + +```json +{ + "pane_id": "%0", + "content": "$ npm test\n\nPASS src/auth.test.ts\nTests: 3 passed\n$", + "cursor_x": 2, + "cursor_y": 4, + "pane_width": 80, + "pane_height": 24, + "pane_in_mode": false, + "pane_mode": null, + "scroll_position": null, + "history_size": 142, + "title": null, + "pane_current_command": "zsh", + "pane_current_path": "/home/user/myproject", + "is_caller": false +} +``` + +```{fastmcp-tool-input} pane_tools.snapshot_pane +``` + +--- + +```{fastmcp-tool} pane_tools.wait_for_content_change +``` + +**Use when** you've sent a command and need to wait for *something* to happen, +but you don't know what the output will look like. Unlike +{tooliconl}`wait-for-text`, this waits for *any* screen change rather than a +specific pattern. + +**Avoid when** you know the expected output — {tooliconl}`wait-for-text` is more +precise and avoids false positives from unrelated output. + +**Side effects:** None. Readonly. Blocks until content changes or timeout. + +**Example:** + +```json +{ + "tool": "wait_for_content_change", + "arguments": { + "pane_id": "%0", + "timeout": 10 + } +} +``` + +Response: + +```json +{ + "changed": true, + "pane_id": "%0", + "elapsed_seconds": 1.234, + "timed_out": false +} +``` + +```{fastmcp-tool-input} pane_tools.wait_for_content_change +``` + +--- + +```{fastmcp-tool} pane_tools.display_message +``` + +**Use when** you need to query arbitrary tmux variables — zoom state, pane +dead flag, client activity, or any `#{format}` string that isn't covered by +other tools. + +**Avoid when** a dedicated tool already provides the information — e.g. use +{tooliconl}`snapshot-pane` for cursor position and mode, or +{tooliconl}`get-pane-info` for standard metadata. + +**Side effects:** None. Readonly. + +**Example:** + +```json +{ + "tool": "display_message", + "arguments": { + "format_string": "zoomed=#{window_zoomed_flag} dead=#{pane_dead}", + "pane_id": "%0" + } +} +``` + +Response (string): + +```text +zoomed=0 dead=0 +``` + +```{fastmcp-tool-input} pane_tools.display_message +``` + ## Act ```{fastmcp-tool} pane_tools.send_keys @@ -254,7 +381,7 @@ Response: "pane_active": "1", "window_id": "@0", "session_id": "$0", - "is_caller": null + "is_caller": false } ``` @@ -326,13 +453,280 @@ Response: "pane_active": "1", "window_id": "@0", "session_id": "$0", - "is_caller": null + "is_caller": false } ``` ```{fastmcp-tool-input} pane_tools.resize_pane ``` +--- + +```{fastmcp-tool} pane_tools.select_pane +``` + +**Use when** you need to focus a specific pane — by ID for a known target, +or by direction (`up`, `down`, `left`, `right`, `last`, `next`, `previous`) +to navigate a multi-pane layout. + +**Side effects:** Changes the active pane in the window. + +**Example:** + +```json +{ + "tool": "select_pane", + "arguments": { + "direction": "down", + "window_id": "@0" + } +} +``` + +Response: + +```json +{ + "pane_id": "%1", + "pane_index": "1", + "pane_width": "80", + "pane_height": "11", + "pane_current_command": "zsh", + "pane_current_path": "/home/user/myproject", + "pane_pid": "12400", + "pane_title": "", + "pane_active": "1", + "window_id": "@0", + "session_id": "$0", + "is_caller": false +} +``` + +```{fastmcp-tool-input} pane_tools.select_pane +``` + +--- + +```{fastmcp-tool} pane_tools.swap_pane +``` + +**Use when** you want to rearrange pane positions without changing content — +e.g. moving a log pane from bottom to top. + +**Side effects:** Exchanges the visual positions of two panes. + +**Example:** + +```json +{ + "tool": "swap_pane", + "arguments": { + "source_pane_id": "%0", + "target_pane_id": "%1" + } +} +``` + +Response: + +```json +{ + "pane_id": "%0", + "pane_index": "1", + "pane_width": "80", + "pane_height": "11", + "pane_current_command": "zsh", + "pane_current_path": "/home/user/myproject", + "pane_pid": "12345", + "pane_title": "", + "pane_active": "1", + "window_id": "@0", + "session_id": "$0", + "is_caller": false +} +``` + +```{fastmcp-tool-input} pane_tools.swap_pane +``` + +--- + +```{fastmcp-tool} pane_tools.pipe_pane +``` + +**Use when** you need to log pane output to a file — useful for monitoring +long-running processes or capturing output that scrolls past the visible +area. + +**Avoid when** you only need a one-time capture — use {tooliconl}`capture-pane` +with `start`/`end` to read scrollback. + +**Side effects:** Starts or stops piping output to a file. Call with +`output_path=null` to stop. + +**Example:** + +```json +{ + "tool": "pipe_pane", + "arguments": { + "pane_id": "%0", + "output_path": "/tmp/build.log" + } +} +``` + +Response (start): + +```text +Piping pane %0 to /tmp/build.log +``` + +**Stopping the pipe:** + +```json +{ + "tool": "pipe_pane", + "arguments": { + "pane_id": "%0", + "output_path": null + } +} +``` + +Response (stop): + +```text +Piping stopped for pane %0 +``` + +```{fastmcp-tool-input} pane_tools.pipe_pane +``` + +--- + +```{fastmcp-tool} pane_tools.enter_copy_mode +``` + +**Use when** you need to scroll through scrollback history in a pane. +Optionally scroll up immediately after entering. Use +{tooliconl}`snapshot-pane` afterward to read the `scroll_position` and +visible content. + +**Side effects:** Puts the pane into copy mode. The pane stops receiving +new output until you exit copy mode. + +**Example:** + +```json +{ + "tool": "enter_copy_mode", + "arguments": { + "pane_id": "%0", + "scroll_up": 50 + } +} +``` + +Response: + +```json +{ + "pane_id": "%0", + "pane_index": "0", + "pane_width": "80", + "pane_height": "24", + "pane_current_command": "zsh", + "pane_current_path": "/home/user/myproject", + "pane_pid": "12345", + "pane_title": "", + "pane_active": "1", + "window_id": "@0", + "session_id": "$0", + "is_caller": false +} +``` + +```{fastmcp-tool-input} pane_tools.enter_copy_mode +``` + +--- + +```{fastmcp-tool} pane_tools.exit_copy_mode +``` + +**Use when** you're done scrolling through scrollback and want the pane to +resume receiving output. + +**Side effects:** Exits copy mode, returning the pane to normal. + +**Example:** + +```json +{ + "tool": "exit_copy_mode", + "arguments": { + "pane_id": "%0" + } +} +``` + +Response: + +```json +{ + "pane_id": "%0", + "pane_index": "0", + "pane_width": "80", + "pane_height": "24", + "pane_current_command": "zsh", + "pane_current_path": "/home/user/myproject", + "pane_pid": "12345", + "pane_title": "", + "pane_active": "1", + "window_id": "@0", + "session_id": "$0", + "is_caller": false +} +``` + +```{fastmcp-tool-input} pane_tools.exit_copy_mode +``` + +--- + +```{fastmcp-tool} pane_tools.paste_text +``` + +**Use when** you need to paste multi-line text into a pane — e.g. a code +block, a config snippet, or a heredoc. Uses tmux paste buffers for clean +multi-line input instead of sending text line-by-line via +{tooliconl}`send-keys`. + +**Side effects:** Pastes text into the pane. With `bracket=true` (default), +uses bracketed paste mode so the terminal knows this is pasted text. + +**Example:** + +```json +{ + "tool": "paste_text", + "arguments": { + "text": "def hello():\n print('world')\n", + "pane_id": "%0" + } +} +``` + +Response (string): + +```text +Text pasted to pane %0 +``` + +```{fastmcp-tool-input} pane_tools.paste_text +``` + ## Destroy ```{fastmcp-tool} pane_tools.kill_pane diff --git a/docs/tools/sessions.md b/docs/tools/sessions.md index dd7ef23a..984f0420 100644 --- a/docs/tools/sessions.md +++ b/docs/tools/sessions.md @@ -150,6 +150,48 @@ Response: ```{fastmcp-tool-input} session_tools.rename_session ``` +--- + +```{fastmcp-tool} session_tools.select_window +``` + +**Use when** you need to switch focus to a different window — by ID, index, +or direction (`next`, `previous`, `last`). + +**Side effects:** Changes the active window in the session. + +**Example:** + +```json +{ + "tool": "select_window", + "arguments": { + "direction": "next", + "session_name": "dev" + } +} +``` + +Response: + +```json +{ + "window_id": "@1", + "window_name": "server", + "window_index": "2", + "session_id": "$0", + "session_name": "dev", + "pane_count": 1, + "window_layout": "b25f,80x24,0,0,2", + "window_active": "1", + "window_width": "80", + "window_height": "24" +} +``` + +```{fastmcp-tool-input} session_tools.select_window +``` + ## Destroy ```{fastmcp-tool} session_tools.kill_session diff --git a/docs/tools/windows.md b/docs/tools/windows.md index 4b022f55..e7bd0cea 100644 --- a/docs/tools/windows.md +++ b/docs/tools/windows.md @@ -94,7 +94,7 @@ Response: "pane_active": "1", "window_id": "@0", "session_id": "$0", - "is_caller": null + "is_caller": false }, { "pane_id": "%1", @@ -108,7 +108,7 @@ Response: "pane_active": "0", "window_id": "@0", "session_id": "$0", - "is_caller": null + "is_caller": false } ] ``` @@ -194,7 +194,7 @@ Response: "pane_active": "0", "window_id": "@0", "session_id": "$0", - "is_caller": null + "is_caller": false } ``` @@ -326,6 +326,48 @@ Response: ```{fastmcp-tool-input} window_tools.resize_window ``` +--- + +```{fastmcp-tool} window_tools.move_window +``` + +**Use when** you need to reorder windows within a session or move a window +to a different session entirely. + +**Side effects:** Changes the window's index or parent session. + +**Example:** + +```json +{ + "tool": "move_window", + "arguments": { + "window_id": "@1", + "destination_index": "1" + } +} +``` + +Response: + +```json +{ + "window_id": "@1", + "window_name": "server", + "window_index": "1", + "session_id": "$0", + "session_name": "dev", + "pane_count": 1, + "window_layout": "b25f,80x24,0,0,2", + "window_active": "0", + "window_width": "80", + "window_height": "24" +} +``` + +```{fastmcp-tool-input} window_tools.move_window +``` + ## Destroy ```{fastmcp-tool} window_tools.kill_window diff --git a/src/libtmux_mcp/models.py b/src/libtmux_mcp/models.py index 563f9119..ebde388a 100644 --- a/src/libtmux_mcp/models.py +++ b/src/libtmux_mcp/models.py @@ -139,3 +139,43 @@ class WaitForTextResult(BaseModel): pane_id: str = Field(description="Pane ID that was polled") elapsed_seconds: float = Field(description="Time spent waiting in seconds") timed_out: bool = Field(description="Whether the timeout was reached") + + +class PaneSnapshot(BaseModel): + """Rich screen capture with metadata: content, cursor, mode, and scroll state.""" + + pane_id: str = Field(description="Pane ID (e.g. '%1')") + content: str = Field(description="Visible pane text") + cursor_x: int = Field(description="Cursor column (0-based)") + cursor_y: int = Field(description="Cursor row (0-based)") + pane_width: int = Field(description="Pane width in columns") + pane_height: int = Field(description="Pane height in rows") + pane_in_mode: bool = Field(description="True if pane is in copy-mode or view-mode") + pane_mode: str | None = Field( + default=None, description="Mode name (e.g. 'copy-mode') or None if normal" + ) + scroll_position: int | None = Field( + default=None, + description="Lines scrolled back in copy mode (None if not in copy mode)", + ) + history_size: int = Field(description="Total scrollback lines available") + title: str | None = Field(default=None, description="Pane title") + pane_current_command: str | None = Field( + default=None, description="Running command" + ) + pane_current_path: str | None = Field( + default=None, description="Current working directory" + ) + is_caller: bool | None = Field( + default=None, + description="True if this is the MCP caller's own pane", + ) + + +class ContentChangeResult(BaseModel): + """Result of waiting for any screen content change.""" + + changed: bool = Field(description="Whether the content changed before timeout") + pane_id: str = Field(description="Pane ID that was polled") + elapsed_seconds: float = Field(description="Time spent waiting in seconds") + timed_out: bool = Field(description="Whether the timeout was reached") diff --git a/src/libtmux_mcp/tools/pane_tools.py b/src/libtmux_mcp/tools/pane_tools.py index 7fb3cc71..314b6e18 100644 --- a/src/libtmux_mcp/tools/pane_tools.py +++ b/src/libtmux_mcp/tools/pane_tools.py @@ -2,8 +2,12 @@ from __future__ import annotations +import contextlib +import pathlib import re +import shlex import typing as t +import uuid from libtmux_mcp._utils import ( ANNOTATIONS_CREATE, @@ -17,10 +21,17 @@ _get_server, _resolve_pane, _resolve_session, + _resolve_window, _serialize_pane, handle_tool_errors, ) -from libtmux_mcp.models import PaneContentMatch, PaneInfo, WaitForTextResult +from libtmux_mcp.models import ( + ContentChangeResult, + PaneContentMatch, + PaneInfo, + PaneSnapshot, + WaitForTextResult, +) if t.TYPE_CHECKING: from fastmcp import FastMCP @@ -617,6 +628,632 @@ def _check() -> bool: ) +@handle_tool_errors +def snapshot_pane( + pane_id: str | None = None, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + socket_name: str | None = None, +) -> PaneSnapshot: + """Take a rich snapshot of a tmux pane: content + cursor + mode + scroll state. + + Returns everything capture_pane and get_pane_info return, plus cursor + position, copy-mode state, and scroll position — in a single call. + Use this instead of separate capture_pane + get_pane_info calls when + you need to reason about cursor location or pane mode. + + Parameters + ---------- + pane_id : str, optional + Pane ID (e.g. '%1'). + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID (e.g. '$1') for pane resolution. + window_id : str, optional + Window ID for pane resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + PaneSnapshot + Rich snapshot with content, cursor, mode, and scroll state. + """ + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + + # Fetch all metadata in a single display-message call. Use the + # printable Unicode glyph ␞ (U+241E, "SYMBOL FOR RECORD SEPARATOR") + # as the delimiter — the same choice libtmux itself uses for + # FORMAT_SEPARATOR. tmux's utf8_strvis (tmux/utf8.c) copies any + # valid UTF-8 multi-byte sequence verbatim, bypassing the vis() + # escape that turns ASCII control chars like 0x1f into literal + # "\037" in display-message output on some tmux builds. And ␞ is + # safe against the false-positive path that a tab delimiter has: + # tabs are legal (if rare) in Linux paths and could realistically + # appear in pane_current_path. + _SEP = "␞" + fmt = _SEP.join( + [ + "#{cursor_x}", + "#{cursor_y}", + "#{pane_width}", + "#{pane_height}", + "#{pane_in_mode}", + "#{pane_mode}", + "#{scroll_position}", + "#{history_size}", + "#{pane_title}", + "#{pane_current_command}", + "#{pane_current_path}", + ] + ) + result = pane.cmd("display-message", "-p", "-t", pane.pane_id, fmt) + raw = result.stdout[0] if result.stdout else "" + # Pad defensively to guarantee 11 fields even if tmux drops an + # unknown format variable on older versions. + parts = (raw.split(_SEP) + [""] * 11)[:11] + + content = "\n".join(pane.capture_pane()) + + pane_in_mode = parts[4] == "1" + pane_mode_raw = parts[5] + scroll_raw = parts[6] + + caller_pane_id = _get_caller_pane_id() + return PaneSnapshot( + pane_id=pane.pane_id or "", + content=content, + cursor_x=int(parts[0]) if parts[0] else 0, + cursor_y=int(parts[1]) if parts[1] else 0, + pane_width=int(parts[2]) if parts[2] else 0, + pane_height=int(parts[3]) if parts[3] else 0, + pane_in_mode=pane_in_mode, + pane_mode=pane_mode_raw if pane_mode_raw else None, + scroll_position=int(scroll_raw) if scroll_raw else None, + history_size=int(parts[7]) if parts[7] else 0, + title=parts[8] if parts[8] else None, + pane_current_command=parts[9] if parts[9] else None, + pane_current_path=parts[10] if parts[10] else None, + is_caller=(pane.pane_id == caller_pane_id if caller_pane_id else None), + ) + + +@handle_tool_errors +def wait_for_content_change( + pane_id: str | None = None, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + timeout: float = 8.0, + interval: float = 0.05, + socket_name: str | None = None, +) -> ContentChangeResult: + """Wait for any content change in a tmux pane. + + Captures the current pane content, then polls until the content differs + or the timeout is reached. Use this after send_keys when you don't know + what the output will be — it waits for "something happened" rather than + a specific pattern. + + Parameters + ---------- + pane_id : str, optional + Pane ID (e.g. '%1'). + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID (e.g. '$1') for pane resolution. + window_id : str, optional + Window ID for pane resolution. + timeout : float + Maximum seconds to wait. Default 8.0. + interval : float + Seconds between polls. Default 0.05 (50ms). + socket_name : str, optional + tmux socket name. + + Returns + ------- + ContentChangeResult + Result with changed status and timing info. + """ + import time + + from libtmux.test.retry import retry_until + + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + + assert pane.pane_id is not None + initial_content = pane.capture_pane() + start_time = time.monotonic() + + def _check() -> bool: + current = pane.capture_pane() + return current != initial_content + + changed = retry_until( + _check, + seconds=timeout, + interval=interval, + raises=False, + ) + + elapsed = time.monotonic() - start_time + return ContentChangeResult( + changed=changed, + pane_id=pane.pane_id, + elapsed_seconds=round(elapsed, 3), + timed_out=not changed, + ) + + +@handle_tool_errors +def select_pane( + pane_id: str | None = None, + direction: t.Literal["up", "down", "left", "right", "last", "next", "previous"] + | None = None, + window_id: str | None = None, + window_index: str | None = None, + session_name: str | None = None, + session_id: str | None = None, + socket_name: str | None = None, +) -> PaneInfo: + """Select (focus) a tmux pane by ID or direction. + + Use this to navigate between panes. Provide either pane_id for direct + selection, or direction for relative navigation within a window. + + Parameters + ---------- + pane_id : str, optional + Pane ID (e.g. '%1') for direct selection. + direction : str, optional + Relative direction: 'up', 'down', 'left', 'right', 'last' + (previously active), 'next', or 'previous'. + window_id : str, optional + Window ID for directional navigation scope. + window_index : str, optional + Window index for directional navigation scope. + session_name : str, optional + Session name for resolution. + session_id : str, optional + Session ID for resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + PaneInfo + The now-active pane. + """ + from fastmcp.exceptions import ToolError + + if pane_id is None and direction is None: + msg = "Provide either pane_id or direction." + raise ToolError(msg) + + server = _get_server(socket_name=socket_name) + + if pane_id is not None: + pane = _resolve_pane(server, pane_id=pane_id) + pane.select() + return _serialize_pane(pane) + + # Directional navigation + _DIRECTION_FLAGS: dict[str, str] = { + "up": "-U", + "down": "-D", + "left": "-L", + "right": "-R", + "last": "-l", + } + + window = _resolve_window( + server, + window_id=window_id, + window_index=window_index, + session_name=session_name, + session_id=session_id, + ) + + assert direction is not None + if direction in _DIRECTION_FLAGS: + window.select_pane(_DIRECTION_FLAGS[direction]) + elif direction in ("next", "previous"): + # Compute the target pane by absolute pane_id rather than using + # tmux's relative pane-target syntax. Two portability issues + # motivate this approach: + # 1. A bare `-t +` / `-t -1` resolves against the attached + # client's current window (tmux cmd-find.c), not the window + # we're targeting. + # 2. The scoped form `@window_id.+` / `.-` works on tmux 3.6+ + # but the relative-offset parser's behavior for prefixed + # window targets varies on older releases (tmux 3.2a still + # falls back to client curw for `@id.+`). Enumerating + # panes and selecting by absolute pane_id sidesteps + # tmux-version variation entirely. + window.refresh() + panes = list(window.panes) + active = next((p for p in panes if p.pane_active == "1"), panes[0]) + idx = panes.index(active) + step = 1 if direction == "next" else -1 + target_pane = panes[(idx + step) % len(panes)] + server.cmd("select-pane", target=target_pane.pane_id) + + # Query the active pane ID directly from tmux to avoid stale cache + target = window.window_id or "" + result = window.cmd("display-message", "-p", "-t", target, "#{pane_id}") + active_pane_id = result.stdout[0] if result.stdout else None + if active_pane_id: + active_pane = server.panes.get(pane_id=active_pane_id, default=None) + if active_pane is not None: + return _serialize_pane(active_pane) + + # Fallback + active_pane = window.active_pane + assert active_pane is not None + return _serialize_pane(active_pane) + + +@handle_tool_errors +def swap_pane( + source_pane_id: str, + target_pane_id: str, + socket_name: str | None = None, +) -> PaneInfo: + """Swap the positions of two panes. + + Exchanges the visual positions of two panes. Both panes must exist. + Use this to rearrange pane layout without changing content. + + Parameters + ---------- + source_pane_id : str + Pane ID of the first pane (e.g. '%1'). + target_pane_id : str + Pane ID of the second pane (e.g. '%2'). + socket_name : str, optional + tmux socket name. + + Returns + ------- + PaneInfo + The source pane after swap (now in target's position). + """ + server = _get_server(socket_name=socket_name) + # Validate both panes exist + source = _resolve_pane(server, pane_id=source_pane_id) + _resolve_pane(server, pane_id=target_pane_id) + + server.cmd("swap-pane", "-s", source_pane_id, "-t", target_pane_id) + source.refresh() + return _serialize_pane(source) + + +@handle_tool_errors +def pipe_pane( + pane_id: str | None = None, + output_path: str | None = None, + append: bool = True, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + socket_name: str | None = None, +) -> str: + """Start or stop piping pane output to a file. + + When output_path is given, starts logging all pane output to the file. + When output_path is None, stops any active pipe for the pane. + + Parameters + ---------- + pane_id : str, optional + Pane ID (e.g. '%1'). + output_path : str, optional + File path to write output to. None stops piping. + append : bool + Whether to append to the file. Default True. If False, overwrites. + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID for pane resolution. + window_id : str, optional + Window ID for pane resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + str + Confirmation message. + """ + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + + if output_path is None: + pane.cmd("pipe-pane") + return f"Piping stopped for pane {pane.pane_id}" + + if not output_path.strip(): + from fastmcp.exceptions import ToolError + + msg = "output_path must be a non-empty path, or None to stop piping." + raise ToolError(msg) + + redirect = ">>" if append else ">" + pane.cmd("pipe-pane", f"cat {redirect} {shlex.quote(output_path)}") + return f"Piping pane {pane.pane_id} to {output_path}" + + +@handle_tool_errors +def display_message( + format_string: str, + pane_id: str | None = None, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + socket_name: str | None = None, +) -> str: + """Query tmux using a format string. + + Expands tmux format variables against a target pane. Use this as a + generic introspection tool to query any tmux variable, e.g. + '#{window_zoomed_flag}', '#{pane_dead}', '#{client_activity}'. + + Parameters + ---------- + format_string : str + tmux format string (e.g. '#{cursor_x} #{cursor_y}'). + pane_id : str, optional + Pane ID (e.g. '%1'). + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID for pane resolution. + window_id : str, optional + Window ID for pane resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + str + Expanded format string result. + """ + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + result = pane.cmd("display-message", "-p", "-t", pane.pane_id, format_string) + return "\n".join(result.stdout) if result.stdout else "" + + +@handle_tool_errors +def enter_copy_mode( + pane_id: str | None = None, + scroll_up: int | None = None, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + socket_name: str | None = None, +) -> PaneInfo: + """Enter copy mode in a tmux pane, optionally scrolling up. + + Use to navigate scrollback history. After entering copy mode, use + snapshot_pane to read the scroll_position and content. + + Parameters + ---------- + pane_id : str, optional + Pane ID (e.g. '%1'). + scroll_up : int, optional + Number of lines to scroll up immediately after entering copy mode. + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID for pane resolution. + window_id : str, optional + Window ID for pane resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + PaneInfo + Serialized pane info. + """ + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + pane.cmd("copy-mode", "-t", pane.pane_id) + if scroll_up is not None and scroll_up > 0: + pane.cmd( + "send-keys", + "-X", + "-N", + str(scroll_up), + "scroll-up", + ) + pane.refresh() + return _serialize_pane(pane) + + +@handle_tool_errors +def exit_copy_mode( + pane_id: str | None = None, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + socket_name: str | None = None, +) -> PaneInfo: + """Exit copy mode in a tmux pane. + + Returns the pane to normal mode. Use after scrolling through + scrollback history. + + Parameters + ---------- + pane_id : str, optional + Pane ID (e.g. '%1'). + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID for pane resolution. + window_id : str, optional + Window ID for pane resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + PaneInfo + Serialized pane info. + """ + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + pane.cmd("send-keys", "-t", pane.pane_id, "-X", "cancel") + pane.refresh() + return _serialize_pane(pane) + + +@handle_tool_errors +def paste_text( + text: str, + pane_id: str | None = None, + bracket: bool = True, + session_name: str | None = None, + session_id: str | None = None, + window_id: str | None = None, + socket_name: str | None = None, +) -> str: + """Paste multi-line text into a pane using tmux paste buffers. + + Uses tmux's load-buffer and paste-buffer for clean multi-line input, + avoiding the issues of sending text line-by-line via send_keys. + Supports bracketed paste mode for terminals that handle it. + + Parameters + ---------- + text : str + The text to paste. + pane_id : str, optional + Pane ID (e.g. '%1'). + bracket : bool + Whether to use bracketed paste mode. Default True. + Bracketed paste wraps the text in escape sequences that tell + the terminal "this is pasted text, not typed input". + session_name : str, optional + Session name for pane resolution. + session_id : str, optional + Session ID for pane resolution. + window_id : str, optional + Window ID for pane resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + str + Confirmation message. + """ + import subprocess + import tempfile + + from fastmcp.exceptions import ToolError + + server = _get_server(socket_name=socket_name) + pane = _resolve_pane( + server, + pane_id=pane_id, + session_name=session_name, + session_id=session_id, + window_id=window_id, + ) + + # Use a unique named tmux buffer so we don't clobber the user's + # unnamed paste buffer, and so we can reliably clean up on error + # paths (paste-buffer -b NAME -d deletes the named buffer). + buffer_name = f"mcp_paste_{uuid.uuid4().hex}" + tmppath: str | None = None + try: + # Write text to a temp file and load into tmux buffer + # (libtmux's cmd() doesn't support stdin). + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: + tmppath = f.name # bind first so cleanup works even if write fails + f.write(text) + + # Build tmux command args for loading the named buffer + tmux_bin: str = getattr(server, "tmux_bin", None) or "tmux" + load_args: list[str] = [tmux_bin] + if server.socket_name: + load_args.extend(["-L", server.socket_name]) + if server.socket_path: + load_args.extend(["-S", str(server.socket_path)]) + load_args.extend(["load-buffer", "-b", buffer_name, tmppath]) + + try: + subprocess.run(load_args, check=True, capture_output=True) + except subprocess.CalledProcessError as e: + stderr = e.stderr.decode(errors="replace").strip() if e.stderr else "" + msg = f"load-buffer failed: {stderr or e}" + raise ToolError(msg) from e + + # Paste from the named buffer. -d deletes only that named buffer, + # leaving any unnamed user buffer intact. + paste_args = ["-b", buffer_name, "-d"] + if bracket: + paste_args.append("-p") # bracketed paste mode + paste_args.extend(["-t", pane.pane_id or ""]) + pane.cmd("paste-buffer", *paste_args) + finally: + if tmppath is not None: + pathlib.Path(tmppath).unlink(missing_ok=True) + # Defensive: the buffer should already be gone (paste-buffer -d + # deletes it), but if paste-buffer failed before -d took effect + # we leak an entry in the tmux server. Best-effort delete. + with contextlib.suppress(Exception): + server.cmd("delete-buffer", "-b", buffer_name) + + return f"Text pasted to pane {pane.pane_id}" + + def register(mcp: FastMCP) -> None: """Register pane-level tools with the MCP instance.""" mcp.tool(title="Send Keys", annotations=ANNOTATIONS_CREATE, tags={TAG_MUTATING})( @@ -648,3 +1285,36 @@ def register(mcp: FastMCP) -> None: mcp.tool(title="Wait For Text", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})( wait_for_text ) + mcp.tool(title="Snapshot Pane", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})( + snapshot_pane + ) + mcp.tool( + title="Wait For Content Change", + annotations=ANNOTATIONS_RO, + tags={TAG_READONLY}, + )(wait_for_content_change) + mcp.tool( + title="Select Pane", annotations=ANNOTATIONS_MUTATING, tags={TAG_MUTATING} + )(select_pane) + mcp.tool(title="Swap Pane", annotations=ANNOTATIONS_CREATE, tags={TAG_MUTATING})( + swap_pane + ) + mcp.tool(title="Pipe Pane", annotations=ANNOTATIONS_CREATE, tags={TAG_MUTATING})( + pipe_pane + ) + mcp.tool(title="Display Message", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})( + display_message + ) + mcp.tool( + title="Enter Copy Mode", + annotations=ANNOTATIONS_CREATE, + tags={TAG_MUTATING}, + )(enter_copy_mode) + mcp.tool( + title="Exit Copy Mode", + annotations=ANNOTATIONS_MUTATING, + tags={TAG_MUTATING}, + )(exit_copy_mode) + mcp.tool(title="Paste Text", annotations=ANNOTATIONS_CREATE, tags={TAG_MUTATING})( + paste_text + ) diff --git a/src/libtmux_mcp/tools/session_tools.py b/src/libtmux_mcp/tools/session_tools.py index 1fd1b302..b46fa78f 100644 --- a/src/libtmux_mcp/tools/session_tools.py +++ b/src/libtmux_mcp/tools/session_tools.py @@ -219,6 +219,86 @@ def kill_session( return f"Session killed: {name}" +@handle_tool_errors +def select_window( + window_id: str | None = None, + window_index: str | None = None, + direction: t.Literal["next", "previous", "last"] | None = None, + session_name: str | None = None, + session_id: str | None = None, + socket_name: str | None = None, +) -> WindowInfo: + """Select (focus) a tmux window by ID, index, or direction. + + Use to navigate between windows. Provide window_id or window_index + for direct selection, or direction for relative navigation. + + Parameters + ---------- + window_id : str, optional + Window ID (e.g. '@1') for direct selection. + window_index : str, optional + Window index for direct selection. + direction : str, optional + Relative direction: 'next', 'previous', or 'last'. + session_name : str, optional + Session name for resolution. + session_id : str, optional + Session ID for resolution. + socket_name : str, optional + tmux socket name. + + Returns + ------- + WindowInfo + The now-active window. + """ + from fastmcp.exceptions import ToolError + + if window_id is None and window_index is None and direction is None: + msg = "Provide window_id, window_index, or direction." + raise ToolError(msg) + + server = _get_server(socket_name=socket_name) + + if window_id is not None or window_index is not None: + from libtmux_mcp._utils import _resolve_window + + window = _resolve_window( + server, + window_id=window_id, + window_index=window_index, + session_name=session_name, + session_id=session_id, + ) + window.select() + return _serialize_window(window) + + # Directional navigation: use the dedicated tmux subcommands so that + # libtmux's Session.cmd injects `-t $session_id` and the navigation + # stays scoped to this session (a bare `-t +` resolves against the + # attached client, not the target session). + session = _resolve_session(server, session_name=session_name, session_id=session_id) + _CMD_MAP = { + "next": "next-window", + "previous": "previous-window", + "last": "last-window", + } + assert direction is not None + subcommand = _CMD_MAP.get(direction) + if subcommand is None: + msg = f"Invalid direction: {direction!r}. Valid: next, previous, last" + raise ToolError(msg) + proc = session.cmd(subcommand) + if proc.stderr: + stderr = " ".join(proc.stderr).strip() + msg = f"tmux {subcommand} failed: {stderr}" + raise ToolError(msg) + + active_window = session.active_window + return _serialize_window(active_window) + + def register(mcp: FastMCP) -> None: """Register session-level tools with the MCP instance.""" mcp.tool(title="List Windows", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})( @@ -235,3 +315,6 @@ def register(mcp: FastMCP) -> None: annotations=ANNOTATIONS_DESTRUCTIVE, tags={TAG_DESTRUCTIVE}, )(kill_session) + mcp.tool( + title="Select Window", annotations=ANNOTATIONS_MUTATING, tags={TAG_MUTATING} + )(select_window) diff --git a/src/libtmux_mcp/tools/window_tools.py b/src/libtmux_mcp/tools/window_tools.py index 548a04a9..8b09b543 100644 --- a/src/libtmux_mcp/tools/window_tools.py +++ b/src/libtmux_mcp/tools/window_tools.py @@ -368,6 +368,62 @@ def resize_window( return _serialize_window(window) +@handle_tool_errors +def move_window( + window_id: str | None = None, + window_index: str | None = None, + session_name: str | None = None, + session_id: str | None = None, + destination_index: str = "", + destination_session: str | None = None, + socket_name: str | None = None, +) -> WindowInfo: + """Move a window to a different index or session. + + Reorder windows within a session or move a window to another session. + + Parameters + ---------- + window_id : str, optional + Window ID (e.g. '@1'). + window_index : str, optional + Window index within the session. + session_name : str, optional + Source session name. + session_id : str, optional + Source session ID. + destination_index : str + Target window index. Default empty string (next available). + destination_session : str, optional + Target session name or ID. Default is current session. + socket_name : str, optional + tmux socket name. + + Returns + ------- + WindowInfo + Serialized window after move. + """ + server = _get_server(socket_name=socket_name) + window = _resolve_window( + server, + window_id=window_id, + window_index=window_index, + session_name=session_name, + session_id=session_id, + ) + window.move_window( + destination=destination_index, + session=destination_session, + ) + # libtmux's Window.move_window skips its own refresh when BOTH a + # non-empty destination index and a target session are passed — in + # that branch session_id stays stale. Refresh unconditionally so + # _serialize_window always reads fresh metadata. + window.refresh() + return _serialize_window(window) + + def register(mcp: FastMCP) -> None: """Register window-level tools with the MCP instance.""" mcp.tool(title="List Panes", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})( @@ -390,3 +446,6 @@ def register(mcp: FastMCP) -> None: mcp.tool( title="Resize Window", annotations=ANNOTATIONS_MUTATING, tags={TAG_MUTATING} )(resize_window) + mcp.tool( + title="Move Window", annotations=ANNOTATIONS_MUTATING, tags={TAG_MUTATING} + )(move_window) diff --git a/tests/test_pane_tools.py b/tests/test_pane_tools.py index 640ce1ca..6dabe190 100644 --- a/tests/test_pane_tools.py +++ b/tests/test_pane_tools.py @@ -6,18 +6,33 @@ import pytest from fastmcp.exceptions import ToolError +from libtmux import exc as libtmux_exc from libtmux.test.retry import retry_until -from libtmux_mcp.models import PaneContentMatch, WaitForTextResult +from libtmux_mcp.models import ( + ContentChangeResult, + PaneContentMatch, + PaneSnapshot, + WaitForTextResult, +) from libtmux_mcp.tools.pane_tools import ( capture_pane, clear_pane, + display_message, + enter_copy_mode, + exit_copy_mode, get_pane_info, kill_pane, + paste_text, + pipe_pane, resize_pane, search_panes, + select_pane, send_keys, set_pane_title, + snapshot_pane, + swap_pane, + wait_for_content_change, wait_for_text, ) @@ -507,3 +522,496 @@ def test_wait_for_text_invalid_regex(mcp_server: Server, mcp_pane: Pane) -> None pane_id=mcp_pane.pane_id, socket_name=mcp_server.socket_name, ) + + +# --------------------------------------------------------------------------- +# snapshot_pane tests +# --------------------------------------------------------------------------- + + +def test_snapshot_pane(mcp_server: Server, mcp_pane: Pane) -> None: + """snapshot_pane returns rich metadata alongside content.""" + result = snapshot_pane( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert isinstance(result, PaneSnapshot) + assert result.pane_id == mcp_pane.pane_id + assert isinstance(result.content, str) + assert result.cursor_x >= 0 + assert result.cursor_y >= 0 + assert result.pane_width > 0 + assert result.pane_height > 0 + assert result.pane_in_mode is False + assert result.pane_mode is None + assert result.history_size >= 0 + + +def test_snapshot_pane_cursor_moves(mcp_server: Server, mcp_pane: Pane) -> None: + """snapshot_pane reflects cursor position changes.""" + mcp_pane.send_keys("echo hello_snapshot", enter=True) + retry_until( + lambda: "hello_snapshot" in "\n".join(mcp_pane.capture_pane()), + 2, + raises=True, + ) + + result = snapshot_pane( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert "hello_snapshot" in result.content + assert result.pane_current_command is not None + + +def test_snapshot_pane_pads_short_display_message_output( + mcp_server: Server, mcp_pane: Pane, monkeypatch: pytest.MonkeyPatch +) -> None: + """snapshot_pane survives a truncated display-message result. + + Older tmux versions may drop unknown format variables (e.g. + `#{pane_mode}`), producing fewer delimited fields than expected. + Defensive padding must guarantee 11 fields so index access in the + parser never raises IndexError. + """ + # Capture the real cmd so non-display-message calls still work. + real_cmd = mcp_pane.__class__.cmd + + def fake_cmd(self, cmd_name, *args, **kwargs): # type: ignore[no-untyped-def] + result = real_cmd(self, cmd_name, *args, **kwargs) + if cmd_name == "display-message": + # Return only the first 2 fields (cursor_x, cursor_y) — + # simulate an old tmux that dropped several unknown format + # variables. Without defensive padding, parts[2..10] would + # IndexError. + parts = result.stdout[0].split("␞") if result.stdout else [""] + result.stdout = ["␞".join(parts[:2])] + return result + + monkeypatch.setattr(mcp_pane.__class__, "cmd", fake_cmd) + + # Must not raise IndexError; missing fields default to zero/None. + result = snapshot_pane( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert isinstance(result, PaneSnapshot) + assert result.pane_width == 0 + assert result.pane_height == 0 + assert result.history_size == 0 + assert result.title is None + assert result.pane_current_command is None + assert result.pane_current_path is None + + +# --------------------------------------------------------------------------- +# wait_for_content_change tests +# --------------------------------------------------------------------------- + + +def test_wait_for_content_change_detects_change( + mcp_server: Server, mcp_pane: Pane +) -> None: + """wait_for_content_change detects screen changes.""" + import threading + + # Send a command after a brief delay to trigger a change + def _send_later() -> None: + import time + + time.sleep(0.2) + mcp_pane.send_keys("echo CHANGE_DETECTED_xyz", enter=True) + + thread = threading.Thread(target=_send_later) + thread.start() + + result = wait_for_content_change( + pane_id=mcp_pane.pane_id, + timeout=3.0, + socket_name=mcp_server.socket_name, + ) + thread.join() + assert isinstance(result, ContentChangeResult) + assert result.changed is True + assert result.timed_out is False + assert result.elapsed_seconds > 0 + + +def test_wait_for_content_change_timeout(mcp_server: Server, mcp_pane: Pane) -> None: + """wait_for_content_change times out when no change occurs.""" + # Wait for the shell prompt to settle before testing for "no change" + import time + + time.sleep(0.5) + + result = wait_for_content_change( + pane_id=mcp_pane.pane_id, + timeout=0.5, + socket_name=mcp_server.socket_name, + ) + assert isinstance(result, ContentChangeResult) + assert result.changed is False + assert result.timed_out is True + + +# --------------------------------------------------------------------------- +# select_pane tests +# --------------------------------------------------------------------------- + + +def test_select_pane_by_id(mcp_server: Server, mcp_session: Session) -> None: + """select_pane focuses a specific pane by ID.""" + window = mcp_session.active_window + pane1 = window.active_pane + assert pane1 is not None + window.split() + + # Select the first pane + result = select_pane( + pane_id=pane1.pane_id, + socket_name=mcp_server.socket_name, + ) + assert result.pane_id == pane1.pane_id + + +def test_select_pane_directional(mcp_server: Server, mcp_session: Session) -> None: + """select_pane navigates using direction.""" + window = mcp_session.active_window + pane1 = window.active_pane + assert pane1 is not None + pane2 = window.split() # creates pane below; pane1 stays active + + # pane1 is active, select "down" should go to pane2 + result = select_pane( + direction="down", + window_id=window.window_id, + socket_name=mcp_server.socket_name, + ) + assert result.pane_id == pane2.pane_id + + +def test_select_pane_requires_target(mcp_server: Server) -> None: + """select_pane raises ToolError when neither pane_id nor direction given.""" + with pytest.raises(ToolError, match="Provide either"): + select_pane(socket_name=mcp_server.socket_name) + + +def test_select_pane_next_previous_respects_target_window( + mcp_server: Server, mcp_session: Session +) -> None: + """select_pane direction=next/previous must anchor to window_id. + + Regression guard: bare `-t +1` / `-t -1` pane targets resolve + against the attached client's current window (tmux cmd-find.c), + not against any earlier -t on the command line. Targeting a + non-active window must use a window-scoped syntax like + `@window_id.+` to actually affect that window. Without the fix, + calling select_pane(direction='next', window_id=w2) when w1 is + the client's active window shifts focus in w1 and leaves w2 + untouched. + """ + w1 = mcp_session.active_window + assert w1.active_pane is not None + w1.split() + w1.split() + w2 = mcp_session.new_window() + w2.split() + w2.split() + + # Make w1 the active window again, so w2 is the NON-active target. + w1.select() + w1.refresh() + w2.refresh() + + w1_before = w1.active_pane.pane_id + assert w2.active_pane is not None + w2_before = w2.active_pane.pane_id + + result = select_pane( + direction="next", + window_id=w2.window_id, + socket_name=mcp_server.socket_name, + ) + + w1.refresh() + w2.refresh() + assert w2.active_pane is not None + w2_after = w2.active_pane.pane_id + assert w1.active_pane is not None + w1_after = w1.active_pane.pane_id + + # Result must describe a pane in w2 (the target), not w1. + w2_pane_ids = {p.pane_id for p in w2.panes} + assert result.pane_id in w2_pane_ids, ( + f"select_pane returned {result.pane_id} which is not in target " + f"window {w2.window_id}'s panes {w2_pane_ids}" + ) + # w2's active pane must have actually changed. + assert w2_after != w2_before, "target window w2's active pane did not change" + # w1's active pane must NOT have changed — the wrong-window bug. + assert w1_after == w1_before, ( + f"select_pane targeting w2 shifted focus in w1 " + f"({w1_before} -> {w1_after}) — anchor missing" + ) + + +# --------------------------------------------------------------------------- +# swap_pane tests +# --------------------------------------------------------------------------- + + +def test_swap_pane(mcp_server: Server, mcp_session: Session) -> None: + """swap_pane exchanges two pane positions.""" + window = mcp_session.active_window + pane1 = window.active_pane + assert pane1 is not None + pane2 = window.split() + + assert pane1.pane_id is not None + assert pane2.pane_id is not None + + result = swap_pane( + source_pane_id=pane1.pane_id, + target_pane_id=pane2.pane_id, + socket_name=mcp_server.socket_name, + ) + assert result.pane_id == pane1.pane_id + + +# --------------------------------------------------------------------------- +# pipe_pane tests +# --------------------------------------------------------------------------- + + +def test_pipe_pane_start_stop( + mcp_server: Server, mcp_pane: Pane, tmp_path: t.Any +) -> None: + """pipe_pane starts writes after start and halts writes after stop.""" + log_file = tmp_path / "pane_output.log" + + result = pipe_pane( + pane_id=mcp_pane.pane_id, + output_path=str(log_file), + socket_name=mcp_server.socket_name, + ) + assert "piping" in result.lower() + + mcp_pane.send_keys("echo START_MARKER_42", enter=True) + retry_until( + lambda: log_file.exists() and "START_MARKER_42" in log_file.read_text(), + 2, + raises=True, + ) + + result = pipe_pane( + pane_id=mcp_pane.pane_id, + output_path=None, + socket_name=mcp_server.socket_name, + ) + assert "stopped" in result.lower() + + size_after_stop = log_file.stat().st_size + mcp_pane.send_keys("echo POST_STOP_MARKER_99", enter=True) + # Poll briefly — if stop worked the file must not grow. + with pytest.raises(libtmux_exc.WaitTimeout): + retry_until( + lambda: log_file.stat().st_size > size_after_stop, + 1, + raises=True, + ) + assert "POST_STOP_MARKER_99" not in log_file.read_text() + + +def test_pipe_pane_quotes_path_with_spaces( + mcp_server: Server, mcp_pane: Pane, tmp_path: t.Any +) -> None: + """pipe_pane survives an output_path containing spaces. + + Without shell-quoting the path, tmux runs `cat >> /tmp/has space.log` + which the shell splits into two arguments — the redirect silently + lands on `/tmp/has` and `space.log` becomes a literal cat argument. + """ + log_file = tmp_path / "has space.log" + marker = "PIPE_PANE_MARKER_42" + + result = pipe_pane( + pane_id=mcp_pane.pane_id, + output_path=str(log_file), + socket_name=mcp_server.socket_name, + ) + assert "piping" in result.lower() + + try: + mcp_pane.send_keys(f"echo {marker}", enter=True) + retry_until( + lambda: log_file.exists() and marker in log_file.read_text(), + 2, + raises=True, + ) + finally: + pipe_pane( + pane_id=mcp_pane.pane_id, + output_path=None, + socket_name=mcp_server.socket_name, + ) + + +def test_pipe_pane_rejects_empty_path(mcp_server: Server, mcp_pane: Pane) -> None: + """pipe_pane raises ToolError when output_path is empty or whitespace.""" + for bad in ("", " ", "\t"): + with pytest.raises(ToolError, match="non-empty"): + pipe_pane( + pane_id=mcp_pane.pane_id, + output_path=bad, + socket_name=mcp_server.socket_name, + ) + + +# --------------------------------------------------------------------------- +# display_message tests +# --------------------------------------------------------------------------- + + +def test_display_message(mcp_server: Server, mcp_pane: Pane) -> None: + """display_message expands tmux format strings.""" + result = display_message( + format_string="#{pane_width}x#{pane_height}", + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert "x" in result + parts = result.split("x") + assert len(parts) == 2 + assert parts[0].isdigit() + assert parts[1].isdigit() + + +def test_display_message_zoomed_flag(mcp_server: Server, mcp_session: Session) -> None: + """display_message queries arbitrary tmux variables.""" + window = mcp_session.active_window + pane = window.active_pane + assert pane is not None + result = display_message( + format_string="#{window_zoomed_flag}", + pane_id=pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert result in ("0", "1") + + +# --------------------------------------------------------------------------- +# enter_copy_mode / exit_copy_mode tests +# --------------------------------------------------------------------------- + + +def test_enter_and_exit_copy_mode(mcp_server: Server, mcp_pane: Pane) -> None: + """enter_copy_mode enters copy mode, exit_copy_mode leaves it.""" + enter_result = enter_copy_mode( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert enter_result.pane_id == mcp_pane.pane_id + + # Verify pane is in copy mode via snapshot + snap = snapshot_pane( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert snap.pane_in_mode is True + + exit_result = exit_copy_mode( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + assert exit_result.pane_id == mcp_pane.pane_id + + +def test_enter_copy_mode_with_scroll(mcp_server: Server, mcp_pane: Pane) -> None: + """enter_copy_mode can scroll up immediately.""" + # Generate some scrollback history + for i in range(20): + mcp_pane.send_keys(f"echo scrollback_line_{i}", enter=True) + retry_until( + lambda: "scrollback_line_19" in "\n".join(mcp_pane.capture_pane()), + 2, + raises=True, + ) + + enter_result = enter_copy_mode( + pane_id=mcp_pane.pane_id, + scroll_up=5, + socket_name=mcp_server.socket_name, + ) + assert enter_result.pane_id == mcp_pane.pane_id + + # Clean up: exit copy mode + exit_copy_mode( + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + + +# --------------------------------------------------------------------------- +# paste_text tests +# --------------------------------------------------------------------------- + + +def test_paste_text(mcp_server: Server, mcp_pane: Pane) -> None: + """paste_text pastes text into a pane via tmux buffer. + + Uses bracket=False and a trailing newline so the shell actually + executes the echo command. Previous versions of this test + relied on the default bracket=True, which is fragile on CI: + bash readline needs a prompt cycle to latch bracketed-paste + mode, and if the paste arrives before that the escape sequences + get consumed as unrecognized input and the marker never reaches + the visible pane buffer. bracket=False sends raw bytes and the + trailing newline forces execution, exercising the full + paste->execute->output round-trip. + """ + result = paste_text( + text="echo PASTE_TEST_marker_xyz\n", + pane_id=mcp_pane.pane_id, + bracket=False, + socket_name=mcp_server.socket_name, + ) + assert "pasted" in result.lower() + + # Verify the echoed marker reaches the pane. 10 seconds is + # generous on local machines (<1s) but tolerates slow CI + # runners where bash cold-start can exceed the default budget. + retry_until( + lambda: "PASTE_TEST_marker_xyz" in "\n".join(mcp_pane.capture_pane()), + 10, + raises=True, + ) + + +def test_paste_text_does_not_leak_named_buffer( + mcp_server: Server, mcp_pane: Pane +) -> None: + """paste_text must not leave its mcp_paste_* buffer behind. + + Regression guard for the pre-fix behavior: the earlier + implementation used tmux's default unnamed buffer AND relied on + `paste-buffer -d` to clean up. If paste-buffer failed mid-flight + the buffer leaked. The fix generates a unique `mcp_paste_` + named buffer per call and adds a best-effort `delete-buffer -b` + in `finally` so the server is left in a clean state on both + success and failure paths. + + The check is portable across every tmux version the CI matrix + tests (3.2a through master): list-buffers with a format string + returns buffer names without any version-specific behavior. + """ + paste_text( + text="echo BUFFER_ISOLATION_test", + pane_id=mcp_pane.pane_id, + socket_name=mcp_server.socket_name, + ) + + listing = mcp_server.cmd("list-buffers", "-F", "#{buffer_name}") + buffer_names = "\n".join(listing.stdout or []) + assert "mcp_paste_" not in buffer_names, ( + f"paste_text leaked a named buffer: {buffer_names!r}" + ) diff --git a/tests/test_session_tools.py b/tests/test_session_tools.py index e3d6b39a..8eb662a3 100644 --- a/tests/test_session_tools.py +++ b/tests/test_session_tools.py @@ -12,6 +12,7 @@ kill_session, list_windows, rename_session, + select_window, ) if t.TYPE_CHECKING: @@ -192,6 +193,78 @@ def test_list_windows_with_filters( second_session.kill() +# --------------------------------------------------------------------------- +# select_window tests +# --------------------------------------------------------------------------- + + +def test_select_window_by_id(mcp_server: Server, mcp_session: Session) -> None: + """select_window focuses a window by ID.""" + win1 = mcp_session.active_window + mcp_session.new_window(window_name="select_target") + + result = select_window( + window_id=win1.window_id, + socket_name=mcp_server.socket_name, + ) + assert result.window_id == win1.window_id + + +def test_select_window_by_index(mcp_server: Server, mcp_session: Session) -> None: + """select_window focuses a window by index.""" + win1 = mcp_session.active_window + mcp_session.new_window(window_name="select_idx") + + result = select_window( + window_index=win1.window_index, + session_name=mcp_session.session_name, + socket_name=mcp_server.socket_name, + ) + assert result.window_id == win1.window_id + + +def test_select_window_direction_next(mcp_server: Server, mcp_session: Session) -> None: + """select_window navigates to next window.""" + win1 = mcp_session.active_window + win2 = mcp_session.new_window(window_name="next_win") + + # Make win1 active + win1.select() + result = select_window( + direction="next", + session_name=mcp_session.session_name, + socket_name=mcp_server.socket_name, + ) + assert result.window_id == win2.window_id + + +def test_select_window_requires_target(mcp_server: Server) -> None: + """select_window raises ToolError without target or direction.""" + with pytest.raises(ToolError, match="Provide"): + select_window(socket_name=mcp_server.socket_name) + + +def test_select_window_last_on_single_window_session_raises( + mcp_server: Server, mcp_session: Session +) -> None: + """select_window last with no prior window must surface the tmux error. + + Regression guard: session.cmd("last-window") on a session that has + never had a previously-active window emits "no last window" on + stderr, but the tool previously discarded the return value and + returned the unchanged active window as if the navigation had + worked. + """ + # The fixture session is freshly created: there is no previously- + # active window for last-window to jump back to. + with pytest.raises(ToolError, match="last-window"): + select_window( + direction="last", + session_name=mcp_session.session_name, + socket_name=mcp_server.socket_name, + ) + + def test_kill_session_requires_target(mcp_server: Server) -> None: """kill_session refuses to kill without an explicit target.""" with pytest.raises(ToolError, match="Refusing to kill"): diff --git a/tests/test_window_tools.py b/tests/test_window_tools.py index a0ff002f..64c7b9b6 100644 --- a/tests/test_window_tools.py +++ b/tests/test_window_tools.py @@ -10,6 +10,7 @@ from libtmux_mcp.tools.window_tools import ( kill_window, list_panes, + move_window, rename_window, resize_window, select_layout, @@ -202,6 +203,74 @@ def test_list_panes_with_filters( assert len(result) >= expected_min_count +# --------------------------------------------------------------------------- +# move_window tests +# --------------------------------------------------------------------------- + + +def test_move_window_reorder(mcp_server: Server, mcp_session: Session) -> None: + """move_window changes a window's index.""" + win = mcp_session.new_window(window_name="move_me") + result = move_window( + window_id=win.window_id, + destination_index="99", + socket_name=mcp_server.socket_name, + ) + assert result.window_id == win.window_id + assert result.window_index == "99" + + +def test_move_window_to_another_session( + mcp_server: Server, mcp_session: Session +) -> None: + """move_window moves a window to a different session.""" + target_session = mcp_server.new_session(session_name="move_target") + win = mcp_session.new_window(window_name="move_cross") + window_id = win.window_id + + result = move_window( + window_id=window_id, + destination_session=target_session.session_id, + socket_name=mcp_server.socket_name, + ) + assert result.window_id == window_id + # Proof the move actually happened: the returned session_id matches + # the destination, and the window no longer lives in the source. + assert result.session_id == target_session.session_id + source_window_ids = {w.window_id for w in mcp_session.windows} + assert window_id not in source_window_ids + + # Cleanup + target_session.kill() + + +def test_move_window_to_another_session_with_index( + mcp_server: Server, mcp_session: Session +) -> None: + """Cross-session move with an explicit destination_index refreshes metadata. + + libtmux's Window.move_window skips its own refresh when BOTH a + non-empty destination index and a target session are provided. The + tool must refresh explicitly, otherwise the returned session_id + would be the pre-move (source) value. + """ + target_session = mcp_server.new_session(session_name="move_target_indexed") + win = mcp_session.new_window(window_name="move_cross_idx") + window_id = win.window_id + + result = move_window( + window_id=window_id, + destination_index="7", + destination_session=target_session.session_id, + socket_name=mcp_server.socket_name, + ) + assert result.window_id == window_id + assert result.window_index == "7" + assert result.session_id == target_session.session_id + + target_session.kill() + + def test_kill_window_requires_window_id(mcp_server: Server) -> None: """kill_window requires window_id as a positional argument.""" with pytest.raises(ToolError, match="missing 1 required positional argument"):