mirror of
https://github.com/NCBM/plyngent.git
synced 2026-07-23 05:55:16 +08:00
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.
This commit is contained in:
@@ -62,7 +62,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init, default local user, se
|
|||||||
Module-level `@tool` handlers. Call `set_workspace_root()` before use.
|
Module-level `@tool` handlers. Call `set_workspace_root()` before use.
|
||||||
|
|
||||||
- **`workspace`**: path resolve under root; path substring denylist; command basename denylist.
|
- **`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`).
|
- **`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.
|
- 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.
|
- CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget.
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ from .file import FILE_TOOLS as FILE_TOOLS
|
|||||||
from .file import edit_replace as edit_replace
|
from .file import edit_replace as edit_replace
|
||||||
from .file import listdir as listdir
|
from .file import listdir as listdir
|
||||||
from .file import read_file as read_file
|
from .file import read_file as read_file
|
||||||
|
from .file import tree as tree
|
||||||
from .file import write_file as write_file
|
from .file import write_file as write_file
|
||||||
from .process import PROCESS_TOOLS as PROCESS_TOOLS
|
from .process import PROCESS_TOOLS as PROCESS_TOOLS
|
||||||
from .process import close_pty as close_pty
|
from .process import close_pty as close_pty
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
from .edit_replace import edit_replace as edit_replace
|
from .edit_replace import edit_replace as edit_replace
|
||||||
from .listdir import listdir as listdir
|
from .listdir import listdir as listdir
|
||||||
from .read import read_file as read_file
|
from .read import read_file as read_file
|
||||||
|
from .tree import tree as tree
|
||||||
from .write import write_file as write_file
|
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]
|
||||||
|
|||||||
@@ -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)
|
||||||
@@ -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)
|
||||||
Reference in New Issue
Block a user