tools: carry the denial reason into the tool message

The four policy gates phrased their refusal themselves, and all four said
"declined or timed out". They read the `Denial` their prompt returned instead:
the command denylist, the private-host fetch policy, the skill write and the
directory-access request each name the answer they got.

`get_fetch_policy_confirm_hook` no longer coerces the answer to `bool`, which
would have reported a timeout as an allow.
This commit is contained in:
2026-10-29 23:43:04 +08:00
parent 79ccf593f6
commit 4c67f6bb04
4 changed files with 44 additions and 29 deletions
+15 -10
View File
@@ -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 = [
+15 -7
View File
@@ -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)
+6 -3
View File
@@ -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)
+8 -9
View File
@@ -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"))