多智能體協作測試：在 CI 中刻意重現腦裂、目標漂移與上下文漂移

Originally published on Loop & Retry — field notes on building LLM agents that survive production.

The multi-agent posts on this blog have covered how a crew breaks (split-brain, goal drift, context drift), how to see it (observability), and how to prevent it at the boundary (tool design). One question is left: how do you know your prevention works before a customer finds out? This post is about writing tests that make each failure happen on purpose, so a regression in your tool boundary fails a build instead of an incident review.

Why your suite is green while your crew is broken

Run a three-agent crew on a happy-path task a hundred times and you will almost never see a coordination failure. That isn't evidence the crew is safe. It's evidence the test schedules agents in a friendly order.

Each failure needs a specific precondition:

• Split-brain needs two agents to read the same key and then both write. Run them sequentially, or let one finish before the next starts, and the second always reads fresh state.
• Goal drift needs several lossy handoffs in a row. A two-hop task rarely drifts far enough to matter.
• Context drift needs a correction to arrive while an agent is mid-task holding the old version. If the objective never changes during a test, nothing can go stale.

So the first rule is that coordination tests cannot be sampled. You don't run the crew and hope the bad interleaving turns up. You construct it. That also means a coordination test almost never needs a real model, and I'd argue it shouldn't use one.

Assert on the state, not on the prose

A model call in the loop makes a test nondeterministic in exactly the place you're trying to pin down. And the output text is the wrong thing to assert on anyway. The question a coordination test asks is not "did the final answer look right?" It's "did the system permit a state it should have refused?" That is a property of the tool layer and the shared store, both of which are ordinary deterministic code.

So replace the model with a scripted agent: a list of tool calls it will make, in order, with no reasoning between them.

from dataclasses import dataclass, field
from typing import Callable

@dataclass
class ScriptedAgent:
    name: str
    steps: list[Callable[["Harness"], dict]]   # each step makes ONE tool call
    cursor: int = 0
    log: list[dict] = field(default_factory=list)

    def done(self) -> bool:
        return self.cursor >= len(self.steps)

    def advance(self, h: "Harness") -> None:
        result = self.steps[self.cursor](h)
        self.log.append({"agent": self.name, "step": self.cursor, "result": result})
        self.cursor += 1

The harness owns the clock. Instead of letting agents run, it takes an explicit schedule, a list of agent names saying whose step runs next:

class Harness:
    def __init__(self, tools, agents: list[ScriptedAgent]):
        self.tools = tools
        self.agents = {a.name: a for a in agents}
        self.scratch: dict = {}   # values an agent read, for a later step to use

    def run(self, schedule: list[str]) -> None:
        for name in schedule:
            agent = self.agents[name]
            assert not agent.done(), f"schedule over-runs {name}"
            agent.advance(self)
        leftover = [n for n, a in self.agents.items() if not a.done()]
        assert not leftover, f"schedule under-runs {leftover}"

Those two asserts matter more than they look. A schedule that silently skips a step produces a test that passes for the wrong reason, and a green test that never exercised the interleaving is worse than no test.

Test 1: split-brain is an interleaving, so enumerate it

Here is the failure from the split-brain post, written as a test against the versioned read_state / write_state pair from the tool-design post. Two agents each read, then each write.

def agent(name, key, new_value):
    def read(h):
        r = h.tools.read_state(key)
        h.scratch[(name, "read")] = r
        return r
    def write(h):
        r = h.scratch[(name, "read")]
        return h.tools.write_state(key, new_value, expected_version=r["version"])
    return ScriptedAgent(name, [read, write])

def test_concurrent_writers_one_is_rejected(store_tools):
    a, b = agent("A", "plan", "ship friday"), agent("B", "plan", "ship monday")
    h = Harness(store_tools, [a, b])

    h.run(["A", "B", "A", "B"])          # both read, then both write

    results = {x.name: x.log[-1]["result"] for x in (a, b)}
    assert results["A"].get("rejected") is not True
    assert results["B"]["rejected"] is True
    assert store_tools.read_state("plan")["value"] == "ship friday"

The schedule ["A", "B", "A", "B"] is the whole test. It is the interleaving that a sequential run never produces. Two things I'd add before calling this done.

Enumerate, don't pick one. For two agents with two steps each there are only six distinct schedules. Generate all of them and assert the invariant on each:

from itertools import permutations

def all_schedules(agents: dict[str, int]) -> set[tuple[str, ...]]:
    # agents maps name -> number of steps; returns every distinct interleaving
    flat = [n for n, k in agents.items() for _ in range(k)]
    return set(permutations(flat))

def test_no_schedule_leaves_a_lost_update(make_store_tools):
    for sched in all_schedules({"A": 2, "B": 2}):
        tools = make_store_tools()
        a, b = agent("A", "plan", "x"), agent("B", "plan", "y")
        Harness(tools, [a, b]).run(list(sched))
        final = tools.read_state("plan")["value"]
        committed = [x.name for x in (a, b)
                     if not x.log[-1]["result"].get("rejected")]
        # The invariant: the stored value belongs to the last COMMITTED write,
        # and every non-committed writer was told so.
        assert final == {"A": "x", "B": "y"}[committed[-1]], sched

Notice what's asserted. It isn't "the result is ship friday." It's an invariant that holds in every schedule: whatever is in the store was written by an agent that was told its write succeeded. That is the definition of no lost update, and it survives you changing the agents, the values, and the order. With three agents and three steps the count is 1,680, which is still milliseconds because nothing is calling a model. Past that, sample schedules with a fixed seed, and print the seed on failure so any red run can be replayed exactly.

Add a positive control. Before you trust this test, run it against the BAD signature, the one with no expected_version. It must fail. If your test passes against the broken tool, it isn't testing what you think, and you've learned that before shipping rather than after.

Test 2: goal drift is a ledger property, not an output property

Goal drift is the hardest of the three to test through behavior, because the symptom is a plausible answer to a subtly different question. You can't write an assert on "plausible." But the tool-design post gave you something checkable: every constraint on the origin task must be carried verbatim or waived with a reason. That is a partition, and a partition can be asserted.

The test drives a handoff chain with scripted paraphrase pressure. Each hop's script tries to drop or reword a constraint, the way a summarizing agent does, and the assertion is that the tool refuses.

ORIGIN = [
    {"id": "c1", "text": "Do not change the public API response shape."},
    {"id": "c2", "text": "No new network calls from the hot path."},
]

def lossy_hop(drop=None, reword=None):
    def step(h):
        carried = [dict(c) for c in ORIGIN if c["id"] != drop]
        if reword:
            for c in carried:
                if c["id"] == reword:
                    c["text"] = "Keep the API mostly stable."   # a paraphrase
        return h.tools.delegate(
            subtask="implement cache layer", origin_task_id="t-1",
            constraints=carried, acceptance_criteria=[], waived_constraints=[],
        )
    return step

def test_silent_drop_is_rejected(delegate_tools):
    a = ScriptedAgent("planner", [lossy_hop(drop="c2")])
    Harness(delegate_tools, [a]).run(["planner"])
    r = a.log[0]["result"]
    assert r["rejected"] and "c2" in r["error"]

def test_paraphrase_is_rejected(delegate_tools):
    a = ScriptedAgent("planner", [lossy_hop(reword="c1")])
    Harness(delegate_tools, [a]).run(["planner"])
    assert a.log[0]["result"]["rejected"]

The second test is the one people skip, and it's the interesting one. A completeness check that only compares constraint ids passes a reworded constraint without a murmur, because the id is still there. The tool has to compare the text against the origin ledger byte for byte. If test_paraphrase_is_rejected passes on your first try, I'd go look for why, because most first implementations of this check fail it.

Then test the chain, not just the hop. Drift compounds, so run five hops where each one carries the previous hop's output, and assert on the final ledger:

def test_five_hop_chain_preserves_ledger(delegate_tools):
    hops = [ScriptedAgent(f"hop{i}", [carry_forward_verbatim]) for i in range(5)]
    h = Harness(delegate_tools, hops)
    h.run([f"hop{i}" for i in range(5)])
    final = delegate_tools.ledger("t-1")
    assert [c["text"] for c in final.constraints] == [c["text"] for c in ORIGIN]

An assertion on the ledger is also the cheap version of what the observability post called a constraint-survival probe, with the difference that it runs in CI with no model in the path.

Test 3: context drift needs a correction injected mid-run

For context drift the test has to change the world while an agent is holding the old copy. The harness needs one more primitive, an event that fires between steps:

class Harness(Harness):
    def run(self, schedule, events: dict[int, Callable] | None = None):
        events = events or {}
        for i, name in enumerate(schedule):
            if i in events:
                events[i](self)            # runs BEFORE step i
            self.agents[name].advance(self)

Now the $32-credit scenario from the context-drift post becomes a deterministic test. A responder reads the refund policy, the policy is corrected, and the responder sends a message drafted against the old version:

def test_send_against_superseded_policy_is_refused(support_tools):
    def read_policy(h):
        r = h.tools.get_objective("ticket-9", agent_id="responder")
        h.scratch["policy"] = r
        return r
    def send(h):
        return h.tools.send_customer_message(
            "ticket-9", "We've credited $32.",
            policy_version=h.scratch["policy"]["version"])

    responder = ScriptedAgent("responder", [read_policy, send])
    h = Harness(support_tools, [responder])

    h.run(["responder", "responder"],
          events={1: lambda h: h.tools.correct_objective("ticket-9", "Credits capped at $20.")})

    result = responder.log[-1]["result"]
    assert result["rejected"] is True
    assert "capped at $20" in result["current_policy"]
    assert support_tools.outbox == []          # nothing reached the customer

The last assert is the one that carries the test. Not "the tool returned a rejection," but "the outbox is empty." A rejection that something downstream ignores is still a leak, and checking the side-effect channel directly is what makes this test hard to fool.

A second test covers the other half of the design: correction fan-out. Two agents read, a third corrects, and the test asserts the holder set contains exactly the two readers. A holder set that misses someone is a stale agent nobody will ever notify.

What to keep out of the suite

Don't assert on model output in the coordination tests. There's a place for evals on whether agents choose well under contention, and evaluating your evals is the caution about how easily those mislead. But the tool-boundary guarantee is deterministic, so test it deterministically. A flaky test that sometimes passes on a real model teaches the team to rerun red builds.

Don't treat schedule coverage as a vanity number. "We test 6 interleavings" means little. The useful question is whether there is a schedule that violates your invariant and isn't in your set. For small N, enumerate. For large N, seed and sample, and keep the seeds of anything that ever failed as permanent regression cases.

Don't let the harness diverge from production tools. The scripted agent should call the real tool functions against a real (in-memory) store, not a mock that returns what you expect. A mock of write_state that implements the version check is a test of your mock. If the production store is a remote service, run the same suite against a local instance of it nightly.

Pair it with the production signal

These tests prove the guard exists. They don't prove it's the only path. The production counterpart is the rejection rate from the instrumentation post: a rising count of rejected writes means contention is growing, and a count stuck at exactly zero for a crew that has several writers is suspicious, not reassuring. Zero rejections can mean perfect coordination, or it can mean a code path that bypasses the guarded tool. A CI test can't tell you which. A metric that was nonzero last month and is zero now can.

The ordering I'd follow: write the invariant first (no lost update, ledger is a partition, nothing is sent on a superseded basis), write the schedule that violates it, confirm the test fails against the unguarded tool, then confirm it passes against the guarded one. A coordination fix without a test that failed first is a fix you can't tell from a coincidence, and a crew's failure modes are exactly the kind that look like coincidence until they don't.