Merge pull request #206 from pchalasani/feat/amux-project-filters

Add repository and directory filters to Amux

Author: pchalasani

claude_code_tools/amux/cli.py

@@ -20,6 +20,7 @@
 import sys
 
 from . import cache, render, scan
+from .filters import directory_argument, select_agents
 from .model import Agent
 
 _FZF_HEADER = (
@@ -46,6 +47,9 @@ def _agents_for_display(max_age: float) -> tuple[list[Agent], bool]:
 
 def _display(agents: list[Agent], args: argparse.Namespace) -> list[Agent]:
     """Refresh ages even for cached rows, then apply the requested view."""
+    agents = select_agents(
+        agents, getattr(args, "repo", None), getattr(args, "directory", None),
+    )
     for agent in agents:
         agent.classify_inactivity(args.dormant_hours)
     if args.dormant:
@@ -112,6 +116,14 @@ def cmd_pick(args: argparse.Namespace) -> int:
         f" --dormant-hours {args.dormant_hours} --sort {args.sort}"
         + (" --dormant" if args.dormant else "")
     )
+    # Keep user values out of fzf's action/placeholder parser as well as the shell.
+    reload_env = dict(os.environ)
+    if args.repo is not None:
+        reload_env["AMUX_RELOAD_REPO"] = args.repo
+        view_flags += ' --repo="$AMUX_RELOAD_REPO"'
+    if args.directory is not None:
+        reload_env["AMUX_RELOAD_DIR"] = args.directory
+        view_flags += ' --dir="$AMUX_RELOAD_DIR"'
     binds = [
         f"ctrl-r:reload({self_cmd} rows --refresh{view_flags})",
         # Opening on cached rows is what makes this instant; refresh the
@@ -153,6 +165,7 @@ def cmd_pick(args: argparse.Namespace) -> int:
         input=render.picker_lines(agents),
         stdout=subprocess.PIPE,
         text=True,
+        env=reload_env,
     )
     if proc.returncode != 0 or not proc.stdout.strip():
         return 0
@@ -215,6 +228,14 @@ def build_parser() -> argparse.ArgumentParser:
             help="seconds a cached scan stays usable (default: 30)",
         )
     for command in (pick, lst, rows):
+        command.add_argument(
+            "--repo", "--project", dest="repo", metavar="NAME",
+            help="exact repository name, including linked Git worktrees",
+        )
+        command.add_argument(
+            "--dir", dest="directory", type=directory_argument, metavar="PATH",
+            help="only agents in this directory or its subdirectories",
+        )
         command.add_argument(
             "--dormant", action="store_true",
             help="show waiting agents with old known submitted input only",
@@ -238,7 +259,20 @@ def main(argv: list[str] | None = None) -> int:
     # only ["pick"] -- reparsing dropped global options, so `amux --max-age 0`
     # silently used the 30s default.
     known = {"pick", "list", "scan", "rows"}
-    if not any(tok in known | {"-h", "--help"} for tok in raw):
+    value_options = {
+        "--repo", "--project", "--dir", "--max-age", "--dormant-hours", "--sort",
+    }
+    tokens = iter(raw)
+    explicit_command = False
+    for token in tokens:
+        if token in value_options:
+            next(tokens, None)
+        elif token in known | {"-h", "--help"}:
+            explicit_command = True
+            break
+        elif not token.startswith("-"):
+            break
+    if not explicit_command:
         raw.insert(0, "pick")
     args = parser.parse_args(raw)
     if not scan.tmux_available():

claude_code_tools/amux/filters.py

@@ -0,0 +1,67 @@
+"""Repository and directory selection for Amux display views."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from .model import Agent
+
+
+def directory_argument(value: str) -> str:
+    """Resolve an existing directory supplied on the command line."""
+    try:
+        path = Path(value).expanduser().resolve()
+        if path.is_dir():
+            return str(path)
+    except (OSError, RuntimeError) as exc:
+        raise argparse.ArgumentTypeError(str(exc)) from exc
+    raise argparse.ArgumentTypeError(f"not a directory: {value!r}")
+
+
+def repository_name(cwd: str) -> str:
+    """Find the shared repository's directory name, including linked worktrees."""
+    if not cwd:
+        return ""
+    try:
+        path = Path(cwd).resolve()
+        for root in (path, *path.parents):
+            marker = root / ".git"
+            if marker.is_dir():
+                return root.name
+            if marker.is_file():
+                pointer = marker.read_text().strip()
+                if not pointer.startswith("gitdir: "):
+                    return ""
+                git_dir = (root / pointer.removeprefix("gitdir: ")).resolve()
+                common_file = git_dir / "commondir"
+                if not common_file.is_file():
+                    return root.name
+                common = (git_dir / common_file.read_text().strip()).resolve()
+                return common.parent.name if common.name == ".git" else common.name
+    except (OSError, RuntimeError, UnicodeError):
+        return ""
+    return ""
+
+
+def select_agents(
+    agents: list[Agent], repo: str | None, directory: str | None,
+) -> list[Agent]:
+    """Apply exact repository-name and directory-subtree filters together."""
+    selected = []
+    names: dict[str, str] = {}
+    for agent in agents:
+        if repo is not None and agent.repo != repo:
+            if agent.cwd not in names:
+                names[agent.cwd] = repository_name(agent.cwd)
+            if names[agent.cwd] != repo:
+                continue
+        if directory:
+            if not agent.cwd:
+                continue
+            try:
+                Path(agent.cwd).resolve().relative_to(directory)
+            except (ValueError, OSError, RuntimeError):
+                continue
+        selected.append(agent)
+    return selected

docs-site/src/content/docs/tools/amux.mdx

@@ -26,6 +26,28 @@ In the picker, rows are sorted by urgency, the right-hand pane shows a live
 preview of whatever is highlighted, `enter` jumps to that pane, and `ctrl-r`
 refreshes.
 
+## Filter by repository or directory
+
+```bash
+amux list --repo proposalwriter
+amux list --repo proposalwriter --dormant --sort oldest
+amux --repo proposalwriter                         # interactive picker
+amux list --dir ~/Git/proposalwriter               # this directory and below
+amux --dir .                                      # picker for the current directory
+```
+
+`--repo NAME` (also `--project NAME`) matches the exact, case-sensitive repository
+name shown by Amux, or the main repository directory name for linked Git
+worktrees. Thus `--repo proposalwriter` also includes sessions in linked
+worktrees with different directory names. This matches local repository names,
+not GitHub `owner/repo` URLs.
+
+`--dir PATH` matches a directory and its descendants, with `~`, relative paths,
+and symlinks resolved. It does not include sibling worktrees outside that
+subtree. Combine both flags to require both matches. Filters work with table,
+JSON, cached results, and the picker; automatic and Ctrl-r refreshes preserve
+them. The shared scan cache always retains agents from all projects.
+
 ## Agent states
 
 Every pane running an agent gets one of four states, sorted most urgent first:

tests/test_amux_filters.py

@@ -0,0 +1,125 @@
+"""Project filters keep linked worktrees and picker refreshes consistent."""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from claude_code_tools.amux import cli
+from claude_code_tools.amux.filters import directory_argument, select_agents
+from claude_code_tools.amux.model import Agent
+
+
+def agent(cwd: Path, repo: str = "", pane: str = "test:1.1") -> Agent:
+    """Make a live-row fixture without touching agent processes."""
+    return Agent(pane, "test", "claude", cwd=str(cwd), repo=repo)
+
+
+def test_repository_filter_includes_linked_worktree(tmp_path: Path) -> None:
+    repo = tmp_path / "project"
+    linked = tmp_path / "feature-tree"
+    subprocess.run(["git", "init", "-q", str(repo)], check=True)
+    subprocess.run([
+        "git", "-C", str(repo), "-c", "user.name=Test",
+        "-c", "user.email=test@example.invalid", "commit", "-qm", "initial",
+        "--allow-empty",
+    ], check=True)
+    subprocess.run([
+        "git", "-C", str(repo), "worktree", "add", "-qb", "feature", str(linked),
+    ], check=True)
+    (linked / "nested").mkdir()
+    rows = [agent(repo, "project"), agent(linked / "nested", "feature-tree"),
+            agent(tmp_path / "unrelated", "project-other")]
+    assert select_agents(rows, "project", None) == rows[:2]
+    assert select_agents(rows, "feature-tree", None) == [rows[1]]
+    assert select_agents(rows, "Project", None) == []
+
+
+def test_directory_boundary_symlink_and_combination(tmp_path: Path) -> None:
+    repo = tmp_path / "project"
+    (repo / "nested").mkdir(parents=True)
+    sibling = tmp_path / "project-other"
+    sibling.mkdir()
+    alias = tmp_path / "alias"
+    alias.symlink_to(repo, target_is_directory=True)
+    rows = [agent(repo, "project"), agent(repo / "nested", "project"),
+            agent(sibling, "project"), agent(repo, "other"),
+            Agent("test:1.9", "test", "claude", repo="project")]
+    selected = select_agents(rows, "project", directory_argument(str(alias)))
+    assert selected == rows[:2]
+    assert select_agents(rows, None, str(repo)) == rows[:2] + [rows[3]]
+
+
+def test_directory_argument_rejects_file_or_missing(tmp_path: Path) -> None:
+    regular = tmp_path / "file"
+    regular.write_text("content")
+    for path in [regular, tmp_path / "missing"]:
+        with pytest.raises(argparse.ArgumentTypeError):
+            directory_argument(str(path))
+
+
+def test_directory_argument_expands_relative_path(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    assert directory_argument(".") == str(tmp_path.resolve())
+
+
+@pytest.mark.parametrize("command", ["list", "pick", "rows"])
+def test_filters_combine_with_dormancy(command: str, tmp_path: Path) -> None:
+    args = cli.build_parser().parse_args([
+        command, "--project", "project", "--dir", str(tmp_path),
+        "--dormant", "--sort", "oldest",
+    ])
+    old = agent(tmp_path, "project")
+    old.state, old.last_input_at = "idle", 1
+    busy = agent(tmp_path, "project", "test:1.2")
+    busy.state, busy.last_input_at = "busy", 1
+    other = agent(tmp_path, "other", "test:1.3")
+    other.state, other.last_input_at = "idle", 1
+    assert cli._display([busy, other, old], args) == [old]
+
+
+@pytest.mark.parametrize("name", ["list", "pick", "scan", "rows"])
+def test_repo_named_like_subcommand_defaults_to_picker(
+    name: str, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    seen = []
+    monkeypatch.setattr(cli.scan, "tmux_available", lambda: True)
+    monkeypatch.setattr(cli, "cmd_pick", lambda args: seen.append(args) or 0)
+    assert cli.main(["--repo", name]) == 0
+    assert seen[0].repo == name
+
+
+def test_picker_refresh_preserves_literal_filter_values(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    directory = tmp_path / "dir (x), {q} $HOME ' space"
+    directory.mkdir()
+    repo = "repo (x), {q} $(echo bad) '"
+    row = agent(directory, repo)
+    captured = {}
+    monkeypatch.setattr(cli.shutil, "which", lambda _: "/usr/bin/fzf")
+    monkeypatch.setattr(cli.sys.stdin, "isatty", lambda: True)
+    monkeypatch.setattr(cli.sys.stdout, "isatty", lambda: True)
+    monkeypatch.setattr(cli, "_agents_for_display", lambda _: ([row], True))
+
+    def run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
+        captured.update(command=cmd, **kwargs)
+        return subprocess.CompletedProcess(cmd, 1, stdout="")
+
+    monkeypatch.setattr(cli.subprocess, "run", run)
+    args = cli.build_parser().parse_args([
+        "pick", "--repo", repo, "--dir", str(directory),
+    ])
+    assert cli.cmd_pick(args) == 0
+    binds = [x for x in captured["command"] if "reload" in x]
+    assert len(binds) == 2
+    assert all('--repo="$AMUX_RELOAD_REPO"' in x for x in binds)
+    assert all('--dir="$AMUX_RELOAD_DIR"' in x for x in binds)
+    assert all(repo not in x and str(directory) not in x for x in binds)
+    assert captured["env"]["AMUX_RELOAD_REPO"] == repo
+    assert captured["env"]["AMUX_RELOAD_DIR"] == str(directory)