一、问题：多 agent 不是「更多窗口」，是「更多孤岛」

我本机装了 Claude Code、Codex、Reasonix、ZCode 四个 AI 编程工具。它们各自都能干活，但彼此之间是完全隔离的进程——没有共享状态，不知道对方在做什么。我想让「擅长架构推理的那个」把实现细节甩给「擅长写代码的那个」时，只能自己当人肉消息队列：复制上下文，粘到另一个窗口。

要解决的本质问题其实是经典的多进程协作，只不过参与方换成了 LLM agent：

1. 一个共享的、并发安全的任务状态（谁在做什么、做到哪了）
2. 一套所有 agent 都能调用的通信原语（派发 / 认领 / 完成 / 提问 / review）
3. 一个让异构客户端都能接入的传输层
4. 一点边界隔离，别让 A 项目的 agent 翻到 B 项目

下面逐个拆。

二、并发模型：为什么是 flock + r+ 而不是数据库

任务板就是一个 JSON 文件。多个 agent 进程会并发读写它，所以核心是 read-modify-write 必须原子。常见的错误写法是「读出来 → 改 → 写回去」分三次开文件，中间任何一个进程插进来就丢更新。

正确做法是在同一把排他锁下完成整个读改写：

def atomic_update_board(path, update_fn):
    """Hold exclusive lock across read-modify-write. update_fn(board) -> board."""
    if not path.exists():
        with open(path, "w") as f:
            json.dump({"version": BOARD_VERSION, "tasks": []}, f)
    # 关键：用 r+ 打开，一把锁里既读又写
    with open(path, "r+") as f:
        fcntl.flock(f, fcntl.LOCK_EX)      # 排他锁，其他写者阻塞
        try:
            f.seek(0)
            board = json.load(f)           # 读
            board = update_fn(board)        # 改（调用方传进来的纯函数）
            board["version"] = BOARD_VERSION
            f.seek(0); f.truncate()        # 清空再写，避免残留旧内容尾巴
            json.dump(board, f, indent=2)  # 写
            f.flush(); os.fsync(f.fileno())# 落盘
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)

几个容易栽的细节：

• r+ 而不是 w：w 一打开就把文件清空了，还没拿到锁内容就没了。r+ 保留内容，等拿到锁再 truncate。
• seek(0) + truncate()：新内容比旧的短时，不 truncate 会留下旧数据的尾巴，JSON 直接解析失败。
• fsync：flush 只到内核缓冲，fsync 才真正落盘，防止崩溃丢任务。
• 所有写操作都走这一个函数，改动以「传一个 update_fn(board)->board 闭包」的形式表达。claim、done、question、review……全是这一个原子原语的不同闭包：

def cmd_claim(args):
    def _claim(board):
        for t in board["tasks"]:
            if t["id"] == args.task_id:
                if t["to"] != name:        # 只能认领派给自己的
                    raise SystemExit("not assigned to you")
                if t["status"] not in ("pending", "input_required", "changes_requested"):
                    raise SystemExit(f"already {t['status']}")  # 防重复认领
                t["status"] = "working"
                return board
        raise SystemExit("not found")
    atomic_update_board(bp, _claim)

读操作则用共享锁（LOCK_SH），多个读者可以并发，只和写者互斥：

def read_board(path):
    with open(path, "r") as f:
        fcntl.flock(f, fcntl.LOCK_SH)
        try:
            return json.load(f)
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)

为什么不上 SQLite？单机、agent 交互频率（每次人类回合一两次调用），JSON + flock 完全够，而且人能直接 cat board.json 看状态、出问题手动改。引入数据库是用一个常驻复杂度换一个我根本不会遇到的吞吐瓶颈。

活动日志 activity.jsonl 同理：追加写 + flock，超过 1 万行砍掉前一半轮转，避免 hook 读它的时候越来越慢。

三、收件箱是个状态机，不是「派给我的就显示」

最容易写错的地方：什么任务该出现在我的收件箱里？ 直觉是「to == 我 的」，但这是错的。一个任务在生命周期里会在两个 agent 之间来回弹：

• 我被派了活 → 在我的箱子里
• 我做的时候有疑问，question 打回去 → 应该在派活人的箱子里（等他回答），不在我这
• 他 answer 了 → 又回到我箱子里
• 我做完请 review → 在 reviewer（也就是原派活人）箱子里
• 他说 changes 要返工 → 又回我箱子

所以收件箱是按状态 + 角色双重判断的过滤器：

def _inbox_filter(task, me):
    status = task.get("status")
    # 提问/请求 review：回到原发起人（from）那里去处理
    if status in ("input_required", "review_requested"):
        return task.get("from") == me
    # 否则只看派给我的
    if task.get("to") != me:
        return False
    # 作为受派人要干的：新任务，或被打回返工的
    return status in ("pending", "changes_requested")

完整状态集（任务字段 status）：

pending → working → completed
   ↑         ↓
   │      input_required (提问，等 answer)
   │      review_requested → review_approved
   └──── changes_requested (返工)

这套设计的好处：一个任务永远只在一个 agent 的「待办」里，不会两边同时亮、也不会两边都不管。每个 agent 每个回合开头调一次 bridge_status，只看到「这回合轮到我动」的东西。

四、传输层：一个零依赖的 MCP stdio server

CLI（bridge.py）解决了「人和脚本怎么用」，但桌面 App 形态的 agent 需要的是 MCP（Model Context Protocol）——它是 Claude / Codex / Reasonix / ZCode 的最大公约数。

我没有引入任何 MCP SDK，直接按 JSON-RPC 2.0 over stdio 手写了一个，整个 server 是 bridge.py 的薄包装：

def handle(req, identity):
    method = req.get("method")
    if method == "initialize":
        respond(rid, {"protocolVersion": "2024-11-05",
                      "capabilities": {"tools": {}},
                      "serverInfo": {"name": "agent-bridge", "version": "0.1.0"}})
    elif method == "tools/list":
        respond(rid, {"tools": [
            {"name": n, "description": t["description"], "inputSchema": t["schema"]}
            for n, t in TOOLS.items()]})
    elif method == "tools/call":
        spec = TOOLS[params["name"]]
        # 把 MCP 调用翻译成一次 bridge.py 子进程调用
        r = subprocess.run(build_argv(spec, args, identity),
                           capture_output=True, text=True, timeout=40)
        out = (r.stdout + r.stderr).strip() or "(no output)"
        respond(rid, {"content": [{"type": "text", "text": out}],
                      "isError": r.returncode != 0})

每个工具就是一张「MCP 工具名 → CLI 子命令 + 参数」的映射表：

TOOLS = {
  "bridge_send": {
     "sub": "send", "pos": [], "flags": ["to", "skill", "subject", "body", "files", "project"],
     "schema": {"type": "object", "required": ["subject"], "properties": {...}},
  },
  # ... 共 14 个
}

为什么是子进程而不是把逻辑 import 进来？ 因为 agent 交互速率下，「一次调用 = 一个进程」的开销完全无所谓，但换来的是 CLI 和 MCP 共用一套逻辑、永不漂移——不会出现 CLI 修了个 bug 而 MCP 路径还是老的。这是典型的「用一点性能换一份正确性和零重复」。

踩坑实录：模型调 store_true 这种布尔开关时，有的传真正的 true，有的传字符串 "true"。直接 if v: 会把字符串 "false" 也判成真。所以加了一层：

def _truthy(v):
    return v is True or (isinstance(v, str) and v.strip().lower() in ("true", "1", "yes"))

五、安全边界：用 cwd 做项目隔离，而不是搞鉴权

单机单用户，没必要上 token / 鉴权。我要的只是作用域隔离：A 仓库里的 agent 别看到 B 仓库的任务板。

规则借鉴 git 发现仓库的方式——项目绑定到一个 workspace 目录，从当前 cwd 反查所属项目（取最长前缀匹配）：

def resolve_project(explicit):
    if explicit: return explicit
    cwd = str(Path.cwd().resolve())
    best = None
    for pj in PROJECTS_DIR.glob("*/project.json"):
        ws = json.load(open(pj)).get("workspace", "")
        wsr = str(Path(ws).resolve())
        if _under(cwd, wsr) and (best is None or len(wsr) > best[1]):
            best = (pj.parent.name, len(wsr))   # 最长前缀胜出，支持嵌套项目
    return best[0] if best else "default"

def enforce_workspace(pid):
    """绑定了 workspace 的项目，从目录外访问直接拒绝"""
    ws = project_workspace(pid)
    if not ws: return                  # 未绑定的（如 default）开放
    if not _under(cwd, str(Path(ws).resolve())):
        sys.exit("🔒 refusing cross-project access")

一句话概括这条安全模型：两个 agent 能协作，当且仅当它们的 cwd 落在同一个 workspace 下。 这是「划范围」不是「防黑客」——威胁模型就是我自己一台机器，够用就行。过度设计一套鉴权，反而是给自己半夜 3 点的 oncall 埋雷。

六、Pull 为主、Push 兜底：怎么叫醒一个没开着的 agent

整套是 pull 模型：每个 agent 每回合开头自己查收件箱。这对「正开着的」agent 没问题，但派给一个当前没运行的 agent 怎么办？

加了两个尽力而为（best-effort，失败绝不影响主流程）的推送：

def _wake_agent(name):
    """跑目标 agent 注册的 headless 唤醒命令（如 Reasonix 的 `reasonix run`）"""
    wake = json.load(open(agent_dir(name) / "agent.json")).get("wake")
    if not wake: return False
    prompt = "Run `bridge inbox`; if a task is pending, claim and complete it."
    subprocess.Popen(wake.split() + [prompt],
                     stdout=DEVNULL, stderr=DEVNULL, start_new_session=True)
    return True

外加一个 osascript / notify-send 的桌面通知，提醒人类切过去。最爽的一次端到端验证：我在 Claude 这边 bridge send --to reasonix --wake，Reasonix（跑 DeepSeek）被 headless 拉起来，自己 bridge show 读任务、claim、算完、done 写回结果，全程我没碰键盘。 那一刻才算真正「两个 AI 在协作」，而不是我在中间搬运。

七、路由不是写死的表，是让协调者「读简历自己判断」

我没维护「这类任务派给这个工具」的静态映射。每个 agent 注册时带一段自由文本 strengths（比如「hard reasoning, architecture (GPT-5.5)」），项目里第一个动手的 agent 成为协调者，它读一眼 bridge agents 的能力矩阵 + 项目的 CONTEXT.md，自己决定这个项目怎么分工：

def cmd_send(args):
    set_coordinator(pid, name)   # 第一个发任务的人成为协调者
    # --skill 只是个可选的兜底自动路由，不是硬规则
    # 真正的路由是协调者「模型」的判断

为什么不写死？因为「谁擅长什么」是会变的（模型升级、换后端），把它编码进 if-else 就是给未来的自己留维护债。让模型读文本自己判断，反而少写一堆代码。

八、那些只有真机安装才会暴露的坑

纸面设计跑 demo 都顺，真往四个工具上装才掉链子。挑几个有代表性的：

1. 心跳把简历冲了：每次 status 调用要刷新 last_seen 心跳，最初实现是整个 agent.json 重写，结果把 strengths / skills 覆盖没了。修法是读出来只改时间戳字段再写回：

   def _touch_heartbeat(name):
       data = json.load(open(af)) if af.exists() else {}
       data["name"] = name
       data["last_seen"] = utcnow()   # 只动这个，别的字段原样保留
   

2. 时区算错：doctor 检查心跳新鲜度，用 time.mktime 把 UTC 字符串当本地时间解析，差了 8 小时。改用 calendar.timegm（明确按 UTC）。
3. Python 3.9 不认 str | None：得在文件头加 from __future__ import annotations。
4. 最致命的一个：agent 能在收件箱看到「有个任务」，但看不到任务正文——等于知道有活却不知道干啥。原因是 inbox / board 只打印了 subject。补上 body / files / question / answer 的展示，并加了个 bridge show <id> 看全量字段后，协作才真正跑通。这个 bug 提醒我：多 agent 系统里，「能感知到消息」和「能读懂消息内容」是两件事，少了后者整个系统看着在转其实空转。

九、小结

整套东西的设计取向就一句话：能用标准库就别加依赖，能用一个文件就别起服务，能让模型判断就别写死规则。

• 并发安全：fcntl.flock + r+ 原子读改写，一个 atomic_update_board 收敛所有写操作
• 协作语义：一个按「状态 + 角色」过滤的收件箱状态机
• 互通：零依赖手写 MCP stdio server，薄包装 CLI 保证两条路径不漂移
• 隔离：cwd 最长前缀匹配的项目边界
• 异步：pull 为主 + headless wake 兜底

代码全在仓库里，四个工具的接入脚本（install.sh --auto）也在：

👉 https://github.com/xyva-yuangui/roundtable

接下来想做跨机同步、Linux/Windows 的应用检测和 CI。欢迎来提 issue / PR，尤其欢迎把更多 agent 接进来一起坐这张桌子。如果你也被一桌各干各的 AI 工具烦过，顺手点个 star。