diff --git a/agent_cli/dev/cleanup.py b/agent_cli/dev/cleanup.py index 0c0dbc57..ce54b074 100644 --- a/agent_cli/dev/cleanup.py +++ b/agent_cli/dev/cleanup.py @@ -4,14 +4,26 @@ import json import subprocess +from dataclasses import dataclass, field from typing import TYPE_CHECKING from . import worktree +from .terminals.tmux import Tmux if TYPE_CHECKING: from pathlib import Path +@dataclass +class RemoveWorktreeResult: + """Outcome of removing a worktree and any tagged tmux windows.""" + + name: str + success: bool + error: str | None = None + warnings: list[str] = field(default_factory=list) + + def find_worktrees_with_no_commits(repo_root: Path) -> list[worktree.WorktreeInfo]: """Find worktrees whose branches have no commits ahead of the default branch.""" worktrees_list = worktree.list_worktrees() @@ -95,18 +107,45 @@ def remove_worktrees( repo_root: Path, *, force: bool = False, -) -> list[tuple[str, bool, str | None]]: +) -> list[RemoveWorktreeResult]: """Remove a list of worktrees. - Returns list of (branch_name, success, error_message) tuples. + Returns a result for each worktree removal attempt. """ - results: list[tuple[str, bool, str | None]] = [] - for wt in worktrees_to_remove: - success, error = worktree.remove_worktree( - wt.path, + return [ + remove_worktree( + wt, + repo_root, force=force, delete_branch=True, - repo_path=repo_root, ) - results.append((wt.branch or wt.path.name, success, error)) - return results + for wt in worktrees_to_remove + ] + + +def remove_worktree( + wt: worktree.WorktreeInfo, + repo_root: Path, + *, + force: bool = False, + delete_branch: bool = False, +) -> RemoveWorktreeResult: + """Remove one worktree and then clean up any tagged tmux windows.""" + removed, error = worktree.remove_worktree( + wt.path, + force=force, + delete_branch=delete_branch, + repo_path=repo_root, + ) + result = RemoveWorktreeResult( + name=wt.branch or wt.path.name, + success=removed, + error=error, + ) + if not removed: + return result + + tmux = Tmux() + tmux_cleanup = tmux.kill_windows_for_worktree(wt.path) + result.warnings.extend(tmux_cleanup.errors) + return result diff --git a/agent_cli/dev/cli.py b/agent_cli/dev/cli.py index 68c690e6..23ed891d 100644 --- a/agent_cli/dev/cli.py +++ b/agent_cli/dev/cli.py @@ -157,6 +157,23 @@ def _resolve_prompt_text( return prompt +def _normalize_tmux_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: + error("tmux session names cannot contain ':'") + + return normalized_session, "tmux" + + def _resolve_dev_new_agent_request( *, start_agent: bool, @@ -440,6 +457,13 @@ def new( 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", ), ] = None, + tmux_session: Annotated[ + str | None, + typer.Option( + "--tmux-session", + help="Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux", + ), + ] = None, hooks: Annotated[ bool, typer.Option( @@ -489,6 +513,7 @@ def new( agent_name_deprecated=agent_name_deprecated, prompt=prompt, ) + tmux_session, multiplexer = _normalize_tmux_session(tmux_session, multiplexer) repo_root = _ensure_git_repo() @@ -571,6 +596,7 @@ def new( task_file, agent_env, multiplexer_name=multiplexer, + tmux_session=tmux_session, ) # Print summary @@ -873,17 +899,19 @@ def remove( if not typer.confirm("Continue?"): raise typer.Abort - removed, remove_err = worktree.remove_worktree( - wt.path, + result = cleanup.remove_worktree( + wt, + repo_root, force=force, delete_branch=delete_branch, - repo_path=repo_root, ) - if removed: + if result.success: success(f"Removed worktree: {wt.path}") + for cleanup_warning in result.warnings: + warn(cleanup_warning) else: - error(remove_err or "Failed to remove worktree") + error(result.error or "Failed to remove worktree") @app.command("path") @@ -1018,6 +1046,13 @@ def start_agent( help="Launch the agent in a specific multiplexer instead of the current terminal. Currently supported: tmux", ), ] = None, + tmux_session: Annotated[ + str | None, + typer.Option( + "--tmux-session", + help="Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux", + ), + ] = None, hooks: Annotated[ bool, typer.Option( @@ -1043,6 +1078,7 @@ 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) repo_root = _ensure_git_repo() @@ -1090,6 +1126,7 @@ def start_agent( task_file, agent_env, multiplexer_name=multiplexer, + tmux_session=tmux_session, ) if handle: info( @@ -1367,11 +1404,13 @@ def _clean_merged_worktrees( info("[dry-run] Would remove the above worktrees") elif yes or typer.confirm("\nRemove these worktrees?"): results = cleanup.remove_worktrees([wt for wt, _ in to_remove], repo_root, force=force) - for branch, ok, remove_err in results: - if ok: - success(f"Removed {branch}") + for result in results: + if result.success: + success(f"Removed {result.name}") + for cleanup_warning in result.warnings: + warn(cleanup_warning) else: - warn(f"Failed to remove {branch}: {remove_err}") + warn(f"Failed to remove {result.name}: {result.error}") def _clean_no_commits_worktrees( @@ -1401,11 +1440,13 @@ def _clean_no_commits_worktrees( info("[dry-run] Would remove the above worktrees") elif yes or typer.confirm("\nRemove these worktrees?"): results = cleanup.remove_worktrees(to_remove, repo_root, force=force) - for branch, ok, remove_err in results: - if ok: - success(f"Removed {branch}") + for result in results: + if result.success: + success(f"Removed {result.name}") + for cleanup_warning in result.warnings: + warn(cleanup_warning) else: - warn(f"Failed to remove {branch}: {remove_err}") + warn(f"Failed to remove {result.name}: {result.error}") @app.command("clean") diff --git a/agent_cli/dev/coding_agents/base.py b/agent_cli/dev/coding_agents/base.py index d329a86e..89b41bda 100644 --- a/agent_cli/dev/coding_agents/base.py +++ b/agent_cli/dev/coding_agents/base.py @@ -143,7 +143,7 @@ def _get_parent_process_names() -> list[str]: - CLI tools that set process.title (like Claude) show their name directly """ try: - import psutil # noqa: PLC0415 + import psutil # type: ignore[import-untyped] # noqa: PLC0415 process = psutil.Process(os.getpid()) names = [] diff --git a/agent_cli/dev/launch.py b/agent_cli/dev/launch.py index 1dcd8b96..0b003772 100644 --- a/agent_cli/dev/launch.py +++ b/agent_cli/dev/launch.py @@ -280,6 +280,7 @@ def _launch_in_tmux( tab_name: str, repo_root: Path | None, multiplexer_name: str | None, + tmux_session: str | None, ) -> TerminalHandle | None: """Launch an agent via tmux and return its pane handle.""" from .terminals.tmux import Tmux # noqa: PLC0415 @@ -289,8 +290,8 @@ def _launch_in_tmux( return None requested_tmux = multiplexer_name == "tmux" - session_name = None - if requested_tmux and not terminal.detect(): + session_name = tmux_session + if session_name is None and requested_tmux and not terminal.detect(): session_name = terminal.session_name_for_repo(repo_root or path) handle = terminal.open_in_session( @@ -305,7 +306,7 @@ def _launch_in_tmux( session_label = ( f" in tmux session {handle.session_name}" - if requested_tmux and 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}") @@ -320,6 +321,7 @@ def _launch_in_terminal( tab_name: str, repo_root: Path | None, multiplexer_name: str | None, + tmux_session: str | None, ) -> tuple[bool, TerminalHandle | None]: """Launch an agent in the resolved terminal.""" if terminal.name == "tmux": @@ -331,6 +333,7 @@ def _launch_in_terminal( tab_name, repo_root, multiplexer_name, + tmux_session, ) return handle is not None, handle @@ -350,13 +353,15 @@ def launch_agent( task_file: Path | None = None, env: dict[str, str] | None = None, multiplexer_name: str | None = None, + tmux_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. """ - terminal = _resolve_launch_terminal(multiplexer_name) + effective_multiplexer_name = "tmux" if tmux_session is not None else multiplexer_name + terminal = _resolve_launch_terminal(effective_multiplexer_name) full_cmd = _build_agent_launch_command( path, agent, extra_args, prompt, task_file, env, terminal ) @@ -370,7 +375,8 @@ def launch_agent( full_cmd, tab_name, repo_root, - multiplexer_name, + effective_multiplexer_name, + tmux_session, ) if launched: return handle diff --git a/agent_cli/dev/terminals/tmux.py b/agent_cli/dev/terminals/tmux.py index faac99c5..fb63b212 100644 --- a/agent_cli/dev/terminals/tmux.py +++ b/agent_cli/dev/terminals/tmux.py @@ -7,6 +7,7 @@ import re import shutil import subprocess +from dataclasses import dataclass from typing import TYPE_CHECKING from .base import Terminal, TerminalHandle @@ -14,6 +15,34 @@ if TYPE_CHECKING: from pathlib import Path +WORKTREE_OPTION = "@agent_cli_worktree" +WINDOW_LIST_FORMAT = "#{window_id}\t#{session_name}\t#{window_name}\t#{@agent_cli_worktree}" + + +@dataclass(frozen=True) +class TmuxWindow: + """A tmux window discovered via cross-session inventory.""" + + window_id: str + session_name: str + window_name: str + + +@dataclass(frozen=True) +class TmuxInventory: + """Tagged tmux windows for a worktree, plus any lookup error.""" + + windows: tuple[TmuxWindow, ...] = () + error: str | None = None + + +@dataclass(frozen=True) +class TmuxCleanupResult: + """Result of killing tagged tmux windows for a worktree.""" + + killed_windows: tuple[TmuxWindow, ...] = () + errors: tuple[str, ...] = () + class Tmux(Terminal): """tmux - Terminal multiplexer.""" @@ -49,6 +78,27 @@ def current_session_name(self) -> str | None: session_name = result.stdout.strip() return session_name or None + def current_window_id(self) -> str | None: + """Get the tmux window id for the current pane, when available.""" + cmd = ["tmux", "display-message", "-p"] + pane_id = os.environ.get("TMUX_PANE") + if pane_id: + cmd.extend(["-t", pane_id]) + cmd.append("#{window_id}") + try: + result = subprocess.run( + cmd, + check=True, + capture_output=True, + text=True, + ) + except subprocess.CalledProcessError as e: + if self._is_server_unavailable_error(e): + return None + return None + window_id = result.stdout.strip() + return window_id or None + def open_in_session( self, path: Path, @@ -84,6 +134,83 @@ def open_in_session( return self._open_window(path, command, tab_name, session_name=session_name) + def list_windows_for_worktree(self, worktree_path: Path) -> TmuxInventory: + """List tagged tmux windows for a worktree across all sessions.""" + if not self.is_available(): + return TmuxInventory() + + normalized_path = self._normalize_worktree_path(worktree_path) + try: + result = subprocess.run( + ["tmux", "list-windows", "-a", "-F", WINDOW_LIST_FORMAT], # noqa: S607 + check=True, + capture_output=True, + text=True, + ) + except subprocess.CalledProcessError as e: + 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)}", + ) + + windows: list[TmuxWindow] = [] + for line in result.stdout.splitlines(): + if not line: + continue + parts = line.split("\t", maxsplit=3) + if len(parts) != 4: # noqa: PLR2004 + continue + window_id, session_name, window_name, tagged_path = parts + if tagged_path != normalized_path: + continue + windows.append( + TmuxWindow( + window_id=window_id, + session_name=session_name, + window_name=window_name, + ), + ) + + return TmuxInventory(windows=tuple(windows)) + + def kill_windows_for_worktree(self, worktree_path: Path) -> TmuxCleanupResult: + """Kill tagged tmux windows for a worktree across all sessions.""" + inventory = self.list_windows_for_worktree(worktree_path) + if inventory.error is not None: + return TmuxCleanupResult(errors=(inventory.error,)) + + killed_windows: list[TmuxWindow] = [] + errors: list[str] = [] + current_window_id = self.current_window_id() if inventory.windows else None + for window in inventory.windows: + if current_window_id is not None and window.window_id == current_window_id: + errors.append( + "Skipped tmux window " + f"{window.window_id} in session {window.session_name} " + "because it is the current window", + ) + continue + try: + subprocess.run( + ["tmux", "kill-window", "-t", window.window_id], # noqa: S607 + check=True, + capture_output=True, + text=True, + ) + 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)}", + ) + continue + killed_windows.append(window) + + return TmuxCleanupResult( + killed_windows=tuple(killed_windows), + errors=tuple(errors), + ) + def open_new_tab( self, path: Path, @@ -160,8 +287,47 @@ def _spawn_target( pane_id = result.stdout.strip() if not pane_id: return None + self._tag_window_for_worktree(pane_id, path) return TerminalHandle( terminal_name=self.name, handle=pane_id, session_name=session_name, ) + + @staticmethod + 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.""" + stderr = exc.stderr.lower() if exc.stderr else "" + return "no server running" in stderr or "no current client" in stderr + + def _tag_window_for_worktree(self, pane_id: str, worktree_path: Path) -> None: + """Tag a tmux window with the owning worktree path for later cleanup.""" + tmux_executable = shutil.which("tmux") + if tmux_executable is None: + return + + subprocess.run( + [ + tmux_executable, + "set-option", + "-w", + "-t", + pane_id, + WORKTREE_OPTION, + self._normalize_worktree_path(worktree_path), + ], + capture_output=True, + check=False, + ) diff --git a/docs/commands/dev.md b/docs/commands/dev.md index 794dc99c..cd19df70 100644 --- a/docs/commands/dev.md +++ b/docs/commands/dev.md @@ -85,6 +85,7 @@ agent-cli dev new [BRANCH] [OPTIONS] | `--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 | +| `--tmux-session` | - | Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux | | `--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 | @@ -297,6 +298,7 @@ agent-cli dev agent NAME [--agent/-a AGENT] [--agent-args ARGS] [--prompt/-p PRO | `--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 | +| `--tmux-session` | - | Reuse or create a specific tmux session for the agent. Implies --multiplexer tmux | | `--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 new file mode 100644 index 00000000..2b42da2f --- /dev/null +++ b/tests/dev/test_cleanup.py @@ -0,0 +1,100 @@ +"""Tests for tmux-aware worktree cleanup helpers.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +from agent_cli.dev.cleanup import remove_worktree, remove_worktrees +from agent_cli.dev.terminals.tmux import TmuxCleanupResult, TmuxWindow +from agent_cli.dev.worktree import WorktreeInfo + + +def _worktree_info() -> WorktreeInfo: + return WorktreeInfo( + path=Path("/repo-worktrees/feature"), + branch="feature", + commit="abc", + is_main=False, + is_detached=False, + is_locked=False, + is_prunable=False, + ) + + +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.""" + wt = _worktree_info() + call_order: list[str] = [] + + def fake_remove(*_args: object, **_kwargs: object) -> tuple[bool, str | None]: + call_order.append("git") + return True, None + + def fake_kill(_path: Path) -> TmuxCleanupResult: + call_order.append("tmux") + return TmuxCleanupResult( + killed_windows=( + TmuxWindow(window_id="@2", session_name="shared", window_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, + ): + mock_tmux_cls.return_value.kill_windows_for_worktree.side_effect = fake_kill + result = remove_worktree(wt, Path("/repo"), force=True, delete_branch=True) + + assert result.success is True + assert call_order == ["git", "tmux"] + + def test_remove_worktree_surfaces_tmux_cleanup_errors_as_warnings(self) -> None: + """Tmux 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, + ): + mock_tmux_cls.return_value.kill_windows_for_worktree.return_value = TmuxCleanupResult( + errors=("Failed to kill tmux window @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"] + + def test_remove_worktree_skips_tmux_cleanup_when_git_removal_fails(self) -> None: + """Tmux cleanup should not run if git worktree removal fails.""" + wt = _worktree_info() + + with ( + patch( + "agent_cli.dev.cleanup.worktree.remove_worktree", + return_value=(False, "cannot remove"), + ), + patch("agent_cli.dev.cleanup.Tmux") as mock_tmux_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() + + def test_remove_worktrees_deletes_branches_for_cleanups(self) -> None: + """Batch cleanups should request branch deletion for each worktree.""" + wt = _worktree_info() + + with patch("agent_cli.dev.cleanup.remove_worktree") as mock_remove: + remove_worktrees([wt], Path("/repo"), force=True) + + mock_remove.assert_called_once_with( + wt, + Path("/repo"), + force=True, + delete_branch=True, + ) diff --git a/tests/dev/test_cli.py b/tests/dev/test_cli.py index 9491835e..d61b4af5 100644 --- a/tests/dev/test_cli.py +++ b/tests/dev/test_cli.py @@ -6,6 +6,7 @@ from pathlib import Path from unittest.mock import patch +import pytest from typer.testing import CliRunner from agent_cli.cli import app @@ -18,6 +19,7 @@ generate_ai_branch_name, generate_random_branch_name, ) +from agent_cli.dev.cleanup import RemoveWorktreeResult from agent_cli.dev.cli import _clean_no_commits_worktrees from agent_cli.dev.cli import app as dev_app from agent_cli.dev.coding_agents.base import CodingAgent @@ -256,7 +258,7 @@ def test_clean_no_commits_force_passed_to_remove_worktree(self) -> None: patch("agent_cli.dev.cleanup.find_worktrees_with_no_commits", return_value=[wt]), patch( "agent_cli.dev.cleanup.remove_worktrees", - return_value=[("feature", True, None)], + return_value=[RemoveWorktreeResult(name="feature", success=True)], ) as mock_remove, ): _clean_no_commits_worktrees( @@ -673,6 +675,92 @@ def test_new_shows_tmux_handle_when_requested(self, tmp_path: Path) -> None: assert "%42" in result.output assert "tmux attach -t agent-cli-repo-1234" in result.output + def test_new_tmux_session_implies_tmux_and_normalizes(self, tmp_path: Path) -> None: + """`--tmux-session` trims whitespace, preserves dots, and forces tmux.""" + 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("tmux", "%42", "my.session name"), + ) 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", + "--tmux-session", + " my.session name ", + "--no-setup", + "--no-copy-env", + "--no-fetch", + "--no-direnv", + ], + ) + + 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 "tmux attach -t 'my.session name'" 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" + 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", return_value=None) as mock_resolve_agent, + patch("agent_cli.dev.cli.launch_agent") as mock_launch, + ): + result = runner.invoke( + app, + [ + "dev", + "new", + "feature", + "--tmux-session", + "shared-session", + "--no-setup", + "--no-copy-env", + "--no-fetch", + "--no-direnv", + ], + ) + + assert result.exit_code == 0 + assert mock_resolve_agent.call_args.args == (False, None, None) + mock_launch.assert_not_called() + def test_new_short_start_agent_alias_warns(self, tmp_path: Path) -> None: """Legacy `-a` should still work but warn to use --start-agent.""" wt_path = tmp_path / "repo-worktrees" / "feature" @@ -847,6 +935,27 @@ def test_new_rejects_empty_prompt_file(self, tmp_path: Path) -> None: assert f"Prompt file is empty: {prompt_file}" in result.output mock_ensure_repo.assert_not_called() + def test_new_rejects_empty_tmux_session(self) -> None: + """Empty tmux 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", "--tmux-session", " "]) + + assert result.exit_code == 1 + assert "--tmux-session cannot be empty" in result.output + mock_ensure_repo.assert_not_called() + + @pytest.mark.parametrize("tmux_session", ["batch:1"]) + def test_new_rejects_tmux_session_with_illegal_characters(self, tmux_session: str) -> None: + """Tmux session names with tmux-illegal characters should fail early.""" + with patch("agent_cli.dev.cli._ensure_git_repo") as mock_ensure_repo: + result = runner.invoke( + app, ["dev", "new", "my-feature", "--tmux-session", tmux_session] + ) + + assert result.exit_code == 1 + assert "tmux session names cannot contain ':'" 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" @@ -994,6 +1103,43 @@ def test_agent_can_launch_in_requested_tmux(self) -> None: assert mock_launch.call_args.kwargs["multiplexer_name"] == "tmux" assert "tmux handle: %42" in result.output + def test_agent_tmux_session_implies_tmux_and_normalizes(self) -> None: + """`dev agent --tmux-session` trims whitespace, preserves dots, and forces tmux.""" + wt = WorktreeInfo( + path=Path("/repo-worktrees/feature"), + branch="feature", + commit="abc", + is_main=False, + is_detached=False, + is_locked=False, + is_prunable=False, + ) + + with ( + patch("agent_cli.dev.cli._ensure_git_repo", return_value=Path("/repo")), + patch("agent_cli.dev.worktree.find_worktree_by_name", return_value=wt), + patch("agent_cli.dev.cli.coding_agents.detect_current_agent") as mock_detect_current, + 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("tmux", "%42", "my.session name"), + ) as mock_launch, + ): + current_agent = mock_detect_current.return_value + current_agent.name = "codex" + current_agent.is_available.return_value = True + result = runner.invoke( + app, + ["dev", "agent", "feature", "--tmux-session", " my.session name "], + ) + + 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 "tmux attach -t 'my.session name'" in result.output + def test_agent_quotes_tmux_attach_hint(self) -> None: """Attach hint should quote session names that contain spaces.""" wt = WorktreeInfo( @@ -1026,6 +1172,28 @@ def test_agent_quotes_tmux_attach_hint(self) -> None: assert result.exit_code == 0 assert "tmux attach -t 'my session'" in result.output + def test_agent_rejects_empty_tmux_session(self) -> None: + """Empty tmux session names should fail before repo/worktree resolution.""" + with patch("agent_cli.dev.cli._ensure_git_repo") as mock_ensure_repo: + result = runner.invoke(app, ["dev", "agent", "feature", "--tmux-session", " "]) + + assert result.exit_code == 1 + assert "--tmux-session cannot be empty" in result.output + mock_ensure_repo.assert_not_called() + + @pytest.mark.parametrize("tmux_session", ["batch:1"]) + def test_agent_rejects_tmux_session_with_illegal_characters(self, tmux_session: str) -> None: + """Tmux session names with tmux-illegal characters should fail early.""" + with patch("agent_cli.dev.cli._ensure_git_repo") as mock_ensure_repo: + result = runner.invoke( + app, + ["dev", "agent", "feature", "--tmux-session", tmux_session], + ) + + assert result.exit_code == 1 + assert "tmux session names cannot contain ':'" in result.output + mock_ensure_repo.assert_not_called() + def test_agent_rejects_empty_prompt_file(self, tmp_path: Path) -> None: """Empty prompt files should fail before repo/worktree resolution.""" prompt_file = tmp_path / "empty.md" @@ -1330,15 +1498,15 @@ def test_rm_force_skips_confirmation(self) -> None: patch("agent_cli.dev.worktree.git_available", return_value=True), patch("agent_cli.dev.worktree.find_worktree_by_name", return_value=mock_wt), patch( - "agent_cli.dev.worktree.remove_worktree", - return_value=(True, None), + "agent_cli.dev.cleanup.remove_worktree", + return_value=RemoveWorktreeResult(name="feature", success=True), ) as mock_remove, ): # With --force, should NOT prompt and should succeed result = runner.invoke(app, ["dev", "rm", "feature", "--force"]) assert result.exit_code == 0 assert "Removed worktree" in result.output - # Verify remove_worktree was called with force=True + # Verify cleanup.remove_worktree was called with force=True mock_remove.assert_called_once() assert mock_remove.call_args[1]["force"] is True @@ -1359,8 +1527,8 @@ def test_rm_yes_skips_confirmation(self) -> None: patch("agent_cli.dev.worktree.git_available", return_value=True), patch("agent_cli.dev.worktree.find_worktree_by_name", return_value=mock_wt), patch( - "agent_cli.dev.worktree.remove_worktree", - return_value=(True, None), + "agent_cli.dev.cleanup.remove_worktree", + return_value=RemoveWorktreeResult(name="feature", success=True), ) as mock_remove, ): # With --yes, should NOT prompt and should succeed @@ -1386,8 +1554,8 @@ def test_rm_without_flags_prompts(self) -> None: patch("agent_cli.dev.worktree.git_available", return_value=True), patch("agent_cli.dev.worktree.find_worktree_by_name", return_value=mock_wt), patch( - "agent_cli.dev.worktree.remove_worktree", - return_value=(True, None), + "agent_cli.dev.cleanup.remove_worktree", + return_value=RemoveWorktreeResult(name="feature", success=True), ) as mock_remove, ): # Without --force or --yes, should prompt (and abort on 'n') @@ -1396,6 +1564,37 @@ def test_rm_without_flags_prompts(self) -> None: # remove_worktree should NOT have been called since user said no mock_remove.assert_not_called() + def test_rm_surfaces_tmux_cleanup_warnings(self) -> None: + """Tmux cleanup failures should warn without failing the removal.""" + mock_wt = WorktreeInfo( + path=Path("/repo-worktrees/feature"), + branch="feature", + commit="abc", + is_main=False, + is_detached=False, + is_locked=False, + is_prunable=False, + ) + + with ( + patch("agent_cli.dev.worktree.get_main_repo_root", return_value=Path("/repo")), + patch("agent_cli.dev.worktree.git_available", return_value=True), + patch("agent_cli.dev.worktree.find_worktree_by_name", return_value=mock_wt), + patch( + "agent_cli.dev.cleanup.remove_worktree", + return_value=RemoveWorktreeResult( + name="feature", + success=True, + warnings=["Failed to inspect tmux windows for /repo-worktrees/feature: boom"], + ), + ), + ): + result = runner.invoke(app, ["dev", "rm", "feature", "--yes"]) + + assert result.exit_code == 0 + assert "Removed worktree" in result.output + assert "Failed to inspect tmux windows" in result.output + class TestFormatEnvPrefix: """Tests for _format_env_prefix function.""" diff --git a/tests/dev/test_launch.py b/tests/dev/test_launch.py index 14869ee7..a3386329 100644 --- a/tests/dev/test_launch.py +++ b/tests/dev/test_launch.py @@ -52,6 +52,68 @@ def test_uses_requested_tmux_outside_tmux(self, tmp_path: Path) -> None: session_name="agent-cli-repo-1234", ) + def test_explicit_tmux_session_takes_precedence_outside_tmux(self, tmp_path: Path) -> None: + """`tmux_session` overrides the repo-derived detached session.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + tmux_terminal = Tmux() + handle = TerminalHandle("tmux", "%42", "shared-session") + + with ( + patch("agent_cli.dev.launch.terminals.get_terminal", return_value=tmux_terminal), + patch("agent_cli.dev.launch.terminals.detect_current_terminal", return_value=None), + patch.object(tmux_terminal, "is_available", return_value=True), + patch.object(tmux_terminal, "detect", return_value=False), + patch.object(tmux_terminal, "session_name_for_repo") as mock_session_name, + patch.object(tmux_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="tmux", tmux_session="shared-session" + ) + + assert result == handle + mock_session_name.assert_not_called() + mock_open.assert_called_once_with( + tmp_path, + "codex", + tab_name="repo@feature", + session_name="shared-session", + ) + + def test_explicit_tmux_session_takes_precedence_inside_tmux(self, tmp_path: Path) -> None: + """Explicit sessions should override the current tmux session too.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + tmux_terminal = Tmux() + handle = TerminalHandle("tmux", "%42", "shared-session") + + with ( + patch("agent_cli.dev.launch.terminals.get_terminal", return_value=tmux_terminal), + patch("agent_cli.dev.launch.terminals.detect_current_terminal", return_value=None), + patch.object(tmux_terminal, "is_available", return_value=True), + patch.object(tmux_terminal, "detect", return_value=True), + patch.object(tmux_terminal, "session_name_for_repo") as mock_session_name, + patch.object(tmux_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, tmux_session="shared-session") + + assert result == handle + mock_session_name.assert_not_called() + mock_open.assert_called_once_with( + tmp_path, + "codex", + tab_name="repo@feature", + session_name="shared-session", + ) + 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 e9a539aa..95d22f41 100644 --- a/tests/dev/test_terminals.py +++ b/tests/dev/test_terminals.py @@ -2,6 +2,7 @@ from __future__ import annotations +import subprocess from pathlib import Path from unittest.mock import MagicMock, patch @@ -15,7 +16,7 @@ get_terminal, ) from agent_cli.dev.terminals.kitty import Kitty -from agent_cli.dev.terminals.tmux import Tmux +from agent_cli.dev.terminals.tmux import Tmux, TmuxInventory, TmuxWindow from agent_cli.dev.terminals.zellij import Zellij @@ -53,9 +54,42 @@ def test_open_new_tab(self) -> None: assert result is True # Check that tmux new-window was called - call_args = mock_run.call_args + call_args = mock_run.call_args_list[0] assert "new-window" in call_args[0][0] + def test_spawn_target_tags_window_with_worktree_path(self) -> None: + """Spawned tmux windows are tagged with the owning worktree path.""" + terminal = Tmux() + with ( + patch("shutil.which", return_value="/usr/bin/tmux"), + patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout="%42\n"), + MagicMock(returncode=0), + ], + ) as mock_run, + ): + handle = terminal._spawn_target( + ["tmux", "new-window", "-t", "repo-session"], + path=Path("/some/path"), + command="echo hello", + tab_name="test", + session_name="repo-session", + ) + + assert handle is not None + assert handle.handle == "%42" + assert mock_run.call_args_list[1].args[0] == [ + "/usr/bin/tmux", + "set-option", + "-w", + "-t", + "%42", + "@agent_cli_worktree", + str(Path("/some/path").resolve()), + ] + def test_session_name_for_repo(self) -> None: """Repo session names are deterministic and shell-safe.""" terminal = Tmux() @@ -189,6 +223,96 @@ def test_open_in_session_retries_when_session_appears_during_race(self) -> None: session_name="repo-session", ) + def test_list_windows_for_worktree_searches_all_sessions(self) -> None: + """Cross-session inventory should filter tagged windows by worktree path.""" + terminal = Tmux() + stdout = ( + "@1\tsession-a\teditor\t/other/path\n" + "@2\tsession-a\tagent-one\t/some/path\n" + "@3\tsession-b\tagent-two\t/some/path" + ) + with ( + patch.object(terminal, "is_available", return_value=True), + patch("subprocess.run", return_value=MagicMock(stdout=stdout)) as mock_run, + ): + inventory = terminal.list_windows_for_worktree(Path("/some/path")) + + assert inventory.error is None + assert inventory.windows == ( + TmuxWindow(window_id="@2", session_name="session-a", window_name="agent-one"), + TmuxWindow(window_id="@3", session_name="session-b", window_name="agent-two"), + ) + assert mock_run.call_args.args[0] == [ + "tmux", + "list-windows", + "-a", + "-F", + "#{window_id}\t#{session_name}\t#{window_name}\t#{@agent_cli_worktree}", + ] + + def test_list_windows_for_worktree_returns_empty_when_no_server_running(self) -> None: + """No running tmux server should be treated as an empty inventory.""" + terminal = Tmux() + error = subprocess.CalledProcessError( + 1, + ["tmux", "list-windows", "-a", "-F", "#{window_id}"], + stderr="no server running on /tmp/tmux-1000/default\n", + ) + with ( + patch.object(terminal, "is_available", return_value=True), + patch("subprocess.run", side_effect=error), + ): + inventory = terminal.list_windows_for_worktree(Path("/some/path")) + + assert inventory == TmuxInventory() + + def test_kill_windows_for_worktree_uses_inventory_window_ids(self) -> None: + """Tagged windows are killed by window id across sessions.""" + terminal = Tmux() + inventory = TmuxInventory( + windows=( + TmuxWindow(window_id="@2", session_name="session-a", window_name="agent-one"), + TmuxWindow(window_id="@3", session_name="session-b", window_name="agent-two"), + ), + ) + with ( + patch.object(terminal, "list_windows_for_worktree", return_value=inventory), + patch.object(terminal, "current_window_id", return_value=None), + patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, + ): + cleanup = terminal.kill_windows_for_worktree(Path("/some/path")) + + assert cleanup.errors == () + assert cleanup.killed_windows == inventory.windows + assert [call.args[0] for call in mock_run.call_args_list] == [ + ["tmux", "kill-window", "-t", "@2"], + ["tmux", "kill-window", "-t", "@3"], + ] + + def test_kill_windows_for_worktree_skips_current_window(self) -> None: + """Cleanup should not kill the tmux window running the command.""" + terminal = Tmux() + inventory = TmuxInventory( + windows=( + TmuxWindow(window_id="@2", session_name="session-a", window_name="agent-one"), + TmuxWindow(window_id="@3", session_name="session-b", window_name="agent-two"), + ), + ) + with ( + patch.object(terminal, "list_windows_for_worktree", return_value=inventory), + patch.object(terminal, "current_window_id", return_value="@2"), + patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, + ): + cleanup = terminal.kill_windows_for_worktree(Path("/some/path")) + + assert cleanup.errors == ( + "Skipped tmux window @2 in session session-a because it is the current window", + ) + assert cleanup.killed_windows == (inventory.windows[1],) + assert [call.args[0] for call in mock_run.call_args_list] == [ + ["tmux", "kill-window", "-t", "@3"], + ] + class TestZellij: """Tests for Zellij terminal.""" diff --git a/tests/dev/test_verification.py b/tests/dev/test_verification.py index c56fe7d4..c9798641 100644 --- a/tests/dev/test_verification.py +++ b/tests/dev/test_verification.py @@ -202,8 +202,9 @@ def test_tmux_new_window_command(self) -> None: tab_name="test-tab", ) # Verify correct flags were used - call_args = mock_run.call_args[0][0] - assert "new-window" in call_args + call_args = next( + args[0] for args, _kwargs in mock_run.call_args_list if "new-window" in args[0] + ) assert "-c" in call_args assert "-n" in call_args assert "test-tab" in call_args