diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 5424540d..6520253c 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -58,6 +58,7 @@ from backend.apps.agents.manager.permissions import path_gate from backend.apps.agents.manager import context_budget from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState from backend.apps.agents.manager.streaming import thinking as thinking_mod +from backend.apps.agents.manager.permissions.decision import effective_policy, request_user_approval from backend.apps.agents.manager.session.workspace_git import _detect_git_identity, _ensure_cwd_git_repo from backend.apps.agents.manager.prompt.tool_catalog import ( FULL_TOOLS, @@ -456,140 +457,18 @@ class AgentManager: # who want a prompt on every command can flip Bash to "ask" in the UI. _DEFAULTS: dict[str, str] = {} - def _default_for(tool_name: str) -> str: - return _DEFAULTS.get(tool_name, "always_allow") - - def _get_effective_policy(tool_name: str) -> str: - """Return 'always_allow', 'deny', or 'ask' for any tool. Keyed through - the shared resolver so the read slot matches the write slot exactly.""" - tools = load_all_tools() - slot = resolve_policy_slot(tool_name, tools) - if slot.store == "builtin": - return _builtin_perms.get(slot.key, _default_for(slot.key)) - if slot.key is not None: - for t in tools: - if t.id == slot.key: - return t.tool_permissions.get(slot.action, "ask") - return _default_for(tool_name) - - def _set_tool_policy(tool_name: str, policy: str) -> None: - """Inverse of _get_effective_policy: persist `policy` into the SAME slot - the gate reads, AND update the live in-memory snapshot, so an 'Always - approve' takes effect for this running agent, not only after a restart. - (The old code wrote the raw tool name to the file and never touched the - captured _builtin_perms, so it behaved like a one-time accept.)""" - tools = load_all_tools() - slot = resolve_policy_slot(tool_name, tools) - if slot.store == "builtin": - _builtin_perms[slot.key] = policy - perms = load_builtin_permissions() - perms[slot.key] = policy - save_builtin_permissions(perms) - return - if slot.key is not None: - for t in tools: - if t.id == slot.key: - t.tool_permissions[slot.action] = policy - save_tool(t) - return - - async def _request_user_approval( - tool_name: str, - tool_input, - sensitive_pattern: str | None = None, - ) -> dict: - """Send an approval request via WebSocket and wait for the user's decision.""" - safe_input = tool_input if isinstance(tool_input, dict) else {} - request_id = uuid4().hex - label, why = (None, None) - if sensitive_pattern: - described = path_gate.describe_sensitive_pattern(sensitive_pattern) - if described: - label, why = described - approval_req = ApprovalRequest( - id=request_id, - session_id=session_id, - tool_name=tool_name, - tool_input=safe_input, - sensitive_pattern=sensitive_pattern, - sensitive_label=label, - sensitive_why=why, - ) - session.pending_approvals.append(approval_req) - session.status = "waiting_approval" - - - await ws_manager.send_to_session(session_id, "agent:status", { - "session_id": session_id, - "status": "waiting_approval", - }) - - decision = await ws_manager.send_approval_request( - session_id, request_id, tool_name, safe_input, - sensitive_pattern=sensitive_pattern, - sensitive_label=label, - sensitive_why=why, - ) - # If the user opted into trusting this pattern, persist now so - # any subsequent prompt against the same pattern (e.g. the - # PreToolUse hook re-evaluating after can_use_tool, or a later - # Write in the same session) skips the modal silently. - if ( - decision.get("behavior") == "allow" - and decision.get("trust_pattern") - and sensitive_pattern - ): - try: - existing = load_trusted_sensitive_paths() - if sensitive_pattern not in existing: - existing.append(sensitive_pattern) - save_trusted_sensitive_paths(existing) - except Exception: - logger.exception("Failed to persist trusted sensitive path") - - # "Always approve" button: persist the tool's policy so it stops - # prompting. The guards above (sensitive/catastrophic) re-fire even - # on always_allow, so this can't disarm an rm -rf or a key-path write. - if decision.get("behavior") == "allow" and decision.get("set_always_allow"): - try: - _set_tool_policy(tool_name, "always_allow") - except Exception: - logger.exception("Failed to persist always-allow for %s", tool_name) - - approval_latency_ms = int((datetime.now() - approval_req.created_at).total_seconds() * 1000) - try: - # Append to the session's approval log so a reload - # restores the full HITL timeline. - session.approval_decisions.append({ - "tool": tool_name, - "behavior": decision.get("behavior"), - "decision_ms": approval_latency_ms, - }) - except Exception: - pass - - session.pending_approvals = [ - a for a in session.pending_approvals if a.id != request_id - ] - session.status = "running" - await ws_manager.send_to_session(session_id, "agent:status", { - "session_id": session_id, - "status": "running", - }) - return decision - async def can_use_tool(tool_name, input_data, context): sensitive_pattern: str | None = None if tool_name != "AskUserQuestion": policy, sensitive_pattern = path_gate.maybe_override_policy( - _get_effective_policy(tool_name), tool_name, input_data + effective_policy(tool_name, _builtin_perms, _DEFAULTS), tool_name, input_data ) if policy == "always_allow": return PermissionResultAllow(updated_input=input_data) if policy == "deny": return PermissionResultDeny(message="Tool denied by permission policy") - decision = await _request_user_approval(tool_name, input_data, sensitive_pattern=sensitive_pattern) + decision = await request_user_approval(session, session_id, tool_name, input_data, _builtin_perms, sensitive_pattern=sensitive_pattern) if decision.get("behavior") == "allow": return PermissionResultAllow( updated_input=decision.get("updated_input", input_data) @@ -680,7 +559,7 @@ class AgentManager: if tool_name and tool_name != "AskUserQuestion": tool_input = input_data.get("tool_input", {}) policy, sensitive_pattern = path_gate.maybe_override_policy( - _get_effective_policy(tool_name), tool_name, tool_input + effective_policy(tool_name, _builtin_perms, _DEFAULTS), tool_name, tool_input ) if policy == "deny": @@ -693,7 +572,7 @@ class AgentManager: } if policy == "ask": - decision = await _request_user_approval(tool_name, tool_input, sensitive_pattern=sensitive_pattern) + decision = await request_user_approval(session, session_id, tool_name, tool_input, _builtin_perms, sensitive_pattern=sensitive_pattern) if decision.get("behavior") == "allow": if tool_use_id: diff --git a/backend/apps/agents/manager/permissions/decision.py b/backend/apps/agents/manager/permissions/decision.py new file mode 100644 index 00000000..3a2abb4d --- /dev/null +++ b/backend/apps/agents/manager/permissions/decision.py @@ -0,0 +1,138 @@ +"""Permission-policy resolution + the HITL approval flow, lifted out of the agent loop. +effective_policy/set_tool_policy resolve and persist a tool's policy through the SAME slot +(builtin vs custom tool), and request_user_approval surfaces the approval card and waits for +the user's decision (the wait itself lives in ws_manager.send_approval_request). builtin_perms +is the live in-memory snapshot the loop also reads, threaded in so an 'Always approve' takes +effect for the running agent, not only after a restart.""" + +import logging +from datetime import datetime +from typing import Dict, Optional +from uuid import uuid4 + +from typeguard import typechecked + +from backend.apps.agents.core.models import AgentSession, ApprovalRequest +from backend.apps.agents.core.ws_manager import ws_manager +from backend.apps.agents.manager.permissions import path_gate +from backend.apps.tools_lib.tools_lib import ( + _load_all as load_all_tools, + _save as save_tool, + load_builtin_permissions, + load_trusted_sensitive_paths, + resolve_policy_slot, + save_builtin_permissions, + save_trusted_sensitive_paths, +) + +logger = logging.getLogger(__name__) + + +@typechecked +def effective_policy(tool_name: str, builtin_perms: Dict[str, str], defaults: Dict[str, str]) -> str: + """'always_allow', 'deny', or 'ask' for any tool, keyed through the shared slot resolver + so the read slot matches the write slot exactly.""" + tools = load_all_tools() + slot = resolve_policy_slot(tool_name, tools) + if slot.store == "builtin": + return builtin_perms.get(slot.key, defaults.get(slot.key, "always_allow")) + if slot.key is not None: + for t in tools: + if t.id == slot.key: + return t.tool_permissions.get(slot.action, "ask") + return defaults.get(tool_name, "always_allow") + + +@typechecked +def set_tool_policy(tool_name: str, policy: str, builtin_perms: Dict[str, str]) -> None: + """Persist `policy` into the SAME slot effective_policy reads AND update the live + in-memory snapshot, so an 'Always approve' takes effect immediately.""" + tools = load_all_tools() + slot = resolve_policy_slot(tool_name, tools) + if slot.store == "builtin": + builtin_perms[slot.key] = policy + perms = load_builtin_permissions() + perms[slot.key] = policy + save_builtin_permissions(perms) + return + if slot.key is not None: + for t in tools: + if t.id == slot.key: + t.tool_permissions[slot.action] = policy + save_tool(t) + return + + +@typechecked +async def request_user_approval( + session: AgentSession, + session_id: str, + tool_name: str, + tool_input: object, + builtin_perms: Dict[str, str], + sensitive_pattern: Optional[str] = None, +) -> Dict[str, object]: + """Send an approval request over WS and wait for the user's decision.""" + safe_input = tool_input if isinstance(tool_input, dict) else {} + request_id = uuid4().hex + label, why = (None, None) + if sensitive_pattern: + described = path_gate.describe_sensitive_pattern(sensitive_pattern) + if described: + label, why = described + approval_req = ApprovalRequest( + id=request_id, + session_id=session_id, + tool_name=tool_name, + tool_input=safe_input, + sensitive_pattern=sensitive_pattern, + sensitive_label=label, + sensitive_why=why, + ) + session.pending_approvals.append(approval_req) + session.status = "waiting_approval" + await ws_manager.send_to_session(session_id, "agent:status", { + "session_id": session_id, + "status": "waiting_approval", + }) + decision = await ws_manager.send_approval_request( + session_id, request_id, tool_name, safe_input, + sensitive_pattern=sensitive_pattern, + sensitive_label=label, + sensitive_why=why, + ) + # Persist a trusted sensitive-path so later prompts for the same pattern skip the modal. + if ( + decision.get("behavior") == "allow" + and decision.get("trust_pattern") + and sensitive_pattern + ): + try: + existing = load_trusted_sensitive_paths() + if sensitive_pattern not in existing: + existing.append(sensitive_pattern) + save_trusted_sensitive_paths(existing) + except Exception: + logger.exception("Failed to persist trusted sensitive path") + # "Always approve": persist the tool policy (the sensitive/catastrophic guards still re-fire). + if decision.get("behavior") == "allow" and decision.get("set_always_allow"): + try: + set_tool_policy(tool_name, "always_allow", builtin_perms) + except Exception: + logger.exception("Failed to persist always-allow for %s", tool_name) + approval_latency_ms = int((datetime.now() - approval_req.created_at).total_seconds() * 1000) + try: + session.approval_decisions.append({ + "tool": tool_name, + "behavior": decision.get("behavior"), + "decision_ms": approval_latency_ms, + }) + except Exception: + pass + session.pending_approvals = [a for a in session.pending_approvals if a.id != request_id] + session.status = "running" + await ws_manager.send_to_session(session_id, "agent:status", { + "session_id": session_id, + "status": "running", + }) + return decision diff --git a/backend/tests/test_tool_policy_slot.py b/backend/tests/test_tool_policy_slot.py index 753dbbcf..bbf62ea8 100644 --- a/backend/tests/test_tool_policy_slot.py +++ b/backend/tests/test_tool_policy_slot.py @@ -45,7 +45,7 @@ def test_slot_for_unknown_mcp_has_no_write_target(): # read/write mirror the dispatch-gate branches in agent_manager -# (_get_effective_policy / _set_tool_policy): both key through resolve_policy_slot. +# (effective_policy / set_tool_policy): both key through resolve_policy_slot. def _read(tool_name, builtin_perms, tools): slot = resolve_policy_slot(tool_name, tools) if slot.store == "builtin":