tools: request_directory_access and grant stores

src/plyngent/config/models.py

@@ -23,6 +23,8 @@ DEFAULT_TOOL_DIRECTIVES = """\
 ### Workspace
 - Explore with `tree`/`listdir`/`glob_paths`/`grep_files`/`read_file` when unsure.
 - Stay under the workspace (or a path from `new_temporary_workspace`). Respect path denylists.
+- Need a path outside the workspace? Call `request_directory_access` \
+(human approves; denylists apply) instead of retrying an escaped path.
 
 ### Files
 - Prefer file tools over shell. `edit_replace` fails usually means a bad match — \

src/plyngent/tools/__init__.py

@@ -1,3 +1,14 @@
+from .access import ACCESS_TOOLS as ACCESS_TOOLS
+from .access import MAX_ACCESS_GRANTS as MAX_ACCESS_GRANTS
+from .access import AccessDecision as AccessDecision
+from .access import active_grant_count as active_grant_count
+from .access import clear_process_access as clear_process_access
+from .access import clear_session_access as clear_session_access
+from .access import get_directory_access_confirm_hook as get_directory_access_confirm_hook
+from .access import grant_process_access as grant_process_access
+from .access import grant_session_access as grant_session_access
+from .access import request_directory_access as request_directory_access
+from .access import set_directory_access_confirm_hook as set_directory_access_confirm_hook
 from .catalog import ToolCatalog as ToolCatalog
 from .catalog import ToolSource as ToolSource
 from .catalog import catalog_scope as catalog_scope
@@ -75,11 +86,13 @@ from .workspace import check_command_allowed as check_command_allowed
 from .workspace import clear_policy_allowed_commands as clear_policy_allowed_commands
 from .workspace import clear_workspace_allowlist as clear_workspace_allowlist
 from .workspace import clear_workspace_root as clear_workspace_root
+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_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
 from .workspace import list_workspace_allowlist as list_workspace_allowlist
 from .workspace import mode_covers as mode_covers
 from .workspace import parse_access_mode as parse_access_mode
@@ -91,3 +104,4 @@ 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_workspace_root as set_workspace_root
+from .workspace import within_workspace_roots as within_workspace_roots

src/plyngent/tools/access.py

@@ -0,0 +1,205 @@
+"""Directory-access grants: request tool plus config / session / process stores.
+
+A grant is one ``path → mode`` pair; its *source* is structural:
+
+- :attr:`WorkspacePolicy.config_allow` — static TOML pre-allow (never prompts),
+- :attr:`SessionState.access_grants` — human-approved for this session,
+- :attr:`WorkspacePolicy.yolo_allow` — ``--yes`` / YOLO auto-approval, process only.
+
+``resolve_path`` consults all three. Grants are a policy gate for path-resolving
+tools, **not** a sandbox: an ``exec`` grant (or any command) can reach the rest
+of the filesystem.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING, NamedTuple
+
+from plyngent.agent import ToolTag, tool
+from plyngent.tools.context import get_session, require_session
+from plyngent.tools.workspace import (
+    AccessMode,
+    denylist_match,
+    get_policy_confirm_timeout,
+    granted_access_mode,
+    mode_covers,
+    parse_access_mode,
+    require_bound_instance,
+    within_workspace_roots,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from plyngent.tools.context import InstanceState, SessionState
+
+# Max active grants across all three stores (bounds prompt/listing noise).
+MAX_ACCESS_GRANTS = 32
+
+_HOOK_KEY = "directory_access_confirm_hook"
+
+
+class AccessDecision(NamedTuple):
+    """Hook approval for one directory-access request.
+
+    ``mode`` is the granted level (may differ from the requested one);
+    ``persist`` is true for a session grant, false for a process-only grant.
+    """
+
+    mode: AccessMode
+    persist: bool = True
+
+
+type DirectoryAccessConfirmHook = Callable[[Path, AccessMode, str, float], AccessDecision | str | None]
+
+
+def get_directory_access_confirm_hook(instance: InstanceState | None = None) -> DirectoryAccessConfirmHook | None:
+    """Return the host-installed confirm hook for this instance, if any."""
+    inst = instance if instance is not None else require_bound_instance()
+    hook = inst.extras.get(_HOOK_KEY)
+    if hook is None or not callable(hook):
+        return None
+
+    def _as_hook(path: Path, mode: AccessMode, reason: str, timeout: float) -> AccessDecision | str | None:
+        result: object = hook(path, mode, reason, timeout)
+        if isinstance(result, AccessDecision | str):
+            return result
+        return None
+
+    return _as_hook
+
+
+def set_directory_access_confirm_hook(
+    hook: DirectoryAccessConfirmHook | None,
+    *,
+    instance: InstanceState | None = None,
+) -> None:
+    """Install (or clear) the confirm hook the CLI uses to ask the human."""
+    inst = instance if instance is not None else require_bound_instance()
+    inst.extras[_HOOK_KEY] = hook
+
+
+def grant_session_access(
+    path: Path | str,
+    mode: AccessMode,
+    *,
+    session: SessionState | None = None,
+) -> Path:
+    """Grant *mode* on *path* for the bound session; returns the resolved path."""
+    resolved = Path(path).expanduser().resolve()
+    sess = session if session is not None else require_session()
+    sess.access_grants[resolved] = mode
+    return resolved
+
+
+def grant_process_access(path: Path | str, mode: AccessMode) -> Path:
+    """Grant *mode* on *path* for this process only (``--yes``); never persisted."""
+    resolved = Path(path).expanduser().resolve()
+    require_bound_instance().workspace.yolo_allow[resolved] = mode
+    return resolved
+
+
+def clear_process_access() -> None:
+    """Drop process-scoped (``--yes``) grants."""
+    require_bound_instance().workspace.yolo_allow.clear()
+
+
+def clear_session_access(*, session: SessionState | None = None) -> None:
+    """Drop live session grants (durable copy is cleared by the host)."""
+    sess = session if session is not None else get_session()
+    if sess is not None:
+        sess.access_grants.clear()
+
+
+def active_grant_count() -> int:
+    """Total live grants across config, session, and process stores."""
+    policy = require_bound_instance().workspace
+    total = len(policy.config_allow) + len(policy.yolo_allow)
+    session = get_session()
+    if session is not None:
+        total += len(session.access_grants)
+    return total
+
+
+def _resolve_target(path: str) -> Path | str:
+    """Resolve a requested path; return a model-facing error string on failure."""
+    try:
+        target = Path(path).expanduser().resolve()
+    except OSError as exc:
+        return f"error: cannot resolve path {path!r}: {exc}"
+    if not target.exists():
+        return f"error: path does not exist: {target}"
+    return target
+
+
+def _precheck_target(target: Path, requested: AccessMode) -> str | None:
+    """Return a short-circuit message when no prompt is needed, else ``None``."""
+    if requested is AccessMode.EXEC and not target.is_dir():
+        return f"error: exec access requires a directory: {target}"
+    matched = denylist_match(target, directory=target.is_dir())
+    if matched is not None:
+        return f"error: path denied by policy (matched {matched!r}); it can never be granted: {target}"
+    if within_workspace_roots(target):
+        return f"already accessible: {target} is inside the workspace"
+    granted = granted_access_mode(target)
+    if granted is not None and mode_covers(granted, requested):
+        return f"already granted: {target} ({granted.name.lower()})"
+    if active_grant_count() >= MAX_ACCESS_GRANTS:
+        return f"error: too many active access grants (max {MAX_ACCESS_GRANTS}); revoke some first"
+    return None
+
+
+def _apply_decision(target: Path, decision: AccessDecision | str | None, timeout: float) -> str:
+    """Record an approved grant or format the denial for the model."""
+    if isinstance(decision, AccessDecision):
+        if decision.persist and get_session() is not None:
+            _ = grant_session_access(target, decision.mode)
+            scope = "session"
+        else:
+            _ = grant_process_access(target, decision.mode)
+            scope = "process"
+        return (
+            f"access granted: {target} ({decision.mode.name.lower()}, {scope})\n"
+            "note: the path denylist still applies; the human can revoke with /grants"
+        )
+    if isinstance(decision, str) and decision.strip():
+        return f"error: access to {target} denied by user; user comment: {decision.strip()}"
+    return f"error: access to {target} denied (declined or timed out after {timeout:g}s)"
+
+
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.SESSION_STATE)
+async def request_directory_access(path: str, reason: str = "", mode: str = "read") -> str:
+    """Request human approval to use a directory (or exact file) outside the workspace.
+
+    Call this instead of retrying a path that failed with ``path escapes
+    workspace root``. ``mode`` is ``read`` (inspect files), ``write`` (also
+    modify files) or ``exec`` (also use it as ``run_argv`` / PTY cwd); the human
+    may approve a different level. Approval lasts for the session (or for this
+    process when ``--yes`` auto-approves). Paths matching the configured
+    denylist can never be granted.
+    """
+    requested = parse_access_mode(mode)
+    if requested is None:
+        return f"error: invalid mode {mode!r} (use read, write, or exec)"
+    resolved = _resolve_target(path)
+    if isinstance(resolved, str):
+        return resolved
+    target = resolved
+    precheck = _precheck_target(target, requested)
+    if precheck is not None:
+        return precheck
+    hook = get_directory_access_confirm_hook()
+    if hook is None:
+        return f"error: access to {target} requires a human confirm (no confirm hook installed; denied)"
+    timeout = get_policy_confirm_timeout()
+    try:
+        decision = hook(target, requested, reason, timeout)
+    except Exception as exc:  # noqa: BLE001 — surface confirm failures to the model
+        return f"error: access confirm for {target} failed: {exc}"
+    return _apply_decision(target, decision, timeout)
+
+
+ACCESS_TOOLS = [
+    request_directory_access,
+]

src/plyngent/tools/catalog.py

@@ -218,6 +218,7 @@ def _ensure_builtin_definitions_registered(catalog: ToolCatalog) -> None:
     import time (usually the process catalog). Test ``catalog_scope(empty=True)``
     overrides must still see builtins for ``default_tool_definitions``.
     """
+    from plyngent.tools.access import ACCESS_TOOLS
     from plyngent.tools.chat import CHAT_TOOLS
     from plyngent.tools.file import FILE_TOOLS
     from plyngent.tools.net import NET_TOOLS
@@ -226,6 +227,7 @@ def _ensure_builtin_definitions_registered(catalog: ToolCatalog) -> None:
     from plyngent.tools.vcs import VCS_TOOLS
 
     for definition in (
+        *ACCESS_TOOLS,
         *FILE_TOOLS,
         *PROCESS_TOOLS,
         *VCS_TOOLS,

src/plyngent/tools/workspace.py

@@ -284,27 +284,46 @@ def _under_any_root(resolved: Path, instance: InstanceState, policy: WorkspacePo
     return False
 
 
-def _grants_cover(grants: Mapping[Path, AccessMode], resolved: Path, required: AccessMode) -> bool:
+def granted_access_mode(resolved: Path) -> AccessMode | None:
-    """Whether any granted root covers *resolved* at *required* mode.
+    """Highest config / session / process mode covering *resolved*, if any."""
-
+    policy = active_workspace_policy()
-    A grant path covers itself and its subtree; a file grant therefore only
+    session = _bound_session()
-    matches that exact file (a file has no children).
+    buckets: list[Mapping[Path, AccessMode]] = [policy.config_allow, policy.yolo_allow]
-    """
+    if session is not None:
-    for root, granted in grants.items():
+        buckets.append(session.access_grants)
-        if mode_covers(granted, required) and (resolved == root or resolved.is_relative_to(root)):
+    best: AccessMode | None = None
-            return True
+    for grants in buckets:
-    return False
+        for root, granted in grants.items():
+            if (resolved == root or resolved.is_relative_to(root)) and (best is None or granted > best):
+                best = granted
+    return best
 
 
 def _path_grant_covers(resolved: Path, required: AccessMode) -> bool:
     """Whether a config / session / process grant covers *resolved*."""
-    policy = active_workspace_policy()
+    granted = granted_access_mode(resolved)
-    if _grants_cover(policy.config_allow, resolved, required):
+    return granted is not None and mode_covers(granted, required)
-        return True
+
-    if _grants_cover(policy.yolo_allow, resolved, required):
+
-        return True
+def within_workspace_roots(resolved: Path) -> bool:
-    session = _bound_session()
+    """Whether *resolved* is under the primary root(s) or a temp allowlist root."""
-    return session is not None and _grants_cover(session.access_grants, resolved, required)
+    instance = require_bound_instance()
+    return _under_any_root(resolved, instance, instance.workspace)
+
+
+def denylist_match(resolved: Path, *, directory: bool = False) -> str | None:
+    """Return the path denylist pattern matching *resolved*, or ``None``.
+
+    *directory* also matches a trailing-separator pattern (``/secret/``) against
+    the directory path itself, not just its children.
+    """
+    resolved_str = str(resolved).replace("\\", "/")
+    if directory and not resolved_str.endswith("/"):
+        resolved_str += "/"
+    for pattern in active_workspace_policy().path_denylist:
+        if pattern and pattern.replace("\\", "/") in resolved_str:
+            return pattern
+    return None
 
 
 def resolve_path(path: str | Path, *, required: AccessMode = AccessMode.WRITE) -> Path:
@@ -329,11 +348,10 @@ def resolve_path(path: str | Path, *, required: AccessMode = AccessMode.WRITE) -
         )
         raise WorkspaceError(msg)
     # Normalize separators so denylist entries like ``/secrets/`` match on Windows.
-    resolved_str = str(resolved).replace("\\", "/")
+    matched = denylist_match(resolved)
-    for pattern in policy.path_denylist:
+    if matched is not None:
-        if pattern and pattern.replace("\\", "/") in resolved_str:
+        msg = f"path denied by policy (matched {matched!r}): {path}"
-            msg = f"path denied by policy (matched {pattern!r}): {path}"
+        raise WorkspaceError(msg)
-            raise WorkspaceError(msg)
     return resolved
 
 

tests/test_tools/test_access.py

@@ -0,0 +1,236 @@
+"""``request_directory_access`` tool and the grant stores."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from plyngent.agent import ToolTag
+from plyngent.tools import (
+    MAX_ACCESS_GRANTS,
+    AccessDecision,
+    AccessMode,
+    SessionState,
+    active_workspace_policy,
+    get_directory_access_confirm_hook,
+    grant_session_access,
+    request_directory_access,
+    resolve_path,
+    set_directory_access_confirm_hook,
+    set_path_denylist,
+)
+from plyngent.tools.context import bind_session
+from tests.test_tools.helpers import call_async
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+@pytest.fixture
+def session() -> Iterator[SessionState]:
+    state = SessionState()
+    with bind_session(state):
+        yield state
+
+
+def _never_called(*_args: object) -> AccessDecision | str | None:
+    msg = "confirm hook should not be called"
+    raise AssertionError(msg)
+
+
+def test_tool_tags_and_registry() -> None:
+    tags = request_directory_access.tags
+    assert tags & ToolTag.LOCAL
+    assert tags & ToolTag.INSTANCE_STATE
+    assert tags & ToolTag.SESSION_STATE
+    assert not (tags & ToolTag.READ_ONLY)
+    from plyngent.tools.catalog import default_tool_definitions
+
+    assert "request_directory_access" in {definition.name for definition in default_tool_definitions()}
+
+
+def test_hook_roundtrip(workspace: object, session: SessionState) -> None:
+    del workspace, session
+    assert get_directory_access_confirm_hook() is None
+    set_directory_access_confirm_hook(_never_called)
+    assert get_directory_access_confirm_hook() is not None
+    set_directory_access_confirm_hook(None)
+    assert get_directory_access_confirm_hook() is None
+
+
+async def test_approve_grants_session_access(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("approved").resolve()
+    calls: list[tuple[Path, AccessMode, str, float]] = []
+
+    def hook(path: Path, mode: AccessMode, reason: str, timeout: float) -> AccessDecision:
+        calls.append((path, mode, reason, timeout))
+        return AccessDecision(AccessMode.WRITE)
+
+    set_directory_access_confirm_hook(hook)
+    out = await call_async(request_directory_access, str(outside), "need the dataset", "read")
+    assert "access granted" in out
+    assert "write, session" in out
+    assert calls == [(outside, AccessMode.READ, "need the dataset", 30.0)]
+    assert session.access_grants[outside] == AccessMode.WRITE
+    target = outside / "data.txt"
+    _ = target.write_text("x", encoding="utf-8")
+    assert resolve_path(str(target), required=AccessMode.WRITE) == target
+
+
+async def test_process_grant_when_not_persisting(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("process").resolve()
+    set_directory_access_confirm_hook(lambda *_a: AccessDecision(AccessMode.READ, persist=False))
+    out = await call_async(request_directory_access, str(outside), mode="read")
+    assert "read, process" in out
+    assert session.access_grants == {}
+    assert active_workspace_policy().yolo_allow[outside] == AccessMode.READ
+
+
+async def test_deny_with_comment(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("denied").resolve()
+    set_directory_access_confirm_hook(lambda *_a: "no, that is my tax data")
+    out = await call_async(request_directory_access, str(outside))
+    assert out.startswith("error: access to")
+    assert "user comment: no, that is my tax data" in out
+    assert session.access_grants == {}
+
+
+async def test_deny_none(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("timeout").resolve()
+    set_directory_access_confirm_hook(lambda *_a: None)
+    out = await call_async(request_directory_access, str(outside))
+    assert "denied (declined or timed out" in out
+    assert session.access_grants == {}
+
+
+async def test_no_hook_denies(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("nohook").resolve()
+    set_directory_access_confirm_hook(None)
+    out = await call_async(request_directory_access, str(outside))
+    assert "no confirm hook installed" in out
+
+
+async def test_denylist_can_never_be_granted(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("secret-zone").resolve()
+    secret = outside / "secret"
+    secret.mkdir()
+    set_path_denylist(["/secret/"])
+    set_directory_access_confirm_hook(_never_called)
+    try:
+        out = await call_async(request_directory_access, str(secret), mode="read")
+        assert "can never be granted" in out
+        assert session.access_grants == {}
+    finally:
+        set_path_denylist(None)
+
+
+async def test_already_accessible_inside_workspace(
+    workspace: object,
+    session: SessionState,
+) -> None:
+    from pathlib import Path
+
+    assert isinstance(workspace, Path)
+    set_directory_access_confirm_hook(_never_called)
+    out = await call_async(request_directory_access, str(workspace), mode="read")
+    assert "already accessible" in out
+    assert session.access_grants == {}
+
+
+async def test_already_granted_skips_prompt(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("granted").resolve()
+    _ = grant_session_access(outside, AccessMode.READ)
+    set_directory_access_confirm_hook(_never_called)
+    out = await call_async(request_directory_access, str(outside), mode="read")
+    assert "already granted" in out
+
+
+async def test_invalid_mode(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace, session
+    outside = tmp_path_factory.mktemp("mode").resolve()
+    out = await call_async(request_directory_access, str(outside), mode="sudo")
+    assert "invalid mode" in out
+
+
+async def test_exec_requires_directory(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace, session
+    outside = tmp_path_factory.mktemp("exec-file").resolve()
+    target = outside / "f.txt"
+    _ = target.write_text("x", encoding="utf-8")
+    out = await call_async(request_directory_access, str(target), mode="exec")
+    assert "exec access requires a directory" in out
+
+
+async def test_missing_path(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace, session
+    missing = tmp_path_factory.mktemp("missing").resolve() / "nope"
+    out = await call_async(request_directory_access, str(missing))
+    assert "does not exist" in out
+
+
+async def test_grant_cap(
+    workspace: object,
+    session: SessionState,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    del workspace
+    outside = tmp_path_factory.mktemp("cap").resolve()
+    policy = active_workspace_policy()
+    for index in range(MAX_ACCESS_GRANTS):
+        policy.config_allow[outside / f"d{index}"] = AccessMode.READ
+    set_directory_access_confirm_hook(_never_called)
+    try:
+        out = await call_async(request_directory_access, str(outside))
+        assert "too many active access grants" in out
+        assert session.access_grants == {}
+    finally:
+        policy.config_allow.clear()

tests/test_tools/test_catalog.py

@@ -6,6 +6,7 @@ import inspect
 
 from plyngent.agent import ToolRegistry, ToolTag, tool
 from plyngent.tools import default_tool_definitions, register_builtin_tools
+from plyngent.tools.access import ACCESS_TOOLS
 from plyngent.tools.catalog import ToolCatalog, ToolSource, catalog_scope, get_catalog, registration_source
 from plyngent.tools.chat import CHAT_TOOLS
 from plyngent.tools.file import FILE_TOOLS
@@ -18,7 +19,7 @@ from plyngent.tools.vcs import VCS_TOOLS
 def test_default_tool_names_match_group_lists() -> None:
     register_builtin_tools()
     selected = default_tool_definitions(surface="local")
-    groups = [*FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS, *NET_TOOLS]
+    groups = [*ACCESS_TOOLS, *FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS, *NET_TOOLS]
     assert sorted(t.name for t in selected) == sorted(t.name for t in groups)
     assert len(selected) == len(groups)