# Testing multi-agent crews: forcing split-brain, goal drift, and context drift to happen in CI > Source: > Published: 2026-10-30 16:37:33+00:00 *Originally published on [Loop & Retry](https://loopandretry.github.io/posts/testing-multi-agent-coordination-failures/?ref=devto) — field notes on building LLM agents that survive production.* The multi-agent posts on this blog have covered how a crew breaks ([split-brain](https://loopandretry.github.io/posts/multi-agent-memory-split-brain/?ref=devto), [goal drift](https://loopandretry.github.io/posts/multi-agent-goal-drift/?ref=devto), [context drift](https://loopandretry.github.io/posts/multi-agent-context-drift/?ref=devto)), how to see it ([observability](https://loopandretry.github.io/posts/multi-agent-observability/?ref=devto)), and how to prevent it at the boundary ([tool design](https://loopandretry.github.io/posts/multi-agent-tool-design-for-coordination/?ref=devto)). 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. 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: 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. 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. ``` python 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: ``` python 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. 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. ``` python 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: ``` php 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. 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: ``` python 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. 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: ``` python 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: ``` python 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. **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](https://loopandretry.github.io/posts/evaluating-your-evals/?ref=devto) 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. 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.