Agent 执行器设计：沙箱、工作树、命令代理与结果回传

2026-05-22 14:32  Agent 接到任务："修复支付模块的金额计算精度问题"
2026-05-22 14:33  Agent 修改了 src/payment/calculate.js
2026-05-22 14:33  Agent 想跑测试，执行了 npm test
2026-05-22 14:34  测试脚本里有 postinstall：rm -rf /tmp/build && ./scripts/sync-prod-data.sh
2026-05-22 14:34  sync-prod-data.sh 把生产数据库的 10 万条订单拉到了本地
2026-05-22 14:35  Agent 把包含真实订单号的日志作为上下文发给了模型

┌─────────────────────────────────────────────────────────────────┐
│                        Task Dispatcher                         │
│                 (领取任务、分配执行器、管理超时)                   │
└────────────────────────────┬────────────────────────────────────┘
                             │  dispatch(task_id, payload)
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│                         Agent Executor                         │
│                                                                 │
│  ┌──────────────┐  ┌───────────────┐  ┌────────────────────┐  │
│  │   Worktree   │  │   Sandbox     │  │  Command Proxy     │  │
│  │   Manager    │→ │   (fs, net,   │→ │  (classify → run   │  │
│  │              │  │    env)       │  │   or ask approval) │  │
│  └──────┬───────┘  └───────┬───────┘  └─────────┬──────────┘  │
│         │                  │                     │              │
│         │           ┌──────▼───────┐      ┌──────▼────────┐   │
│         │           │ Agent 进程    │      │ 命令审计日志   │   │
│         │           │ (Claude/...)  │      │               │   │
│         │           └──────┬───────┘      └───────────────┘   │
│         │                  │                                   │
│         ▼                  ▼                                   │
│  ┌──────────────────────────────────────────────────────────┐ │
│  │                    Artifact Collector                    │ │
│  │  (git diff, test report, log tail, error trace)          │ │
│  └───────────────────────────┬──────────────────────────────┘ │
└──────────────────────────────┼─────────────────────────────────┘
                               │  upload(task_id, artifacts)
                               ▼
                    ┌─────────────────────┐
                    │   Artifact Store    │
                    │   (S3 / OSS / Minio) │
                    └─────────────────────┘

任务入队
    │
    ▼
┌──────────────────────────┐
│ 1. git worktree add      │  → .worktrees/task-<id>-<branch>
│    --detach <base-commit> │    (基于目标分支最新 commit)
└──────────┬───────────────┘
           │
           ▼
┌──────────────────────────┐
│ 2. Agent 在此目录工作     │  → 所有文件改动只影响此 worktree
│    修改代码、跑测试、提交  │    主分支完全不受影响
└──────────┬───────────────┘
           │
           ▼
┌──────────────────────────┐
│ 3. 收集 artifact         │  → git diff, git log, test report
│    打包上传 Artifact Store │
└──────────┬───────────────┘
           │
           ▼
┌──────────────────────────┐
│ 4. git worktree remove   │  → 清理磁盘
│    (任务成功时保留分支)    │    失败任务立即清理
└──────────────────────────┘

import subprocess
import shutil
from pathlib import Path

class WorktreeManager:
    """Git Worktree 生命周期管理"""

    def __init__(self, repo_root: str, worktree_base: str = ".worktrees"):
        self.repo_root = Path(repo_root)
        self.worktree_base = Path(worktree_base)
        self.worktree_base.mkdir(exist_ok=True)

    def create(self, task_id: str, base_branch: str = "main") -> Path:
        """为任务创建独立 worktree，返回路径"""
        # 基于目标分支最新 commit，而不是当前 HEAD
        base_commit = self._run_git("rev-parse", base_branch).strip()
        branch_name = f"agent/task-{task_id}"
        worktree_path = self.worktree_base / f"task-{task_id}"

        # 创建新分支 + worktree
        self._run_git(
            "worktree", "add", "-b", branch_name,
            str(worktree_path), base_commit
        )
        return worktree_path

    def collect_diff(self, worktree_path: Path, base_branch: str = "main") -> str:
        """收集 worktree 相对 base_branch 的所有改动"""
        return self._run_git(
            "-C", str(worktree_path),
            "diff", f"origin/{base_branch}...HEAD"
        )

    def cleanup(self, worktree_path: Path, keep_branch: bool = False):
        """清理 worktree，可选保留分支（用于 PR）"""
        self._run_git("worktree", "remove", "--force", str(worktree_path))
        if not keep_branch:
            # 提取分支名
            branch = worktree_path.name.replace("task-", "agent/task-")
            self._run_git("branch", "-D", branch)

    def _run_git(self, *args) -> str:
        result = subprocess.run(
            ["git"] + list(args),
            cwd=self.repo_root,
            capture_output=True, text=True, timeout=30, check=True
        )
        return result.stdout

# executor-sandbox.yaml
sandbox:
  # 文件系统：只允许访问 worktree 目录 + 显式声明的共享目录
  filesystem:
    allowed_paths:
      - path: "{{worktree}}"           # 当前任务的 worktree（动态替换）
        mode: readwrite
      - path: "/tmp/agent-{{task_id}}" # 任务专属临时目录
        mode: readwrite
      - path: "{{repo_root}}/node_modules"
        mode: readonly                  # 依赖目录只读
      - path: "{{repo_root}}/docs"
        mode: readonly                  # 文档目录只读
    denied_paths:
      - "/etc"
      - "/var"
      - "~/.ssh"
      - "~/.aws"
      - "~/.gnupg"
      - "/root"
      # 即使 worktree 在 /home 下，也不允许越权到父目录
      - "{{worktree}}/.."

  # 网络：默认拒绝所有出站，只放行显式域名
  network:
    default_policy: deny
    allowed_egress:
      - host: "registry.npmjs.org"     # npm 安装依赖
        ports: [443]
      - host: "pypi.org"               # pip 安装依赖
        ports: [443]
      - host: "github.com"             # 拉 tag、release
        ports: [443]
      - host: "api.openai.com"         # 模型 API（如果需要）
        ports: [443]
      - host: "api.anthropic.com"
        ports: [443]
    denied_egress:
      - "0.0.0.0/8"                   # 内网全禁
      - "10.0.0.0/8"
      - "172.16.0.0/12"
      - "192.168.0.0/16"
      - "169.254.0.0/16"              # 云元数据服务

  # 环境变量：脱敏，禁止把密钥传给 Agent 子进程
  env:
    deny_patterns:
      - "*PASSWORD*"
      - "*SECRET*"
      - "*TOKEN*"
      - "*KEY*"
      - "AWS_*"
      - "DATABASE_URL"
    allow_explicit:
      NODE_ENV: "test"
      CI: "true"

import re
from enum import Enum
from dataclasses import dataclass
from typing import Optional

class RiskLevel(Enum):
    SAFE = "safe"            # 只读、无副作用，直接执行
    MODERATE = "moderate"    # 有副作用但可逆，记录审计后执行
    HIGH = "high"            # 高风险，需人工审批
    FORBIDDEN = "forbidden"  # 绝对禁止，直接拒绝

@dataclass
class CommandVerdict:
    level: RiskLevel
    reason: str
    approver: Optional[str] = None  # 高风险命令的指定审批人

class CommandClassifier:
    """命令分级器：根据预定义规则对命令分类"""

    # 只读命令：查看状态、读文件、跑测试
    SAFE_PATTERNS = [
        r"^ls(\s|$)", r"^cat\s+", r"^head\s+", r"^tail\s+",
        r"^grep\s+", r"^find\s+", r"^wc\s+", r"^file\s+",
        r"^git\s+(status|log|diff|show|branch|tag)\b",
        r"^npm\s+test(\s|$)", r"^npm\s+run\s+(test|lint|build)(\s|$)",
        r"^pytest\b", r"^jest\b", r"^mocha\b",
        r"^echo\s+", r"^pwd$", r"^which\s+", r"^type\s+",
    ]

    # 中等风险：可逆的修改
    MODERATE_PATTERNS = [
        r"^git\s+(add|commit|checkout|reset|revert)\b",
        r"^npm\s+install\s+", r"^pip\s+install\s+",
        r"^mkdir\s+", r"^cp\s+", r"^mv\s+", r"^touch\s+",
        r"^chmod\s+", r"^chown\s+",
    ]

    # 高风险：不可逆或影响外部
    HIGH_PATTERNS = [
        r"^git\s+(push|merge|rebase)\b",
        r"^rm\s+-rf\s+/", r"^rm\s+-rf\s+~",
        r"curl\s+.*\s(-X\s*POST|-X\s*PUT|-X\s*DELETE)\s+",
        r"npm\s+publish\b", r"pip\s+upload\b",
        r"docker\s+(push|rm|stop)\b",
        r"kubectl\s+(apply|delete|rollout)\b",
        r"^ssh\s+", r"^scp\s+",
    ]

    # 绝对禁止
    FORBIDDEN_PATTERNS = [
        r"rm\s+-rf\s+/\s*$",          # 删根目录
        r":\(\)\s*\{\s*:\|:&\s*\};:",  # fork bomb
        r"mkfs\.",                      # 格式化磁盘
        r"dd\s+if=.+of=/dev/",         # 写设备
        r"chmod\s+-R\s+777\s+/",       # 全网开放权限
        r"curl.*169\.254\.169\.254",    # 元数据服务
        r"env\s*\|\s*.*PASSWORD",       # 暴露密钥
    ]

    def classify(self, command: str) -> CommandVerdict:
        """对命令分级，返回 verdict"""
        cmd_stripped = command.strip()

        # 先检查绝对禁止
        for pattern in self.FORBIDDEN_PATTERNS:
            if re.search(pattern, cmd_stripped):
                return CommandVerdict(
                    level=RiskLevel.FORBIDDEN,
                    reason=f"命中禁止规则：{pattern}"
                )

        # 再检查高风险
        for pattern in self.HIGH_PATTERNS:
            if re.search(pattern, cmd_stripped):
                return CommandVerdict(
                    level=RiskLevel.HIGH,
                    reason=f"高风险命令，需人工审批：{cmd_stripped[:80]}"
                )

        # 检查中等风险
        for pattern in self.MODERATE_PATTERNS:
            if re.search(pattern, cmd_stripped):
                return CommandVerdict(
                    level=RiskLevel.MODERATE,
                    reason=f"可逆修改，记录审计：{cmd_stripped[:80]}"
                )

        # 最后检查只读
        for pattern in self.SAFE_PATTERNS:
            if re.search(pattern, cmd_stripped):
                return CommandVerdict(
                    level=RiskLevel.SAFE,
                    reason=f"只读命令：{cmd_stripped[:80]}"
                )

        # 未知命令默认为高风险——宁可误拦不可放过
        return CommandVerdict(
            level=RiskLevel.HIGH,
            reason=f"未识别命令，默认高风险：{cmd_stripped[:80]}"
        )

┌────────────────────────────────────────────────────────────┐
│                    Command Proxy                            │
└─────────────────────────────┬──────────────────────────────┘
                              │
            ┌─────────────────┼─────────────────┐
            │                 │                 │
            ▼                 ▼                 ▼
      ┌──────────┐      ┌──────────┐      ┌──────────┐
      │  SAFE    │      │ MODERATE │      │   HIGH   │
      │ 直接执行  │      │ 记录+执行 │      │  暂停等待 │
      │ 无审批    │      │ 无审批    │      │  人工审批 │
      └────┬─────┘      └────┬─────┘      └────┬─────┘
           │                 │                 │
           │                 │         ┌───────┴────────┐
           │                 │         │ 审批人收到通知   │
           │                 │         │ 点"同意"或"拒绝"│
           │                 │         │ 超时 5min 自动拒│
           │                 │         └───────┬────────┘
           │                 │                 │
           ▼                 ▼                 ▼
      ┌─────────────────────────────────────────────┐
      │        命令审计日志（所有分级都记录）         │
      │  task_id | command | level | verdict | ts   │
      └─────────────────────────────────────────────┘

artifact-store/
└── tasks/
    └── task-<id>/
        ├── manifest.json           # 元数据：任务ID、开始/结束时间、使用模型
        ├── diff.patch              # git diff 完整输出
        ├── files-changed.json      # 改动文件列表 + 每文件增删行数
        ├── commands.jsonl          # 执行过的所有命令（按时间）
        ├── agent-log.txt           # Agent 对话 / 思考日志
        ├── test-report.xml         # 测试结果（JUnit 格式）
        ├── coverage.json           # 覆盖率报告（如果有）
        ├── errors.json             # 错误与异常汇总
        └── review-summary.md       # 给 reviewer 的摘要（Agent 自生成）

import json
from datetime import datetime, timezone
from pathlib import Path

class ArtifactCollector:
    """任务结束后收集所有产物"""

    def __init__(self, worktree: Path, task_id: str, store_root: Path):
        self.worktree = worktree
        self.task_id = task_id
        self.store_dir = store_root / "tasks" / f"task-{task_id}"
        self.store_dir.mkdir(parents=True, exist_ok=True)
        self.wt_manager = WorktreeManager(str(worktree.parent))

    def collect_all(self, base_branch: str, started_at: str) -> dict:
        """收集全部 artifact，返回 manifest"""
        manifest = {
            "task_id": self.task_id,
            "base_branch": base_branch,
            "started_at": started_at,
            "finished_at": datetime.now(timezone.utc).isoformat(),
            "artifacts": {},
        }

        # 1. Diff
        diff = self.wt_manager.collect_diff(self.worktree, base_branch)
        (self.store_dir / "diff.patch").write_text(diff)
        manifest["artifacts"]["diff"] = {
            "path": "diff.patch",
            "size_bytes": len(diff.encode()),
            "files_changed": diff.count("\ndiff --git"),
        }

        # 2. 改动文件列表
        files_changed = self._collect_files_changed(base_branch)
        (self.store_dir / "files-changed.json").write_text(
            json.dumps(files_changed, indent=2)
        )
        manifest["artifacts"]["files_changed"] = {
            "path": "files-changed.json",
            "count": len(files_changed),
        }

        # 3. 测试结果（如果跑了的话）
        test_report = self.worktree / "test-results.xml"
        if test_report.exists():
            (self.store_dir / "test-report.xml").write_text(
                test_report.read_text()
            )
            manifest["artifacts"]["test_report"] = {
                "path": "test-report.xml",
                "passed": self._count_test_results(test_report, "passed"),
                "failed": self._count_test_results(test_report, "failed"),
            }

        # 4. 命令执行日志
        commands_log = self._collect_commands_log()
        (self.store_dir / "commands.jsonl").write_text(commands_log)
        manifest["artifacts"]["commands"] = {
            "path": "commands.jsonl",
            "count": len(commands_log.strip().splitlines()),
        }

        # 5. 自生成 review summary
        review = self._generate_review_summary(diff, files_changed)
        (self.store_dir / "review-summary.md").write_text(review)
        manifest["artifacts"]["review_summary"] = {
            "path": "review-summary.md",
        }

        # 写入 manifest
        (self.store_dir / "manifest.json").write_text(
            json.dumps(manifest, indent=2)
        )
        return manifest

    def _collect_files_changed(self, base_branch: str) -> list[dict]:
        """解析 diff 得到每个文件 + 增删行数"""
        stat = self.wt_manager._run_git(
            "-C", str(self.worktree),
            "diff", f"--numstat", f"origin/{base_branch}...HEAD"
        )
        result = []
        for line in stat.strip().splitlines():
            if not line.strip():
                continue
            added, deleted, path = line.split("\t")
            result.append({
                "path": path,
                "added": int(added) if added.isdigit() else 0,
                "deleted": int(deleted) if deleted.isdigit() else 0,
            })
        return result

    def _count_test_results(self, report_path: Path, status: str) -> int:
        """简易 JUnit 解析"""
        content = report_path.read_text()
        return content.count(f'<failure') if status == "failed" else content.count('testsuite')

    def _collect_commands_log(self) -> str:
        """从命令代理收集本次任务执行过的所有命令"""
        log_path = Path("/var/log/agent-proxy") / f"{self.task_id}.jsonl"
        return log_path.read_text() if log_path.exists() else ""

    def _generate_review_summary(self, diff: str, files: list[dict]) -> str:
        """给 reviewer 的摘要——让 reviewer 30 秒内知道改了啥"""
        total_added = sum(f["added"] for f in files)
        total_deleted = sum(f["deleted"] for f in files)
        file_list = "\n".join(
            f"- `{f['path']}` (+{f['added']}/-{f['deleted']})" for f in files
        )
        return (
            f"## 任务 {self.task_id} 改动摘要\n\n"
            f"**统计**：{len(files)} 个文件，+{total_added}/-{total_deleted} 行\n\n"
            f"**改动文件**：\n{file_list}\n\n"
            f"**完整 diff**：见 `diff.patch`\n"
        )

# 看起来安全，实际危险
ls -la | xargs -I {} rm {}          # ls 命中 SAFE，整体却是 rm
git log --oneline; rm -rf /tmp/data  # git log 命中 SAFE，分号后是 FORBIDDEN
echo $(cat /etc/shadow)              # echo 命中 SAFE，实际在读敏感文件

# 错误做法：只匹配整条字符串
if command.startswith("ls"):
    return RiskLevel.SAFE  # 危险！ls | xargs rm 也通过了

# 正确做法：解析管道和分号，对每段分级
def classify_pipeline(pipeline: str) -> RiskLevel:
    segments = split_pipeline(pipeline)  # 按 | ; && || 拆分
    levels = [classify_single(seg) for seg in segments]
    # 返回最高风险等级
    return max(levels, key=lambda l: RISK_ORDER.index(l))

# 清理超过 7 天、且没有对应活跃任务的 worktree
0 3 * * * /opt/agent-platform/scripts/cleanup-orphan-worktrees.sh --max-age=7d