From f079ab73dff7d18fdbb72938ebe95219754db091 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 14 Jul 2026 20:19:49 +0800 Subject: [PATCH] core/tools/file: add tree with depth, entry, and VCS filters Workspace-relative tree; always skip VCS dirs; skip hidden dirs by default; max_depth and max_entries caps. --- CLAUDE.md | 2 +- src/plyngent/tools/__init__.py | 1 + src/plyngent/tools/file/__init__.py | 3 +- src/plyngent/tools/file/tree.py | 149 ++++++++++++++++++++++++++++ tests/test_tools/test_tree.py | 85 ++++++++++++++++ 5 files changed, 238 insertions(+), 2 deletions(-) create mode 100644 src/plyngent/tools/file/tree.py create mode 100644 tests/test_tools/test_tree.py diff --git a/CLAUDE.md b/CLAUDE.md index 4514763..9bf9326 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -62,7 +62,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init, default local user, se Module-level `@tool` handlers. Call `set_workspace_root()` before use. - **`workspace`**: path resolve under root; path substring denylist; command basename denylist. -- **`file`**: `read_file`, `write_file`, `listdir`, `edit_replace` (first occurrence). +- **`file`**: `read_file`, `write_file`, `listdir`, `tree` (depth/entry caps, skip VCS + optional hidden dirs), `edit_replace` (first occurrence). - **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); PTY `open_pty` / `read_pty` / `write_pty` / `close_pty` (**Unix only**: `pty`+`fork`). - PTY: structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)`; session limit/idle TTL/output budget; close SIGTERM→SIGKILL. - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget. diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index a1886ea..aae0a80 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -2,6 +2,7 @@ from .file import FILE_TOOLS as FILE_TOOLS from .file import edit_replace as edit_replace from .file import listdir as listdir from .file import read_file as read_file +from .file import tree as tree from .file import write_file as write_file from .process import PROCESS_TOOLS as PROCESS_TOOLS from .process import close_pty as close_pty diff --git a/src/plyngent/tools/file/__init__.py b/src/plyngent/tools/file/__init__.py index ee36c14..033a096 100644 --- a/src/plyngent/tools/file/__init__.py +++ b/src/plyngent/tools/file/__init__.py @@ -1,6 +1,7 @@ from .edit_replace import edit_replace as edit_replace from .listdir import listdir as listdir from .read import read_file as read_file +from .tree import tree as tree from .write import write_file as write_file -FILE_TOOLS = [read_file, write_file, listdir, edit_replace] +FILE_TOOLS = [read_file, write_file, listdir, tree, edit_replace] diff --git a/src/plyngent/tools/file/tree.py b/src/plyngent/tools/file/tree.py new file mode 100644 index 0000000..fd5ec17 --- /dev/null +++ b/src/plyngent/tools/file/tree.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from plyngent.agent import tool +from plyngent.tools.workspace import WorkspaceError, resolve_path + +if TYPE_CHECKING: + from pathlib import Path + +DEFAULT_MAX_DEPTH = 4 +DEFAULT_MAX_ENTRIES = 50 + +# Always skipped directory basenames (VCS metadata). +VCS_DIR_NAMES: frozenset[str] = frozenset( + { + ".git", + ".hg", + ".svn", + ".bzr", + "CVS", + ".jj", + "_darcs", + ".fossil", + } +) + + +@dataclass(frozen=True, slots=True) +class _TreeLimits: + max_depth: int + max_entries: int + skip_hidden_dirs: bool + + +def _skip_directory(name: str, *, skip_hidden_dirs: bool) -> bool: + if name in VCS_DIR_NAMES: + return True + return bool(skip_hidden_dirs and name.startswith(".")) + + +def _list_children(directory: Path, *, skip_hidden_dirs: bool) -> list[Path] | str: + try: + children = list(directory.iterdir()) + except OSError as exc: + return f"error: cannot list {directory.name}: {exc}" + visible: list[Path] = [] + for child in children: + try: + is_dir = child.is_dir() + except OSError: + continue + if is_dir and _skip_directory(child.name, skip_hidden_dirs=skip_hidden_dirs): + continue + visible.append(child) + # Directories first, then files; alphabetical within each group. + visible.sort(key=lambda p: (not p.is_dir(), p.name.casefold())) + return visible + + +def _render_tree( + directory: Path, + *, + prefix: str, + depth: int, + limits: _TreeLimits, + lines: list[str], +) -> None: + if depth >= limits.max_depth: + return + + children = _list_children(directory, skip_hidden_dirs=limits.skip_hidden_dirs) + if isinstance(children, str): + lines.append(f"{prefix}{children}") + return + + truncated = len(children) > limits.max_entries + shown = children[: limits.max_entries] + for index, child in enumerate(shown): + is_last = index == len(shown) - 1 and not truncated + branch = "└── " if is_last else "├── " + extension = " " if is_last else "│ " + try: + is_dir = child.is_dir() + except OSError: + lines.append(f"{prefix}{branch}{child.name} [error: stat failed]") + continue + if is_dir: + lines.append(f"{prefix}{branch}{child.name}/") + if depth + 1 < limits.max_depth: + _render_tree( + child, + prefix=prefix + extension, + depth=depth + 1, + limits=limits, + lines=lines, + ) + else: + lines.append(f"{prefix}{branch}{child.name}") + + if truncated: + more = len(children) - limits.max_entries + lines.append(f"{prefix}└── … ({more} more entries not shown)") + + +@tool +def tree( + path: str = ".", + *, + max_depth: int = DEFAULT_MAX_DEPTH, + max_entries: int = DEFAULT_MAX_ENTRIES, + skip_hidden_dirs: bool = True, +) -> str: + """Show a directory tree under the workspace. + + Always skips VCS metadata directories (``.git``, ``.hg``, ``.svn``, …). + By default skips other dot-directories (not hidden files). Use + ``skip_hidden_dirs=false`` to include them. + + ``max_depth`` limits how deep directories are expanded (1 = origin + children). + ``max_entries`` caps how many entries are listed per directory. + """ + if max_depth < 1: + return "error: max_depth must be >= 1" + if max_entries < 1: + return "error: max_entries must be >= 1" + + try: + origin = resolve_path(path) + except WorkspaceError as exc: + return f"error: {exc}" + if not origin.is_dir(): + return f"error: not a directory: {path}" + + root_label = path.rstrip("/\\") or "." + lines = [f"{root_label}/"] + _render_tree( + origin, + prefix="", + depth=0, + limits=_TreeLimits( + max_depth=max_depth, + max_entries=max_entries, + skip_hidden_dirs=skip_hidden_dirs, + ), + lines=lines, + ) + return "\n".join(lines) diff --git a/tests/test_tools/test_tree.py b/tests/test_tools/test_tree.py new file mode 100644 index 0000000..42b51b4 --- /dev/null +++ b/tests/test_tools/test_tree.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from pathlib import Path + +from plyngent.tools.file import tree +from tests.test_tools.helpers import call_sync + + +def _build_sample(root: Path) -> None: + _ = (root / "a.txt").write_text("a", encoding="utf-8") + _ = (root / ".hidden_file").write_text("h", encoding="utf-8") + (root / "src").mkdir() + _ = (root / "src" / "main.py").write_text("m", encoding="utf-8") + (root / "src" / "nested").mkdir() + _ = (root / "src" / "nested" / "deep.txt").write_text("d", encoding="utf-8") + (root / ".hidden_dir").mkdir() + _ = (root / ".hidden_dir" / "secret.txt").write_text("s", encoding="utf-8") + (root / ".git").mkdir() + _ = (root / ".git" / "config").write_text("g", encoding="utf-8") + (root / "pkg").mkdir() + for i in range(60): + _ = (root / "pkg" / f"f{i:02d}.txt").write_text("x", encoding="utf-8") + + +def test_tree_basic_skips_vcs_and_hidden_dirs(workspace: object) -> None: + assert isinstance(workspace, Path) + _build_sample(workspace) + out = call_sync(tree, ".") + assert "a.txt" in out + assert ".hidden_file" in out # hidden *files* still shown + assert "src/" in out + assert "main.py" in out + assert ".git" not in out + assert ".hidden_dir" not in out + assert "secret.txt" not in out + + +def test_tree_include_hidden_dirs(workspace: object) -> None: + assert isinstance(workspace, Path) + _build_sample(workspace) + out = call_sync(tree, ".", skip_hidden_dirs=False) + assert ".hidden_dir/" in out + assert "secret.txt" in out + assert ".git" not in out # VCS still always skipped + + +def test_tree_max_depth(workspace: object) -> None: + assert isinstance(workspace, Path) + _build_sample(workspace) + out = call_sync(tree, ".", max_depth=1) + assert "src/" in out + assert "main.py" not in out + assert "nested" not in out + + +def test_tree_max_entries(workspace: object) -> None: + assert isinstance(workspace, Path) + _build_sample(workspace) + out = call_sync(tree, "pkg", max_depth=2, max_entries=10) + assert "more entries not shown" in out + # only first 10 of 60 files listed as entries + listed = [line for line in out.splitlines() if line.strip().endswith(".txt")] + assert len(listed) == 10 # noqa: PLR2004 + + +def test_tree_origin_subdir(workspace: object) -> None: + assert isinstance(workspace, Path) + _build_sample(workspace) + out = call_sync(tree, "src") + assert out.startswith("src/") + assert "main.py" in out + assert "deep.txt" in out + + +def test_tree_not_directory(workspace: object) -> None: + assert isinstance(workspace, Path) + _build_sample(workspace) + out = call_sync(tree, "a.txt") + assert "error" in out + + +def test_tree_invalid_limits(workspace: object) -> None: + del workspace + assert "max_depth" in call_sync(tree, ".", max_depth=0) + assert "max_entries" in call_sync(tree, ".", max_entries=0)