Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 48 additions & 9 deletions agent_cli/dev/cleanup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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
67 changes: 54 additions & 13 deletions agent_cli/dev/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -571,6 +596,7 @@ def new(
task_file,
agent_env,
multiplexer_name=multiplexer,
tmux_session=tmux_session,
)

# Print summary
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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(
Expand All @@ -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()

Expand Down Expand Up @@ -1090,6 +1126,7 @@ def start_agent(
task_file,
agent_env,
multiplexer_name=multiplexer,
tmux_session=tmux_session,
)
if handle:
info(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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")
Expand Down
2 changes: 1 addition & 1 deletion agent_cli/dev/coding_agents/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = []
Expand Down
16 changes: 11 additions & 5 deletions agent_cli/dev/launch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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(
Expand All @@ -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}")
Expand All @@ -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":
Expand All @@ -331,6 +333,7 @@ def _launch_in_terminal(
tab_name,
repo_root,
multiplexer_name,
tmux_session,
)
return handle is not None, handle

Expand All @@ -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
)
Expand All @@ -370,7 +375,8 @@ def launch_agent(
full_cmd,
tab_name,
repo_root,
multiplexer_name,
effective_multiplexer_name,
tmux_session,
)
if launched:
return handle
Expand Down
Loading
Loading