QA swarm: 60 agents test jq 1.6 and check each other's bugs

@app.post("/api/boards/{board}/messages")
def send(board: str, message: NewMessage) -> dict:
    with pool.connection() as db:
        # One writer at a time, so ids commit in order: a reader paging with after= never skips one.
        db.execute("LOCK TABLE messages IN SHARE ROW EXCLUSIVE MODE")
        (id,) = db.execute(
            "INSERT INTO messages (board, author, text) VALUES (%s, %s, %s) RETURNING id",
            (board, message.author, message.text[:2000]),
        ).fetchone()
    return {"id": id}

@app.get("/api/boards/{board}/messages")
def read(board: str, after: int = 0, limit: int = Query(50, le=200)) -> list[dict]:
    """The board's messages with an id greater than `after`, oldest first."""
    with pool.connection() as db:
        rows = db.execute(
            "SELECT id, author, text, at FROM messages WHERE board = %s AND id > %s ORDER BY id LIMIT %s",
            (board, after, limit),
        ).fetchall()
    return [
        {"id": id, "author": author, "text": text, "at": at.isoformat(timespec="seconds")}
        for id, author, text, at in rows
    ]

class Member(agents.SessionInput):
    """Who this session is on the board."""

    input_id = "swarm.member"
    name: str

@agents.tool
async def send_swarm_message(text: str, me: Annotated[Member, agents.Depends()]) -> str:
    """Post a message on the swarm's board, where every member and the coordinator can read it."""
    return f"posted as message {await post(me.name, text)}"

@agents.tool(annotations=agents.ToolAnnotations(read_only_hint=True, destructive_hint=False))
async def read_message(after: int = 0, limit: int = 50) -> str:
    """Read the board's messages with an id greater than `after`, oldest first, at most `limit` (200 at most)."""
    response = await call_board("GET", "/messages", params={"after": after, "limit": max(1, min(limit, 200))})
    lines = [f"{m['id']} {m['author']}: {m['text']}" for m in response.json()]
    return "\n".join(lines) or "(no newer messages)"

@agents.tool
async def assign(member: str, task: str) -> str:
    """Give one member its task for the next round. The task is posted on the board."""
    if member not in sessions:
        return f"no member {member}: the members are {', '.join(sessions)}"
    assignments[member] = task
    await post("coordinator", f"-> {member}: {task}")
    return f"{member} assigned"

@agents.tool
async def finish(reasons: list[str], report: str) -> str:
    """End the run.

    `reasons`: one line per slice, why its coverage is good enough. `report`: the bug report in Markdown.
    """
    finished.update(reasons=reasons, report=report)
    return "run ended"

member = agents.Agent(
    model="mistral-large-4",
    name="qa-member",
    instructions=(
        "You are one member of a QA swarm testing the jq command-line JSON processor installed in your "
        "sandbox (run `jq` with your shell tool; there is no network). A bug is wrong output, a crash, a "
        "hang or an error on valid input, judged against the jq manual. The swarm talks only through its "
        "board: send_swarm_message posts, read_message(after=, limit=) reads. Post one line per message:\n"
        "- 'FINDING [<slice>] `<one shell command>` prints <got>; expected <expected>'\n"
        "- 'REPRODUCED #<id>: <what you saw>' or 'NOT REPRODUCED #<id>: <why>'\n"
        "REPRODUCED means two things: you saw the same output, and the jq 1.6 manual says otherwise. "
        "If jq 1.6 behaves as its manual documents (even if a later jq changed it), post NOT REPRODUCED "
        "with the manual's words.\n"
        "Run every command before you post it, post each finding as soon as you have it, and never post "
        "a finding already on the board. Keep each command small: no huge outputs, no endless loops.\n"
        "Each round is short: run at most 8 shell commands (one command may test several cases), post, "
        "then answer with one line: what you covered.\n"
        "Each jq run is limited to 128 MB of memory and 20 seconds. 'cannot allocate memory', exit 124 or "
        "an abort on a small input is a finding: name the input and its size."
    ),
    tools=[send_swarm_message, read_message],
    environment=agents.LocalDockerSandbox(image="swarm-jq16", network=agents.Network.none()),
)

FROM python:3.12-bookworm
RUN apt-get update && apt-get install -y --no-install-recommends jq && rm -rf /var/lib/apt/lists/*
# Each jq run gets 128 MB of virtual memory and 20 seconds, so a runaway jq dies in its own sandbox
# (exit 124 on the time limit, "cannot allocate memory" or exit 5/134 on the memory limit)
# instead of filling the Docker VM. The members see the limits in their instructions.
# The wrapper comes first in PATH; jq itself stays in /usr/bin.
RUN printf '#!/bin/sh\nulimit -v 131072\nexec timeout 20 /usr/bin/jq "$@"\n' > /usr/local/bin/jq && chmod +x /usr/local/bin/jq

coordinator = agents.Agent(
    model="mistral-large-4",
    name="qa-coordinator",
    instructions=(
        "You coordinate a QA swarm that tests jq 1.6. Members post FINDING, REPRODUCED and NOT REPRODUCED "
        "lines on the board. After each round, read the new messages with read_message(after=...). "
        "Then either call assign once per member with its next task, or call finish. When you are asked, "
        "the round is over and every member has stopped: never wait for more messages.\n"
        "A task is one of: reproduce given #ids (never by the member that found them), probe a gap in a "
        "slice, or dig deeper on an area where bugs cluster. Each task names its slice.\n"
        "Call finish only when every finding has been reproduced or refuted by a second member and each "
        "slice is covered well enough. In finish, give one line of reasons per slice, and a report listing "
        "each reproduced bug: its message id, the command, what jq printed and what it should print."
    ),
    tools=[read_message, send_swarm_message, assign, finish],
)

async def timed_turn(session: agents.Session, prompt: str, seconds: float = ROUND_SECONDS):
    """One turn of at most `seconds`. A turn still running then is interrupted, and has ended when this returns."""
    state = await session.run(prompt, timeout=seconds)
    if isinstance(state, agents.Busy):
        raise RuntimeError("the session still runs an earlier turn")
    if isinstance(state, agents.Failed):
        raise RuntimeError(f"the turn failed: {state.error}")
    if isinstance(state, agents.UnknownOutcome):
        raise RuntimeError(f"the turn ended with outcome {state.outcome}")
    if isinstance(state, agents.Running):
        turn = session.turn(state.turn_id)
        await turn.interrupt()
        if isinstance(await turn.result(timeout=60), agents.Running):
            raise RuntimeError("the turn did not stop after its interrupt")
    return state

after = 0
for round_number in range(1, MAX_ROUNDS + 1):
    if finished:  # the coordinator ended the run in round 0
        break
    tasks = [(name, task) for name, task in assignments.items() if name in sessions]
    idle = [name for name in sessions if name not in assignments]
    assignments.clear()
    round_started = time.monotonic()
    states = await asyncio.gather(
        *(run_member(name, f"Round {round_number}. Your task: {task}") for name, task in tasks)
    )
    late = sum(isinstance(state, agents.Running) for state in states)
    messages = await read_round(after)
    kinds = {
        k: sum(m["text"].startswith(k) for m in messages) for k in ("FINDING", "REPRODUCED", "NOT REPRODUCED")
    }
    seconds = time.monotonic() - round_started
    print(
        f"round {round_number}: {len(tasks)} members, {late} late, {seconds:.0f} s; "
        f"board +{len(messages)} {kinds}; crashes so far {len(crashes)}"
    )
    fresh = f" {', '.join(restarted)} failed this round, so their task is not done." if restarted else ""
    fresh += f" {', '.join(idle)} had no task this round." if idle else ""
    restarted.clear()
    await ask(
        f"Round {round_number} is over and every member has stopped.{fresh} Read the board after "
        f"message {after}, paging with after= until there are no newer messages. Then assign each "
        "member its next task, or finish.",
    )
    after = messages[-1]["id"] if messages else after
    if finished:
        print(f"the coordinator ended the run after round {round_number}")
        break
else:
    print(f"stopped by the safety cap of {MAX_ROUNDS} rounds")
    await ask(FINAL_ASK.format(after=after))

async def restart(name: str, error: Exception) -> None:
    """Give a member whose turn failed a fresh session for the next round, or drop it when none starts."""
    crashes.append(f"{name}: {type(error).__name__}: {str(error)[:200]}")
    print(f"crash: {crashes[-1]}", file=sys.stderr, flush=True)
    await asyncio.gather(sessions[name].cancel(), return_exceptions=True)
    try:
        sessions[name] = await member.start_session(session_inputs={Member.input_id: Member(name=name)})
        restarted.append(name)
    except Exception as start_error:
        del sessions[name]
        print(f"dropped {name}: {start_error}", file=sys.stderr, flush=True)

async def run_member(name: str, prompt: str):
    """One member's turn: its state, or None when it failed."""
    try:
        async with at_once:
            return await timed_turn(sessions[name], prompt)
    except Exception as error:
        await restart(name, error)
        return None

async def start_coordinator() -> agents.Session:
    return await coordinator.start_session(session_inputs={Member.input_id: Member(name="coordinator")})

async def ask(prompt: str, seconds: float = ROUND_SECONDS) -> None:
    """One coordinator turn. When it fails, a fresh coordinator reads the whole board, then gets the same
    prompt once more."""
    global lead
    try:
        await timed_turn(lead, prompt, seconds)
    except Exception as error:
        crashes.append(f"coordinator: {type(error).__name__}: {str(error)[:200]}")
        print(f"crash: {crashes[-1]}", file=sys.stderr, flush=True)
        await asyncio.gather(lead.cancel(), return_exceptions=True)
        lead = await start_coordinator()
        catch_up = "You take over from a coordinator that failed: first read the whole board from message 0, "
        await timed_turn(lead, catch_up + "paging with after= until there are no newer messages. " + prompt, seconds)

docker run -d --name swarm-board-db -p 5432:5432 \
  -e POSTGRES_USER=board -e POSTGRES_PASSWORD=board -e POSTGRES_DB=board postgres:17
uv sync
POSTGRES_HOST=127.0.0.1 POSTGRES_USER=board POSTGRES_PASSWORD=board POSTGRES_DB=board \
  uv run uvicorn main:app --port 8000

curl -fsSL https://mistral.ai/cli/install.sh | bash
export PATH="$HOME/.mistral/bin:$PATH"

{"flags": {"mistral_cli_apps": true}}

mistral login
mistral deploy --from-disk --name swarm-board

export SWARM_BOARD_URL=<the URL the deploy printed>

docker build -t swarm-jq16 sandbox
uv sync
export MISTRAL_API_KEY=...
uv run python qa_swarm.py 60

uv run python qa_swarm.py --report qa-20261105-045621