Mission: an agent and its builders build jq from its manual, scored on jq's own tests

A planner agent builds jq 1.6 in Python, the package jqpy/, from the jq manual alone. The work takes hours, so the planner works in milestones. Code, not the agent, scores each milestone. From milestone 2, the planner gives features to builder subagents. They run in parallel in its sandbox. After each milestone, a hook runs jq's own test suite on the build: 307 cases kept outside the agents' workspace. The hook puts the score on the first line of the planner's answer. A judge reads that answer and decides whether to run another milestone. To steer the running milestone, you type a line.

Read it left to right: each milestone is one turn of the planner, scored before the judge sees it.

In our run, two milestones took 26 minutes and 6.9M tokens, and the build passed 129, then 219, of jq's 307 tests.

Demonstrates: subagents in the planner's sandbox, a skill from a directory, a post_agent_turn hook that runs a command in the sandbox and replaces the answer, turns with idempotency keys and a timeout, live events, steering, a judge with structured output, and agents serve, where the mission outlives its driver.

Run: from the example's folder, with Docker running and MISTRAL_API_KEY set: uv run python drive.py jq-1 2

The planner owns the core: the lexer, the parser and the interpreter. Each builder owns one file under jqpy/builtins/, which the planner names when it starts it, so two builders never edit the same file. Both load the jq-manual skill: the jq 1.6 manual, its 198 examples as dev_tests.py, and those layout rules.

agent = agents.Agent(
    model="mistral-large-4",
    name="jq-mission",
    instructions=PLANNER,
    environment=(
        agents.MistralSandbox(default_exec_timeout_seconds=300)  # dev_tests.py takes minutes; the default is 30 s
        if os.environ.get("MISSION_SANDBOX") == "mistral"  # served, an agent with subagents needs it
        else agents.LocalDockerSandbox()
    ),
    skills=[agents.LocalSkillBundle.from_directory(HERE / "skills" / "jq-manual")],
    subagents=[
        agents.Subagent(
            name="builder",
            description="Implements one jq feature in the one jqpy file you name, from the manual sections you give.",
            instructions=BUILDER,
            hooks=[Tokens()],  # no Score: the score is the planner's, once its builders are done
            builtins=agents.Builtins(subagents=False),
        )
    ],
    subagent_policy=agents.SubagentPolicy(concurrent_child_turn_limit=4),
    hooks=[Score(), Tokens()],
)

agent = agents.Agent(
    model="mistral-large-4",
    name="jq-mission",
    instructions=PLANNER,
    environment=(
        agents.MistralSandbox(default_exec_timeout_seconds=300)  # dev_tests.py takes minutes; the default is 30 s
        if os.environ.get("MISSION_SANDBOX") == "mistral"  # served, an agent with subagents needs it
        else agents.LocalDockerSandbox()
    ),
    skills=[agents.LocalSkillBundle.from_directory(HERE / "skills" / "jq-manual")],
    subagents=[
        agents.Subagent(
            name="builder",
            description="Implements one jq feature in the one jqpy file you name, from the manual sections you give.",
            instructions=BUILDER,
            hooks=[Tokens()],  # no Score: the score is the planner's, once its builders are done
            builtins=agents.Builtins(subagents=False),
        )
    ],
    subagent_policy=agents.SubagentPolicy(concurrent_child_turn_limit=4),
    hooks=[Score(), Tokens()],
)

The builders run in the planner's sandbox and write to the same workspace. At most four run at once, and Builtins(subagents=False) keeps a builder from starting builders of its own. The Tokens hook adds up the tokens of the planner's and the builders' model calls.

On your machine, the build runs in a Docker sandbox. With MISSION_SANDBOX=mistral, it runs in a MistralSandbox, which needs the cloud_user workspace role, shown as Mistral Cloud User in the admin console.

The agents check their own work on the manual's examples. The score that counts comes from elsewhere: jq 1.6's own test suite, held_out/jq.test, which stays in the example's folder. After each planner turn, the Score hook copies it and score.py into the sandbox, outside the workspace, runs the score, and puts its line before the planner's report:

class Score(agents.Hook):
    """After each planner turn, run jq's held-out tests on the build and put the score first in the answer."""

    async def post_agent_turn(self, ctx, answer):
        score = await ctx.sandbox.write_file("score.py", SCORE)
        tests = await ctx.sandbox.write_file("jq.test", HELD_OUT)
        run = await ctx.sandbox.execute(f"python3 {score} {tests}", timeout_seconds=300)
        lines = run.stdout.strip().splitlines()
        line = lines[-1] if run.exit_code == 0 and lines else f"score failed: {run.stderr[-300:]}"
        return agents.hook.PostAgentTurnHookAccept(
            acceptance=agents.hook.ReplaceAssistantContent(
                content=[agents.TextContentBlock(text=f"{line}\n\n{answer.text}")]
            )
        )

class Score(agents.Hook):
    """After each planner turn, run jq's held-out tests on the build and put the score first in the answer."""

    async def post_agent_turn(self, ctx, answer):
        score = await ctx.sandbox.write_file("score.py", SCORE)
        tests = await ctx.sandbox.write_file("jq.test", HELD_OUT)
        run = await ctx.sandbox.execute(f"python3 {score} {tests}", timeout_seconds=300)
        lines = run.stdout.strip().splitlines()
        line = lines[-1] if run.exit_code == 0 and lines else f"score failed: {run.stderr[-300:]}"
        return agents.hook.PostAgentTurnHookAccept(
            acceptance=agents.hook.ReplaceAssistantContent(
                content=[agents.TextContentBlock(text=f"{line}\n\n{answer.text}")]
            )
        )

score.py reads the tests, then deletes them and itself, so the next milestone never finds them in the sandbox. It runs each case in a new Python process that gets the program and its input, never the expected output. The system lets that process start no other process, so the build cannot run a jq binary itself. It prints one line, such as jq.test 219/307.

drive.py starts each milestone as a turn, with an idempotency key made of the mission's name and the milestone's number. It waits 40 minutes at most and prints each tool call of the planner as it happens. A milestone still running then gets a steer: interrupt your builders, then report. An interrupt of the planner's turn would stop the planner only, and its builders would keep editing files. drive.py waits 10 more minutes, which includes the score's run, then interrupts the milestone. The next one starts from the files it wrote, and a builder still running then reports to it:

turn = await session.start_turn(milestone(n), idempotency_key=f"{name}:m{n}")  # a rerun gets the same turn
if isinstance(turn, agents.Busy):
    sys.exit(f"session {session.id} runs turn {turn.active_turn_id}, not a milestone: interrupt it first")
if served:
    SAVED.mkdir(exist_ok=True)
    saved_file.write_text(json.dumps({**saved, "session_id": session.id, "running": n}))
keyboard.turn = turn
record = await turn.result(after=saved.get("cursor"), timeout=MILESTONE_SECONDS, on_event=show)
if isinstance(record, agents.Running):  # time is up: an interrupt would stop the planner, not its builders
    await turn.steer(TIME_UP)
    record = await turn.result(after=record, timeout=10 * 60, on_event=show)  # the score included
if isinstance(record, agents.Running):
    await turn.interrupt()  # a builder still running reports to the next milestone
    record = await turn.result()

turn = await session.start_turn(milestone(n), idempotency_key=f"{name}:m{n}")  # a rerun gets the same turn
if isinstance(turn, agents.Busy):
    sys.exit(f"session {session.id} runs turn {turn.active_turn_id}, not a milestone: interrupt it first")
if served:
    SAVED.mkdir(exist_ok=True)
    saved_file.write_text(json.dumps({**saved, "session_id": session.id, "running": n}))
keyboard.turn = turn
record = await turn.result(after=saved.get("cursor"), timeout=MILESTONE_SECONDS, on_event=show)
if isinstance(record, agents.Running):  # time is up: an interrupt would stop the planner, not its builders
    await turn.steer(TIME_UP)
    record = await turn.result(after=record, timeout=10 * 60, on_event=show)  # the score included
if isinstance(record, agents.Running):
    await turn.interrupt()  # a builder still running reports to the next milestone
    record = await turn.result()

The judge is a second agent, on mistral-medium-latest, with output_type=Verdict: a done flag and one sentence of why. It says done when the score passes 90% or the report shows the build is stuck:

async def judged(report: str) -> bool:
    """Ask the judge whether the mission is done."""
    match await judge.run(report):
        case agents.Completed(output=Verdict(done=True, why=why)):
            print(f"done: {why}")
            return True
        case agents.Completed(output=Verdict(why=why)):
            print(f"{DIM}go on: {why}{RESET}")
        case other:
            print(f"{DIM}the judge gave no verdict ({type(other).__name__}): go on{RESET}")
    return False

async def judged(report: str) -> bool:
    """Ask the judge whether the mission is done."""
    match await judge.run(report):
        case agents.Completed(output=Verdict(done=True, why=why)):
            print(f"done: {why}")
            return True
        case agents.Completed(output=Verdict(why=why)):
            print(f"{DIM}go on: {why}{RESET}")
        case other:
            print(f"{DIM}the judge gave no verdict ({type(other).__name__}): go on{RESET}")
    return False

Type a line and press Enter: drive.py adds it to the running milestone with turn.steer(). Between two milestones there is no turn to steer, and drive.py prints not sent, as it does when turn.steer() returns TurnConflict because the milestone has just ended:

async def steer(self, line: str) -> None:
    # steer() adds the line to this turn only; it never starts a turn between two milestones
    try:
        steered = self.turn is not None and isinstance(await self.turn.steer(line), agents.Turn)
    except agents.AgentsError as error:  # the session ended, or the connection did
        print(f"  {DIM}↳ not sent: {error}{RESET}")
        return
    if steered:
        print(f"  {DIM}↳ steered: {line}{RESET}")
    else:
        print(f"  {DIM}↳ not sent: no milestone is running{RESET}")

async def steer(self, line: str) -> None:
    # steer() adds the line to this turn only; it never starts a turn between two milestones
    try:
        steered = self.turn is not None and isinstance(await self.turn.steer(line), agents.Turn)
    except agents.AgentsError as error:  # the session ended, or the connection did
        print(f"  {DIM}↳ not sent: {error}{RESET}")
        return
    if steered:
        print(f"  {DIM}↳ steered: {line}{RESET}")
    else:
        print(f"  {DIM}↳ not sent: no milestone is running{RESET}")

With agents serve, or when you deploy with Studio (self-hosted or Mistral-managed), the planner and its builders run in the served agent. drive.py --served starts the turns, waits, and asks the judge. Kill the driver and run it again: the same idempotency keys return the same session and the same running milestone, and the driver attaches to it. It keeps the session's id and the last milestone's cursor in .missions/.

From a second terminal, watch.py follows the same session, and steers whichever milestone runs:

async def main(name: str) -> None:
    session_id = json.loads((SAVED / f"{name}.json").read_text())["session_id"]
    session = agents.Agent.load_from(agent_name=agent.name).session(session_id)
    print(f"watching mission {name} · session {session_id}")
    keyboard = Keyboard()
    reading = asyncio.create_task(keyboard.read()) if sys.stdin.isatty() else None
    async for event in session.iter_events():
        show(event)
        match event:
            case agents.TurnStarted(turn_id=turn_id) if turn_id is not None:
                keyboard.turn = session.turn(turn_id)  # the milestone that runs now
            case agents.TurnEnded(state=state):
                keyboard.turn = None
                if state.text:
                    print(state.text.splitlines()[0])  # the milestone's score
    if reading is not None:
        reading.cancel()

async def main(name: str) -> None:
    session_id = json.loads((SAVED / f"{name}.json").read_text())["session_id"]
    session = agents.Agent.load_from(agent_name=agent.name).session(session_id)
    print(f"watching mission {name} · session {session_id}")
    keyboard = Keyboard()
    reading = asyncio.create_task(keyboard.read()) if sys.stdin.isatty() else None
    async for event in session.iter_events():
        show(event)
        match event:
            case agents.TurnStarted(turn_id=turn_id) if turn_id is not None:
                keyboard.turn = session.turn(turn_id)  # the milestone that runs now
            case agents.TurnEnded(state=state):
                keyboard.turn = None
                if state.text:
                    print(state.text.splitlines()[0])  # the milestone's score
    if reading is not None:
        reading.cancel()

For agents serve, the key's user needs a workspace role with Workflows access, such as Developer or Workspace Contributor; a workspace administrator assigns it. Served, an agent with subagents needs a MistralSandbox: set MISSION_SANDBOX=mistral in the shell of agents serve, or add ENV MISSION_SANDBOX=mistral to the Dockerfile that agents deploy writes. The key's user, or the deployment's service account, then needs the cloud_user role.

jq's manual and tests are from jq 1.6, under the MIT license (skills/jq-manual/COPYING).

You need Docker. From the example's folder:

uv sync
export MISTRAL_API_KEY=...
uv run python drive.py jq-1 2        # at most 2 milestones; without a number, at most 6

uv sync
export MISTRAL_API_KEY=...
uv run python drive.py jq-1 2        # at most 2 milestones; without a number, at most 6

To run the build in a MistralSandbox, set MISSION_SANDBOX=mistral.

Served, in three terminals:

MISSION_SANDBOX=mistral uv run agents serve     # terminal 1
uv run python drive.py jq-1 6 --served          # terminal 2: kill it, run it again, it attaches
uv run python watch.py jq-1                     # terminal 3: the live events; type a line to steer

MISSION_SANDBOX=mistral uv run agents serve     # terminal 1
uv run python drive.py jq-1 6 --served          # terminal 2: kill it, run it again, it attaches
uv run python watch.py jq-1                     # terminal 3: the live events; type a line to steer

The test needs no key: uv run pytest -o asyncio_mode=auto test_mission.py.

Two milestones in a MistralSandbox (MISSION_SANDBOX=mistral), on your machine. Long runs of tool calls are folded, and long report lines end in …:

The two milestones took 26 minutes. In milestone 2, the four builders wrote their modules at the same time, and the score rose from 129 to 219. Of the 6.9M tokens, 97% are input: each model call sends the conversation so far, and the planner reads long sections of the manual.

The served run below shows milestone 1, with long report lines cut at [...]. The driver was killed 4 minutes into the milestone and run again 20 seconds later. It attaches to the running milestone and replays its events from the start:

Meanwhile, in watch.py, a line typed during milestone 1 asked for the dev_tests.py count on the first line of the report. The report above opens with it:

Served, the model calls run in the served agent, not in the driver, so the driver prints no token count.

• Another spec and its tests: swap the manual and the held-out suite for your own: a library ported to another language, a parser built from its grammar, a client built from its API reference. Keep the suite out of the agents' workspace, and let the hook score each milestone.
• Another stop rule: change the judge's instructions, for example to stop when a milestone raises the score by less than 5 tests.
• Longer or wider milestones: raise MILESTONE_SECONDS, or concurrent_child_turn_limit for more builders at once.