diff --git a/src/plyngent/tools/access.py b/src/plyngent/tools/access.py index 18b2281..3cc79b6 100644 --- a/src/plyngent/tools/access.py +++ b/src/plyngent/tools/access.py @@ -17,7 +17,7 @@ from __future__ import annotations from pathlib import Path from typing import TYPE_CHECKING, NamedTuple -from plyngent.agent import ToolTag, tool +from plyngent.agent import Denial, ToolTag, tool from plyngent.tools.context import get_session, require_session from plyngent.tools.workspace import ( AccessMode, @@ -52,7 +52,7 @@ class AccessDecision(NamedTuple): persist: bool = True -type DirectoryAccessConfirmHook = Callable[[Path, AccessMode, str, float], AccessDecision | str | None] +type DirectoryAccessConfirmHook = Callable[[Path, AccessMode, str, float], AccessDecision | str | Denial | None] def get_directory_access_confirm_hook(instance: InstanceState | None = None) -> DirectoryAccessConfirmHook | None: @@ -62,9 +62,9 @@ def get_directory_access_confirm_hook(instance: InstanceState | None = None) -> if hook is None or not callable(hook): return None - def _as_hook(path: Path, mode: AccessMode, reason: str, timeout: float) -> AccessDecision | str | None: + def _as_hook(path: Path, mode: AccessMode, reason: str, timeout: float) -> AccessDecision | str | Denial | None: result: object = hook(path, mode, reason, timeout) - if isinstance(result, AccessDecision | str): + if isinstance(result, AccessDecision | str | Denial): return result return None @@ -232,8 +232,12 @@ def _precheck_target(target: Path, requested: AccessMode) -> str | None: 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.""" +def _apply_decision(target: Path, decision: AccessDecision | str | Denial | None) -> str: + """Record an approved grant or format the denial for the model. + + The prompt says whether the human declined or the clock ran out, so a timeout + is never reported as a decision. + """ if isinstance(decision, AccessDecision): if decision.persist and get_session() is not None: _ = grant_session_access(target, decision.mode) @@ -245,9 +249,10 @@ def _apply_decision(target: Path, decision: AccessDecision | str | None, timeout 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)" + denial = Denial.from_answer(decision) + text = f"error: access to {target} denied ({denial.clause()})" + comment = denial.user_comment + return f"{text}; user comment: {comment}" if comment else text @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.SESSION_STATE) @@ -281,7 +286,7 @@ async def request_directory_access(path: str, reason: str = "", mode: str = "rea 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) + return _apply_decision(target, decision) ACCESS_TOOLS = [ diff --git a/src/plyngent/tools/net/grants.py b/src/plyngent/tools/net/grants.py index d9a50c6..5bbd15c 100644 --- a/src/plyngent/tools/net/grants.py +++ b/src/plyngent/tools/net/grants.py @@ -10,10 +10,11 @@ from plyngent.tools.workspace import WorkspaceError, require_bound_instance if TYPE_CHECKING: from collections.abc import Callable, Sequence + from plyngent.agent.tools import Denial from plyngent.tools.context import InstanceState # (host, port, url, timeout_seconds) -> True to allow for this process/instance. -type FetchPolicyConfirmHook = Callable[[str, int, str, float], bool] +type FetchPolicyConfirmHook = Callable[[str, int, str, float], bool | Denial] def get_private_grants(instance: InstanceState | None = None) -> set[str]: @@ -48,8 +49,14 @@ def get_fetch_policy_confirm_hook(instance: InstanceState | None = None) -> Fetc if hook is None or not callable(hook): return None - def _as_hook(host: str, port: int, url: str, timeout_seconds: float) -> bool: - return bool(hook(host, port, url, timeout_seconds)) + def _as_hook(host: str, port: int, url: str, timeout_seconds: float) -> bool | Denial: + from plyngent.agent import Denial as DenialAnswer + + answer: object = hook(host, port, url, timeout_seconds) + # Pass the reason through: only an explicit True allows. + if answer is True or isinstance(answer, DenialAnswer): + return answer + return False return _as_hook @@ -89,7 +96,7 @@ async def ensure_host_allowed( raise FetchPolicyError(msg) # PRIVATE - from plyngent.agent import Approval + from plyngent.agent import Approval, Denial from plyngent.tools.context import record_approval if has_private_grant(parsed.host, parsed.port, instance=inst): @@ -109,14 +116,15 @@ async def ensure_host_allowed( ) raise FetchPolicyError(msg) try: - allowed = bool(hook(parsed.host, parsed.port, parsed.url, timeout)) + answer = hook(parsed.host, parsed.port, parsed.url, timeout) except Exception as exc: msg = f"fetch_local blocked private/loopback host {parsed.host}:{parsed.port} (confirm failed: {exc})" raise FetchPolicyError(msg) from exc - if not allowed: + if answer is not True: + denial = Denial.from_answer(answer) msg = ( f"fetch_local blocked private/loopback host {parsed.host}:{parsed.port} " - f"(user declined or timed out after {timeout:g}s; not skipped by YOLO)" + f"({assessment.reason}; {denial.clause()}; not skipped by YOLO)" ) raise FetchPolicyError(msg) _ = grant_private_host(parsed.host, parsed.port, instance=inst) diff --git a/src/plyngent/tools/skills.py b/src/plyngent/tools/skills.py index c101795..a9eefb2 100644 --- a/src/plyngent/tools/skills.py +++ b/src/plyngent/tools/skills.py @@ -18,7 +18,7 @@ from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, cast -from plyngent.agent import Approval, ToolTag, tool +from plyngent.agent import Approval, Denial, ToolTag, tool from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS from plyngent.skills import ( SKILL_FILE, @@ -261,9 +261,11 @@ class SkillWriteDecision: # True keeps the grant for the rest of this chat session (the standing # alternative is ``[skills].allow_write`` in the config). session: bool = False + # Why a denial: declined, timed out, cancelled, or nobody could be asked. + denial: Denial | None = None -type SkillWriteConfirmHook = Callable[[Path, str, str, float], SkillWriteDecision | bool | None] +type SkillWriteConfirmHook = Callable[[Path, str, str, float], SkillWriteDecision | Denial | bool | None] @dataclass(frozen=True, slots=True) @@ -329,7 +331,8 @@ def _prompt_write( return f"error: skill write confirm failed: {exc}" allowed = decision is True or (isinstance(decision, SkillWriteDecision) and decision.allow) if not allowed: - return f"error: skill write to {skill_dir} denied (declined or timed out after {policy.timeout:g}s)" + denial = decision.denial if isinstance(decision, SkillWriteDecision) else None + return f"error: skill write to {skill_dir} denied ({(denial or Denial.from_answer(decision)).clause()})" keys_session = isinstance(decision, SkillWriteDecision) and decision.session if keys_session and session is not None: session.skill_write_roots.add(skill_dir) diff --git a/src/plyngent/tools/workspace.py b/src/plyngent/tools/workspace.py index 222758e..cf3482c 100644 --- a/src/plyngent/tools/workspace.py +++ b/src/plyngent/tools/workspace.py @@ -10,6 +10,7 @@ from plyngent.tools.command_scan import basename, unwrap_command if TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence + from plyngent.agent.tools import Denial from plyngent.tools.context import InstanceState, SessionState DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset( @@ -89,7 +90,7 @@ def mode_covers(granted: AccessMode, required: AccessMode) -> bool: # Hook: (basename, argv, timeout_seconds) -> True allow for this session basename. -type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool] +type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool | Denial] class WorkspaceError(ValueError): @@ -571,7 +572,7 @@ def check_command_allowed(argv: list[str]) -> None: says it: ``user (process)`` for this call's answer, ``grant (process)`` when an earlier one already covers this program. """ - from plyngent.agent import Approval + from plyngent.agent import Approval, Denial from plyngent.tools.context import record_approval shape_error = argv_shape_error(argv) @@ -590,19 +591,17 @@ def check_command_allowed(argv: list[str]) -> None: why = f": {hit.keyword!r}" if hit.keyword else "" hook = policy.policy_confirm_hook if hook is None: - msg = f"command denied by policy (basename {binary!r} is blocked{why})" + msg = f"command denied by policy (basename {binary!r} is blocked{why}; no policy confirm hook)" raise WorkspaceError(msg) timeout = policy.policy_confirm_timeout_seconds try: - allowed = bool(hook(binary, list(argv), timeout)) + answer = hook(binary, list(argv), timeout) except Exception as exc: msg = f"command denied by policy (basename {binary!r}; confirm failed: {exc})" raise WorkspaceError(msg) from exc - if not allowed: - msg = ( - f"command denied by policy (basename {binary!r} is blocked{why}; " - f"user declined or timed out after {timeout:g}s)" - ) + if answer is not True: + denial = Denial.from_answer(answer) + msg = f"command denied by policy (basename {binary!r} is blocked{why}; {denial.clause()})" raise WorkspaceError(msg) policy.policy_allowed_commands.add(binary) record_approval(Approval("user", "process"))