15 · Meeting the real world

The stop button that does not stop

Everything so far was instant, because the model and the shell are fakes. A real build takes ten minutes, and while it runs, nothing else in your process is running at all.

~70 min · 2 labs · builds on C Where does it go?

This lesson builds on lesson 14. New here? Start at lesson 01, or carry on: every lab is self-contained.

Your lesson 14 gate is up. The check is an allow-list: it looks the command's first word up in a dict of commands it knows. The model asks for bash with rm -r build, and "rm" is not a key in that dict, so the check raises KeyError. Does the command run?

Yes. The check crashed, so it never said no, and a broken check is no reason to stop working

That is the logger's rule, not the door's. A door that opens when its lock breaks is not a door.

No. A check that raised has not said yes, so the call is refused and the model is told

Fail closed, and the refusal travels as a result, so the run carries on and the model can try something else.

No, and worse: the KeyError comes out of the run and your process ends on a traceback

The lesson 4 boundary is still under all of this. Nothing that comes out of a tool, gate included, reaches your terminal.

The build directory is still there, and the model got one sentence it can act on, although the sentence is poor: your gate reported the KeyError's own text, which is the missing key and nothing else. Fail closed and tell the model: two lessons in one line of output.

import harness, lab

ws = lab.Workspace({"build/out.txt": "artifacts\n", "src/app.py": "print(1)\n"})
shell = lab.Shell(ws)

ALLOWED = {"pytest": True, "echo": True, "cat": True}


def only_known_commands(name, arguments):
    """Refuses any bash command whose first word is not on the list above."""
    if name != "bash":
        return None
    first = arguments["command"].split()[0]
    return None if ALLOWED[first] else f"{first} is not on the allow-list"


tools = [harness.guard(harness.make_bash_tool(shell), only_known_commands)]
system = harness.build_system_prompt(tools, [], [], "/work/shop", "2026-03-14")
model = lab.ScriptedModel([
    lab.reply(lab.call("bash", {"command": "rm -r build"})),
    lab.say("Nothing was deleted."),
])
h = harness.Harness(model, system, tools)
for event in h.prompt("Clear the build directory."):
    pass

print(lab.show(list(h.messages)))
try:
    ws.read_text("build/out.txt")
    print("build/out.txt: still there")
except FileNotFoundError:
    print("build/out.txt: gone")

user -> "Clear the build directory."
assistant -> toolCall c1 bash({"command": "rm -r build"})
toolResult c1 -> "Tool call blocked: the check failed: 'rm'"  [is_error]
assistant -> "Nothing was deleted."
build/out.txt: still there

Lesson 4, eleven lessons back. You decided how wide the tool boundary's except should be and settled on Exception. Below, a tool raises lab.PowerCut, the stand-in lesson 4 gave you for something that is not a tool failing. What does the record hold when the dust settles?

The prompt and the reply with its call, and no result for that call. Nothing else

It went straight past the boundary and out of the run. Remember the shape: a call with no result.

A result marked is_error saying PowerCut, and the loop asks the model what to do next

That is what happens to everything a tool fails with. This is not the tool failing; it is the program being told to end.

A result, made up by the loop, saying the call was interrupted, so that the record stays valid

Lesson 10 does write such a result, but into the view it computes, never into the record. The record is what happened.

Out of the run came the PowerCut, and the record stopped where it stood: U A[c1], a call nobody answered. Hold on to that shape. By the end of this page you will have made it on purpose, and lesson 10 will make it harmless.

import harness, lab


def execute(arguments):
    """A tool that does not fail. Something else is ending the program it runs in."""
    raise lab.PowerCut("the machine is going down")


tool = {"name": "build", "description": "Build the project.",
        "parameters": {"type": "object", "properties": {}}, "execute": execute}
system = harness.build_system_prompt([tool], [], [], "/work/shop", "2026-03-14")
model = lab.ScriptedModel([lab.reply(lab.call("build")), lab.say("unused")])
h = harness.Harness(model, system, [tool])

try:
    for event in h.prompt("Build it."):
        pass
except lab.PowerCut as exc:
    print("out of h.prompt() came:", type(exc).__name__, "-", exc)

print()
print("the record:", lab.shape(list(h.messages)))
print(lab.show(list(h.messages)))
print("model calls:", len(model.calls))
print("is_running:", h.is_running)

out of h.prompt() came: PowerCut - the machine is going down

the record: U A[c1]
user -> "Build it."
assistant -> toolCall c1 build({})
model calls: 1
is_running: False

Warm-up: async in three cells

This lesson needs async, await and one more thing about tasks. If you have written them before, skip this; it is about five minutes. Read each cell and decide what it prints before you press Run.

A function with await in it. What does calling it do?

# An async function is a function with `await` in it. This one is four lines.
import asyncio


async def job():
    print("  inside: starting")
    await asyncio.sleep(0)            # the floor is offered here
    print("  inside: finishing")


print("before the call")
running = job()
print("after the call, job() handed back a", type(running).__name__)
print("still nothing above this line came from inside the function.")
print()
await running
print("and now it has run, top to bottom")

before the call
after the call, job() handed back a coroutine
still nothing above this line came from inside the function.

  inside: starting
  inside: finishing
and now it has run, top to bottom

Two jobs, one thread of control. One pair has an await in its loop, the other has none. In what order do they print?

# Two jobs, one thread of control. Who runs when?
import asyncio


async def polite(name, n):
    for i in range(1, n + 1):
        print(f"  {name}{i}")
        await asyncio.sleep(0)        # the only place this loop offers the floor


async def busy(name, n):
    for i in range(1, n + 1):
        print(f"  {name}{i}")         # no await anywhere in this loop


print("two jobs with an await in the loop:")
await asyncio.gather(polite("a", 3), polite("b", 3))
print()
print("two jobs with no await in the loop:")
await asyncio.gather(busy("a", 3), busy("b", 3))

two jobs with an await in the loop:
  a1
  b1
  a2
  b2
  a3
  b3

two jobs with no await in the loop:
  a1
  a2
  a3
  b1
  b2
  b3

A job waiting for something that never comes, and somebody who wants it to stop. Watch where the interruption lands, and what cancel() does the instant it is called.

# A job that is waiting for something that never comes, and somebody who wants it to stop.
import asyncio


async def waiting():
    print("  job: waiting for the build")
    try:
        await asyncio.get_running_loop().create_future()      # never resolves
    except asyncio.CancelledError:
        print("  job: it arrived here, inside the await")
        raise
    finally:
        print("  job: tidying up on the way out")


task = asyncio.ensure_future(waiting())
await asyncio.sleep(0)                # let the job reach its await
print("task.done() before:", task.done())
task.cancel()                         # only asks
print("task.done() right after cancel():", task.done())
try:
    await task                        # a cancelled task still has to be waited for
except asyncio.CancelledError:
    print("whoever waited for it is told too")
print("task.cancelled():", task.cancelled())

  job: waiting for the build
task.done() before: False
task.done() right after cancel(): False
  job: it arrived here, inside the await
  job: tidying up on the way out
whoever waited for it is told too
task.cancelled(): True

Three facts to carry out of that: calling an async function runs none of it; a job holds the floor until it reaches an await, and then whoever else is ready gets a turn; and task.cancel() does not stop anything on the spot, it arranges for CancelledError to be raised inside the await the job is sitting in, next time the job gets a turn.

Minute three

Every run on this course has been instant. The model answers in the same breath it is asked, and lab.Shell runs pytest between two characters of output. That was a kindness, and it is over.

A real model can take thirty seconds to answer. A real build takes ten minutes. So: your agent is three minutes into a ten-minute build, the user has changed their mind, and there is a Stop button on the screen. Pressing it calls a function of yours. Before you read on, ask the smaller question.

Here is the whole call stack while that build runs, from your frontend down. Tap every place at which the Stop button's handler could run.

for event in h.prompt("Run the tests."):   # your frontend
    render(event)

# One lap of that for-loop runs all of this, in one go,
# to fetch one event:
    Harness._run  ->  run_agent      # a generator of events
        run_tool(tools, call)
            guard's execute  ->  check(name, arguments)
                bash's execute  ->  shell.run("pytest")
                    # ten minutes in here

in your frontend, in the instant between one event arriving and your asking for the next
inside bash's execute, while the command runs
inside run_agent, while it waits for the model's reply
inside run_tool, while it turns a failure into a result
inside guard's execute, while check decides

One place, and it lasts an instant. Everything under that for line is one long function call, and this program has one thread of control running it: while any of it is running, your frontend is not. You could start a second thread; nothing you have written does, and lesson 16 will show you what that costs. The cell below prints the stack as it really happens.

Your lesson 14 harness on a ten-minute build. Every line marked frontend: is a moment your code had the floor; the indented lines come from inside bash. The Stop button's handler is press_stop(), and the last line says how often it ran.

import harness, lab

ws = lab.Workspace({"src/app.py": "print(1)\n"})
bash = harness.make_bash_tool(lab.Shell(ws))
really_run = bash["execute"]

stop = {"pressed": False}           # the Stop button's handler sets this flag. Nothing else does.


def press_stop():                   # wired to the button in your terminal, three minutes in
    stop["pressed"] = True


def execute(arguments):
    print("          | bash: the build starts. Ten minutes.")
    print("          | bash: minute 3. The user presses Stop.")
    print("          | bash: minute 10. The build ends. Nobody was asked anything.")
    return really_run(arguments)


tools = [{**bash, "execute": execute}]
system = harness.build_system_prompt(tools, [], [], "/work/shop", "2026-03-14")
model = lab.ScriptedModel([lab.reply(lab.call("bash", {"command": "pytest"})),
                           lab.say("Two tests failed.")])
h = harness.Harness(model, system, tools)

looks = 0
for event in h.prompt("Run the tests."):
    looks += 1
    print(f"frontend: {event['type']}")
    if stop["pressed"]:             # your frontend looks at the flag here, and only here
        print("frontend: Stop seen. Walking away.")
        break

print()
print("times your frontend held the floor:", looks)
print("press_stop() ever ran:", stop["pressed"])

frontend: agent_start
frontend: turn_start
frontend: message_end
frontend: message_end
frontend: tool_execution_start
          | bash: the build starts. Ten minutes.
          | bash: minute 3. The user presses Stop.
          | bash: minute 10. The build ends. Nobody was asked anything.
frontend: message_end
frontend: tool_execution_end
frontend: turn_end
frontend: turn_start
frontend: message_end
frontend: turn_end
frontend: agent_end

times your frontend held the floor: 12
press_stop() ever ran: False

Twelve moments with the floor, and the build sits inside one of them. Your frontend asked for the next event, and got it back ten minutes later. In between, the only code running in that process was shell.run. The button is drawn, the handler is written, and nothing can call it. This is not a bug in your harness. It is what a plain generator is: it gives the floor away between events, and never inside one.

Where this failure comes from: Tau's agent loop is async on its very first line (src/tau_agent/loop.py:52), and every Tau block on this course so far has carried the same footnote: read async for as for until lesson 15. This is lesson 15.

The same file, a different colour

Python has a second kind of function, the kind from the warm-up. Inside one, await means: I am going to be waiting for a while; whoever else is ready may have the floor until I am done. That is the only thing you need from async, and it is enough to give the Stop button a turn.

You do not have to redesign anything for it. There is one rule, and it decides for you, function by function:

A function is async if and only if it waits, directly or through
something it calls, on the model or on a tool.

Your starter is the lesson 14 file with that colour rule applied by hand, so you do not spend an hour typing await. It is shown beside lesson 14's file in the lab's diff, and it is worth two minutes of your time, because of what is in it.

Your starter, asked about itself: one row per function, and whether it is async or plain. bash's execute and the one guard wraps it in are in there too. One row is worth a second look: Harness.prompt is plain, because it makes the run and hands it back without taking a single step in it. So when does a run begin?

import harness, inspect, lab


def colour(f):
    return "async" if inspect.iscoroutinefunction(f) or inspect.isasyncgenfunction(f) else "plain"


ws = lab.Workspace({})
bash = harness.make_bash_tool(lab.Shell(ws))
guarded = harness.guard(bash, harness.deny_destructive)

rows = [
    ("1 messages", "user_message", harness.user_message),
    ("1 messages", "error_message", harness.error_message),
    ("1 messages", "render", harness.render),
    ("2 tools", "str_arg", harness.str_arg),
    ("2 tools", "tool_specs", harness.tool_specs),
    ("2 tools", "truncate_tail", harness.truncate_tail),
    ("2 tools", "run_tool", harness.run_tool),
    ("2 tools", "make_bash_tool", harness.make_bash_tool),
    ("2 tools", "bash's execute", bash["execute"]),
    ("2 tools", "guard", harness.guard),
    ("2 tools", "the guarded execute", guarded["execute"]),
    ("2 tools", "deny_destructive", harness.deny_destructive),
    ("3 loop", "run_agent", harness.run_agent),
    ("3 loop", "context_for_model", harness.context_for_model),
    ("3 loop", "repair_tool_history", harness.repair_tool_history),
    ("4 harness", "Harness.prompt", harness.Harness.prompt),
    ("4 harness", "Harness._run", harness.Harness._run),
    ("4 harness", "Harness.subscribe", harness.Harness.subscribe),
    ("5 frontends", "FinalTextRenderer.render", harness.FinalTextRenderer.render),
    ("5 frontends", "JsonRenderer.finish", harness.JsonRenderer.finish),
    ("6 environment", "SessionLog.append_message", harness.SessionLog.append_message),
    ("6 environment", "persist_to", harness.persist_to),
    ("6 environment", "estimate_tokens", harness.estimate_tokens),
    ("6 environment", "summarize", harness.summarize),
    ("6 environment", "find_cut", harness.find_cut),
    ("6 environment", "compact", harness.compact),
    ("6 environment", "build_system_prompt", harness.build_system_prompt),
]

print(f"{'region':<16}{'in your file':<28}colour")
for region, name, f in rows:
    print(f"{region:<16}{name:<28}{colour(f)}")
print()
print("async:", sum(colour(f) == "async" for _, _, f in rows),
      " plain:", sum(colour(f) == "plain" for _, _, f in rows))

region          in your file                colour
1 messages      user_message                plain
1 messages      error_message               plain
1 messages      render                      plain
2 tools         str_arg                     plain
2 tools         tool_specs                  plain
2 tools         truncate_tail               plain
2 tools         run_tool                    async
2 tools         make_bash_tool              plain
2 tools         bash's execute              async
2 tools         guard                       plain
2 tools         the guarded execute         async
2 tools         deny_destructive            plain
3 loop          run_agent                   async
3 loop          context_for_model           plain
3 loop          repair_tool_history         plain
4 harness       Harness.prompt              plain
4 harness       Harness._run                async
4 harness       Harness.subscribe           plain
5 frontends     FinalTextRenderer.render    plain
5 frontends     JsonRenderer.finish         plain
6 environment   SessionLog.append_message   plain
6 environment   persist_to                  plain
6 environment   estimate_tokens             plain
6 environment   summarize                   async
6 environment   find_cut                    plain
6 environment   compact                     async
6 environment   build_system_prompt         plain

async: 7  plain: 20

Seven of twenty-seven. Now the other side of that listing. Tap everything the recolouring did not change.

the order in which messages are appended to the record
the event types, and the order they arrive in
SessionLog, and what one line of the log holds
what run_tool promises: one call in, exactly one result out
how many arguments run_tool takes
the assertions in the earlier lessons' hidden tests

One parameter arrived, in three kinds of place: signal, on run_tool, on run_agent and on every execute (and error_message gained a stop_reason). That is the whole list of design changes. A hundred and forty-five hidden tests from lessons 2 to 14 run against this file unchanged: the driver awaits what needs awaiting and the assertions never find out. async is a colour, not an architecture, and what it buys you is one thing: a place where your code can stop and let somebody else run.

Who is somebody else?

Nobody, yet. A Stop button needs something to set, and the run needs something to look at. The something is small: an object with a flag, cancel() to set it and is_cancelled() to read it. The Harness makes one for each run and hands it to the loop, which passes it to the model call and to every tool. Its name in the code is signal, and your starter already has the parameters, unused, in every signature.

That is the machinery, and it is not the lesson. The interesting questions are about what the run owes the record on the way out. The next two predictions run against the finished reference file, so they work before your lab does.

One reply asks for three files at once: c1, c2, c3. The first read is running when the user presses Stop. What happens to the second and the third?

They are skipped, and the run ends. There is nothing to say about work nobody did

Say it to the record, then. A reply that asks for three calls and a transcript that answers one is the shape lesson 10 spent a whole page repairing.

They run anyway. They were already asked for, and a flag cannot reach into a batch that has started

The batch is a for loop that you wrote. It can look at the flag before each lap as easily as after.

Neither is run, and each still gets a result, marked as an error, saying the run was stopped

One call in, exactly one result out, on this path as on every other.

a.py keeps its real contents, because that read had already finished; c2 and c3 come back Operation aborted, marked is_error. Then a last assistant message says the run was stopped. Count the model calls: one. A second request would be paid for, and its answer thrown away.

Where this failure comes from: Tau's loop makes the same choice in the same words. Before each call it asks whether the signal is cancelled, and if it is, the call's result is the text Operation aborted with the error flag set (src/tau_agent/loop.py:304-306).

import harness, lab

SYSTEM = "You are a careful assistant."
box, looked = {}, []


async def look(arguments, signal=None):
    looked.append(arguments["path"])
    if len(looked) == 1:
        print("(the user presses Stop while the first read is running)")
        box["h"].cancel()
    return f"contents of {arguments['path']}"


tool = {"name": "look", "description": "Read one file.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}}},
        "execute": look}
calls = [lab.call("look", {"path": path}) for path in ("a.py", "b.py", "c.py")]
model = lab.ScriptedModel([lab.reply(*calls), lab.say("This reply costs money after Stop.")])
h = box["h"] = harness.Harness(model, SYSTEM, [tool])
await lab.drive(h.prompt("Look at all three files."))

print("files really read:", looked)
print()
print(lab.show(list(h.messages), stop_reason=True))
print("shape:            ", lab.shape(list(h.messages)))
print("model calls:      ", len(model.calls))
print("record valid:     ", lab.validate(list(h.messages), record=True) == [])

(the user presses Stop while the first read is running)
files really read: ['a.py']

user -> "Look at all three files."
assistant -> toolCall c1 look({"path": "a.py"}) + toolCall c2 look({"path": "b.py"}) + toolCall c3 look({"path": "c.py"})  [stop_reason=toolUse]
toolResult c1 -> "contents of a.py"
toolResult c2 -> "Operation aborted"  [is_error]
toolResult c3 -> "Operation aborted"  [is_error]
assistant -> (nothing)  [stop_reason=aborted]  [error: Operation aborted]
shape:             U A[c1,c2,c3] R(c1) R(c2) R(c3) A(aborted)
model calls:       1
record valid:      True

Now the tool that is running. Your bash is inside await shell.arun("sleep inf"), which waits for the command and looks at nothing else. Stop is pressed. When does it stop?

At once. cancel() set the flag, and the flag is what Stop means

The flag is a boolean in an object. Setting it changes one byte of memory. Nothing anywhere reads that byte unless you wrote the line that reads it.

At the next await. The run is async now, so the interruption lands at the first place the tool gives the floor away

Giving the floor away is how the button's handler gets to run at all. It is not a delivery. The handler set a flag; an await does not read flags.

Never. That command has no end, and nothing between it and the flag is looking

Two hundred turns of the event loop later the run is exactly where it was. The fix is not clever, and it is in the same cell.

Two tools, one Stop. The one that polls is finished one turn of the event loop later, with Command cancelled as the last line of its result. The one that only awaits is still going after two hundred. This is cooperative cancellation: cancelling a run is asking it to stop, and only code that looks can hear you.

Where this failure comes from: Tau's bash does not await the command either. It waits on the command and on a watcher that reads the token every 50 milliseconds, whichever finishes first, and kills the process if the command was not the one (src/tau_coding/tools.py:769-793, polling at src/tau_coding/tools.py:812-814). The last line it appends is Command cancelled (src/tau_coding/tools.py:681-682), which is where your text comes from.

import harness, lab

SYSTEM = "You are a careful assistant."


def deaf_tool():
    """A build tool that awaits the command and looks at nothing else."""
    shell = lab.Shell(lab.Workspace({}))

    async def execute(arguments, signal=None):
        return (await shell.arun("sleep inf"))[0]

    return {"name": "build", "description": "Build the project. Takes ten minutes.",
            "parameters": {"type": "object", "properties": {}}, "execute": execute}


async def press_stop_on(label, tool, call):
    model = lab.ScriptedModel([lab.reply(call), lab.say("unused")])
    h = harness.Harness(model, SYSTEM, [tool])
    async with lab.background(h.prompt("Build it.")) as task:
        await lab.ticks(20)
        h.cancel()                                  # the flag is set here
        turns = 0
        while turns < 200 and not task.done():
            await lab.tick()
            turns += 1
        print(f"{label}:")
        print(f"  Stop, then {turns} turn(s) of the event loop:",
              "the run is over" if task.done() else "the run is STILL GOING")
        print("  record:", lab.shape(list(h.messages)))
        if task.done():
            print("  last line of the tool's result:",
                  repr(h.messages[2]["content"].splitlines()[-1]))


bash = harness.make_bash_tool(lab.Shell(lab.Workspace({})))
await press_stop_on("a tool that looks at the flag (your bash)", bash,
                    lab.call("bash", {"command": "sleep inf"}))
print()
await press_stop_on("a tool that never looks", deaf_tool(), lab.call("build"))

a tool that looks at the flag (your bash):
  Stop, then 1 turn(s) of the event loop: the run is over
  record: U A[c1] R(c1) A(aborted)
  last line of the tool's result: 'Command cancelled'

a tool that never looks:
  Stop, then 200 turn(s) of the event loop: the run is STILL GOING
  record: U A[c1]

So the flag has to be read, by hand, wherever a run spends time. Tap every place where your run spends time that will have to do the reading.

at the top of each turn, before the model is called
in run_tool, before it runs a tool
inside bash, while the command runs
in context_for_model, while it builds the request
in repair_tool_history, while it pairs calls with results
in persist_to, while it writes a line to the log

Three places, and they are the three that wait. The other three are arithmetic over lists in memory: they are finished before anybody could press anything, and a check in them would be a line that is never true. (persist_to writes to a workspace that is a dict; on a real disk that line stops being free, and you would think again.) The rule is not "check everywhere", it is "check wherever you wait". Each of the three owes the record something different, and that is the lab.

Build: ask the run to stop

The rungs took the wrong designs away one at a time: a flag that stops things by itself, an interruption that arrives at the next await, a batch that skips its remaining calls without answering them, a paid request after Stop. What is left is about thirty lines, spread over the three places where your run waits, plus the flag itself.

Five things to write, about thirty lines, in ten marked gaps. The signatures and the signal=None parameters are already there; you change no signature you already have.

CancelToken, three methods. A new one is not cancelled, cancel() sets it, and it stays set. Beside it, ABORTED = "Operation aborted": a stopped run says that in two places, as the result of every call it did not run and as the last message's error_message.
error_message(text, stop_reason="error"). Lesson 5's body, with the new parameter used. One argument still means an in-band error; stop_reason="aborted" is the same message with a different label.
Harness, about five lines in three places. A fresh token for every run, made where is_running turns True; cancel() cancels the current run's token and must do nothing, and not raise, when there is no run; _run hands it to run_agent as signal=.
run_agent, three small changes. At the top of a turn, after the max_turns check and before the model call: a cancelled signal means the reply is error_message(ABORTED, stop_reason="aborted") and the model is not called at all. A reply labelled aborted ends the run the way an errored one does. And run_tool needs the signal too. context_for_model gets the matching one-word change: an empty aborted message is as unwelcome in a request as an empty error one.
run_tool and bash. A cancelled signal means the tool is not run and the call still gets its one result, ABORTED, with is_error set; a tool that is run is handed the signal. In bash, await shell.arun(command) becomes a poll loop over process = shell.start(command), which returns at once: process.done(), process.kill(), process.output() once it is done. Until it is done, a cancelled signal ends the call with "Command cancelled", and otherwise await asyncio.sleep(0) gives the floor away for one turn of the event loop, which is when Stop's code gets to run.

Two things the tests are strict about. The hidden tests count turns of the event loop, never seconds, so await asyncio.sleep(0) is the only wait you need. And however bash's execute is left, the command must not still be running when it is over.

One thing is done for you, and the diff marks it # given in lesson 15: both renderers now treat a stopped run as a run that did not end well. A frontend that prints an empty answer and exits zero because the user pressed Stop is wrong, but it is not today's idea.

One token for the harness's life, or a fresh one per run? Imagine the user stopping run 1, then typing again. And when exactly does a run begin: when prompt() is called, or when the consumer asks for the first event?
Take the three waits one at a time and ask what the record owes each. The top of the turn owes an assistant message that says the run was stopped, and owes the user no paid request. run_tool owes every call a result, whether or not the tool ran. bash owes a result that says why there is no output, and owes the machine a killed process. For the harness: a run begins where is_running turns True, because a frontend hands the run to a task and the user can press Stop before the task takes its first step; that is where the token has to exist. cancel() with nothing running is not a mistake, it is a user who was a moment late, and it must do nothing.
In outline. CancelToken: one attribute, set in __init__, set again by cancel(), returned by is_cancelled(). Harness.prompt: after the run guard, a new token on self; cancel: one line on that token; _run: one more keyword argument. run_agent: the turn-top if gains a branch between the max_turns branch and the model call; failed compares against two labels instead of one; the run_tool call gains an argument. run_tool: a branch before the "not found" branch, guarded by signal is not None and signal.is_cancelled(), setting the same two names the other branches set; the execute call gains an argument. bash: start, then while not process.done(): — inside, return the cancelled text if the signal says so, else sleep zero; then, outside the loop, the output. Wrap the loop so that the process is killed on every way out.

Thirty-odd lines, and the button on the screen is wired to something. Look at what you did not write: a thread, a timeout, a second loop, or one character inside repair_tool_history. Your run now stops because three pieces of slow code agreed to look.

What changed since lesson 14

The line-by-line diff needs JavaScript. The whole file this exercise starts from is printed at the end of it.

One reply asks for three tools and the first, while it runs, calls h.cancel(). The first call keeps its real result, the second and third are not run and each gets the result "Operation aborted" marked is_error, the model is not called again, and the run ends with an assistant message whose stop_reason is "aborted".
The events of that stopped run are the canonical sequence: every call still has its start, its recorded result and its end; then a last turn holding only the aborted message; then agent_end. Every message on the record was announced, in order.
CancelToken() starts clear, cancel() sets it and it stays set. run_tool with a cancelled token does not run the tool and answers "Operation aborted"; run_agent with a cancelled token records the prompt and the aborted message and calls nobody. With a clear token both behave as before.
signal is optional: run_agent called with no signal, as every caller written before this lesson calls it, runs its tools and ends normally.
error_message(text) is still an "error" message; error_message(text, stop_reason="aborted") is the same message with stop_reason "aborted".
h.cancel() called after prompt() has returned, before anyone has taken the run's first event, still stops the run: no model call, and the record is the prompt and the aborted message.
cancel() on a harness that never ran, or between runs, does nothing. After run 1 was stopped, run 2 on the same harness runs its tools and ends normally.
After a stopped run the next prompt is accepted. The request it sends holds everything that happened except the empty aborted message, which stays on the record.
A follow-up queued during a run that is then stopped is not delivered into the stopped run: it waits, and opens a turn of the next run.
Stop pressed while the request is on its way: the model is handed the run's signal, so it gives up at once with an "aborted" reply, and nothing in it is acted on.
A tool's execute(arguments, signal) receives the run's token, bare or behind lesson 14's guard: it reads False, and True once h.cancel() has been called.
bash is running `sleep inf`. Within five turns of the event loop after h.cancel() the command has been killed, its result ends with the line "Command cancelled", and the run has ended with the aborted message. The same through lesson 14's guard.
A command that takes three turns of the event loop and then prints is waited for: its output comes back as an ordinary result.
Both renderers' finish() return False for a run whose last message is "aborted", and FinalTextRenderer prints the reason, not an empty answer.

import asyncio # given in lesson 15 import json from collections import deque # == 1 messages == # A conversation is a list of messages. Each message is a dict tagged with the role that spoke. def user_message(text): """A message from the user. Its content is a plain string.""" return {"role": "user", "content": text} def text_of(message): """The words of an assistant message. Its content is a list of blocks, not a string (lesson 2 explains why); the words are in the blocks of type "text".""" return "".join(block["text"] for block in message["content"] if block["type"] == "text") def tool_calls(message): """The tool calls an assistant message asks for, in the order it asked. Worked out from `content` on every call and stored nowhere else.""" return [block for block in message["content"] if block["type"] == "toolCall"] def error_message(text, stop_reason="error"): """An assistant message that says why a run stopped: nothing in `content`, the reason in `error_message`, and stop_reason "error", or "aborted" when it was the user who stopped it. A stop is written into the record, not raised.""" # --- your code --- # Lesson 5's body. The signature has a new parameter; use it. return {"role": "assistant", "content": [], "stop_reason": "error", "error_message": text} # --- end --- # --- your code --- # ABORTED, the text "Operation aborted". A stopped run says it in two places: as the result of # every tool call it did not run, and as the error_message of its last assistant message. # --- end --- def render(messages): """The transcript as plain text, one line per message, tool calls as name(arguments): what a summary is written from, and what estimate_tokens measures.""" lines = [] for m in messages: if m["role"] == "assistant": calls = [f"{c['name']}({json.dumps(c['arguments'])})" for c in tool_calls(m)] words = " ".join([text_of(m)] + calls) else: words = m["content"] lines.append(f"{m['role']}: {words}".replace("\n", " ")) return "\n".join(lines) # given in lesson 15: execute is async now, and is handed the run's signal # == 2 tools == # A tool is a dict. "name", "description" and "parameters" are advertised to the model. # "execute" is an async function, execute(arguments, signal=None) -> str, that only this program # ever awaits. # "snippet" and "guidelines" are optional and are for the system prompt: one line on what the # tool is for, and advice on when to use it. A schema can say what; only words can say when. def str_arg(arguments, name): """The string argument `name`, checked. The schema asked the model for a string and nobody made it send one, so every tool reads its arguments through here. Raises ValueError("<name> must be a string") when it is missing or not a string.""" value = arguments.get(name) if not isinstance(value, str): raise ValueError(f"{name} must be a string") return value def int_arg(arguments, name, default): """The optional whole-number argument `name`, checked; `default` if the model left it out. Raises ValueError("<name> must be an integer") for anything else.""" value = arguments.get(name) if value is None: return default if not isinstance(value, int): raise ValueError(f"{name} must be an integer") return value def tool_specs(tools): """What the model is told about the tools: everything except the function.""" return [{"name": tool["name"], "description": tool["description"], "parameters": tool["parameters"]} for tool in tools] async def run_tool(tools, call, signal=None): """Run one tool call and return the toolResult message that answers it. One call in, exactly one result out, on every path. Tools just raise; this is the one place that turns a failure into a result the model can read. is_error means the tool could not do its job. It never raises Exception. After Stop (a cancelled `signal`) the tool is not run, and the call still gets its one result: ABORTED. A tool that is run is handed the signal.""" # --- your code --- # Below is lesson 4's body, recoloured. `signal` is a CancelToken or None, and nothing looks # at it yet. Two changes: # - if the signal is cancelled, do not run the tool: the result is ABORTED, is_error True. # One call in, one result out, on this path too # - hand the signal on to the tool: execute(arguments, signal) tools_by_name = {tool["name"]: tool for tool in tools} tool = tools_by_name.get(call["name"]) if tool is None: content, is_error = f"Tool {call['name']} not found", True else: try: content, is_error = await tool["execute"](call["arguments"]), False except Exception as exc: content, is_error = str(exc), True return {"role": "toolResult", "tool_call_id": call["id"], "tool_name": call["name"], "content": content, "is_error": is_error} # --- end --- MAX_LINES = 2000 # The budget of one tool result. Whatever a tool returns is re-read, and MAX_BYTES = 50_000 # paid for, on every later turn. Bytes, not characters: "é" is two. def truncate_head(text, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The start of `text` that fits the budget, for files: (kept, lines_shown). Whole lines only, newlines included, sized in UTF-8 bytes. If even the first line is too big, the answer is ("", 0): half a line is worse than none.""" kept, size = [], 0 for line in text.splitlines(keepends=True): size += len(line.encode()) if len(kept) == max_lines or size > max_bytes: break kept.append(line) return "".join(kept), len(kept) def truncate_tail(text, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The end of `text` that fits the budget, for command output, where the verdict comes last: (kept, lines_shown). Whole lines only, newlines included, sized in UTF-8 bytes.""" kept, size = [], 0 for line in reversed(text.splitlines(keepends=True)): size += len(line.encode()) if len(kept) == max_lines or size > max_bytes: break kept.append(line) return "".join(reversed(kept)), len(kept) def make_read_tool(ws, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The `read` tool, over the workspace `ws`. One call returns at most the budget, and says so in a last line that tells the model how to ask for the rest.""" # given in lesson 15: async, because run_tool awaits every tool; `signal` is not used yet async def execute(arguments, signal=None): path = str_arg(arguments, "path") offset, limit = int_arg(arguments, "offset", 1), int_arg(arguments, "limit", None) if offset < 1 or (limit is not None and limit < 1): raise ValueError("offset and limit must be 1 or more") if not ws.exists(path): raise FileNotFoundError( f"File not found: {path}. Files here: {', '.join(ws.listdir('.'))}") lines = ws.read_text(path).splitlines(keepends=True) wanted = lines[offset - 1:] if limit is None else lines[offset - 1:offset - 1 + limit] kept, shown = truncate_head("".join(wanted), max_lines, max_bytes) if offset > max(len(lines), 1): raise ValueError(f"offset {offset} is past the end of {path}: {len(lines)} lines") if wanted and shown == 0: skip = f" Use offset={offset + 1} to skip it." if offset < len(lines) else "" return (f"[Line {offset} of {path} is {len(wanted[0].encode())} bytes, over the " f"{max_bytes}-byte budget of one read, so none of it is shown.{skip}]") last = offset + shown - 1 if last < len(lines): kept += (f"[Showing lines {offset}-{last} of {len(lines)}. " f"Use offset={last + 1} to continue.]") return kept return { "name": "read", "description": "Read a text file and return its contents. One call returns at most " f"{max_lines} lines or {max_bytes} bytes; use offset to read on.", "parameters": { "type": "object", "properties": {"path": {"type": "string", "description": "Path of the file to read."}, "offset": {"type": "integer", "description": "First line, from 1."}, "limit": {"type": "integer", "description": "How many lines."}}, "required": ["path"], }, "execute": execute, "snippet": "Read file contents", "guidelines": ["Use read to examine files instead of cat or sed.", "Show file paths clearly when working with files."], } def make_write_tool(ws): """The `write` tool, over the workspace `ws`: create a file, or replace all of it.""" # given in lesson 15: async, because run_tool awaits every tool async def execute(arguments, signal=None): path, content = str_arg(arguments, "path"), str_arg(arguments, "content") ws.write_text(path, content) return f"Successfully wrote to {path}." return { "name": "write", "description": "Write a text file. Creates the file, or replaces its whole contents.", "parameters": { "type": "object", "properties": {"path": {"type": "string", "description": "Path of the file to write."}, "content": {"type": "string", "description": "The new contents."}}, "required": ["path", "content"], }, "execute": execute, "snippet": "Create or overwrite files", "guidelines": ["Use write only for new files or complete rewrites.", "Show file paths clearly when working with files."], } def make_bash_tool(shell, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The `bash` tool: run one command line in `shell` and return what it printed. A command that fails has still been run: its exit code is news for the model, not a tool error, so it comes back as an ordinary result that ends with a status line.""" async def execute(arguments, signal=None): # --- your code --- # `await shell.arun(command)` waits for the command and looks at nothing else: against # `sleep inf` no flag will ever stop it. Replace that line with a poll loop over # process = shell.start(command) starts the command and returns at once # process.done() -> bool process.kill() # process.output() -> (output, code) only once it is done # Until the process is done: if the signal is cancelled, kill the process and return # "Command cancelled"; otherwise `await asyncio.sleep(0)`, which gives the floor away # for one turn of the event loop (that is when Stop's code gets to run), and look again. # However execute is left, the command must not be left running. output, code = await shell.arun(str_arg(arguments, "command")) # --- end --- total = len(output.splitlines()) output, shown = truncate_tail(output, max_lines, max_bytes) if shown < total: output += f"\n[Showing the last {shown} of {total} lines. Narrow the command for more.]" if code != 0: output += f"\nCommand exited with code {code}" return output return { "name": "bash", "description": "Run a shell command and return its output (stdout and stderr together). " f"Only the last {max_lines} lines or {max_bytes} bytes come back.", "parameters": { "type": "object", "properties": {"command": {"type": "string", "description": "The command line."}}, "required": ["command"], }, "execute": execute, "snippet": "Run shell commands", "guidelines": ["Say what a command is for before you run it."], } def guard(tool, check): """`tool` behind a door: the same name, schema and everything else, so neither the model nor the loop can tell, with an execute that first asks check(name, arguments): None, or the reason to refuse. A refusal is raised, so run_tool makes it the call's one result, is_error and all. A check that raises has not said yes: the call is blocked (fail closed).""" # given in lesson 15: async, because it waits on the real tool, which it hands the signal. # check() stays a plain function: it waits on nothing async def execute(arguments, signal=None): try: reason = check(tool["name"], arguments) except Exception as exc: # a check that crashed has not said yes reason = f"the check failed: {exc}" if reason is not None: raise PermissionError(f"Tool call blocked: {reason}") return await tool["execute"](arguments, signal) return {**tool, "execute": execute} def deny_destructive(name, arguments): """A check for guard: refuses a bash command that contains "rm -rf". A deny-list is a speed bump, not a sandbox: `rm -fr build` walks straight past it.""" if name == "bash" and "rm -rf" in str(arguments.get("command")): return "rm -rf deletes files for good. Ask the user to run it themselves if they want it." return None def confirm_with(human): """A check for guard that asks a person. human.ask(question) answers True for yes; anything else is a no, and the model is told so.""" def check(name, arguments): if human.ask(f"Allow {name} {json.dumps(arguments)}?"): return None return "the user said no" return check # == 3 loop == # The code that talks to the model. The model keeps nothing between calls. # given in lesson 15: async, because it waits on the model and on tools; and one new # parameter, signal async def run_agent(model, system, messages, tools, prompt, *, max_turns=None, get_steering=None, get_follow_ups=None, signal=None): """Put `prompt` to the model and keep going until it stops asking for tools, a reply comes back failed (stop_reason "error"), `max_turns` model calls have been made (None: no limit), or `signal` (a CancelToken, or None) is cancelled: the run then ends with an "aborted" message. An async generator of events: dicts with a "type", which say what happened and draw nothing. They come in balanced pairs on every exit: agent_start ... agent_end around the run, and turn_start ... turn_end around each turn. The one ordering rule is mutate, then announce: a message is appended to `messages` (the caller's list, the record) first, and only then is its message_end yielded. The final message is the last of agent_end's "messages". The user may speak during a run; the loop asks for their words only where a user message keeps the transcript valid. `get_steering` and `get_follow_ups` take no arguments and return a list of user messages, often empty: steering is asked for once a turn's whole tool batch is answered, follow-ups only when the run would otherwise end. The loop never sees a queue.""" new = [] # this run's messages, for agent_end def record(message): """Append `message` to the record, then make the event that announces it.""" messages.append(message) new.append(message) return {"type": "message_end", "message": message} def ask(get): """The user messages a callable has for us: [] if it has none, or was never passed.""" return list(get()) if get else [] yield {"type": "agent_start"} turn, calls, failed = 0, [], False # User messages waiting to open the next turn. Steering sent while idle goes in with the prompt. pending = [user_message(prompt)] + ask(get_steering) while pending: # outer loop: the prompt, then each follow-up while calls or pending: # inner loop: one pass is one turn turn += 1 yield {"type": "turn_start"} for message in pending: # only here, at the top of a turn, may a user speak yield record(message) pending = [] # --- your code --- # 1 of 3, the top of a turn. After the max_turns check and before the model call: if # the signal is cancelled, the reply is error_message(ABORTED, stop_reason="aborted") # and the model is not called (a real request costs money after Stop). Otherwise the # model is called as before, and handed the signal as a fourth argument. if max_turns is not None and turn > max_turns: # checked before the call, never after reply = error_message(f"Agent stopped after max_turns={max_turns}") else: reply = await model.acomplete(system, context_for_model(messages), tool_specs(tools)) # --- end --- yield record(reply) # --- your code --- # 2 of 3. A reply that failed: none of it is acted on and nobody is asked for more. # The same goes for one that was stopped; the model itself answers "aborted" when # Stop lands while it is answering. failed = reply["stop_reason"] == "error" # --- end --- calls = [] if failed else tool_calls(reply) for call in calls: yield {"type": "tool_execution_start", "call": call} # --- your code --- # 3 of 3. run_tool needs the signal too. result = await run_tool(tools, call) # --- end --- yield record(result) # no yield between the tool returning and this yield {"type": "tool_execution_end", "call": call, "result": result} yield {"type": "turn_end", "message": reply} if not failed: pending = ask(get_steering) # every call of the turn has its result if not failed: pending = ask(get_follow_ups) # no calls, no steering: the run would end here yield {"type": "agent_end", "messages": new} # the one way out, so every exit is balanced def context_for_model(messages): """What the model is sent. The record is not the view: the record keeps what happened, and this builds, for every request, a transcript a provider accepts. Assistant messages that failed or were stopped and are empty are left out (one that got something out stays in), then the calls and results are paired up. Returns a new list; the record is never changed.""" # --- your code --- # Lesson 10's body. An empty "aborted" message is as unwelcome in a request as an empty # "error" one. kept = [m for m in messages if not (m["role"] == "assistant" and not m["content"] and m.get("stop_reason") == "error")] return repair_tool_history(kept) # --- end --- INTERRUPTED = ("Tool call interrupted: no result was recorded. It may not have run, or may have " "run partly; check before repeating it.") def repair_tool_history(messages): """A copy of `messages` in which every tool call is followed at once by exactly one result, in call order. A recorded result is kept and moved to its place; of two for one call the first wins; a result whose call is nowhere is dropped (nobody knows the arguments of the call, so it cannot be invented); a call with no result gets INTERRUPTED, marked is_error. Pure and idempotent: a new list of the same message objects, equal to a valid input. Call ids are taken to be unique.""" recorded = {} # call id -> the first result recorded for it for m in messages: if m["role"] == "toolResult": recorded.setdefault(m["tool_call_id"], m) repaired = [] for m in messages: if m["role"] == "toolResult": continue # results are put back below, after their call repaired.append(m) if m["role"] == "assistant": for call in tool_calls(m): repaired.append(recorded.get(call["id"]) or { "role": "toolResult", "tool_call_id": call["id"], "tool_name": call["name"], "content": INTERRUPTED, "is_error": True}) return repaired # == 4 harness == # One transcript, one writer. The loop keeps the mechanics and stays stateless; this holds the list. # given in lesson 15: the class line and its docstring class CancelToken: """The Stop flag of ONE run. cancel() only sets it; nothing stops until slow code looks: the loop at the top of each turn, run_tool before each call, the model, a polling tool.""" # --- your code --- # Three methods: __init__(self), cancel(self), is_cancelled(self) -> bool. # A new token is not cancelled; a cancelled one stays cancelled. # --- end --- # given in lesson 15: the docstring's words on cancel class Harness: """Owns the one transcript, and the settings that stay the same from run to run, and lets one run at a time write to it. Harness(model, system, tools, messages=(), max_turns=None) copies `messages` into a list of its own. .messages a tuple, a snapshot of the transcript: nobody outside can append to it or pop from it. (The dicts inside are not copied.) .is_running True from the call of prompt() until that run is exhausted, closed or dies. .prompt(text) returns the run: run_agent's events, over the harness's own list. While a run is going on it raises RuntimeError("already running; use steer() or follow_up()"), and it raises on the caller's line, so prompt is a plain def and the async generator is _run. .steer(text), .follow_up(text) queue a user message: steer for the next gap between turns ("use spaces"), follow_up for when the run would otherwise end ("now run the tests"). Callable at any time, by anyone. A queued message is not history until the loop asks for it, one message per ask. Queues outlive runs. .subscribe(listener) listener(event) hears every event of every run, after the record has changed and before the run's consumer is handed the event. Returns the function that undoes it. A listener that raises ends the run: listeners are trusted code, and a saver that cannot save must not be silent. .replace_messages(messages) swaps in a whole new transcript, for whoever rebuilt it from the log. The one door in, and shut during a run: RuntimeError. .cancel() Stop: cancels the current run's token, from the moment prompt() returns. Every run gets a fresh token, so when idle this does nothing.""" # --- your code --- # Below is lesson 14's Harness, recoloured. About 5 new lines, in three places: # - prompt: a fresh CancelToken for every run, kept on self. A run begins here, where # is_running turns True, so a Stop pressed before its first step is not lost # - cancel: cancel the current run's token. With nothing running it must do nothing, and # must not raise # - _run: hand the token to run_agent as signal= # One of the hidden tests. The model asks for three tools in one reply; the first, while it # runs, does what a user at the keyboard would do: # def look(arguments): # runs as c1 of the batch [c1, c2, c3] # if arguments["path"] == "a.py": # h.cancel() # return "contents of " + arguments["path"] # async for event in h.prompt("Look at all three files."): # pass # assert lab.shape(h.messages) == "U A[c1,c2,c3] R(c1) R(c2) R(c3) A(aborted)" # assert [m["content"] for m in h.messages[2:5]] == [ # "contents of a.py", "Operation aborted", "Operation aborted"] # assert len(model.calls) == 1 # no paid request follows a Stop def __init__(self, model, system, tools, messages=(), max_turns=None): self._model, self._system, self._tools = model, system, list(tools) self._messages, self._max_turns = list(messages), max_turns self._running = False self._steering, self._follow_ups = deque(), deque() self._listeners = [] @property def messages(self): return tuple(self._messages) @property def is_running(self): return self._running def prompt(self, text): if self._running: # checked now, not at the first next() raise RuntimeError("already running; use steer() or follow_up()") self._running = True return self._run(text) def steer(self, text): self._steering.append(user_message(text)) def follow_up(self, text): self._follow_ups.append(user_message(text)) def subscribe(self, listener): self._listeners.append(listener) def unsubscribe(): if listener in self._listeners: # undoing twice is harmless self._listeners.remove(listener) return unsubscribe def _notify(self, event): for listener in list(self._listeners): # a copy: a listener may unsubscribe while called listener(event) def replace_messages(self, messages): if self._running: raise RuntimeError("cannot replace messages during a run") self._messages = list(messages) def cancel(self): pass def _drain(self, queue): """What the loop is handed when it asks: the oldest queued message, one at a time.""" return [queue.popleft()] if queue else [] async def _run(self, text): run = run_agent(self._model, self._system, self._messages, self._tools, text, max_turns=self._max_turns, get_steering=lambda: self._drain(self._steering), get_follow_ups=lambda: self._drain(self._follow_ups)) try: async for event in run: self._notify(event) # notify, then yield: whoever stops listening to yield event # the run, the listeners have already heard finally: # exhausted, closed by the consumer, or dead await run.aclose() self._running = False # --- end --- # == 5 frontends == # A frontend is a fold over the events of one run: render(event) for each event, then finish(), # which says whether the run ended well. The loop never learns what its events look like. # given in lesson 15: a run that was stopped ("aborted") did not end well either class JsonRenderer: """Every event as one line of JSON, for another program to read.""" def __init__(self): self.failed = False def render(self, event): print(json.dumps(event)) if (event["type"] == "message_end" and event["message"].get("stop_reason") in ("error", "aborted")): self.failed = True def finish(self): return not self.failed # given in lesson 15: "aborted" is reported like "error" class FinalTextRenderer: """Only the answer, for `> answer.txt`: nothing is printed while the run goes on. finish() prints the text of the last assistant message and returns True; if that message failed or was stopped (stop_reason "error" or "aborted") it prints "Error: <its error_message>" instead and returns False.""" def __init__(self): self.last = None # the newest assistant message seen so far def render(self, event): if event["type"] == "message_end" and event["message"]["role"] == "assistant": self.last = event["message"] def finish(self): if self.last is None: return False if self.last["stop_reason"] in ("error", "aborted"): print(f"Error: {self.last['error_message']}") return False print(text_of(self.last)) return True # == 6 environment == # What a session needs around the harness, starting with a disk. Regions 3 and 4 know none of it. class SessionLog: """A session on disk: the file `path` in the workspace `ws`, one JSON object per line, one line per message: {"id": "e1", "type": "message", "message": {...}}. Lines are appended and never edited. The log is the truth and the transcript is what replay() makes of it, so resuming is Harness(..., messages=log.replay()). Holds no state but `ws` and `path`.""" def __init__(self, ws, path): self.ws, self.path = ws, path def entries(self): """Every entry, oldest first; [] if there is no file yet. A line that is not JSON (a torn write) raises ValueError naming its line number. It is never skipped: the next append would be glued onto it, and both entries lost.""" text = self.ws.read_text(self.path) if self.ws.exists(self.path) else "" entries = [] for number, line in enumerate(text.splitlines(), 1): try: entries.append(json.loads(line)) except ValueError: raise ValueError(f"{self.path}, line {number}: not a complete log entry") from None return entries def append_message(self, message): """Append one message as one line and return its entry id, "e" + its line number. The number is counted by re-reading the file, so two logs over one file cannot disagree.""" entry = {"id": f"e{len(self.entries()) + 1}", "type": "message", "message": message} self.ws.append_text(self.path, json.dumps(entry) + "\n") return entry["id"] def append_compaction(self, summary, first_kept_id): """Append a compaction entry and return its id. From now on the entries before `first_kept_id` are read as `summary`. Nothing is deleted.""" entry = {"id": f"e{len(self.entries()) + 1}", "type": "compaction", "summary": summary, "first_kept_id": first_kept_id} self.ws.append_text(self.path, json.dumps(entry) + "\n") return entry["id"] def rows(self): """(entry id, message) for every message of the transcript, in order. A compaction entry is a rule for reading what came before it: the rows before its first_kept_id become one row, (its own id, the summary as a user message), and the rows from there on stay.""" rows = [] for entry in self.entries(): if entry["type"] == "message": rows.append((entry["id"], entry["message"])) elif entry["type"] == "compaction": ids, first = [row_id for row_id, _ in rows], entry["first_kept_id"] kept = rows[ids.index(first):] if first in ids else [] summary = user_message("Previous conversation summary:\n" + entry["summary"]) rows = [(entry["id"], summary)] + kept return rows def replay(self): """The transcript: the state is computed from what happened, never saved.""" return [message for _, message in self.rows()] def persist_to(log): """A listener for Harness.subscribe that appends every completed message to `log`, as it completes: saving is nobody's job but this one's, whoever consumes the run.""" def listener(event): if event["type"] == "message_end": log.append_message(event["message"]) return listener def estimate_tokens(messages): """About how many tokens `messages` cost: four characters to a token. No tokenizer, the same answer every time, and it undercounts: leave room.""" return len(render(messages)) // 4 SUMMARY_SYSTEM = ("You summarise a conversation between a user and a coding agent. Do NOT continue " "the conversation and do NOT answer any question in it. Reply with the summary " "only: the goal, what is done, what comes next, exact file paths and errors.") # given in lesson 15: async, because it waits on the model async def summarize(model, messages): """A summary of `messages`, from one model call that is no part of any transcript: its own system prompt, one user message holding the rendered conversation in <conversation> tags, and no tools, so there is nothing to do but summarise. Raises RuntimeError if the reply failed or is empty.""" wrapped = user_message(f"<conversation>\n{render(messages)}\n</conversation>") reply = await model.acomplete(SUMMARY_SYSTEM, [wrapped], []) summary = text_of(reply).strip() if reply["stop_reason"] == "error" or not summary: raise RuntimeError(f"no summary: {reply.get('error_message', 'the reply was empty')}") return summary def find_cut(rows, keep_recent_tokens): """Where to cut `rows` (SessionLog.rows()) for a compaction: the index of the first row to keep. Rows before it are summarised. 0 means all of it fits in `keep_recent_tokens`. The cut is at a turn boundary, a user message, and failing that at least never on a tool result: a kept tail that opens on a result whose call was summarised away is a 400.""" cut, kept = len(rows), 0 while cut > 0 and kept < keep_recent_tokens: # walk back from the newest row cut -= 1 kept += estimate_tokens([rows[cut][1]]) if kept < keep_recent_tokens: return 0 for index in range(cut, len(rows)): # snap forward to the next user message if rows[index][1]["role"] == "user": return index while cut < len(rows) and rows[cut][1]["role"] == "toolResult": cut += 1 # no user message left: step past the results return cut # given in lesson 15: async, because it waits on summarize async def compact(log, model, *, keep_recent_tokens): """Compact the session in `log`: summarise the old rows and append ONE compaction entry. The log only grows; replay() now gives [summary] + [kept tail]. False, with nothing appended, if there is nothing to summarise or nothing would be kept, or if the summary failed: housekeeping never costs the user their turn.""" rows = log.rows() cut = find_cut(rows, keep_recent_tokens) if cut <= 0 or cut >= len(rows): return False try: summary = await summarize(model, [message for _, message in rows[:cut]]) except RuntimeError: return False log.append_compaction(summary, rows[cut][0]) return True # given in lesson 15: async, because it waits on compact async def maybe_compact(harness, log, model, *, window, reserve, keep_recent_tokens): """Compact if the transcript has come within `reserve` tokens of the context `window`, then give the harness what the log now replays. For the code around the harness to call before each prompt, never during a run. True if it compacted.""" if harness.is_running: raise RuntimeError("cannot compact during a run") if estimate_tokens(harness.messages) <= window - reserve: return False if not await compact(log, model, keep_recent_tokens=keep_recent_tokens): return False harness.replace_messages(log.replay()) return True def discover_context(ws, cwd): """The project's standing instructions: every AGENTS.md from the top of the workspace `ws` down to the directory `cwd` (a plain path such as "services/api", no ".."), as [(path, text)]. Broad to specific, each path once: the file nearest the work comes last and has the last word. Whoever wrote these files is writing your system prompt.""" folders = [""] for part in cwd.split("/"): if part not in ("", "."): folders.append(folders[-1] + part + "/") return [(folder + "AGENTS.md", ws.read_text(folder + "AGENTS.md")) for folder in folders if ws.exists(folder + "AGENTS.md")] def build_system_prompt(tools, context_files, skills, cwd, today): """The system prompt, computed from what is really there: the enabled `tools` (a line for each "snippet", then every "guidelines" entry once, first seen first), `context_files` [(path, text)] wrapped with their paths in the order given, and an index of `skills` [{"name", "description", "path"}]. A skill's body stays on disk and `read` is its loader, so without a tool named read there is no index. The date and the directory come last. It reads no clock and no disk: the same arguments give the same bytes, which tests and prefix caches rely on.""" guidelines = [] for tool in tools: for guideline in tool.get("guidelines", []): if guideline not in guidelines: # a list, not a set(): a set has no order to keep guidelines.append(guideline) tool_lines = [f"- {tool['name']}: {tool['snippet']}" for tool in tools if tool.get("snippet")] sections = ["You are a coding assistant working in the user's project."] if tool_lines: sections.append("Available tools:\n" + "\n".join(tool_lines)) if guidelines: sections.append("Guidelines:\n" + "\n".join(f"- {guideline}" for guideline in guidelines)) for path, text in context_files: sections.append(f'<project_instructions path="{path}">\n{text}\n</project_instructions>') if skills and any(tool["name"] == "read" for tool in tools): index = [f"- {skill['name']}: {skill['description']} ({skill['path']})" for skill in skills] sections.append("Skills are files of instructions for particular tasks. When a task " "matches a description, read the file first.\n" + "\n".join(index)) sections.append(f"Current date: {today}\nCurrent working directory: {cwd}") return "\n\n".join(sections)Running code needs JavaScript. Above is the file this exercise starts from (harness.py). You can run its tests on your own machine: python lab/lab_check.py --harness harness.py --lesson 15 tests.py

Lesson 5 left you a list of the ways a run can end, and every one of them left a transcript you could send again. That list just got one longer, and the new one obeys the same rule.

Figure 15.1 Every exit is balanced, and cancel is now one of them.

Exit 1, the ordinary one: the reply asks for no tools, so the loop returns it. The last thing on the record is a plain assistant message.
Exit 2, max_turns, checked at the top of a turn. The last message is an assistant marked error, carrying Agent stopped after max_turns=1.
Exit 3, the reply itself failed. The loop acts on none of it and leaves an assistant marked error carrying the provider's text.
Not an exit: an unknown tool gives an error result, and the loop goes round again. A result is news, not an ending.
The exit you must never build: a max_turns check between the reply and its tool calls, crossed out here, would leave a call with no result.
Exit 4, new today: cancel, checked at the top of a turn beside max_turns. The last message is an assistant marked aborted carrying Operation aborted, and no model call was made for it.

The tool that will not take a hint

Your own bash polls, because you wrote it. The next tool will not be yours. It will be a library call, or a colleague's, or a good tool written before anybody had a Stop button, and it will sit inside one await for ten minutes without ever looking at the signal you handed it. You saw one of those two rungs ago: two hundred turns of the event loop and the run had not moved.

There is a blunter instrument, and the trade calls it a hard cancel. From the warm-up: task.cancel() arranges for CancelledError to be raised inside whatever the job is awaiting. Point it at the task consuming your run and the exception surfaces inside the tool, in the await it is stuck in, whether or not that tool ever agreed to anything.

Which puts it in a place you have thought about before.

That exception climbs out of the tool and reaches run_tool, which is one try around one call. Should your boundary catch it? In a sentence or two: what kind of agent do you get if it does, and what does the same question mean for lesson 14's guard, which wraps the same call one layer out?

A tool result is a sentence written for the model. What would this one say, and what would the model do about it?

No. A cancelled tool did not fail: the program it runs in is being stopped, and that is news for whoever pressed the button, not for the model. Catch it and you get an agent that answers Stop with "the tool failed", takes that as a reason to think again, and carries on spending money. The guard has the same duty one layer out: it may catch what check throws, because a crashed check is a refusal, but it must not wrap the real tool in anything wider than Exception, or Stop arrives at the model as Tool call blocked and the run continues.

Here is what you said in lesson 4, when the question was still abstract:

You were asked whether that except should catch everything, and the answer was no: some things that come out of a tool are not that tool failing. lab.PowerCut was the stand-in. This is the real one.

Python takes the same side, as it happens: since version 3.8 CancelledError has not been an Exception, so except Exception lets it through without being asked. Your lab writes the clause out anyway, above the wide one, because the next person to edit that boundary will not know any of this.

Common answers, and what each one misses

"Catch it and return a result saying the run was cancelled, so the transcript stays tidy." Tidy, and the run does not end: the loop takes that result, goes round, and asks the model for another turn. Stop has to be something the loop cannot answer.
"Catch it, then re-raise after cleaning up." That is a finally, and it is the right instinct in the wrong clause. finally runs on every way out and changes nothing about what is caught; your bash already uses one to kill its process.
"It does not matter, since CancelledError is not an Exception anyway." True today, and the clause is a sentence to the next reader, which is the only reason it exists. Rules that hold by accident are the ones that get widened.

The same hung tool and the same task.cancel(), through three boundaries: run_tool catching Exception, run_tool catching BaseException, and a guard that catches BaseException around the real tool. Read what each hands back to the loop.

# run_tool, cut down to the one `try` that matters, and a tool that is stuck in an await.
import asyncio


async def ten_minute_build(arguments, signal=None):
    await asyncio.get_running_loop().create_future()          # waits for ever
    return "build finished"


async def run_tool(catching):
    try:
        return "toolResult: " + await ten_minute_build({})
    except catching as exc:
        return f"toolResult: the tool failed ({type(exc).__name__})  [is_error]"


async def guarded_run_tool(catching):
    """Lesson 14's guard around the same tool, widened the same way."""
    async def guarded():
        try:
            return await ten_minute_build({})
        except catching as exc:
            raise PermissionError(f"Tool call blocked: {type(exc).__name__}")
    try:
        return "toolResult: " + await guarded()
    except Exception as exc:
        return f"toolResult: {exc}  [is_error]"


async def press_stop(label, coroutine):
    task = asyncio.ensure_future(coroutine)
    await asyncio.sleep(0)                                    # let the tool reach its await
    task.cancel()
    try:
        print(f"{label:<34} {await task}")
    except asyncio.CancelledError:
        print(f"{label:<34} nothing. The cancellation went straight through.")


await press_stop("run_tool, except Exception", run_tool(Exception))
await press_stop("run_tool, except BaseException", run_tool(BaseException))
await press_stop("the guard, except BaseException", guarded_run_tool(BaseException))

run_tool, except Exception         nothing. The cancellation went straight through.
run_tool, except BaseException     toolResult: the tool failed (CancelledError)  [is_error]
the guard, except BaseException    toolResult: Tool call blocked: CancelledError  [is_error]

So the polite path and the blunt one are the same mechanism seen twice: ask, wait a little, and if nothing happens, insist. That is the whole shape of stop_run, and it is the last thing you write on this course before the capstone.

Figure 15.2 One Stop, and everything that has to look at it. A hollow mark is a look that found the flag clear; a filled one found it set.

Time runs down the page, in three lanes: the model, run_agent, and bash. Three looks at the flag find it clear. Then c1 starts, and while it runs the user presses Stop, which sets the flag. Nothing happens.
The tool notices first. bash is polling, so on its next lap it sees the flag, kills the command, and the call c1 gets the result Command cancelled.
The loop notices next. c2 and c3 are not run, and each gets Operation aborted, because every call still gets exactly one result. At the top of the following turn the flag is set, so no model call is made: an assistant message marked aborted is appended, and agent_end closes the run.
The backstop, for a tool that never looks. stop_run sets the flag, waits grace turns of the event loop, and then calls task.cancel(), which raises CancelledError inside the tool's await. run_tool lets it through, so c1 ends with no result at all, and stop_run returns "hard".

Two gaps, about fifteen lines.

run_tool, two lines: the explicit except asyncio.CancelledError: raise, above the wide clause, so that nobody ever widens the wide clause.
stop_run(harness, task, grace=5), returning "clean" or "hard". Ask first: harness.cancel(), then give the flag up to grace turns of the event loop to be noticed, stopping early if the task finishes. If it finished, that is "clean". If it did not, task.cancel(), then wait for the task, and that is "hard". Either way the task is over when stop_run returns, because a cancelled task that nobody waits for is a task nobody knows is over.

One subtlety, and the tests will find it. Waiting for the cancelled task raises that task's CancelledError in stop_run, and swallowing it is the whole point. But stop_run is itself running in somebody's task, and that somebody may be stopping it, in which case the CancelledError is not the task's, it is stop_run's own, and a function that eats its own cancellation cannot be stopped. asyncio.current_task().cancelling() is how the two are told apart: it counts the cancellations aimed at the task you are in, so anything above zero means this one is yours to pass on.

Your starter carries the reference bash from the last lab, whose poll loop kills the process however the loop is left. If you paste in your own and it kills the command only on the path where it saw the flag, one test will fail and say so: a hard cancel arrives inside the poll loop's sleep, by a way out that path never sees.

Two questions. In run_tool: the new clause returns nothing and builds no result, so what is the only statement in it? In stop_run: if the run stops politely on turn 3 of a grace of 5, what should the remaining two turns be spent on?
stop_run is five steps in a straight line, no nesting beyond one try. Cancel the harness. Loop up to grace times, leaving the loop early if task.done(), and spending each turn on await asyncio.sleep(0) so that the run gets its chance. Ask task.done() once more: if it is done, you are finished and it was clean. Otherwise cancel the task and wait for it, catching the CancelledError that the wait raises. Inside that catch, one question decides whether to swallow it or re-raise it, and asyncio.current_task().cancelling() answers it. Then say it was hard.
In outline. run_tool: above the wide clause, one naming asyncio.CancelledError, whose body is a single word. stop_run, in order: cancel the harness; a for over range(grace) that leaves early when the task is done and otherwise spends the turn on a zero sleep; then ask once whether the task is done and, if it is, say so and stop there. Past that point: cancel the task; a try whose body is one await of the task; an except for asyncio.CancelledError holding one if, on the count asyncio.current_task().cancelling() returns, whose body is a bare raise; and the other answer, last, outside the try. There is more than one correct shape for that end — asyncio.wait on the task raises nothing at all and needs no except. The reference keeps the exception in view because this lab is about that exception; when you have passed, write it the other way and see.

Fifteen lines, and your harness can now be stopped whether the tool it is waiting on co-operates or not. The hard path leaves a call with no result, which is the shape you predicted at the top of this page. Nothing was built to clean it up.

What changed since the previous lab

The line-by-line diff needs JavaScript. The whole file this exercise starts from is printed at the end of it.

bash is running `sleep inf`. stop_run(h, task, grace=5) returns "clean": the flag was enough, the task finished by itself and was not cancelled, and the record ends with the cancelled command's result and the aborted message.
A tool awaits `sleep inf` and never looks at its signal. stop_run returns "hard": the task was cancelled, CancelledError went up through run_tool without becoming a result, the run is over and is_running is False.
bash is running `sleep inf` and is given no grace at all: stop_run(h, task, grace=0) returns "hard", and the command has been killed all the same.
The same deaf tool behind lesson 14's guard: stop_run returns "hard", the task is cancelled, and the guard has not turned the cancellation into a refusal.
After a hard stop the record equals log.replay(), and two further prompts are answered: the killed call is shown to the model as interrupted, and nothing new was built for that.
A tool that sees the flag and then needs ten turns of the loop to tidy up: stop_run with grace=50 returns "clean" and the tool's result is recorded; with grace=2 it returns "hard". A run that is already over is "clean" at once.
While stop_run is waiting for a cancelled task that is slow to die, stop_run's own task is cancelled: the CancelledError is not swallowed, so whoever cancelled it sees it end as cancelled.
A job that includes a slow bash command is closed (aclose) after every k events. At every k the run is over, the log equals the record, two further prompts are answered, and no tool that ran is shown to the model as interrupted.

import asyncio import json from collections import deque # == 1 messages == # A conversation is a list of messages. Each message is a dict tagged with the role that spoke. def user_message(text): """A message from the user. Its content is a plain string.""" return {"role": "user", "content": text} def text_of(message): """The words of an assistant message. Its content is a list of blocks, not a string (lesson 2 explains why); the words are in the blocks of type "text".""" return "".join(block["text"] for block in message["content"] if block["type"] == "text") def tool_calls(message): """The tool calls an assistant message asks for, in the order it asked. Worked out from `content` on every call and stored nowhere else.""" return [block for block in message["content"] if block["type"] == "toolCall"] def error_message(text, stop_reason="error"): """An assistant message that says why a run stopped: nothing in `content`, the reason in `error_message`, and stop_reason "error", or "aborted" when it was the user who stopped it. A stop is written into the record, not raised.""" return {"role": "assistant", "content": [], "stop_reason": stop_reason, "error_message": text} ABORTED = "Operation aborted" # what a stopped run says, in a tool result and in its message def render(messages): """The transcript as plain text, one line per message, tool calls as name(arguments): what a summary is written from, and what estimate_tokens measures.""" lines = [] for m in messages: if m["role"] == "assistant": calls = [f"{c['name']}({json.dumps(c['arguments'])})" for c in tool_calls(m)] words = " ".join([text_of(m)] + calls) else: words = m["content"] lines.append(f"{m['role']}: {words}".replace("\n", " ")) return "\n".join(lines) # == 2 tools == # A tool is a dict. "name", "description" and "parameters" are advertised to the model. # "execute" is an async function, execute(arguments, signal=None) -> str, that only this program # ever awaits. # "snippet" and "guidelines" are optional and are for the system prompt: one line on what the # tool is for, and advice on when to use it. A schema can say what; only words can say when. def str_arg(arguments, name): """The string argument `name`, checked. The schema asked the model for a string and nobody made it send one, so every tool reads its arguments through here. Raises ValueError("<name> must be a string") when it is missing or not a string.""" value = arguments.get(name) if not isinstance(value, str): raise ValueError(f"{name} must be a string") return value def int_arg(arguments, name, default): """The optional whole-number argument `name`, checked; `default` if the model left it out. Raises ValueError("<name> must be an integer") for anything else.""" value = arguments.get(name) if value is None: return default if not isinstance(value, int): raise ValueError(f"{name} must be an integer") return value def tool_specs(tools): """What the model is told about the tools: everything except the function.""" return [{"name": tool["name"], "description": tool["description"], "parameters": tool["parameters"]} for tool in tools] async def run_tool(tools, call, signal=None): """Run one tool call and return the toolResult message that answers it. One call in, exactly one result out, on every path. Tools just raise; this is the one place that turns a failure into a result the model can read. is_error means the tool could not do its job. It never raises Exception. After Stop (a cancelled `signal`) the tool is not run, and the call still gets its one result: ABORTED. A tool that is run is handed the signal. Task cancellation is no tool failure: CancelledError goes up through here untouched.""" tools_by_name = {tool["name"]: tool for tool in tools} tool = tools_by_name.get(call["name"]) if signal is not None and signal.is_cancelled(): content, is_error = ABORTED, True elif tool is None: content, is_error = f"Tool {call['name']} not found", True else: # --- your code --- # task.cancel() raises asyncio.CancelledError inside whatever the tool is awaiting. It is # not a tool failure and must not become a result: say so with a clause of its own that # re-raises it. (Since Python 3.8 it is not an Exception, so the clause below would let # it through anyway; the explicit clause is there for whoever next edits this boundary.) try: content, is_error = await tool["execute"](call["arguments"], signal), False except Exception as exc: content, is_error = str(exc), True # --- end --- return {"role": "toolResult", "tool_call_id": call["id"], "tool_name": call["name"], "content": content, "is_error": is_error} MAX_LINES = 2000 # The budget of one tool result. Whatever a tool returns is re-read, and MAX_BYTES = 50_000 # paid for, on every later turn. Bytes, not characters: "é" is two. def truncate_head(text, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The start of `text` that fits the budget, for files: (kept, lines_shown). Whole lines only, newlines included, sized in UTF-8 bytes. If even the first line is too big, the answer is ("", 0): half a line is worse than none.""" kept, size = [], 0 for line in text.splitlines(keepends=True): size += len(line.encode()) if len(kept) == max_lines or size > max_bytes: break kept.append(line) return "".join(kept), len(kept) def truncate_tail(text, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The end of `text` that fits the budget, for command output, where the verdict comes last: (kept, lines_shown). Whole lines only, newlines included, sized in UTF-8 bytes.""" kept, size = [], 0 for line in reversed(text.splitlines(keepends=True)): size += len(line.encode()) if len(kept) == max_lines or size > max_bytes: break kept.append(line) return "".join(reversed(kept)), len(kept) def make_read_tool(ws, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The `read` tool, over the workspace `ws`. One call returns at most the budget, and says so in a last line that tells the model how to ask for the rest.""" async def execute(arguments, signal=None): path = str_arg(arguments, "path") offset, limit = int_arg(arguments, "offset", 1), int_arg(arguments, "limit", None) if offset < 1 or (limit is not None and limit < 1): raise ValueError("offset and limit must be 1 or more") if not ws.exists(path): raise FileNotFoundError( f"File not found: {path}. Files here: {', '.join(ws.listdir('.'))}") lines = ws.read_text(path).splitlines(keepends=True) wanted = lines[offset - 1:] if limit is None else lines[offset - 1:offset - 1 + limit] kept, shown = truncate_head("".join(wanted), max_lines, max_bytes) if offset > max(len(lines), 1): raise ValueError(f"offset {offset} is past the end of {path}: {len(lines)} lines") if wanted and shown == 0: skip = f" Use offset={offset + 1} to skip it." if offset < len(lines) else "" return (f"[Line {offset} of {path} is {len(wanted[0].encode())} bytes, over the " f"{max_bytes}-byte budget of one read, so none of it is shown.{skip}]") last = offset + shown - 1 if last < len(lines): kept += (f"[Showing lines {offset}-{last} of {len(lines)}. " f"Use offset={last + 1} to continue.]") return kept return { "name": "read", "description": "Read a text file and return its contents. One call returns at most " f"{max_lines} lines or {max_bytes} bytes; use offset to read on.", "parameters": { "type": "object", "properties": {"path": {"type": "string", "description": "Path of the file to read."}, "offset": {"type": "integer", "description": "First line, from 1."}, "limit": {"type": "integer", "description": "How many lines."}}, "required": ["path"], }, "execute": execute, "snippet": "Read file contents", "guidelines": ["Use read to examine files instead of cat or sed.", "Show file paths clearly when working with files."], } def make_write_tool(ws): """The `write` tool, over the workspace `ws`: create a file, or replace all of it.""" async def execute(arguments, signal=None): path, content = str_arg(arguments, "path"), str_arg(arguments, "content") ws.write_text(path, content) return f"Successfully wrote to {path}." return { "name": "write", "description": "Write a text file. Creates the file, or replaces its whole contents.", "parameters": { "type": "object", "properties": {"path": {"type": "string", "description": "Path of the file to write."}, "content": {"type": "string", "description": "The new contents."}}, "required": ["path", "content"], }, "execute": execute, "snippet": "Create or overwrite files", "guidelines": ["Use write only for new files or complete rewrites.", "Show file paths clearly when working with files."], } def make_bash_tool(shell, max_lines=MAX_LINES, max_bytes=MAX_BYTES): """The `bash` tool: run one command line in `shell` and return what it printed. A command that fails has still been run: its exit code is news for the model, not a tool error, so it comes back as an ordinary result that ends with a status line.""" async def execute(arguments, signal=None): process = shell.start(str_arg(arguments, "command")) try: while not process.done(): # slow code looks at the flag on every lap if signal is not None and signal.is_cancelled(): return "Command cancelled" await asyncio.sleep(0) # give the floor away: Stop's code runs here finally: process.kill() # however we leave, nothing is left running output, code = process.output() total = len(output.splitlines()) output, shown = truncate_tail(output, max_lines, max_bytes) if shown < total: output += f"\n[Showing the last {shown} of {total} lines. Narrow the command for more.]" if code != 0: output += f"\nCommand exited with code {code}" return output return { "name": "bash", "description": "Run a shell command and return its output (stdout and stderr together). " f"Only the last {max_lines} lines or {max_bytes} bytes come back.", "parameters": { "type": "object", "properties": {"command": {"type": "string", "description": "The command line."}}, "required": ["command"], }, "execute": execute, "snippet": "Run shell commands", "guidelines": ["Say what a command is for before you run it."], } def guard(tool, check): """`tool` behind a door: the same name, schema and everything else, so neither the model nor the loop can tell, with an execute that first asks check(name, arguments): None, or the reason to refuse. A refusal is raised, so run_tool makes it the call's one result, is_error and all. A check that raises has not said yes: the call is blocked (fail closed).""" async def execute(arguments, signal=None): try: reason = check(tool["name"], arguments) except Exception as exc: # a check that crashed has not said yes reason = f"the check failed: {exc}" if reason is not None: raise PermissionError(f"Tool call blocked: {reason}") return await tool["execute"](arguments, signal) return {**tool, "execute": execute} def deny_destructive(name, arguments): """A check for guard: refuses a bash command that contains "rm -rf". A deny-list is a speed bump, not a sandbox: `rm -fr build` walks straight past it.""" if name == "bash" and "rm -rf" in str(arguments.get("command")): return "rm -rf deletes files for good. Ask the user to run it themselves if they want it." return None def confirm_with(human): """A check for guard that asks a person. human.ask(question) answers True for yes; anything else is a no, and the model is told so.""" def check(name, arguments): if human.ask(f"Allow {name} {json.dumps(arguments)}?"): return None return "the user said no" return check # == 3 loop == # The code that talks to the model. The model keeps nothing between calls. async def run_agent(model, system, messages, tools, prompt, *, max_turns=None, get_steering=None, get_follow_ups=None, signal=None): """Put `prompt` to the model and keep going until it stops asking for tools, a reply comes back failed (stop_reason "error"), `max_turns` model calls have been made (None: no limit), or `signal` (a CancelToken, or None) is cancelled: the run then ends with an "aborted" message. An async generator of events: dicts with a "type", which say what happened and draw nothing. They come in balanced pairs on every exit: agent_start ... agent_end around the run, and turn_start ... turn_end around each turn. The one ordering rule is mutate, then announce: a message is appended to `messages` (the caller's list, the record) first, and only then is its message_end yielded. The final message is the last of agent_end's "messages". The user may speak during a run; the loop asks for their words only where a user message keeps the transcript valid. `get_steering` and `get_follow_ups` take no arguments and return a list of user messages, often empty: steering is asked for once a turn's whole tool batch is answered, follow-ups only when the run would otherwise end. The loop never sees a queue.""" new = [] # this run's messages, for agent_end def record(message): """Append `message` to the record, then make the event that announces it.""" messages.append(message) new.append(message) return {"type": "message_end", "message": message} def ask(get): """The user messages a callable has for us: [] if it has none, or was never passed.""" return list(get()) if get else [] yield {"type": "agent_start"} turn, calls, failed = 0, [], False # User messages waiting to open the next turn. Steering sent while idle goes in with the prompt. pending = [user_message(prompt)] + ask(get_steering) while pending: # outer loop: the prompt, then each follow-up while calls or pending: # inner loop: one pass is one turn turn += 1 yield {"type": "turn_start"} for message in pending: # only here, at the top of a turn, may a user speak yield record(message) pending = [] if max_turns is not None and turn > max_turns: # checked before the call, never after reply = error_message(f"Agent stopped after max_turns={max_turns}") elif signal is not None and signal.is_cancelled(): # no paid request follows a Stop reply = error_message(ABORTED, stop_reason="aborted") else: reply = await model.acomplete(system, context_for_model(messages), tool_specs(tools), signal) yield record(reply) # A failed or stopped reply: none of it is acted on and nobody is asked for more. failed = reply["stop_reason"] in ("error", "aborted") calls = [] if failed else tool_calls(reply) for call in calls: yield {"type": "tool_execution_start", "call": call} result = await run_tool(tools, call, signal) yield record(result) # no yield between the tool returning and this yield {"type": "tool_execution_end", "call": call, "result": result} yield {"type": "turn_end", "message": reply} if not failed: pending = ask(get_steering) # every call of the turn has its result if not failed: pending = ask(get_follow_ups) # no calls, no steering: the run would end here yield {"type": "agent_end", "messages": new} # the one way out, so every exit is balanced def context_for_model(messages): """What the model is sent. The record is not the view: the record keeps what happened, and this builds, for every request, a transcript a provider accepts. Assistant messages that failed or were stopped and are empty are left out (one that got something out stays in), then the calls and results are paired up. Returns a new list; the record is never changed.""" kept = [m for m in messages if not (m["role"] == "assistant" and not m["content"] and m.get("stop_reason") in ("error", "aborted"))] return repair_tool_history(kept) INTERRUPTED = ("Tool call interrupted: no result was recorded. It may not have run, or may have " "run partly; check before repeating it.") def repair_tool_history(messages): """A copy of `messages` in which every tool call is followed at once by exactly one result, in call order. A recorded result is kept and moved to its place; of two for one call the first wins; a result whose call is nowhere is dropped (nobody knows the arguments of the call, so it cannot be invented); a call with no result gets INTERRUPTED, marked is_error. Pure and idempotent: a new list of the same message objects, equal to a valid input. Call ids are taken to be unique.""" recorded = {} # call id -> the first result recorded for it for m in messages: if m["role"] == "toolResult": recorded.setdefault(m["tool_call_id"], m) repaired = [] for m in messages: if m["role"] == "toolResult": continue # results are put back below, after their call repaired.append(m) if m["role"] == "assistant": for call in tool_calls(m): repaired.append(recorded.get(call["id"]) or { "role": "toolResult", "tool_call_id": call["id"], "tool_name": call["name"], "content": INTERRUPTED, "is_error": True}) return repaired # == 4 harness == # One transcript, one writer. The loop keeps the mechanics and stays stateless; this holds the list. class CancelToken: """The Stop flag of ONE run. cancel() only sets it; nothing stops until slow code looks: the loop at the top of each turn, run_tool before each call, the model, a polling tool.""" def __init__(self): self._cancelled = False def cancel(self): self._cancelled = True def is_cancelled(self): return self._cancelled class Harness: """Owns the one transcript, and the settings that stay the same from run to run, and lets one run at a time write to it. Harness(model, system, tools, messages=(), max_turns=None) copies `messages` into a list of its own. .messages a tuple, a snapshot of the transcript: nobody outside can append to it or pop from it. (The dicts inside are not copied.) .is_running True from the call of prompt() until that run is exhausted, closed or dies. .prompt(text) returns the run: run_agent's events, over the harness's own list. While a run is going on it raises RuntimeError("already running; use steer() or follow_up()"), and it raises on the caller's line, so prompt is a plain def and the async generator is _run. .steer(text), .follow_up(text) queue a user message: steer for the next gap between turns ("use spaces"), follow_up for when the run would otherwise end ("now run the tests"). Callable at any time, by anyone. A queued message is not history until the loop asks for it, one message per ask. Queues outlive runs. .subscribe(listener) listener(event) hears every event of every run, after the record has changed and before the run's consumer is handed the event. Returns the function that undoes it. A listener that raises ends the run: listeners are trusted code, and a saver that cannot save must not be silent. .replace_messages(messages) swaps in a whole new transcript, for whoever rebuilt it from the log. The one door in, and shut during a run: RuntimeError. .cancel() Stop: cancels the current run's token, from the moment prompt() returns. Every run gets a fresh token, so when idle this does nothing.""" def __init__(self, model, system, tools, messages=(), max_turns=None): self._model, self._system, self._tools = model, system, list(tools) self._messages, self._max_turns = list(messages), max_turns self._running = False self._steering, self._follow_ups = deque(), deque() self._listeners = [] self._signal = CancelToken() @property def messages(self): return tuple(self._messages) @property def is_running(self): return self._running def prompt(self, text): if self._running: # checked now, not at the first next() raise RuntimeError("already running; use steer() or follow_up()") self._running = True self._signal = CancelToken() # fresh: yesterday's Stop must not stop this run return self._run(text) def steer(self, text): self._steering.append(user_message(text)) def follow_up(self, text): self._follow_ups.append(user_message(text)) def subscribe(self, listener): self._listeners.append(listener) def unsubscribe(): if listener in self._listeners: # undoing twice is harmless self._listeners.remove(listener) return unsubscribe def _notify(self, event): for listener in list(self._listeners): # a copy: a listener may unsubscribe while called listener(event) def replace_messages(self, messages): if self._running: raise RuntimeError("cannot replace messages during a run") self._messages = list(messages) def cancel(self): self._signal.cancel() # when idle: the token of a run that is over def _drain(self, queue): """What the loop is handed when it asks: the oldest queued message, one at a time.""" return [queue.popleft()] if queue else [] async def _run(self, text): run = run_agent(self._model, self._system, self._messages, self._tools, text, max_turns=self._max_turns, signal=self._signal, get_steering=lambda: self._drain(self._steering), get_follow_ups=lambda: self._drain(self._follow_ups)) try: async for event in run: self._notify(event) # notify, then yield: whoever stops listening to yield event # the run, the listeners have already heard finally: # exhausted, closed by the consumer, or dead await run.aclose() self._running = False # == 5 frontends == # A frontend is a fold over the events of one run: render(event) for each event, then finish(), # which says whether the run ended well. The loop never learns what its events look like. class JsonRenderer: """Every event as one line of JSON, for another program to read.""" def __init__(self): self.failed = False def render(self, event): print(json.dumps(event)) if (event["type"] == "message_end" and event["message"].get("stop_reason") in ("error", "aborted")): self.failed = True def finish(self): return not self.failed class FinalTextRenderer: """Only the answer, for `> answer.txt`: nothing is printed while the run goes on. finish() prints the text of the last assistant message and returns True; if that message failed or was stopped (stop_reason "error" or "aborted") it prints "Error: <its error_message>" instead and returns False.""" def __init__(self): self.last = None # the newest assistant message seen so far def render(self, event): if event["type"] == "message_end" and event["message"]["role"] == "assistant": self.last = event["message"] def finish(self): if self.last is None: return False if self.last["stop_reason"] in ("error", "aborted"): print(f"Error: {self.last['error_message']}") return False print(text_of(self.last)) return True async def stop_run(harness, task, grace=5): """Stop the run that `task` is consuming, politely and then not. harness.cancel() sets the flag, and code that looks at it gets `grace` turns of the event loop to finish: "clean". Code that never looks is cancelled as a task, which raises CancelledError inside whatever it is awaiting, and is waited for: "hard". Either way the task is over when this returns. Only that task's CancelledError is swallowed: if stop_run itself is being cancelled, it goes.""" # --- your code --- # About 12 lines. # 1. harness.cancel() # 2. up to `grace` times: if task.done(), stop waiting; else `await asyncio.sleep(0)` # 3. if the task is done, return "clean" # 4. task.cancel(), then `await task` inside try / except asyncio.CancelledError # (a cancelled task that nobody awaits is a task nobody knows is over), return "hard" # 5. in that except: if asyncio.current_task().cancelling() is not 0, the CancelledError # is stop_run's own (somebody is cancelling IT): re-raise # One of the hidden tests, with a tool that never looks at its signal: # async def execute(arguments, signal=None): # return (await shell.arun("sleep inf"))[0] # task = asyncio.ensure_future(consume(h.prompt("Build it."))) # the run, in a task # ... # assert await stop_run(h, task, grace=5) == "hard" # assert task.cancelled() and not h.is_running # assert lab.shape(h.messages) == "U A[c1]" # a killed call has no result... # (then two more prompts, and both are answered: ...because of lesson 10) # --- end --- # == 6 environment == # What a session needs around the harness, starting with a disk. Regions 3 and 4 know none of it. class SessionLog: """A session on disk: the file `path` in the workspace `ws`, one JSON object per line, one line per message: {"id": "e1", "type": "message", "message": {...}}. Lines are appended and never edited. The log is the truth and the transcript is what replay() makes of it, so resuming is Harness(..., messages=log.replay()). Holds no state but `ws` and `path`.""" def __init__(self, ws, path): self.ws, self.path = ws, path def entries(self): """Every entry, oldest first; [] if there is no file yet. A line that is not JSON (a torn write) raises ValueError naming its line number. It is never skipped: the next append would be glued onto it, and both entries lost.""" text = self.ws.read_text(self.path) if self.ws.exists(self.path) else "" entries = [] for number, line in enumerate(text.splitlines(), 1): try: entries.append(json.loads(line)) except ValueError: raise ValueError(f"{self.path}, line {number}: not a complete log entry") from None return entries def append_message(self, message): """Append one message as one line and return its entry id, "e" + its line number. The number is counted by re-reading the file, so two logs over one file cannot disagree.""" entry = {"id": f"e{len(self.entries()) + 1}", "type": "message", "message": message} self.ws.append_text(self.path, json.dumps(entry) + "\n") return entry["id"] def append_compaction(self, summary, first_kept_id): """Append a compaction entry and return its id. From now on the entries before `first_kept_id` are read as `summary`. Nothing is deleted.""" entry = {"id": f"e{len(self.entries()) + 1}", "type": "compaction", "summary": summary, "first_kept_id": first_kept_id} self.ws.append_text(self.path, json.dumps(entry) + "\n") return entry["id"] def rows(self): """(entry id, message) for every message of the transcript, in order. A compaction entry is a rule for reading what came before it: the rows before its first_kept_id become one row, (its own id, the summary as a user message), and the rows from there on stay.""" rows = [] for entry in self.entries(): if entry["type"] == "message": rows.append((entry["id"], entry["message"])) elif entry["type"] == "compaction": ids, first = [row_id for row_id, _ in rows], entry["first_kept_id"] kept = rows[ids.index(first):] if first in ids else [] summary = user_message("Previous conversation summary:\n" + entry["summary"]) rows = [(entry["id"], summary)] + kept return rows def replay(self): """The transcript: the state is computed from what happened, never saved.""" return [message for _, message in self.rows()] def persist_to(log): """A listener for Harness.subscribe that appends every completed message to `log`, as it completes: saving is nobody's job but this one's, whoever consumes the run.""" def listener(event): if event["type"] == "message_end": log.append_message(event["message"]) return listener def estimate_tokens(messages): """About how many tokens `messages` cost: four characters to a token. No tokenizer, the same answer every time, and it undercounts: leave room.""" return len(render(messages)) // 4 SUMMARY_SYSTEM = ("You summarise a conversation between a user and a coding agent. Do NOT continue " "the conversation and do NOT answer any question in it. Reply with the summary " "only: the goal, what is done, what comes next, exact file paths and errors.") async def summarize(model, messages): """A summary of `messages`, from one model call that is no part of any transcript: its own system prompt, one user message holding the rendered conversation in <conversation> tags, and no tools, so there is nothing to do but summarise. Raises RuntimeError if the reply failed or is empty.""" wrapped = user_message(f"<conversation>\n{render(messages)}\n</conversation>") reply = await model.acomplete(SUMMARY_SYSTEM, [wrapped], []) summary = text_of(reply).strip() if reply["stop_reason"] == "error" or not summary: raise RuntimeError(f"no summary: {reply.get('error_message', 'the reply was empty')}") return summary def find_cut(rows, keep_recent_tokens): """Where to cut `rows` (SessionLog.rows()) for a compaction: the index of the first row to keep. Rows before it are summarised. 0 means all of it fits in `keep_recent_tokens`. The cut is at a turn boundary, a user message, and failing that at least never on a tool result: a kept tail that opens on a result whose call was summarised away is a 400.""" cut, kept = len(rows), 0 while cut > 0 and kept < keep_recent_tokens: # walk back from the newest row cut -= 1 kept += estimate_tokens([rows[cut][1]]) if kept < keep_recent_tokens: return 0 for index in range(cut, len(rows)): # snap forward to the next user message if rows[index][1]["role"] == "user": return index while cut < len(rows) and rows[cut][1]["role"] == "toolResult": cut += 1 # no user message left: step past the results return cut async def compact(log, model, *, keep_recent_tokens): """Compact the session in `log`: summarise the old rows and append ONE compaction entry. The log only grows; replay() now gives [summary] + [kept tail]. False, with nothing appended, if there is nothing to summarise or nothing would be kept, or if the summary failed: housekeeping never costs the user their turn.""" rows = log.rows() cut = find_cut(rows, keep_recent_tokens) if cut <= 0 or cut >= len(rows): return False try: summary = await summarize(model, [message for _, message in rows[:cut]]) except RuntimeError: return False log.append_compaction(summary, rows[cut][0]) return True async def maybe_compact(harness, log, model, *, window, reserve, keep_recent_tokens): """Compact if the transcript has come within `reserve` tokens of the context `window`, then give the harness what the log now replays. For the code around the harness to call before each prompt, never during a run. True if it compacted.""" if harness.is_running: raise RuntimeError("cannot compact during a run") if estimate_tokens(harness.messages) <= window - reserve: return False if not await compact(log, model, keep_recent_tokens=keep_recent_tokens): return False harness.replace_messages(log.replay()) return True def discover_context(ws, cwd): """The project's standing instructions: every AGENTS.md from the top of the workspace `ws` down to the directory `cwd` (a plain path such as "services/api", no ".."), as [(path, text)]. Broad to specific, each path once: the file nearest the work comes last and has the last word. Whoever wrote these files is writing your system prompt.""" folders = [""] for part in cwd.split("/"): if part not in ("", "."): folders.append(folders[-1] + part + "/") return [(folder + "AGENTS.md", ws.read_text(folder + "AGENTS.md")) for folder in folders if ws.exists(folder + "AGENTS.md")] def build_system_prompt(tools, context_files, skills, cwd, today): """The system prompt, computed from what is really there: the enabled `tools` (a line for each "snippet", then every "guidelines" entry once, first seen first), `context_files` [(path, text)] wrapped with their paths in the order given, and an index of `skills` [{"name", "description", "path"}]. A skill's body stays on disk and `read` is its loader, so without a tool named read there is no index. The date and the directory come last. It reads no clock and no disk: the same arguments give the same bytes, which tests and prefix caches rely on.""" guidelines = [] for tool in tools: for guideline in tool.get("guidelines", []): if guideline not in guidelines: # a list, not a set(): a set has no order to keep guidelines.append(guideline) tool_lines = [f"- {tool['name']}: {tool['snippet']}" for tool in tools if tool.get("snippet")] sections = ["You are a coding assistant working in the user's project."] if tool_lines: sections.append("Available tools:\n" + "\n".join(tool_lines)) if guidelines: sections.append("Guidelines:\n" + "\n".join(f"- {guideline}" for guideline in guidelines)) for path, text in context_files: sections.append(f'<project_instructions path="{path}">\n{text}\n</project_instructions>') if skills and any(tool["name"] == "read" for tool in tools): index = [f"- {skill['name']}: {skill['description']} ({skill['path']})" for skill in skills] sections.append("Skills are files of instructions for particular tasks. When a task " "matches a description, read the file first.\n" + "\n".join(index)) sections.append(f"Current date: {today}\nCurrent working directory: {cwd}") return "\n\n".join(sections)Running code needs JavaScript. Above is the file this exercise starts from (harness.py). You can run its tests on your own machine: python lab/lab_check.py --harness harness.py --lesson 15 tests.py

Two ten-minute builds on one harness, each stopped in its third minute: first a bash command that watches its signal, then a tool that never looks. A session log is subscribed throughout. After the hard stop, two more prompts are sent. No tests: read what comes back, and read the last line. Once lab 2 has passed, this runs on your code.

import asyncio, harness, lab

SYSTEM = "You are a careful assistant."
ws = lab.Workspace({})
shell = lab.Shell(ws)


async def consume(run):                     # a frontend, with the rendering left out
    async for event in run:
        pass


def deaf_build():
    """A build tool that waits for the command and never looks at its signal."""
    async def execute(arguments, signal=None):
        return (await shell.arun("sleep 600"))[0]
    return {"name": "build", "description": "Build the project. Takes ten minutes.",
            "parameters": {"type": "object", "properties": {}}, "execute": execute}


def job(request):
    newest = [m for m in request.messages if m["role"] == "user"][-1]["content"]
    if newest.startswith("Run the tests"):
        return lab.reply(lab.call("bash", {"command": "sleep 600"}))
    if newest.startswith("Build"):
        return lab.reply(lab.call("build"))
    return lab.say(f"Answer {sum(m['role'] == 'user' for m in request.messages)}.")


model = lab.ScriptedModel([lab.forever(job)])
tools = [harness.make_bash_tool(shell), deaf_build()]
h = harness.Harness(model, SYSTEM, tools)
log = harness.SessionLog(ws, "session.jsonl")
h.subscribe(harness.persist_to(log))


async def stop_it(prompt):
    task = asyncio.ensure_future(consume(h.prompt(prompt)))
    for _ in range(20):                     # the build gets going
        await asyncio.sleep(0)
    return await harness.stop_run(h, task, grace=5)


print("Run 1. The ten-minute command is `bash`, which looks at its signal.")
print("  stop_run said:", await stop_it("Run the tests."))
print("  record:", lab.shape(list(h.messages)))
print("  the command's result ends:", repr(h.messages[2]["content"].splitlines()[-1]))
print()
print("Run 2. The ten-minute command is a tool that never looks.")
before = len(h.messages)
print("  stop_run said:", await stop_it("Build it."))
print("  new on the record:", lab.shape(list(h.messages)[before:]), " is_running:", h.is_running)
print()
print("Two more prompts on the same harness:")
for text in ("What happened?", "Thank you."):
    await lab.drive(h.prompt(text))
    print(f"  {text:<16} -> {harness.text_of(h.messages[-1])}")
print()
sent = model.calls[-1].messages
for message in sent:
    if message.get("tool_name") == "build":
        print("  what the model was shown for the killed call:")
        print("   ", message["content"])
print()
print("log lines:", len(log.entries()), " messages on the record:", len(h.messages),
      " they agree:", log.replay() == list(h.messages))

Run 1. The ten-minute command is `bash`, which looks at its signal.
  stop_run said: clean
  record: U A[c1] R(c1) A(aborted)
  the command's result ends: 'Command cancelled'

Run 2. The ten-minute command is a tool that never looks.
  stop_run said: hard
  new on the record: U A[c2]  is_running: False

Two more prompts on the same harness:
  What happened?   -> Answer 3.
  Thank you.       -> Answer 4.

  what the model was shown for the killed call:
    Tool call interrupted: no result was recorded. It may not have run, or may have run partly; check before repeating it.

log lines: 10  messages on the record: 10  they agree: True

clean, then hard, and then the session simply carries on. The killed call has no result on the record and never will: the record says what happened, and what happened is that nobody knows whether the build ran. The next request shows the model Tool call interrupted in its place, because context_for_model repairs the view before every request, and it has done that since lesson 10. You wrote nothing today to make that work.

Hold a conversation: it knows what was said, and who said it.
Run a tool the model asks for and show it the result.
Keep going until the model stops asking.
Tell the model when a tool fails, and carry on.
Stop a runaway, and survive a provider failure.
Keep every tool result within a budget.
Report what it is doing, as events, to any frontend.
Own its transcript: one writer at a time.
Take your input mid-run, at a safe point.
Send a valid transcript even after an interruption.
Survive a power cut: an append-only log, resumed by replay.
Outlast the context window: summarise the old, keep the recent word for word, delete nothing.
Brief the model from what is really there: the enabled tools, the project's files, an index of skills.
Refuse a tool call at the door, in your code, and hand the model the reason.
Be stopped mid-build, politely or not, and be usable a second later.

A colleague is adding a Stop button to their own agent and says: "Easy. A boolean on the harness, and cancel() sets it." In a sentence or two, what would you tell them?

Their harness is three minutes into a ten-minute build. What has to happen next, and who has to have written it?

The boolean is right and it is about a fifth of the work. Setting it stops nothing: every slow piece of the run has to look at it, by hand, and anything that never looks needs an interruption raised inside the await it is stuck in. And the button is only reachable at all because the run gives the floor away while it waits.

Common answers, and what each one misses

"Just raise an exception from cancel()." cancel() runs on the button's thread of control, not the run's. The exception would end the button handler and leave the build going.
"Check the flag at the top of the loop." Necessary, and it is the check that saves the money. It is also the check that a ten-minute tool never reaches.
"Kill the process." An honest answer, and it throws away the transcript, the log's last lines, and any chance of the user typing again. Everything on this page exists to make Stop cheaper than that.

await is how one thread waits politely. A cancel flag stops nothing until slow code looks at it.

Tau cancels the same way, in the same two layers: a token every slow piece checks, and a CancelledError that the tool boundary is told, in writing, to let through.

if signal is not None and signal.is_cancelled():
    content, is_error = ABORTED, True
elif tool is None:
    content, is_error = f"Tool {call['name']} not found", True
else:
    try:
        content, is_error = await tool["execute"](call["arguments"], signal), False
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        content, is_error = str(exc), True

if blocked:
    result = _error_result(block_reason or "Tool execution was blocked")
    is_error = True
elif signal is not None and signal.is_cancelled():
    result = _error_result("Operation aborted")
    is_error = True
# ...
try:
    result = await tool.execute(call.id, call.arguments, signal, on_update)
    return result, False, updates
except asyncio.CancelledError:
    raise
except Exception as exc:  # noqa: BLE001 - tools are an isolation boundary
    return _error_result(str(exc)), True, updates

The same parts, piece by piece. This is the lesson where the standing footnote retires: Tau has been async all along, and from here you read it as it is written.

The token is a protocol with one method, so any object with is_cancelled() will do (src/tau_agent/provider.py:13-16). The implementation Tau ships is your CancelToken, line for line (src/tau_agent/harness.py:51-59).
One fresh token per run (src/tau_agent/harness.py:165-166), and cancel() sets whichever is current (src/tau_agent/harness.py:117-119).
A reply labelled error or aborted ends the run, in exactly those two words (src/tau_agent/loop.py:148), and an empty one of either kind is left out of the next request while staying on the record (src/tau_agent/loop.py:185-201). That function is context_for_model under another name.
The fake provider honours the token by ending its stream (src/tau_ai/fake.py:35-39), and so does the wait between retries, so a backoff of thirty seconds does not swallow a Stop (src/tau_ai/retry.py:46-62).
Both layers at once, in Tau's own test: harness.cancel() on one line and task.cancel() on the next (tests/test_agent_harness.py:260-265), which is stop_run with the waiting left out. The terminal does the same when you press Escape: cancel the session, then cancel the worker (src/tau_coding/tui/app.py:6172-6177).

Where the two differ, and why. Three places, and the first is ours to own.

Nothing in Tau produces aborted today. The value is in the vocabulary (src/tau_agent/messages.py:154) and seven files read it, but no provider or loop in src/ ever sets it, so in Tau a Stop surfaces as an error terminal or as no terminal at all. Ours sets it, because a run the user stopped and a run the provider failed are different news for whoever reads the log tomorrow.
Tau's loop does not check the token between turns. Its turn top has the max_turns check and nothing beside it (src/tau_agent/loop.py:112-113); the check before each tool call is the only one, and stopping the model call itself is left to the provider. Ours checks once per turn, which is the check that keeps a request from being paid for after Stop.
Tau clears the current token only if it is still the same one (src/tau_agent/harness.py:203-204), against a later run having replaced it while an older finally was pending. Your run guard makes that impossible, so the line would be dead code and the lab does not ask for it.

A gap in Tau, found by writing this lesson. Tau's prompt_message is a plain def that marks the run as running and hands back the generator (src/tau_agent/harness.py:147-150), and the token is made on the first line of that generator, which does not execute until the consumer asks for the first event. Meanwhile cancel() does nothing when there is no token. The question below is about Tau's code, so nothing waits on it.

A frontend does what frontends do: task = ensure_future(consume(h.prompt(text))), and the user hits Stop before that task has taken a single step. With Tau's ordering, what happens?

Nothing. cancel() finds no token, and the run then starts with a fresh one that nobody has cancelled

The Stop is dropped on the floor. Yours makes the token in prompt(), on the line where is_running turns True, for exactly this reason.

The run is stopped: the flag was set before the run began, so the first check reads it

There was no flag to set. Which object would cancel() have written to?

The run had already begun, because h.prompt(text) started it

Lesson 7. Calling a generator function runs none of it; the body waits for the first request for a value.

Your harness records the prompt and one aborted message, and calls the model zero times. Tau's ordering calls the model, gets an answer, and hands the user the answer to the question they cancelled. Both classes are in the cell, and the only difference between them is which line makes the token.

import harness, lab

SYSTEM = "You are a careful assistant."


class TauOrder(harness.Harness):
    """Your harness with Tau's ordering: prompt() only marks the run as running, and the token
    is made on the first line of the run generator, which does not execute until the consumer
    asks for the first event. cancel() looks for a token and does nothing if there is none."""

    def prompt(self, text):
        run = super().prompt(text)
        self._signal = None                     # no token yet (tau_agent/harness.py:117-119)

        async def token_on_the_first_step():
            self._signal = harness.CancelToken()     # (tau_agent/harness.py:165-166)
            async for event in run:
                yield event

        return token_on_the_first_step()

    def cancel(self):
        if self._signal is not None:
            self._signal.cancel()


async def stop_before_the_first_step(label, made):
    model = lab.ScriptedModel([lab.say("Deleted 4,000 files. You are welcome.")])
    h = made(model, SYSTEM, [])
    run = h.prompt("Tidy the repo.")            # a frontend hands this to a task...
    h.cancel()                                  # ...and the user hits Stop before it runs
    await lab.drive(run)
    print(f"{label}:")
    print("  record:", lab.shape(list(h.messages)), " model calls:", len(model.calls))
    print("  last message:", repr(harness.text_of(h.messages[-1])
                                  or h.messages[-1].get("error_message")))


await stop_before_the_first_step("yours: the token is made in prompt()", harness.Harness)
print()
await stop_before_the_first_step("Tau's order: the token is made at the first step", TauOrder)

yours: the token is made in prompt():
  record: U A(aborted)  model calls: 0
  last message: 'Operation aborted'

Tau's order: the token is made at the first step:
  record: U A  model calls: 1
  last message: 'Deleted 4,000 files. You are welcome.'

What Tau adds. Tools get a token of their own, a second one-method protocol with the same shape as the provider's (src/tau_agent/tools.py:15-18), so a tool can take a Stop button without knowing anything about providers. bash does not poll in its own body: it waits on the command and on a watcher of that token at the same time, and acts on whichever finishes first (src/tau_coding/tools.py:783-793), so its own body burns nothing while it waits. It then kills the whole process group, not one process (src/tau_coding/tools.py:1174-1180). A cancelled bash also refuses before it starts, with the same words it would have ended with (src/tau_coding/tools.py:622-623). And on a cancelled exit Tau's finally does durably what lesson 10 taught you to do only in the view: it appends a synthetic result for every dangling call, with the text Tool call interrupted by user (src/tau_agent/tool_history.py:16), and pushes those messages to subscribers inside suppress(Exception) so that a broken listener cannot swallow the cancellation on its way out (src/tau_agent/harness.py:192-205). That is the bug lesson 11's comparison described, fixed, and it is why Tau's record holds a result where yours holds a hole. Neither is wrong: Tau writes down that nobody knows, and you compute the same sentence fresh on every request.

Where Tau is weaker. The watcher polls every 50 milliseconds, so a cancelled command can keep the machine for another fiftieth of a second, and every waiting tool pays for a task of its own whether or not anyone ever cancels. More seriously, the gap above is real: between prompt() and the consumer's first step, Tau has no token, and a Stop in that window is lost without a trace. And aborted is declared and unwritten, so a tool that is cancelled and a provider that failed look the same in a saved session.

Where yours is weaker. grace counts turns of the event loop, which is a fine unit in a lab where all time is fake and a poor one on a real machine, where you would want seconds. Your bash polls with await asyncio.sleep(0), which is a busy loop: it burns the processor for as long as the command runs. [general] On a real event loop you would sleep for a small interval instead, as Tau does, or wait on the token itself. Nothing in your harness times out on its own, so a hung provider waits for ever unless somebody presses Stop. And your token has one bit; it cannot say who cancelled, or why, or offer a reason to put in the message.

src/tau_agent/loop.py:301-313,355-361 · pinned to commit 9fe6a71 · view on GitHub

A new case. This time nothing is wrong with the tools: the model call hangs for sixty seconds. You have two mechanisms. Which one stops it against the lab's model, and which one against a provider that calls urllib on a worker thread?

The token stops both. That is what a cancel flag is for, and both are code that waits

Both wait; only one looks. The lab's model reads the signal on the way in. urllib has never heard of it.

Cancelling the task stops both, because in an async program the thread is just another thing being awaited

Cancelling the task ends your waiting. The thread is not awaiting anything; it is inside a socket read, and it carries on until the socket answers or times out.

The token stops the lab's model, because it looks. Against the thread only the hard cancel gets your run back, and even then the request itself keeps going

Two different questions: when does the user get their prompt back, and when does the work actually end. Only the first has a general answer.

Against a model that reads the signal, stop_run says clean and the record holds one aborted message. Against one that never looks, it says hard and the record holds the prompt alone: the reply never arrived, so there is nothing to write down. [general] A real provider client usually does take a cancellation token or an abort signal, and closes the connection; one that does not leaves a request billing in the background while your user is already typing the next thing.

import asyncio, harness, lab

SYSTEM = "You are a careful assistant."


class BlockingProvider:
    """A provider that goes away for sixty seconds and looks at nothing while it is gone.
    That is urllib on a worker thread, written as one class."""

    def __init__(self):
        self.calls = []

    async def acomplete(self, system, messages, tools=(), signal=None):
        self.calls.append(messages)
        await asyncio.get_running_loop().create_future()        # comes back in sixty seconds


async def consume(run):
    async for event in run:
        pass


async def press_stop_during_the_model_call(label, model):
    h = harness.Harness(model, SYSTEM, [])
    task = asyncio.ensure_future(consume(h.prompt("Summarise the repo.")))
    await asyncio.sleep(0)                                      # the request is on its way
    how = await harness.stop_run(h, task, grace=5)
    print(f"{label}:")
    print(f"  stop_run said {how!r}; the record is {lab.shape(list(h.messages))!r}")


await press_stop_during_the_model_call("the lab's model, which looks at the signal",
                                       lab.ScriptedModel([lab.say("unused")]))
print()
await press_stop_during_the_model_call("a provider that never looks", BlockingProvider())

the lab's model, which looks at the signal:
  stop_run said 'clean'; the record is 'U A(aborted)'

a provider that never looks:
  stop_run said 'hard'; the record is 'U'

You hit

a Stop button with nothing behind it: three minutes into a ten-minute build, no code of yours was running

You built

CancelToken, ABORTED, Harness.cancel(), three checks where the run waits, a polling bash, and stop_run()

The principle

await is how one thread waits politely; a cancel flag stops nothing until slow code looks at it

Your harness now

run_tool
run_agent
context_for_model
repair_tool_history
Harness
SessionLog
persist_to
summarize
find_cut
compact
maybe_compact
discover_context
build_system_prompt
guard
deny_destructive
confirm_with
CancelToken
ABORTED
Harness.cancel
stop_run

Still open

Fifteen lessons of harness, and it has never met a model. A real provider does not take your dicts and does not return them: it sends back a stream of lines in which a tool call's arguments arrive as fragments of a JSON string. Lesson 16.

Previous Checkpoint C Where does it go?Next 16 Capstone: out of the browser Sixteen lessons of harness, and it has never met a model. What has to change?