五个编码 Agent、五个分支、一个数据库：给每个任务一个自己的模拟器

10 月 28 日，Azure Cosmos DB 博客发了一篇由 Jay Gordon 署名的文章，讲的不是怎么写代码，而是并行编码 Agent 的一种失败方式。场景很具体：几个 Agent 同时改同一个应用，一个在改调度行为，一个在加仓库规则，一个在更新测试；每个都有自己的分支，甚至自己的 git worktree，但他们都连同一个开发数据库。于是 A 重置夹具的时候，B 正在断言一次写入；某个测试改了一份文档，另一个任务却期待它保持不变；一个清理脚本把库删了，而另一套测试还在跑。文章用一句话把问题钉住了：代码是隔离的，状态不是。

分支之间不打架，是因为它们真的不共享目录。数据没有这个特权。

一、Cosmos DB 没有迁移脚本，不代表没有共享状态

文章先排除了一个常见的安慰：Cosmos DB 不像关系型数据库那样需要表结构迁移，但这并不等于共享状态的碰撞会消失。应用依赖的东西依然存在——文档的形状、分区键的选择、索引策略、以及测试数据的可预期性。搬掉 schema 迁移只解决了一半问题，剩下那一半是「同一份数据被两个任务同时当成自己的」。这句话对任何用文档库做后端的团队都成立：你省掉的是迁移，不是并发。

# Spawn an emulator that owns its lifetime, its ports and its data.
# The article's helper does exactly this; the shape is worth copying.
import docker, socket, time

client = docker.from_env()

def port():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p

def start_env(task_id: str):
    db_port, ready_port = port(), port()
    c = client.containers.run(
        "mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:latest",
        detach=True,
        name=f"cosmos-{task_id}",          # invocation-owned name
        ports={"8081/tcp": ("127.0.0.1", db_port),
               "8080/tcp": ("127.0.0.1", ready_port)},
        labels={"owner": "agent-task", "task": task_id},
        environment={"AZURE_COSMOS_EMULATOR_PARTITION_COUNT": "3"},
    )
    wait_ready(ready_port, timeout=180)
    return {"id": c.id, "endpoint": f"https://127.0.0.1:{db_port}"}

# Internal ports stay 8081 and 8080. Only the host side is dynamic,
# bound to loopback, so two tasks never fight over 8081.

二、隔离单位是任务，不是分支

文章提出的做法可以概括成一句话：给每个编码任务一个可丢弃的数据环境——独立 worktree、独立模拟器、已知夹具、独立测试、自己负责清理。它借用的是 git worktree 的隔离模型，但明确声明只借模型不借机制：不创建写时复制快照，不克隆生产账号，也不做数据库内容的合并。每个环境从同一份版本化的合成夹具开始，可以独立修改，可以整体丢弃。更妙的是那个对照实验的设计：两个环境故意使用完全相同的身份——同一个数据库名 fleetchargetest-agent-isolation、同一个容器 dispatchUpdates、同一个文档 ID current-dispatch、同一个分区键 SEA-01、同一个起始值 BAY-2，然后 A 改成 BAY-7、B 改成 BAY-8。如果身份不同，「没撞上」可能只是运气；身份完全相同，还能各自读到自己的值，才证明边界真的落在模拟器实例上。

# Status codes lie. Read the stored item back through an independent client.
from azure.cosmos.aio import CosmosClient

async def verify(env_a, env_b):
    async with CosmosClient(env_a["endpoint"], credential=KEY) as a, \
               CosmosClient(env_b["endpoint"], credential=KEY) as b:
        ca = a.get_database_client("fleetchargetest-agent-isolation") \
              .get_container_client("dispatchUpdates")
        cb = b.get_database_client("fleetchargetest-agent-isolation") \
              .get_container_client("dispatchUpdates")

        # Same database, same container, same document id, same partition key.
        doc_a = await ca.read_item("current-dispatch", partition_key="SEA-01")
        doc_b = await cb.read_item("current-dispatch", partition_key="SEA-01")
        assert doc_a["pickupBay"] == "BAY-7" and doc_a["updateRef"] == "agent-a-dispatch"
        assert doc_b["pickupBay"] == "BAY-8" and doc_b["updateRef"] == "agent-b-dispatch"

        # A control document that must never move.
        for c in (ca, cb):
            pdx = await c.read_item("current-dispatch", partition_key="PDX-02")
            assert pdx["pickupBay"] == "BAY-99"
            assert pdx["updateRef"] == "dispatch-099"

# Deliberately reusing the same identities is the point. Different ids
# would let a leak pass unnoticed; identical ids force the boundary to be
# the emulator instance, which is what you actually want to test.

三、验证存储状态，而不是验证请求成功

文章里最实用的一条工程纪律是：不要用「请求返回 200」当作通过。做法是两次写入完成后，用独立的 SDK 客户端对同一个 item 做一次点读（read_item，带同一个分区键），不仅校验 ID，还校验仓库、取货位和更新引用；同时读一份对照文档，确认它在两个环境里都没有被动过。更进一步，它验证生命周期的边界：把 A 删掉，确认它确实不在了，再用全新客户端读 B，B 必须还带着自己的更新。这一条特别值得抄，因为它把「清理」从「希望脚本没跑错」变成了可断言的测试——很多团队的数据污染就是这么进来的：某个收尾脚本删多了，而没有任何测试会发现。

两个模拟器，内部端口都是 8081/8080，宿主机端口由 Docker 分配。

# The runner hands each test subprocess its own configuration.
# It never rewrites a shared .env or mutates process globals.
import os, subprocess, sys

TASKS = {
    "dispatch":   ["tests/unit/test_dispatch.py", "tests/integration/test_feed.py"],
    "fleet-rules": ["tests/unit/test_fleet_rules.py", "tests/integration/test_delivery_wave.py"],
}

def run(task: str):
    env = start_env(task)                       # owned container, dynamic ports
    try:
        seed_fixtures(env["endpoint"])         # versioned synthetic fixtures
        for path in TASKS[task]:
            code = subprocess.run(
                [sys.executable, "-m", "pytest", path, "-q"],
                env={**os.environ,
                     "COSMOS_ENDPOINT": env["endpoint"],
                     "COSMOS_KEY": TEST_KEY,
                     "COSMOS_READY_URL": env["ready_url"]},
            ).returncode
            if code != 0:
                raise SystemExit(code)
        verify_stored_state(env)
    finally:
        cleanup_owned(env["id"])                # only what we created

# Same command locally and in CI: python -m scripts.agent_ci --task dispatch
# An agent can therefore pre-flight exactly what CI will run.

四、CI 里也要一任务一模拟器

本地隔离只是一半。文章把同样的模式搬进了 GitHub Actions：新增的 Agent task CI 工作流用一个矩阵，两个任务预设，dispatch 跑调度单元测试加 feed 集成测试，fleet-rules 跑仓库规则单元测试加配送波次工具集成测试。每个矩阵条目拿到自己的 Ubuntu runner，并启动一个属于自己的模拟器，宿主端口动态分配、绑定回环地址；模拟器内部的 8081（数据库）和 8080（就绪）保持不变。这里有两个细节很容易做错。第一，配置必须显式传给子进程——文章明确说示例没有改写共享的 .env，也没有改动进程级环境变量；扩到编码 Agent 启动器时，这条规则更重要，因为多个子进程共享父进程环境是隐形的串台。第二，矩阵的 fail-fast 应该关掉，否则第一个任务红了会取消其余任务，你就失去了「这是任务级问题还是环境级问题」的判断依据。

# .github/workflows/agentic-ci.yml
# One matrix entry per task preset. Each gets its own runner and emulator.
name: Agent task CI
on:
  pull_request:
  workflow_dispatch:

jobs:
  task:
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        task: [dispatch, fleet-rules]
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - name: Run isolated task checks
        run: uv run --locked --extra test python -m scripts.agent_ci --task ${{ matrix.task }}
      - name: Upload environment report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: agent-task-${{ matrix.task }}
          path: artifacts/${{ matrix.task }}/

# fail-fast: false matters more than it looks. With fail-fast on, the
# first red task cancels the rest and you lose the signal about whether
# the failure is task-specific or environmental.

五、失败必须长得像失败

最后一节是整篇文章里最像老工程师写的部分：读回断言失败时，进程必须以非零状态退出，先清理自己拥有的容器，并且不产出成功报告。文章的原话是，一次失败的断言绝不能被当成一次成功的环境运行。落地到代码上就是三件事：诊断信息在清理之前抓取（崩掉的模拟器也欠你一份日志）、清理之后要断言容器确实不存在、以及 JSON 结果报告只在「验证 + 清理」都完成之后才写。每个调用还有自己独立的日志目录和一个被 git 忽略的 artifacts 目录。这套规矩听起来啰嗦，但它解决的是一个特别贵的问题——当你有一堆并行的编码 Agent，最危险的不是报错，而是安静地通过。

每个矩阵条目一条独立 runner，一个自己的模拟器。

# Failure must look like failure: nonzero exit, cleanup, no report.
import json, pathlib, shutil

def finalize(task, env, ok: bool, detail: dict):
    art = pathlib.Path("artifacts") / task
    art.mkdir(parents=True, exist_ok=True)
    try:
        if not ok:
            # Diagnostics are captured BEFORE cleanup; a crashed emulator
            # still owes you its logs.
            (art / "diagnostics.log").write_text(collect_diagnostics(env))
            (art / "failure.json").write_text(json.dumps(detail, indent=2))
            raise SystemExit(2)          # never exit 0 on a failed assertion
        cleanup_owned(env["id"])
        gone = assert_container_absent(env["id"])
        (art / "result.json").write_text(json.dumps(
            {**detail, "cleanup_verified": gone}, indent=2))
    finally:
        if art.exists() and not (art / "result.json").exists():
            shutil.rmtree(art, ignore_errors=True)

# The rule from the article, restated as code: the JSON result report is
# written only after verification AND cleanup. A failed assertion must not
# look like a successful environment run.