@dataclassclass Tool:    name: str # 工具名    description: str # 说明书    schema: Dict[str, Any] # 参数形状    handler: Callable # 干活    risk: str = ”read” # read | write

for step in range(1, max_steps + 1):    reply = model.complete(ctx.render())    tool = tools_by_name.get(reply[”tool”])    if tool is None: # 交卷        return reply[”content”]    try:        policy.check(tool, reply[”args”])        result = tool.handler(**reply[”args”])    except (PermissionError, TypeError) as e:        result = ”拒绝/参数错: %s” % e    ctx.add(”tool”, result) # 观察回灌

def check(tool, args):    resolve(args[”path”]) # 先查越界    if tool.risk == ”read”:        return ”allow”    return ”allow(after-approval)”

[step 1] 工具 write_file         args={”path”: ”hello.txt”, ...}         策略: allow(after-approval) write         观察: OK 写入 27 字节 -> hello.txt[step 2] 工具 read_file         args={”path”: ”hello.txt”}         观察: （共 2 行，第 1-2 行）[step 3] 工具 write_file         args={”path”: ”../escape.txt”}         拦截: 路径越界: ../escape.txt         观察: PERMISSION_DENIED[step 4] 工具 list_dir         观察: hello.txt[step 5] 工具 write_file         args={”path”: ”note.md”}         参数错: missing argument         观察: BAD_ARGS------------------------------------最终回答: hello.txt 已写入并读回；          越界与缺参写入都被拦下，没有落盘。统计: 5 步 / 5 次工具调用 / 5 次策略判定

#!/usr/bin/env python3# -*- coding: utf-8 -*-\"\"\"mini_harness.py —— 200 行以内的最小 agent harness（离线可跑）
只保留内核五件套：  1. 模型适配层   （这里用脚本化假模型，换成真 provider 即可）  2. 工具注册表   （名字 / 描述 / 参数 schema / 处理器 / 风险等级）  3. 主循环       （模型 -> 工具 -> 观察 -> 再模型，直到不再调用工具）  4. 上下文预算   （超预算就裁剪，并显式告知模型被裁掉了多少）  5. 策略与观测   （路径越界拦截、写操作审批、JSONL 全量留痕）\"\"\"import jsonimport osimport timefrom dataclasses import dataclassfrom typing import Any, Callable, Dict, List
ROOT = os.path.dirname(os.path.abspath(__file__))WORKSPACE = os.path.join(ROOT, \"sandbox\")LOG_PATH = os.path.join(ROOT, \"transcript.jsonl\")

def resolve(path: str) -> str:    \"\"\"把相对路径锁死在 workspace 里，越界直接抛错。\"\"\"    root = os.path.realpath(WORKSPACE)    full = os.path.realpath(os.path.join(root, path))    if not (full == root or full.startswith(root + os.sep)):        raise PermissionError(\"路径越界: \" + path)    return full

@dataclassclass Tool:    name: str    description: str    schema: Dict[str, Any]    handler: Callable[..., str]    risk: str = \"read\"          # read | write

def read_file(path: str, offset: int = 1, limit: int = 100) -> str:    full = resolve(path)    with open(full, encoding=\"utf-8\") as f:        lines = f.read().splitlines()    chunk = lines[offset - 1: offset - 1 + limit]    head = \"（共 %d 行，本次返回第 %d-%d 行）\
\" % (        len(lines), offset, offset + len(chunk) - 1)    return head + \"\
\".join(chunk)

def write_file(path: str, content: str) -> str:    full = resolve(path)    os.makedirs(os.path.dirname(full) or WORKSPACE, exist_ok=True)    data = content.encode(\"utf-8\")    with open(full, \"wb\") as f:        f.write(data)    return \"OK 写入 %d 字节 -> %s\" % (len(data), os.path.relpath(full, WORKSPACE))

def list_dir(path: str = \".\") -> str:    full = resolve(path)    names = sorted(os.listdir(full))    return \"\
\".join(names) if names else \"(空目录)\"

TOOLS: List[Tool] = [    Tool(\"read_file\", \"读工作区内的文本文件；返回带行号范围提示的内容。\",         {\"path\": \"文件相对路径\", \"offset\": \"起始行(可选)\", \"limit\": \"最多行数(可选)\"},         read_file, \"read\"),    Tool(\"write_file\", \"在工作区内写入/覆盖文本文件，会创建父目录。\",         {\"path\": \"文件相对路径\", \"content\": \"完整文本内容\"},         write_file, \"write\"),    Tool(\"list_dir\", \"列出工作区某个目录下的条目。\",         {\"path\": \"目录相对路径(可选，默认.)\"},         list_dir, \"read\"),]

class Policy:    \"\"\"最小权限模型：读免审批，写要审批，越界一律拒绝（不消耗审批额度）。\"\"\"
    def __init__(self, auto_approve: bool = True):        self.auto_approve = auto_approve        self.decisions: List[Dict[str, str]] = []
    def check(self, tool: Tool, args: Dict[str, Any]) -> str:        try:            if \"path\" in args:                resolve(args[\"path\"])          # 先做越界检查        except PermissionError as exc:            self.decisions.append({\"tool\": tool.name, \"verdict\": \"deny\"})            raise        if tool.risk == \"read\":            verdict = \"allow\"        else:            verdict = \"allow\" if self.auto_approve else \"deny\"            if verdict == \"allow\":                verdict = \"allow(after-approval)\"        self.decisions.append({\"tool\": tool.name, \"verdict\": verdict})        return verdict

class Context:    \"\"\"上下文预算：保留 system + 首个任务 + 最近 N 条，其余折叠成一行提示。\"\"\"
    def __init__(self, system: str, keep_recent: int = 6):        self.system = system        self.keep_recent = keep_recent        self.messages: List[Dict[str, str]] = []        self.trimmed = 0
    def add(self, role: str, content: str) -> None:        self.messages.append({\"role\": role, \"content\": content})
    def render(self) -> List[Dict[str, str]]:        if len(self.messages) <= self.keep_recent:            return [{\"role\": \"system\", \"content\": self.system}] + self.messages        head, recent = self.messages[:1], self.messages[-(self.keep_recent - 1):]        self.trimmed = len(self.messages) - len(head) - len(recent)        note = {\"role\": \"system\",                \"content\": \"[上下文压缩] 已省略 %d 条较早消息\" % self.trimmed}        return [{\"role\": \"system\", \"content\": self.system}] + head + [note] + recent

class ScriptedModel:    \"\"\"假模型：按剧本走五个回合，最后交卷。
    剧本里故意埋了两个坑：越界写、漏参数。用来观察 harness    怎么把「失败」变成模型能读懂的观察，而不是直接崩掉。    \"\"\"
    PLAN = {        1: (\"write_file\", {\"path\": \"hello.txt\",                           \"content\": \"hello, harness\
second line\
\"}),        2: (\"read_file\", {\"path\": \"hello.txt\"}),        3: (\"write_file\", {\"path\": \"../escape.txt\", \"content\": \"should be blocked\"}),        4: (\"list_dir\", {\"path\": \".\"}),        5: (\"write_file\", {\"path\": \"note.md\"}),      # 故意漏掉 content    }
    def __init__(self):        self.turn = 0
    def complete(self, messages):        self.turn += 1        if self.turn in self.PLAN:            name, args = self.PLAN[self.turn]            return {\"tool\": name, \"args\": dict(args)}        return {\"tool\": None, \"args\": {}, \"content\":                \"hello.txt 已写入并读回校验；\"                \"越界写入与缺参写入都被拦下，没有落盘。\"}

class Harness:    def __init__(self, model, tools, policy, ctx, max_steps=8):        self.model, self.tools, self.policy, self.ctx = model, tools, policy, ctx        self.max_steps = max_steps        self.tools_by_name = {t.name: t for t in tools}        self.events: List[Dict[str, Any]] = []
    def log(self, **event) -> None:        event[\"ts\"] = round(time.time(), 3)        self.events.append(event)        with open(LOG_PATH, \"a\", encoding=\"utf-8\") as f:            f.write(json.dumps(event, ensure_ascii=False) + \"\
\")
    def run(self, task: str) -> str:        self.ctx.add(\"user\", task)        for step in range(1, self.max_steps + 1):            reply = self.model.complete(self.ctx.render())            tool = self.tools_by_name.get(reply.get(\"tool\") or \"\")            if tool is None:                return reply.get(\"content\", \"(无输出)\")            args = reply.get(\"args\", {})            print(\"[step %d] 模型选择工具 %s  args=%s\"                  % (step, tool.name, json.dumps(args, ensure_ascii=False)))            self.log(step=step, kind=\"tool_call\", tool=tool.name, args=args)            start = time.time()            try:                verdict = self.policy.check(tool, args)                print(\"          策略: %s (risk=%s)\" % (verdict, tool.risk))                result = tool.handler(**args)            except PermissionError as exc:                result = \"PERMISSION_DENIED: %s\" % exc                print(\"          拦截: %s\" % exc)            except TypeError as exc:                result = \"BAD_ARGS: %s\" % exc                print(\"          参数错: %s\" % exc)            cost = round((time.time() - start) * 1000, 2)            print(\"          观察: %s  (%sms)\"                  % (result.splitlines()[0][:60], cost))            self.ctx.add(\"tool\", result)            self.log(step=step, kind=\"tool_result\", tool=tool.name,                     result=result, ms=cost)        return \"(达到最大步数 %d，未收敛)\" % self.max_steps

def main() -> None:    os.makedirs(WORKSPACE, exist_ok=True)    if os.path.exists(LOG_PATH):        os.remove(LOG_PATH)    ctx = Context(system=\"你是 mini harness 里的执行器：先做事，再汇报，不许编造。\",                  keep_recent=6)    harness = Harness(ScriptedModel(), TOOLS, Policy(auto_approve=True), ctx)    print(\"任务: 写 hello.txt 并读回校验\
\" + \"-\" * 56)    answer = harness.run(\"在工作区写 hello.txt 并读回校验。\")    print(\"-\" * 56)    print(\"最终回答:\", answer)    print(\"统计: %d 步 / %d 次工具调用 / %d 次策略判定 / 上下文裁剪 %d 条\"          % (len([e for e in harness.events if e[\"kind\"] == \"tool_result\"]),             len([e for e in harness.events if e[\"kind\"] == \"tool_call\"]),             len(harness.policy.decisions), ctx.trimmed))    print(\"留痕:\", os.path.relpath(LOG_PATH, ROOT))

if __name__ == \"__main__\":    main()