JOJO
8d665ad6de
Implemented a new permission mode between and , including backend policy, runtime behavior, frontend display, and documentation updates. Backend & policy changes: - Added to permission mode validation and persistence paths. - Updated tool permission evaluation: workspace-local direct pass, out-of-workspace requires approval. - Kept readonly-first flow and added auto-approval decision path when permission denial occurs. - Added approval reason/decider support in tool approval manager. - Improved rejection tool-content format to natural language: '工具调用被拒绝\n原因：...' - Fixed permission-mode sync edge cases on conversation create/load and restart-first-switch behavior. Approval agent subsystem: - Added and . - Added lightweight auto-approval config at (name/url/key/model/extra_params + timeouts). - Added optional transcript debug switch in code and transcript output under logs/approval_agent. - Aligned transcript saving toward cumulative messages format and captured reasoning/content/tool_calls/tool sequence. Frontend changes: - Added option to personalization and permission menus. - Reworked approval panel auto-review display into a dedicated block with simplified lines: start, command, final approve/reject + reason. - Added delayed sidebar auto-close (10s) after approval resolved. - Added stricter permission-switch verification and rollback on mismatch/error. Docs updated: - Synced AGENTS.md, README.md, and docs/host_sandbox_and_permission_model.md with new permission mode, approval-agent behavior, config, and current constraints (including docker caveat).
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class ToolApprovalManager:
|
|
def __init__(self):
|
|
self._items: Dict[str, Dict[str, Any]] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def create_request(
|
|
self,
|
|
*,
|
|
username: str,
|
|
conversation_id: Optional[str],
|
|
task_id: Optional[str],
|
|
tool_call_id: Optional[str],
|
|
tool_name: str,
|
|
arguments: Dict[str, Any],
|
|
preview: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
approval_id = f"approval_{uuid.uuid4().hex}"
|
|
item = {
|
|
"approval_id": approval_id,
|
|
"username": username,
|
|
"conversation_id": conversation_id,
|
|
"task_id": task_id,
|
|
"tool_call_id": tool_call_id,
|
|
"tool_name": tool_name,
|
|
"arguments": arguments or {},
|
|
"preview": preview or {},
|
|
"status": "pending",
|
|
"created_at": time.time(),
|
|
"decided_at": None,
|
|
"decision": None,
|
|
}
|
|
with self._lock:
|
|
self._items[approval_id] = item
|
|
return dict(item)
|
|
|
|
def get(self, approval_id: str) -> Optional[Dict[str, Any]]:
|
|
with self._lock:
|
|
item = self._items.get(approval_id)
|
|
return dict(item) if item else None
|
|
|
|
def list_pending(self, username: str, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
with self._lock:
|
|
rows = []
|
|
for item in self._items.values():
|
|
if item.get("username") != username:
|
|
continue
|
|
if item.get("status") != "pending":
|
|
continue
|
|
if conversation_id and item.get("conversation_id") != conversation_id:
|
|
continue
|
|
rows.append(dict(item))
|
|
rows.sort(key=lambda x: x.get("created_at", 0.0))
|
|
return rows
|
|
|
|
def decide(
|
|
self,
|
|
approval_id: str,
|
|
username: str,
|
|
decision: str,
|
|
*,
|
|
reason: Optional[str] = None,
|
|
decider: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
normalized = str(decision or "").strip().lower()
|
|
if normalized not in {"approved", "rejected"}:
|
|
raise ValueError("decision 仅支持 approved / rejected")
|
|
with self._lock:
|
|
item = self._items.get(approval_id)
|
|
if not item:
|
|
raise KeyError("审批请求不存在")
|
|
if item.get("username") != username:
|
|
raise PermissionError("无权限操作该审批请求")
|
|
if item.get("status") != "pending":
|
|
return dict(item)
|
|
item["status"] = normalized
|
|
item["decision"] = normalized
|
|
item["decided_at"] = time.time()
|
|
if isinstance(reason, str) and reason.strip():
|
|
item["reason"] = reason.strip()
|
|
if isinstance(decider, str) and decider.strip():
|
|
item["decider"] = decider.strip()
|
|
return dict(item)
|