多 Agent 协调新方案：Git Refs 与 CRDT

协调两个使用 Git Refs 和 CRDT 的编码智能体——智能体实验室日志

  智能体实验室日志

指南
    术语表

实践 · 智能体工程 · Git

      级别：高级
      阅读和实验时间：60 分钟
      结果：保留两个智能体各自独立完成的更改，而无需将它们的分支合并到工作分支

并行编码智能体失败的主要原因，并不是 Git 无法合并文本，而是一个分支被要求同时表示三种不同的事物：私有工作、共享协调状态以及已接受的项目状态。
      在本实验中，两个编码智能体从同一个任务开始，在彼此不可见的情况下工作，通过私有 Git refs 发布不可变操作，并在各自独立的会话结束后恢复。一个小型无冲突复制数据类型（CRDT）会合并它们的意图。协调器从不将任何智能体分支合并到工作分支中，但最终任务仍会包含双方的更改，并且可以仅依靠 Git 重新生成。

为什么仅靠分支还不够

为什么仅靠分支还不够

前置条件与安全边界

前置条件与安全边界

步骤 1：创建仓库

步骤 1：创建仓库

步骤 2：实现 CRDT 归约器

步骤 2：实现 CRDT 归约器

步骤 3：发布两个独立更改

步骤 3：发布两个独立更改

步骤 4：通过 Git refs 同步

步骤 4：通过 Git refs 同步

步骤 5：物化已接受状态

步骤 5：物化已接受状态

步骤 6：验证会话恢复

步骤 6：验证会话恢复

验证协议

验证协议

连接真实的编码智能体

连接真实的编码智能体

      本实验使用一个包含两个可独立编辑字段的任务。智能体 Alpha 将任务状态从 open 更改为 in_progress。智能体 Beta 添加一项验证要求。每个智能体都将自己的更改记录为一个独立的不可变操作，并且只推进自己的 ref。

一个确定性归约器直接从 Git 对象数据库读取两个 refs，验证其中的操作，并且无论以何种顺序提供这些 refs，都会创建相同的物化任务。已接受分支接收的是一个生成文件，而不是任何智能体工作区的合并结果。

实验结束时，必须验证以下所有性质：

智能体从同一个基础提交开始；

智能体从同一个基础提交开始；

任何智能体都不需要另一个智能体的检出目录或对话历史；

任何智能体都不需要另一个智能体的检出目录或对话历史；

每个智能体都有一个私有发布 ref；

每个智能体都有一个私有发布 ref；

操作标识符不会发生冲突；

操作标识符不会发生冲突；

先应用 Alpha 再应用 Beta，与先应用 Beta 再应用 Alpha，会得到相同的状态；

先应用 Alpha 再应用 Beta，与先应用 Beta 再应用 Alpha，会得到相同的状态；

重放同一个操作不会导致该操作被应用两次；

重放同一个操作不会导致该操作被应用两次；

最终任务包含两个独立更改；

最终任务包含两个独立更改；

工作分支中不存在来自任何智能体分支的合并提交；

工作分支中不存在来自任何智能体分支的合并提交；

新会话可以仅通过 refs 和已提交对象重建结果。

新会话可以仅通过 refs 和已提交对象重建结果。

这里的“无冲突”意味着什么

        这并不意味着任意并发源代码编辑都会在语义上兼容。它意味着智能体在已接受的工作树之外发布结构化意图，并且定义的操作集具有确定性的收敛规则。在生成状态成为已接受的项目状态之前，仍然需要人工或自动化审查。

为什么仅靠分支还不够

      传统分支记录的是一系列文件系统快照。这对源代码很有用，但分支无法说明两个快照表示的是重复工作、独立新增内容，还是相互竞争的决策。如果两个智能体编辑同一个 JSON 任务文件，Git 看到的只是文本行。协调器需要的是具有标识、作者、逻辑顺序和明确语义的操作。

常见的“每个智能体一个分支”设计也存在协调缺口。智能体可能正确完成了提交，却在创建拉取请求之前消失。另一个会话可能知道分支名称，却不知道任务的前提假设。第三个进程可能合并两个分支并解决文本冲突，但在此过程中意外丢弃了某个智能体的意图。

我们将其划分为三个层次：

私有执行状态。每个智能体都可以使用自己的目录、临时文件、提示词和未提交的实验。

私有执行状态。每个智能体都可以使用自己的目录、临时文件、提示词和未提交的实验。

已发布的协调状态。不可变操作可以通过 refs/agents/ 下的 refs 访问。

已发布的协调状态。不可变操作可以通过 refs/agents/ 下的 refs 访问。

已接受的项目状态。协调器将经过验证的操作物化到 main 上，而不合并智能体分支。Git 提交会成为不可变的传输信封。它的哈希值提供内容寻址，父提交提供来源信息，而 ref 则提供一个可移动指针，指向智能体最新发布的信封。Git 是传输与保留层；CRDT 定义并发意图如何收敛。

已接受的项目状态。协调器将经过验证的操作物化到 main 上，而不合并智能体分支。

  Git 提交会成为不可变的传输信封。它的哈希值提供内容寻址，父提交提供来源信息，而 ref 则提供一个可移动指针，指向智能体最新发布的信封。Git 是传输与保留层；CRDT 定义并发意图如何收敛。

本实验保留以下名称：

refs/heads/main
refs/agents/alpha/task-42
refs/agents/beta/task-42

      `refs/agents/` 下的 refs 是普通的 Git refs，但不会在常规的 git branch 输出中显示为本地分支。智能体使用 git update-ref 更新它们。协调器使用 git show 读取它们的提交树，并且绝不会将它们检出并覆盖工作分支。

      共享数据类型是一个只增操作集合。每个操作都有一个全局唯一标识符。集合并集具有交换律、结合律和幂等性，因此重复传递或不同的发现顺序都不会改变集合成员。

任务本身有两种字段策略：

status 是一个按照逻辑元组排序的最后写入者胜出寄存器；

status 是一个按照逻辑元组排序的最后写入者胜出寄存器；

requirements 是一个以稳定需求标识符为键的只增集合。最后写入者胜出寄存器使用 Lamport 时钟，并以参与者标识符作为确定性的平局决胜规则。时钟建立的是可复现的顺序；它并不能证明较晚的决策更加正确。

requirements 是一个以稳定需求标识符为键的只增集合。

  最后写入者胜出寄存器使用 Lamport 时钟，并以参与者标识符作为确定性的平局决胜规则。时钟建立的是可复现的顺序；它并不能证明较晚的决策更加正确。

{
  "op_id": "alpha:task-42:1",
  "actor": "alpha",
  "clock": 1,
  "task": "task-42",
  "kind": "set_status",
  "value": "in_progress"
}

      操作文件位于 ops/<actor>/<op-id>.json。每个文件只保存一个操作，这可以避免并发追加造成的数据损坏，并使重复标识符变得可观察。智能体绝不能在操作发布后编辑它；如需修正，必须创建一个新操作。

为什么归约器要对输入排序

      集合并集决定存在哪些操作，但物化器仍然需要生成稳定的输出字节。因此，归约器会根据 op_id 去重，拒绝标识符相同但载荷不同的操作，并在归约前对操作排序。JSON 键和需求列表也会按照固定顺序序列化。

具体案例：两个智能体更改同一个任务

      任务 task-42 的初始状态如下：

{
  "id": "task-42",
  "title": "Prevent duplicate payment retries",
  "status": "open",
  "requirements": []
}

      两个智能体接收相同的基础修订版本，但承担不同的职责：

参与者
          独立任务
          发布的操作

Alpha
          认领该任务并进行实现
          set_status("in_progress")

Beta
          添加一个验证条件
          add_requirement("retry-idempotency-test", ...)

这些操作涉及不同的逻辑组件。它们可以组合起来，而无需让 Git 合并 tasks/task-42.json 的两个已编辑副本。

前置条件与安全边界

Python 3.9 或更高版本，仅使用标准库；

Python 3.9 或更高版本，仅使用标准库；

兼容 POSIX 的 shell；

兼容 POSIX 的 shell；

一个空的、可随时丢弃的目录。

一个空的、可随时丢弃的目录。

git --version
python3 --version

      请在一个新建的、用完即可丢弃的目录中运行以下实验，不要在现有仓库内运行。这些命令会创建仓库、提交、引用和临时工作区。如果你的环境已经提供了作者身份信息，请替换示例中的作者身份。

在生产系统中，不应授予 AI 智能体对所有引用不受限制的访问权限。应仅允许每个参与者更新自己的命名空间，并由可信的协调器管理已接受分支。

步骤 1：创建仓库

创建并进入一个空目录：

mkdir agent-ref-crdt-lab
cd agent-ref-crdt-lab
git init -b main
git config user.name "Agent Lab"
git config user.email "lab@example.invalid"

mkdir -p base ops tools tasks

创建 base/task-42.json：

{
  "id": "task-42",
  "title": "Prevent duplicate payment retries",
  "status": "open",
  "requirements": []
}

# Agent ref and CRDT laboratory

Immutable agent operations are published under refs/agents/.
The accepted task in tasks/ is generated by tools/reduce.py.
Do not edit a published operation in place.

提交公共基础版本：

git add README.md base/task-42.json
git commit -m "Initialize task coordination laboratory"
BASE_COMMIT=$(git rev-parse HEAD)
printf '%s\n' "$BASE_COMMIT"

      请在当前终端会话中保留打印出的提交标识符。两个 AI 智能体都将使用这个确切的提交作为其发布内容的父提交，从而证明二者都不是基于对方的提交创建的。

步骤 2：实现 CRDT 归并器

      使用以下代码创建 tools/reduce.py：

#!/usr/bin/env python3
import argparse
import hashlib
import json
import subprocess
import sys
from pathlib import Path

ALLOWED_STATUS = {"open", "in_progress", "blocked", "done"}
ALLOWED_KINDS = {"set_status", "add_requirement"}

def git(*args):
    completed = subprocess.run(
        ["git", *args],
        check=True,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    return completed.stdout

def canonical(value):
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")

def validate(operation):
    required = {"op_id", "actor", "clock", "task", "kind", "value"}
    if set(operation) != required:
        raise ValueError(
            f"{operation.get('op_id', '<unknown>')}: invalid keys"
        )
    if not isinstance(operation["op_id"], str) or not operation["op_id"]:
        raise ValueError("op_id must be a non-empty string")
    if not isinstance(operation["actor"], str) or not operation["actor"]:
        raise ValueError(f"{operation['op_id']}: invalid actor")
    if not isinstance(operation["clock"], int) or operation["clock"] < 1:
        raise ValueError(f"{operation['op_id']}: invalid clock")
    if operation["task"] != "task-42":
        raise ValueError(f"{operation['op_id']}: unexpected task")
    if operation["kind"] not in ALLOWED_KINDS:
        raise ValueError(f"{operation['op_id']}: unsupported kind")

value = operation["value"]
    if operation["kind"] == "set_status":
        if value not in ALLOWED_STATUS:
            raise ValueError(f"{operation['op_id']}: invalid status")
    else:
        if not isinstance(value, dict) or set(value) != {"id", "text"}:
            raise ValueError(f"{operation['op_id']}: invalid requirement")
        if not all(isinstance(value[key], str) and value[key]
                   for key in ("id", "text")):
            raise ValueError(f"{operation['op_id']}: empty requirement field")

def read_base():
    with Path("base/task-42.json").open(encoding="utf-8") as stream:
        return json.load(stream)

def operations_from_ref(ref):
    paths = git("ls-tree", "-r", "--name-only", ref, "--", "ops").splitlines()
    operations = []
    for path in paths:
        if not path.endswith(".json"):
            continue
        raw = git("show", f"{ref}:{path}")
        operation = json.loads(raw)
        validate(operation)
        operations.append(operation)
    return operations

def collect(refs):
    by_id = {}
    encoded_by_id = {}

for ref in refs:
        for operation in operations_from_ref(ref):
            op_id = operation["op_id"]
            encoded = canonical(operation)
            if op_id in by_id and encoded_by_id[op_id] != encoded:
                raise ValueError(f"divergent payloads for {op_id}")
            by_id[op_id] = operation
            encoded_by_id[op_id] = encoded

return sorted(
        by_id.values(),
        key=lambda item: (
            item["clock"],
            item["actor"],
            item["op_id"],
        ),
    )

def reduce_operations(base, operations):
    status = base["status"]
    status_version = (0, "base", "base:0")
    requirements = {
        item["id"]: item for item in base.get("requirements", [])
    }

for operation in operations:
        version = (
            operation["clock"],
            operation["actor"],
            operation["op_id"],
        )
        if operation["kind"] == "set_status":
            if version > status_version:
                status = operation["value"]
                status_version = version
        elif operation["kind"] == "add_requirement":
            requirement = operation["value"]
            existing = requirements.get(requirement["id"])
            if existing is not None and existing != requirement:
                raise ValueError(
                    "divergent requirement payload for "
                    + requirement["id"]
           

在创建 AI 智能体的发布内容之前提交归并器：

chmod +x tools/reduce.py
git add tools/reduce.py
git commit -m "Add deterministic task operation reducer"
BASE_COMMIT=$(git rev-parse HEAD)

      在这里更新 BASE_COMMIT 非常重要：两个 AI 智能体的提交都必须包含归并器的父版本，尽管它们的目录树只会分别添加各自的操作。

步骤 3：发布两个独立变更

      Git worktree 可以让每个 AI 智能体拥有独立的文件系统，同时共享同一个对象数据库。worktree 对本实验而言很方便，但它并不是收敛算法的一部分。不同的克隆同样可以将这些引用发布到共享的裸仓库。

从公共基础版本创建一个分离 HEAD 的 worktree：

git worktree add --detach .agent-alpha "$BASE_COMMIT"
mkdir -p .agent-alpha/ops/alpha

创建 .agent-alpha/ops/alpha/alpha-task-42-1.json：

{
  "op_id": "alpha:task-42:1",
  "actor": "alpha",
  "clock": 1,
  "task": "task-42",
  "kind": "set_status",
  "value": "in_progress"
}

提交该操作并发布私有引用：

git -C .agent-alpha add ops/alpha/alpha-task-42-1.json
git -C .agent-alpha commit -m "Alpha claims task-42"
ALPHA_COMMIT=$(git -C .agent-alpha rev-parse HEAD)
git update-ref refs/agents/alpha/task-42 \
  "$ALPHA_COMMIT" \
  0000000000000000000000000000000000000000

      全零的旧值表示“仅在引用不存在时创建”。这是一种比较并交换（compare-and-swap）护栏：如果意外发现已有引用，发布操作就会失败，而不会悄无声息地覆盖另一个会话的指针。

从同一个基础版本创建另一个分离 HEAD 的 worktree：

git worktree add --detach .agent-beta "$BASE_COMMIT"
mkdir -p .agent-beta/ops/beta

创建 .agent-beta/ops/beta/beta-task-42-1.json：

{
  "op_id": "beta:task-42:1",
  "actor": "beta",
  "clock": 1,
  "task": "task-42",
  "kind": "add_requirement",
  "value": {
    "id": "retry-idempotency-test",
    "text": "Verify that replaying the same payment retry key creates one charge."
  }
}

提交并发布 Beta 的引用：

git -C .agent-beta add ops/beta/beta-task-42-1.json
git -C .agent-beta commit -m "Beta adds retry verification requirement"
BETA_COMMIT=$(git -C .agent-beta rev-parse HEAD)
git update-ref refs/agents/beta/task-42 \
  "$BETA_COMMIT" \
  0000000000000000000000000000000000000000

test "$(git rev-parse refs/agents/alpha/task-42^)" = "$BASE_COMMIT"
test "$(git rev-parse refs/agents/beta/task-42^)" = "$BASE_COMMIT"

git merge-base --is-ancestor \
  refs/agents/alpha/task-42 \
  refs/agents/beta/task-42 && exit 1 || true

git merge-base --is-ancestor \
  refs/agents/beta/task-42 \
  refs/agents/alpha/task-42 && exit 1 || true

      前两个断言证明两次发布具有相同的父提交。接下来的两个断言要求任何一个 AI 智能体的提交都不能是另一个提交的祖先。

步骤 4：通过 Git 引用进行同步

      在不切换分支的情况下列出协调命名空间：

git for-each-ref \
  --format='%(refname) %(objectname)' \
  refs/agents/

      直接从相应的提交树中检查已发布的操作：

git show \
  refs/agents/alpha/task-42:ops/alpha/alpha-task-42-1.json

git show \
  refs/agents/beta/task-42:ops/beta/beta-task-42-1.json

      此时 main 尚未移动。确认已接受的工作树中不包含任何智能体操作：

test ! -e ops/alpha/alpha-task-42-1.json
test ! -e ops/beta/beta-task-42-1.json
git status --short

当智能体使用独立克隆时

      共享仓库必须显式传输自定义命名空间，因为默认 fetch 通常只跟踪 refs/heads/*。智能体可以使用以下命令推送其私有 ref：

git push origin \
  HEAD:refs/agents/alpha/task-42

协调器可以使用以下命令获取所有智能体 ref：

git fetch origin \
  '+refs/agents/*:refs/agents/*'

      开头的加号允许非快进替换，因此对于经过安全加固的部署而言权限过于宽松。应优先采用服务端策略，要求以快进方式发布，或使用不可变且特定于每一代的 ref 名称。

步骤 5：物化已接受的状态

先归约 Alpha，再归约 Beta

python3 tools/reduce.py \
  --ref refs/agents/alpha/task-42 \
  --ref refs/agents/beta/task-42 \
  --output /tmp/task-ab.json \
  --print-digest

先归约 Beta，再归约 Alpha

python3 tools/reduce.py \
  --ref refs/agents/beta/task-42 \
  --ref refs/agents/alpha/task-42 \
  --output /tmp/task-ba.json \
  --print-digest

cmp /tmp/task-ab.json /tmp/task-ba.json

      cmp 必须成功退出。这里检查的是逐字节相等，而不仅仅是语义相似。

检查物化后的任务

python3 -m json.tool /tmp/task-ab.json

生成的文档应具有以下结构：

{
  "id": "task-42",
  "requirements": [
    {
      "id": "retry-idempotency-test",
      "text": "Verify that replaying the same payment retry key creates one charge."
    }
  ],
  "status": "in_progress",
  "title": "Prevent duplicate payment retries"
}

      这并非宣称得到的基准测试结果，而是由两个操作固件所决定的确定性预期状态：Alpha 提供胜出的状态操作，而 Beta 贡献唯一的需求。

仅提交生成的状态

mkdir -p tasks
cp /tmp/task-ab.json tasks/task-42.json
git add tasks/task-42.json
git commit -m "Materialize coordinated task-42 state"

      这个普通提交只有一个父提交：此前的 main 提交。它不会合并任何一个私有智能体 ref。已接受的分支包含协调器选定的结果，而这些 ref 则保留重新生成该结果所需的来源信息。

步骤 6：证明能够跨会话恢复

      对话记录和工作树目录被有意排除在恢复契约之外。确认临时工作树处于干净状态后，将其删除：

git -C .agent-alpha status --porcelain
git -C .agent-beta status --porcelain
git worktree remove .agent-alpha
git worktree remove .agent-beta

      在仓库中新建一个 shell 会话。不要恢复 ALPHA_COMMIT、BETA_COMMIT，也不要恢复任何智能体提示词。根据持久状态重新构建：

git for-each-ref --format='%(refname)' refs/agents/

python3 tools/reduce.py \
  --ref refs/agents/alpha/task-42 \
  --ref refs/agents/beta/task-42 \
  --output /tmp/task-recovered.json

cmp tasks/task-42.json /tmp/task-recovered.json

      比较成功即证明物化状态不依赖任何临时检出目录或内存中的会话上下文。由于自定义 ref 使相应提交保持可达，Git 会保留这些操作。

验证协议

      逐项运行以下检查。命令无输出且退出状态为零，即表示断言通过。

git show-ref --verify --quiet refs/agents/alpha/task-42
git show-ref --verify --quiet refs/agents/beta/task-42

2. 两个操作均可达

git cat-file -e \
  refs/agents/alpha/task-42:ops/alpha/alpha-task-42-1.json

git cat-file -e \
  refs/agents/beta/task-42:ops/beta/beta-task-42-1.json

3. 交付顺序不影响结果

python3 tools/reduce.py \
  --ref refs/agents/alpha/task-42 \
  --ref refs/agents/beta/task-42 \
  --output /tmp/order-one.json

python3 tools/reduce.py \
  --ref refs/agents/beta/task-42 \
  --ref refs/agents/alpha/task-42 \
  --output /tmp/order-two.json

cmp /tmp/order-one.json /tmp/order-two.json

4. 重复交付不会造成影响

python3 tools/reduce.py \
  --ref refs/agents/alpha/task-42 \
  --ref refs/agents/alpha/task-42 \
  --ref refs/agents/beta/task-42 \
  --output /tmp/duplicate.json

cmp /tmp/order-one.json /tmp/duplicate.json

5. 两项逻辑变更均得到保留

python3 - <<'PY'
import json
from pathlib import Path

task = json.loads(Path("tasks/task-42.json").read_text())
assert task["status"] == "in_progress"
assert {
    item["id"] for item in task["requirements"]
} == {"retry-idempotency-test"}
PY

6. 工作分支中不存在智能体合并

test "$(git rev-list --parents -n 1 HEAD | wc -w)" -eq 2
test "$(git rev-list --merges "$BASE_COMMIT"..HEAD | wc -l)" -eq 0

      对于普通的单父提交，第一个命令的输出包含两个单词：提交本身及其父提交。此检查假设 BASE_COMMIT 之后的实验历史中仅包含物化提交。

7. 已提交的状态可重现

python3 tools/reduce.py \
  --ref refs/agents/alpha/task-42 \
  --ref refs/agents/beta/task-42 \
  --output /tmp/rebuilt.json

cmp tasks/task-42.json /tmp/rebuilt.json
git status --short

记录证据，而非预先写好的结论

        保存你自己运行时的 Git 版本、Python 版本、基础提交、ref 对象标识符、归约器提交、命令、退出状态和生成的摘要。在你的环境中所有断言全部成功之前，不要声称该实验已经通过。

连接真实的编程智能体

      上述 shell 命令以确定性的方式模拟智能体，从而无需凭据或模型提供商即可测试协调机制。使用真实智能体替换这些固件时，应将其输出限制为相同的操作模式。