"""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_, 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"(? (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//, 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()