From fb2c816eff6c120f0f3c3242bb55446026535b88 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Wed, 10 Jun 2026 10:13:03 -0700 Subject: [PATCH 1/3] Add zellij multiplexer support to agent-cli dev Zellij >= 0.44.0 gained stable tab IDs, initial commands for new tabs, and detached-session control, enabling tmux-parity support: - New Multiplexer ABC (shared session naming, open_in_session, attach_command); Tmux and Zellij both implement it - --multiplexer zellij on dev new/agent: creates or reuses a detached repo-scoped session when outside zellij - --zellij-session flag mirroring --tmux-session - Worktree cleanup closes zellij tabs by matching pane cwd, across all live sessions, skipping the current tab - Legacy write-chars tab opening kept for zellij < 0.44 inside a session - Evidence-backed verification tests (docs/PRs + live probes on 0.44.3) --- agent_cli/dev/cleanup.py | 10 +- agent_cli/dev/cli.py | 109 +++++++---- agent_cli/dev/launch.py | 70 +++---- agent_cli/dev/terminals/__init__.py | 3 +- agent_cli/dev/terminals/base.py | 43 ++++ agent_cli/dev/terminals/tmux.py | 26 +-- agent_cli/dev/terminals/zellij.py | 290 ++++++++++++++++++++++++++- docs/commands/dev.md | 6 +- tests/dev/test_cleanup.py | 36 +++- tests/dev/test_cli.py | 85 +++++++- tests/dev/test_launch.py | 130 +++++++++++- tests/dev/test_terminals.py | 294 +++++++++++++++++++++++++++- tests/dev/test_verification.py | 102 +++++++++- 13 files changed, 1078 insertions(+), 126 deletions(-) diff --git a/agent_cli/dev/cleanup.py b/agent_cli/dev/cleanup.py index ce54b0740..ea41fdd03 100644 --- a/agent_cli/dev/cleanup.py +++ b/agent_cli/dev/cleanup.py @@ -9,6 +9,7 @@ from . import worktree from .terminals.tmux import Tmux +from .terminals.zellij import Zellij if TYPE_CHECKING: from pathlib import Path @@ -16,7 +17,7 @@ @dataclass class RemoveWorktreeResult: - """Outcome of removing a worktree and any tagged tmux windows.""" + """Outcome of removing a worktree and any multiplexer tabs/windows it owned.""" name: str success: bool @@ -130,7 +131,7 @@ def remove_worktree( force: bool = False, delete_branch: bool = False, ) -> RemoveWorktreeResult: - """Remove one worktree and then clean up any tagged tmux windows.""" + """Remove one worktree and then clean up any tmux windows and zellij tabs it owned.""" removed, error = worktree.remove_worktree( wt.path, force=force, @@ -145,7 +146,8 @@ def remove_worktree( if not removed: return result - tmux = Tmux() - tmux_cleanup = tmux.kill_windows_for_worktree(wt.path) + tmux_cleanup = Tmux().kill_windows_for_worktree(wt.path) result.warnings.extend(tmux_cleanup.errors) + zellij_cleanup = Zellij().close_tabs_for_worktree(wt.path) + result.warnings.extend(zellij_cleanup.errors) return result diff --git a/agent_cli/dev/cli.py b/agent_cli/dev/cli.py index bf20a36cc..6016ae79b 100644 --- a/agent_cli/dev/cli.py +++ b/agent_cli/dev/cli.py @@ -4,7 +4,6 @@ import json import os -import shlex import shutil import subprocess from pathlib import Path @@ -164,21 +163,44 @@ def _resolve_prompt_text( return prompt -def _normalize_tmux_session( +def _normalize_multiplexer_session( tmux_session: str | None, - multiplexer: Literal["tmux"] | None, -) -> tuple[str | None, Literal["tmux"] | None]: - """Normalize `--tmux-session` and make it imply tmux launches.""" - if tmux_session is None: - return None, multiplexer - - normalized_session = tmux_session.strip() - if not normalized_session: - error("--tmux-session cannot be empty") - if "." in normalized_session or ":" in normalized_session: - error("tmux session names cannot contain '.' or ':'") - - return normalized_session, "tmux" + zellij_session: str | None, + multiplexer: Literal["tmux", "zellij"] | None, +) -> tuple[str | None, Literal["tmux", "zellij"] | None]: + """Normalize `--tmux-session`/`--zellij-session` and make them imply a multiplexer.""" + if tmux_session is not None and zellij_session is not None: + error("Cannot use --tmux-session and --zellij-session together") + + if tmux_session is not None: + if multiplexer == "zellij": + error("--tmux-session cannot be combined with --multiplexer zellij") + normalized_session = tmux_session.strip() + if not normalized_session: + error("--tmux-session cannot be empty") + if "." in normalized_session or ":" in normalized_session: + error("tmux session names cannot contain '.' or ':'") + return normalized_session, "tmux" + + if zellij_session is not None: + if multiplexer == "tmux": + error("--zellij-session cannot be combined with --multiplexer tmux") + normalized_session = zellij_session.strip() + if not normalized_session: + error("--zellij-session cannot be empty") + return normalized_session, "zellij" + + return None, multiplexer + + +def _attach_hint(handle: terminals.TerminalHandle) -> str | None: + """Shell command to attach to the session holding a launched agent, if any.""" + if handle.session_name is None: + return None + terminal = terminals.get_terminal(handle.terminal_name) + if isinstance(terminal, terminals.Multiplexer): + return terminal.attach_command(handle.session_name) + return None def _resolve_dev_new_agent_request( @@ -457,12 +479,12 @@ def new( ), ] = None, multiplexer: Annotated[ - Literal["tmux"] | None, + Literal["tmux", "zellij"] | None, typer.Option( "--multiplexer", "-m", case_sensitive=False, - help="Launch the agent in a specific multiplexer. Currently supported: tmux. When started outside tmux, creates or reuses a detached session and reports the pane handle", + help="Launch the agent in a specific multiplexer. Currently supported: tmux, zellij (zellij requires >= 0.44.0). When started outside the multiplexer, creates or reuses a detached session and reports the tab/pane handle", ), ] = None, tmux_session: Annotated[ @@ -472,6 +494,13 @@ def new( help="Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux", ), ] = None, + zellij_session: Annotated[ + str | None, + typer.Option( + "--zellij-session", + help="Reuse or create a specific zellij session for the agent. Implies --multiplexer zellij", + ), + ] = None, hooks: Annotated[ bool, typer.Option( @@ -521,7 +550,11 @@ def new( agent_name_deprecated=agent_name_deprecated, prompt=prompt, ) - tmux_session, multiplexer = _normalize_tmux_session(tmux_session, multiplexer) + multiplexer_session, multiplexer = _normalize_multiplexer_session( + tmux_session, + zellij_session, + multiplexer, + ) repo_root = _ensure_git_repo() runtime_config = _runtime_config_from_ctx(ctx) @@ -606,7 +639,7 @@ def new( task_file, agent_env, multiplexer_name=multiplexer, - tmux_session=tmux_session, + multiplexer_session=multiplexer_session, ) # Print summary @@ -618,13 +651,15 @@ def new( summary_lines.append( f"[bold]Agent Handle:[/bold] {agent_handle.handle} ({agent_handle.terminal_name})", ) - if agent_handle.session_name and agent_handle.terminal_name == "tmux": - summary_lines.append(f"[bold]tmux Session:[/bold] {agent_handle.session_name}") + if agent_handle.session_name and agent_handle.terminal_name == "cmux": + summary_lines.append(f"[bold]cmux Workspace:[/bold] {agent_handle.session_name}") + elif agent_handle.session_name: summary_lines.append( - f"[bold]Attach:[/bold] tmux attach -t {shlex.quote(agent_handle.session_name)}", + f"[bold]{agent_handle.terminal_name} Session:[/bold] {agent_handle.session_name}", ) - elif agent_handle.session_name and agent_handle.terminal_name == "cmux": - summary_lines.append(f"[bold]cmux Workspace:[/bold] {agent_handle.session_name}") + attach = _attach_hint(agent_handle) + if attach: + summary_lines.append(f"[bold]Attach:[/bold] {attach}") console.print() console.print( @@ -1051,12 +1086,12 @@ def start_agent( ), ] = None, multiplexer: Annotated[ - Literal["tmux"] | None, + Literal["tmux", "zellij"] | None, typer.Option( "--multiplexer", "-m", case_sensitive=False, - help="Launch the agent in a specific multiplexer instead of the current terminal. Currently supported: tmux", + help="Launch the agent in a specific multiplexer instead of the current terminal. Currently supported: tmux, zellij (zellij requires >= 0.44.0)", ), ] = None, tmux_session: Annotated[ @@ -1066,6 +1101,13 @@ def start_agent( help="Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux", ), ] = None, + zellij_session: Annotated[ + str | None, + typer.Option( + "--zellij-session", + help="Reuse or create a specific zellij session for the agent. Implies --multiplexer zellij", + ), + ] = None, hooks: Annotated[ bool, typer.Option( @@ -1091,7 +1133,11 @@ def start_agent( agent_name = agent_name or agent_name_deprecated prompt = _resolve_prompt_text(prompt, prompt_file=prompt_file) - tmux_session, multiplexer = _normalize_tmux_session(tmux_session, multiplexer) + multiplexer_session, multiplexer = _normalize_multiplexer_session( + tmux_session, + zellij_session, + multiplexer, + ) repo_root = _ensure_git_repo() runtime_config = _runtime_config_from_ctx(ctx) @@ -1141,16 +1187,13 @@ def start_agent( task_file, agent_env, multiplexer_name=multiplexer, - tmux_session=tmux_session, + multiplexer_session=multiplexer_session, ) if handle: + attach = _attach_hint(handle) info( f"{handle.terminal_name} handle: {handle.handle}" - + ( - f" (attach with: tmux attach -t {shlex.quote(handle.session_name)})" - if handle.session_name and handle.terminal_name == "tmux" - else "" - ), + + (f" (attach with: {attach})" if attach else ""), ) return diff --git a/agent_cli/dev/launch.py b/agent_cli/dev/launch.py index 11ebde79a..301cd4ff0 100644 --- a/agent_cli/dev/launch.py +++ b/agent_cli/dev/launch.py @@ -280,45 +280,27 @@ def _tab_name_for_path(path: Path) -> tuple[Path | None, str]: return repo_root, tab_name -def _launch_in_tmux( +def _launch_in_multiplexer( path: Path, - agent: CodingAgent, - terminal: terminals.Terminal, + terminal: terminals.Multiplexer, full_cmd: str, tab_name: str, repo_root: Path | None, - multiplexer_name: str | None, - tmux_session: str | None, + *, + requested: bool, + session_override: str | None, ) -> TerminalHandle | None: - """Launch an agent via tmux and return its pane handle.""" - from .terminals.tmux import Tmux # noqa: PLC0415 - - if not isinstance(terminal, Tmux): - warn("Could not open new tab in tmux") - return None - - requested_tmux = multiplexer_name == "tmux" - session_name = tmux_session - if session_name is None and requested_tmux and not terminal.detect(): + """Launch an agent in a multiplexer and return its tab/pane handle.""" + session_name = session_override + if session_name is None and requested and not terminal.detect(): session_name = terminal.session_name_for_repo(repo_root or path) - handle = terminal.open_in_session( + return terminal.open_in_session( path, full_cmd, tab_name=tab_name, session_name=session_name, ) - if handle is None: - warn("Could not open new tab in tmux") - return None - - session_label = ( - f" in tmux session {handle.session_name}" - if (requested_tmux or tmux_session is not None) and handle.session_name - else " in new tmux tab" - ) - success(f"Started {agent.name}{session_label}") - return handle def _launch_in_cmux( @@ -359,25 +341,36 @@ def _launch_in_terminal( tab_name: str, repo_root: Path | None, multiplexer_name: str | None, - tmux_session: str | None, + multiplexer_session: str | None, ) -> tuple[bool, TerminalHandle | None]: """Launch an agent in the resolved terminal.""" if terminal.name == "cmux": handle = _launch_in_cmux(path, agent, terminal, full_cmd, tab_name, repo_root) return handle is not None, handle - if terminal.name == "tmux": - handle = _launch_in_tmux( + if isinstance(terminal, terminals.Multiplexer): + requested = multiplexer_name == terminal.name or multiplexer_session is not None + handle = _launch_in_multiplexer( path, - agent, terminal, full_cmd, tab_name, repo_root, - multiplexer_name, - tmux_session, + requested=requested, + session_override=multiplexer_session, ) - return handle is not None, handle + if handle is not None: + session_label = ( + f" in {terminal.name} session {handle.session_name}" + if requested and handle.session_name + else f" in new {terminal.name} tab" + ) + success(f"Started {agent.name}{session_label}") + return True, handle + if requested: + warn(f"Could not open new tab in {terminal.name}") + return False, None + # Fall through to plain tab opening (e.g. zellij < 0.44 inside a session) if terminal.open_new_tab(path, full_cmd, tab_name=tab_name): success(f"Started {agent.name} in new {terminal.name} tab") @@ -395,15 +388,14 @@ def launch_agent( task_file: Path | None = None, env: dict[str, str] | None = None, multiplexer_name: str | None = None, - tmux_session: str | None = None, + multiplexer_session: str | None = None, ) -> TerminalHandle | None: """Launch agent in a new terminal tab. Agents are interactive TUIs that need a proper terminal. Priority: tmux/zellij tab > terminal tab > print instructions. """ - effective_multiplexer_name = "tmux" if tmux_session is not None else multiplexer_name - terminal = _resolve_launch_terminal(effective_multiplexer_name) + terminal = _resolve_launch_terminal(multiplexer_name) full_cmd = _build_agent_launch_command( path, agent, extra_args, prompt, task_file, env, terminal ) @@ -417,8 +409,8 @@ def launch_agent( full_cmd, tab_name, repo_root, - effective_multiplexer_name, - tmux_session, + multiplexer_name, + multiplexer_session, ) if launched: return handle diff --git a/agent_cli/dev/terminals/__init__.py b/agent_cli/dev/terminals/__init__.py index d69f5e8e1..627ed7ed8 100644 --- a/agent_cli/dev/terminals/__init__.py +++ b/agent_cli/dev/terminals/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations -from .base import Terminal, TerminalHandle +from .base import Multiplexer, Terminal, TerminalHandle from .registry import ( detect_current_terminal, get_all_terminals, @@ -11,6 +11,7 @@ ) __all__ = [ + "Multiplexer", "Terminal", "TerminalHandle", "detect_current_terminal", diff --git a/agent_cli/dev/terminals/base.py b/agent_cli/dev/terminals/base.py index e47882665..47d9157eb 100644 --- a/agent_cli/dev/terminals/base.py +++ b/agent_cli/dev/terminals/base.py @@ -2,12 +2,15 @@ from __future__ import annotations +import hashlib import os +import re from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TYPE_CHECKING if TYPE_CHECKING: + import subprocess from pathlib import Path @@ -61,6 +64,46 @@ def __repr__(self) -> str: # noqa: D105 return f"<{self.__class__.__name__} {self.name!r} ({status})>" +class Multiplexer(Terminal): + """Terminal multiplexer that can host named sessions with addressable tabs/windows.""" + + def session_name_for_repo(self, repo_root: Path) -> str: + """Build a deterministic session name for a repo (safe for tmux and zellij).""" + repo_slug = re.sub(r"[^A-Za-z0-9_-]+", "-", repo_root.name).strip("-") or "repo" + repo_hash = hashlib.sha256(str(repo_root).encode()).hexdigest()[:8] + return f"agent-cli-{repo_slug[:24]}-{repo_hash}" + + @abstractmethod + def current_session_name(self) -> str | None: + """Get the session name this process is running inside, when available.""" + + @abstractmethod + def attach_command(self, session_name: str) -> str: + """Shell command a user can run to attach to a session.""" + + @abstractmethod + def open_in_session( + self, + path: Path, + command: str | None = None, + tab_name: str | None = None, + *, + session_name: str | None = None, + ) -> TerminalHandle | None: + """Open a tab/window and return its handle. + + If ``session_name`` is omitted, the current session is used. + When a named session does not exist yet, it is created in detached mode. + """ + + +def subprocess_error_text(exc: subprocess.CalledProcessError) -> str: + """Extract a useful stderr/stdout payload from a subprocess error.""" + stderr = exc.stderr.strip() if exc.stderr else "" + stdout = exc.stdout.strip() if exc.stdout else "" + return stderr or stdout or str(exc) + + def _get_term_program() -> str | None: """Get the TERM_PROGRAM environment variable.""" return os.environ.get("TERM_PROGRAM") diff --git a/agent_cli/dev/terminals/tmux.py b/agent_cli/dev/terminals/tmux.py index b23a9cfaf..dae595fb4 100644 --- a/agent_cli/dev/terminals/tmux.py +++ b/agent_cli/dev/terminals/tmux.py @@ -2,15 +2,14 @@ from __future__ import annotations -import hashlib import os -import re +import shlex import shutil import subprocess from dataclasses import dataclass from typing import TYPE_CHECKING -from .base import Terminal, TerminalHandle +from .base import Multiplexer, TerminalHandle, subprocess_error_text if TYPE_CHECKING: from pathlib import Path @@ -44,7 +43,7 @@ class TmuxCleanupResult: errors: tuple[str, ...] = () -class Tmux(Terminal): +class Tmux(Multiplexer): """tmux - Terminal multiplexer.""" name = "tmux" @@ -58,11 +57,9 @@ def is_available(self) -> bool: """Check if tmux is available.""" return shutil.which("tmux") is not None - def session_name_for_repo(self, repo_root: Path) -> str: - """Build a deterministic tmux-safe session name for a repo.""" - repo_slug = re.sub(r"[^A-Za-z0-9_-]+", "-", repo_root.name).strip("-") or "repo" - repo_hash = hashlib.sha256(str(repo_root).encode()).hexdigest()[:8] - return f"agent-cli-{repo_slug[:24]}-{repo_hash}" + def attach_command(self, session_name: str) -> str: + """Shell command a user can run to attach to a tmux session.""" + return f"tmux attach -t {shlex.quote(session_name)}" def current_session_name(self) -> str | None: """Get the current tmux session name.""" @@ -156,7 +153,7 @@ def list_windows_for_worktree(self, worktree_path: Path) -> TmuxInventory: if self._is_server_unavailable_error(e): return TmuxInventory() return TmuxInventory( - error=f"Failed to inspect tmux windows for {normalized_path}: {self._error_text(e)}", + error=f"Failed to inspect tmux windows for {normalized_path}: {subprocess_error_text(e)}", ) windows: list[TmuxWindow] = [] @@ -208,7 +205,7 @@ def kill_windows_for_worktree(self, worktree_path: Path) -> TmuxCleanupResult: except subprocess.CalledProcessError as e: errors.append( "Failed to kill tmux window " - f"{window.window_id} in session {window.session_name}: {self._error_text(e)}", + f"{window.window_id} in session {window.session_name}: {subprocess_error_text(e)}", ) continue killed_windows.append(window) @@ -306,13 +303,6 @@ def _normalize_worktree_path(path: Path) -> str: """Normalize a worktree path for tmux window tagging and lookup.""" return str(path.resolve(strict=False)) - @staticmethod - def _error_text(exc: subprocess.CalledProcessError) -> str: - """Extract a useful stderr/stdout payload from a tmux subprocess error.""" - stderr = exc.stderr.strip() if exc.stderr else "" - stdout = exc.stdout.strip() if exc.stdout else "" - return stderr or stdout or str(exc) - @staticmethod def _is_server_unavailable_error(exc: subprocess.CalledProcessError) -> bool: """Detect tmux errors that mean there is no server/client to inspect.""" diff --git a/agent_cli/dev/terminals/zellij.py b/agent_cli/dev/terminals/zellij.py index ca1b119dc..85b7f7ace 100644 --- a/agent_cli/dev/terminals/zellij.py +++ b/agent_cli/dev/terminals/zellij.py @@ -1,20 +1,57 @@ -"""Zellij terminal multiplexer adapter.""" +"""Zellij terminal multiplexer adapter. + +Multiplexer-style control (detached sessions, addressable tabs, cross-session +inventory) requires zellij >= 0.44.0, which returns stable tab IDs from +``new-tab``, accepts an initial command for new tabs, and supports +``list-panes --json`` / ``close-tab-by-id`` against detached sessions. +Older zellij versions fall back to the legacy in-session tab opening. +""" from __future__ import annotations +import json import os +import re +import shlex import shutil import subprocess import time -from typing import TYPE_CHECKING +from dataclasses import dataclass +from pathlib import Path + +from .base import Multiplexer, TerminalHandle, subprocess_error_text + +# Minimum zellij version for CLI tab control (tab IDs, initial commands, +# list-panes --json, close-tab-by-id). See module docstring. +MIN_CONTROL_VERSION = (0, 44, 0) + + +@dataclass(frozen=True) +class ZellijTab: + """A zellij tab discovered via cross-session inventory.""" + + tab_id: int + session_name: str + tab_name: str + + +@dataclass(frozen=True) +class ZellijInventory: + """Zellij tabs owned by a worktree, plus any lookup error.""" + + tabs: tuple[ZellijTab, ...] = () + error: str | None = None -from .base import Terminal -if TYPE_CHECKING: - from pathlib import Path +@dataclass(frozen=True) +class ZellijCleanupResult: + """Result of closing zellij tabs for a worktree.""" + closed_tabs: tuple[ZellijTab, ...] = () + errors: tuple[str, ...] = () -class Zellij(Terminal): + +class Zellij(Multiplexer): """Zellij - A terminal workspace with batteries included.""" name = "zellij" @@ -28,6 +65,14 @@ def is_available(self) -> bool: """Check if Zellij is available.""" return shutil.which("zellij") is not None + def current_session_name(self) -> str | None: + """Get the current zellij session name.""" + return os.environ.get("ZELLIJ_SESSION_NAME") or None + + def attach_command(self, session_name: str) -> str: + """Shell command a user can run to attach to a zellij session.""" + return f"zellij attach {shlex.quote(session_name)}" + def open_new_tab( self, path: Path, @@ -40,10 +85,132 @@ def open_new_tab( """ if not self.is_available(): return False + if self._supports_cli_control(): + return self.open_in_session(path, command, tab_name) is not None + return self._open_new_tab_legacy(path, command, tab_name) + + def open_in_session( + self, + path: Path, + command: str | None = None, + tab_name: str | None = None, + *, + session_name: str | None = None, + ) -> TerminalHandle | None: + """Open a zellij tab and return its handle. + + If ``session_name`` is omitted, the current zellij session is used. + When a named session does not exist yet, it is created in detached mode. + Requires zellij >= 0.44.0. + """ + if not self.is_available() or not self._supports_cli_control(): + return None + + if session_name is None: + if not self.detect(): + return None + session_name = self.current_session_name() + zellij_cmd = ["zellij"] + else: + if not self._ensure_session(session_name): + return None + zellij_cmd = ["zellij", "--session", session_name] + + cmd = [*zellij_cmd, "action", "new-tab", "--cwd", str(path)] + if tab_name: + cmd.extend(["--name", tab_name]) + if command: + # The initial command is argv-style (no shell), so wrap it + cmd.extend(["--", "/bin/sh", "-c", command]) + try: + result = subprocess.run(cmd, check=True, capture_output=True, text=True) + except subprocess.CalledProcessError: + return None + return TerminalHandle( + terminal_name=self.name, + handle=result.stdout.strip(), + session_name=session_name, + ) + + def list_tabs_for_worktree(self, worktree_path: Path) -> ZellijInventory: + """List zellij tabs with a pane working in a worktree, across all live sessions.""" + if not self.is_available() or not self._supports_cli_control(): + return ZellijInventory() + + normalized_path = self._normalize_worktree_path(worktree_path) + tabs: dict[tuple[str, int], ZellijTab] = {} + errors: list[str] = [] + for session_name in self._live_session_names(): + panes, error = self._list_panes(session_name) + if error is not None: + errors.append(error) + continue + for pane in panes: + tab_id = pane.get("tab_id") + if ( + pane.get("is_plugin") + or tab_id is None + or not self._pane_in_worktree(pane, normalized_path) + ): + continue + tabs.setdefault( + (session_name, tab_id), + ZellijTab( + tab_id=tab_id, + session_name=session_name, + tab_name=pane.get("tab_name") or "", + ), + ) + + return ZellijInventory(tabs=tuple(tabs.values()), error="; ".join(errors) or None) + + def close_tabs_for_worktree(self, worktree_path: Path) -> ZellijCleanupResult: + """Close zellij tabs working in a worktree, across all live sessions.""" + inventory = self.list_tabs_for_worktree(worktree_path) + errors: list[str] = [inventory.error] if inventory.error else [] + + current_tab = self._current_tab() if inventory.tabs else None + closed_tabs: list[ZellijTab] = [] + for tab in inventory.tabs: + if current_tab == (tab.session_name, tab.tab_id): + errors.append( + f"Skipped zellij tab {tab.tab_id} in session {tab.session_name} " + "because it is the current tab", + ) + continue + try: + subprocess.run( + [ # noqa: S607 + "zellij", + "--session", + tab.session_name, + "action", + "close-tab-by-id", + str(tab.tab_id), + ], + check=True, + capture_output=True, + text=True, + ) + except subprocess.CalledProcessError as e: + errors.append( + f"Failed to close zellij tab {tab.tab_id} in session " + f"{tab.session_name}: {subprocess_error_text(e)}", + ) + continue + closed_tabs.append(tab) + + return ZellijCleanupResult(closed_tabs=tuple(closed_tabs), errors=tuple(errors)) + def _open_new_tab_legacy( + self, + path: Path, + command: str | None, + tab_name: str | None, + ) -> bool: + """Open a tab on zellij < 0.44 by typing the command into the focused pane.""" try: - # Create new tab using zellij action - # Workaround: --cwd requires --layout due to bug (github.com/zellij-org/zellij/issues/2981) + # Workaround: --cwd requires --layout on zellij < 0.43 (zellij-org/zellij#2981) cmd = ["zellij", "action", "new-tab", "--layout", "default", "--cwd", str(path)] if tab_name: cmd.extend(["--name", tab_name]) @@ -76,3 +243,110 @@ def open_new_tab( return True except subprocess.CalledProcessError: return False + + def _supports_cli_control(self) -> bool: + """Whether the installed zellij supports tab IDs and detached-session control.""" + version = self._cli_version() + return version is not None and version >= MIN_CONTROL_VERSION + + @staticmethod + def _cli_version() -> tuple[int, int, int] | None: + """Parse the installed zellij version (e.g. "zellij 0.44.3").""" + try: + result = subprocess.run( + ["zellij", "--version"], # noqa: S607 + check=True, + capture_output=True, + text=True, + ) + except (subprocess.CalledProcessError, OSError): + return None + match = re.search(r"(\d+)\.(\d+)\.(\d+)", result.stdout) + if match is None: + return None + major, minor, patch = match.groups() + return int(major), int(minor), int(patch) + + @staticmethod + def _ensure_session(session_name: str) -> bool: + """Create a detached session if it does not exist yet.""" + result = subprocess.run( + ["zellij", "attach", "--create-background", session_name], # noqa: S607 + check=False, + capture_output=True, + text=True, + ) + if result.returncode == 0: + return True + # Non-interactive attach to an existing session fails with this message + return "already exists" in f"{result.stdout}\n{result.stderr}".lower() + + @staticmethod + def _live_session_names() -> list[str]: + """List running (non-EXITED) zellij session names.""" + try: + result = subprocess.run( + ["zellij", "list-sessions", "--no-formatting"], # noqa: S607 + check=True, + capture_output=True, + text=True, + ) + except (subprocess.CalledProcessError, OSError): + # zellij exits non-zero when there are no sessions at all + return [] + names: list[str] = [] + for raw_line in result.stdout.splitlines(): + line = raw_line.strip() + if not line or "(EXITED" in line or line.lower().startswith("no active"): + continue + names.append(line.split()[0]) + return names + + @staticmethod + def _list_panes(session_name: str) -> tuple[list[dict], str | None]: + """List panes of a session as JSON dicts, with any lookup error.""" + try: + result = subprocess.run( + ["zellij", "--session", session_name, "action", "list-panes", "--json"], # noqa: S607 + check=True, + capture_output=True, + text=True, + ) + panes = json.loads(result.stdout) + except subprocess.CalledProcessError as e: + return [], ( + f"Failed to inspect zellij panes in session {session_name}: " + f"{subprocess_error_text(e)}" + ) + except json.JSONDecodeError as e: + return [], f"Failed to parse zellij panes in session {session_name}: {e}" + return panes, None + + @staticmethod + def _pane_in_worktree(pane: dict, normalized_path: str) -> bool: + """Whether a pane's working directory is inside a worktree.""" + pane_cwd = pane.get("pane_cwd") + if not pane_cwd: + return False + try: + return Path(pane_cwd).resolve(strict=False).is_relative_to(normalized_path) + except (OSError, ValueError): + return False + + @staticmethod + def _normalize_worktree_path(path: Path) -> str: + """Normalize a worktree path for pane cwd matching.""" + return str(path.resolve(strict=False)) + + def _current_tab(self) -> tuple[str, int] | None: + """Identify the (session, tab id) holding this process's pane, when inside zellij.""" + session_name = self.current_session_name() + pane_id = os.environ.get("ZELLIJ_PANE_ID") + if session_name is None or pane_id is None: + return None + panes, _error = self._list_panes(session_name) + for pane in panes: + if not pane.get("is_plugin") and str(pane.get("id")) == pane_id: + tab_id = pane.get("tab_id") + return (session_name, tab_id) if tab_id is not None else None + return None diff --git a/docs/commands/dev.md b/docs/commands/dev.md index be100e66a..5088bfc60 100644 --- a/docs/commands/dev.md +++ b/docs/commands/dev.md @@ -87,8 +87,9 @@ agent-cli dev new [BRANCH] [OPTIONS] | `--agent-args` | - | Extra CLI args for the agent. Can be repeated. Example: --agent-args='--dangerously-skip-permissions' | | `--prompt, -p` | - | Initial task for the AI agent. Saved to a unique file in .claude/ to avoid conflicts. Implies starting the agent. Example: --prompt='Fix the login bug' | | `--prompt-file, -P` | - | Read the agent prompt from a file. Useful for long prompts to avoid shell quoting. Implies starting the agent | -| `--multiplexer, -m` | - | Launch the agent in a specific multiplexer. Currently supported: tmux. When started outside tmux, creates or reuses a detached session and reports the pane handle | +| `--multiplexer, -m` | - | Launch the agent in a specific multiplexer. Currently supported: tmux, zellij (zellij requires >= 0.44.0). When started outside the multiplexer, creates or reuses a detached session and reports the tab/pane handle | | `--tmux-session` | - | Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux | +| `--zellij-session` | - | Reuse or create a specific zellij session for the agent. Implies --multiplexer zellij | | `--hooks/--no-hooks` | `true` | Run built-in agent preparation (like Codex auto-trust) and configured pre-launch hooks before starting the agent | | `--verbose, -v` | `false` | Stream output from setup commands instead of hiding it | @@ -305,8 +306,9 @@ agent-cli dev agent NAME [--agent/-a AGENT] [--agent-args ARGS] [--prompt/-p PRO | `--agent-args` | - | Extra CLI args for the agent. Example: --agent-args='--dangerously-skip-permissions' | | `--prompt, -p` | - | Initial task for the agent. Saved to a unique file in .claude/ to avoid conflicts. Example: --prompt='Add unit tests for auth' | | `--prompt-file, -P` | - | Read the agent prompt from a file instead of command line | -| `--multiplexer, -m` | - | Launch the agent in a specific multiplexer instead of the current terminal. Currently supported: tmux | +| `--multiplexer, -m` | - | Launch the agent in a specific multiplexer instead of the current terminal. Currently supported: tmux, zellij (zellij requires >= 0.44.0) | | `--tmux-session` | - | Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux | +| `--zellij-session` | - | Reuse or create a specific zellij session for the agent. Implies --multiplexer zellij | | `--hooks/--no-hooks` | `true` | Run built-in agent preparation (like Codex auto-trust) and configured pre-launch hooks before starting the agent | diff --git a/tests/dev/test_cleanup.py b/tests/dev/test_cleanup.py index 2b42da2f8..9d2452081 100644 --- a/tests/dev/test_cleanup.py +++ b/tests/dev/test_cleanup.py @@ -1,4 +1,4 @@ -"""Tests for tmux-aware worktree cleanup helpers.""" +"""Tests for multiplexer-aware worktree cleanup helpers.""" from __future__ import annotations @@ -7,6 +7,7 @@ from agent_cli.dev.cleanup import remove_worktree, remove_worktrees from agent_cli.dev.terminals.tmux import TmuxCleanupResult, TmuxWindow +from agent_cli.dev.terminals.zellij import ZellijCleanupResult, ZellijTab from agent_cli.dev.worktree import WorktreeInfo @@ -25,8 +26,8 @@ def _worktree_info() -> WorktreeInfo: class TestCleanup: """Tests for shared worktree cleanup orchestration.""" - def test_remove_worktree_cleans_tmux_after_git_removal(self) -> None: - """Git removal should happen before tagged tmux windows are killed.""" + def test_remove_worktree_cleans_multiplexers_after_git_removal(self) -> None: + """Git removal should happen before tmux windows and zellij tabs are cleaned.""" wt = _worktree_info() call_order: list[str] = [] @@ -42,34 +43,49 @@ def fake_kill(_path: Path) -> TmuxCleanupResult: ), ) + def fake_close(_path: Path) -> ZellijCleanupResult: + call_order.append("zellij") + return ZellijCleanupResult( + closed_tabs=(ZellijTab(tab_id=2, session_name="shared", tab_name="agent"),), + ) + with ( patch("agent_cli.dev.cleanup.worktree.remove_worktree", side_effect=fake_remove), patch("agent_cli.dev.cleanup.Tmux") as mock_tmux_cls, + patch("agent_cli.dev.cleanup.Zellij") as mock_zellij_cls, ): mock_tmux_cls.return_value.kill_windows_for_worktree.side_effect = fake_kill + mock_zellij_cls.return_value.close_tabs_for_worktree.side_effect = fake_close result = remove_worktree(wt, Path("/repo"), force=True, delete_branch=True) assert result.success is True - assert call_order == ["git", "tmux"] + assert call_order == ["git", "tmux", "zellij"] - def test_remove_worktree_surfaces_tmux_cleanup_errors_as_warnings(self) -> None: - """Tmux cleanup failures should not fail an already-removed worktree.""" + def test_remove_worktree_surfaces_multiplexer_cleanup_errors_as_warnings(self) -> None: + """Multiplexer cleanup failures should not fail an already-removed worktree.""" wt = _worktree_info() with ( patch("agent_cli.dev.cleanup.worktree.remove_worktree", return_value=(True, None)), patch("agent_cli.dev.cleanup.Tmux") as mock_tmux_cls, + patch("agent_cli.dev.cleanup.Zellij") as mock_zellij_cls, ): mock_tmux_cls.return_value.kill_windows_for_worktree.return_value = TmuxCleanupResult( errors=("Failed to kill tmux window @2 in session shared: boom",), ) + mock_zellij_cls.return_value.close_tabs_for_worktree.return_value = ZellijCleanupResult( + errors=("Failed to close zellij tab 2 in session shared: boom",), + ) result = remove_worktree(wt, Path("/repo")) assert result.success is True - assert result.warnings == ["Failed to kill tmux window @2 in session shared: boom"] + assert result.warnings == [ + "Failed to kill tmux window @2 in session shared: boom", + "Failed to close zellij tab 2 in session shared: boom", + ] - def test_remove_worktree_skips_tmux_cleanup_when_git_removal_fails(self) -> None: - """Tmux cleanup should not run if git worktree removal fails.""" + def test_remove_worktree_skips_multiplexer_cleanup_when_git_removal_fails(self) -> None: + """Multiplexer cleanup should not run if git worktree removal fails.""" wt = _worktree_info() with ( @@ -78,12 +94,14 @@ def test_remove_worktree_skips_tmux_cleanup_when_git_removal_fails(self) -> None return_value=(False, "cannot remove"), ), patch("agent_cli.dev.cleanup.Tmux") as mock_tmux_cls, + patch("agent_cli.dev.cleanup.Zellij") as mock_zellij_cls, ): result = remove_worktree(wt, Path("/repo")) assert result.success is False assert result.error == "cannot remove" mock_tmux_cls.return_value.kill_windows_for_worktree.assert_not_called() + mock_zellij_cls.return_value.close_tabs_for_worktree.assert_not_called() def test_remove_worktrees_deletes_branches_for_cleanups(self) -> None: """Batch cleanups should request branch deletion for each worktree.""" diff --git a/tests/dev/test_cli.py b/tests/dev/test_cli.py index 46e26f275..ee7753dc3 100644 --- a/tests/dev/test_cli.py +++ b/tests/dev/test_cli.py @@ -720,9 +720,57 @@ def test_new_tmux_session_implies_tmux_and_normalizes(self, tmp_path: Path) -> N assert result.exit_code == 0 assert mock_launch.call_args.kwargs["multiplexer_name"] == "tmux" - assert mock_launch.call_args.kwargs["tmux_session"] == "my_session name" + assert mock_launch.call_args.kwargs["multiplexer_session"] == "my_session name" assert "tmux attach -t 'my_session name'" in result.output + def test_new_zellij_session_implies_zellij_and_normalizes(self, tmp_path: Path) -> None: + """`--zellij-session` trims whitespace and forces zellij.""" + wt_path = tmp_path / "repo-worktrees" / "feature" + wt_path.mkdir(parents=True) + + with ( + patch("agent_cli.dev.cli._ensure_git_repo", return_value=Path("/repo")), + patch( + "agent_cli.dev.worktree.create_worktree", + return_value=CreateWorktreeResult( + success=True, + path=wt_path, + branch="feature", + ), + ), + patch("agent_cli.dev.cli.resolve_editor", return_value=None), + patch("agent_cli.dev.cli.resolve_agent") as mock_resolve_agent, + patch("agent_cli.dev.cli.prepare_agent_launch"), + patch("agent_cli.dev.cli.merge_agent_args", return_value=None), + patch("agent_cli.dev.cli.get_agent_env", return_value={}), + patch( + "agent_cli.dev.cli.launch_agent", + return_value=TerminalHandle("zellij", "3", "my session"), + ) as mock_launch, + ): + mock_agent = mock_resolve_agent.return_value + mock_agent.is_available.return_value = True + result = runner.invoke( + app, + [ + "dev", + "new", + "feature", + "--start-agent", + "--zellij-session", + " my session ", + "--no-setup", + "--no-copy-env", + "--no-fetch", + "--no-direnv", + ], + ) + + assert result.exit_code == 0 + assert mock_launch.call_args.kwargs["multiplexer_name"] == "zellij" + assert mock_launch.call_args.kwargs["multiplexer_session"] == "my session" + assert "zellij attach 'my session'" in result.output + def test_new_tmux_session_does_not_imply_start_agent(self, tmp_path: Path) -> None: """`--tmux-session` is only a placement flag for `dev new`.""" wt_path = tmp_path / "repo-worktrees" / "feature" @@ -956,6 +1004,39 @@ def test_new_rejects_tmux_session_with_illegal_characters(self, tmux_session: st assert "tmux session names cannot contain '.' or ':'" in result.output mock_ensure_repo.assert_not_called() + def test_new_rejects_empty_zellij_session(self) -> None: + """Empty zellij session names should fail before repo/worktree work starts.""" + with patch("agent_cli.dev.cli._ensure_git_repo") as mock_ensure_repo: + result = runner.invoke(app, ["dev", "new", "my-feature", "--zellij-session", " "]) + + assert result.exit_code == 1 + assert "--zellij-session cannot be empty" in result.output + mock_ensure_repo.assert_not_called() + + def test_new_rejects_conflicting_session_flags(self) -> None: + """`--tmux-session` and `--zellij-session` are mutually exclusive.""" + with patch("agent_cli.dev.cli._ensure_git_repo") as mock_ensure_repo: + result = runner.invoke( + app, + ["dev", "new", "my-feature", "--tmux-session", "a", "--zellij-session", "b"], + ) + + assert result.exit_code == 1 + assert "Cannot use --tmux-session and --zellij-session together" in result.output + mock_ensure_repo.assert_not_called() + + def test_new_rejects_session_flag_for_other_multiplexer(self) -> None: + """A session flag for one multiplexer cannot combine with the other multiplexer.""" + with patch("agent_cli.dev.cli._ensure_git_repo") as mock_ensure_repo: + result = runner.invoke( + app, + ["dev", "new", "my-feature", "--multiplexer", "zellij", "--tmux-session", "a"], + ) + + assert result.exit_code == 1 + assert "--tmux-session cannot be combined with --multiplexer zellij" in result.output + mock_ensure_repo.assert_not_called() + def test_new_skips_launch_preparation_when_hooks_are_disabled(self, tmp_path: Path) -> None: """`--no-hooks` should bypass built-in preparation and configured hooks.""" wt_path = tmp_path / "repo-worktrees" / "feature" @@ -1137,7 +1218,7 @@ def test_agent_tmux_session_implies_tmux_and_normalizes(self) -> None: assert result.exit_code == 0 assert mock_launch.call_args.kwargs["multiplexer_name"] == "tmux" - assert mock_launch.call_args.kwargs["tmux_session"] == "my_session name" + assert mock_launch.call_args.kwargs["multiplexer_session"] == "my_session name" assert "tmux attach -t 'my_session name'" in result.output def test_agent_quotes_tmux_attach_hint(self) -> None: diff --git a/tests/dev/test_launch.py b/tests/dev/test_launch.py index 56a3c523a..e26dbc9b2 100644 --- a/tests/dev/test_launch.py +++ b/tests/dev/test_launch.py @@ -14,6 +14,7 @@ from agent_cli.dev.terminals import TerminalHandle from agent_cli.dev.terminals.cmux import Cmux from agent_cli.dev.terminals.tmux import Tmux +from agent_cli.dev.terminals.zellij import Zellij class TestLaunchAgent: @@ -107,7 +108,7 @@ def test_explicit_tmux_session_takes_precedence_outside_tmux(self, tmp_path: Pat patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), ): result = launch_agent( - tmp_path, agent, multiplexer_name="tmux", tmux_session="shared-session" + tmp_path, agent, multiplexer_name="tmux", multiplexer_session="shared-session" ) assert result == handle @@ -138,7 +139,75 @@ def test_explicit_tmux_session_takes_precedence_inside_tmux(self, tmp_path: Path patch("agent_cli.dev.launch.worktree.get_main_repo_root", return_value=Path("/repo")), patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), ): - result = launch_agent(tmp_path, agent, tmux_session="shared-session") + result = launch_agent( + tmp_path, agent, multiplexer_name="tmux", multiplexer_session="shared-session" + ) + + assert result == handle + mock_session_name.assert_not_called() + mock_open.assert_called_once_with( + tmp_path, + "codex", + tab_name="feature", + session_name="shared-session", + ) + + def test_uses_requested_zellij_outside_zellij(self, tmp_path: Path) -> None: + """Explicit zellij launch uses a detached repo session when not already in zellij.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + zellij_terminal = Zellij() + handle = TerminalHandle("zellij", "3", "agent-cli-repo-1234") + + with ( + patch("agent_cli.dev.launch.terminals.get_terminal", return_value=zellij_terminal), + patch("agent_cli.dev.launch.terminals.detect_current_terminal", return_value=None), + patch.object(zellij_terminal, "is_available", return_value=True), + patch.object(zellij_terminal, "detect", return_value=False), + patch.object( + zellij_terminal, + "session_name_for_repo", + return_value="agent-cli-repo-1234", + ) as mock_session_name, + patch.object(zellij_terminal, "open_in_session", return_value=handle) as mock_open, + patch("agent_cli.dev.launch.worktree.get_main_repo_root", return_value=Path("/repo")), + patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), + ): + result = launch_agent(tmp_path, agent, multiplexer_name="zellij") + + assert result == handle + mock_session_name.assert_called_once_with(Path("/repo")) + mock_open.assert_called_once_with( + tmp_path, + "codex", + tab_name="feature", + session_name="agent-cli-repo-1234", + ) + + def test_explicit_zellij_session_takes_precedence(self, tmp_path: Path) -> None: + """An explicit zellij session overrides the repo-derived detached session.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + zellij_terminal = Zellij() + handle = TerminalHandle("zellij", "3", "shared-session") + + with ( + patch("agent_cli.dev.launch.terminals.get_terminal", return_value=zellij_terminal), + patch("agent_cli.dev.launch.terminals.detect_current_terminal", return_value=None), + patch.object(zellij_terminal, "is_available", return_value=True), + patch.object(zellij_terminal, "detect", return_value=False), + patch.object(zellij_terminal, "session_name_for_repo") as mock_session_name, + patch.object(zellij_terminal, "open_in_session", return_value=handle) as mock_open, + patch("agent_cli.dev.launch.worktree.get_main_repo_root", return_value=Path("/repo")), + patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), + ): + result = launch_agent( + tmp_path, agent, multiplexer_name="zellij", multiplexer_session="shared-session" + ) assert result == handle mock_session_name.assert_not_called() @@ -149,6 +218,63 @@ def test_explicit_tmux_session_takes_precedence_inside_tmux(self, tmp_path: Path session_name="shared-session", ) + def test_detected_zellij_uses_current_session(self, tmp_path: Path) -> None: + """Inside zellij without --multiplexer, the current session is used.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + zellij_terminal = Zellij() + handle = TerminalHandle("zellij", "3", "current-session") + + with ( + patch( + "agent_cli.dev.launch.terminals.detect_current_terminal", + return_value=zellij_terminal, + ), + patch.object(zellij_terminal, "is_available", return_value=True), + patch.object(zellij_terminal, "detect", return_value=True), + patch.object(zellij_terminal, "session_name_for_repo") as mock_session_name, + patch.object(zellij_terminal, "open_in_session", return_value=handle) as mock_open, + patch("agent_cli.dev.launch.worktree.get_main_repo_root", return_value=Path("/repo")), + patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), + ): + result = launch_agent(tmp_path, agent) + + assert result == handle + mock_session_name.assert_not_called() + mock_open.assert_called_once_with( + tmp_path, + "codex", + tab_name="feature", + session_name=None, + ) + + def test_detected_zellij_falls_back_to_plain_tab_when_unsupported(self, tmp_path: Path) -> None: + """Auto-detected zellij falls back to open_new_tab when CLI control is unavailable.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + zellij_terminal = Zellij() + + with ( + patch( + "agent_cli.dev.launch.terminals.detect_current_terminal", + return_value=zellij_terminal, + ), + patch.object(zellij_terminal, "is_available", return_value=True), + patch.object(zellij_terminal, "detect", return_value=True), + patch.object(zellij_terminal, "open_in_session", return_value=None), + patch.object(zellij_terminal, "open_new_tab", return_value=True) as mock_open_tab, + patch("agent_cli.dev.launch.worktree.get_main_repo_root", return_value=Path("/repo")), + patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), + ): + result = launch_agent(tmp_path, agent) + + assert result is None + mock_open_tab.assert_called_once() + def test_uses_wrapper_script_for_requested_tmux(self, tmp_path: Path) -> None: """Prompt launches still use the wrapper script for explicit tmux sessions.""" task_file = tmp_path / ".claude" / "TASK.md" diff --git a/tests/dev/test_terminals.py b/tests/dev/test_terminals.py index 4ed16407c..561444c41 100644 --- a/tests/dev/test_terminals.py +++ b/tests/dev/test_terminals.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import shlex import subprocess from pathlib import Path @@ -20,7 +21,7 @@ from agent_cli.dev.terminals.cmux import Cmux from agent_cli.dev.terminals.kitty import Kitty from agent_cli.dev.terminals.tmux import Tmux, TmuxInventory, TmuxWindow -from agent_cli.dev.terminals.zellij import Zellij +from agent_cli.dev.terminals.zellij import Zellij, ZellijInventory, ZellijTab class TestTmux: @@ -381,6 +382,297 @@ def test_is_available(self) -> None: with patch("shutil.which", return_value="/usr/bin/zellij"): assert terminal.is_available() is True + def test_current_session_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Current session name comes from ZELLIJ_SESSION_NAME.""" + monkeypatch.setenv("ZELLIJ_SESSION_NAME", "my-session") + assert Zellij().current_session_name() == "my-session" + + def test_attach_command_quotes_session_name(self) -> None: + """Attach hints quote session names for the shell.""" + assert Zellij().attach_command("my session") == "zellij attach 'my session'" + + def test_cli_version_parsing(self) -> None: + """Version is parsed from `zellij --version` output.""" + with patch("subprocess.run", return_value=MagicMock(stdout="zellij 0.44.3\n")): + assert Zellij._cli_version() == (0, 44, 3) + + def test_open_in_session_creates_detached_session_when_missing(self) -> None: + """Outside zellij, a named session is created in detached mode if absent.""" + terminal = Zellij() + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(terminal, "_supports_cli_control", return_value=True), + patch( + "subprocess.run", + side_effect=[ + MagicMock(returncode=0), # attach --create-background + MagicMock(returncode=0, stdout="3\n"), # action new-tab + ], + ) as mock_run, + ): + handle = terminal.open_in_session( + Path("/some/path"), + "echo hello", + tab_name="feature", + session_name="repo-session", + ) + + assert handle is not None + assert handle.handle == "3" + assert handle.session_name == "repo-session" + assert mock_run.call_args_list[0].args[0] == [ + "zellij", + "attach", + "--create-background", + "repo-session", + ] + assert mock_run.call_args_list[1].args[0] == [ + "zellij", + "--session", + "repo-session", + "action", + "new-tab", + "--cwd", + "/some/path", + "--name", + "feature", + "--", + "/bin/sh", + "-c", + "echo hello", + ] + + def test_open_in_session_reuses_existing_session(self) -> None: + """An existing session is reused when create-background reports it exists.""" + terminal = Zellij() + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(terminal, "_supports_cli_control", return_value=True), + patch( + "subprocess.run", + side_effect=[ + MagicMock(returncode=1, stdout="Session already exists\n", stderr=""), + MagicMock(returncode=0, stdout="4\n"), + ], + ), + ): + handle = terminal.open_in_session( + Path("/some/path"), + "echo hello", + session_name="repo-session", + ) + + assert handle is not None + assert handle.handle == "4" + + def test_open_in_session_uses_current_session_inside_zellij( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Inside zellij without a named session, the current session is used.""" + monkeypatch.setenv("ZELLIJ", "0") + monkeypatch.setenv("ZELLIJ_SESSION_NAME", "current-session") + terminal = Zellij() + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(terminal, "_supports_cli_control", return_value=True), + patch("subprocess.run", return_value=MagicMock(returncode=0, stdout="2\n")) as mock_run, + ): + handle = terminal.open_in_session(Path("/some/path"), "echo hello") + + assert handle is not None + assert handle.session_name == "current-session" + assert mock_run.call_args.args[0][:3] == ["zellij", "action", "new-tab"] + + def test_open_in_session_requires_modern_zellij(self) -> None: + """Multiplexer control is gated on zellij >= 0.44.0.""" + terminal = Zellij() + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(Zellij, "_cli_version", return_value=(0, 43, 1)), + patch("subprocess.run") as mock_run, + ): + handle = terminal.open_in_session( + Path("/some/path"), + "echo hello", + session_name="repo-session", + ) + + assert handle is None + mock_run.assert_not_called() + + def test_open_new_tab_falls_back_to_legacy_on_old_zellij( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Zellij < 0.44 opens tabs via the legacy write-chars path.""" + monkeypatch.setenv("ZELLIJ", "0") + terminal = Zellij() + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(Zellij, "_cli_version", return_value=(0, 41, 2)), + patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, + patch("time.sleep"), + ): + result = terminal.open_new_tab(Path("/some/path"), "echo hello", tab_name="feature") + + assert result is True + commands = [call.args[0] for call in mock_run.call_args_list] + assert commands == [ + [ + "zellij", + "action", + "new-tab", + "--layout", + "default", + "--cwd", + "/some/path", + "--name", + "feature", + ], + ["zellij", "action", "write-chars", "echo hello"], + ["zellij", "action", "write", "10"], + ] + + def test_open_new_tab_uses_open_in_session_on_modern_zellij( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Zellij >= 0.44 opens tabs with an initial command instead of write-chars.""" + monkeypatch.setenv("ZELLIJ", "0") + monkeypatch.setenv("ZELLIJ_SESSION_NAME", "current-session") + terminal = Zellij() + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(Zellij, "_cli_version", return_value=(0, 44, 0)), + patch("subprocess.run", return_value=MagicMock(returncode=0, stdout="2\n")) as mock_run, + ): + result = terminal.open_new_tab(Path("/some/path"), "echo hello", tab_name="feature") + + assert result is True + assert mock_run.call_args.args[0] == [ + "zellij", + "action", + "new-tab", + "--cwd", + "/some/path", + "--name", + "feature", + "--", + "/bin/sh", + "-c", + "echo hello", + ] + + def test_list_tabs_for_worktree_matches_pane_cwd(self, tmp_path: Path) -> None: + """Tabs are inventoried by pane cwd across live sessions, skipping dead ones.""" + terminal = Zellij() + worktree = tmp_path / "wt" + worktree.mkdir() + sessions_output = ( + "work [Created 5s ago] \ndead [Created 1day 2h ago] (EXITED - attach to resurrect)\n" + ) + panes = [ + {"id": 0, "is_plugin": False, "tab_id": 0, "tab_name": "main", "pane_cwd": "/other"}, + { + "id": 1, + "is_plugin": False, + "tab_id": 2, + "tab_name": "feature", + "pane_cwd": str(worktree), + }, + {"id": 2, "is_plugin": True, "tab_id": 3, "tab_name": "plug", "pane_cwd": ""}, + ] + with ( + patch.object(terminal, "is_available", return_value=True), + patch.object(terminal, "_supports_cli_control", return_value=True), + patch( + "subprocess.run", + side_effect=[ + MagicMock(returncode=0, stdout=sessions_output), + MagicMock(returncode=0, stdout=json.dumps(panes)), + ], + ) as mock_run, + ): + inventory = terminal.list_tabs_for_worktree(worktree) + + assert inventory.error is None + assert inventory.tabs == (ZellijTab(tab_id=2, session_name="work", tab_name="feature"),) + # Only the live session is inspected + assert mock_run.call_args_list[1].args[0] == [ + "zellij", + "--session", + "work", + "action", + "list-panes", + "--json", + ] + + def test_close_tabs_for_worktree_closes_by_id(self, tmp_path: Path) -> None: + """Worktree tabs are closed via close-tab-by-id.""" + terminal = Zellij() + inventory = ZellijInventory( + tabs=(ZellijTab(tab_id=2, session_name="work", tab_name="feature"),), + ) + with ( + patch.object(terminal, "list_tabs_for_worktree", return_value=inventory), + patch.object(terminal, "_current_tab", return_value=None), + patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, + ): + result = terminal.close_tabs_for_worktree(tmp_path) + + assert result.closed_tabs == inventory.tabs + assert result.errors == () + assert mock_run.call_args.args[0] == [ + "zellij", + "--session", + "work", + "action", + "close-tab-by-id", + "2", + ] + + def test_close_tabs_for_worktree_skips_current_tab(self, tmp_path: Path) -> None: + """The tab running this process is never closed.""" + terminal = Zellij() + inventory = ZellijInventory( + tabs=( + ZellijTab(tab_id=2, session_name="work", tab_name="feature"), + ZellijTab(tab_id=5, session_name="work", tab_name="other"), + ), + ) + with ( + patch.object(terminal, "list_tabs_for_worktree", return_value=inventory), + patch.object(terminal, "_current_tab", return_value=("work", 2)), + patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, + ): + result = terminal.close_tabs_for_worktree(tmp_path) + + assert result.closed_tabs == (inventory.tabs[1],) + assert len(result.errors) == 1 + assert "current tab" in result.errors[0] + assert len(mock_run.call_args_list) == 1 + + def test_current_tab_resolves_own_pane(self, monkeypatch: pytest.MonkeyPatch) -> None: + """The current tab is resolved from ZELLIJ_PANE_ID via list-panes.""" + monkeypatch.setenv("ZELLIJ_SESSION_NAME", "work") + monkeypatch.setenv("ZELLIJ_PANE_ID", "7") + terminal = Zellij() + panes = [ + {"id": 1, "is_plugin": False, "tab_id": 0}, + {"id": 7, "is_plugin": False, "tab_id": 4}, + ] + with patch( + "subprocess.run", return_value=MagicMock(returncode=0, stdout=json.dumps(panes)) + ): + assert terminal._current_tab() == ("work", 4) + + def test_live_session_names_handles_no_sessions(self) -> None: + """A non-zero exit from list-sessions means there are no sessions.""" + error = subprocess.CalledProcessError(1, "zellij") + with patch("subprocess.run", side_effect=error): + assert Zellij._live_session_names() == [] + class TestKitty: """Tests for Kitty terminal.""" diff --git a/tests/dev/test_verification.py b/tests/dev/test_verification.py index c97986417..8d5a167cb 100644 --- a/tests/dev/test_verification.py +++ b/tests/dev/test_verification.py @@ -38,7 +38,7 @@ from agent_cli.dev.terminals.kitty import Kitty from agent_cli.dev.terminals.tmux import Tmux from agent_cli.dev.terminals.warp import Warp -from agent_cli.dev.terminals.zellij import Zellij +from agent_cli.dev.terminals.zellij import MIN_CONTROL_VERSION, Zellij class TestTerminalDetection: @@ -210,24 +210,112 @@ def test_tmux_new_window_command(self) -> None: assert "test-tab" in call_args def test_zellij_new_tab_command_syntax(self) -> None: - """Zellij uses `zellij action new-tab --cwd --name `. + """Zellij >= 0.44 runs commands via `new-tab --cwd --name -- `. Evidence: - Source: zellij action new-tab --help + Source: zellij action new-tab --help (zellij 0.44.3) Output: -c, --cwd Change the working directory of the new tab -n, --name Name of the new tab - Verified: 2026-01-11 via `zellij action new-tab --help` + [-- ...] Optional initial command to run in the new tab + "Returns: The created tab's ID as a single number on stdout" + Initial command support and returned tab IDs were added in 0.44.0: + CHANGELOG 0.44.0: "feat: command sequences, conditionally blocking + CLI commands" (zellij-org/zellij#4546) and "return pane/tab IDs from + plugin and CLI methods creating them" (zellij-org/zellij#4690) + `--cwd` without `--layout` works since 0.43.0 (zellij-org/zellij#4273), + which is why the modern path drops the `--layout default` workaround. + Verified: 2026-06-10 live on zellij 0.44.3 — `zellij --session X action + new-tab --name t --cwd /private/tmp -- bash -c 'pwd > /tmp/out'` printed + the tab ID and the pane's cwd was /private/tmp """ - # Syntax verified via --help, implementation tested elsewhere + assert MIN_CONTROL_VERSION == (0, 44, 0) + + def test_zellij_session_targeting_from_outside(self) -> None: + """Zellij actions target other sessions via the global `--session` flag. + + Evidence: + Source: Zellij CLI documentation + URL: https://zellij.dev/documentation/controlling-zellij-through-cli + Quote: "Commands can also be issued to a different Zellij session: + `$ zellij --session pretentious-cat action new-pane`" + Verified: 2026-06-10 live on zellij 0.44.3 against a fully detached + session (no attached clients) + """ + # Implementation verified in test_terminals.py via open_in_session/list/close tests + + def test_zellij_create_background_session(self) -> None: + """`zellij attach --create-background ` creates a detached session. + + Evidence: + Source: zellij attach --help (zellij 0.44.3) + Quote: "-b, --create-background Create a detached session in the + background if one does not exist" + Added in 0.40.0: CHANGELOG "feat(cli): allow starting a session in the + background (detached)" (zellij-org/zellij#3257, #3265) + Verified: 2026-06-10 live on zellij 0.44.3 — first call exits 0 and the + session appears in `zellij list-sessions`; a second non-interactive call + exits 1 with "Session already exists" (it does NOT block), which is why + _ensure_session treats that message as success + """ + # Implementation verified in test_terminals.py: test_open_in_session_reuses_existing_session + + def test_zellij_list_panes_json_fields(self) -> None: + """`zellij action list-panes --json` exposes pane_cwd, tab_id, and tab_name. + + Evidence: + Source: zellij action list-panes --help and live output (zellij 0.44.3) + `list-panes` was added in 0.44.0: CHANGELOG "feat: allow querying tab + info ... as well as general info about all tabs" (zellij-org/zellij#4695) + Verified: 2026-06-10 live on zellij 0.44.3 — JSON pane objects contain + keys: id, is_plugin, tab_id, tab_name, pane_cwd (among others), and the + command works against detached sessions via `--session`. + Panes inside zellij also receive ZELLIJ_PANE_ID (verified live by + dumping `env` in a spawned tab), which _current_tab uses to find the + tab owning this process. + """ + # Implementation verified in test_terminals.py: test_list_tabs_for_worktree_matches_pane_cwd + # and test_current_tab_resolves_own_pane + + def test_zellij_close_tab_by_id(self) -> None: + """Specific tabs are closed from outside via `zellij action close-tab-by-id `. + + Evidence: + Source: zellij action close-tab-by-id --help (zellij 0.44.3) + Added in 0.44.0: CHANGELOG "feat: add --pane-id and --tab-id to all + relevant CLI actions" (zellij-org/zellij#4846) + Note: `go-to-tab-name` + `close-tab` does NOT work on detached sessions + (no focused tab without an attached client), so stable tab IDs are the + only reliable way to close tabs from outside. + Verified: 2026-06-10 live on zellij 0.44.3 — closed a tab in a detached + session by ID; `query-tab-names` confirmed removal + """ + # Implementation verified in test_terminals.py: test_close_tabs_for_worktree_closes_by_id + + def test_zellij_list_sessions_format(self) -> None: + """`zellij list-sessions --no-formatting` marks dead sessions with (EXITED. + + Evidence: + Source: zellij list-sessions --help (zellij 0.44.3) + Quote: "-n, --no-formatting Do not add colors and formatting to the + list (useful for parsing)" + Format: " [Created ...]" for live sessions and + " [Created ...] (EXITED - attach to resurrect)" for dead + ones; exits non-zero when no sessions exist. + Note: `--short` prints bare names but does not distinguish dead + sessions, so liveness parsing must use --no-formatting. + Verified: 2026-06-10 live on zellij 0.44.3 + """ + # Implementation verified in test_terminals.py: test_live_session_names_handles_no_sessions def test_zellij_write_enter_byte(self) -> None: - """Zellij sends Enter key via `zellij action write 10` (byte 10 = newline). + """Zellij < 0.44 sends Enter via `zellij action write 10` (byte 10 = newline). Evidence: Source: zellij action write --help Quote: "Write bytes to the terminal" - Note: Byte 10 is ASCII newline (Enter key) + Note: Byte 10 is ASCII newline (Enter key); only used on the legacy + (zellij < 0.44) tab-opening path, where new-tab cannot run a command Verified: 2026-01-11 via `zellij action write --help` """ # Byte 10 = newline verified via ASCII table From 9df3c929d545d84938b6fa80b35356181651b6c8 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Wed, 10 Jun 2026 10:42:13 -0700 Subject: [PATCH 2/3] Address PR review feedback on zellij support - Parse session names from list-sessions with a regex so names containing spaces are preserved (was: first token only) - Fail safe in close_tabs_for_worktree: when inside zellij and the current tab cannot be resolved, skip all tabs in the current session instead of risking closing our own tab - Cache the zellij version check per instance to avoid repeated `zellij --version` subprocess calls --- agent_cli/dev/terminals/zellij.py | 27 +++++++++++--- tests/dev/test_terminals.py | 62 ++++++++++++++++++++++++++++++- 2 files changed, 82 insertions(+), 7 deletions(-) diff --git a/agent_cli/dev/terminals/zellij.py b/agent_cli/dev/terminals/zellij.py index 85b7f7ace..44ecb51d8 100644 --- a/agent_cli/dev/terminals/zellij.py +++ b/agent_cli/dev/terminals/zellij.py @@ -169,7 +169,9 @@ def close_tabs_for_worktree(self, worktree_path: Path) -> ZellijCleanupResult: inventory = self.list_tabs_for_worktree(worktree_path) errors: list[str] = [inventory.error] if inventory.error else [] - current_tab = self._current_tab() if inventory.tabs else None + current_session = self.current_session_name() + in_zellij = current_session is not None and os.environ.get("ZELLIJ_PANE_ID") is not None + current_tab = self._current_tab() if inventory.tabs and in_zellij else None closed_tabs: list[ZellijTab] = [] for tab in inventory.tabs: if current_tab == (tab.session_name, tab.tab_id): @@ -178,6 +180,14 @@ def close_tabs_for_worktree(self, worktree_path: Path) -> ZellijCleanupResult: "because it is the current tab", ) continue + # Fail safe: if we are inside zellij but could not resolve our own tab, + # never close tabs in the session we are running in. + if in_zellij and current_tab is None and tab.session_name == current_session: + errors.append( + f"Skipped zellij tab {tab.tab_id} in session {tab.session_name} " + "because the current tab could not be determined", + ) + continue try: subprocess.run( [ # noqa: S607 @@ -244,10 +254,14 @@ def _open_new_tab_legacy( except subprocess.CalledProcessError: return False + _supports_control: bool | None = None + def _supports_cli_control(self) -> bool: """Whether the installed zellij supports tab IDs and detached-session control.""" - version = self._cli_version() - return version is not None and version >= MIN_CONTROL_VERSION + if self._supports_control is None: + version = self._cli_version() + self._supports_control = version is not None and version >= MIN_CONTROL_VERSION + return self._supports_control @staticmethod def _cli_version() -> tuple[int, int, int] | None: @@ -296,10 +310,13 @@ def _live_session_names() -> list[str]: return [] names: list[str] = [] for raw_line in result.stdout.splitlines(): + # Lines look like " [Created ...]" with an optional + # "(EXITED - attach to resurrect)" suffix; names may contain spaces. line = raw_line.strip() - if not line or "(EXITED" in line or line.lower().startswith("no active"): + match = re.match(r"(.+?) \[Created ", line) + if match is None or "(EXITED" in line[match.end() :]: continue - names.append(line.split()[0]) + names.append(match.group(1)) return names @staticmethod diff --git a/tests/dev/test_terminals.py b/tests/dev/test_terminals.py index 561444c41..7c436b0f6 100644 --- a/tests/dev/test_terminals.py +++ b/tests/dev/test_terminals.py @@ -608,8 +608,14 @@ def test_list_tabs_for_worktree_matches_pane_cwd(self, tmp_path: Path) -> None: "--json", ] - def test_close_tabs_for_worktree_closes_by_id(self, tmp_path: Path) -> None: + def test_close_tabs_for_worktree_closes_by_id( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: """Worktree tabs are closed via close-tab-by-id.""" + monkeypatch.delenv("ZELLIJ_SESSION_NAME", raising=False) + monkeypatch.delenv("ZELLIJ_PANE_ID", raising=False) terminal = Zellij() inventory = ZellijInventory( tabs=(ZellijTab(tab_id=2, session_name="work", tab_name="feature"),), @@ -632,8 +638,14 @@ def test_close_tabs_for_worktree_closes_by_id(self, tmp_path: Path) -> None: "2", ] - def test_close_tabs_for_worktree_skips_current_tab(self, tmp_path: Path) -> None: + def test_close_tabs_for_worktree_skips_current_tab( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: """The tab running this process is never closed.""" + monkeypatch.setenv("ZELLIJ_SESSION_NAME", "work") + monkeypatch.setenv("ZELLIJ_PANE_ID", "7") terminal = Zellij() inventory = ZellijInventory( tabs=( @@ -653,6 +665,33 @@ def test_close_tabs_for_worktree_skips_current_tab(self, tmp_path: Path) -> None assert "current tab" in result.errors[0] assert len(mock_run.call_args_list) == 1 + def test_close_tabs_for_worktree_fails_safe_when_current_tab_unknown( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Inside zellij, tabs in the current session are kept if own tab lookup fails.""" + monkeypatch.setenv("ZELLIJ_SESSION_NAME", "work") + monkeypatch.setenv("ZELLIJ_PANE_ID", "7") + terminal = Zellij() + inventory = ZellijInventory( + tabs=( + ZellijTab(tab_id=2, session_name="work", tab_name="feature"), + ZellijTab(tab_id=3, session_name="elsewhere", tab_name="feature"), + ), + ) + with ( + patch.object(terminal, "list_tabs_for_worktree", return_value=inventory), + patch.object(terminal, "_current_tab", return_value=None), + patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, + ): + result = terminal.close_tabs_for_worktree(tmp_path) + + assert result.closed_tabs == (inventory.tabs[1],) + assert len(result.errors) == 1 + assert "could not be determined" in result.errors[0] + assert len(mock_run.call_args_list) == 1 + def test_current_tab_resolves_own_pane(self, monkeypatch: pytest.MonkeyPatch) -> None: """The current tab is resolved from ZELLIJ_PANE_ID via list-panes.""" monkeypatch.setenv("ZELLIJ_SESSION_NAME", "work") @@ -673,6 +712,25 @@ def test_live_session_names_handles_no_sessions(self) -> None: with patch("subprocess.run", side_effect=error): assert Zellij._live_session_names() == [] + def test_live_session_names_preserves_spaces_and_skips_dead(self) -> None: + """Session names with spaces parse correctly; EXITED and notice lines are skipped.""" + output = ( + "my session [Created 5s ago] \n" + "plain [Created 2m 3s ago] \n" + "old one [Created 1day 2h ago] (EXITED - attach to resurrect)\n" + "No active zellij sessions found.\n" + ) + with patch("subprocess.run", return_value=MagicMock(returncode=0, stdout=output)): + assert Zellij._live_session_names() == ["my session", "plain"] + + def test_supports_cli_control_caches_version_lookup(self) -> None: + """The zellij version is only queried once per instance.""" + terminal = Zellij() + with patch.object(Zellij, "_cli_version", return_value=(0, 44, 3)) as mock_version: + assert terminal._supports_cli_control() is True + assert terminal._supports_cli_control() is True + mock_version.assert_called_once() + class TestKitty: """Tests for Kitty terminal.""" From b6de62f8feadf68b24f212800b24f3b2dcdcbd66 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Wed, 10 Jun 2026 11:02:42 -0700 Subject: [PATCH 3/3] Fix zellij tests on Windows: compare against str(path), not literal POSIX paths --- tests/dev/test_terminals.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/tests/dev/test_terminals.py b/tests/dev/test_terminals.py index 7c436b0f6..f4436deb9 100644 --- a/tests/dev/test_terminals.py +++ b/tests/dev/test_terminals.py @@ -399,6 +399,7 @@ def test_cli_version_parsing(self) -> None: def test_open_in_session_creates_detached_session_when_missing(self) -> None: """Outside zellij, a named session is created in detached mode if absent.""" terminal = Zellij() + path = Path("/some/path") with ( patch.object(terminal, "is_available", return_value=True), patch.object(terminal, "_supports_cli_control", return_value=True), @@ -411,7 +412,7 @@ def test_open_in_session_creates_detached_session_when_missing(self) -> None: ) as mock_run, ): handle = terminal.open_in_session( - Path("/some/path"), + path, "echo hello", tab_name="feature", session_name="repo-session", @@ -433,7 +434,7 @@ def test_open_in_session_creates_detached_session_when_missing(self) -> None: "action", "new-tab", "--cwd", - "/some/path", + str(path), "--name", "feature", "--", @@ -508,13 +509,14 @@ def test_open_new_tab_falls_back_to_legacy_on_old_zellij( """Zellij < 0.44 opens tabs via the legacy write-chars path.""" monkeypatch.setenv("ZELLIJ", "0") terminal = Zellij() + path = Path("/some/path") with ( patch.object(terminal, "is_available", return_value=True), patch.object(Zellij, "_cli_version", return_value=(0, 41, 2)), patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, patch("time.sleep"), ): - result = terminal.open_new_tab(Path("/some/path"), "echo hello", tab_name="feature") + result = terminal.open_new_tab(path, "echo hello", tab_name="feature") assert result is True commands = [call.args[0] for call in mock_run.call_args_list] @@ -526,7 +528,7 @@ def test_open_new_tab_falls_back_to_legacy_on_old_zellij( "--layout", "default", "--cwd", - "/some/path", + str(path), "--name", "feature", ], @@ -542,12 +544,13 @@ def test_open_new_tab_uses_open_in_session_on_modern_zellij( monkeypatch.setenv("ZELLIJ", "0") monkeypatch.setenv("ZELLIJ_SESSION_NAME", "current-session") terminal = Zellij() + path = Path("/some/path") with ( patch.object(terminal, "is_available", return_value=True), patch.object(Zellij, "_cli_version", return_value=(0, 44, 0)), patch("subprocess.run", return_value=MagicMock(returncode=0, stdout="2\n")) as mock_run, ): - result = terminal.open_new_tab(Path("/some/path"), "echo hello", tab_name="feature") + result = terminal.open_new_tab(path, "echo hello", tab_name="feature") assert result is True assert mock_run.call_args.args[0] == [ @@ -555,7 +558,7 @@ def test_open_new_tab_uses_open_in_session_on_modern_zellij( "action", "new-tab", "--cwd", - "/some/path", + str(path), "--name", "feature", "--",