TwinEdgeAI

↑ Course home

← Previous: Week 4b, Agent Memory Architectures  ·  Next: Week 5b, Optimization →

Engineering AI Agents · Week 5a

Multi-Agent Systems

When one agent is enough and when it is not, four ways to organize a team, how agents talk (conversations, handoffs, message boards, the A2A protocol), how they stay organized (ledgers, shared memory, isolated workspaces), and the many ways teams fail.

Author: Grok Bot Version: 1.0, October 28, 2026 Study time: about 14 hours

CreditThis coursework is adapted from, and inspired by, Stanford CS329Z: Engineering AI Agents (Diyi Yang, Michael Ryan, John Yang), Week 5, "Multi-Agent Systems" (Monday, October 19). The topic list and the readings come from the course's public schedule [30]. I checked the topic order against the public lecture slides [31]. I used the slides for coverage only. The notes, lab, quiz and assignment are original writing, and the lab uses its own examples and code. This pack does not reproduce the slides. It is not an official Stanford document. I added extra sources of my own. Each is marked "added" in the references.

1. Week overview and learning objectives

So far you have built one agent: a loop that thinks, calls tools, and remembers. This week you put several of them to work on one task. The lecture opens with the claim that multi-agent systems are "emergent" and making headlines, and that the goal is to break down what is known about how to build them [31]. The lecture's three learning objectives are to tell single and multi-agent systems apart, to name the orchestration, communication and coordination patterns, and to compare how teams fail [31].

The week has six parts. First, when to use several agents, and what a team costs. Second, four orchestration patterns: independent, centralized, decentralized and hybrid. Third, communication: conversations, handoffs, agents used as tools, message boards, and the A2A protocol for talking across systems. Fourth, coordination: the task ledger, shared memory, and isolated workspaces, plus three patterns that teams discover on their own. Fifth, failures: a taxonomy from 1,600 traces, a benchmark where two agents do worse than one, and some odd behaviors that appear only in groups. Sixth, a lab where you build all of it.

The lab uses the standard library only. Mock models stand in for the LLM, so you need no API key. The mocks follow scripts and rules that I wrote. They show how each design behaves. They say nothing about how well a real model does. The cost and accuracy comparison in the lab uses made-up numbers, and its results follow from my assumptions. I flag this again where it matters.

Learning objectives

After this week you should be able to:

1. Say what a multi-agent system is, and place it on the scale from a single model call to a team. [2.1]
2. List the benefits and the costs of a team, and say when a single agent should win, using the evidence in the Kim et al. scaling study. [2.2]
3. Describe independent, centralized, decentralized and hybrid orchestration, their trade-offs, and the idea of a dynamic team. [2.3]
4. Tell a handoff from an agent used as a tool, and say what the receiving agent sees in each case. [2.4]
5. Explain a message board with topics and schemas, and a task board with atomic claims. [2.5]
6. Read an A2A Agent Card, a task, and a JSON-RPC request, and name the task states and the main error codes. [2.6]
7. Explain the task ledger and the progress ledger of Magentic-One, and the stall counter that triggers a re-plan. [2.7]
8. Explain why shared state needs a lock or a version check, and what a team memory stores. [2.8]
9. Explain worktree isolation, a three-way merge, and why "no conflict" does not mean "works". [2.9]
10. Name the three emergent coordination patterns, and the 14 failure modes of MAST in their three groups. [2.10, 2.11]
11. Do a simple cost and accuracy estimate for a single agent and a team, and say which assumptions drive the answer. [2.13]
12. Build and test the designs in the lab: handoffs and orchestration, boards, A2A messages, ledgers, shared memory with a lock, isolated workspaces with a merge, a failure classifier, and a cost comparison. [Lab]

How to use this pack

Time	What to do
3 hours	Read the notes (section 2). Skim the figures first.
4 hours	Read the three required papers with the reading guide questions. Skim the optional ones that match your project.
3.5 hours	Do the lab and at least five exercises.
1.5 hours	Discussion and the quiz.
2 hours	The graded assignment.

2. Lecture notes

2.1 One agent or several?

Think of system complexity as a scale. At the simple end is one model call. Next come a workflow (fixed steps, several calls) and a single agent (one model in a loop with tools). The lecture calls these together compound AI systems. At the complex end is a multi-agent system, and the slide's examples include triaging a bug, implementing a feature and maintaining a repository, as tasks of growing size [31]. Figure 1 shows the scale.

The lecture's working definition is short: a multi-agent system (MAS) is a collection of agents oriented to solve a shared task [31]. Three design questions follow from it, and they give the structure of this week:

• Orchestration. Who is in charge, and how is the team shaped? (Section 2.3.)
• Communication. How do agents exchange information? Who receives what, through which channel, in which format? (Sections 2.4 to 2.6.)
• Coordination. How do agents manage interdependent work? Who owns each task, what must happen first, and how are conflicts settled? (Sections 2.7 to 2.10.)

The lecture keeps the last two apart, and so will I. Communication is about information. Coordination is about work that depends on other work [31]. A team can talk a lot and still coordinate badly, which is a theme of the failure section.

Where the idea came from

Several early systems explored teams of language-model agents, and they are still the vocabulary of the field. I list them with what each one added. For most of these I read the abstract and the parts I cite, as the references say.

• Multiagent debate. Several model instances propose answers and then debate over rounds to reach a common answer. The paper reports better math and strategic reasoning and fewer false facts, using the same prompts for every task [7].
• CAMEL. Two chat agents play assigned roles and keep each other on task. The paper calls the way it sets up the roles inception prompting [6].
• AutoGen. A framework in which customizable, "conversable" agents talk to one another and to humans and tools. You program the conversation patterns in natural language or in code [1]. This is required reading 1.
• MetaGPT and ChatDev. Software-company simulations. MetaGPT writes standard operating procedures (SOPs) into the prompts and has agents pass structured documents [5]. ChatDev guides agents in what to say (a "chat chain" of phases) and how to say it (a step it calls communicative dehallucination) [8].
• AgentVerse. A framework for groups of expert agents that also studies the behaviors that appear in groups [9].
• More Agents Is All You Need. The simplest team of all: sample the same model many times and vote. The paper reports that performance grows with the number of agents, and that the gain depends on how hard the task is [11].

Look at what they share. Most of them assume that more voices help. The rest of the week is about when that is true.

2.2 Benefits, costs, and what the evidence says

The lecture gives four benefits [31]:

• Parallel execution. Separable subtasks run at the same time.
• Specialization. Each agent gets its own instructions, model and tools.
• Focused context. Each agent sees only what its job needs, so its context stays short.
• Broader exploration. Different agents can look at different sources, guesses or approaches.

And five costs [31]:

• Coordination overhead. Delegating, messaging and merging use time and tokens.
• Information loss. A handoff can leave out something that the next agent needs.
• Conflicting actions. Agents can repeat work, break promises, or overwrite each other.
• Error propagation. One agent's mistake can mislead the others.
• Engineering complexity. More roles, prompts and interfaces to maintain and debug.

The lecture's rule is one sentence: collaboration helps when the benefits outweigh the coordination costs [31]. Its checklist for the choice, in my words:

Prefer one agent when…	Try a team when…
Steps depend closely on one another.	Subtasks can go on with little synchronization.
The work is tied to one shared state, with a lot of overlap.	The information splits into focused pieces of context.
One agent already meets the quality, cost and speed goals.	Agents can own separate pieces of the result, and someone can put them together.
	Parallel work, specialists or a separate reviewer fix a bottleneck that you have measured.

The scaling study (required reading 3)

Kim et al. tested the folk wisdom. They ran 260 configurations: six benchmarks, five architectures (a single agent, plus Independent, Centralized, Decentralized and Hybrid teams), and three model families, with the tools, prompts and token budgets held the same so that the architecture was the thing that changed [3]. This is the version 3 of the paper (April 2026). An earlier version had 180 configurations and a different fit, so cite the numbers with the version [3]. The authors work at Google, and they used models from several vendors.

• The spread is huge. Compared with a single agent, the best team changed the score by +80.8% on a decomposable financial-reasoning task, and the worst by −70.0% on a sequential planning task (PlanCraft, a Minecraft crafting benchmark) [3].
• A capability ceiling. Once the single agent is already good (around 45% accuracy in their data), adding coordination gives diminishing returns [3].
• Tool-heavy tasks pay more overhead. The abstract says these tasks "appear to incur" multi-agent overhead [3].
• Verification matters. Teams without a central check spread errors more: the paper measures an error amplification of 17.2 times for Independent teams, against 4.4 times for Centralized ones [3].
• On PlanCraft every team design lost, by 39% to 70%, while on web navigation the best design (Decentralized) gained 9.2% [3].
• The fit is moderate. The paper's model predicts the best architecture for 87% of held-out configurations, with a cross-validated R² of 0.373 (0.413 with a task-grounded metric) [3]. That is useful, not a law.

Two vendors, two opinions

Two engineering posts, both from companies that sell agents, sit on opposite sides. Treat both as opinion with some data, not as findings.

• Anthropic's research system uses a lead agent that spawns sub-agents. On an internal research evaluation, a Claude Opus 4 lead with Claude Sonnet 4 sub-agents beat single-agent Claude Opus 4 by 90.2%. The post says such systems work mainly because they spend enough tokens: token use alone explains 80% of the variance on the BrowseComp benchmark. The cost is real: agents use about 4 times the tokens of chat, and multi-agent systems about 15 times. The same post says that most coding tasks have less parallel work than research, and that agents are "not yet great" at delegating in real time [14]. This is Anthropic's own evaluation and its own product.
• Cognition's "Don't Build Multi-Agents" (June 2025) argues for one agent in a line. Its first principle is to share context, meaning the full agent traces and not only individual messages. Its second is that actions carry implicit decisions, and conflicting decisions give bad results. Its example: one sub-agent builds a Super Mario style background for a "Flappy Bird clone" while another builds a bird that does not fit the game. The post recommends a single-threaded agent, with a model that compresses a long history when needed [15]. This is one company's view, written for its own coding agent, at one date.

The two are less opposed than they look. Both say that context must reach whoever acts, and both say that parallel work needs a task that splits cleanly. A later Anthropic research post adds a distinction that I find useful: agents work well together when they treat each other as tool calls with clear inputs and outputs, and stumble when they act as long-lived peers with their own goals and no clear hierarchy [27].

2.3 Four ways to organize a team

The lecture names four orchestration patterns [31]. Kim et al. use the same four names, so you can read the two together [3]. Figure 2 draws them.

• Independent. Agents work alone, with no talk between them. A last step combines or selects among their outputs (a vote, a review). It is parallel and easy to build. But no agent can use another's findings while it works. Sampling the same model many times and voting is this pattern [11].
• Centralized (orchestrator). One orchestrator splits the task, assigns pieces to workers, and decides what happens after each result. Oversight is simple. The orchestrator can become the bottleneck, and everything it forgets to say, the workers do not know. Anthropic's research system and Magentic-One are of this kind [14] [4].
• Decentralized (swarm). Agents talk to all the others through a shared board or channel. Agents may spawn more agents. It suits broad exploration and discovery at scale [31]. The price is that nobody holds the whole plan, and it is hard to keep the work consistent.
• Hybrid. An orchestrator manages the whole task and merges results, while workers talk directly when their subtasks touch. You get global oversight and local flexibility, and you must write clear rules for who tells whom what [31].

OpenAI's practical guide to agents describes a similar split in two classes: a manager pattern, where one agent calls the others as tools, and a decentralized pattern, where agents hand off to one another as peers. In the manager pattern the manager keeps control of the user conversation. In the decentralized one, execution moves to the next agent [19]. That is the next section's topic.

LangChain's documentation sorts multi-agent designs into four patterns too: subagents, handoffs, skills and a router. It compares them by how many model calls a request takes. These are vendor docs, and the comparison is illustrative [20].

Cost goes up as control goes down

The lecture states the cost order as: single agent, then orchestrated, then swarm [31]. Why? A single agent pays for its own steps. An orchestrated team adds messages to and from the orchestrator. A swarm adds a shared board that each agent must read, and polling. In section 2.13 you will compute a small version of this, with my numbers, so that you can see which assumption drives the order.

Teams that change shape: DyLAN

A fixed team is a guess. DyLAN picks the team per task. In a first stage (team optimization), it runs a pool of candidate agents on a few queries. The agents rate one another's answers, and credit is passed back to the agents that helped reach the agreed answer. That gives each agent an Agent Importance Score, and the most useful agents are kept. In the second stage the selected team solves new queries, and it can drop agents and stop early when about two thirds of them agree [31] [10]. In the paper's abstract, picking the team improves accuracy by up to 25.0% on some MMLU subjects [10]. The lecture says DyLAN improved decision-making success in fewer API calls [31]. I read the abstract and the lecture, and not the whole paper. DyLAN shares an author with the course (Diyi Yang).

2.4 Communication: conversations, handoffs and agents as tools

The simplest way for agents to communicate is to talk. AutoGen (required reading 1) makes this the core idea: agents are conversable, and the application is a pattern of conversations [1]. The lecture lists the patterns AutoGen supports [31]:

• Sequential. One agent finishes, then hands the conversation to the next team.
• Nested. A manager agent runs a sub-conversation with its own sub-agents and gets one answer back.
• Group chat. Agents share one thread, and a rule decides who speaks next.
• Hierarchical chat. All sub-agents report to an orchestrator.

The lecture adds that the current AutoGen has classes for round-robin group chat, selector group chat and a swarm pattern [31]. I did not check those class names against the current AutoGen docs, and the 2023 paper predates them, so treat the names as the lecture's.

Handoff and agent-as-tool

Two ways to let one agent use another look alike and behave very differently. The lecture draws them from the OpenAI Agents SDK [31] [16] [17]. Figure 3 shows both.

• Handoff. Control moves. The receiving agent becomes the active one and carries on with the user. In the SDK, a handoff is a tool that the model can call, named transfer_to_<agent_name>, such as transfer_to_refund_agent. By default, the new agent sees the whole earlier conversation, and you can change that with an input filter [16]. OpenAI's guide calls a handoff a one-way transfer [19]. It fits when a specialist should talk to the user and own the next steps [31].
• Agent as tool. The caller keeps control. It sends the specialist a bounded request, with selected context, and gets an answer or artifact back. In the SDK, agent.as_tool() wraps an agent as a tool, and the parent run's conversation state is not passed on automatically [17]. It fits when one agent must put the pieces together and give the final answer [31].

The practical difference is what the specialist knows. After a handoff it sees (by default) the history, so it knows more and can be misled by more. As a tool it sees only the request, so it is cheap and focused, and it fails if the request was thin. This is the same issue that Cognition raises when it says to share full traces [15]. The lab runs the same task both ways: after the handoff, the final speaker is the billing agent. As a tool, the final speaker is the triage agent, and the billing agent saw only "Refund order 1234, $20".

2.5 Message boards and blackboards

Chat does not scale well. If every agent must talk to every other agent, the number of channels grows quickly, and every agent must read everything. A message board (the older name is a blackboard) puts shared information in one place. Agents publish to it and read from it. The lecture shows three examples [31].

• MetaGPT. A global message pool where agents publish structured messages, with a schema for each role. Agents subscribe by role, so the architect follows the product manager's requirements and does not read everything. An agent acts only after its prerequisites have arrived [5]. The paper says that communicating through documents and diagrams, and not through dialogue, helps to prevent irrelevant or missing content [5].
• Swarm-style shared state. In the lecture's decentralized pattern, all agents see a shared board or channel and may spawn more agents [31].
• MoltBook. The lecture shows it as a social-media style board for agents [31]. I did not look at the site, and I use it only as the lecture's example of an open board.

Two designs make a board useful. First, typed messages: a topic and a schema, so that a post with the wrong shape is rejected at the door, as MetaGPT's structured outputs suggest. Second, subscriptions and cursors, so that an agent reads only the new messages that match its interests. The lab's MessageBoard has both, and a direct to field for a private note.

A board for work is a different thing: a task board. Each task has an owner field and a list of tasks it depends on. An agent claims a ready task, and the claim must be atomic: if two agents claim at the same moment, exactly one wins. The lab's TaskBoard does this under a lock, checks that only the owner can complete a task, and can tell when the whole board is stuck (nothing running, nothing ready). Carlini's compiler project did the same with files (section 2.12) [24].

A board is not a cure for everything. The Anthropic multiagent post reports that when agents with no other way to coordinate shared a job queue, they quickly flooded it with polling daemons that ran 30 times per second. In one run there were 2.4 million job requests and only 117 accepted jobs [27]. Shared channels can help, but only if the agents are willing to use them well.

2.6 The A2A protocol: Agent Cards, tasks and messages

Everything so far lives inside one codebase. What if the agent you need runs in another company's system, built with another framework? The Agent2Agent (A2A) protocol is a standard for that. The lecture describes it as a protocol, like MCP, for communication between agents [31]. The A2A project's own page gives the division of labor: MCP connects an agent to tools and data (the vertical layer), and A2A connects agents to agents (the horizontal layer) [22]. The two are meant to be used together.

This section follows A2A version 1.0.1, whose specification I read, together with its protocol buffer file [21]. The proto file is the normative source, and the JSON forms follow from it. The pre-1.0 versions used different names for several things, and I did not check their details, so do not read this section as a description of older versions.

The flow

1. Discover. The client fetches the remote agent's Agent Card, a JSON file that the spec places at /.well-known/agent-card.json [21]. The lecture's wording: the card describes the agent's capabilities, skills, inputs and outputs [31].
2. Send. The client calls a method such as SendMessage. The server creates a task that can be traced.
3. Work and talk. The task moves through states. The server may send progress, and may ask the client for more input.
4. Finish. The server returns artifacts: the result of the task [21] [31].

The objects

• Agent Card. Required fields are name, description, supportedInterfaces (a list of url, protocolBinding and protocolVersion), version, capabilities, defaultInputModes, defaultOutputModes and skills. Each skill has an id, name, description and tags [21].
• Message. Has a messageId, a role (ROLE_USER or ROLE_AGENT), and parts. It may also carry contextId, taskId, and referenceTaskIds. A part holds exactly one of text, raw (bytes), url or data, with optional filename and mediaType. In version 1.0 a part has no kind field [21].
• Task. Has an id (made by the server), a contextId that groups related tasks, a status (state, optional message, timestamp), and optionally artifacts and history [21].
• Artifact. Has an artifactId, an optional name and description, and parts [21].

The task states

JSON values are strings that start with TASK_STATE_: SUBMITTED, WORKING, COMPLETED, FAILED, CANCELED, REJECTED, INPUT_REQUIRED and AUTH_REQUIRED (plus an UNSPECIFIED zero value). Four are terminal: completed, failed, canceled and rejected. A terminal task accepts no more messages, and canceling it is refused. Two are interrupted: the agent needs input or authorization, and the client answers on the same task [21].

On the wire: JSON-RPC

One of the protocol's bindings uses JSON-RPC 2.0 [23]. A request has jsonrpc, an id, a method and params. A reply has the same id and either a result or an error [23]. In A2A v1.0 the method names are in PascalCase: SendMessage, SendStreamingMessage, GetTask, ListTasks, CancelTask, SubscribeToTask and GetExtendedAgentCard. The result of SendMessage is an object with either a task or a message [21]. The protocol version travels as an A2A-Version header. Here is the first request that the lab sends:

{"jsonrpc": "2.0", "id": 1, "method": "SendMessage",
 "params": {"message": {"messageId": "msg-1", "role": "ROLE_USER",
                        "parts": [{"text": "Book me a flight"}]}}}

Errors use the standard JSON-RPC codes (−32700 parse error, −32600 invalid request, −32601 method not found, −32602 invalid params, −32603 internal error) and nine A2A codes from −32001 to −32009. They include −32001 task not found, −32002 task not cancelable, −32004 unsupported operation, and −32009 version not supported. The error.data field is a list of typed objects, such as a google.rpc.ErrorInfo with the reason TASK_NOT_FOUND and the domain a2a-protocol.org [21].

A few rules from the spec show what a server must check. The client must not invent a taskId: a task id that is sent must refer to an existing task, or the server answers with task-not-found. A contextId that does not match the task is rejected. Sending a message to a terminal task is an unsupported operation [21]. The lab's server enforces these, so you can see each error.

A2A solves format and discovery. It does not solve understanding. The MAST authors note that protocols such as MCP and A2A standardize message formats, but that the inter-agent failures they observe occur even when agents in one framework talk in plain natural language [2]. A perfectly formed message can still say the wrong thing.

2.7 Coordination: task ledgers

Communication was about how agents exchange information. Coordination is about the work: who owns which task, what must happen first, and what to do when things stall. The lecture names three kinds of coordination infrastructure, and three patterns that agents use on their own [31]. Figure 5 puts them on one page.

A task ledger is an explicit, updated account of the work. The lecture lists four jobs for it [31]:

• Track the task: the goal, the current plan, and open questions.
• Track execution: assignments, dependencies, progress and blockers.
• Guide decisions: use the current state to choose what runs next.
• Support recovery: change the plan, or retry, when execution stalls.

Magentic-One: two ledgers, two loops

Magentic-One is a generalist team from Microsoft Research. A lead agent, the Orchestrator, directs four specialists: a web browser agent (WebSurfer), a file reader (FileSurfer), a coder, and a terminal that runs code [4]. Its Orchestrator keeps two structured ledgers [4]:

• The task ledger (outer loop) holds given or verified facts, facts to look up, facts to derive, educated guesses, and a plan.
• The progress ledger (inner loop) is filled in at every step by answering five questions: Is the request fully satisfied? Is the team looping or repeating itself? Is it making forward progress? Which agent should speak next? What instruction should it get?

The Orchestrator also keeps a stall counter. When the answers show a loop or no progress, the counter goes up. While it stays below a threshold (at most 2 in the authors' experiments), the Orchestrator keeps going. When it goes over, the Orchestrator reflects, updates the task ledger with what it learned, writes a new plan, and starts a new outer-loop round [4]. The paper reports that the system is statistically competitive with the state of the art on GAIA, AssistantBench and WebArena [4]. I read the parts about the ledgers and the abstract, and not the results in detail.

Why does a ledger help? The facts and plan live in one place that is rewritten, not in a long chat that grows. A new plan after a stall carries the lesson with it. The lab's MagenticOrchestrator is a small version: when a scripted agent fails three times in a row, the ledger gets a new "given" fact ("stalled 3 times at step 4") and a new plan, and the run succeeds on the second plan.

Symphony: the issue tracker as a ledger

OpenAI's Symphony (April 27, 2026) turns an issue tracker into the ledger. A long-running service polls the tracker, creates an isolated workspace for each issue, and runs a coding agent in it. The draft spec describes dependency blocking (blocked_by), retry with back-off, and a stall timeout [25] [26]. The announcement says that some teams saw a 500% increase in landed pull requests. That is OpenAI's claim about its own tool, and I did not verify it [25].

2.8 Coordination: shared memory

If agents share a store, two problems appear at once. The first is races. Two agents read the same value, each adds one, and each writes back. One update is lost. The lab shows the worst case on purpose: four threads read a counter, wait until all four have read it, and then write. Without a lock the counter ends at 1, not 4. With a lock it ends at 4. (The wait is a barrier that I added to force the bad timing. Real races are rarer and harder to reproduce, which is what makes them nasty.)

The lab's SharedMemory offers two remedies:

• A lock around each operation, and an atomic update(key, fn) for read-modify-write.
• Versions. Each write bumps a version number. A writer may say expected_version=n, and the write fails if somebody else wrote first. This is optimistic concurrency. The loser reads again and retries.

The second problem is what to remember across tasks. Week 4b gave you memory for one agent. A team has more to keep: who did what, in what order, and which team-level lessons came out. G-Memory builds a three-tier memory for a team [12]:

• An interaction graph with the exact utterances between agents for each task.
• A query graph that stores earlier tasks, whether they succeeded, and how they relate.
• An insight graph of general lessons distilled from experience.

The lecture maps these to reflections (insights), an index of episodes (queries) and the episodic log (interactions) [31]. When a new task arrives, the system finds similar earlier queries, then goes up to related insights and down to short summaries of the relevant interactions, and gives each agent a version that fits its role [12]. The authors report gains of up to 20.89% on an embodied-action benchmark and 10.12% on knowledge QA, across three multi-agent frameworks and five benchmarks [12]. Two other things in that paper are useful. Memory designs made for one agent often did not help a team, and some made it worse. And the authors say that team trajectories are long (up to 10 times the tokens of single-agent runs in their figure), so you cannot just paste the history back in [12]. These are the authors' own results on their chosen benchmarks. I read the paper's main text and tables.

The lab's TeamMemory is a toy with the same three tiers: a log of interactions, an index of tasks with their outcomes, and a short list of distilled insights. Its recall is word overlap, not the paper's method. Treat it as a picture of the shape.

2.9 Coordination: isolated workspaces

If five agents edit the same folder at once, the last writer wins and the others' work disappears without a sound. The lecture's fix is worktree isolation: each agent gets its own checkout and branch, and finished work is merged, conflicts are resolved, and the combined result is tested. The trade-off: isolation avoids interference during the work, and creates integration work afterward, because the copies drift apart [31].

The idea comes from git. A repository has one main working tree and can have extra linked working trees attached to it, each with its own checked-out branch [28]. Real systems use it, or its cousins:

• Symphony gives each issue its own workspace directory [26].
• Carlini's compiler agents each ran in their own container and cloned a shared bare upstream repository. They pulled, merged, pushed, and removed their lock. "Merge conflicts are frequent," he writes, and the agents resolved them [24].
• Anthropic's research system tells sub-agents to write their output to a file system, to avoid the "game of telephone" through the lead agent [14].

Three-way merge, by hand

To combine two edited copies, you need the common starting point. A three-way merge compares each copy to that base, and works by regions of lines:

1. If only one side changed a region, take that change.
2. If both sides changed a region in the same way, take it once.
3. If both changed the same region in different ways, or their changes touch, you have a conflict. Write both versions with markers (<<<<<<<, =======, >>>>>>>) and stop for a decision.

The lab's merge3 does this on lines. It also handles a delete on one side and an edit on the other, which it reports as a conflict. A merge that reports no conflict is not proof that the result works. Agent A can rename a function in one file while agent B adds a call to the old name in another file. Git sees two clean edits. The program is broken. That is why the lecture says to test the combined result [31], and why merge_workspaces takes a dictionary of checks to run on the merged files.

2.10 Patterns that agents use on their own

The lecture names three coordination patterns that were seen in CooperBench. In that benchmark each of two agents gets its own feature to add to the same repository. The features are compatible, but they touch overlapping code, so the agents must coordinate [31] [13]. The authors report that these behaviors are rare, and they found them by running a large number of simulations [13]:

• Role division: agents agree on who handles which part.
• Resource division: agents avoid collisions by splitting a shared resource, such as who edits which lines or files.
• Negotiation: agents with conflicting approaches settle on one plan.

The patterns are a good checklist of what a team needs to do. But seeing the words in a chat does not mean the plan was kept, as the next section shows. CooperBench is co-authored by two of the course's instructors (Michael Ryan and Diyi Yang), so the lecture is close to its source [13].

2.11 Why multi-agent systems fail

The MAST paper (required reading 2) starts from a plain observation: the gains of multi-agent systems on popular benchmarks are often small [2]. To find out why, the authors collected more than 1,600 annotated traces (MAST-Data, 1,642 in the figure) from seven multi-agent frameworks, whose failure rates in their tests ranged from 41% to 86.7%. They built the taxonomy by working through 150 traces with expert annotators, and the annotators agreed with a Cohen's kappa of 0.88. The result is 14 failure modes in three groups [2]. They also built an LLM-as-judge annotator, which agrees with humans at a kappa of 0.77 (0.79 on systems that it was not tuned on) [2].

ID	Mode	Share	What it looks like (my wording)
1. System design (44.2%)
1.1	Disobey task specification	11.8%	The result ignores a stated requirement.
1.2	Disobey role specification	1.5%	An agent does a job that belongs to another role.
1.3	Step repetition	15.7%	The team repeats a step it has already done.
1.4	Loss of conversation history	2.8%	Earlier context is dropped, and an agent asks again or contradicts itself.
1.5	Unaware of termination conditions	12.4%	The team keeps going after it should have stopped.
2. Inter-agent misalignment (32.3%)
2.1	Conversation reset	2.2%	The dialogue restarts for no reason.
2.2	Fail to ask for clarification	6.8%	An agent guesses at something that it should have asked about.
2.3	Task derailment	7.4%	The work drifts away from the task.
2.4	Information withholding	0.8%	An agent has something another needs and does not say it.
2.5	Ignored other agent's input	1.9%	A peer's advice or correction has no effect.
2.6	Reasoning-action mismatch	13.2%	What an agent says it will do differs from what it does.
3. Task verification (23.5%)
3.1	Premature termination	6.2%	The team stops before the work is done.
3.2	No or incomplete verification	8.2%	Nobody checks, or checks only part.
3.3	Incorrect verification	9.1%	A check passes work that is wrong.

Shares are from the paper's taxonomy figure [2]. The paper's text gives 0.85% for mode 2.4 where the figure says 0.8%, and a second figure over 210 traces gives slightly different group shares (41.8%, 36.9% and 21.3%). The paper calls them the prevalence of each mode and category in its analysis of 1,642 traces. They add up to 100%, so I call them shares. The "what it looks like" column is my own short reading, not the paper's definitions.

Four things stand out. First, no single group dominates. Design, misalignment and verification each account for a large share, so one fix will not do. Second, the biggest single modes are step repetition (15.7%), reasoning-action mismatch (13.2%), unawareness of when to stop (12.4%) and disobeying the task (11.8%). Third, verification is a group of its own: skipped, incomplete or wrong checks together are 23.5%. Fourth, diagnosis is hard: the paper notes that similar surface behavior (such as missing information) can come from different causes, such as withholding (2.4), ignoring input (2.5) or lost history (1.4) [2]. That is why a fine-grained list helps.

The authors tried two simple fixes on two systems: sharper role specifications (+9.4% on ChatDev) and a high-level verification step (+15.6% on ProgramDev). They conclude that such patches help but are not enough, and that the failures need deeper changes to how teams are designed [2].

Coordination failures: CooperBench

CooperBench measures the same thing from another side. It has more than 600 coding tasks from 12 libraries in 4 languages, and in each task two agents get two features to build in one repository [13]. Its headline: on average, success is 30% lower when two agents split the work than when one agent does both. For the strongest models (GPT-5 and Claude Sonnet 4.5), the paper says two agents reach only 25%, about half of the single-agent figure [13]. It names three causes [13]:

• Communication: messages that are vague, late or wrong, questions that go unanswered, and floods of repeated status updates.
• Commitment: agents break promises, make claims about the code that they have not checked, and ignore agreed integration points.
• Expectation: agents hold wrong beliefs about what their partner is doing, repeat work, and overwrite changes they think will merge cleanly.

In the lecture's words, a single agent solving both tasks is better than two agents that must coordinate [31]. This is an intentionally conflicting benchmark, since the two features touch the same code. It is a stress test for coordination. It is not an estimate of how all teams do.

Behavioral failures: groups behave differently

The Anthropic multiagent post (a research post, not a peer-reviewed paper) reports failures that do not occur in one agent at all [27]. They come from many similar agents together:

• Conformity. Agents with the same model, context and scaffold act alike, so a bad decision is made many times. In one run, 18 of 30 agents created a git branch with the same name. When a swarm was asked to create something impressive, over half chose to build a ray tracer or a self-hosting compiler. In an iterated prisoner's dilemma with communication, all agents settled on one strategy and defected together [27].
• Incompatible goals. Three instances of one model, each asked to migrate the same backend to a different language, formed a turf war. Each assumed the others were blocking it, and the models sabotaged one another, for example by disabling accounts and killing processes in a loop [27].

The same post has a result in favor of swarms. On vulnerability finding, a coordinating swarm of 45 agents with a shared forum found 266 vulnerabilities in a 27-million-token run, against 21 for independent agents in a 6.5-million-token run. The post adds three things that change the reading: about half of the swarm's finds were outside the directories where the independent agents were sent; inside those directories the two methods looked comparable per token; and only 12 vulnerabilities were found by both [27]. So the honest summary is "complementary, and the swarm explored more", not "swarms are 12 times better". The post is from Anthropic and describes its own experiments.

2.12 Case study: sixteen agents write a compiler

The lecture's optional case study is Nicholas Carlini's project at Anthropic: sixteen Claude agents wrote a C compiler in Rust that can build the Linux kernel [24] [31]. The numbers, from his post: nearly 2,000 sessions over two weeks, about 2 billion input tokens and 140 million output tokens, a bill just under $20,000, and a 100,000-line compiler that builds Linux 6.9 for x86, ARM and RISC-V [24]. It is an Anthropic showcase, and the author calls it a very early research prototype.

Look at how it maps onto this week's ideas:

Idea	What the project did [24]
Orchestration	A decentralized swarm. There was no orchestration agent. Each agent picked the "next most obvious" problem, and each ran in a loop.
Task claiming	An agent wrote a lock file, such as current_tasks/parse_if_statement.txt. If two agents tried the same task, git's synchronization made the second pick another one. (The lab's TaskBoard.claim is the in-memory version.)
Isolation	Each agent worked in its own container on a clone of a shared upstream, pulled and merged, pushed, and removed its lock. Merge conflicts were frequent.
Shared memory	Agents kept notes: a running document of failed approaches and what remained.
Verification	Most of the effort went into the environment: tests, feedback and an oracle. When all agents hit the same kernel bug, the fix was to use GCC as a known-good compiler and compile a random subset of files with it, so that different agents could work on different failing files.

The lecture's lessons from the case, in my words [31]: strong tests tell agents whether they are making progress; design the task so that it has separable parts; make coordination explicit (claim, clone, merge); keep knowledge across sessions in documents and notes; give feedback that is short and fast to read; and give agents complementary goals. A related Anthropic post on long-running agents recommends the same habits for a single agent that spans many context windows: a feature list, a progress file, and a commit at the end of each session [29].

One more lesson sits in the middle of the post: the kernel build was one giant task, so sixteen agents hit the same bug and overwrote each other's fixes. Parallelism needs separable work, and sometimes you have to build the separation yourself (the oracle) [24].

2.13 A worked cost example

Here is a small model that you can do on paper. Every number in it is made up by me, and the lab's simulate function runs the same model. The point is to see which assumptions create the result. A task has 8 items and every item must be right.

Assumption (made up)	Value
Chance that a step is right with a fresh context	0.92
Skill lost for each earlier item in the same context ("dilution")	0.015
Tokens: a step; re-reading each earlier item; one message	60; 15; 40
Workers in a team	4

One agent, 8 items. The agent's chance on item i (counting from 0) is 0.92 − 0.015i. All 8 must be right: 0.92 × 0.905 × … × 0.815 = 0.319. Tokens: Σ(60 + 15i) for i = 0 to 7 = 480 + 15 × 28 = 900.

Orchestrator with 4 workers. Each worker gets 2 items, so every context holds at most one earlier item. Four items are at chance 0.92 and four at 0.905: 0.92⁴ × 0.905⁴ = 0.481. Tokens: 2 messages for the orchestrator (80), plus for each item 2 messages (80), 60 for the step, and 15 for the 4 items that follow another one. 80 + 8 × 140 + 4 × 15 = 1,260. The team costs 1.4 times the tokens and succeeds 1.5 times as often, in 4 rounds instead of 8.

Now see what the model assumes. The team wins here only because the lone agent gets worse as its context grows (the dilution). Set the dilution to zero and both succeed with 0.92⁸ = 0.513, and the team has cost more for no gain. One of the lab exercises asks you to check this. In the sequential case, where each step needs the previous result and a hand-off loses a value 10% of the time, the orchestrator's chance is (0.92 × 0.9)⁸ = 0.221, below the single agent's 0.319. These are the same two forces as in the evidence of section 2.2: a team gains from focused contexts, and loses from lost information.

The lab's demo runs all seven designs for both kinds of task. The rates come from 2,000 random trials, so they carry noise of a point or two: with 200,000 trials, the independent-voting row in the parallel case is 0.77, and the demo prints 0.79.

Task	Design	Success	Tokens	Rounds
parallel	single	0.32	900	8
parallel	single + verify	0.72	1,483	16
parallel	independent (5 voters)	0.79	4,900	9
parallel	orchestrator	0.48	1,260	4
parallel	orchestrator + verify	0.82	2,136	4
parallel	swarm	0.48	970	2
parallel	hybrid	0.48	1,340	4
sequential	single	0.32	900	8
sequential	single + verify	0.72	1,483	16
sequential	independent	0.40	4,900	9
sequential	orchestrator	0.22	1,200	17
sequential	orchestrator + verify	0.63	2,150	17
sequential	swarm	0.32	1,480	8
sequential	hybrid	0.41	1,760	17

All values are from my toy model with made-up parameters (the lab's demo, section 9, seed 0). The swarm's parallel cost (970) is lower than the orchestrator's (1,260) because my model gives the swarm no orchestrator messages and charges only for a small board read; real swarms may differ. The swarm's sequential success of 0.32 equals the single agent's by coincidence of my numbers.

What to take from the table, as claims about this toy model: (1) A check on every item is worth more than the choice of architecture: single + verify (0.72) beats the unchecked orchestrator (0.48) and the unchecked swarm. (2) On sequential work, the unchecked team is worse than one agent (0.22 against 0.32), and a single agent with a check (0.72) beats the checked team (0.63). (3) Independent voting works well here only because I assumed that errors are independent apart from a 3% share of items that are hard in the same way for all voters. If real errors are correlated, as a shared model would make them, the gain shrinks. (4) Cost and accuracy move together only up to a point: the voting team spends 5.4 times the tokens of the single agent for 0.79 against 0.72 for the checked single agent, which spends 1.6 times. None of this is a measurement of a real system. Kim et al. found that the answer depends on the task, and so does this toy [3].

2.14 Week 5a in one page

• Start with one agent. Add agents for a measured reason: parallel work, specialists, focused context, wider exploration. Team costs are overhead, lost information, conflicts, error spread and complexity [31].
• The evidence is mixed and task-dependent. The best team design in the scaling study gained 80.8% on one task and the worst lost 70% on another. Above a single-agent baseline of about 45%, coordination had diminishing returns [3]. Vendors disagree: Anthropic reports large gains on research [14], Cognition argues for one thread [15].
• Four orchestration patterns: independent, centralized, decentralized, hybrid. Cost order: single, orchestrated, swarm. Teams can also be chosen per task (DyLAN) [31] [10].
• Communication: conversations (AutoGen), handoffs (control moves), agents as tools (caller keeps control), message boards (MetaGPT-style, typed and subscribed), and A2A across systems (Agent Cards, tasks, artifacts, JSON-RPC) [1] [5] [16] [17] [21].
• Coordination: a task ledger (Magentic-One's two ledgers and stall counter), shared memory with locks or version checks (and G-Memory's three tiers), and isolated workspaces with a merge and a test of the result [4] [12] [28].
• Emergent patterns: role division, resource division, negotiation [13].
• Failures: 14 MAST modes in three groups (design, misalignment, verification); the curse of coordination (two agents do about 30% worse than one on CooperBench); and group-only behaviors such as conformity and turf wars [2] [13] [27].
• Case: sixteen agents, no orchestrator, lock files, clones and merges, strong tests, and an oracle to split one giant task [24].

3. Reading guide

Three readings are required by the course. For each you get: why it matters, the key ideas, and questions to think about as you read. Read the questions first. The optional table at the end points to the sources that I used for the notes. For some of those I read only the abstract, and the references say which.

Optional readings

Source	Why it is worth your time
Khatua et al., CooperBench [13]	The curse of coordination, with the three causes and three emergent patterns. Read the failure analysis and the message examples. Two of the authors are the course's instructors.
Fourney et al., Magentic-One [4]	The task ledger and progress ledger, and the stall counter. Read the section on the Orchestrator and its two ledgers.
Hong et al., MetaGPT [5]; Qian et al., ChatDev [8]; Li et al., CAMEL [6]; Chen et al., AgentVerse [9]	The early role-based teams. In MetaGPT, find the shared message pool and the subscription rule.
Du et al., multiagent debate [7]; Li et al., More Agents Is All You Need [11]; Liu et al., DyLAN [10]	Teams as an ensemble. Ask in each: what keeps the agents' errors from being the same?
Zhang et al., G-Memory [12]	Memory for a team (insight, query and interaction graphs). Compare with Week 4b.
Anthropic, multi-agent research system [14]; Cognition, Don't Build Multi-Agents [15]	The two sides of the debate. Read both in a row, and write down what they agree on.
OpenAI Agents SDK, handoffs and tools [16] [17]; OpenAI guide to building agents [19]; Swarm [18]; LangGraph docs [20]	The handoff and agent-as-tool patterns in code.
A2A specification [21]; A2A and MCP [22]; JSON-RPC 2.0 [23]	The protocol. Read the sections on the Agent Card, the task lifecycle and the error table.
Carlini, building a C compiler [24]; Anthropic on long-running agents [29]	A swarm in practice, and good habits for work that spans sessions.
OpenAI Symphony: announcement [25] and specification [26]; git worktree docs [28]	Per-issue workspaces and a tracker as a ledger. Read the spec's sections on the workspace and retries.
Anthropic, patterns and problems in multiagent systems [27]	Conformity, incompatible goals, and the swarm against independent search.

4. Hands-on lab: teams of mock agents

Time: about 3.5 hours. Goal: build small, testable versions of this week's designs, break them on purpose, and see what each one costs. You do not need an API key or any package. Every "language model" is a mock: a scripted list of replies, or a few lines of Python that act like a model.

What you will learn

• How a handoff and an agent-as-tool differ in code, and what each specialist sees.
• How an orchestrator splits, retries and merges, and how a hybrid team reads its peers' notes.
• How a typed message board and an atomic task board work, and how a swarm can deadlock.
• What an A2A Agent Card, a task and a JSON-RPC exchange look like, and what a server must reject.
• How a task ledger, a progress ledger and a stall counter drive a re-plan.
• How shared state is lost without a lock, and how versions catch a stale write.
• How isolated workspaces, a three-way merge and a post-merge check work, and what a shared folder loses.
• How to inject a failure into a team and detect it, one MAST mode at a time.
• How to compare a single agent and a team on cost and accuracy, and why every number needs its assumptions.

Setup

1. You need a recent Python 3. There is nothing to install. The code uses only the standard library. I ran it on Python 3.13 only.
2. Save the code below as multi_agent_toolkit.py. Or download it: multi_agent_toolkit.py.
3. Save the tests as test_multi_agent_toolkit.py in the same folder. Or download: test_multi_agent_toolkit.py. The test code is not printed on this page, to keep the page shorter. It is in the download, and the names of all the tests are listed after the exercises.
4. Run python multi_agent_toolkit.py for the nine-section demo, and python test_multi_agent_toolkit.py for the tests. The demo and the tests are deterministic. The tests take about a second. The merge and workspace code writes to temporary folders and removes them.

The code: multi_agent_toolkit.py

The code has eight parts, matching Figure 7, and a demo at the end. The comments name each choice that I made and the sources did not. The file is long. Read the part you are working on, with the matching section of the notes open.

"""multi_agent_toolkit.py - Week 5a lab: multi-agent systems with mock models.

Standard library only (tested on Python 3.13). Every "model" here is a scripted list of
replies or a few lines of Python that act like one. All task data, skill levels, costs and
failure scripts are MADE UP for teaching. Nothing here measures a real model.

Parts
  1  Agents: handoff vs agent-as-tool, a swarm runner, orchestrator-worker, a hybrid team
  2  A message board (MetaGPT-like) and a task board with atomic claims (a swarm's workspace)
  3  A2A v1.0-shaped Agent Cards, Task/Message/Artifact objects and JSON-RPC 2.0 messages
  4  Magentic-One-style task ledger and progress ledger
  5  Shared memory with locking, plus a toy three-tier team memory
  6  Worktree-like isolation in temp dirs, with a three-way merge that detects conflicts
  7  Failure injection into a scripted team, and a MAST-style rule-based classifier
  8  A single-agent vs multi-agent cost/accuracy simulation (made-up parameters)
"""
from __future__ import annotations

import difflib
import itertools
import json
import math
import os
import random
import re
import shutil
import sys
import tempfile
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime, timezone


# --------------------------------------------------------------------------- 0. mocks
def count_tokens(text) -> int:
    """Rough token proxy: words plus punctuation marks. Not a real tokenizer."""
    return len(re.findall(r"\w+|[^\w\s]", str(text)))


class ScriptedLLM:
    """A mock model: returns the next scripted reply and records every prompt."""

    def __init__(self, replies):
        self.replies = list(replies)
        self.calls = []

    def __call__(self, prompt):
        self.calls.append(prompt)
        if not self.replies:
            raise RuntimeError("ScriptedLLM: script exhausted")
        return self.replies.pop(0)


@dataclass
class CostMeter:
    calls: int = 0
    tokens_in: int = 0
    tokens_out: int = 0
    messages: int = 0

    def add(self, prompt, reply):
        self.calls += 1
        self.tokens_in += count_tokens(prompt)
        self.tokens_out += count_tokens(reply)

    @property
    def tokens(self):
        return self.tokens_in + self.tokens_out


# --------------------------------------------------------------------------- 1. agents
class MaxStepsExceeded(RuntimeError):
    pass


@dataclass
class Handoff:
    """Returned by a transfer tool: control moves to `target`."""
    target: "Agent"
    note: str = ""


class Agent:
    """An agent is a policy plus tools. policy(history, ctx) returns {"say": text} or
    {"call": tool_name, "args": {...}}. Handoffs are registered as tools named
    transfer_to_<agent_name>, the naming the OpenAI Agents SDK docs describe."""

    def __init__(self, name, policy, tools=None, handoffs=None, description=""):
        self.name = name
        self.policy = policy
        self.description = description
        self.tools = dict(tools or {})
        self.handoffs = list(handoffs or [])
        for h in self.handoffs:
            self.tools["transfer_to_" + h.name] = (lambda ctx, _h=h, **kw: Handoff(_h, kw.get("note", "")))

    def as_tool(self, name=None):
        return AgentTool(self, name or self.name)


def run_agent(agent, history, ctx=None, max_steps=8):
    """Run one agent until it speaks or hands off. Appends to `history` in place.
    Returns ("say", text) or ("handoff", Handoff)."""
    ctx = {} if ctx is None else ctx
    for _ in range(max_steps):
        step = agent.policy(history, ctx)
        if "say" in step:
            history.append({"role": "assistant", "agent": agent.name, "content": step["say"]})
            return "say", step["say"]
        if "call" not in step:
            raise ValueError(f"{agent.name}: policy returned neither say nor call: {step!r}")
        name, args = step["call"], step.get("args", {})
        history.append({"role": "assistant", "agent": agent.name, "call": name, "args": args})
        tool = agent.tools.get(name)
        if tool is None:
            history.append({"role": "tool", "agent": agent.name, "name": name, "content": f"error: unknown tool {name!r}"})
            continue
        out = tool(ctx, **args)
        if isinstance(out, Handoff):
            history.append({"role": "tool", "agent": agent.name, "name": name, "content": f"handoff to {out.target.name}"})
            return "handoff", out
        history.append({"role": "tool", "agent": agent.name, "name": name, "content": str(out)})
    raise MaxStepsExceeded(f"{agent.name} used {max_steps} steps without speaking")


class AgentTool:
    """Agent-as-tool: the caller keeps control. The specialist sees ONLY the request
    string (a fresh history), and returns text. It may not hand off."""

    def __init__(self, agent, name):
        self.agent, self.name = agent, name

    def __call__(self, ctx, request):
        hist = [{"role": "user", "content": request}]
        kind, out = run_agent(self.agent, hist, dict(ctx))
        if kind == "handoff":
            return f"error: {self.agent.name} tried to hand off inside a tool call"
        return out


@dataclass
class SwarmResult:
    agent: str
    messages: list
    handoffs: list
    reply: str


def run_swarm(start, messages, ctx=None, max_turns=6, history_filter=None):
    """Swarm-style loop: one active agent at a time, shared history, handoff by tool call.
    history_filter(history) models an SDK-style input filter: what the next agent sees."""
    ctx = {} if ctx is None else ctx
    history = list(messages)
    current, hops = start, []
    for _ in range(max_turns):
        kind, out = run_agent(current, history, ctx)
        if kind == "say":
            return SwarmResult(current.name, history, hops, out)
        hops.append((current.name, out.target.name))
        current = out.target
        if history_filter:
            history[:] = history_filter(history)
    raise MaxStepsExceeded(f"swarm exceeded {max_turns} turns; hops so far: {hops}")


class WorkerError(RuntimeError):
    pass


@dataclass
class OrchResult:
    answer: str
    results: list
    errors: list
    meter: CostMeter


def run_orchestrator(task, plan, workers, synthesize, max_retries=1, parallel=True):
    """Orchestrator-worker. plan(task) -> [(worker_name, instruction)]. Each worker is a
    callable instruction -> text and may raise WorkerError; failed subtasks are retried
    up to max_retries times. The orchestrator then synthesizes. Workers run in threads."""
    meter = CostMeter()
    subtasks = plan(task)
    meter.messages += len(subtasks)

    def one(item):
        name, instruction = item
        last = None
        for _ in range(max_retries + 1):
            try:
                out = workers[name](instruction)
                meter.add(instruction, out)
                return out, None
            except WorkerError as e:
                last = f"{name}: {e}"
        return None, last

    if parallel and len(subtasks) > 1:
        with ThreadPoolExecutor(max_workers=len(subtasks)) as ex:
            pairs = list(ex.map(one, subtasks))
    else:
        pairs = [one(s) for s in subtasks]
    results = [p[0] for p in pairs]
    errors = [p[1] for p in pairs if p[1]]
    answer = synthesize(task, results)
    meter.messages += len(subtasks)
    return OrchResult(answer, results, errors, meter)


def run_hybrid(task, plan, workers, board, synthesize, rounds=2):
    """Hybrid: an orchestrator assigns work; workers ALSO read each other's notes on a
    board between rounds. plan(task) -> [(worker_name, instruction)].
    workers[name](instruction, peer_notes) -> text. Notes are published after each round."""
    subtasks = plan(task)
    last = [None] * len(subtasks)
    for r in range(rounds):
        new = []
        for i, (name, instruction) in enumerate(subtasks):
            notes = [m["content"] for m in board.history() if m["sender"] != name]
            new.append(workers[name](instruction, notes))
        for i, (name, _) in enumerate(subtasks):
            board.publish(name, "notes", new[i])
        last = new
    return synthesize(task, last)


# --------------------------------------------------------------------------- 2. boards
class SchemaError(ValueError):
    pass


class MessageBoard:
    """A MetaGPT-like shared message pool: agents publish structured messages and
    subscribe to topics. A message with `to` is visible only to that agent."""

    def __init__(self, schemas=None):
        self._lock = threading.Lock()
        self._msgs = []
        self._subs = {}
        self._cursor = {}
        self.schemas = dict(schemas or {})

    def subscribe(self, agent, topics):
        with self._lock:
            self._subs[agent] = set(topics)
            self._cursor.setdefault(agent, 0)

    def publish(self, sender, topic, content, cause_by=None, to=None):
        need = self.schemas.get(topic)
        if need is not None:
            if not isinstance(content, dict) or any(k not in content for k in need):
                raise SchemaError(f"topic {topic!r} needs keys {sorted(need)}")
        with self._lock:
            msg = {"id": len(self._msgs) + 1, "sender": sender, "topic": topic,
                   "content": content, "cause_by": cause_by, "to": to}
            self._msgs.append(msg)
            return msg["id"]

    def read_new(self, agent):
        with self._lock:
            subs = self._subs.get(agent, set())
            cur = self._cursor.get(agent, 0)
            out = [m for m in self._msgs if m["id"] > cur and m["sender"] != agent
                   and ((m["to"] is None and m["topic"] in subs) or m["to"] == agent)]
            self._cursor[agent] = len(self._msgs)
            return out

    def history(self, topic=None):
        with self._lock:
            return [m for m in self._msgs if topic is None or m["topic"] == topic]


class TaskBoard:
    """A swarm's shared task list. claim() is atomic, so two agents never get the same task.
    Tasks can depend on other tasks. Like Carlini's lock files, but in memory."""

    def __init__(self):
        self._lock = threading.Lock()
        self.tasks = {}
        self.claims = []

    def add(self, task_id, payload, deps=()):
        with self._lock:
            if task_id in self.tasks:
                raise KeyError(f"duplicate task {task_id}")
            self.tasks[task_id] = {"payload": payload, "deps": tuple(deps), "owner": None, "result": None, "done": False}

    def claim(self, agent):
        with self._lock:
            for tid in sorted(self.tasks):
                t = self.tasks[tid]
                if t["owner"] is None and all(self.tasks[d]["done"] for d in t["deps"] if d in self.tasks):
                    t["owner"] = agent
                    self.claims.append((tid, agent))
                    return tid
            return None

    def complete(self, task_id, agent, result):
        with self._lock:
            t = self.tasks[task_id]
            if t["owner"] != agent:
                raise PermissionError(f"{agent} does not own {task_id}")
            t["result"], t["done"] = result, True

    def results(self):
        with self._lock:
            return {k: v["result"] for k, v in self.tasks.items() if v["done"]}

    def finished(self):
        with self._lock:
            return all(t["done"] for t in self.tasks.values())

    def stuck(self):
        """True if nothing is running or ready but some task is not done (a dependency cycle)."""
        with self._lock:
            if all(t["done"] for t in self.tasks.values()):
                return False
            running = any(t["owner"] is not None and not t["done"] for t in self.tasks.values())
            ready = any(t["owner"] is None and all(self.tasks[d]["done"] for d in t["deps"] if d in self.tasks)
                        for t in self.tasks.values())
            return not running and not ready


def run_board_swarm(agents, board, timeout=5.0):
    """Peer agents (name -> callable(payload, results_so_far) -> text) pull tasks from the board."""
    errors = []
    start = time.time()

    def loop(name):
        fn = agents[name]
        while not board.finished():
            if board.stuck():
                errors.append("deadlock: dependency cycle")
                return
            if time.time() - start > timeout:
                errors.append("timeout")
                return
            tid = board.claim(name)
            if tid is None:
                time.sleep(0.001)
                continue
            payload = board.tasks[tid]["payload"]
            try:
                board.complete(tid, name, fn(payload, board.results()))
            except Exception as e:  # a failed agent releases nothing: the task stays owned (visible bug)
                errors.append(f"{name} failed {tid}: {e}")
                return

    threads = [threading.Thread(target=loop, args=(n,)) for n in agents]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return errors


# --------------------------------------------------------------------------- 3. A2A (v1.0 shapes)
A2A_VERSION = "1.0"
WELL_KNOWN_PATH = "/.well-known/agent-card.json"
TERMINAL = {"TASK_STATE_COMPLETED", "TASK_STATE_FAILED", "TASK_STATE_CANCELED", "TASK_STATE_REJECTED"}
INTERRUPTED = {"TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED"}
TASK_STATES = ["TASK_STATE_UNSPECIFIED", "TASK_STATE_SUBMITTED", "TASK_STATE_WORKING", "TASK_STATE_COMPLETED",
               "TASK_STATE_FAILED", "TASK_STATE_CANCELED", "TASK_STATE_INPUT_REQUIRED", "TASK_STATE_REJECTED",
               "TASK_STATE_AUTH_REQUIRED"]
# name -> (JSON-RPC code, message). Codes are from spec section 5.4 and 9.5.
RPC_ERRORS = {
    "JSONParseError": (-32700, "Invalid JSON payload"),
    "InvalidRequestError": (-32600, "Request payload validation error"),
    "MethodNotFoundError": (-32601, "Method not found"),
    "InvalidParamsError": (-32602, "Invalid parameters"),
    "InternalError": (-32603, "Internal error"),
    "TaskNotFoundError": (-32001, "Task not found"),
    "TaskNotCancelableError": (-32002, "Task cannot be canceled"),
    "PushNotificationNotSupportedError": (-32003, "Push notifications are not supported"),
    "UnsupportedOperationError": (-32004, "This operation is not supported"),
    "ContentTypeNotSupportedError": (-32005, "Incompatible content types"),
    "InvalidAgentResponseError": (-32006, "Invalid agent response"),
    "ExtendedAgentCardNotConfiguredError": (-32007, "Extended agent card not configured"),
    "ExtensionSupportRequiredError": (-32008, "Extension support required"),
    "VersionNotSupportedError": (-32009, "Protocol version not supported"),
}
KNOWN_METHODS = ["SendMessage", "SendStreamingMessage", "GetTask", "ListTasks", "CancelTask", "SubscribeToTask",
                 "GetExtendedAgentCard"]


class A2AError(Exception):
    def __init__(self, name, data=None, message=None):
        self.name, self.data = name, data
        self.code, std = RPC_ERRORS[name]
        self.message = message or std
        super().__init__(f"{name} ({self.code}): {self.message}")


def _reason(name):
    return re.sub(r"(?<!^)(?=[A-Z])", "_", name.removesuffix("Error")).upper()


def error_info(name, **metadata):
    # reason strings other than TASK_NOT_FOUND are my own convention, not quoted from the spec
    return [{"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": _reason(name),
             "domain": "a2a-protocol.org", "metadata": {k: str(v) for k, v in metadata.items()}}]


def bad_request(*violations):
    return [{"@type": "type.googleapis.com/google.rpc.BadRequest",
             "fieldViolations": [{"field": f, "description": d} for f, d in violations]}]


def rpc_error(req_id, err):
    body = {"code": err.code, "message": err.message}
    if err.data:
        body["data"] = err.data
    return {"jsonrpc": "2.0", "id": req_id, "error": body}


def make_agent_card(name, description, url, skills, version="1.0.0", streaming=False,
                    push=False, input_modes=("text/plain",), output_modes=("text/plain",)):
    return {
        "name": name, "description": description,
        "supportedInterfaces": [{"url": url, "protocolBinding": "JSONRPC", "protocolVersion": A2A_VERSION}],
        "version": version,
        "capabilities": {"streaming": streaming, "pushNotifications": push, "extendedAgentCard": False},
        "defaultInputModes": list(input_modes), "defaultOutputModes": list(output_modes),
        "skills": [dict(s) for s in skills],
    }


def validate_agent_card(card):
    """Check the REQUIRED fields of a v1.0 AgentCard (a.proto, field_behavior = REQUIRED)."""
    errs = []
    for k in ("name", "description", "supportedInterfaces", "version", "capabilities",
              "defaultInputModes", "defaultOutputModes", "skills"):
        if k not in card:
            errs.append(f"missing {k}")
    for k in card:
        if "_" in k:
            errs.append(f"field {k!r} is snake_case; JSON must be camelCase")
    ifaces = card.get("supportedInterfaces")
    if isinstance(ifaces, list):
        if not ifaces:
            errs.append("supportedInterfaces is empty")
        for i, x in enumerate(ifaces):
            for k in ("url", "protocolBinding", "protocolVersion"):
                if k not in x:
                    errs.append(f"supportedInterfaces[{i}] missing {k}")
    for i, s in enumerate(card.get("skills") or []):
        for k in ("id", "name", "description", "tags"):
            if k not in s:
                errs.append(f"skills[{i}] missing {k}")
    return errs


def text_message(text, role="ROLE_USER", message_id=None, task_id=None, context_id=None):
    m = {"messageId": message_id or str(uuid.uuid4()), "role": role, "parts": [{"text": text}]}
    if task_id:
        m["taskId"] = task_id
    if context_id:
        m["contextId"] = context_id
    return m


class A2AServer:
    """An in-process A2A server speaking the JSON-RPC binding. No network. The handler is
    handler(text, info) -> (kind, text) with kind in completed, input_required, failed,
    rejected, message. info has taskId, contextId, history (list of texts)."""

    def __init__(self, card, handler, clock=None, id_factory=None, supported_versions=(A2A_VERSION,)):
        self.card, self.handler = card, handler
        self.clock = clock or (lambda: datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"))
        self._ids = id_factory or (lambda: str(uuid.uuid4()))
        self.supported_versions = tuple(supported_versions)
        self.tasks, self._seen, self._pending = {}, {}, []
        self._lock = threading.RLock()

    def get(self, path):
        """Mimic GET on the well-known path."""
        return self.card if path == WELL_KNOWN_PATH else None

    # ---- helpers
    def _status(self, state, text=None):
        st = {"state": state, "timestamp": self.clock()}
        if text is not None:
            st["message"] = {"messageId": self._ids(), "role": "ROLE_AGENT", "parts": [{"text": text}]}
        return st

    @staticmethod
    def _validate_message(m):
        v = []
        if not isinstance(m, dict):
            return [("message", "message is required and must be an object")]
        if not isinstance(m.get("messageId"), str) or not m.get("messageId"):
            v.append(("message.messageId", "messageId is required"))
        if m.get("role") != "ROLE_USER":
            v.append(("message.role", "role must be ROLE_USER for a client message"))
        parts = m.get("parts")
        if not isinstance(parts, list) or not parts:
            v.append(("message.parts", "At least one part is required"))
        else:
            for i, p in enumerate(parts):
                n = sum(1 for k in ("text", "raw", "url", "data") if k in p) if isinstance(p, dict) else 0
                if n != 1:
                    v.append((f"message.parts[{i}]", "exactly one of text, raw, url, data is required"))
        return v

    def _view(self, task, history_length=None, artifacts=True):
        t = json.loads(json.dumps(task))
        if history_length is not None:
            h = t.get("history", [])
            t["history"] = h[-history_length:] if history_length > 0 else []
            if not t["history"]:
                t.pop("history")
        if not artifacts:
            t.pop("artifacts", None)
        return t

    def _text_of(self, m):
        return " ".join(p["text"] for p in m["parts"] if "text" in p)

    def _process(self, task):
        with self._lock:
            if task["status"]["state"] in TERMINAL:
                return
            task["status"] = self._status("TASK_STATE_WORKING")
            texts = [self._text_of(m) for m in task["history"]]
            info = {"taskId": task["id"], "contextId": task["contextId"], "history": texts}
        try:
            kind, text = self.handler(texts[-1], info)
        except Exception:
            kind, text = "failed", "agent error"
        with self._lock:
            if task["status"]["state"] in TERMINAL:  # canceled while working
                return
            if kind == "completed":
                task.setdefault("artifacts", []).append(
                    {"artifactId": f"artifact-{len(task.get('artifacts', [])) + 1}", "name": "result",
                     "parts": [{"text": text}]})
                task["status"] = self._status("TASK_STATE_COMPLETED")
            elif kind in ("input_required", "failed", "rejected"):
                state = {"input_required": "TASK_STATE_INPUT_REQUIRED", "failed": "TASK_STATE_FAILED",
                         "rejected": "TASK_STATE_REJECTED"}[kind]
                task["status"] = self._status(state, text)
            else:
                raise A2AError("InvalidAgentResponseError", error_info("InvalidAgentResponseError", kind=kind))
            reply = task["status"].get("message") or {"messageId": self._ids(), "role": "ROLE_AGENT",
                                                     "parts": [{"text": text}]}
            task["history"].append(dict(reply, taskId=task["id"], contextId=task["contextId"]))

    def run_pending(self):
        """Process tasks that were accepted with returnImmediately."""
        done = 0
        while self._pending:
            tid = self._pending.pop(0)
            self._process(self.tasks[tid])
            done += 1
        return done

    # ---- methods
    def _send_message(self, params, headers):
        v = (headers or {}).get("A2A-Version")
        if v is not None and v not in self.supported_versions:
            raise A2AError("VersionNotSupportedError", error_info("VersionNotSupportedError", requested=v))
        if not isinstance(params, dict):
            raise A2AError("InvalidParamsError", bad_request(("params", "params must be an object")))
        viol = self._validate_message(params.get("message"))
        if viol:
            raise A2AError("InvalidParamsError", bad_request(*viol))
        msg = params["message"]
        cfg = params.get("configuration") or {}
        for p in msg["parts"]:
            mt = p.get("mediaType")
            if mt and mt not in self.card["defaultInputModes"]:
                raise A2AError("ContentTypeNotSupportedError", error_info("ContentTypeNotSupportedError", mediaType=mt))
        with self._lock:
            if msg["messageId"] in self._seen:  # idempotent on messageId (the spec says MAY)
                return self._seen[msg["messageId"]]
            tid, ctx_id = msg.get("taskId"), msg.get("contextId")
            if tid:
                task = self.tasks.get(tid)
                if task is None:
                    raise A2AError("TaskNotFoundError", error_info("TaskNotFoundError", taskId=tid))
                if ctx_id and ctx_id != task["contextId"]:
                    raise A2AError("InvalidParamsError", bad_request(
                        ("message.contextId", "contextId does not match the task's contextId")))
                if task["status"]["state"] in TERMINAL:
                    raise A2AError("UnsupportedOperationError", error_info(
                        "UnsupportedOperationError", taskId=tid, state=task["status"]["state"]))
                task["history"].append(dict(msg, taskId=tid, contextId=task["contextId"]))
            else:
                ctx_id = ctx_id or self._ids()
                task = {"id": self._ids(), "contextId": ctx_id,
                        "status": self._status("TASK_STATE_SUBMITTED"), "history": []}
                task["history"].append(dict(msg, taskId=task["id"], contextId=ctx_id))
                self.tasks[task["id"]] = task
        if cfg.get("returnImmediately"):
            self._pending.append(task["id"])
        else:
            self._process(task)
        res = {"task": self._view(task, cfg.get("historyLength"))}
        with self._lock:
            self._seen[msg["messageId"]] = res
        return res

    def _get_task(self, params):
        tid = (params or {}).get("id")
        if not isinstance(tid, str) or not tid:
            raise A2AError("InvalidParamsError", bad_request(("id", "id is required")))
        with self._lock:
            task = self.tasks.get(tid)
            if task is None:
                raise A2AError("TaskNotFoundError", error_info("TaskNotFoundError", taskId=tid))
            return self._view(task, params.get("historyLength"))

    def _list_tasks(self, params):
        params = params or {}
        size = params.get("pageSize", 50)
        if not isinstance(size, int) or not 1 <= size <= 100:
            raise A2AError("InvalidParamsError", bad_request(("pageSize", "pageSize must be between 1 and 100")))
        start = int(params.get("pageToken") or 0)
        with self._lock:
            tasks = [t for t in self.tasks.values()
                     if (not params.get("contextId") or t["contextId"] == params["contextId"])
                     and (not params.get("status") or t["status"]["state"] == params["status"])]
            page = tasks[start:start + size]
            nxt = str(start + size) if start + size < len(tasks) else ""
            return {"tasks": [self._view(t, params.get("historyLength"), bool(params.get("includeArtifacts")))
                              for t in page],
                    "nextPageToken": nxt, "pageSize": len(page), "totalSize": len(tasks)}

    def _cancel(self, params):
        tid = (params or {}).get("id")
        if not isinstance(tid, str) or not tid:
            raise A2AError("InvalidParamsError", bad_request(("id", "id is required")))
        with self._lock:
            task = self.tasks.get(tid)
            if task is None:
                raise A2AError("TaskNotFoundError", error_info("TaskNotFoundError", taskId=tid))
            if task["status"]["state"] in TERMINAL:
                raise A2AError("TaskNotCancelableError", error_info(
                    "TaskNotCancelableError", taskId=tid, state=task["status"]["state"]))
            task["status"] = self._status("TASK_STATE_CANCELED")
            return self._view(task)

    def handle(self, request, headers=None):
        """Take a raw JSON string or a dict. Return a JSON-RPC response dict."""
        req_id = None
        try:
            if isinstance(request, (str, bytes)):
                try:
                    request = json.loads(request)
                except ValueError:
                    raise A2AError("JSONParseError")
            ok = (isinstance(request, dict) and request.get("jsonrpc") == "2.0"
                  and isinstance(request.get("method"), str) and "id" in request)
            if isinstance(request, dict):
                req_id = request.get("id")
            if not ok:
                raise A2AError("InvalidRequestError")
            method, params = request["method"], request.get("params")
            if method == "SendMessage":
                result = self._send_message(params, headers)
            elif method == "GetTask":
                result = self._get_task(params)
            elif method == "ListTasks":
                result = self._list_tasks(params)
            elif method == "CancelTask":
                result = self._cancel(params)
            elif method == "SendStreamingMessage" and not self.card["capabilities"].get("streaming"):
                raise A2AError("UnsupportedOperationError", error_info("UnsupportedOperationError", method=method))
            elif method in KNOWN_METHODS:
                raise A2AError("UnsupportedOperationError", error_info("UnsupportedOperationError", method=method))
            else:
                raise A2AError("MethodNotFoundError")
            return {"jsonrpc": "2.0", "id": req_id, "result": result}
        except A2AError as e:
            return rpc_error(req_id, e)
        except Exception:
            return rpc_error(req_id, A2AError("InternalError"))


class A2AClient:
    """Builds JSON-RPC requests, keeps a wire log, and raises A2AError on error responses."""

    def __init__(self, server, version=A2A_VERSION, message_ids=None):
        self.server, self.version = server, version
        self._mid = message_ids or (lambda: str(uuid.uuid4()))
        self._n = itertools.count(1)
        self.card, self.log = None, []

    def discover(self):
        card = self.server.get(WELL_KNOWN_PATH)
        errs = validate_agent_card(card or {})
        if errs:
            raise ValueError("bad agent card: " + "; ".join(errs))
        if not any(i["protocolBinding"] == "JSONRPC" for i in card["supportedInterfaces"]):
            raise ValueError("no JSONRPC interface")
        self.card = card
        return card

    def call(self, method, params):
        req = {"jsonrpc": "2.0", "id": next(self._n), "method": method, "params": params}
        resp = self.server.handle(json.dumps(req), headers={"A2A-Version": self.version})
        self.log.append((req, resp))
        if "error" in resp:
            e = resp["error"]
            name = next((k for k, (c, _) in RPC_ERRORS.items() if c == e["code"]), "InternalError")
            raise A2AError(name, e.get("data"), e["message"])
        return resp["result"]

    def send(self, text, task_id=None, context_id=None, return_immediately=False, history_length=None):
        cfg = {}
        if return_immediately:
            cfg["returnImmediately"] = True
        if history_length is not None:
            cfg["historyLength"] = history_length
        params = {"message": text_message(text, message_id=self._mid(), task_id=task_id, context_id=context_id)}
        if cfg:
            params["configuration"] = cfg
        return self.call("SendMessage", params)["task"]

    def get_task(self, task_id, history_length=None):
        p = {"id": task_id}
        if history_length is not None:
            p["historyLength"] = history_length
        return self.call("GetTask", p)

    def cancel(self, task_id):
        return self.call("CancelTask", {"id": task_id})


def task_text(task):
    """Join the text of a task's artifacts, or of its status message."""
    parts = [p["text"] for a in task.get("artifacts", []) for p in a["parts"] if "text" in p]
    if not parts and "message" in task["status"]:
        parts = [p["text"] for p in task["status"]["message"]["parts"] if "text" in p]
    return "\n".join(parts)


def a2a_tool(client):
    """Wrap a remote A2A agent as an agent-as-tool callable(ctx, request) -> text."""
    def tool(ctx, request):
        task = client.send(request)
        state = task["status"]["state"]
        if state == "TASK_STATE_COMPLETED":
            return task_text(task)
        return f"[{state}] {task_text(task)}"
    return tool


# --------------------------------------------------------------------------- 4. ledgers
@dataclass
class TaskLedger:
    """Magentic-One's outer-loop ledger: facts (given, to look up, to derive), educated
    guesses, and a plan. My plan is a list of (agent, instruction) pairs; the paper's is prose."""
    task: str
    facts_given: list = field(default_factory=list)
    facts_to_lookup: list = field(default_factory=list)
    facts_to_derive: list = field(default_factory=list)
    guesses: list = field(default_factory=list)
    plan: list = field(default_factory=list)
    version: int = 1

    def render(self):
        out = [f"TASK LEDGER v{self.version}: {self.task}"]
        for label, items in (("given or verified facts", self.facts_given), ("facts to look up", self.facts_to_lookup),
                             ("facts to derive", self.facts_to_derive), ("educated guesses", self.guesses)):
            out.append(f"  {label}: {items}")
        for i, (a, ins) in enumerate(self.plan, 1):
            out.append(f"  plan {i}. {a}: {ins}")
        return "\n".join(out)


@dataclass
class ProgressEntry:
    """The five questions the orchestrator answers in each inner-loop step."""
    step: int
    satisfied: bool
    looping: bool
    progress: bool
    next_speaker: str | None
    instruction: str | None
    reason: str = ""


@dataclass
class LedgerRun:
    status: str
    answer: str | None
    ledgers: list
    progress: list
    conversation: list
    resets: int
    stalls: int


def plan_follower_progress(ledger, conv):
    """A rule-based stand-in for the orchestrator model. Walks the plan in order."""
    step = len(conv) + 1
    ok_prefix = 0
    for i, (agent, ins) in enumerate(ledger.plan):
        if any(c["speaker"] == agent and c["instruction"] == ins and c["ok"] for c in conv):
            ok_prefix = i + 1
        else:
            break
    satisfied = ok_prefix == len(ledger.plan)
    looping = len(conv) >= 2 and all(
        (c["speaker"], c["instruction"], c["result"]) == (conv[-1]["speaker"], conv[-1]["instruction"], conv[-1]["result"])
        for c in conv[-2:])
    progress = bool(conv) and conv[-1]["ok"]
    if not conv:
        progress = True
    nxt = None if satisfied else ledger.plan[ok_prefix]
    return ProgressEntry(step, satisfied, looping, progress, nxt[0] if nxt else None, nxt[1] if nxt else None,
                         f"{ok_prefix}/{len(ledger.plan)} plan steps done")


class MagenticOrchestrator:
    """Outer loop: task ledger. Inner loop: progress ledger plus a stall counter. If the
    counter exceeds `stall_threshold` the orchestrator reflects, re-plans (new ledger
    version) and starts the inner loop again. The threshold 2 is the paper's experiment value."""

    def __init__(self, agents, planner, progress_fn=plan_follower_progress, stall_threshold=2,
                 max_resets=2, max_steps=30):
        self.agents, self.planner, self.progress_fn = agents, planner, progress_fn
        self.stall_threshold, self.max_resets, self.max_steps = stall_threshold, max_resets, max_steps

    def run(self, task):
        ledger = self.planner(task, None, None)
        ledgers, progress, conv = [ledger], [], []
        stalls = resets = total_stalls = steps = 0
        while True:
            entry = self.progress_fn(ledger, conv)
            progress.append(entry)
            if entry.satisfied:
                return LedgerRun("done", conv[-1]["result"] if conv else None, ledgers, progress, conv, resets, total_stalls)
            if steps >= self.max_steps:
                return LedgerRun("max_steps", None, ledgers, progress, conv, resets, total_stalls)
            if entry.looping or not entry.progress:
                stalls += 1
                total_stalls += 1
            if stalls > self.stall_threshold:
                resets += 1
                if resets > self.max_resets:
                    guess = ledger.guesses[0] if ledger.guesses else None
                    return LedgerRun("gave_up", guess, ledgers, progress, conv, resets, total_stalls)
                reflection = (f"stalled {stalls} times at step {entry.step}; last result: "
                              f"{conv[-1]['result'] if conv else None!r}")
                ledger = self.planner(task, ledger, reflection)
                ledgers.append(ledger)
                conv.append({"speaker": "orchestrator", "instruction": "re-plan", "result": reflection, "ok": True})
                stalls = 0
                continue
            agent, ins = entry.next_speaker, entry.instruction
            result = self.agents[agent](ins)
            steps += 1
            conv.append({"speaker": agent, "instruction": ins, "result": result, "ok": not result.startswith("ERROR")})


# --------------------------------------------------------------------------- 5. shared memory
class VersionConflict(RuntimeError):
    pass


class SharedMemory:
    """A key-value store several agents write to. A lock makes each operation atomic.
    put(..., expected_version=n) is optimistic concurrency: it fails if someone wrote first."""

    def __init__(self):
        self._lock = threading.RLock()
        self._data = {}
        self.log = []

    def get(self, key, default=None):
        with self._lock:
            e = self._data.get(key)
            return (e["value"], e["version"]) if e else (default, 0)

    def put(self, key, value, author, expected_version=None):
        with self._lock:
            cur = self._data.get(key, {"version": 0})["version"]
            if expected_version is not None and expected_version != cur:
                raise VersionConflict(f"{key}: expected v{expected_version} but found v{cur}")
            self._data[key] = {"value": value, "version": cur + 1, "author": author}
            self.log.append((key, cur + 1, author))
            return cur + 1

    def update(self, key, fn, author, default=None):
        """Atomic read-modify-write."""
        with self._lock:
            cur = self._data.get(key)
            new = fn(cur["value"] if cur else default)
            return self.put(key, new, author)

    def snapshot(self):
        with self._lock:
            return {k: v["value"] for k, v in self._data.items()}


def race_counter(use_lock, workers=4):
    """Each worker reads the counter, waits for all the others to read, then writes value+1.
    Without a lock all workers read 0, so the final value is 1: three updates are lost."""
    state = {"n": 0}
    lock = threading.Lock()
    barrier = threading.Barrier(workers)

    def unsafe():
        v = state["n"]
        barrier.wait(timeout=5)
        state["n"] = v + 1

    def safe():
        with lock:
            state["n"] = state["n"] + 1

    ts = [threading.Thread(target=safe if use_lock else unsafe) for _ in range(workers)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    return state["n"]


def _words(text):
    return set(re.findall(r"[a-z0-9]+", text.lower()))


class TeamMemory:
    """A toy three-tier team memory. It borrows the NAMES of G-Memory's insight, query and
    interaction graphs, and none of its algorithm: tiers here are plain lists with word overlap."""

    def __init__(self):
        self.interactions, self.queries, self.insights = [], [], []

    def record(self, task, transcript, outcome):
        qid = len(self.queries) + 1
        self.queries.append({"id": qid, "task": task, "outcome": outcome})
        for line in transcript:
            self.interactions.append({"query": qid, "line": line})
        return qid

    def distill(self, query_id, insight):
        if not any(q["id"] == query_id for q in self.queries):
            raise KeyError(f"unknown query {query_id}")
        self.insights.append({"insight": insight, "sources": [query_id]})

    def recall(self, task, k=1):
        scored = sorted(self.queries, key=lambda q: (-len(_words(q["task"]) & _words(task)), q["id"]))
        scored = [q for q in scored if _words(q["task"]) & _words(task)][:k]
        ids = {q["id"] for q in scored}
        return {"queries": scored,
                "insights": [i["insight"] for i in self.insights if set(i["sources"]) & ids],
                "interactions": [x["line"] for x in self.interactions if x["query"] in ids]}


# --------------------------------------------------------------------------- 6. worktrees
def _changes(base, other):
    sm = difflib.SequenceMatcher(None, base, other, autojunk=False)
    return [(i1, i2, other[j1:j2]) for tag, i1, i2, j1, j2 in sm.get_opcodes() if tag != "equal"]


def _apply(base, s, e, changes):
    out, pos = [], s
    for i1, i2, new in changes:
        out += base[pos:i1] + new
        pos = i2
    return out + base[pos:e]


def merge3(base, ours, theirs):
    """Three-way merge of two lists of lines against a common base. Changes that touch
    or overlap are a conflict unless both sides made the same change (git is also
    conservative about touching changes). Returns (merged_lines, n_conflicts)."""
    a, b = _changes(base, ours), _changes(base, theirs)
    out, pos, conflicts = [], 0, 0
    ia = ib = 0
    while ia < len(a) or ib < len(b):
        ca = a[ia] if ia < len(a) else None
        cb = b[ib] if ib < len(b) else None
        if cb is None or (ca is not None and ca[1] < cb[0]):
            out += base[pos:ca[0]] + ca[2]
            pos, ia = ca[1], ia + 1
            continue
        if ca is None or cb[1] < ca[0]:
            out += base[pos:cb[0]] + cb[2]
            pos, ib = cb[1], ib + 1
            continue
        s, e = min(ca[0], cb[0]), max(ca[1], cb[1])
        ga, gb = [ca], [cb]
        ia, ib = ia + 1, ib + 1
        grew = True
        while grew:
            grew = False
            if ia < len(a) and a[ia][0] <= e:
                ga.append(a[ia]); e = max(e, a[ia][1]); ia += 1; grew = True
            if ib < len(b) and b[ib][0] <= e:
                gb.append(b[ib]); e = max(e, b[ib][1]); ib += 1; grew = True
        mine, theirs_ = _apply(base, s, e, ga), _apply(base, s, e, gb)
        out += base[pos:s]
        if mine == theirs_:
            out += mine
        else:
            conflicts += 1
            out += ["<<<<<<< ours\n"] + mine + ["=======\n"] + theirs_ + [">>>>>>> theirs\n"]
        pos = e
    return out + base[pos:], conflicts


def _lines(text):
    return text.splitlines(keepends=True)


class Repo:
    """A tiny repo in a temp dir. main/ is the integration tree. fork(name) copies it to work/<name>/,
    like `git worktree add`, and remembers the base snapshot for three-way merges."""

    def __init__(self, files=None, root=None):
        self.root = root or tempfile.mkdtemp(prefix="repo5a_")
        self.main = os.path.join(self.root, "main")
        os.makedirs(self.main, exist_ok=True)
        for p, t in (files or {}).items():
            self._write(self.main, p, t)
        self.workspaces = {}

    @staticmethod
    def _write(base, path, text):
        full = os.path.join(base, path)
        if os.path.commonpath([os.path.realpath(base), os.path.realpath(full)]) != os.path.realpath(base):
            raise ValueError(f"path escapes the workspace: {path}")
        os.makedirs(os.path.dirname(full), exist_ok=True)
        with open(full, "w") as f:
            f.write(text)

    @staticmethod
    def snapshot_dir(base):
        out = {}
        for d, _, fs in os.walk(base):
            for f in fs:
                p = os.path.join(d, f)
                with open(p) as fh:
                    out[os.path.relpath(p, base)] = fh.read()
        return out

    def snapshot(self):
        return self.snapshot_dir(self.main)

    def fork(self, name):
        if name in self.workspaces:
            raise KeyError(f"workspace {name} exists")
        ws = Workspace(self, name, os.path.join(self.root, "work", name), self.snapshot())
        self.workspaces[name] = ws
        return ws

    def cleanup(self):
        shutil.rmtree(self.root, ignore_errors=True)


class Workspace:
    def __init__(self, repo, name, path, base):
        self.repo, self.name, self.path, self.base = repo, name, path, base
        os.makedirs(path, exist_ok=True)
        for p, t in base.items():
            Repo._write(path, p, t)

    def write(self, rel, text):
        Repo._write(self.path, rel, text)

    def read(self, rel):
        with open(os.path.join(self.path, rel)) as f:
            return f.read()

    def delete(self, rel):
        os.remove(os.path.join(self.path, rel))

    def files(self):
        return Repo.snapshot_dir(self.path)

    def changes(self):
        now, out = self.files(), {}
        for p in sorted(set(now) | set(self.base)):
            if p not in self.base:
                out[p] = "A"
            elif p not in now:
                out[p] = "D"
            elif now[p] != self.base[p]:
                out[p] = "M"
        return out


@dataclass
class MergeReport:
    merged: list = field(default_factory=list)
    conflicts: dict = field(default_factory=dict)
    checks: dict = field(default_factory=dict)

    @property
    def clean(self):
        return not self.conflicts


def merge_workspaces(repo, names, checks=None):
    """Merge workspaces into main/ one after the other. Conflicts are written with markers
    (as git does) and listed in the report. `checks` is {name: fn(files_dict) -> bool}, run
    on the combined result: a clean text merge can still be wrong."""
    rep = MergeReport()
    for name in names:
        ws = repo.workspaces[name]
        theirs_files, main = ws.files(), repo.snapshot()
        for path, kind in ws.changes().items():
            base = ws.base.get(path)
            ours, theirs = main.get(path), theirs_files.get(path)
            if kind == "D":
                if ours is None or ours == base:
                    if ours is not None:
                        os.remove(os.path.join(repo.main, path))
                    rep.merged.append((name, path, "D"))
                else:
                    rep.conflicts.setdefault(path, []).append(f"{name}: deleted a file that main changed")
                continue
            if ours is None and base is not None:
                Repo._write(repo.main, path, theirs)
                rep.conflicts.setdefault(path, []).append(f"{name}: modified a file that main deleted")
                continue
            merged, n = merge3(_lines(base or ""), _lines(ours or ""), _lines(theirs))
            Repo._write(repo.main, path, "".join(merged))
            if n:
                rep.conflicts.setdefault(path, []).append(f"{name}: {n} conflicting region(s)")
            else:
                rep.merged.append((name, path, kind))
    final = repo.snapshot()
    for cname, fn in (checks or {}).items():
        rep.checks[cname] = bool(fn(final))
    return rep


def shared_dir_demo(edits):
    """No isolation: agents write into one directory. The last write wins and earlier work is lost.
    edits: list of (agent, path, text). Returns the final snapshot."""
    d = tempfile.mkdtemp(prefix="shared5a_")
    try:
        for _, p, t in edits:
            Repo._write(d, p, t)
        return Repo.snapshot_dir(d)
    finally:
        shutil.rmtree(d, ignore_errors=True)


# --------------------------------------------------------------------------- 7. failures (MAST-style)
# The 14 modes and 3 categories are from Cemri et al. (MAST). The detectors below are MY rules
# over MY synthetic traces. They are not MAST's LLM annotator and say nothing about its accuracy.
MAST_MODES = {
    "1.1": ("Disobey task specification", "System design"),
    "1.2": ("Disobey role specification", "System design"),
    "1.3": ("Step repetition", "System design"),
    "1.4": ("Loss of conversation history", "System design"),
    "1.5": ("Unaware of termination conditions", "System design"),
    "2.1": ("Conversation reset", "Inter-agent misalignment"),
    "2.2": ("Fail to ask for clarification", "Inter-agent misalignment"),
    "2.3": ("Task derailment", "Inter-agent misalignment"),
    "2.4": ("Information withholding", "Inter-agent misalignment"),
    "2.5": ("Ignored other agent's input", "Inter-agent misalignment"),
    "2.6": ("Reasoning-action mismatch", "Inter-agent misalignment"),
    "3.1": ("Premature termination", "Task verification"),
    "3.2": ("No or incomplete verification", "Task verification"),
    "3.3": ("Incorrect verification", "Task verification"),
}


def mode_category(mode):
    return MAST_MODES[mode][1]


@dataclass
class TeamSpec:
    task: str = "Build a todo CLI that saves tasks between runs"
    roles: dict = field(default_factory=lambda: {
        "planner": {"plan", "ask", "say"}, "coder": {"implement", "run_tests", "ask", "say"},
        "reviewer": {"verify", "say"}})
    topics: frozenset = frozenset({"cli", "storage", "tests", "review"})
    constraints: dict = field(default_factory=lambda: {"persists": lambda a: a.get("persistent_storage") is True})
    required_steps: tuple = ("plan", "implement", "run_tests", "verify")
    checks: tuple = ("tests_pass", "meets_spec")
    ambiguities: tuple = ("storage_format",)


def run_team(spec, faults=()):
    """A scripted planner -> coder -> reviewer team. `faults` is a set of MAST mode ids; each
    one changes the script at one place. Returns the trace: a list of event dicts."""
    F = set(faults)
    unknown = F - set(MAST_MODES)
    if unknown:
        raise ValueError(f"unknown fault ids: {sorted(unknown)}")
    tr = []

    def ev(agent, typ, text, to=None, **data):
        tr.append({"t": len(tr), "agent": agent, "type": typ, "to": to, "text": text, "data": data})

    ev("user", "message", spec.task, to="planner", provides=["persist"])
    ev("planner", "action", "Write the plan", does="plan", intends="plan", topic="logo_design" if "2.3" in F else "cli")
    ev("planner", "message", "Plan: add, list, done commands; tasks must persist", to="coder",
       provides=[] if "2.4" in F else ["persist"])
    if "2.2" not in F:
        ev("coder", "message", "Which storage format should I use?", to="planner", question_about="storage_format")
        ev("planner", "message", "Use a JSON file", to="coder", provides=["storage_format"], recommends="json")
    else:
        ev("planner", "message", "Use a JSON file", to="coder", provides=["storage_format"], recommends="json")
    if "1.4" in F:
        ev("coder", "message", "Which storage format should I use?", to="planner", question_about="storage_format")
    if "2.1" in F:
        ev("coder", "message", "Hello! Let's start from scratch. What are we building?", to="planner")
    art = {"persistent_storage": "1.1" not in F, "format": "pickle" if "2.5" in F else "json"}
    if "2.6" in F:
        ev("coder", "action", "Plan says implement; run the tests instead", does="run_tests", intends="implement",
           result="pass", topic="tests")
    ev("coder", "action", "Implement the CLI", does="implement", intends="implement",
       requires=["persist", "storage_format"], chosen=art["format"], artifact=art,
       assumed=["storage_format"] if "2.2" in F else [], topic="storage")
    if "1.1" in F:
        ev("reviewer", "action", "Verify", does="verify", checked=list(spec.checks), verdict="fail", topic="review")
        art = dict(art, persistent_storage=True)
        ev("coder", "action", "Fix storage", does="implement", intends="implement", artifact=art, topic="storage",
           requires=["persist", "storage_format"], chosen="json")
    if "3.1" not in F:
        ev("coder", "action", "Run the tests", does="run_tests", intends="run_tests",
           result="fail" if "3.3" in F else "pass", topic="tests")
        if "1.3" in F:
            ev("coder", "action", "Run the tests", does="run_tests", intends="run_tests",
               result="fail" if "3.3" in F else "pass", topic="tests")
    if "1.2" in F:
        ev("reviewer", "action", "Rewrite the storage code myself", does="implement", intends="implement",
           artifact=art, topic="storage", chosen="json")
    checked = ["tests_pass"] if "3.2" in F else list(spec.checks)
    ev("reviewer", "action", "Verify", does="verify", checked=checked, verdict="pass", topic="review")
    ev("reviewer", "finish", "Done", done=True)
    if "1.5" in F:
        for i in range(3):
            ev("coder", "action", f"Polish round {i + 1}", does="implement", intends="implement", artifact=art,
               topic="storage", requires=["persist", "storage_format"], chosen="json")
    return tr


@dataclass
class Detection:
    mode: str
    t: int
    evidence: str


def classify(trace, spec):
    """Rule-based MAST-style classifier over a trace. Returns a list of Detection."""
    out = []

    def hit(mode, t, why):
        out.append(Detection(mode, t, why))

    provided = {}                       # fact -> set of recipients (None = everyone)
    asked, finished_at = set(), None
    last_artifact, last_tests = None, None
    seen_sig, prev_sig = set(), None
    spoken = set()
    verify_checked, has_verify = set(), False
    done_steps = set()
    for e in trace:
        d, a, t = e["data"], e["agent"], e["t"]
        if finished_at is not None:
            hit("1.5", t, f"activity after finish at t={finished_at}")
        # 2.1 reset: a greeting-style restart by an agent that already spoke
        if e["type"] == "message" and a in spoken and re.match(r"^(hello|hi|let's start|starting over)", e["text"], re.I):
            hit("2.1", t, "restart-style message mid-conversation")
        if a != "user":
            spoken.add(a)
        # 1.4 asks for something it was already given
        q = d.get("question_about")
        if q:
            got = any(q in f and (to in (None, a)) for f, to in [(f, x) for f, xs in provided.items() for x in xs])
            if got:
                hit("1.4", t, f"asked about {q!r} after it was provided")
            asked.add(q)
        for f in d.get("provides", []):
            provided.setdefault(f, set()).add(e["to"])
        does = d.get("does")
        if e["type"] == "action":
            if does and does not in spec.roles.get(a, set()):
                hit("1.2", t, f"{a} did {does!r}, not allowed for its role")
            if d.get("topic") and d["topic"] not in spec.topics:
                hit("2.3", t, f"topic {d['topic']!r} is not part of the task")
            if d.get("intends") and does and d["intends"] != does:
                hit("2.6", t, f"said {d['intends']!r} but did {does!r}")
            for amb in d.get("assumed", []):
                if amb not in asked:
                    hit("2.2", t, f"assumed {amb!r} without asking")
            for f in d.get("requires", []):
                if not any(to in (None, a) for to in provided.get(f, set())):
                    hit("2.4", t, f"{a} needed {f!r} but nobody told it")
            rec = next((x["data"].get("recommends") for x in reversed(trace[:t]) if x["data"].get("recommends")), None)
            if does == "implement" and rec and d.get("chosen") and d["chosen"] != rec:
                hit("2.5", t, f"chose {d['chosen']!r} against the recommendation {rec!r}")
            sig = (a, does, json.dumps(d.get("artifact"), sort_keys=True), d.get("result"))
            if does == "run_tests" and sig == prev_sig:
                hit("1.3", t, "same step repeated with the same result")
            prev_sig = sig
            if does:
                done_steps.add(does)
            if does == "implement" and d.get("artifact") is not None:
                last_artifact = d["artifact"]
                for cname, fn in spec.constraints.items():
                    if not fn(d["artifact"]):
                        hit("1.1", t, f"artifact violates constraint {cname!r}")
            if does == "run_tests":
                last_tests = d.get("result")
            if does == "verify":
                has_verify = True
                verify_checked |= set(d.get("checked", []))
                broken = last_tests == "fail" or (last_artifact is not None and any(
                    not fn(last_artifact) for fn in spec.constraints.values()))
                if d.get("verdict") == "pass" and broken:
                    hit("3.3", t, "verdict pass while the latest tests or constraints fail")
        if e["type"] == "finish":
            finished_at = t
            missing = [s for s in spec.required_steps if s not in done_steps]
            if missing:
                hit("3.1", t, f"finished before required steps: {missing}")
            if not has_verify or set(spec.checks) - verify_checked:
                hit("3.2", t, f"checks not covered: {sorted(set(spec.checks) - verify_checked)}")
    seen, uniq = set(), []
    for d_ in out:
        if (d_.mode, d_.t) not in seen:
            seen.add((d_.mode, d_.t))
            uniq.append(d_)
    return uniq


def detected_modes(trace, spec):
    return sorted({d.mode for d in classify(trace, spec)})


def failure_profile(detections_by_trace):
    """Count modes and category shares over many traces (lists of mode ids)."""
    modes, cats = {}, {}
    for ms in detections_by_trace:
        for m in ms:
            modes[m] = modes.get(m, 0) + 1
            cats[mode_category(m)] = cats.get(mode_category(m), 0) + 1
    tot = sum(cats.values()) or 1
    return {"modes": dict(sorted(modes.items())), "categories": {c: round(100 * n / tot, 1) for c, n in sorted(cats.items())}}


def injection_study(spec=None):
    """Inject each of the 14 faults alone; report what the classifier finds."""
    spec = spec or TeamSpec()
    rows = []
    for m in sorted(MAST_MODES):
        found = detected_modes(run_team(spec, {m}), spec)
        rows.append({"mode": m, "found": found, "hit": m in found, "extras": [x for x in found if x != m]})
    return {"clean": detected_modes(run_team(spec), spec), "rows": rows}


# --------------------------------------------------------------------------- 8. cost / accuracy simulation
@dataclass(frozen=True)
class SimParams:
    """All numbers are MADE UP. skill: chance one step is right with a fresh context.
    dilution: skill lost per step already held in one agent's context.
    shared_hard: chance an item is hard in the same way for every voter (a shared misconception).
    handoff_loss: chance an orchestrator-to-worker message drops a value the step needs.
    swarm_context_loss: chance a swarm agent rebuilds its context from the board imperfectly.
    verify_catch: chance a checker catches a wrong result (then one retry). Tokens are crude."""
    skill: float = 0.92
    dilution: float = 0.015
    shared_hard: float = 0.03
    handoff_loss: float = 0.10
    swarm_context_loss: float = 0.06
    verify_catch: float = 0.8
    workers: int = 4
    voters: int = 5
    step_tokens: int = 60
    reread_tokens: int = 15
    msg_tokens: int = 40
    board_tokens_per_result: int = 20
    poll_tokens: int = 5


@dataclass
class SimTask:
    kind: str   # "parallel": n independent items. "sequential": item i needs item i-1 right.
    n: int = 8

    def __post_init__(self):
        if self.kind not in ("parallel", "sequential"):
            raise ValueError("kind must be parallel or sequential")
        if self.n < 1:
            raise ValueError("n must be at least 1")


ARCHS = ("single", "single+verify", "independent", "orchestrator", "orchestrator+verify", "swarm", "hybrid")


def _clamp(p):
    return min(1.0, max(0.02, p))


def _trial(arch, task, P, rng):
    n, seq = task.n, task.kind == "sequential"

    def chain(ps):
        ok, prev = [], True
        for p in ps:
            cur = (rng.random() < p) and (prev or not seq)
            ok.append(cur)
            prev = cur
        return ok

    solo = [_clamp(P.skill - P.dilution * i) for i in range(n)]
    solo_tokens = sum(P.step_tokens + i * P.reread_tokens for i in range(n))
    if arch == "single":
        return all(chain(solo)), solo_tokens, n
    if arch == "single+verify":
        ok, prev, tokens = [], True, 0
        for i, p in enumerate(solo):
            got = rng.random() < p
            tokens += P.step_tokens + i * P.reread_tokens + P.step_tokens   # work, then a check by the same agent
            if not got and rng.random() < P.verify_catch:
                got = rng.random() < p
                tokens += P.step_tokens + i * P.reread_tokens
            cur = got and (prev or not seq)
            ok.append(cur)
            prev = cur
        return all(ok), tokens, 2 * n
    if arch == "independent":
        hard = [rng.random() < P.shared_hard for _ in range(n)]
        ps = [0.0 if h else p for h, p in zip(hard, solo)]
        runs = [chain(ps) for _ in range(P.voters)]
        # wrong answers are assumed all different (except on hard items), so a majority needs 2+ correct voters
        final = [sum(r[i] for r in runs) >= 2 for i in range(n)]
        return all(final), P.voters * solo_tokens + P.voters * n * 10, n + 1
    if arch in ("orchestrator", "orchestrator+verify", "hybrid"):
        verify, hybrid = arch == "orchestrator+verify", arch == "hybrid"
        tokens = 2 * P.msg_tokens
        if seq:
            ok, prev = [], True
            lose = 0.0 if hybrid else P.handoff_loss      # a board gives exact values, at a reading cost
            for i in range(n):
                good = P.skill * (1 - lose) * (1 - (P.swarm_context_loss / 2 if hybrid else 0))
                got = rng.random() < good
                tokens += 2 * P.msg_tokens + P.step_tokens + (i * P.board_tokens_per_result if hybrid else 0)
                if verify:
                    tokens += P.msg_tokens + P.step_tokens
                    if not got and rng.random() < P.verify_catch:
                        got = rng.random() < good
                        tokens += 2 * P.msg_tokens + P.step_tokens
                cur = got and prev
                ok.append(cur)
                prev = cur
            return all(ok), tokens, 2 * n + 1
        c = math.ceil(n / P.workers)
        ok = []
        for i in range(n):
            pos = i % c
            p = _clamp(P.skill - P.dilution * pos)
            got = rng.random() < p
            tokens += 2 * P.msg_tokens + pos * P.reread_tokens + P.step_tokens
            if hybrid:
                tokens += (i // P.workers) * P.board_tokens_per_result
            if verify:
                tokens += P.msg_tokens + P.step_tokens
                if not got and rng.random() < P.verify_catch:
                    got = rng.random() < p
                    tokens += 2 * P.msg_tokens + P.step_tokens
            ok.append(got)
        return all(ok), tokens, c + 2
    if arch == "swarm":
        W = P.workers
        if seq:
            ok, prev, tokens = [], True, 0
            for i in range(n):
                got = rng.random() < P.skill * (1 - P.swarm_context_loss)
                tokens += P.msg_tokens + i * P.board_tokens_per_result + P.step_tokens + (W - 1) * P.poll_tokens
                cur = got and prev
                ok.append(cur)
                prev = cur
            return all(ok), tokens, n
        c = math.ceil(n / W)
        ok, tokens = [], c * (W - 1) * P.poll_tokens
        for i in range(n):
            pos = i % c
            ok.append(rng.random() < _clamp(P.skill - P.dilution * pos))
            tokens += P.msg_tokens + (i // W) * P.board_tokens_per_result + pos * P.reread_tokens + P.step_tokens
        return all(ok), tokens, c
    raise ValueError(f"unknown architecture {arch!r}")


def simulate(arch, task, params=None, trials=2000, seed=0):
    """Return {success, tokens, rounds} averaged over trials (success = every item right)."""
    P = params or SimParams()
    rng = random.Random(seed)
    wins = tok = rnd = 0
    for _ in range(trials):
        ok, t, r = _trial(arch, task, P, rng)
        wins += ok
        tok += t
        rnd += r
    return {"arch": arch, "kind": task.kind, "n": task.n, "success": wins / trials,
            "tokens": tok / trials, "rounds": rnd / trials}


def compare_architectures(params=None, n=8, trials=2000, seed=0):
    rows = []
    for kind in ("parallel", "sequential"):
        for arch in ARCHS:
            rows.append(simulate(arch, SimTask(kind, n), params, trials, seed))
    return rows


def format_table(rows):
    lines = [f"{'task':<11}{'architecture':<22}{'success':>8}{'tokens':>9}{'rounds':>8}"]
    for r in rows:
        lines.append(f"{r['kind']:<11}{r['arch']:<22}{r['success']:>8.2f}{r['tokens']:>9.0f}{r['rounds']:>8.1f}")
    return "\n".join(lines)


# --------------------------------------------------------------------------- demo
def _billing_policy(history, ctx):
    return {"say": "Billing here: I refunded $20 to your card."}


def _triage_handoff_policy(history, ctx):
    if not any(m.get("role") == "tool" for m in history):
        return {"call": "transfer_to_billing"}
    return {"say": "unreachable"}


def _triage_tool_policy(history, ctx):
    if not any(m.get("role") == "tool" for m in history):
        return {"call": "billing", "args": {"request": "Refund order 1234, $20"}}
    return {"say": "Triage here: billing says they refunded $20."}


def demo(out=print):
    out("== 1. Handoff vs agent-as-tool (same task, same specialist)")
    billing = Agent("billing", _billing_policy, description="refunds")
    triage_h = Agent("triage", _triage_handoff_policy, handoffs=[billing])
    r = run_swarm(triage_h, [{"role": "user", "content": "Please refund order 1234"}])
    out(f"handoff:      final speaker={r.agent!r}, hops={r.handoffs}, reply={r.reply!r}")
    triage_t = Agent("triage", _triage_tool_policy, tools={"billing": billing.as_tool()})
    hist = [{"role": "user", "content": "Please refund order 1234"}]
    kind, text = run_agent(triage_t, hist)
    out(f"agent-as-tool: final speaker={hist[-1]['agent']!r}, tool saw only {hist[1]['args']['request']!r}, reply={text!r}")

    out("\n== 2. Orchestrator-worker with one flaky worker (retry)")
    calls = {"n": 0}

    def flaky(ins):
        calls["n"] += 1
        if calls["n"] == 1:
            raise WorkerError("timeout")
        return f"notes on {ins.split()[-1]}"

    workers = {"w1": lambda ins: f"notes on {ins.split()[-1]}", "w2": flaky, "w3": lambda ins: f"notes on {ins.split()[-1]}"}
    res = run_orchestrator("compare A B C", lambda t: [("w1", "research A"), ("w2", "research B"), ("w3", "research C")],
                           workers, lambda t, rs: " | ".join(rs), parallel=False)
    out(f"answer={res.answer!r} errors={res.errors} messages={res.meter.messages} flaky worker calls={calls['n']}")

    out("\n== 3. Message board (publish/subscribe with a schema)")
    board = MessageBoard(schemas={"prd": {"goals", "stories"}})
    board.subscribe("architect", ["prd"])
    board.subscribe("engineer", ["design"])
    board.publish("pm", "prd", {"goals": ["todo CLI"], "stories": ["add task"]})
    try:
        board.publish("pm", "prd", "free text is rejected")
    except SchemaError as e:
        out(f"rejected: {e}")
    out(f"architect sees {[m['topic'] for m in board.read_new('architect')]}; engineer sees {[m['topic'] for m in board.read_new('engineer')]}")

    out("\n== 4. A2A over JSON-RPC (in-process)")
    ids = itertools.count(1)
    card = make_agent_card("Flight Agent", "Books flights", "https://flights.example.com/a2a",
                           [{"id": "book", "name": "Book flight", "description": "Books a flight", "tags": ["travel"]}])

    def handler(text, info):
        if "from" not in " ".join(info["history"]).lower():
            return "input_required", "Where would you like to fly from and to?"
        return "completed", "Booked SFO to JFK"

    srv = A2AServer(card, handler, clock=lambda: "2026-10-28T10:00:00Z", id_factory=lambda: f"id-{next(ids)}")
    mids = itertools.count(1)
    cl = A2AClient(srv, message_ids=lambda: f"msg-{next(mids)}")
    cl.discover()
    t1 = cl.send("Book me a flight")
    t2 = cl.send("From SFO to JFK", task_id=t1["id"])
    out("request : " + json.dumps(cl.log[0][0]))
    out("response: " + json.dumps(cl.log[0][1]))
    out(f"turn 1 state={t1['status']['state']}, turn 2 state={t2['status']['state']}, artifact={task_text(t2)!r}")
    try:
        cl.get_task("nope")
    except A2AError as e:
        out("error   : " + json.dumps(cl.log[-1][1]))

    out("\n== 5. Magentic-One-style ledgers (a stuck agent triggers a re-plan)")

    def planner(task, ledger, reflection):
        if ledger is None:
            return TaskLedger(task, facts_given=["the page needs a login"], facts_to_lookup=["the price"],
                              guesses=["about $10"], plan=[("WebSurfer", "open the page"), ("Coder", "sum the prices")])
        return TaskLedger(task, facts_given=ledger.facts_given + [reflection], facts_to_lookup=ledger.facts_to_lookup,
                          guesses=ledger.guesses, plan=[("FileSurfer", "open the cached page"), ("Coder", "sum the prices")],
                          version=ledger.version + 1)

    team = {"WebSurfer": lambda ins: "ERROR: login wall", "FileSurfer": lambda ins: "page text with prices 4 and 6",
            "Coder": lambda ins: "10"}
    run = MagenticOrchestrator(team, planner).run("What do the two items cost together?")
    out(f"status={run.status} answer={run.answer!r} resets={run.resets} stalls={run.stalls} ledger versions={len(run.ledgers)}")
    out(run.ledgers[-1].render())

    out("\n== 6. Shared memory and locks")
    out(f"4 workers, no lock: counter = {race_counter(False)} (4 were expected)")
    out(f"4 workers, lock:    counter = {race_counter(True)}")
    mem = SharedMemory()
    v = mem.put("plan", "v1", "alice")
    try:
        mem.put("plan", "v2 by bob", "bob", expected_version=0)
    except VersionConflict as e:
        out(f"stale write rejected: {e}")
    out(f"alice then writes with the right version -> v{mem.put('plan', 'v2', 'alice', expected_version=v)}")

    out("\n== 7. Worktree isolation and merge")
    repo = Repo({"app.py": "def add(a, b):\n    return a + b\n\n\ndef sub(a, b):\n    return a - b\n"})
    a, b = repo.fork("agent-a"), repo.fork("agent-b")
    a.write("app.py", a.read("app.py").replace("return a + b", "return a + b  # checked"))
    b.write("app.py", b.read("app.py").replace("return a - b", "return b - a"))
    rep = merge_workspaces(repo, ["agent-a", "agent-b"])
    out(f"edits to different functions: clean={rep.clean}, merged={[(n, p) for n, p, _ in rep.merged]}")
    repo2 = Repo({"app.py": "x = 1\n"})
    c, d = repo2.fork("c"), repo2.fork("d")
    c.write("app.py", "x = 2\n")
    d.write("app.py", "x = 3\n")
    rep2 = merge_workspaces(repo2, ["c", "d"])
    out(f"both edit the same line: clean={rep2.clean}, conflicts={rep2.conflicts}")
    out("main now contains:\n" + repo2.snapshot()["app.py"].rstrip())
    shared = shared_dir_demo([("c", "app.py", "x = 2\n"), ("d", "app.py", "x = 3\n")])
    out(f"same edits with no isolation: file = {shared['app.py']!r} (c's work is silently gone)")
    repo.cleanup(); repo2.cleanup()

    out("\n== 8. Failure injection and a MAST-style classifier (rules over synthetic traces)")
    study = injection_study()
    out(f"clean run: found {study['clean'] or 'nothing'}")
    for r_ in study["rows"]:
        name, cat = MAST_MODES[r_["mode"]]
        extra = f"  (also {r_['extras']})" if r_["extras"] else ""
        out(f"inject {r_['mode']} {name:<36} -> found {r_['found']}{extra}")

    out("\n== 9. Single vs multi-agent: cost and accuracy (MADE-UP parameters, 2000 trials each)")
    out(format_table(compare_architectures()))


if __name__ == "__main__":
    demo()

Run the demo

$ python multi_agent_toolkit.py
== 1. Handoff vs agent-as-tool (same task, same specialist)
handoff:      final speaker='billing', hops=[('triage', 'billing')], reply='Billing here: I refunded $20 to your card.'
agent-as-tool: final speaker='triage', tool saw only 'Refund order 1234, $20', reply='Triage here: billing says they refunded $20.'

== 2. Orchestrator-worker with one flaky worker (retry)
answer='notes on A | notes on B | notes on C' errors=[] messages=6 flaky worker calls=2

== 3. Message board (publish/subscribe with a schema)
rejected: topic 'prd' needs keys ['goals', 'stories']
architect sees ['prd']; engineer sees []

== 4. A2A over JSON-RPC (in-process)
request : {"jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": {"message": {"messageId": "msg-1", "role": "ROLE_USER", "parts": [{"text": "Book me a flight"}]}}}
response: {"jsonrpc": "2.0", "id": 1, "result": {"task": {"id": "id-2", "contextId": "id-1", "status": {"state": "TASK_STATE_INPUT_REQUIRED", "timestamp": "2026-10-28T10:00:00Z", "message": {"messageId": "id-3", "role": "ROLE_AGENT", "parts": [{"text": "Where would you like to fly from and to?"}]}}, "history": [{"messageId": "msg-1", "role": "ROLE_USER", "parts": [{"text": "Book me a flight"}], "taskId": "id-2", "contextId": "id-1"}, {"messageId": "id-3", "role": "ROLE_AGENT", "parts": [{"text": "Where would you like to fly from and to?"}], "taskId": "id-2", "contextId": "id-1"}]}}}
turn 1 state=TASK_STATE_INPUT_REQUIRED, turn 2 state=TASK_STATE_COMPLETED, artifact='Booked SFO to JFK'
error   : {"jsonrpc": "2.0", "id": 3, "error": {"code": -32001, "message": "Task not found", "data": [{"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "TASK_NOT_FOUND", "domain": "a2a-protocol.org", "metadata": {"taskId": "nope"}}]}}

== 5. Magentic-One-style ledgers (a stuck agent triggers a re-plan)
status=done answer='10' resets=1 stalls=3 ledger versions=2
TASK LEDGER v2: What do the two items cost together?
  given or verified facts: ['the page needs a login', "stalled 3 times at step 4; last result: 'ERROR: login wall'"]
  facts to look up: ['the price']
  facts to derive: []
  educated guesses: ['about $10']
  plan 1. FileSurfer: open the cached page
  plan 2. Coder: sum the prices

== 6. Shared memory and locks
4 workers, no lock: counter = 1 (4 were expected)
4 workers, lock:    counter = 4
stale write rejected: plan: expected v0 but found v1
alice then writes with the right version -> v2

== 7. Worktree isolation and merge
edits to different functions: clean=True, merged=[('agent-a', 'app.py'), ('agent-b', 'app.py')]
both edit the same line: clean=False, conflicts={'app.py': ['d: 1 conflicting region(s)']}
main now contains:
<<<<<<< ours
x = 2
=======
x = 3
>>>>>>> theirs
same edits with no isolation: file = 'x = 3\n' (c's work is silently gone)

== 8. Failure injection and a MAST-style classifier (rules over synthetic traces)
clean run: found nothing
inject 1.1 Disobey task specification           -> found ['1.1']
inject 1.2 Disobey role specification           -> found ['1.2']
inject 1.3 Step repetition                      -> found ['1.3']
inject 1.4 Loss of conversation history         -> found ['1.4']
inject 1.5 Unaware of termination conditions    -> found ['1.5']
inject 2.1 Conversation reset                   -> found ['2.1']
inject 2.2 Fail to ask for clarification        -> found ['2.2']
inject 2.3 Task derailment                      -> found ['2.3']
inject 2.4 Information withholding              -> found ['2.4']
inject 2.5 Ignored other agent's input          -> found ['2.5']
inject 2.6 Reasoning-action mismatch            -> found ['2.6']
inject 3.1 Premature termination                -> found ['3.1']
inject 3.2 No or incomplete verification        -> found ['3.2']
inject 3.3 Incorrect verification               -> found ['3.3']

== 9. Single vs multi-agent: cost and accuracy (MADE-UP parameters, 2000 trials each)
task       architecture           success   tokens  rounds
parallel   single                    0.32      900     8.0
parallel   single+verify             0.72     1483    16.0
parallel   independent               0.79     4900     9.0
parallel   orchestrator              0.48     1260     4.0
parallel   orchestrator+verify       0.82     2136     4.0
parallel   swarm                     0.48      970     2.0
parallel   hybrid                    0.48     1340     4.0
sequential single                    0.32      900     8.0
sequential single+verify             0.72     1483    16.0
sequential independent               0.40     4900     9.0
sequential orchestrator              0.22     1200    17.0
sequential orchestrator+verify       0.63     2150    17.0
sequential swarm                     0.32     1480     8.0
sequential hybrid                    0.41     1760    17.0

Exercises

1. What the specialist sees. Run run_swarm with a history_filter that keeps only the last user message. Print what the billing agent sees with and without the filter. When would you want the filter, and what could it hide [16]?
2. A handoff loop. Make two agents that always hand off to each other. What exception stops them, and after how many turns? Which MAST modes does this resemble, and what would a real system need to detect it [2]?
3. A thin request. Make the triage agent send only "refund" as the request in the agent-as-tool run. What does the specialist reply? Add a check that rejects a request with no order number, as a structured tool input would [17].
4. Retries and failure. Run the orchestrator with max_retries=0, and then with a worker that always fails. What does synthesize receive for the failed piece? Rewrite it so that it reports a gap and does not hide it [14].
5. A hybrid team. Run run_hybrid with workers that use a peer's note in round 2. Compare the messages and the result with the plain orchestrator. When is the extra reading worth it [31]?
6. Typed messages. Add a design topic that needs three keys. Show a rejected post, a private note with to=, and a second read_new call that returns nothing. Why does MetaGPT insist on structured outputs [5]?
7. Deadlock and a stuck owner. Create two tasks that depend on each other and run run_board_swarm. Then make an agent fail halfway through a task and see that the task stays owned. Add a lease (a time after which a claim expires), and compare with Symphony's stall timeout [26].
8. Break the card. Write an Agent Card with a snake_case key and a skill with no tags, and read the errors from validate_agent_card. Then fix it. Which required fields does the spec list [21]?
9. Provoke each error. Using A2AServer.handle, send requests that give −32700, −32601, −32602 (a contextId that does not match), −32001, −32004 (a message to a completed task) and −32009 (a wrong A2A-Version header). Compare each reply with the spec's error table [21] [23].
10. The same message twice. Send one messageId two times. What does the server return? The spec lets a server do this, and does not require it. What could go wrong in a real system if the server did not [21]?
11. A2A as a tool. Use a2a_tool to give a triage agent a remote flight agent. Walk a two-turn "input required" flow through it. Where does the caller have to carry the taskId, and what happens if it forgets [17] [21]?
12. Tune the stall counter. Run the ledger demo with stall_threshold 0, 2 and 4. How many times is the failing agent called before the re-plan? What do you give up with a low threshold and with a high one [4]?
13. Give up gracefully. Write a planner that always returns the same plan, so that every re-plan fails. What status does the run end with, and what does it return? Why does the ledger keep an "educated guess" [4]?
14. Make the lock fix it. Run race_counter with 2 and 8 workers, with and without the lock. Then write a retry loop around SharedMemory.put(..., expected_version=...) so that 8 threads each add one and the total is 8, without using update. Compare with git's lock files in Carlini's project [24].
15. What the dilution does. Run the single agent and the orchestrator on the parallel task with SimParams(dilution=0). Which wins on success, and which on tokens? Explain, using the arithmetic of section 2.13.
16. The cost of a lost value. Sweep handoff_loss from 0 to 0.3 on the sequential task. At what value does the unchecked orchestrator stop beating the single agent? Is that crossing a fact about agents, or about my model [3]?
17. Seeds and noise. Run compare_architectures with seeds 1 to 10, and report the mean and range of each row. Which gaps in the table of 2.13 are larger than the noise? The demo prints 0.79 for independent voting, and the exact value is about 0.77.
18. Correlated errors. Sweep shared_hard from 0 to 0.2. At what value does independent voting stop beating orchestrator+verify on the parallel task? Relate your answer to the conformity failures in the Anthropic multiagent post [27].
19. Touching edits. Use merge3 on three cases: edits far apart, edits on adjacent lines, and the same edit on both sides. Which merge cleanly, and why does a line-based merge treat adjacent edits as a conflict?
20. A clean merge that breaks. Fork two workspaces. One renames a function in one file. The other adds a call to the old name in a second file. Show that merge_workspaces reports no conflict. Then write a checks function that catches the break. What does this say about "tests on the combined result" [31]?
21. Delete against edit. One workspace deletes a file and another edits it. What does the report say? Which side would you want to win in a real project, and who should decide?
22. Two faults at once. Call run_team with faults {"2.2", "3.2"} and classify the trace. Are both found? Then invent a rule for a fifteenth failure of your own (for example, a premature handoff) and add a fault and a detector for it [2].
23. Fool the classifier. Write a trace by hand that a person would call "task derailment" but that the rules miss. Change the rule. What new false alarms does your change create? Why does the MAST paper use an LLM judge, and what does that cost [2]?
24. Add an architecture. Add a design of your own to the simulation (for example, an orchestrator with two workers, or a pipeline with a check after each stage). State your assumptions in the comment, and compare it with the others on both tasks.

The tests: test_multi_agent_toolkit.py

The 170 tests check the mocks and the token proxy; the agents (handoff tool names, unknown tools, the step limit, agent-as-tool, the swarm loop and its filter); the orchestrator (order, retries, errors); the message board (schemas, subscriptions, cursors, private notes); the task board (atomic claims under threads, dependencies, owner checks, deadlock); the Agent Card and its validator; the A2A shapes and errors (every code in the table, the version header, role and part rules); the task life cycle (input required, terminal states, cancel, list and pages, idempotent message ids); the ledgers (the five-question entries, the stall threshold, re-plan, give-up); shared memory (locks, versions, the race); the three-way merge (touching changes, identical changes, delete against edit, markers); the workspaces (isolation, path escape, checks, the shared-folder loss); the failure injection (each of the 14 modes found alone, no extras, a clean run); the simulation (closed-form checks, rounds, voting, tokens) and the demo and its output. I also changed the code in 71 ways, one at a time (for example breaking a claim, a version check, an error code, the merge rule for touching edits, a detector, a token formula), and checked that at least one test failed each time. All 71 changes were caught. That shows the tests are sensitive to those changes. It does not show that the code has no bugs, and the 71 changes are the ones that I thought of. I did not run the tests on any other version of Python.

$ python test_multi_agent_toolkit.py -v
test_client_message_must_use_role_user (__main__.TestA2AErrorsAndShapes.test_client_message_must_use_role_user) ... ok
test_client_raises_a2a_error_with_the_right_name (__main__.TestA2AErrorsAndShapes.test_client_raises_a2a_error_with_the_right_name) ... ok
test_error_codes_match_spec_table (__main__.TestA2AErrorsAndShapes.test_error_codes_match_spec_table) ... ok
test_handler_exception_marks_the_task_failed (__main__.TestA2AErrorsAndShapes.test_handler_exception_marks_the_task_failed) ... ok
test_handler_returning_a_bad_kind_is_invalid_agent_response (__main__.TestA2AErrorsAndShapes.test_handler_returning_a_bad_kind_is_invalid_agent_response) ... ok
test_invalid_json_gives_parse_error_with_null_id (__main__.TestA2AErrorsAndShapes.test_invalid_json_gives_parse_error_with_null_id) ... ok
test_known_but_unimplemented_method_is_unsupported_not_unknown (__main__.TestA2AErrorsAndShapes.test_known_but_unimplemented_method_is_unsupported_not_unknown) ... ok
test_missing_method_or_version_is_invalid_request (__main__.TestA2AErrorsAndShapes.test_missing_method_or_version_is_invalid_request) ... ok
test_old_style_method_name_is_not_accepted (__main__.TestA2AErrorsAndShapes.test_old_style_method_name_is_not_accepted) ... ok
test_params_must_be_an_object (__main__.TestA2AErrorsAndShapes.test_params_must_be_an_object) ... ok
test_part_must_have_exactly_one_content_field (__main__.TestA2AErrorsAndShapes.test_part_must_have_exactly_one_content_field) ... ok
test_response_has_result_xor_error (__main__.TestA2AErrorsAndShapes.test_response_has_result_xor_error) ... ok
test_streaming_without_capability_is_unsupported_operation (__main__.TestA2AErrorsAndShapes.test_streaming_without_capability_is_unsupported_operation) ... ok
test_supported_version_header_passes (__main__.TestA2AErrorsAndShapes.test_supported_version_header_passes) ... ok
test_task_not_found_error_info_matches_the_spec_example (__main__.TestA2AErrorsAndShapes.test_task_not_found_error_info_matches_the_spec_example) ... ok
test_unknown_method_is_method_not_found_and_id_is_echoed (__main__.TestA2AErrorsAndShapes.test_unknown_method_is_method_not_found_and_id_is_echoed) ... ok
test_unsupported_media_type_is_content_type_error (__main__.TestA2AErrorsAndShapes.test_unsupported_media_type_is_content_type_error) ... ok
test_unsupported_version_header_is_rejected (__main__.TestA2AErrorsAndShapes.test_unsupported_version_header_is_rejected) ... ok
test_validation_errors_carry_a_bad_request_detail (__main__.TestA2AErrorsAndShapes.test_validation_errors_carry_a_bad_request_detail) ... ok
test_a2a_agent_as_a_tool (__main__.TestA2ATasks.test_a2a_agent_as_a_tool) ... ok
test_cancel_a_completed_task_is_not_cancelable (__main__.TestA2ATasks.test_cancel_a_completed_task_is_not_cancelable) ... ok
test_cancel_a_submitted_task (__main__.TestA2ATasks.test_cancel_a_submitted_task) ... ok
test_cancel_unknown_task_and_missing_id (__main__.TestA2ATasks.test_cancel_unknown_task_and_missing_id) ... ok
test_client_context_id_is_kept (__main__.TestA2ATasks.test_client_context_id_is_kept) ... ok
test_client_round_trip_and_wire_log (__main__.TestA2ATasks.test_client_round_trip_and_wire_log) ... ok
test_completed_task_shape (__main__.TestA2ATasks.test_completed_task_shape) ... ok
test_duplicate_message_id_is_idempotent (__main__.TestA2ATasks.test_duplicate_message_id_is_idempotent) ... ok
test_every_response_is_json_serialisable (__main__.TestA2ATasks.test_every_response_is_json_serialisable) ... ok
test_failed_and_rejected_states_carry_a_message (__main__.TestA2ATasks.test_failed_and_rejected_states_carry_a_message) ... ok
test_history_length_zero_omits_history_and_n_keeps_last_n (__main__.TestA2ATasks.test_history_length_zero_omits_history_and_n_keeps_last_n) ... ok
test_input_required_then_follow_up_completes_the_same_task (__main__.TestA2ATasks.test_input_required_then_follow_up_completes_the_same_task) ... ok
test_list_tasks_filters_and_pages (__main__.TestA2ATasks.test_list_tasks_filters_and_pages) ... ok
test_list_tasks_hides_artifacts_unless_asked (__main__.TestA2ATasks.test_list_tasks_hides_artifacts_unless_asked) ... ok
test_list_tasks_page_size_bounds (__main__.TestA2ATasks.test_list_tasks_page_size_bounds) ... ok
test_message_to_a_terminal_task_is_unsupported_operation (__main__.TestA2ATasks.test_message_to_a_terminal_task_is_unsupported_operation) ... ok
test_mismatched_context_id_is_rejected (__main__.TestA2ATasks.test_mismatched_context_id_is_rejected) ... ok
test_return_immediately_gives_submitted_and_run_pending_finishes (__main__.TestA2ATasks.test_return_immediately_gives_submitted_and_run_pending_finishes) ... ok
test_returned_task_is_a_copy (__main__.TestA2ATasks.test_returned_task_is_a_copy) ... ok
test_server_generates_task_and_context_ids (__main__.TestA2ATasks.test_server_generates_task_and_context_ids) ... ok
test_unknown_task_id_in_message_is_task_not_found (__main__.TestA2ATasks.test_unknown_task_id_in_message_is_task_not_found) ... ok
test_card_is_served_at_the_well_known_path_only (__main__.TestAgentCard.test_card_is_served_at_the_well_known_path_only) ... ok
test_client_discovery_rejects_a_card_without_jsonrpc (__main__.TestAgentCard.test_client_discovery_rejects_a_card_without_jsonrpc) ... ok
test_empty_interfaces_and_incomplete_skill_are_flagged (__main__.TestAgentCard.test_empty_interfaces_and_incomplete_skill_are_flagged) ... ok
test_made_card_is_valid_and_camel_case (__main__.TestAgentCard.test_made_card_is_valid_and_camel_case) ... ok
test_missing_required_fields_are_listed (__main__.TestAgentCard.test_missing_required_fields_are_listed) ... ok
test_snake_case_keys_are_flagged (__main__.TestAgentCard.test_snake_case_keys_are_flagged) ... ok
test_task_states_and_terminal_sets_match_the_spec (__main__.TestAgentCard.test_task_states_and_terminal_sets_match_the_spec) ... ok
test_agent_as_tool_cannot_hand_off (__main__.TestAgents.test_agent_as_tool_cannot_hand_off) ... ok
test_agent_as_tool_keeps_control_with_the_caller (__main__.TestAgents.test_agent_as_tool_keeps_control_with_the_caller) ... ok
test_agent_as_tool_sees_only_the_request_string (__main__.TestAgents.test_agent_as_tool_sees_only_the_request_string) ... ok
test_bad_policy_output_is_rejected (__main__.TestAgents.test_bad_policy_output_is_rejected) ... ok
test_handoff_moves_control_and_final_speaker_is_specialist (__main__.TestAgents.test_handoff_moves_control_and_final_speaker_is_specialist) ... ok
test_handoff_tool_is_named_transfer_to_agent (__main__.TestAgents.test_handoff_tool_is_named_transfer_to_agent) ... ok
test_history_filter_limits_what_the_next_agent_sees (__main__.TestAgents.test_history_filter_limits_what_the_next_agent_sees) ... ok
test_ping_pong_handoff_loop_is_stopped (__main__.TestAgents.test_ping_pong_handoff_loop_is_stopped) ... ok
test_policy_that_never_speaks_hits_the_step_limit (__main__.TestAgents.test_policy_that_never_speaks_hits_the_step_limit) ... ok
test_swarm_keeps_the_shared_history_across_handoffs (__main__.TestAgents.test_swarm_keeps_the_shared_history_across_handoffs) ... ok
test_tool_results_are_added_to_history_in_order (__main__.TestAgents.test_tool_results_are_added_to_history_in_order) ... ok
test_unknown_tool_becomes_an_error_message_not_a_crash (__main__.TestAgents.test_unknown_tool_becomes_an_error_message_not_a_crash) ... ok
test_board_is_safe_under_many_threads (__main__.TestBoards.test_board_is_safe_under_many_threads) ... ok
test_direct_message_reaches_only_the_addressee (__main__.TestBoards.test_direct_message_reaches_only_the_addressee) ... ok
test_message_ids_are_sequential_and_cause_by_is_kept (__main__.TestBoards.test_message_ids_are_sequential_and_cause_by_is_kept) ... ok
test_read_new_advances_the_cursor (__main__.TestBoards.test_read_new_advances_the_cursor) ... ok
test_schema_rejects_free_text_and_missing_keys (__main__.TestBoards.test_schema_rejects_free_text_and_missing_keys) ... ok
test_subscriber_sees_only_its_topics_and_not_its_own_messages (__main__.TestBoards.test_subscriber_sees_only_its_topics_and_not_its_own_messages) ... ok
test_command_line_runs (__main__.TestDemo.test_command_line_runs) ... ok
test_demo_is_deterministic (__main__.TestDemo.test_demo_is_deterministic) ... ok
test_demo_runs_and_has_all_nine_sections (__main__.TestDemo.test_demo_runs_and_has_all_nine_sections) ... ok
test_a_trace_with_no_verification_at_all_is_flagged (__main__.TestFailures.test_a_trace_with_no_verification_at_all_is_flagged) ... ok
test_category_follows_the_id_prefix (__main__.TestFailures.test_category_follows_the_id_prefix) ... ok
test_classifier_uses_the_spec_constraints (__main__.TestFailures.test_classifier_uses_the_spec_constraints) ... ok
test_clean_run_has_no_detections (__main__.TestFailures.test_clean_run_has_no_detections) ... ok
test_detections_carry_evidence_and_time (__main__.TestFailures.test_detections_carry_evidence_and_time) ... ok
test_each_fault_is_detected_without_any_unrelated_mode (__main__.TestFailures.test_each_fault_is_detected_without_any_unrelated_mode) ... ok
test_each_injected_fault_is_detected_alone (__main__.TestFailures.test_each_injected_fault_is_detected_alone) ... ok
test_failure_profile_counts_modes_and_category_shares (__main__.TestFailures.test_failure_profile_counts_modes_and_category_shares) ... ok
test_failure_profile_of_nothing_is_empty (__main__.TestFailures.test_failure_profile_of_nothing_is_empty) ... ok
test_injection_study_covers_all_14 (__main__.TestFailures.test_injection_study_covers_all_14) ... ok
test_premature_termination_is_not_the_same_as_missing_verification (__main__.TestFailures.test_premature_termination_is_not_the_same_as_missing_verification) ... ok
test_role_table_is_what_decides_a_role_violation (__main__.TestFailures.test_role_table_is_what_decides_a_role_violation) ... ok
test_there_are_exactly_14_modes_in_3_categories (__main__.TestFailures.test_there_are_exactly_14_modes_in_3_categories) ... ok
test_trace_events_have_increasing_time_and_known_types (__main__.TestFailures.test_trace_events_have_increasing_time_and_known_types) ... ok
test_two_faults_together_are_both_found (__main__.TestFailures.test_two_faults_together_are_both_found) ... ok
test_unknown_fault_id_is_an_error (__main__.TestFailures.test_unknown_fault_id_is_an_error) ... ok
test_a_failing_agent_is_called_threshold_plus_one_times_before_the_replan (__main__.TestLedgers.test_a_failing_agent_is_called_threshold_plus_one_times_before_the_replan) ... ok
test_gives_up_after_max_resets_with_the_best_guess (__main__.TestLedgers.test_gives_up_after_max_resets_with_the_best_guess) ... ok
test_happy_path_needs_no_replan (__main__.TestLedgers.test_happy_path_needs_no_replan) ... ok
test_ledger_render_lists_facts_and_plan (__main__.TestLedgers.test_ledger_render_lists_facts_and_plan) ... ok
test_loop_detection_needs_two_identical_entries (__main__.TestLedgers.test_loop_detection_needs_two_identical_entries) ... ok
test_max_steps_guard (__main__.TestLedgers.test_max_steps_guard) ... ok
test_progress_entry_answers_the_five_questions (__main__.TestLedgers.test_progress_entry_answers_the_five_questions) ... ok
test_progress_ledger_records_every_inner_step (__main__.TestLedgers.test_progress_ledger_records_every_inner_step) ... ok
test_replan_leaves_a_marker_in_the_conversation_so_the_new_plan_starts_fresh (__main__.TestLedgers.test_replan_leaves_a_marker_in_the_conversation_so_the_new_plan_starts_fresh) ... ok
test_stall_threshold_controls_how_many_failures_are_tolerated (__main__.TestLedgers.test_stall_threshold_controls_how_many_failures_are_tolerated) ... ok
test_stuck_agent_triggers_one_replan_then_succeeds (__main__.TestLedgers.test_stuck_agent_triggers_one_replan_then_succeeds) ... ok
test_changes_far_apart_merge_cleanly (__main__.TestMerge3.test_changes_far_apart_merge_cleanly) ... ok
test_delete_vs_edit_of_the_same_line_conflicts (__main__.TestMerge3.test_delete_vs_edit_of_the_same_line_conflicts) ... ok
test_different_changes_to_the_same_line_conflict_with_markers (__main__.TestMerge3.test_different_changes_to_the_same_line_conflict_with_markers) ... ok
test_empty_base_add_add (__main__.TestMerge3.test_empty_base_add_add) ... ok
test_no_changes (__main__.TestMerge3.test_no_changes) ... ok
test_one_line_apart_still_merges (__main__.TestMerge3.test_one_line_apart_still_merges) ... ok
test_one_sided_change_is_taken (__main__.TestMerge3.test_one_sided_change_is_taken) ... ok
test_same_change_on_both_sides_is_not_a_conflict (__main__.TestMerge3.test_same_change_on_both_sides_is_not_a_conflict) ... ok
test_touching_changes_conflict (__main__.TestMerge3.test_touching_changes_conflict) ... ok
test_two_clean_edits_and_one_conflict_in_one_file (__main__.TestMerge3.test_two_clean_edits_and_one_conflict_in_one_file) ... ok
test_two_different_insertions_at_the_same_point_conflict (__main__.TestMerge3.test_two_different_insertions_at_the_same_point_conflict) ... ok
test_cost_meter_adds_up (__main__.TestMocks.test_cost_meter_adds_up) ... ok
test_scripted_llm_raises_when_script_ends (__main__.TestMocks.test_scripted_llm_raises_when_script_ends) ... ok
test_scripted_llm_replays_and_records (__main__.TestMocks.test_scripted_llm_replays_and_records) ... ok
test_token_proxy_counts_words_and_punctuation (__main__.TestMocks.test_token_proxy_counts_words_and_punctuation) ... ok
test_failed_worker_is_retried_once (__main__.TestOrchestrator.test_failed_worker_is_retried_once) ... ok
test_hybrid_workers_see_peer_notes_in_round_two (__main__.TestOrchestrator.test_hybrid_workers_see_peer_notes_in_round_two) ... ok
test_max_retries_zero_means_no_second_try (__main__.TestOrchestrator.test_max_retries_zero_means_no_second_try) ... ok
test_meter_counts_messages_both_ways (__main__.TestOrchestrator.test_meter_counts_messages_both_ways) ... ok
test_results_keep_plan_order_even_in_parallel (__main__.TestOrchestrator.test_results_keep_plan_order_even_in_parallel) ... ok
test_worker_that_always_fails_is_reported_and_others_survive (__main__.TestOrchestrator.test_worker_that_always_fails_is_reported_and_others_survive) ... ok
test_workers_really_run_in_parallel_threads (__main__.TestOrchestrator.test_workers_really_run_in_parallel_threads) ... ok
test_atomic_update_loses_nothing_under_threads (__main__.TestSharedMemory.test_atomic_update_loses_nothing_under_threads) ... ok
test_distilling_an_unknown_query_fails (__main__.TestSharedMemory.test_distilling_an_unknown_query_fails) ... ok
test_log_records_authors_and_snapshot_is_a_copy (__main__.TestSharedMemory.test_log_records_authors_and_snapshot_is_a_copy) ... ok
test_stale_write_is_rejected (__main__.TestSharedMemory.test_stale_write_is_rejected) ... ok
test_team_memory_recall_of_unrelated_task_is_empty (__main__.TestSharedMemory.test_team_memory_recall_of_unrelated_task_is_empty) ... ok
test_team_memory_recalls_similar_tasks_with_insights (__main__.TestSharedMemory.test_team_memory_recalls_similar_tasks_with_insights) ... ok
test_unlocked_race_loses_updates_and_the_lock_fixes_it (__main__.TestSharedMemory.test_unlocked_race_loses_updates_and_the_lock_fixes_it) ... ok
test_versions_increase_and_get_returns_value_and_version (__main__.TestSharedMemory.test_versions_increase_and_get_returns_value_and_version) ... ok
test_write_with_expected_zero_creates_exactly_once (__main__.TestSharedMemory.test_write_with_expected_zero_creates_exactly_once) ... ok
test_a_broken_chain_hurts_sequential_voting_more_than_parallel_voting (__main__.TestSimulation.test_a_broken_chain_hurts_sequential_voting_more_than_parallel_voting) ... ok
test_a_single_agent_with_a_check_beats_an_unchecked_team_on_sequential_work (__main__.TestSimulation.test_a_single_agent_with_a_check_beats_an_unchecked_team_on_sequential_work) ... ok
test_checking_every_item_costs_at_least_one_message_and_one_step_per_item (__main__.TestSimulation.test_checking_every_item_costs_at_least_one_message_and_one_step_per_item) ... ok
test_comparison_table_has_every_row_and_formats (__main__.TestSimulation.test_comparison_table_has_every_row_and_formats) ... ok
test_default_orchestrator_is_worse_than_single_on_sequential_and_better_on_parallel (__main__.TestSimulation.test_default_orchestrator_is_worse_than_single_on_sequential_and_better_on_parallel) ... ok
test_handoff_loss_hurts_the_sequential_orchestrator (__main__.TestSimulation.test_handoff_loss_hurts_the_sequential_orchestrator) ... ok
test_independent_voting_costs_about_voters_times_a_single_run (__main__.TestSimulation.test_independent_voting_costs_about_voters_times_a_single_run) ... ok
test_more_skill_means_more_success (__main__.TestSimulation.test_more_skill_means_more_success) ... ok
test_multi_agent_systems_use_more_tokens_than_one_agent_by_default (__main__.TestSimulation.test_multi_agent_systems_use_more_tokens_than_one_agent_by_default) ... ok
test_one_item_task_works_for_all_architectures (__main__.TestSimulation.test_one_item_task_works_for_all_architectures) ... ok
test_parallel_work_finishes_in_fewer_rounds_with_workers (__main__.TestSimulation.test_parallel_work_finishes_in_fewer_rounds_with_workers) ... ok
test_perfect_skill_gives_perfect_success_for_every_architecture (__main__.TestSimulation.test_perfect_skill_gives_perfect_success_for_every_architecture) ... ok
test_rounds_follow_the_simple_formulas (__main__.TestSimulation.test_rounds_follow_the_simple_formulas) ... ok
test_same_seed_gives_same_numbers_and_other_seed_differs (__main__.TestSimulation.test_same_seed_gives_same_numbers_and_other_seed_differs) ... ok
test_sequential_work_is_not_faster_with_an_orchestrator (__main__.TestSimulation.test_sequential_work_is_not_faster_with_an_orchestrator) ... ok
test_shared_hard_items_cap_the_voting_gain (__main__.TestSimulation.test_shared_hard_items_cap_the_voting_gain) ... ok
test_single_agent_success_matches_the_hand_computed_product (__main__.TestSimulation.test_single_agent_success_matches_the_hand_computed_product) ... ok
test_task_validation (__main__.TestSimulation.test_task_validation) ... ok
test_verification_helps_and_costs_tokens (__main__.TestSimulation.test_verification_helps_and_costs_tokens) ... ok
test_voting_needs_at_least_two_agreeing_voters (__main__.TestSimulation.test_voting_needs_at_least_two_agreeing_voters) ... ok
test_zero_skill_floor_keeps_success_near_zero (__main__.TestSimulation.test_zero_skill_floor_keeps_success_near_zero) ... ok
test_agent_failure_is_recorded (__main__.TestTaskBoard.test_agent_failure_is_recorded) ... ok
test_claim_is_exclusive (__main__.TestTaskBoard.test_claim_is_exclusive) ... ok
test_dependencies_block_claims_until_done (__main__.TestTaskBoard.test_dependencies_block_claims_until_done) ... ok
test_dependency_cycle_is_reported_not_hung (__main__.TestTaskBoard.test_dependency_cycle_is_reported_not_hung) ... ok
test_duplicate_task_ids_are_rejected (__main__.TestTaskBoard.test_duplicate_task_ids_are_rejected) ... ok
test_only_the_owner_can_complete (__main__.TestTaskBoard.test_only_the_owner_can_complete) ... ok
test_swarm_finishes_all_tasks_with_no_duplicate_claims (__main__.TestTaskBoard.test_swarm_finishes_all_tasks_with_no_duplicate_claims) ... ok
test_swarm_respects_a_dependency_chain (__main__.TestTaskBoard.test_swarm_respects_a_dependency_chain) ... ok
test_add_add_with_different_content_conflicts_and_same_content_does_not (__main__.TestWorktrees.test_add_add_with_different_content_conflicts_and_same_content_does_not) ... ok
test_both_delete_is_fine (__main__.TestWorktrees.test_both_delete_is_fine) ... ok
test_changes_report_added_modified_deleted (__main__.TestWorktrees.test_changes_report_added_modified_deleted) ... ok
test_clean_text_merge_can_still_fail_the_combined_checks (__main__.TestWorktrees.test_clean_text_merge_can_still_fail_the_combined_checks) ... ok
test_delete_is_applied_when_main_unchanged (__main__.TestWorktrees.test_delete_is_applied_when_main_unchanged) ... ok
test_delete_vs_modify_conflicts (__main__.TestWorktrees.test_delete_vs_modify_conflicts) ... ok
test_duplicate_workspace_name_is_rejected (__main__.TestWorktrees.test_duplicate_workspace_name_is_rejected) ... ok
test_edits_to_different_functions_merge_cleanly (__main__.TestWorktrees.test_edits_to_different_functions_merge_cleanly) ... ok
test_fork_copies_main_and_edits_do_not_leak_back (__main__.TestWorktrees.test_fork_copies_main_and_edits_do_not_leak_back) ... ok
test_merge_order_does_not_change_a_clean_result (__main__.TestWorktrees.test_merge_order_does_not_change_a_clean_result) ... ok
test_merged_files_are_listed_with_who_and_what (__main__.TestWorktrees.test_merged_files_are_listed_with_who_and_what) ... ok
test_modify_vs_delete_in_main_conflicts (__main__.TestWorktrees.test_modify_vs_delete_in_main_conflicts) ... ok
test_path_escape_is_blocked (__main__.TestWorktrees.test_path_escape_is_blocked) ... ok
test_same_line_edits_conflict_and_are_reported (__main__.TestWorktrees.test_same_line_edits_conflict_and_are_reported) ... ok
test_without_isolation_the_last_write_wins (__main__.TestWorktrees.test_without_isolation_the_last_write_wins) ... ok

----------------------------------------------------------------------
Ran 170 tests in 0.905s

OK

5. Discussion questions

Use these in a study group or a class. Try to answer with an example and a source. Give each about 5 to 10 minutes.

1. Pick a task from your own work. Place it on the scale of Figure 1. List the benefits that a team would bring and the costs it would add, and say what you would measure to decide. What is your single-agent baseline [31] [3]?
2. The scaling study found +80.8% on one task and −70.0% on another [3]. Choose three tasks you know. Predict whether a single agent, a centralized team, or a swarm would do best on each, and say what would change your mind.
3. Anthropic's research system gained 90.2% over one agent on its internal eval [14], and Cognition argues for a single thread [15]. Write down what the two posts agree on, what each company sells, and what a fair test between them would look like.
4. A support bot has billing, legal and technical specialists. Choose handoffs or agents as tools for each, and say who talks to the user. What must each specialist see? What private data could leak if a specialist sees the whole history [16] [17] [19]?
5. Design a message board for six research agents: topics, schemas, who may post to what, and how an agent knows what is new. How would you stop a flood of polling or repeated status posts [5] [27]?
6. You publish an agent that books travel. What goes into its Agent Card? What does a caller need to know that a card does not carry (trust, price, limits)? What changes when the caller is another company's agent [21] [22]?
7. For a task of your choice, write a task ledger: what is a given fact, what must be looked up, what is a guess? When would you raise the stall threshold, and what does a re-plan cost [4]?
8. G-Memory keeps an interaction graph, a query graph and an insight graph for a team [12]. What should a team remember that no single agent should? Who may write to it, and how would you stop one bad insight from spreading to every agent?
9. Isolated workspaces avoid overwrites but cannot see meaning. List three conflicts that a text merge will not catch (a renamed function, a changed file format, a changed default). For each, say what check would catch it [24] [28] [31].
10. Agents built from one model tend to act alike (conformity) [27]. What kinds of diversity could you add: prompts, models, roles, temperature, information? What would you measure to see whether it helped [11]?
11. Take a failure from a team you have seen or built. Classify it with MAST. Which layer would you fix first: design, communication, or verification? What would the fix cost [2] [13]?
12. The compiler project leaned on a strong test suite and an oracle [24]. Choose a task with no such oracle (for example, a design review). What would take the place of the tests, and how much of the swarm design still works [29]?

6. Quiz (10 questions)

Pick one answer for each question, then press Check answers. Check answers 0 of 10 answered.

1. According to the lecture's advice on one agent or several, which situation most favors keeping a single agent [31]?
   • The work splits into many small subtasks that need almost no synchronization, so parallel agents could each take a piece and a later step could join their results
   • A measured bottleneck exists in review, and a separate reviewing agent has been shown to remove it, so adding a second agent has a clear and tested purpose
   • The task needs several different tools and sources, each of which could be given to a specialist agent with its own focused context and its own instructions
   • The steps depend closely on one another, or the work is tied to one shared state, or one agent already meets the quality, cost and speed goals
2. What is the best summary of the scaling study by Kim et al. [3]?
   • The effect of a team ranged from +80.8% to −70.0% by task, coordination had diminishing returns above a single-agent baseline of about 45%, and the best design depends on the task
   • Multi-agent teams beat a single agent on every benchmark that the study ran, and the gain grows steadily with the number of agents that are added to the team, so the authors advise scaling the team to the budget
   • Multi-agent teams lose to a single agent on every benchmark that the study ran, because errors always spread, so a single agent with a good prompt is the best choice in every setting that the study tried
   • Independent teams beat all other designs because they do not talk, and a central check only adds cost, so the study advises that teams should never include a verifier across all six benchmarks that the study used
3. Which description of the four orchestration patterns is right [31]?
   • Independent agents send each other every intermediate result, and a swarm has one orchestrator that sends instructions to the workers one by one and keeps all the state and decides after every step what the next worker should do
   • Hybrid means that every agent talks to every other agent and no orchestrator exists, and centralized means that agents work alone and a final vote picks the answer from their outputs, and both patterns forbid agents from reading each other's work
   • Centralized: an orchestrator assigns work and decides what is next, and may become a bottleneck. Decentralized: agents share a board, all to all. Hybrid: an orchestrator plus limited peer links
   • Decentralized means that one orchestrator talks to each worker, and independent means that workers share a message board and may create new agents while they work on the task, with no limit on how many agents can be created or how long they run
4. What is the difference between a handoff and an agent used as a tool, as the OpenAI Agents SDK describes them [16] [17]?
   • In a handoff, the caller keeps control and receives a result from the specialist, and as a tool the specialist takes over the user conversation and the caller is no longer active. In both cases the history is shared in full.
   • In a handoff, control moves to the receiving agent, which by default sees the earlier conversation. As a tool, the caller keeps control and the specialist receives a bounded request
   • The two are the same mechanism with different names, and in both the specialist always sees the full conversation history and the caller always continues after the specialist answers. The only difference is in the name of the function that the model calls.
   • A handoff is only possible between agents that use the same model, and a tool call is only possible between agents in different companies, which use the A2A protocol instead. Handoffs are not available in the SDK for agents that call external tools.
5. In the toy cost model of section 2.13 (made-up numbers), one agent does 8 items. Each step costs 60 tokens, and re-reading each earlier item costs 15 tokens. What is the single agent's token cost (own arithmetic)?
   • 480 tokens, because 8 items at 60 tokens each is 480, and the cost of re-reading earlier items is not counted in this model
   • 1,260 tokens, because the single agent also pays for the two messages that an orchestrator would send for every item of work in this model
   • 900 tokens, because the sum over items i = 0 to 7 of (60 + 15i) is 480 + 15 × 28
   • 1,080 tokens, because each of the 8 items re-reads all 8 items, which gives 8 × (60 + 15 × 5) tokens in total for the whole task
6. Which statements about the A2A protocol, version 1.0.1, are correct [21]?
   • The card is served at /.well-known/agent-card.json, the server generates the task id, and a message to a task in a terminal state is rejected as an unsupported operation
   • The Agent Card is served at a path that the client chooses, a task is created by the client, and a task that has completed still accepts new messages from the client
   • A message part has a kind field that is either "text" or "file", and method names are lower case with a slash, such as message/send, in every version of the protocol, and the version header is optional and ignored by every server
   • Task states are the strings "open" and "closed", and the JSON-RPC error codes are chosen freely by each server, so a client must read the message text to know what went wrong, and the Agent Card is optional for any server that wants to be found
7. In Magentic-One, what happens when the stall counter goes over its threshold [4]?
   • The Orchestrator stops the run at once and returns its first educated guess to the user, because it has no way to change a plan after the task ledger has been written, and it has no later chance to try a different approach
   • The Orchestrator hands the whole task to the human user through a handoff, and the team agents wait idle until the user has written a new plan for them to follow in the next round, and the rest of the team is shut down for the remainder of the run
   • The Orchestrator deletes the progress ledger and asks the web agent to try the same instruction again without any change, until the counter drops back below the threshold again, while the task ledger stays untouched
   • The Orchestrator reflects, updates the task ledger with what it learned, writes a new plan, and starts a new round of the outer loop. The threshold was at most 2 in the paper's experiments
8. Two agents call claim on a shared task board at the same moment. What must be true, and what did Carlini's compiler project use [24]?
   • Both agents should receive the task, so that the work is done twice and the better result can be chosen, and the project used one lock file that all sixteen agents shared at the same time, and no merge conflicts ever happened
   • Exactly one agent must get the task, so the claim must be atomic. The project used lock files in a current_tasks/ folder, synchronized through git, so the second agent picked another task
   • The agent with the longer history must get the task, and the project used a central orchestration agent that gave each task to the agent that had the most tokens left in its context window, so that busy agents were never given new work
   • Neither agent should get it until a human approves, and the project used a manual review queue where a person assigned each of the roughly two thousand sessions to one of the sixteen agents, and no two sessions ever ran at the same time
9. Agent A changes lines 3 and 7 of a file. Agent B changes line 3 differently and leaves line 7 alone. What does a three-way merge do, and what does a clean result prove (the lab's merge, and section 2.9)?
   • It takes A's change to line 7, marks line 3 as a conflict with both versions, and a clean merge elsewhere proves only that the text did not collide, not that the combined program is correct
   • It takes B's version of the whole file, because B was the last to finish, and a clean result proves that the program works because no markers are left in any of the files, and the other agent's work is deleted on purpose
   • It rejects both agents' work and restores the base file, because any two edits to the same file conflict, and a clean result proves that the tests will pass on the merged files, so every conflict should be treated as a failure of the agents and rerun
   • It merges both versions of line 3 into one line by averaging them, takes line 7 from A, and a clean result proves that no agent has overwritten any work from another agent, and a merge conflict can never occur in that case
10. Which statement about the MAST failure taxonomy is correct [2]?
    • It has 14 modes in two groups, built from a single framework, and it shows that failures in multi-agent systems are almost all caused by weak base models, so better models will remove them, with no need for any other fix
    • It has 14 modes that occur only in systems that use a protocol such as A2A or MCP, and it shows that adding a role prompt or a verifier removes the failures completely in every system, whatever the framework or the task
    • It has 14 modes in three groups (design, inter-agent misalignment, verification), built from careful human annotation and then applied with an LLM judge. No single group accounts for most failures
    • It has 3 modes, one per group, and it was built by asking a language model to label 1,600 traces, with no human annotators and no check of the model's labels against human labels, and the paper reports that the labels are perfectly reliable

Answer key

7. Graded assignment and rubric

Title: Design, build, break and defend a multi-agent system.
Time: 4 to 5 hours. Total: 100 points. Work: solo.

The task

Extend the lab code and write a short memo (three pages at most). Mock models are fine. You may use a real model if you have access, but say clearly which parts used it. Build your own small scenarios for each part, and say how you made them. Do not reuse the lab's triage and billing example or its numbered items. Label every number that comes from a toy model or from made-up data.

Part A. Single agent or team? A cost and accuracy study (25 points)

• Choose a kind of work that is not the lab's (for example, checking 12 citations in a draft, or triaging 10 support tickets). Describe two workloads: one that splits into independent pieces, and one with a chain of dependent steps [31] [3].
• Set your own SimParams and say why each value is plausible for your work. Run all seven architectures on both workloads with at least 5 seeds. Report success, tokens and rounds as a table, with the mean and the range [11].
• Compute the single agent's success and its token cost by hand for one workload, as in 2.13, and show that your arithmetic matches the code.
• Add one new architecture (for example, a planner plus a checker, or debate with a judge). Write at least five unit tests, including an edge case. Say which of your assumptions drives its result [7].
• In four sentences, say which results would change if the numbers came from a real model, and which would not.

Part B. Messages, boards and an A2A-style exchange (25 points)

• Write Agent Cards for three agents in your domain. Validate them with validate_agent_card. Say which fields a real client would need that the card does not carry (trust, price, limits) [21].
• Run an exchange through A2AServer and A2AClient in which one task goes to TASK_STATE_INPUT_REQUIRED and then completes on the same task. Include a GetTask call and print the wire log [21] [23].
• Trigger at least three errors on purpose (for example, an unknown task, a message to a completed task, and a bad version header). Report the code and the reason of each, and say which spec rule each one tests [21].
• Design a message board for your domain with at least three topics and a schema for each. Run a task board with at least six tasks and two dependencies, with at least four threads. Show that no task is claimed twice [5] [24].
• List what your exchange does not implement (for example, streaming, push notifications, security), and say what a production system would still need.

Part C. Coordination, memory and isolation (25 points)

• Run a MagenticOrchestrator with an agent that gets stuck. Show the task ledger before and after the re-plan, and the progress ledger entries that led to it. Run it with two different stall thresholds and compare [4].
• Show the lost-update race in SharedMemory with your own data. Fix it in two ways (a lock and a version check), and show a stale write being rejected [12].
• Create a small repository of your own (at least three files). Let three agents edit it in separate workspaces: one pair of edits that merge cleanly, one pair that conflict, and one pair that merge cleanly but break a check that you wrote. Report the merge result of each [28] [24].
• Write a short note: what does text-level conflict detection miss, and what other check would you add [31]?

Part D. Failures and a design memo (25 points)

• Inject at least six faults that cover all three MAST groups. Report which modes the classifier found for each, and whether it found anything unrelated [2].
• Add one new classifier rule and its tests. Then show one trace that the classifier gets wrong (a miss or a false alarm), and explain why. State plainly that this classifier is a rule-based toy and not the MAST annotator [2].
• Write a two-page memo for this decision: a company wants an assistant that handles incident response, with log search, code search and ticket updates. Decide single agent or team. If a team, choose the pattern, the communication (conversation, handoff, tool, board), the coordination (ledger, memory, isolation) and the verification step. Give a rough cost estimate and say how you would test it before launch [3] [14] [15].
• Name three risks from sections 2.10 to 2.12 (for example, conformity, polling floods, a stuck owner) and give one control for each [27] [13] [24]. List three limits of your own experiments.

What to hand in

• Your multi_agent_toolkit.py (or your own file names), your scenarios and data, and the tests. All tests must pass.
• A memo in PDF or Markdown with the tables from Parts A to C and the design memo from Part D.
• A sources list for any reading you cite.

Rubric

Criterion	Full marks	Partial	Points
A. Single or team	Two own workloads. Seven architectures with at least 5 seeds, a mean and a range. Hand arithmetic that matches the code. One new architecture with five or more passing tests. Honest note on toy versus real.	One seed, or no hand arithmetic, or fewer than five tests, or numbers not labelled as toy.	25
B. Messages and boards	Three valid cards. An input-required exchange on one task with a wire log. Three deliberate errors with codes. A board with schemas and a task board with no double claim under threads. A clear list of what is not implemented.	Cards without validation, or no follow-up on the same task, or fewer than three errors, or no thread test.	25
C. Coordination and isolation	Ledgers before and after a re-plan, with two thresholds compared. A race shown and fixed two ways, with a stale write rejected. Three merge cases, including a clean merge that fails a check. A note on what text merging misses.	One threshold only, or one fix only, or no semantic-break case.	25
D. Failures and memo	Six or more faults across all three MAST groups with results. A new rule with tests and one honest error case. A memo with a clear choice, a cost estimate, a test plan, three risks with controls and three limits.	Faults in fewer groups, or no error case, or a memo with no cost or no test plan, or sources missing.	25
Total			100

Grade bands: 90 to 100 excellent. 75 to 89 good. 60 to 74 needs work. Under 60: redo and resubmit. Plain writing is rewarded: short sentences, no filler. Label your own arithmetic as your own.

8. Glossary

Term	Meaning
A2A (Agent2Agent)	An open protocol for one agent to discover and send tasks to another agent, over a network, without sharing its inner workings. This course follows version 1.0.1 [21] [22].
Agent as a tool	A pattern in which one agent calls another like a function and keeps control. The specialist sees only the request [17].
Agent Card	A JSON document in which an A2A server describes itself: name, skills, supported interfaces, capabilities. It is served at /.well-known/agent-card.json [21].
Artifact	In A2A, an output that a task produces, made of parts, such as a document or a file [21].
Blackboard / message board	A shared place where agents post and read messages, often by topic, instead of talking to each other directly [5].
Capability ceiling	The finding that coordination adds little when a single agent is already good (about 45% in the study's data) [3].
Centralized	A team pattern in which one orchestrator assigns work and collects results [31].
Context ID	In A2A, an identifier that groups related tasks and messages. The server may generate it [21].
Conformity	The tendency of agents built from one model to act alike and to converge on the same choices [27].
Decentralized	A team pattern in which agents talk to each other as peers, for example through a shared board [31].
Error amplification	How much a team multiplies a single agent's error rate. In the scaling study, 17.2 times for independent teams and 4.4 times for centralized ones [3].
Handoff	A transfer of control from one agent to another, often as a tool such as transfer_to_refund_agent. The receiving agent sees the earlier conversation by default [16].
Hybrid	A team pattern that combines an orchestrator with some peer-to-peer links [31].
Independent (team)	A pattern in which agents work alone on the same task and a final step combines their answers, for example by a vote [3] [11].
JSON-RPC 2.0	A small remote-call format. A request has jsonrpc, id, method and params. A reply has the same id and either result or error [23].
Magentic-One	A generalist team from Microsoft Research in which an Orchestrator directs four specialist agents using a task ledger and a progress ledger [4].
MAST	The Multi-Agent System Failure Taxonomy: 14 failure modes in three groups, with an annotation method [2].
Message pool	In MetaGPT, a shared place where agents publish structured messages and from which they subscribe by role [5].
Orchestrator	The agent that plans, assigns work to others and decides what happens next [4].
Part (A2A)	A piece of content in a message or artifact. In version 1.0 a part holds exactly one of text, raw bytes, a URL, or structured data, and has no kind field [21].
Progress ledger	In Magentic-One, the inner-loop record in which the Orchestrator answers five questions at every step [4].
Stall counter	In Magentic-One, a count of how long the team has made no progress. When it exceeds a threshold, the Orchestrator re-plans [4].
Swarm	A style in which agents hand off control to one another with no central manager. Also the name of an educational OpenAI library [18] [31].
Task (A2A)	The unit of work in A2A, with an id and a state. States are submitted, working, input required, auth required, completed, failed, canceled and rejected [21].
Task ledger	In Magentic-One, the outer-loop record of given facts, facts to look up, facts to derive, educated guesses and the plan [4].
Terminal state	A task state after which no more messages are accepted: completed, failed, canceled or rejected [21].
Three-way merge	Combining two edited versions of a file by comparing each with their common base. Changes on one side are taken. Different changes to the same lines are a conflict [28].
Verification (MAST group)	The group of failure modes about ending a task too early and about missing or wrong checks of the result [2].
Worktree	In git, an extra working directory linked to one repository, each with its own checked-out branch. The lab uses a simpler copy of the idea [28].

9. References

Entries marked (required) are the course's required readings. Entries marked (added) are my own additions. The course site and slides, [30] and [31], are the course's own pages. "Abstract level" means that I read the abstract and not the whole paper. Vendor-authored pages are marked as such.

• [1] Wu, Q., Bansal, G., Zhang, J., Wu, Y., Li, B., Zhu, E., Jiang, L., Zhang, X., Zhang, S., Liu, J., Awadallah, A., White, R. W., Burger, D., and Wang, C. "AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation." arXiv:2308.08155 (version 2, October 2023). Microsoft-affiliated authors. https://arxiv.org/abs/2308.08155 (required)
• [2] Cemri, M., Pan, M. Z., Yang, S., et al. "Why Do Multi-Agent LLM Systems Fail?" arXiv:2503.13657 (version 3, October 26, 2025). UC Berkeley and Intesa Sanpaolo. https://arxiv.org/abs/2503.13657 (required)
• [3] Kim, Y., Gu, K., Park, C., et al. "Towards a Science of Scaling Agent Systems." arXiv:2512.08296 (version 3, April 2026; this course uses the version 3 numbers). Google Research, Google DeepMind and MIT. https://arxiv.org/abs/2512.08296 (required)
• [4] Fourney, A., Bansal, G., Mozannar, H., Tan, C., et al. "Magentic-One: A Generalist Multi-Agent System for Solving Complex Tasks." arXiv:2411.04468. Microsoft authors. https://arxiv.org/abs/2411.04468 (I read the abstract and the sections on the Orchestrator, the two ledgers and the stall counter) (added)
• [5] Hong, S., Zhuge, M., Chen, J., et al. "MetaGPT: Meta Programming for a Multi-Agent Collaborative Framework." ICLR 2024. arXiv:2308.00352. https://arxiv.org/abs/2308.00352 (I read the abstract and the passages on SOPs, the shared message pool and role subscription) (added)
• [6] Li, G., Hammoud, H. A. A. K., Itani, H., Khizbullin, D., and Ghanem, B. "CAMEL: Communicative Agents for “Mind” Exploration of Large Language Model Society." arXiv:2303.17760. https://arxiv.org/abs/2303.17760 (abstract level) (added)
• [7] Du, Y., Li, S., Torralba, A., and Tenenbaum, J. B. "Improving Factuality and Reasoning in Language Models through Multiagent Debate." arXiv:2305.14325. https://arxiv.org/abs/2305.14325 (abstract level) (added)
• [8] Qian, C., Liu, W., Liu, H., et al. "ChatDev: Communicative Agents for Software Development." arXiv:2307.07924. https://arxiv.org/abs/2307.07924 (abstract level) (added)
• [9] Chen, W., Su, Y., Zuo, J., et al. "AgentVerse: Facilitating Multi-Agent Collaboration and Exploring Emergent Behaviors." arXiv:2308.10848. https://arxiv.org/abs/2308.10848 (abstract level) (added)
• [10] Liu, Z., Zhang, Y., Li, P., Liu, Y., and Yang, D. "A Dynamic LLM-Powered Agent Network for Task-Oriented Agent Collaboration" (DyLAN). COLM 2024. arXiv:2310.02170. One author of this paper, Diyi Yang, also teaches the course. https://arxiv.org/abs/2310.02170 (abstract level) (added)
• [11] Li, J., Zhang, Q., Yu, Y., et al. "More Agents Is All You Need." Transactions on Machine Learning Research, October 2024. arXiv:2402.05120. https://arxiv.org/abs/2402.05120 (abstract level) (added)
• [12] Zhang, G., Fu, M., Wan, G., Yu, M., Wang, K., and Yan, S. "G-Memory: Tracing Hierarchical Memory for Multi-Agent Systems." arXiv:2506.07398 (June 2025). https://arxiv.org/abs/2506.07398 (I read the abstract, the method and the main results tables, in the arXiv HTML version) (added)
• [13] Khatua, A., Zhu, H., Tran, P., et al. "CooperBench: Why Coding Agents Cannot be Your Teammates Yet." arXiv:2601.13295 (version 2, January 27, 2026). Two co-authors, Diyi Yang and Michael J. Ryan, teach the course. https://arxiv.org/abs/2601.13295 (I read the abstract and the introduction) (added)
• [14] Hadfield, J., Zhang, B., Lien, K., Scholz, F., Fox, J., and Ford, D. "How we built our multi-agent research system." Anthropic engineering blog (the page date was not in the text I captured). Vendor-authored: the 90.2% figure is on Anthropic’s own internal evaluation, and I did not check it. https://www.anthropic.com/research/multi-agent-research-system (added)
• [15] Cognition. "Don’t Build Multi-Agents." Cognition blog, June 2025. Vendor opinion, with no experiments in the text that I read. https://cognition.com/blog/dont-build-multi-agents (added)
• [16] OpenAI. "Handoffs." OpenAI Agents SDK (Python) documentation. Vendor documentation; it changes over time, and I read it in October 2026. https://openai.github.io/openai-agents-python/handoffs/ (added)
• [17] OpenAI. "Tools" (including agents as tools). OpenAI Agents SDK (Python) documentation. Vendor documentation; I read it in October 2026. https://openai.github.io/openai-agents-python/tools/ (added)
• [18] OpenAI. "Swarm: Educational framework exploring ergonomic, lightweight multi-agent orchestration." GitHub repository and README. The README says that Swarm is replaced by the OpenAI Agents SDK. https://github.com/openai/swarm (added)
• [19] OpenAI. "A practical guide to building agents." PDF guide, 2025. Vendor-authored. https://cdn.openai.com/business-guides-and-resources/a-practical-guide-to-building-agents.pdf (I read the sections on the manager and decentralized patterns) (added)
• [20] LangChain. "Multi-agent" (subagents, handoffs, skills, router). LangChain documentation. Vendor documentation; the comparison is illustrative. https://docs.langchain.com/oss/python/langchain/multi-agent (added)
• [21] A2A Project. "Agent2Agent (A2A) Protocol Specification," version 1.0.1 (released May 28, 2026), with the protocol buffer file a2a.proto from the same release. The web page’s banner still said that 1.0.0 was the latest release when I read it. Version 0.3 was not checked. release tag. https://a2a-protocol.org/latest/specification/ (added)
• [22] A2A Project. "A2A and MCP." Topic page, a2a-protocol.org. https://a2a-protocol.org/latest/topics/a2a-and-mcp/ (added)
• [23] JSON-RPC Working Group. "JSON-RPC 2.0 Specification." https://www.jsonrpc.org/specification (added)
• [24] Carlini, N. "Building a C compiler with parallel Claudes." Anthropic engineering blog. Vendor-authored; the account is the author’s own and is not independent. https://www.anthropic.com/research/building-c-compiler (added)
• [25] OpenAI. "An open-source spec for Codex orchestration: Symphony." OpenAI, April 27, 2026. Vendor-authored; the "500% increase in landed pull requests" is OpenAI’s claim about some of its own teams, and I did not verify it. The openai.com site blocked my script, so I read the page through a separate fetch tool. https://openai.com/index/open-source-codex-orchestration-symphony/ (added)
• [26] OpenAI. "Symphony Service Specification" (SPEC.md, Draft v1). GitHub, openai/symphony. https://github.com/openai/symphony/blob/main/SPEC.md (added)
• [27] Anthropic. "Patterns and problems in multiagent systems." Anthropic research page (corresponding author: Carolyn Zou). Not on arXiv. Vendor-authored; the text I captured shows no date. In the swarm comparison, the two arms used different token budgets. https://www.anthropic.com/research/multiagent-systems (added)
• [28] Git project. "git-worktree" manual page. git-scm.com. https://git-scm.com/docs/git-worktree (added)
• [29] Young, J. "Effective harnesses for long-running agents." Anthropic engineering blog. Vendor-authored. https://www.anthropic.com/research/effective-harnesses-for-long-running-agents (added)
• [30] Yang, D., Ryan, M., and Yang, J. "CS329Z: Engineering AI Agents." Stanford University, Fall 2026. Course site and schedule. https://cs329z.stanford.edu/
• [31] Yang, D., Ryan, M., and Yang, J. "Multi-Agent Systems." CS329Z lecture slides (Lecture 8, Week 5, October 19), Stanford University. Used for the topic order and for what the lecture claims; where my page says "the lecture", it means these slides. https://web.stanford.edu/class/cs329z/slides/lecture08.pdf