mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
core/cli: hand the trash switch to the tools
`[trash] enabled` now reaches `delete_path` through the bound instance's workspace policy, the same channel `[agent] path_denylist` already travels: set when the session state is built and re-read by `reload_config_from_disk`, so a config reload in the middle of a chat changes the next delete too. The policy flag is what `delete_path` consults, so a host that never sets it gets the safe default (deletes go to the trash) instead of a silent no-op.
This commit is contained in:
@@ -347,8 +347,10 @@ async def _run_chat( # noqa: C901, PLR0915 — chat orchestration
|
||||
quiet=quiet,
|
||||
mcp_manager=mcp_manager,
|
||||
)
|
||||
# Path denylist and policy confirm live on instance.workspace (no process bag).
|
||||
# Path denylist, trash switch and policy confirm live on
|
||||
# instance.workspace (no process bag).
|
||||
state.instance_state.workspace.path_denylist = tuple(store.agent_config.path_denylist or ())
|
||||
state.instance_state.workspace.trash_enabled = store.trash_config.enabled
|
||||
from plyngent.tools.net import set_ssrf_assume_public_cidrs
|
||||
|
||||
# Clash Fake-IP etc.: treat configured CIDRs as public for fetch_local SSRF.
|
||||
|
||||
@@ -655,6 +655,7 @@ class ReplState:
|
||||
|
||||
self.config.reload()
|
||||
self.instance_state.workspace.path_denylist = tuple(self.config.agent_config.path_denylist or ())
|
||||
self.instance_state.workspace.trash_enabled = self.config.trash_config.enabled
|
||||
# Roots and their switches came from the file that was just re-read.
|
||||
self._install_skills()
|
||||
|
||||
|
||||
@@ -101,6 +101,7 @@ from .workspace import denylist_match as denylist_match
|
||||
from .workspace import get_command_denylist as get_command_denylist
|
||||
from .workspace import get_path_denylist as get_path_denylist
|
||||
from .workspace import get_policy_confirm_timeout as get_policy_confirm_timeout
|
||||
from .workspace import get_trash_enabled as get_trash_enabled
|
||||
from .workspace import get_workspace_root as get_workspace_root
|
||||
from .workspace import grant_policy_command as grant_policy_command
|
||||
from .workspace import granted_access_mode as granted_access_mode
|
||||
@@ -116,5 +117,6 @@ from .workspace import set_path_denylist as set_path_denylist
|
||||
from .workspace import set_policy_confirm_hook as set_policy_confirm_hook
|
||||
from .workspace import set_policy_confirm_timeout as set_policy_confirm_timeout
|
||||
from .workspace import set_static_read_roots as set_static_read_roots
|
||||
from .workspace import set_trash_enabled as set_trash_enabled
|
||||
from .workspace import set_workspace_root as set_workspace_root
|
||||
from .workspace import within_workspace_roots as within_workspace_roots
|
||||
|
||||
@@ -5,7 +5,13 @@ 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
|
||||
from plyngent.tools.workspace import (
|
||||
AccessMode,
|
||||
WorkspaceError,
|
||||
get_trash_enabled,
|
||||
get_workspace_root,
|
||||
resolve_path,
|
||||
)
|
||||
|
||||
|
||||
def _kind(path: Path) -> str:
|
||||
@@ -164,7 +170,8 @@ async def delete_path(path: str, *, recursive: bool = False, permanent: bool = F
|
||||
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.
|
||||
are scratch space, not something to restore. A host with ``[trash] enabled =
|
||||
false`` removes the path outright, as ``permanent=true`` does.
|
||||
"""
|
||||
try:
|
||||
target = resolve_path(path, required=AccessMode.WRITE)
|
||||
@@ -181,7 +188,7 @@ async def delete_path(path: str, *, recursive: bool = False, permanent: bool = F
|
||||
target,
|
||||
path,
|
||||
recursive=recursive,
|
||||
permanent=permanent or under_system_temp(target),
|
||||
permanent=permanent or not get_trash_enabled() or under_system_temp(target),
|
||||
)
|
||||
except OSError as exc:
|
||||
return f"error: delete failed: {exc}"
|
||||
|
||||
@@ -120,6 +120,8 @@ class WorkspacePolicy:
|
||||
# read them, nothing may write them. Not a grant, so they stay out of
|
||||
# ``/grants``; writes go through the owning tool's own confirm.
|
||||
static_read: list[Path] = field(default_factory=list)
|
||||
# ``[trash] enabled``: deletes go to the system trash unless cleared.
|
||||
trash_enabled: bool = True
|
||||
|
||||
|
||||
def _bound_instance() -> InstanceState | None:
|
||||
@@ -188,6 +190,16 @@ def get_path_denylist() -> tuple[str, ...]:
|
||||
return active_workspace_policy().path_denylist
|
||||
|
||||
|
||||
def set_trash_enabled(*, enabled: bool) -> None:
|
||||
"""Set whether ``delete_path`` moves to the system trash (``[trash]``)."""
|
||||
active_workspace_policy().trash_enabled = bool(enabled)
|
||||
|
||||
|
||||
def get_trash_enabled() -> bool:
|
||||
"""Whether deletes go to the system trash rather than being final."""
|
||||
return active_workspace_policy().trash_enabled
|
||||
|
||||
|
||||
def set_command_denylist(names: list[str] | tuple[str, ...] | frozenset[str] | None) -> None:
|
||||
"""Set denied command basenames (None restores defaults)."""
|
||||
policy = active_workspace_policy()
|
||||
|
||||
@@ -5,6 +5,7 @@ from pathlib import Path
|
||||
import pytest
|
||||
|
||||
from plyngent.tools.file import copy_path, delete_path, fs_ops, move_path, read_file, write_file
|
||||
from plyngent.tools.workspace import set_trash_enabled
|
||||
from tests.test_tools.helpers import call_sync
|
||||
|
||||
|
||||
@@ -169,3 +170,16 @@ def test_permanent_delete_skips_the_trash(workspace: object, monkeypatch: pytest
|
||||
|
||||
assert out == "deleted file t.txt"
|
||||
assert not (workspace / "t.txt").exists()
|
||||
|
||||
|
||||
def test_the_host_switch_turns_the_trash_off(workspace: object, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""``[trash] enabled = false`` restores the behaviour from before the trash."""
|
||||
assert isinstance(workspace, Path)
|
||||
_elsewhere_than_temp(monkeypatch)
|
||||
set_trash_enabled(enabled=False)
|
||||
_ = call_sync(write_file, "t.txt", "x")
|
||||
|
||||
out = call_sync(delete_path, "t.txt")
|
||||
|
||||
assert out == "deleted file t.txt"
|
||||
assert not (workspace / "t.txt").exists()
|
||||
|
||||
@@ -7,12 +7,14 @@ from plyngent.tools import (
|
||||
WorkspaceError,
|
||||
active_workspace_policy,
|
||||
check_command_allowed,
|
||||
get_trash_enabled,
|
||||
get_workspace_root,
|
||||
mode_covers,
|
||||
parse_access_mode,
|
||||
resolve_path,
|
||||
set_command_denylist,
|
||||
set_path_denylist,
|
||||
set_trash_enabled,
|
||||
)
|
||||
from plyngent.tools.context import SessionState, bind_session
|
||||
|
||||
@@ -55,6 +57,16 @@ def test_access_mode_parse_and_covers() -> None:
|
||||
assert not mode_covers(AccessMode.READ, AccessMode.WRITE)
|
||||
|
||||
|
||||
def test_trash_switch(workspace: object) -> None:
|
||||
"""Deletes go to the trash until a host clears the switch."""
|
||||
del workspace
|
||||
assert get_trash_enabled() is True
|
||||
set_trash_enabled(enabled=False)
|
||||
assert get_trash_enabled() is False
|
||||
set_trash_enabled(enabled=True)
|
||||
assert get_trash_enabled() is True
|
||||
|
||||
|
||||
def test_config_grant_modes(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
|
||||
del workspace
|
||||
outside = tmp_path_factory.mktemp("granted")
|
||||
|
||||
Reference in New Issue
Block a user