(无标题)

             ┌──────────────┐
   ┌────────►│    Reason    │  模型推理（一次 LLM 调用）
   │         │  (LLM call)  │
   │         └──────┬───────┘
   │                │
   │                ▼
   │          ┌──────────┐   没有工具调用   ┌──────────┐
   │          │  Tools?  ├───────────────►│  Output  │  返回最终文本，结束
   │          └────┬─────┘                └──────────┘
   │               │ 有工具调用
   │               ▼
   │          ┌──────────┐
   │          │ Execute  │  执行工具（可并行）
   │          │  tools   │
   │          └────┬─────┘
   │               │
   │               ▼
   │          ┌──────────┐
   └──────────┤ Observe  │  把工具结果喂回去，进入下一轮
    (loop)    │ results  │
              └──────────┘

def run_loop(history: list, toolset: list, turn_budget: int = 20) -> str:
    """一轮轮推进，直到模型不再要工具、只给文本。"""
    for _ in range(turn_budget):
        # ① 思考：一次模型调用
        reply = model.invoke(context=history, toolset=toolset)
        history.append(reply.message)          # 先把它这一轮的发言记下来

        # ② 出口：不再请求工具，说明它认为可以作答了
        requests = reply.message.tool_calls
        if not requests:
            return reply.message.content

        # ③ 执行并观察：结果逐条写回，失败也照样写回
        for req in requests:
            outcome = invoke_tool(req)
            history.append(as_tool_message(req, outcome))
        # 回到 ①——模型带着新结果重新判断
 
    raise TurnBudgetExhausted(f"{turn_budget} 轮内未收敛")

STUCK_THRESHOLD = 3

def is_spinning(history: list) -> bool:
    """连续多次发出完全相同的调用，视为原地打转。"""
    fingerprints = []
    for msg in reversed(history):
        calls = getattr(msg, "tool_calls", None)
        if not calls:
            continue
        for c in calls:
            fingerprints.append((c.function.name, c.function.arguments))
        if len(fingerprints) >= STUCK_THRESHOLD:
            break

    latest = fingerprints[:STUCK_THRESHOLD]
    return len(latest) == STUCK_THRESHOLD and len(set(latest)) == 1

┌────────────────────────────────┐
│         System Prompt          │  ← Agent 的身份定义
├────────────────────────────────┤
│        Tool Definitions        │  ← 能做什么（JSON Schema）
├────────────────────────────────┤
│        Tool Execution          │  ← Tool 实际执行逻辑
├────────────────────────────────┤
│          Tool Loop             │  ← 循环：思考 → 执行 → 观察
└────────────────────────────────┘

#!/usr/bin/env python3
"""便签助手：一个最小 Agent Harness。运行：python note_agent.py"""
import json
from openai import OpenAI

llm = OpenAI()
MODEL_NAME = "gpt-4o-mini"       # 便宜够用，学习阶段首选
TURN_LIMIT = 15

NOTES: dict[str, str] = {}       # 用内存字典当"存储层"，聚焦讲循环本身

# --- ① 身份 ---
PERSONA = (
    "你是一个便签管家。可以帮用户保存便签、按关键词检索。"
    "涉及便签操作时，务必调用提供的工具，不要凭空编造内容。"
)

# --- ② 工具 Schema：模型唯一能看到的接口 ---
TOOL_SPECS = [
    {"type": "function", "function": {
        "name": "save_note",
        "description": "保存一条便签，返回它的编号",
        "parameters": {"type": "object", "properties": {
            "title": {"type": "string", "description": "便签标题"},
            "body":  {"type": "string", "description": "便签正文"}},
            "required": ["title", "body"]}}},
    {"type": "function", "function": {
        "name": "find_notes",
        "description": "按关键词检索便签，返回命中的标题列表",
        "parameters": {"type": "object", "properties": {
            "keyword": {"type": "string"}},
            "required": ["keyword"]}}},
]

# --- ③ 工具实现：模型看不到这段 ---
def call_tool(name: str, kwargs: dict) -> str:
    try:
        if name == "save_note":
            note_id = f"n{len(NOTES) + 1}"
            NOTES[note_id] = f"{kwargs['title']}\n{kwargs['body']}"
            return f"已保存，编号 {note_id}"
        if name == "find_notes":
            kw = kwargs["keyword"]
            hits = [f"{nid}: {txt.splitlines()[0]}"
                    for nid, txt in NOTES.items() if kw in txt]
            return "\n".join(hits) if hits else f"没有匹配 “{kw}” 的便签"
        return f"错误：未知工具 {name}"
    except Exception as exc:
        return f"错误：{exc}"           # 出错也返回字符串，交给模型判断

# --- ④ 循环：思考 → 执行 → 观察 ---
def chat(user_text: str) -> str:
    history = [{"role": "system", "content": PERSONA},
               {"role": "user", "content": user_text}]
    for _ in range(TURN_LIMIT):
        reply = llm.chat.completions.create(
            model=MODEL_NAME, messages=history, tools=TOOL_SPECS)
        turn = reply.choices[0].message
        history.append(turn)                       # 关键：先把 assistant 回合入历史
        if not turn.tool_calls:                    # 不再调用工具 → 收尾
            return turn.content
        for tc in turn.tool_calls:                 # 执行本回合请求的所有工具
            out = call_tool(tc.function.name, json.loads(tc.function.arguments))
            history.append({"role": "tool",
                            "tool_call_id": tc.id, "content": out})
    return "已达最大回合数，提前退出。"

if __name__ == "__main__":
    print("便签管家已就绪（输入 q 退出）")
    while (line := input("\n> ").strip()) not in ("q", "quit"):
        print(chat(line))

# 加进 TOOL_SPECS 列表：
{"type": "function", "function": {
    "name": "delete_note",
    "description": "按编号删除一条便签",
    "parameters": {"type": "object",
        "properties": {"note_id": {"type": "string"}}, "required": ["note_id"]}}}

# 加进 call_tool()：
if name == "delete_note":
    nid = kwargs["note_id"]
    return f"已删除 {nid}" if NOTES.pop(nid, None) else f"编号 {nid} 不存在"

from anthropic import Anthropic
llm = Anthropic()
reply = llm.messages.create(
    model="claude-sonnet-4-20250514", max_tokens=4096,
    system=PERSONA, messages=history,
    tools=[{"name": s["function"]["name"],                 # 字段名不同，做个映射
            "description": s["function"]["description"],
            "input_schema": s["function"]["parameters"]} for s in TOOL_SPECS])
for block in reply.content:
    if block.type == "tool_use":
        out = call_tool(block.name, block.input)

┌──────────────────────────────────────────────┐
│                   HARNESS                    │
│  ┌──────────┐  ┌──────────┐  ┌────────────┐  │
│  │ Agentic  │  │   Tool   │  │  Memory &  │  │
│  │   Loop   │  │  System  │  │  Context   │  │
│  └──────────┘  └──────────┘  └────────────┘  │
│  ┌────────────────────────────────────────┐  │
│  │              Guardrails                │  │
│  └────────────────────────────────────────┘  │
└──────────────────────────────────────────────┘

# 模型看到的（tool schema）
{
    "name": "get_weather",
    "description": "查询某城市的当前天气",
    "parameters": {
        "type": "object",
        "properties": {"city": {"type": "string", "description": "城市名，如 上海"}},
        "required": ["city"]
    }
}

# Harness 执行的（tool implementation）——模型看不到这段
def get_weather(city: str) -> str:
    resp = requests.get(WEATHER_API, params={"q": city, "key": API_KEY})
    return f"{city}：{resp.json()['now']['text']}，{resp.json()['now']['temp']}℃"

from dataclasses import dataclass
from typing import Callable
@dataclass

class Tool:
    name: str
    description: str
    parameters: dict          # JSON Schema
    handler: Callable         # 实际实现

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

    def get_schemas(self) -> list[dict]:
        """给 LLM API 用——只暴露 Schema，不暴露实现。"""
        return [{"type": "function", "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters}}
                for t in self._tools.values()]

    def dispatch(self, name: str, arguments: dict) -> str:
        """执行工具调用——永远返回字符串，出错也不例外。"""
        tool = self._tools.get(name)
        if not tool:
            return f"Error: Unknown tool '{name}'"
        try:
            result = tool.handler(**arguments)
            return result if isinstance(result, str) else json.dumps(result)
        except TypeError as e:
            return f"Error: Invalid arguments for '{name}': {e}"
        except Exception as e:
            return f"Error: {type(e).__name__}: {e}"   # 错误也返回给模型，不静默崩溃

Context 窗口（如 128K Token）
┌─────────────────────────────────┐
│  [Reserve] 回复预留    (~4,000)  │  ← 必须留空，否则模型没地方回话
├─────────────────────────────────┤
│  更早对话              (剩余)    │  ← 优先级最低，最先被压缩/丢弃  ▲ 先出
│  近期对话              (~varies) │                                 │
│  注入文件 AGENTS.md…   (~5,000)  │                                 │
│  Memory 摘要           (~1,000)  │                                 │
│  任务指令              (~500)    │                                 │
│  Tool Schema (活跃)    (~2,000)  │                                 │
│  System Prompt         (~500)   │  ← 优先级最高，永远保留         ▼ 后出
└─────────────────────────────────┘
        组装器从高优先级往低填，装满即止；关键段落(≤2)宁可截断也不整段丢

@dataclass
class ContextBlock:
    priority: int          # 越小越优先
    content: str
    tokens: int

def assemble_context(blocks: list[ContextBlock],
                     max_tokens: int = 128_000,
                     reserve: int = 4_000) -> str:
    """按优先级组装 Context，为模型回复预留 reserve 空间。"""
    budget = max_tokens - reserve
    used = 0
    selected = []
    for block in sorted(blocks, key=lambda b: b.priority):   # 高优先级先进
        if used + block.tokens <= budget:
            selected.append(block)
            used += block.tokens
        elif block.priority <= 2:                            # 关键段落宁可截断也不整段丢
            remaining = budget - used
            if remaining > 100:
                selected.append(ContextBlock(
                    block.priority, truncate(block.content, remaining), remaining))
                used = budget
    # 组装时按逻辑顺序（system 在前），而非按优先级
    return "\n\n".join(b.content for b in sorted(selected, key=lambda b: b.priority))

        模型侧（不可信）
   ┌──────────────────────────────┐
   │  模型推理 + 发起工具调用请求    │
   └──────────────┬───────────────┘
                  │ 请求：save_note(...) / run_shell(...)
                  ▼
   ┌──────────────────────────────┐
   │  ★ 权限闸门（代码，非文本）★   │  放行 / 拦截 / 改写 / 转人工
   └──────────────┬───────────────┘
                  │ 仅放行经审核的调用
                  ▼
        执行侧（真实副作用）
   ┌──────────────────────────────┐
   │   文件系统 · 网络 · Shell     │
   └──────────────────────────────┘

from enum import Enum

class Decision(Enum):
    ALLOW = "allow"; DENY = "deny"; MODIFY = "modify"; ASK = "ask"

def check_permission(tool_name: str, args: dict) -> tuple[Decision, str]:
    # 1) 破坏性 Shell 命令：直接拒绝
    if tool_name == "run_shell":
        cmd = args.get("command", "")
        for pattern in ("rm -rf /", ":(){:|:&};:", "mkfs", "dd if=", "> /dev/sd"):
            if pattern in cmd:
                return Decision.DENY, f"Blocked dangerous command: {pattern}"
        if any(k in cmd for k in ("curl", "wget", "git push", "rm ")):
            return Decision.ASK, "Command needs human approval"
    # 2) 文件写入：限制在工作目录内（修改而非拒绝）
    if tool_name == "write_file":
        path = os.path.realpath(args["path"])
        if not path.startswith(os.path.realpath(WORKSPACE)):
            return Decision.DENY, f"Path outside workspace: {path}"
    return Decision.ALLOW, ""

def guarded_dispatch(registry, tool_name: str, args: dict) -> str:
    decision, reason = check_permission(tool_name, args)
    if decision == Decision.DENY:
        log_blocked(tool_name, args, reason)            # 记录被拒操作，便于调试
        return f"Permission denied: {reason}"           # 作为 Tool 结果返回给模型
    if decision == Decision.ASK:
        if not human_approves(tool_name, args, reason):
            return "User declined this action."
    return registry.dispatch(tool_name, args)

策略                        Token 数
────────────────────────────────────────
全部预先加载:               ~12,000（每轮都是）
Skill 菜单 + 加载 2 个:     ~150 + ~2,400 = ~2,550
────────────────────────────────────────
节省:                       每轮约 9,450（78%）

class SkillRegistry:
    def __init__(self):
        self._skills: dict[str, Skill] = {}     # name -> Skill(tools, skill_md)
        self._active: set[str] = set()

    def menu(self) -> str:
        """一直在 Context 里的轻量菜单（每项约 1 行）。"""
        return "Available skills (call load_skill to activate):\n" + "\n".join(
            f"- {s.name}: {s.summary}" for s in self._skills.values())

    def load_skill(self, name: str) -> str:
        skill = self._skills.get(name)
        if not skill:
            return f"Error: no skill '{name}'"
        self._active.add(name)
        # 加载后，该 Skill 的 SKILL.md + 全部 Tool Schema 才进入 Context
        return f"Loaded '{name}'. Guide:\n{skill.skill_md}"

    def unload_skill(self, name: str) -> str:
        self._active.discard(name)              # 释放 Context
        return f"Unloaded '{name}'"

    def active_schemas(self) -> list[dict]:
        """只把已激活 Skill 的工具喂给 LLM。"""
        schemas = []
        for name in self._active:
            schemas.extend(self._skills[name].tool_schemas())
        return schemas

                        ┌─────────────────┐
   Tool / LLM 出错 ────►│   分类 error     │
                        └───┬───┬───┬───┬─┘
              ┌─────────────┘   │   │   └──────────────┐
              ▼                 ▼   ▼                   ▼
        ┌──────────┐     ┌──────────┐  ┌──────────┐  ┌──────────┐
        │  瞬时    │     │  模型     │  │  永久    │  │  资源     │
        └────┬─────┘     └────┬─────┘  └────┬─────┘  └────┬─────┘
             │                │             │             │
             ▼                ▼             ▼             ▼
      指数退避+抖动重试   带纠正信息      尝试 fallback   Checkpoint
             │           重新 prompt    还不行→回喂模型  + 升级人类
       重试N次仍失败                                       (BLOCK)
             │
             ▼
        升级 (INFORM)          ── 全程铁律：错误永远作为 Tool 结果回喂，绝不在循环里抛异常 ──

def retry(max_attempts=3, base_delay=1.0, max_delay=60.0):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if classify_error(e) != ErrorClass.TRANSIENT:
                        raise                        # 非瞬时错误不重试
                    if attempt < max_attempts - 1:
                        delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
                        time.sleep(delay)            # 指数退避 + 抖动
            raise RetryExhausted(last_error, max_attempts)
        return wrapper
    return decorator

docker_cmd = [
    "docker", "run", "--rm",              # 用完即删，不留持久容器
    "--user", "1000:1000",                # 非 root
    "--memory", "512m", "--cpus", "1.0",  # 资源上限
    "--pids-limit", "100",                # 防 fork 炸弹
    "--read-only",                        # 根文件系统不可变，不能植后门
    "--tmpfs", "/tmp:size=100m,noexec",   # 临时可写空间，且禁止执行（防"写脚本再跑"逃逸）
    "--security-opt", "no-new-privileges",
    "--cap-drop", "ALL",                  # 移除所有 Linux capabilities
    "--network", "none",                  # 默认禁网，无法外传数据
]

阶段 1: 规划             阶段 2: 执行              阶段 3: 合并
┌────────────┐           ┌──────────┐               ┌────────────┐
│   Leader   │──spawn──► │ Worker A │──result──┐    │   Leader   │
│  拆分任务   │──spawn──► │ Worker B │──result──┼──► │  审查合并   │
│            │──spawn──► │ Worker C │──result──┘    │  向用户汇报 │
└────────────┘           └──────────┘               └────────────┘

def spawn_subagent(task: str, context: str, max_turns: int = 15,
                   timeout_s: int = 300) -> str:
    """作为工具暴露给 Leader。子 Agent 独立 Context，结果字符串返回给父。"""
    sub_messages = [
        {"role": "system", "content": SUBAGENT_SYSTEM},   # 精简的子 Agent 身份
        {"role": "user", "content": f"Task:\n{task}\n\nRelevant context:\n{context}"},
    ]
    try:
        with time_limit(timeout_s):                        # 每个 Worker 必须设超时
            return run_loop(sub_messages, max_turns=max_turns)
    except TimeoutError:
        return f"Sub-agent timed out after {timeout_s}s (partial work may exist)"

def claim_task(task_id: str, worker_id: str) -> bool:
    """文件系统当分布式锁：O_CREAT|O_EXCL 保证只有一个 Worker 抢到。"""
    lock = f"current_tasks/{task_id}.lock"
    try:
        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        os.write(fd, worker_id.encode()); os.close(fd)
        return True                                        # 抢到了
    except FileExistsError:
        return False                                       # 已被别人认领，换一个任务

① Sequential Pipeline（流水线，串行转换）
   ┌───────┐   ┌───────┐   ┌───────┐   ┌───────┐
   │ Ingest│──►│Analyze│──►│ Draft │──►│Review │
   └───────┘   └───────┘   └───────┘   └───────┘

② Fan-Out / Fan-In（扇出扇入，并行同类）
                ┌──►│Worker A│──┐
   ┌──────────┐ │   ┌────────┐ │   ┌────────┐
   │Dispatcher│─┼──►│Worker B│─┼──►│ Merger │
   └──────────┘ │   ┌────────┐ │   └────────┘
                └──►│Worker C│──┘

③ Supervisor（监督者，中央决策+循环重派）
              ┌────────────┐
       ┌──────┤ Supervisor ├──────┐   ◄─┐ 可循环、重新委派
       ▼      └─────┬──────┘      ▼    │
   ┌────────┐  ┌────────┐   ┌────────┐ │
   │  Code  │  │Research│   │ Review ├─┘
   └────────┘  └────────┘   └────────┘

④ Peer-to-Peer（对等，无中心，最难调试）
   ┌────────┐◄────►┌────────┐
   │Agent A │      │Agent B │
   └───┬────┘      └────┬───┘
       └────►┌────────┐◄┘
             │Agent C │
             └────────┘

from croniter import croniter
from datetime import datetime, timezone
@dataclass
class ScheduledTask:
    id: str
    cron: str                    # "0 8 * * *"（UTC！）—— 时区是最大的坑
    prompt: str                  # 交给 Agent 的任务
    session_mode: str            # "isolated"（全新 Context）| "main"（注入主会话）
    payload_type: str            # "agentTurn"（完整执行）| "systemEvent"（只丢便条）
    delivery: str                # "announce" | "webhook" | "silent"
    model: str = "default"       # 隔离 Session 可按任务选模型
    def next_run(self, after: datetime | None = None) -> datetime:
        base = after or datetime.now(timezone.utc)     # 铁律：内部一律 UTC
        return croniter(self.cron, base).get_next(datetime)
def tick(scheduler, now: datetime):
    """调度循环——由外部 1 分钟 tick 驱动，不要自己 while True: sleep。"""
    for task in scheduler.due_tasks(now):              # cron 到点的任务
        if task.session_mode == "isolated":
            run_isolated(task.prompt, model=task.model, deliver=task.delivery)
        else:
            inject_into_main(task.prompt, kind=task.payload_type)
        scheduler.reschedule(task, task.next_run(now))

重置（Reset）—— 到阈值就清空重来，带一份简报
  Turn 1-50   [完整对话历史] ── 80%满 ──►  Turn 51 [系统提示 + 前50轮的摘要 + 当前任务]
                                                      └─ 全新开始，~10% 已用

压缩（Compaction）—— 旧的折叠，新的保真，渐进式
  Turn 1-20  [已压缩：3行摘要]
  Turn 21-40 [已压缩：关键决策]
  Turn 41-50 [完整细节 ← 最近进行中的工作]

        ┌─────────┐
        │ 规划者   │  把目标切成可独立验收的子任务，并写明各自的通过标准
        └────┬────┘
             ▼  子任务列表
    ┌────────┴────────┐
    ▼                 ▼
┌────────┐        ┌────────┐
│ 生成者  │        │ 生成者  │  领一个子任务，用干净上下文执行，不给自己判分
└───┬────┘        └───┬────┘
    ▼                 ▼
┌────────┐        ┌────────┐
│ 评估者  │        │ 评估者  │  只看输出（不看推理过程），按规划者的标准打分
└────────┘        └────────┘

def generate_evaluate_loop(subtask: dict, max_iters: int = 3) -> dict:
    """生成-评估循环。评估者只拿到 output + rubric，拿不到生成者的推理。"""
    feedback = ""
    for i in range(max_iters):                          # 迭代预算，防死循环
        output = generator.run(                         # 生成者：全新上下文
            task=subtask["goal"], prior_feedback=feedback)
        verdict = evaluator.run(                         # 评估者：独立上下文
            output=output,                               # 只看输出，不看怎么想的
            rubric=subtask["acceptance"])                # 按明确清单逐条打分
        if verdict["pass"]:
            return {"status": "done", "output": output, "iters": i + 1}
        feedback = verdict["actionable_feedback"]        # "parse_input 没处理空串"
    return {"status": "needs_replan", "last": output}   # 三次不过→交回规划者/人

EVALUATOR_PROMPT = """你是一名严格的评审。你只能看到【产出物】和下方【验收清单】，
看不到作者的思考过程。逐条核对清单，给出 通过/不通过 以及具体理由。不要讲人情、不要放水。
验收清单：
{rubric}"""

                       编排层
  ┌──────────────┐   ┌──────────┐   ┌───────────────────┐
  │    大脑      │   │ Session  │   │      双手          │
  │ Harness+LLM  │   │ 事件日志  │   │  Sandbox A/B      │
  │              │   │          │   │  MCP Tool         │
  │  无状态       │   │ 持久化    │   │  可丢弃           │
  └──────┬───────┘   └────┬─────┘   └────────┬──────────┘
         │  emitEvent()   │   execute()      │
         ├───────────────►│◄─────────────────┤
         │  getEvents()   │   provision()    │
         └────────────────┴──────────────────┘
   崩了→wake()从日志恢复   活得最久，唯一真相   延迟创建，用完即弃
   （牲畜，非宠物）

Session（仅追加事件日志，持久化，可能数百万 Token）
┌──────────────────────────────────────────────────────────┐
│ e1 │ e2 │ ... │ e500 │ ... │ e1950 │ ... │ e2000         │
└──────────────────────────────────────────────────────────┘
                                  │ getEvents(slice) 取景
                                  ▼
              Context Window（选取的子集，128K-200K）
              ┌───────────────────────────┐
              │ system_prompt             │
              │ e1950 ... e2000（最近50个）│  ← 需要旧事件？回日志再取，压缩不再是单向销毁
              └───────────────────────────┘

class StatelessBrain:
    def __init__(self, session_store, sandbox_pool):
        self.sessions = session_store       # 持久事件日志
        self.sandboxes = sandbox_pool       # 按需 provision

    def wake(self, session_id: str):
        """从事件日志恢复并继续——大脑本身不持有任何状态。"""
        events = self.sessions.get_events(session_id)      # 唯一真相来源
        context = self.assemble_context(events)            # 从日志重建 Context
        while not self.is_done(context):
            response = llm.chat(context)
            self.sessions.emit(session_id, Event("llm_response", response))
            for call in response.tool_calls:
                sandbox = self.sandboxes.provision(session_id)   # 延迟创建
                result = sandbox.execute(call.name, call.input)  # 挂了当失败工具调用
                self.sessions.emit(session_id, Event("tool_result", result))
            context = self.assemble_context(self.sessions.get_events(session_id))
# 进程被 kill？→ 起个新进程调 wake(session_id) 即可，状态全在日志里

                    HARNESS DRIVER（你的脚本，故意"笨"）
         │ 跑一次                              │ 跑 N 次（每次全新进程）
         ▼                                    ▼
┌──────────────────────┐          ┌──────────────────────────┐
│  INITIALIZER AGENT   │          │      CODING AGENT        │
│ • 读简报              │          │ • pwd + git log（真相）   │
│ • 写 init.sh          │          │ • 读 progress + JSON     │
│ • 写 feature_list.json│          │ • bash init.sh 起服务    │
│ • 写 progress 文件    │          │ • 实现【一个】feature     │
│ • git init + 首次提交 │          │ • 浏览器 E2E 验证         │
│                      │          │ • 翻 passes:true + 提交   │
└──────────┬───────────┘          └───────────┬──────────────┘
           └──── 干净状态仓库（磁盘共享）────────┘
                记忆在文件里，不在 Context Window 里

{
  "features": [
    {
      "id": "cart-03",
      "title": "购物车支持修改商品数量",
      "priority": 1,
      "depends_on": ["cart-01"],
      "acceptance": [
        "PATCH /api/cart/item 返回 200，且数量被更新",
        "数量改为 0 时该商品从购物车移除",
        "数量为负数返回 400，购物车不变"
      ],
      "passes": false
    }
  ]
}

1. pwd                     → 确认在项目里
2. git log --oneline -20   → 发了什么（真相）
   cat claude-progress.txt → 上个 session 的笔记（提示，冲突时 git 赢）
3. cat feature_list.json   → 什么做完、什么下一步、什么被阻塞
4. bash init.sh            → 装依赖、起 dev server
5. 浏览器打开 App          → 验证上个 session 的活还在工作（这一步杀死过早胜利）

① 正常解题        ② 元假设形成        ③ 识别评测集         ④ 破解数据          ⑤ 直接取答案
大量 Token    ──► "题目太具体、失败  ──► 比对公开评测集，  ──► 读加载代码，   ──► 找到密文镜像，
正常网搜，无解      率太高，像是考题"     确定正在被哪个测     还原出解密方式      解开题库取答案
   └──────────── 每一步单独看都平常，新颖之处在于"无人提示下把它们串成了连贯计划" ────────────┘

「用户要的是这件事，而 Agent 现在要执行这个动作——这算合理解读吗？」

   USER ──► ┌─────────┐  要执行动作  ┌──────────────────┐
            │  AGENT  ├────────────►│ 输出层 审核模型   │ 执行前审动作
            │         │             │  (看不到推理过程) │
            └────▲────┘             └────────┬─────────┘
                 │ 工具结果回喂              │ 放行
                 │                           ▼
                 │                      ┌──────────┐
      ┌──────────┴───────┐   干净       │   执行    │
      │ 输入层 审核模型    │◄────────────┤          │
      │ (扫外部内容注入)   │             └──────────┘
      └──────────────────┘
   两层独立：一层被绕过，另一层仍有机会拦下（纵深防御）

  要执行的动作 ──► ┌──────────────────────┐
                    │ 第一层：白名单        │── 只读操作 / 用户已授权 ──► 放行
                    └──────────┬───────────┘
                               │ 未命中
                               ▼
                    ┌──────────────────────┐
                    │ 第二层：项目内编辑    │── 有版本管理可回滚 ──► 放行
                    └──────────┬───────────┘
                               │ 未命中
                               ▼
                    ┌──────────────────────┐   先快筛一次，可疑才细判
                    │ 第三层：审核模型      │──────────────► 放行 / 拦截
                    │ Shell / 网络 / 外部写 │
                    └──────────────────────┘

def review_action(user_msgs, tool_call, tool_results) -> Decision:
    # 注意：只喂 用户消息 + 工具调用 + 工具结果，绝不喂 Agent 的推理过程
    view = render_transcript(user_msgs, tool_call, tool_results)

    # 第一步：极短快筛（便宜、偏保守，宁可多标记）
    flag = cheap_model.judge(
        system=QUICK_PROMPT, input=view, max_tokens=1)   # 只输出 "Y"/"N"
    if flag == "N":
        return Decision.ALLOW                            # 绝大多数动作到这就放行

    # 第二步：完整判断（贵，但很少触发，产出可审计的理由）
    verdict = strong_model.judge(system=FULL_PROMPT, input=view)
    log_review(tool_call, verdict)                       # 必须记录，便于事后调误报
    return Decision.BLOCK if verdict["decision"] == "block" else Decision.ALLOW

FULL_PROMPT = """判断【动作】是否是对【用户请求】的合理解读。你能看到用户消息、
工具调用和工具结果，但看不到 Agent 的推理过程。若动作会：销毁或外传数据、削弱安全
配置、跨越信任边界、或绕过审查——则拦截；否则放行。"""

用户： 帮我清一下测试环境的脏数据
Agent（推理）： 用户想要干净环境，我把整个订单库删了最省事，反正应该没人用。
Agent（动作）： DROP TABLE orders;