diff --git a/src/plyngent/tools/file/fs_ops.py b/src/plyngent/tools/file/fs_ops.py index 78c23d6..98b6495 100644 --- a/src/plyngent/tools/file/fs_ops.py +++ b/src/plyngent/tools/file/fs_ops.py @@ -4,6 +4,7 @@ import shutil from pathlib import Path from plyngent.agent import ToolTag, invalidate_lineno_read, tool +from plyngent.tools.trash import move_to_trash, under_system_temp from plyngent.tools.workspace import AccessMode, WorkspaceError, get_workspace_root, resolve_path @@ -124,37 +125,46 @@ async def move_path(src: str, dst: str, *, overwrite: bool = False) -> str: 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) +def _discard(target: Path, path: str, *, kind: str, permanent: bool, detail: str = "") -> str: + """Move *target* to the system trash (default), or delete it for good.""" + if permanent: + _remove_existing(target) invalidate_lineno_read(str(target)) - return f"deleted directory {path} (recursive)" - try: - _ = next(target.iterdir()) - except StopIteration: - target.rmdir() - invalidate_lineno_read(str(target)) - return f"deleted empty directory {path}" - return f"error: directory not empty: {path} (set recursive=true)" + return f"deleted {kind} {path}{detail}" + error = move_to_trash(target) + if error is not None: + return error + invalidate_lineno_read(str(target)) + return f"deleted {kind} {path} (moved to system trash)" -def _delete_target(target: Path, path: str, *, recursive: bool) -> str: +def _delete_target(target: Path, path: str, *, recursive: bool, permanent: bool) -> str: + if target.is_dir() and not target.is_symlink(): + if recursive: + return _discard(target, path, kind="directory", permanent=permanent, detail=" (recursive)") + try: + _ = next(target.iterdir()) + except StopIteration: + return _discard(target, path, kind="empty directory", permanent=permanent) + return f"error: directory not empty: {path} (set recursive=true)" if target.is_symlink() or target.is_file(): - kind = _kind(target) - target.unlink() - invalidate_lineno_read(str(target)) - return f"deleted {kind} {path}" - if target.is_dir(): - return _delete_directory(target, path, recursive=recursive) + return _discard(target, path, kind=_kind(target), permanent=permanent) return f"error: unsupported path type: {path}" @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, writes=("path",)) -async def delete_path(path: str, *, recursive: bool = False) -> str: +async def delete_path(path: str, *, recursive: bool = False, permanent: 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. + The path goes to the system trash (the desktop's own trash / Recycle Bin), so + a deletion can be undone there; ``permanent=true`` removes it for good + instead. Files and empty directories are removed always, while non-empty + directories require ``recursive=true`` (uses ``shutil.rmtree``). The + workspace root cannot be deleted. When the trash is unavailable for this + path, nothing is deleted and the error says how to delete on purpose. + + Paths under the OS temporary directory are deleted for good either way: they + are scratch space, not something to restore. """ try: target = resolve_path(path, required=AccessMode.WRITE) @@ -167,6 +177,11 @@ async def delete_path(path: str, *, recursive: bool = False) -> str: if target == root: return "error: cannot delete the workspace root" try: - return _delete_target(target, path, recursive=recursive) + return _delete_target( + target, + path, + recursive=recursive, + permanent=permanent or under_system_temp(target), + ) except OSError as exc: return f"error: delete failed: {exc}" diff --git a/src/plyngent/tools/temp_workspace.py b/src/plyngent/tools/temp_workspace.py index 7d6fd47..0ca016d 100644 --- a/src/plyngent/tools/temp_workspace.py +++ b/src/plyngent/tools/temp_workspace.py @@ -6,6 +6,7 @@ import tempfile from pathlib import Path from plyngent.agent import ToolTag, tool +from plyngent.tools.trash import under_system_temp from plyngent.tools.workspace import ( MAX_TEMPORARY_WORKSPACES, WorkspaceError, @@ -32,22 +33,14 @@ def _sanitize_prefix(prefix: str) -> str: return cleaned -def _is_under_system_temp(path: Path) -> bool: - """True if *path* is under the OS temporary directory (safe to rmtree).""" - try: - temp_root = Path(tempfile.gettempdir()).expanduser().resolve() - resolved = path.expanduser().resolve() - _ = resolved.relative_to(temp_root) - except OSError, ValueError: - return False - return True - - def cleanup_temporary_workspaces() -> int: """Remove directories created by :func:`new_temporary_workspace` (chat exit). Only deletes paths still under the system temp dir. Returns the number of directories removed. No-op when no instance is bound. + + These are scratch directories, so they are removed for good rather than + moved to the trash the delete tool would use. """ from plyngent.tools.context import get_instance @@ -55,7 +48,7 @@ def cleanup_temporary_workspaces() -> int: return 0 removed = 0 for path in pop_owned_temporary_workspaces(): - if not _is_under_system_temp(path): + if not under_system_temp(path): remove_workspace_allowlist(path) continue if path.is_dir(): diff --git a/src/plyngent/tools/trash.py b/src/plyngent/tools/trash.py new file mode 100644 index 0000000..210a03a --- /dev/null +++ b/src/plyngent/tools/trash.py @@ -0,0 +1,61 @@ +"""Move a path to the system trash instead of destroying it. + +``delete_path`` hands every removal to :func:`move_to_trash`, so a delete the +human approved (or YOLO skipped) can still be undone from the desktop's own +trash. The platform work is ``send2trash``: the freedesktop trash specification +on Linux, the Finder trash on macOS, the Recycle Bin on Windows. + +Scratch paths under the OS temporary directory are the exception — they are +deleted for good, since nobody restores them and a trash on the ``/tmp`` volume +would only collect them (see :func:`under_system_temp`). +""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + + +def is_available() -> bool: + """True when ``send2trash`` can be imported. + + It is a required dependency, so ``False`` means a broken install rather than + an expected absence; the delete path then reports the problem instead of + destroying the file. + """ + from importlib.util import find_spec + + return find_spec("send2trash") is not None + + +def under_system_temp(path: Path) -> bool: + """True if *path* is under the OS temporary directory (scratch space).""" + try: + temp_root = Path(tempfile.gettempdir()).expanduser().resolve() + resolved = path.expanduser().resolve() + _ = resolved.relative_to(temp_root) + except OSError, ValueError: + return False + return True + + +def move_to_trash(path: Path) -> str | None: + """Send *path* to the system trash; return an error text, or ``None`` on success. + + A failure is always reported, never downgraded to a permanent delete: the + human was told the path is recoverable, so a trash that is unavailable for + this path leaves the file untouched and says how to remove it on purpose. + """ + if not is_available(): + return "error: cannot move to trash: send2trash is not installed (use permanent=true)" + + from send2trash import TrashPermissionError, send2trash + + try: + send2trash(str(path)) + except TrashPermissionError as exc: + # A volume with no trash directory and no permission to create one. + return f"error: cannot move to trash: {exc} (use permanent=true)" + except OSError as exc: + return f"error: cannot move to trash: {exc}" + return None diff --git a/tests/test_tools/test_fs_ops.py b/tests/test_tools/test_fs_ops.py index 037b244..085096b 100644 --- a/tests/test_tools/test_fs_ops.py +++ b/tests/test_tools/test_fs_ops.py @@ -2,7 +2,9 @@ from __future__ import annotations from pathlib import Path -from plyngent.tools.file import copy_path, delete_path, move_path, read_file, write_file +import pytest + +from plyngent.tools.file import copy_path, delete_path, fs_ops, move_path, read_file, write_file from tests.test_tools.helpers import call_sync @@ -75,6 +77,8 @@ def test_delete_nonempty_requires_recursive(workspace: object) -> None: out = call_sync(delete_path, "box") assert "not empty" in out assert (workspace / "box" / "a.txt").is_file() + # The workspace fixture lives under the system temp dir, so it is removed + # for good rather than trashed (see the trash tests below). out2 = call_sync(delete_path, "box", recursive=True) assert "recursive" in out2 assert not (workspace / "box").exists() @@ -97,3 +101,71 @@ def test_missing_source(workspace: object) -> None: 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") + + +def _elsewhere_than_temp(monkeypatch: pytest.MonkeyPatch) -> None: + """Pretend the workspace is not scratch space, so deletes use the trash.""" + monkeypatch.setattr(fs_ops, "under_system_temp", lambda _path: False) + + +def test_a_temp_path_is_deleted_for_good(workspace: object) -> None: + assert isinstance(workspace, Path) + _ = call_sync(write_file, "t.txt", "x") + out = call_sync(delete_path, "t.txt") + assert "trash" not in out + assert not (workspace / "t.txt").exists() + + +def test_delete_goes_to_the_trash(workspace: object, monkeypatch: pytest.MonkeyPatch) -> None: + assert isinstance(workspace, Path) + _elsewhere_than_temp(monkeypatch) + trashed: list[Path] = [] + monkeypatch.setattr(fs_ops, "move_to_trash", trashed.append) + _ = call_sync(write_file, "t.txt", "x") + + out = call_sync(delete_path, "t.txt") + + assert out == "deleted file t.txt (moved to system trash)" + assert [path.name for path in trashed] == ["t.txt"] + + +def test_recursive_delete_goes_to_the_trash(workspace: object, monkeypatch: pytest.MonkeyPatch) -> None: + assert isinstance(workspace, Path) + _elsewhere_than_temp(monkeypatch) + trashed: list[Path] = [] + monkeypatch.setattr(fs_ops, "move_to_trash", trashed.append) + _ = call_sync(write_file, "box/a.txt", "a") + + out = call_sync(delete_path, "box", recursive=True) + + assert out == "deleted directory box (moved to system trash)" + assert [path.name for path in trashed] == ["box"] + + +def test_a_refused_trash_keeps_the_path(workspace: object, monkeypatch: pytest.MonkeyPatch) -> None: + assert isinstance(workspace, Path) + _elsewhere_than_temp(monkeypatch) + refused = "error: cannot move to trash: nope (use permanent=true)" + monkeypatch.setattr(fs_ops, "move_to_trash", lambda _target: refused) + _ = call_sync(write_file, "t.txt", "x") + + out = call_sync(delete_path, "t.txt") + + assert out == refused + assert (workspace / "t.txt").is_file() + + +def test_permanent_delete_skips_the_trash(workspace: object, monkeypatch: pytest.MonkeyPatch) -> None: + assert isinstance(workspace, Path) + + def _never(_target: Path) -> str | None: + msg = "a permanent delete must not touch the trash" + raise AssertionError(msg) + + monkeypatch.setattr(fs_ops, "move_to_trash", _never) + _ = call_sync(write_file, "t.txt", "x") + + out = call_sync(delete_path, "t.txt", permanent=True) + + assert out == "deleted file t.txt" + assert not (workspace / "t.txt").exists() diff --git a/tests/test_tools/test_trash.py b/tests/test_tools/test_trash.py new file mode 100644 index 0000000..ea9694b --- /dev/null +++ b/tests/test_tools/test_trash.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import os +import sys +from pathlib import Path + +import pytest +import send2trash + +from plyngent.tools import trash +from plyngent.tools.trash import is_available, move_to_trash, under_system_temp + +# The freedesktop implementation is the only one whose trash location can be +# redirected for a test; macOS and Windows move into their own system trash. +freedesktop_only = pytest.mark.skipif(sys.platform != "linux", reason="freedesktop trash only") + + +@pytest.fixture +def home_trash(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Point the freedesktop implementation at a trash under *tmp_path*. + + ``plat_other`` reads its trash locations from module globals computed at + import time and picks "home" vs "external volume" by device number, so both + are pinned here — otherwise a ``/tmp`` on a filesystem of its own would + create a real trash directory on that volume. + """ + plat_other = pytest.importorskip("send2trash.plat_other") + root = tmp_path / "Trash" + monkeypatch.setattr(plat_other, "XDG_DATA_HOME", os.fsencode(str(tmp_path))) + monkeypatch.setattr(plat_other, "HOMETRASH_B", os.fsencode(str(root))) + monkeypatch.setattr(plat_other, "HOMETRASH", str(root)) + monkeypatch.setattr(plat_other, "get_dev", lambda _path: 1) + # A GIO-backed install would otherwise bypass every patch above. + monkeypatch.setattr(send2trash, "send2trash", plat_other.send2trash) + return root + + +def test_the_library_is_installed() -> None: + assert is_available() + + +@freedesktop_only +def test_move_to_trash_lands_in_the_trash(home_trash: Path, tmp_path: Path) -> None: + target = tmp_path / "gone.txt" + _ = target.write_text("x") + + assert move_to_trash(target) is None + + assert not target.exists() + assert (home_trash / "files" / "gone.txt").read_text() == "x" + entry = (home_trash / "info" / "gone.txt.trashinfo").read_text() + assert entry.startswith("[Trash Info]") + assert "Path=" in entry + assert "DeletionDate=" in entry + + +@freedesktop_only +def test_move_to_trash_moves_a_directory_in_one_piece(home_trash: Path, tmp_path: Path) -> None: + target = tmp_path / "box" + target.mkdir() + _ = (target / "inner.txt").write_text("x") + + assert move_to_trash(target) is None + + assert not target.exists() + assert (home_trash / "files" / "box" / "inner.txt").read_text() == "x" + + +@freedesktop_only +def test_trashing_twice_keeps_both_copies(home_trash: Path, tmp_path: Path) -> None: + first = tmp_path / "dup.txt" + second = tmp_path / "sub" / "dup.txt" + second.parent.mkdir() + _ = first.write_text("one") + _ = second.write_text("two") + + assert move_to_trash(first) is None + assert move_to_trash(second) is None + + names = sorted(path.name for path in (home_trash / "files").iterdir()) + assert names == ["dup 1.txt", "dup.txt"] + + +def test_a_refused_volume_names_the_escape_hatch(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + target = tmp_path / "gone.txt" + _ = target.write_text("x") + + def _refuse(_paths: object) -> None: + raise send2trash.TrashPermissionError(str(target)) + + monkeypatch.setattr(send2trash, "send2trash", _refuse) + + error = move_to_trash(target) + + assert error is not None + assert error.startswith("error: cannot move to trash:") + assert "permanent=true" in error + assert target.is_file() + + +def test_an_oserror_is_reported(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + target = tmp_path / "gone.txt" + _ = target.write_text("x") + + def _fail(_paths: object) -> None: + raise OSError("disk on fire") + + monkeypatch.setattr(send2trash, "send2trash", _fail) + + assert move_to_trash(target) == "error: cannot move to trash: disk on fire" + assert target.is_file() + + +def test_a_broken_install_leaves_the_file_alone(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + target = tmp_path / "gone.txt" + _ = target.write_text("x") + monkeypatch.setattr(trash, "is_available", lambda: False) + + error = move_to_trash(target) + + assert error is not None + assert "send2trash is not installed" in error + assert "permanent=true" in error + assert target.is_file() + + +def test_under_system_temp(tmp_path: Path) -> None: + assert under_system_temp(tmp_path) + assert under_system_temp(tmp_path / "nested" / "file.txt") + assert not under_system_temp(Path("/plyngent-not-temp/deep/file.txt")) diff --git a/typings/send2trash/__init__.pyi b/typings/send2trash/__init__.pyi new file mode 100644 index 0000000..1232cb0 --- /dev/null +++ b/typings/send2trash/__init__.pyi @@ -0,0 +1,9 @@ +# Minimal stubs for ``send2trash`` (a required dependency). +# Only the surface ``plyngent.tools.trash`` uses is declared; the real package is +# pure Python over ctypes (freedesktop trash spec / Finder trash / Recycle Bin). + +from collections.abc import Iterable + +class TrashPermissionError(PermissionError): ... + +def send2trash(paths: str | Iterable[str]) -> None: ...