feat(perm): 添加会话权限模式管理功能

实现会话权限模式管理功能,包括:
1. 在 pty_process 中定义三种权限模式标志
2. 添加 /perm 命令用于修改会话权限模式
3. 新增 run_command 工具用于执行 bot 控制命令
4. 在会话管理中支持权限模式设置
5. 添加完整的测试用例和文档说明
This commit is contained in:
Yuyao Huang (Sam)
2026-03-29 06:46:45 +08:00
parent 8dab229aaf
commit cbeafa35a5
8 changed files with 302 additions and 60 deletions
+24 -3
View File
@@ -10,7 +10,7 @@ from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Dict, List, Optional
from agent.pty_process import run_claude
from agent.pty_process import run_claude, DEFAULT_PERMISSION_MODE, VALID_PERMISSION_MODES
from agent.audit import log_interaction
logger = logging.getLogger(__name__)
@@ -30,6 +30,7 @@ class Session:
started: bool = False
idle_timeout: int = DEFAULT_IDLE_TIMEOUT
cc_timeout: float = DEFAULT_CC_TIMEOUT
permission_mode: str = field(default_factory=lambda: DEFAULT_PERMISSION_MODE)
def touch(self) -> None:
self.last_activity = asyncio.get_event_loop().time()
@@ -70,6 +71,7 @@ class SessionManager:
owner_id: str = "",
idle_timeout: int = DEFAULT_IDLE_TIMEOUT,
cc_timeout: float = DEFAULT_CC_TIMEOUT,
permission_mode: str = DEFAULT_PERMISSION_MODE,
) -> Session:
async with self._lock:
session = Session(
@@ -78,12 +80,14 @@ class SessionManager:
owner_id=owner_id,
idle_timeout=idle_timeout,
cc_timeout=cc_timeout,
permission_mode=permission_mode,
)
self._sessions[conv_id] = session
self._save()
logger.info(
"Created session %s (owner=...%s) in %s (idle=%ds, cc=%.0fs)",
conv_id, owner_id[-8:] if owner_id else "-", working_dir, idle_timeout, cc_timeout,
"Created session %s (owner=...%s) in %s (idle=%ds, cc=%.0fs, perm=%s)",
conv_id, owner_id[-8:] if owner_id else "-", working_dir,
idle_timeout, cc_timeout, permission_mode,
)
return session
@@ -98,6 +102,7 @@ class SessionManager:
cwd = session.cwd
cc_session_id = session.cc_session_id
cc_timeout = session.cc_timeout
permission_mode = session.permission_mode
first_message = not session.started
if first_message:
session.started = True
@@ -116,6 +121,7 @@ class SessionManager:
cc_session_id=cc_session_id,
resume=not first_message,
timeout=cc_timeout,
permission_mode=permission_mode,
)
log_interaction(
conv_id=conv_id,
@@ -158,6 +164,7 @@ class SessionManager:
cc_session_id=cc_session_id,
resume=not first_message,
timeout=cc_timeout,
permission_mode=permission_mode,
)
log_interaction(
@@ -195,10 +202,24 @@ class SessionManager:
"started": s.started,
"idle_timeout": s.idle_timeout,
"cc_timeout": s.cc_timeout,
"permission_mode": s.permission_mode,
}
for s in sessions
]
def set_permission_mode(self, conv_id: str, mode: str, user_id: Optional[str] = None) -> None:
"""Change the permission mode for an existing session."""
session = self._sessions.get(conv_id)
if session is None:
raise KeyError(f"No session for conv_id={conv_id!r}")
if session.owner_id and user_id and session.owner_id != user_id:
raise PermissionError(f"Session {conv_id} belongs to another user")
if mode not in VALID_PERMISSION_MODES:
raise ValueError(f"Invalid permission mode {mode!r}. Valid: {VALID_PERMISSION_MODES}")
session.permission_mode = mode
self._save()
logger.info("Set permission_mode=%s for session %s", mode, conv_id)
def _save(self) -> None:
try:
data = {cid: s.to_dict() for cid, s in self._sessions.items()}
+13 -4
View File
@@ -17,12 +17,22 @@ def strip_ansi(text: str) -> str:
return ANSI_ESCAPE.sub("", text)
PERMISSION_MODE_FLAGS: dict[str, list[str]] = {
"bypassPermissions": ["--dangerously-skip-permissions"],
"acceptEdits": ["--permission-mode", "acceptEdits"],
"plan": ["--permission-mode", "plan"],
}
VALID_PERMISSION_MODES = list(PERMISSION_MODE_FLAGS)
DEFAULT_PERMISSION_MODE = "bypassPermissions"
async def run_claude(
prompt: str,
cwd: str,
cc_session_id: str | None = None,
resume: bool = False,
timeout: float = 300.0,
permission_mode: str = DEFAULT_PERMISSION_MODE,
) -> str:
"""
Run `claude -p <prompt>` in the given directory and return the output.
@@ -35,11 +45,10 @@ async def run_claude(
- Subsequent calls: passed as --resume so CC has full history.
resume: If True, use --resume instead of --session-id.
timeout: Maximum seconds to wait before giving up.
permission_mode: One of 'bypassPermissions', 'acceptEdits', 'plan'.
"""
base_args = [
"--dangerously-skip-permissions",
"-p", prompt,
]
perm_flags = PERMISSION_MODE_FLAGS.get(permission_mode, PERMISSION_MODE_FLAGS[DEFAULT_PERMISSION_MODE])
base_args = perm_flags + ["-p", prompt]
if cc_session_id:
if resume: