diff --git a/CLAUDE.md b/CLAUDE.md index ba96ec1..fb78e5d 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`, `tree` (depth/entry caps, skip VCS + optional hidden dirs), `edit_replace` (first occurrence). +- **`file`**: `read_file`, `write_file`, `listdir`, `tree` (depth/entry caps, skip VCS + optional hidden dirs), `edit_replace` (first occurrence), `copy_path` / `move_path` / `delete_path` (shutil; dirs supported). - **`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 aae0a80..1231005 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -1,6 +1,9 @@ from .file import FILE_TOOLS as FILE_TOOLS +from .file import copy_path as copy_path +from .file import delete_path as delete_path from .file import edit_replace as edit_replace from .file import listdir as listdir +from .file import move_path as move_path from .file import read_file as read_file from .file import tree as tree from .file import write_file as write_file diff --git a/src/plyngent/tools/file/__init__.py b/src/plyngent/tools/file/__init__.py index 033a096..a8594f3 100644 --- a/src/plyngent/tools/file/__init__.py +++ b/src/plyngent/tools/file/__init__.py @@ -1,7 +1,19 @@ from .edit_replace import edit_replace as edit_replace +from .fs_ops import copy_path as copy_path +from .fs_ops import delete_path as delete_path +from .fs_ops import move_path as move_path 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, tree, edit_replace] +FILE_TOOLS = [ + read_file, + write_file, + listdir, + tree, + edit_replace, + copy_path, + move_path, + delete_path, +] diff --git a/src/plyngent/tools/file/fs_ops.py b/src/plyngent/tools/file/fs_ops.py new file mode 100644 index 0000000..e491a76 --- /dev/null +++ b/src/plyngent/tools/file/fs_ops.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import shutil +from pathlib import Path + +from plyngent.agent import tool +from plyngent.tools.workspace import WorkspaceError, get_workspace_root, resolve_path + + +def _kind(path: Path) -> str: + if path.is_dir(): + return "directory" + if path.is_file(): + return "file" + if path.is_symlink(): + return "symlink" + return "path" + + +def _remove_existing(path: Path) -> None: + if path.is_dir() and not path.is_symlink(): + shutil.rmtree(path) + elif path.exists() or path.is_symlink(): + path.unlink() + + +def _resolve_pair(src: str, dst: str) -> tuple[Path, Path, Path] | str: + try: + source = resolve_path(src) + dest = resolve_path(dst) + root = get_workspace_root() + except WorkspaceError as exc: + return f"error: {exc}" + return source, dest, root + + +def _prepare_dest(source: Path, dest: Path, root: Path, *, overwrite: bool, dst_label: str) -> Path | str: + """Resolve copy/move destination; return error string on conflict.""" + if source.is_file() and dest.is_dir(): + dest = resolve_path(str((dest / source.name).relative_to(root))) + if dest.exists() or dest.is_symlink(): + if not overwrite: + return f"error: destination exists: {dst_label} (set overwrite=true)" + _remove_existing(dest) + dest.parent.mkdir(parents=True, exist_ok=True) + return dest + + +def _copy_or_move_validated( # noqa: PLR0913 + source: Path, + dest: Path, + root: Path, + *, + src: str, + dst: str, + overwrite: bool, + action: str, +) -> str: + prepared = _prepare_dest(source, dest, root, overwrite=overwrite, dst_label=dst) + if isinstance(prepared, str): + return prepared + dest = prepared + try: + if action == "copy": + if source.is_dir() and not source.is_symlink(): + _ = shutil.copytree(source, dest, symlinks=True) + label = "directory" + else: + _ = shutil.copy2(source, dest, follow_symlinks=False) + label = _kind(source) + return f"copied {label} {src} -> {dest.relative_to(root)}" + moved = Path(shutil.move(str(source), str(dest))).resolve() + return f"moved {_kind(moved)} {src} -> {moved.relative_to(root)}" + except WorkspaceError as exc: + return f"error: {exc}" + except OSError as exc: + return f"error: {action} failed: {exc}" + + +@tool +def copy_path(src: str, dst: str, *, overwrite: bool = False) -> str: + """Copy a file or directory under the workspace (``shutil.copy2`` / ``copytree``). + + ``dst`` parent directories are created as needed. If ``dst`` exists, set + ``overwrite=true`` to replace it. + """ + pair = _resolve_pair(src, dst) + if isinstance(pair, str): + return pair + source, dest, root = pair + if not source.exists() and not source.is_symlink(): + return f"error: source does not exist: {src}" + return _copy_or_move_validated( + source, dest, root, src=src, dst=dst, overwrite=overwrite, action="copy" + ) + + +@tool +def move_path(src: str, dst: str, *, overwrite: bool = False) -> str: + """Move/rename a file or directory under the workspace (``shutil.move``).""" + pair = _resolve_pair(src, dst) + if isinstance(pair, str): + return pair + source, dest, root = pair + if not source.exists() and not source.is_symlink(): + return f"error: source does not exist: {src}" + if source == root: + return "error: cannot move the workspace root" + return _copy_or_move_validated( + source, dest, root, src=src, dst=dst, overwrite=overwrite, action="move" + ) + + +def _delete_directory(target: Path, path: str, *, recursive: bool) -> str: + if recursive: + shutil.rmtree(target) + return f"deleted directory {path} (recursive)" + try: + _ = next(target.iterdir()) + except StopIteration: + target.rmdir() + return f"deleted empty directory {path}" + return f"error: directory not empty: {path} (set recursive=true)" + + +def _delete_target(target: Path, path: str, *, recursive: bool) -> str: + if target.is_symlink() or target.is_file(): + kind = _kind(target) + target.unlink() + return f"deleted {kind} {path}" + if target.is_dir(): + return _delete_directory(target, path, recursive=recursive) + return f"error: unsupported path type: {path}" + + +@tool +def delete_path(path: str, *, recursive: bool = False) -> str: + """Delete a file or directory under the workspace. + + Files and empty directories are removed always. Non-empty directories require + ``recursive=true`` (uses ``shutil.rmtree``). The workspace root cannot be deleted. + """ + try: + target = resolve_path(path) + root = get_workspace_root() + except WorkspaceError as exc: + return f"error: {exc}" + + if not target.exists() and not target.is_symlink(): + return f"error: path does not exist: {path}" + if target == root: + return "error: cannot delete the workspace root" + try: + return _delete_target(target, path, recursive=recursive) + except OSError as exc: + return f"error: delete failed: {exc}" diff --git a/tests/test_tools/test_fs_ops.py b/tests/test_tools/test_fs_ops.py new file mode 100644 index 0000000..fc2c9e2 --- /dev/null +++ b/tests/test_tools/test_fs_ops.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +from pathlib import Path + +from plyngent.tools.file import copy_path, delete_path, move_path, read_file, write_file +from tests.test_tools.helpers import call_sync + + +def test_copy_file(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "a.txt", "hello") + out = call_sync(copy_path, "a.txt", "b.txt") + assert "copied" in out + assert call_sync(read_file, "b.txt") == "hello" + assert (workspace / "a.txt").is_file() + + +def test_copy_directory(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "src/x.txt", "x") + _ = call_sync(write_file, "src/nested/y.txt", "y") + out = call_sync(copy_path, "src", "dst") + assert "directory" in out + assert call_sync(read_file, "dst/x.txt") == "x" + assert call_sync(read_file, "dst/nested/y.txt") == "y" + + +def test_copy_overwrite(workspace: object) -> None: + del workspace + _ = call_sync(write_file, "a.txt", "old") + _ = call_sync(write_file, "b.txt", "new") + assert "exists" in call_sync(copy_path, "b.txt", "a.txt") + out = call_sync(copy_path, "b.txt", "a.txt", overwrite=True) + assert "copied" in out + assert call_sync(read_file, "a.txt") == "new" + + +def test_move_file(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "old.txt", "data") + out = call_sync(move_path, "old.txt", "renamed.txt") + assert "moved" in out + assert not (workspace / "old.txt").exists() + assert call_sync(read_file, "renamed.txt") == "data" + + +def test_move_directory(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "d/a.txt", "a") + out = call_sync(move_path, "d", "e") + assert "moved" in out + assert not (workspace / "d").exists() + assert call_sync(read_file, "e/a.txt") == "a" + + +def test_delete_file(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "t.txt", "x") + out = call_sync(delete_path, "t.txt") + assert "deleted" in out + assert not (workspace / "t.txt").exists() + + +def test_delete_empty_dir(workspace: object) -> None: + assert isinstance(workspace, Path) + (workspace / "empty").mkdir() + out = call_sync(delete_path, "empty") + assert "deleted" in out + assert not (workspace / "empty").exists() + + +def test_delete_nonempty_requires_recursive(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "box/a.txt", "a") + out = call_sync(delete_path, "box") + assert "not empty" in out + assert (workspace / "box" / "a.txt").is_file() + out2 = call_sync(delete_path, "box", recursive=True) + assert "recursive" in out2 + assert not (workspace / "box").exists() + + +def test_delete_workspace_root_forbidden(workspace: object) -> None: + del workspace + out = call_sync(delete_path, ".") + assert "workspace root" in out + + +def test_move_workspace_root_forbidden(workspace: object) -> None: + del workspace + out = call_sync(move_path, ".", "elsewhere") + assert "workspace root" in out + + +def test_missing_source(workspace: object) -> None: + del workspace + assert "does not exist" in call_sync(copy_path, "nope", "x") + assert "does not exist" in call_sync(move_path, "nope", "x") + assert "does not exist" in call_sync(delete_path, "nope")