mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
core/tools: send a deleted path to the system trash
`delete_path` is the only destructive file entry point the agent has, and it carries `ToolTag.YOLO` — under `--yes` / `/yolo` it runs with no confirmation at all. A delete is now a move into the desktop's own trash (freedesktop trash on Linux, Finder trash on macOS, Recycle Bin on Windows) through `send2trash`, so the human can put the file back with the file manager they already have. The new `permanent=true` argument is the old behaviour, chosen on purpose. A trashing that fails is reported, never downgraded to a deletion: the human was told the path is recoverable, so a volume with no writable trash directory leaves the file where it is and the error names `permanent=true` as the way to delete it anyway. A broken install (the library is a required dependency) is reported the same way. Scratch paths under the OS temporary directory are still deleted for good — nobody restores them, and a trash on the `/tmp` volume would only collect them. The rule now lives in `tools/trash.py` (`under_system_temp`) and the temporary workspace cleanup uses it too, instead of keeping a second copy of the check.
This commit is contained in:
@@ -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}"
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
@@ -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"))
|
||||
@@ -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: ...
|
||||
Reference in New Issue
Block a user