《模型竞赛转向执行层》

#!/usr/bin/env python3
"""把 Claude Code 三类失败 Hook 统一记录为 JSONL；不改变原始决策。"""
import json
import pathlib
import sys
from datetime import datetime, timezone

payload = json.load(sys.stdin)
event = str(payload.get("hook_event_name", "未知事件"))
record = {
    "时间": datetime.now(timezone.utc).isoformat(),
    "事件": event,
    "会话ID": payload.get("session_id"),
    "智能体ID": payload.get("agent_id"),
    "工具": payload.get("tool_name"),
    "工具输入": payload.get("tool_input"),
    "错误": payload.get("error") or payload.get("error_type"),
    "原始事件": payload,
}
path = pathlib.Path(".claude/failure-timeline.jsonl")
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as file:
    file.write(json.dumps(record, ensure_ascii=False) + "\n")

# 只记录，不批准、不拒绝、不要求重试。
print("{}")

{
  "hooks": {
    "PermissionDenied": [
      {
        "matcher": "*",
        "hooks": [
          {"type": "command", "command": "python .claude/hooks/failure_timeline.py", "timeout": 10}
        ]
      }
    ],
    "PostToolUseFailure": [
      {
        "matcher": "*",
        "hooks": [
          {"type": "command", "command": "python .claude/hooks/failure_timeline.py", "timeout": 10}
        ]
      }
    ],
    "StopFailure": [
      {
        "matcher": "rate_limit|overloaded|authentication_failed|billing_error|server_error|unknown",
        "hooks": [
          {"type": "command", "command": "python .claude/hooks/failure_timeline.py", "timeout": 10}
        ]
      }
    ]
  }
}

#!/usr/bin/env python3
import json
import pathlib
from collections import Counter

path = pathlib.Path(".claude/failure-timeline.jsonl")
rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
counts = Counter(row.get("事件", "未知事件") for row in rows)
print(f"失败记录总数：{len(rows)}")
for name, count in counts.most_common():
    print(f"- {name}：{count} 次")
print("最近三条：")
for row in rows[-3:]:
    print(f"- {row.get('时间')}｜{row.get('事件')}｜工具={row.get('工具')}｜错误={row.get('错误')}")

python summarize_failures.py

hermes kanban init
TASK_ID=$(hermes kanban create "为订单接口增加幂等保护" \
  --assignee implementer \
  --workspace worktree \
  --branch "hermes/order-idempotency" \
  --body "目标：重复提交同一幂等键时只创建一笔订单。验收：1）新增并发重复请求测试；2）全部原有测试通过；3）不得记录完整支付令牌；4）完成后请求 reviewer 复核。" \
  --json | python -c "import json,sys; print(json.load(sys.stdin)['id'])")
printf '%s\n' "$TASK_ID"

kanban_request_review(
  summary="已增加幂等键唯一约束与并发测试，候选变更等待复核",
  metadata={
    "changed_files": ["src/orders.py", "tests/test_orders.py"],
    "verification": ["python -m pytest tests/test_orders.py -q：12项通过"],
    "acceptance": {
      "重复请求只创建一笔订单": true,
      "原有测试通过": true,
      "未记录完整支付令牌": true
    },
    "residual_risk": ["尚未进行跨区域数据库延迟测试"]
  },
  reviewer="reviewer"
)

kanban_request_changes(
  reason="并发测试只覆盖单进程；请增加两个独立连接同时提交同一幂等键的用例。"
)

kanban_complete(
  summary="独立复核通过：验收条件全部满足，并发重复提交测试稳定通过。",
  metadata={
    "reviewed_files": ["src/orders.py", "tests/test_orders.py"],
    "verification": ["python -m pytest tests/test_orders.py -q：13项通过"],
    "decision": "通过"
  }
)

hermes kanban show "$TASK_ID"
hermes kanban runs "$TASK_ID"

#!/usr/bin/env python3
"""Claude 批量队列：先真实计数，再按 RPM/ITPM/OTPM 令牌桶排队。"""
import json
import os
import sys
import time
import urllib.error
import urllib.request

API_KEY = os.environ["ANTHROPIC_API_KEY"]
MODEL = os.environ["CLAUDE_MODEL"]
RPM = int(os.getenv("CLAUDE_RPM", "30"))
ITPM = int(os.getenv("CLAUDE_ITPM", "50000"))
OTPM = int(os.getenv("CLAUDE_OTPM", "10000"))
MAX_OUTPUT = int(os.getenv("CLAUDE_MAX_OUTPUT", "800"))
VERSION = "2023-06-01"

class Bucket:
    def __init__(self, capacity: int):
        self.capacity = float(capacity)
        self.tokens = float(capacity)
        self.rate = float(capacity) / 60.0
        self.updated = time.monotonic()

    def wait_for(self, amount: int):
        if amount > self.capacity:
            raise SystemExit(f"单个任务需要 {amount}，超过该分钟桶容量 {int(self.capacity)}")
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens >= amount:
                self.tokens -= amount
                return
            time.sleep(max(0.05, (amount - self.tokens) / self.rate))

requests_bucket = Bucket(RPM)
input_bucket = Bucket(ITPM)
output_bucket = Bucket(OTPM)

def post(path: str, body: dict):
    req = urllib.request.Request(
        "https://api.anthropic.com" + path,
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={
            "x-api-key": API_KEY,
            "anthropic-version": VERSION,
            "content-type": "application/json",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=180) as resp:
            return json.loads(resp.read().decode("utf-8")), dict(resp.headers)
    except urllib.error.HTTPError as exc:
        retry_after = exc.headers.get("retry-after")
        detail = exc.read().decode("utf-8", "replace")
        if exc.code == 429 and retry_after:
            time.sleep(float(retry_after))
            return post(path, body)
        raise SystemExit(f"请求失败，状态码 {exc.code}；请检查密钥、额度或请求格式。详情：{detail}")

def run_job(text: str, number: int):
    body = {"model": MODEL, "messages": [{"role": "user", "content": text}]}
    count, _ = post("/v1/messages/count_tokens", body)
    input_tokens = int(count["input_tokens"])

    # 计数请求也占 RPM；正式生成前为请求、输入和最大输出预留预算。
    requests_bucket.wait_for(1)
    input_bucket.wait_for(input_tokens)
    output_bucket.wait_for(MAX_OUTPUT)

    result, headers = post("/v1/messages", {**body, "max_tokens": MAX_OUTPUT})
    answer = "".join(block.get("text", "") for block in result.get("content", []))
    actual_output = int(result.get("usage", {}).get("output_tokens", 0))
    request_id = headers.get("request-id", "未返回")
    print(json.dumps({
        "任务": number,
        "输入词元": input_tokens,
        "输出词元": actual_output,
        "请求ID": request_id,
        "回答": answer,
    }, ensure_ascii=False))

for index, line in enumerate(sys.stdin, 1):
    if line.strip():
        run_job(line.strip(), index)

export ANTHROPIC_API_KEY="你的密钥"
export CLAUDE_MODEL="你账号当前可用的模型ID"
export CLAUDE_RPM="30"
export CLAUDE_ITPM="50000"
export CLAUDE_OTPM="10000"
export CLAUDE_MAX_OUTPUT="800"
python claude_rate_queue.py < jobs.txt > results.jsonl