---
name: Butler-Shell 系统设计
description: 基于 Inngest 与 tmux 的持久化 Shell 助手 (多通道桥接)
type: project
---

# Butler-Shell 系统设计

## 概述

Butler-Shell 是一个通过 QQ 接收指令、在远程服务器上维护持久化 Shell 会话的系统。核心特性：

- **双模式切换**：自动在自然语言模式与 PTY 透传交互模式间切换
- **命令准入**：危险命令需要用户审批
- **上下文压缩**：`/compact` 指令压缩会话上下文
- **可扩展 Skills**：插件化功能扩展

## 技术栈

| 组件 | 技术选型 |
|------|----------|
| 任务编排 | Inngest (Python SDK) + 自托管 Server |
| PTY 容器 | tmux |
| QQ 接入 | NapCatQQ (WebSocket) |
| 状态存储 | Inngest State |
| 日志存储 | SQLite |
| Agent Logic | Claude API |

### 日志存储 Schema

```sql
CREATE TABLE shell_logs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    user_id TEXT NOT NULL,
    command TEXT NOT NULL,
    output TEXT,
    executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    approved BOOLEAN DEFAULT TRUE,
    approval_id TEXT,
    mode TEXT DEFAULT 'nl',
    dangerous BOOLEAN DEFAULT FALSE
);

CREATE INDEX idx_session_logs ON shell_logs(session_id);
CREATE INDEX idx_user_logs ON shell_logs(user_id);
CREATE INDEX idx_timestamp_logs ON shell_logs(executed_at);
```

## 架构

```
┌─────────────────────────────────────────────────────────┐
│                    QQ Gateway                           │
│              (NapCatQQ WebSocket Handler)               │
└─────────────────────┬───────────────────────────────────┘
                      │ 消息事件
┌─────────────────────▼───────────────────────────────────┐
│                 Inngest Functions                       │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐       │
│  │handle_msg   │ │guardrail    │ │compaction   │       │
│  └─────────────┘ └─────────────┘ └─────────────┘       │
└─────────────────────┬───────────────────────────────────┘
                      │ 状态管理
┌─────────────────────▼───────────────────────────────────┐
│              Session Manager (TmuxWrapper)              │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐    │
│  │create_session│ │capture_pane │ │detect_mode   │    │
│  └──────────────┘ └──────────────┘ └──────────────┘    │
└─────────────────────┬───────────────────────────────────┘
                      │ PTY 控制
┌─────────────────────▼───────────────────────────────────┐
│                    tmux sessions                        │
│              [user_001] [user_002] ...                  │
└─────────────────────────────────────────────────────────┘
```

## 安全设计

### 认证与授权

| 机制 | 说明 |
|------|------|
| 用户白名单 | 仅允许配置文件中的 QQ 用户 ID 使用系统 |
| 消息签名验证 | 验证 NapCatQQ 消息来源合法性 |
| 会话令牌 | 每个会话生成唯一 UUID，防止会话劫持 |

```python
# 白名单检查
ALLOWED_USERS = os.getenv("BUTLER_ALLOWED_USERS", "").split(",")

def is_authorized(user_id: str) -> bool:
    return user_id in ALLOWED_USERS
```

### 命令安全检测

危险命令检测使用多层防护：

1. **正则匹配**：基础检测
2. **命令解析**：使用 `shlex.split()` 解析命令结构
3. **嵌套检测**：检测命令替换、管道执行等绕过方式

```python
import shlex
import re

DANGEROUS_PATTERNS = [
    # 基础危险命令
    r"\brm\s+(-[rf]+\s+|.*-rf)",
    r"\bchmod\s+(-R\s+)?777",
    r"\bchown\s+",
    r"\bdd\s+",
    r"\bmkfs\b",
    r"\b(fdisk|parted)\b",
    r">\s*/dev/sd",
    r"\biptables\b",
    r"\bshutdown\b",
    r"\breboot\b",
    # 绕过防护
    r"\bsudo\b",                    # sudo 提权
    r"\bsh\s+-c\b",                 # 命令嵌套
    r"\bbash\s+-c\b",               # 命令嵌套
    r"\|.*\b(bash|sh)\b",           # 管道执行
    r"\$\([^)]+\)",                 # 命令替换
    r"`[^`]+`",                     # 反引号命令替换
    r"fork\s*\(\s*\)",              # fork bomb
]

def is_dangerous_command(cmd: str) -> bool:
    # 正则匹配
    if any(re.search(p, cmd, re.IGNORECASE) for p in DANGEROUS_PATTERNS):
        return True
    
    # 解析命令结构
    try:
        parts = shlex.split(cmd)
        if any(p in DANGEROUS_COMMANDS for p in parts):
            return True
    except ValueError:
        # 解析失败，保守处理
        return True
    
    return False
```

### 敏感信息处理

- **终端输出过滤**：移除匹配密码、API Key 模式的文本
- **日志脱敏**：Token、密钥等不记录明文
- **会话隔离**：每个用户独立 tmux session

```python
SENSITIVE_PATTERNS = [
    r"(password|passwd|pwd)\s*=\s*\S+",
    r"(api[_-]?key|token)\s*=\s*\S+",
    r"Bearer\s+\S+",
    r"-----BEGIN.*KEY-----",
]

def sanitize_output(output: str) -> str:
    for pattern in SENSITIVE_PATTERNS:
        output = re.sub(pattern, "[REDACTED]", output, flags=re.IGNORECASE)
    return output
```

## 模式切换规则

| 切换方向 | 触发条件 | 用户可干预 |
|----------|----------|------------|
| NL → INTERACTIVE | 检测到交互进程启动 | 否 |
| INTERACTIVE → NL | 检测到进程退出 | 是，发送 `/nl` |
| INTERACTIVE → NL | 发送 `C-d` 或 `:q` | 是 |
| 强制 NL | 发送 `/force-nl` | 是 |

### 模式检测降级

当模式检测失败时（如 `/proc` 访问权限问题）：

- 默认使用 `NATURAL_LANGUAGE` 模式
- 记录 WARN 级别日志
- 提示用户可手动切换：`/interactive` 进入交互模式

## 会话生命周期管理

### 会话策略

| 配置项 | 默认值 | 说明 |
|--------|--------|------|
| `SESSION_IDLE_TIMEOUT` | 3600s | 空闲超时后自动销毁 |
| `MAX_SESSIONS_PER_USER` | 1 | 每用户最大会话数 |
| `MAX_TOTAL_SESSIONS` | 100 | 全局最大会话数 |

### 会话清理

定时任务每小时执行：

1. 扫描所有会话的最后活跃时间
2. 超过 `SESSION_IDLE_TIMEOUT` 的会话标记为待清理
3. 发送通知给用户（可选）
4. 优雅关闭 tmux 会话

```python
import asyncio

async def cleanup_idle_sessions():
    while True:
        await asyncio.sleep(3600)  # 每小时执行
        for session in get_all_sessions():
            if session.idle_seconds > SESSION_IDLE_TIMEOUT:
                await notify_user(session.user_id, "会话已超时，自动关闭")
                tmux.kill_session(session.session_id)
```

## 项目结构

```
butler-shell/
├── src/
│   └── butler/
│       ├── __init__.py
│       ├── main.py                    # FastAPI 入口 + Inngest Serve
│       │
│       ├── gateway/                   # QQ 消息接入层
│       │   ├── __init__.py
│       │   ├── napcat.py              # NapCatQQ WebSocket 客户端
│       │   └── events.py              # 消息事件定义
│       │
│       ├── workflows/                 # Inngest 工作流
│       │   ├── __init__.py
│       │   ├── handle_message.py      # 主消息处理流
│       │   ├── guardrail.py           # 命令准入审批
│       │   └── compaction.py          # 上下文压缩
│       │
│       ├── session/                   # Tmux 会话管理
│       │   ├── __init__.py
│       │   ├── wrapper.py             # TmuxWrapper 核心类
│       │   ├── detector.py            # 模式检测器
│       │   └── state.py               # 会话状态模型
│       │
│       ├── skills/                    # Skill 扩展系统
│       │   ├── __init__.py
│       │   ├── base.py                # SkillManager 基类
│       │   ├── log_skill.py           # 日志记录 Skill
│       │   └── system_skill.py        # 系统信息 Skill
│       │
│       └── config.py                  # 配置管理
│
├── tests/
│   ├── conftest.py
│   ├── test_wrapper.py
│   ├── test_detector.py
│   └── test_workflows.py
│
├── docs/
│   └── specs/
│
├── pyproject.toml
├── requirements.txt
└── docker-compose.yml                 # Inngest Server + 依赖
```

## 核心数据模型

### SessionMode & PtyState

```python
from enum import Enum
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class SessionMode(str, Enum):
    """会话运行模式"""
    NATURAL_LANGUAGE = "nl"      # 自然语言模式 - 通过 LLM 解析
    INTERACTIVE = "interactive"  # 交互模式 - 直接透传到 PTY

class PtyState(str, Enum):
    """PTY 进程状态"""
    IDLE = "idle"           # 空闲，等待输入
    RUNNING = "running"     # 正在执行命令
    BLOCKED = "blocked"     # 阻塞中（如等待 I/O）
    UNKNOWN = "unknown"     # 状态检测失败

class SessionState(BaseModel):
    """会话状态快照"""
    session_id: str = Field(..., description="tmux session 名称")
    user_id: str = Field(..., description="QQ 用户 ID")
    mode: SessionMode = Field(default=SessionMode.NATURAL_LANGUAGE)
    
    # Shell 状态
    pwd: str = Field(default="", description="当前工作目录")
    env: dict[str, str] = Field(default_factory=dict, description="关键环境变量")
    last_output: str = Field(default="", description="最近终端输出摘要")
    
    # 交互模式追踪
    interactive_process: Optional[str] = Field(
        default=None, 
        description="当前交互进程名 (vim, top, etc.)"
    )
    
    # 元数据
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
```

### MessageContext

```python
class MessageContext(BaseModel):
    """消息处理上下文"""
    message_id: str
    user_id: str
    content: str
    is_command: bool = False          # 是否以 / 开头
    requires_approval: bool = False   # 危险命令标记
    session: Optional[SessionState] = None
```

## TmuxWrapper 核心 API

```python
class TmuxWrapper:
    """tmux 会话封装器"""
    
    # 交互模式进程名单
    INTERACTIVE_PROCESSES = frozenset([
        "vim", "nvim", "vi",
        "top", "htop", "btop",
        "less", "more", "man",
        "claude", "aider",
        "python", "node", "irb", "ipython",
    ])
    
    # 快捷键映射
    SPECIAL_KEYS = {
        "C-c", "C-d", "C-z", "C-a", "C-e",
        "Esc", "Tab", "Up", "Down", "Left", "Right",
        "Enter", "Space",
    }
    
    def create_session(self, user_id: str) -> str:
        """为用户创建独立 tmux session"""
        
    def session_exists(self, session_name: str) -> bool:
        """检查 session 是否存在"""
        
    def kill_session(self, session_name: str) -> None:
        """销毁 session"""
        
    def send_keys(self, session_name: str, keys: str, enter: bool = True) -> None:
        """向 session 发送按键"""
        
    def send_raw(self, session_name: str, data: str) -> None:
        """发送原始数据（交互模式精确控制）"""
        
    def capture_pane(self, session_name: str, as_ansi: bool = False) -> str:
        """抓取终端屏幕内容"""
        
    def get_active_process(self, session_name: str) -> Optional[str]:
        """获取 tmux 窗口中的活跃进程名"""
        
    def get_pty_state(self, session_name: str) -> PtyState:
        """检测 PTY 是否在等待输入（通过 /proc 状态监控）"""
        
    def detect_mode(self, session_name: str) -> SessionMode:
        """混合模式检测：进程名单匹配 + PTY 状态检测"""
```

## Inngest 工作流

### handle_im_message

```
Step 1: get-or-create-session → butler_<user_id>
Step 2: load-state → SessionState
Step 3: detect-mode → NATURAL_LANGUAGE | INTERACTIVE

INTERACTIVE 模式:
  └─ send_keys(content, enter=False) → capture_pane() → 返回快照

NATURAL_LANGUAGE 模式:
  ├─ Skill 匹配? → execute Skill → 返回结果
  ├─ 危险命令? → 发送 im/command-needs-approval 事件
  ├─ 特殊指令? → 执行对应逻辑
  └─ send_keys(content, enter=True) → capture_pane() → 返回输出
```

### command_guardrail

```
Step 1: send-confirmation-request → 发送确认请求到 QQ
Step 2: wait-approval → ctx.step.wait_for_event (最长 5 分钟)
Step 3: 处理用户回复 → 执行或取消
```

### context_compaction

```
Step 1: get-pwd → 获取当前工作目录
Step 2: get-env → 获取关键环境变量
Step 3: get-output → 抓取最近终端输出
Step 4: summarize → LLM 总结（可选）
Step 5: 更新状态快照 → ctx.state.set()
```

## 消息处理流程

```
QQ 消息到达
     │
     ▼
NapCatGateway ──→ 转换为 NapCatEvent
     │
     ▼
Inngest Event ──→ 发送 im/message 事件
     │
     ▼
handle_im_message workflow
     │
     ├─ INTERACTIVE 模式 ──→ send_keys + capture_pane
     │
     └─ NATURAL_LANGUAGE 模式
          ├─ Skill 匹配 ──→ execute Skill
          ├─ 危险命令 ──→ command_guardrail
          └─ 正常命令 ──→ execute + capture
     │
     ▼
NapCatGateway ──→ 发送 QQ 回复
```

## 危险命令检测

扩展集合，包括：

| 命令模式 | 说明 |
|----------|------|
| `rm -rf` | 递归强制删除 |
| `chmod 777` | 开放所有权限 |
| `chown` | 修改所有者 |
| `dd` | 磁盘操作 |
| `mkfs` | 格式化 |
| `fdisk`/`parted` | 分区操作 |
| `> /dev/sd*` | 写入磁盘设备 |
| `iptables` | 防火墙规则 |
| `shutdown`/`reboot` | 系统关机/重启 |

## Skill 扩展系统

### 基类 API

```python
class SkillBase(ABC):
    name: str
    description: str
    triggers: list[str]
    
    @abstractmethod
    async def execute(self, ctx: SkillContext) -> SkillResult:
        pass
    
    def matches(self, content: str) -> bool:
        """检查是否匹配触发条件"""
```

### 内置 Skills

| Skill | 触发器 | 功能 |
|-------|--------|------|
| LogSkill | `/log`, `/history` | 记录和检索 Shell 交互日志 |
| SystemSkill | `/sys`, `/system`, `/load` | 获取系统负载和资源信息 |

## 错误处理

### 错误分类

| 类型 | 说明 | 处理方式 |
|------|------|----------|
| TRANSIENT | 暂时性错误 | Inngest 自动重试 |
| PERMANENT | 永久性错误 | 返回用户友好消息 |
| USER_ERROR | 用户输入错误 | 提示正确用法 |

### 错误分类详情

| 错误类型 | 具体错误 | 分类 | 处理策略 |
|----------|----------|------|----------|
| tmux session 不存在 | `SessionNotFoundError` | TRANSIENT | 自动重建 session |
| tmux 命令超时 | `TmuxTimeoutError` | TRANSIENT | 重试 3 次，指数退避 |
| NapCat 连接断开 | `GatewayDisconnectedError` | TRANSIENT | 重连 + 重试 |
| 用户无权限 | `UnauthorizedError` | PERMANENT | 直接返回错误消息 |
| 命令被拒绝 | `CommandRejectedError` | PERMANENT | 记录日志，通知用户 |
| 审批超时 | `ApprovalTimeoutError` | USER_ERROR | 提示重新发起 |

### 重试策略

```python
RETRY_CONFIG = {
    "max_retries": 3,
    "initial_delay": 1,  # 秒
    "max_delay": 30,
    "multiplier": 2,     # 指数退避
}

async def retry_with_backoff(fn, *args, **kwargs):
    delay = RETRY_CONFIG["initial_delay"]
    for attempt in range(RETRY_CONFIG["max_retries"]):
        try:
            return await fn(*args, **kwargs)
        except TransientError as e:
            if attempt == RETRY_CONFIG["max_retries"] - 1:
                raise
            await asyncio.sleep(delay)
            delay = min(delay * RETRY_CONFIG["multiplier"], RETRY_CONFIG["max_delay"])
```

### 关键错误类型

- `SessionNotFoundError`: 会话不存在（可恢复）
- `TmuxCommandError`: tmux 命令执行失败（可恢复）
- `ApprovalTimeoutError`: 审批超时

## 测试策略

| 层级 | 测试类型 | 覆盖目标 |
|------|----------|----------|
| TmuxWrapper | 单元测试 | 100% 方法覆盖 |
| 模式检测 | 单元测试 | 所有进程类型 |
| 工作流 | 集成测试 | 主路径 + 错误路径 |
| Skill | 单元测试 | 所有触发场景 |
| 端到端 | E2E 测试 | 完整消息流程 |

### E2E 测试场景

| 场景 | 描述 | 验证点 |
|------|------|--------|
| 新用户首次交互 | 发送消息，创建会话 | 会话创建成功 |
| 正常命令执行 | `ls -la` | 返回正确输出 |
| 危险命令审批 | `rm -rf /tmp/test` | 进入审批流程 |
| 审批超时 | 不响应审批请求 | 5分钟后超时取消 |
| 审批确认 | 回复"确认" | 命令执行成功 |
| 审批拒绝 | 回复"取消" | 命令不执行 |
| 交互模式切换 | 在 vim 中发送按键 | 正确透传 |
| 模式自动恢复 | 退出 vim | 切换回 NL 模式 |
| 手动模式切换 | 发送 `/interactive` | 进入交互模式 |
| 未授权用户 | 白名单外用户发消息 | 返回拒绝消息 |
| 会话超时清理 | 等待超过超时时间 | 会话被清理 |

## 可观测性

### 日志级别

| 级别 | 使用场景 |
|------|----------|
| DEBUG | 详细执行流程、模式检测详情 |
| INFO | 会话创建/销毁、命令执行 |
| WARN | 模式检测失败、降级操作 |
| ERROR | 命令执行失败、外部服务异常 |

### 关键指标

| 指标名 | 类型 | 说明 |
|--------|------|------|
| `butler_sessions_active` | Gauge | 当前活跃会话数 |
| `butler_commands_total` | Counter | 命令执行总数 (按状态分组) |
| `butler_command_duration_seconds` | Histogram | 命令执行耗时 |
| `butler_mode_switches_total` | Counter | 模式切换次数 |
| `butler_approvals_total` | Counter | 审批请求总数 |

### 健康检查端点

| 端点 | 说明 | 检查项 |
|------|------|--------|
| `GET /health` | 服务存活 | 进程运行状态 |
| `GET /ready` | 服务就绪 | NapCat 连接、tmux 可用性 |

## 配置管理

### 配置文件

```yaml
# config.yaml
butler:
  session:
    idle_timeout: 3600
    max_per_user: 1
    max_total: 100
    
  guardrail:
    enabled: true
    approval_timeout: 300
    danger_patterns:
      - "rm -rf"
      - "sudo"
      - "chmod 777"
      
  security:
    allowed_users:
      - "12345678"
      - "87654321"
      
  logging:
    level: INFO
    path: /var/log/butler/shell.log
    
  napcat:
    ws_url: ws://localhost:3001
    token: ${NAPCAT_TOKEN}
```

### 配置热更新

支持运行时重新加载配置：

- `POST /config/reload` 触发重载
- 监听 `SIGHUP` 信号

## 部署配置

### docker-compose.yml

```yaml
version: '3.8'
services:
  inngest:
    image: inngest/inngest:latest
    ports:
      - "8288:8288"
    command: start --url http://host.docker.internal:8000/api/inngest
    extra_hosts:
      - "host.docker.internal:host-gateway"
  
  napcat:
    image: mlikiowa/napcat-docker:latest
    ports:
      - "3001:3001"
    volumes:
      - ./napcat-data:/app/napcat/config
```

### 环境变量

```bash
INNGEST_EVENT_KEY=your_event_key
INNGEST_SIGNING_KEY=your_signing_key
NAPCAT_WS_URL=ws://localhost:3001
NAPCAT_TOKEN=your_napcat_token
ANTHROPIC_API_KEY=your_claude_api_key
```

## 实施计划

按以下顺序实施：

1. **Phase 1: 核心会话层**
   - TmuxWrapper 实现
   - 模式检测器实现
   - 单元测试

2. **Phase 2: Inngest 工作流**
   - handle_message 工作流
   - guardrail 审批流程
   - 集成测试

3. **Phase 3: QQ Gateway**
   - NapCatQQ 集成
   - 消息转换
   - 端到端测试

4. **Phase 4: Skills 系统**
   - Skill 基类
   - LogSkill 实现
   - SystemSkill 实现

5. **Phase 5: 完善**
   - 上下文压缩
   - 错误处理优化
   - Mock 测试脚本