Add zellij multiplexer support to agent-cli dev

agent_cli/dev/cleanup.py

@@ -9,14 +9,15 @@
 
 from . import worktree
 from .terminals.tmux import Tmux
+from .terminals.zellij import Zellij
 
 if TYPE_CHECKING:
     from pathlib import Path
 
 
 @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

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
 

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

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",