多 Agent 安全沙箱：每个 Agent 独立容器运行、Hermes 汇总结果、隔离执行

┌─────────────────────────────────────────────────────────┐
│                    安全沙箱核心目标                        │
│                                                         │
│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
│  │  执行隔离     │  │  资源限制     │  │  审计追踪      │ │
│  │              │  │              │  │               │ │
│  │ • 独立文件系统│  │ • CPU 配额    │  │ • 操作日志     │ │
│  │ • 独立网络栈  │  │ • 内存上限    │  │ • 文件变更追踪  │ │
│  │ • 独立进程空间│  │ • 磁盘配额    │  │ • 网络调用记录  │ │
│  │ • 独立环境变量│  │ • 网络带宽    │  │ • 资源使用统计  │ │
│  └──────────────┘  └──────────────┘  └───────────────┘ │
│                                                         │
│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
│  │  快速恢复     │  │  权限控制     │  │  结果聚合      │ │
│  │              │  │              │  │               │ │
│  │ • 容器快照    │  │ • 最小权限原则 │  │ • 标准化输出   │ │
│  │ • 一键重建    │  │ • 白名单网络   │  │ • 冲突检测     │ │
│  │ • 状态回滚    │  │ • 只读系统层   │  │ • 合并策略     │ │
│  └──────────────┘  └──────────────┘  └───────────────┘ │
└─────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                        Hermes 调度中心                            │
│                                                                 │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌───────────────┐  │
│  │ 任务分解器│  │ 沙箱管理器│  │ 结果聚合器│  │ 安全审计器     │  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └──────┬────────┘  │
│       │              │              │               │           │
│  ┌────┴──────────────┴──────────────┴───────────────┴───────┐  │
│  │                    Container Orchestrator                  │  │
│  │                 (Docker Compose / Kubernetes)              │  │
│  └────┬──────────────┬──────────────┬──────────────┬───────┘  │
│       │              │              │              │           │
└───────┼──────────────┼──────────────┼──────────────┼───────────┘
        ▼              ▼              ▼              ▼
┌──────────────┐┌──────────────┐┌──────────────┐┌──────────────┐
│ Agent A 沙箱  ││ Agent B 沙箱  ││ Agent C 沙箱  ││ Agent D 沙箱  │
│              ││              ││              ││              │
│ • 独立镜像    ││ • 独立镜像    ││ • 独立镜像    ││ • 独立镜像    │
│ • CPU: 2C    ││ • CPU: 1C    ││ • CPU: 4C    ││ • CPU: 2C    │
│ • MEM: 4GB   ││ • MEM: 2GB   ││ • MEM: 8GB   ││ • MEM: 4GB   │
│ • NET: 受限  ││ • NET: 内部  ││ • NET: 受限  ││ • NET: 受限  │
│ • FS: 隔离   ││ • FS: 隔离   ││ • FS: 隔离   ││ • FS: 隔离   │
│ • Timeout:5m ││ • Timeout:3m ││ • Timeout:10m││ • Timeout:5m │
└──────────────┘└──────────────┘└──────────────┘└──────────────┘

┌──────────────────────────────────────────────────────┐
│                    L4: 网络隔离层                       │
│  • 独立的网络命名空间                                   │
│  • 仅允许白名单域名访问                                 │
│  • 出口流量审计                                        │
├──────────────────────────────────────────────────────┤
│                    L3: 资源限制层                       │
│  • cgroups 资源配额                                    │
│  • OOM Killer 保护                                     │
│  • 磁盘 I/O 限制                                       │
├──────────────────────────────────────────────────────┤
│                    L2: 文件系统隔离层                    │
│  • OverlayFS 读写层                                    │
│  • 临时工作目录                                        │
│  • 执行后自动清理                                      │
├──────────────────────────────────────────────────────┤
│                    L1: 进程隔离层                       │
│  • PID 命名空间隔离                                    │
│  • 非 root 用户运行                                    │
│  • seccomp 系统调用过滤                                │
└──────────────────────────────────────────────────────┘

# sandbox/Dockerfile.agent-base
FROM python:3.11-slim

# 安全：非 root 用户
RUN groupadd -g 1001 agentuser && \
    useradd -u 1001 -g agentuser -m -s /bin/bash agentuser

# 安装必要工具
RUN apt-get update && apt-get install -y --no-install-recommends \
    git curl jq tini \
    && rm -rf /var/lib/apt/lists/*

# 设置工作目录
WORKDIR /workspace

# 安全：只读根文件系统 + 可写 /workspace
USER agentuser

ENTRYPOINT ["tini", "--"]
CMD ["python", "-u", "/agent/run.py"]

# sandbox/manager.py
import docker
import json
import time
import uuid
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class AgentRole(str, Enum):
    RESEARCHER = "researcher"
    CODER = "coder"
    REVIEWER = "reviewer"
    TESTER = "tester"

@dataclass
class SandboxConfig:
    """沙箱配置"""
    agent_id: str = field(default_factory=lambda: f"agent-{uuid.uuid4().hex[:8]}")
    role: AgentRole = AgentRole.CODER
    image: str = "agent-sandbox:latest"
    cpu_limit: float = 2.0       # CPU 核心数
    mem_limit: str = "4g"        # 内存限制
    disk_limit: str = "10g"      # 磁盘限制
    network_mode: str = "sandbox-net"
    timeout: int = 300           # 执行超时(秒)
    allowed_hosts: list = field(default_factory=list)
    environment: dict = field(default_factory=dict)
    volumes: dict = field(default_factory=dict)

@dataclass
class SandboxResult:
    """沙箱执行结果"""
    agent_id: str
    status: str  # success, failed, timeout, error
    stdout: str
    stderr: str
    exit_code: int
    execution_time: float
    resources_used: dict
    output_artifacts: list = field(default_factory=list)

class SandboxManager:
    """沙箱管理器"""

    def __init__(self, docker_client: Optional[docker.DockerClient] = None):
        self.client = docker_client or docker.from_env()
        self.active_sandboxes: dict[str, 'SandboxInstance'] = {}
        self._setup_network()

    def _setup_network(self):
        """创建隔离网络"""
        try:
            self.client.networks.get("sandbox-net")
        except docker.errors.NotFound:
            self.client.networks.create(
                "sandbox-net",
                driver="bridge",
                internal=True,  # 无外部网络访问
                options={"com.docker.network.bridge.enable_icc": "false"}
            )
            print("[沙箱网络] 已创建隔离网络 sandbox-net")

    def create_sandbox(self, config: SandboxConfig) -> 'SandboxInstance':
        """创建并启动沙箱"""
        instance = SandboxInstance(self.client, config)
        instance.start()
        self.active_sandboxes[config.agent_id] = instance
        print(f"[沙箱] 已启动 {config.agent_id} ({config.role.value})")
        return instance

    def execute_in_sandbox(
        self,
        config: SandboxConfig,
        task_code: str,
        task_input: dict
    ) -> SandboxResult:
        """在沙箱中执行任务"""
        instance = self.create_sandbox(config)

        try:
            # 写入任务代码
            instance.write_file("/agent/run.py", task_code)
            instance.write_file(
                "/agent/input.json",
                json.dumps(task_input, ensure_ascii=False)
            )

            # 执行任务
            result = instance.execute(timeout=config.timeout)
            return result

        except Exception as e:
            return SandboxResult(
                agent_id=config.agent_id,
                status="error",
                stdout="",
                stderr=str(e),
                exit_code=-1,
                execution_time=0,
                resources_used={},
            )

        finally:
            instance.cleanup()
            del self.active_sandboxes[config.agent_id]

    def aggregate_results(self, results: list[SandboxResult]) -> dict:
        """汇总所有沙箱结果"""
        summary = {
            "total": len(results),
            "success": sum(1 for r in results if r.status == "success"),
            "failed": sum(1 for r in results if r.status == "failed"),
            "timeout": sum(1 for r in results if r.status == "timeout"),
            "error": sum(1 for r in results if r.status == "error"),
            "total_time": sum(r.execution_time for r in results),
            "results": [],
        }

        for r in results:
            summary["results"].append({
                "agent_id": r.agent_id,
                "status": r.status,
                "execution_time": f"{r.execution_time:.1f}s",
                "output_preview": r.stdout[:200] if r.stdout else "",
                "error_preview": r.stderr[:200] if r.stderr else "",
            })

        return summary


class SandboxInstance:
    """沙箱实例"""

    def __init__(self, client: docker.DockerClient, config: SandboxConfig):
        self.client = client
        self.config = config
        self.container = None

    def start(self):
        """启动容器"""
        self.container = self.client.containers.run(
            image=self.config.image,
            name=self.config.agent_id,
            detach=True,
            cpu_quota=int(self.config.cpu_limit * 100000),
            cpu_period=100000,
            mem_limit=self.config.mem_limit,
            network=self.config.network_mode,
            environment=self.config.environment,
            security_opt=["no-new-privileges:true"],
            read_only=True,  # 根文件系统只读
            tmpfs={"/tmp": "exec,size=1g"},  # 临时目录可写
            cap_drop=["ALL"],  # 移除所有能力
            cap_add=["NET_BIND_SERVICE"],  # 仅添加必要能力
            ulimits=[
                docker.types.Ulimit(name="nofile", soft=1024, hard=1024),
                docker.types.Ulimit(name="nproc", soft=256, hard=256),
            ],
        )

    def write_file(self, path: str, content: str):
        """向沙箱写入文件"""
        import tarfile, io
        tarstream = io.BytesIO()
        with tarfile.open(fileobj=tarstream, mode='w') as tar:
            data = content.encode('utf-8')
            tarinfo = tarfile.TarInfo(name=path.lstrip('/'))
            tarinfo.size = len(data)
            tar.addfile(tarinfo, io.BytesIO(data))
        tarstream.seek(0)
        self.container.put_archive('/', tarstream)

    def execute(self, timeout: int = 300) -> SandboxResult:
        """在沙箱中执行任务"""
        import time
        start = time.time()

        try:
            exec_result = self.container.exec_run(
                cmd="python -u /agent/run.py",
                stdout=True,
                stderr=True,
                workdir="/workspace",
            )

            elapsed = time.time() - start
            return SandboxResult(
                agent_id=self.config.agent_id,
                status="success" if exec_result.exit_code == 0 else "failed",
                stdout=exec_result.output.decode('utf-8', errors='replace'),
                stderr="",
                exit_code=exec_result.exit_code,
                execution_time=elapsed,
                resources_used=self._get_resource_stats(),
            )

        except docker.errors.APIError as e:
            elapsed = time.time() - start
            return SandboxResult(
                agent_id=self.config.agent_id,
                status="timeout" if "timeout" in str(e).lower() else "error",
                stdout="",
                stderr=str(e),
                exit_code=-1,
                execution_time=elapsed,
                resources_used={},
            )

    def _get_resource_stats(self) -> dict:
        """获取资源使用统计"""
        try:
            stats = self.container.stats(stream=False)
            return {
                "cpu_percent": stats["cpu_stats"]["cpu_usage"]["total_usage"],
                "memory_usage": stats["memory_stats"]["usage"],
                "memory_limit": stats["memory_stats"]["limit"],
            }
        except Exception:
            return {}

    def cleanup(self):
        """清理沙箱"""
        if self.container:
            try:
                self.container.stop(timeout=5)
                self.container.remove(force=True)
                print(f"[沙箱] 已清理 {self.config.agent_id}")
            except Exception as e:
                print(f"[沙箱] 清理失败 {self.config.agent_id}: {e}")

# sandbox/k8s/agent-sandbox.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: agent-sandbox

---
apiVersion: v1
kind: LimitRange
metadata:
  name: agent-limits
  namespace: agent-sandbox
spec:
  limits:
    - default:
        cpu: "2"
        memory: "4Gi"
        ephemeral-storage: "10Gi"
      defaultRequest:
        cpu: "500m"
        memory: "1Gi"
        ephemeral-storage: "2Gi"
      type: Container

---
apiVersion: v1
kind: NetworkPolicy
metadata:
  name: agent-network-policy
  namespace: agent-sandbox
spec:
  podSelector:
    matchLabels:
      app: agent-sandbox
  policyTypes:
    - Egress
    - Ingress
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              name: internal-services
      ports:
        - protocol: TCP
          port: 443
    - to:
        - ipBlock:
            cidr: 10.0.0.0/8
      ports:
        - protocol: TCP
          port: 53
        - protocol: UDP
          port: 53

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-coder
  namespace: agent-sandbox
  labels:
    app: agent-sandbox
    role: coder
spec:
  replicas: 1
  selector:
    matchLabels:
      app: agent-sandbox
      role: coder
  template:
    metadata:
      labels:
        app: agent-sandbox
        role: coder
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1001
        fsGroup: 1001
      containers:
        - name: agent
          image: agent-sandbox:latest
          resources:
            requests:
              cpu: "500m"
              memory: "1Gi"
            limits:
              cpu: "2"
              memory: "4Gi"
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
          volumeMounts:
            - name: workspace
              mountPath: /workspace
            - name: tmp
              mountPath: /tmp
      volumes:
        - name: workspace
          emptyDir:
            sizeLimit: 10Gi
        - name: tmp
          emptyDir:
            medium: Memory
            sizeLimit: 1Gi

# hermes/sandbox_scheduler.py
import asyncio
from typing import Optional
from dataclasses import dataclass

@dataclass
class TaskSpec:
    """任务规格"""
    task_id: str
    role: str
    code: str
    input_data: dict
    priority: int = 0
    timeout: int = 300
    cpu: float = 2.0
    memory: str = "4g"

class SandboxScheduler:
    """Hermes 沙箱调度器"""

    def __init__(self):
        self.manager = SandboxManager()
        self.task_queue: asyncio.Queue = asyncio.Queue()
        self.results: list[SandboxResult] = []

    async def submit_tasks(self, tasks: list[TaskSpec]):
        """提交任务到沙箱队列"""
        for task in tasks:
            await self.task_queue.put(task)
            print(f"[调度器] 任务入队: {task.task_id} ({task.role})")

    async def execute_all(self, max_parallel: int = 4) -> dict:
        """并行执行所有任务"""
        semaphore = asyncio.Semaphore(max_parallel)
        tasks = []

        while not self.task_queue.empty():
            task = await self.task_queue.get()
            tasks.append(self._run_in_sandbox(task, semaphore))

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 处理结果
        sandbox_results = []
        for r in results:
            if isinstance(r, SandboxResult):
                sandbox_results.append(r)
            else:
                print(f"[调度器] 任务异常: {r}")

        # 汇总
        summary = self.manager.aggregate_results(sandbox_results)
        return summary

    async def _run_in_sandbox(
        self, task: TaskSpec, semaphore: asyncio.Semaphore
    ) -> SandboxResult:
        """在沙箱中运行单个任务"""
        async with semaphore:
            config = SandboxConfig(
                agent_id=task.task_id,
                role=AgentRole(task.role),
                cpu_limit=task.cpu,
                mem_limit=task.memory,
                timeout=task.timeout,
            )
            return await asyncio.to_thread(
                self.manager.execute_in_sandbox,
                config,
                task.code,
                task.input_data,
            )

# hermes/examples/parallel_code_review.py

async def run_parallel_code_review():
    """并行代码审查示例"""
    scheduler = SandboxScheduler()

    # 定义 4 个审查 Agent 任务
    tasks = [
        TaskSpec(
            task_id="reviewer-style",
            role="reviewer",
            code="""
import json, sys

with open('/agent/input.json') as f:
    code = json.load(f)['code']

# 检查代码风格
issues = []
lines = code.split('\\n')
for i, line in enumerate(lines, 1):
    if len(line) > 120:
        issues.append(f"Line {i}: line too long ({len(line)} > 120)")
    if '\\t' in line:
        issues.append(f"Line {i}: tab character found")

result = {
    "agent": "style-reviewer",
    "issues": issues,
    "total_lines": len(lines),
    "style_score": max(0, 100 - len(issues) * 5)
}
print(json.dumps(result))
""",
            input_data={"code": open("main.py").read()},
            timeout=60,
            cpu=1.0,
            memory="2g",
        ),
        TaskSpec(
            task_id="reviewer-security",
            role="reviewer",
            code="""
import json, re, sys

with open('/agent/input.json') as f:
    code = json.load(f)['code']

# 安全检查
patterns = {
    "hardcoded_secret": r'(?:password|secret|api_key|token)\\s*=\\s*["\\'][^"\\']+["\\']',
    "sql_injection": r'execute\\s*\\(.*%.*%',
    "eval_usage": r'\\beval\\s*\\(',
    "shell_injection": r'os\\.system\\s*\\(',
}

findings = []
for name, pattern in patterns.items():
    matches = re.finditer(pattern, code)
    for m in matches:
        findings.append({"type": name, "line": code[:m.start()].count('\\n') + 1})

result = {
    "agent": "security-reviewer",
    "findings": findings,
    "risk_level": "HIGH" if any(f["type"] == "hardcoded_secret" for f in findings) else "LOW"
}
print(json.dumps(result))
""",
            input_data={"code": open("main.py").read()},
            timeout=60,
            cpu=1.0,
            memory="2g",
        ),
        TaskSpec(
            task_id="reviewer-complexity",
            role="reviewer",
            code="""
import json, ast, sys

with open('/agent/input.json') as f:
    code = json.load(f)['code']

# 计算圈复杂度
tree = ast.parse(code)
complexity = 1
for node in ast.walk(tree):
    if isinstance(node, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
        complexity += 1
    elif isinstance(node, ast.BoolOp):
        complexity += len(node.values) - 1

result = {
    "agent": "complexity-reviewer",
    "cyclomatic_complexity": complexity,
    "rating": "A" if complexity <= 5 else "B" if complexity <= 10 else "C" if complexity <= 20 else "D"
}
print(json.dumps(result))
""",
            input_data={"code": open("main.py").read()},
            timeout=60,
            cpu=1.0,
            memory="2g",
        ),
        TaskSpec(
            task_id="reviewer-docs",
            role="reviewer",
            code="""
import json, ast, sys

with open('/agent/input.json') as f:
    code = json.load(f)['code']

# 文档检查
tree = ast.parse(code)
functions = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)]
classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]

missing_docstrings = []
for func in functions:
    if not (func.body and isinstance(func.body[0], ast.Expr) and isinstance(func.body[0].value, ast.Constant)):
        missing_docstrings.append(func.name)

for cls in classes:
    if not (cls.body and isinstance(cls.body[0], ast.Expr) and isinstance(cls.body[0].value, ast.Constant)):
        missing_docstrings.append(f"class {cls.name}")

result = {
    "agent": "docs-reviewer",
    "total_functions": len(functions),
    "total_classes": len(classes),
    "missing_docstrings": missing_docstrings,
    "doc_coverage": round((1 - len(missing_docstrings) / max(1, len(functions) + len(classes))) * 100, 1)
}
print(json.dumps(result))
""",
            input_data={"code": open("main.py").read()},
            timeout=60,
            cpu=1.0,
            memory="2g",
        ),
    ]

    # 提交并执行
    await scheduler.submit_tasks(tasks)
    summary = await scheduler.execute_all(max_parallel=4)

    print("\n" + "=" * 60)
    print("📊 沙箱执行汇总")
    print("=" * 60)
    print(f"总任务数: {summary['total']}")
    print(f"成功: {summary['success']}")
    print(f"失败: {summary['failed']}")
    print(f"超时: {summary['timeout']}")
    print(f"错误: {summary['error']}")
    print(f"总耗时: {summary['total_time']:.1f}s")
    print()

    for r in summary['results']:
        status_icon = "✅" if r['status'] == 'success' else "❌"
        print(f"  {status_icon} {r['agent_id']} | {r['status']} | {r['execution_time']}")
        if r['output_preview']:
            print(f"     输出: {r['output_preview'][:100]}...")


# 运行
asyncio.run(run_parallel_code_review())

{
  "defaultAction": "SCMP_ACT_ERRNO",
  "archMap": [
    {
      "architecture": "SCMP_ARCH_X86_64",
      "syscalls": [
        { "names": ["read", "write", "open", "close", "stat", "fstat",
                     "mmap", "mprotect", "brk", "exit", "exit_group",
                     "clone", "execve", "fork", "wait4", "getpid",
                     "getuid", "getgid", "geteuid", "getegid",
                     "access", "pipe", "select", "socket", "connect",
                     "sendto", "recvfrom", "gettimeofday", "clock_gettime",
                     "nanosleep", "getdents", "getdents64", "lseek",
                     "readlink", "getcwd", "uname", "fcntl", "flock",
                     "fsync", "fdatasync", "truncate", "ftruncate",
                     "getxattr", "listxattr", "removexattr", "setxattr",
                     "prctl", "arch_prctl", "set_tid_address",
                     "set_robust_list", "get_robust_list",
                     "futex", "epoll_create", "epoll_ctl", "epoll_wait",
                     "eventfd", "timerfd_create", "timerfd_settime",
                     "signalfd", "epoll_create1"],
          "action": "SCMP_ACT_ALLOW"
        }
      ]
    }
  ]
}

# sandbox/network_policy.py

ALLOWED_DOMAINS = [
    "api.openai.com",
    "api.anthropic.com",
    "registry.npmjs.org",
    "pypi.org",
    "files.pythonhosted.org",
]

def generate_dns_whitelist(domains: list[str]) -> str:
    """生成 DNS 白名单 iptables 规则"""
    rules = []
    rules.append("# 默认拒绝所有出站")
    rules.append("iptables -P OUTPUT DROP")

    # 允许 DNS 查询
    rules.append("iptables -A OUTPUT -p udp --dport 53 -j ACCEPT")
    rules.append("iptables -A OUTPUT -p tcp --dport 53 -j ACCEPT")

    # 允许白名单域名
    for domain in domains:
        rules.append(f"iptables -A OUTPUT -d {domain} -p tcp --dport 443 -j ACCEPT")

    # 允许回环
    rules.append("iptables -A OUTPUT -o lo -j ACCEPT")

    return "\n".join(rules)

┌──────────────────────────────────────────────────────────────────┐
│                    Hermes 沙箱监控面板                              │
│                                                                  │
│  运行中: ████████░░░░░░░░░░░░░░░░░░░░░  4/10 沙箱              │
│  CPU:    ██████░░░░░░░░░░░░░░░░░░░░░░░  32%                     │
│  MEM:    ██████████░░░░░░░░░░░░░░░░░░░  48%                     │
│                                                                  │
│  ┌──────────────┬──────────┬─────────┬────────┬───────────────┐  │
│  │ Agent ID     │ 状态     │ CPU     │ 内存   │ 运行时间       │  │
│  ├──────────────┼──────────┼─────────┼────────┼───────────────┤  │
│  │ reviewer-01  │ 🟢 运行  │ 1.2C    │ 1.2GB  │ 00:02:34      │  │
│  │ reviewer-02  │ 🟢 运行  │ 0.8C    │ 0.9GB  │ 00:02:31      │  │
│  │ tester-01    │ 🟢 运行  │ 1.8C    │ 2.1GB  │ 00:01:45      │  │
│  │ coder-01     │ 🟡 等待  │ 0.0C    │ 0.0GB  │ 排队中        │  │
│  └──────────────┴──────────┴─────────┴────────┴───────────────┘  │
│                                                                  │
│  最近事件:                                                        │
│  [18:34:12] reviewer-01 完成: style check passed (2.3s)          │
│  [18:34:10] tester-01  启动 (timeout=300s)                       │
│  [18:34:05] reviewer-02 发现 1 个安全问题                        │
│  [18:34:01] coder-01   排队中 (等待可用沙箱)                       │
└──────────────────────────────────────────────────────────────────┘

# sandbox/auditor.py
import logging
from datetime import datetime

class SandboxAuditor:
    """沙箱审计器"""

    def __init__(self, log_path: str = "/var/log/hermes/sandbox-audit.log"):
        self.logger = logging.getLogger("sandbox-audit")
        self.logger.setLevel(logging.INFO)
        handler = logging.FileHandler(log_path)
        handler.setFormatter(logging.Formatter(
            '%(asctime)s | %(levelname)s | %(message)s'
        ))
        self.logger.addHandler(handler)

    def log_sandbox_event(self, event_type: str, agent_id: str, details: dict):
        """记录沙箱事件"""
        self.logger.info(
            f"EVENT={event_type} | AGENT={agent_id} | DETAILS={details}"
        )

    def log_resource_usage(self, agent_id: str, stats: dict):
        """记录资源使用"""
        self.logger.info(
            f"RESOURCE | AGENT={agent_id} | "
            f"CPU={stats.get('cpu_percent', 0)} | "
            f"MEM={stats.get('memory_usage', 0)} | "
            f"TIME={datetime.utcnow().isoformat()}"
        )

    def log_security_violation(self, agent_id: str, violation: str):
        """记录安全违规"""
        self.logger.warning(
            f"SECURITY_VIOLATION | AGENT={agent_id} | {violation}"
        )