11 · Memory that outlives the process

Pull the plug

The agent has written two of its three files. Then somebody trips over the power cable. Everything your harness knows is in a Python list.

~65 min · 2 labs · builds on B Transcript doctor II

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

Lesson 10's repair_tool_history runs before every request, including the thousands of requests whose history is fine. Hand it a list that is already valid. Are the messages that come back the same objects you put in?

No. It returns copies, so that nothing done to the view can ever reach the record

That is the record-and-view rule applied one level too deep: the belief that separate lists need separate messages. The lists are separate. The messages are shared, because nobody edits a message.

Yes: a new list, holding the very same message objects

A new list protects the record's order and length. Copying every message on every request would cost time and buy nothing.

Yes, and the list is the same object too. There was nothing to repair, so there is nothing to build

That is the stored list and the sent list being one list again, on the days when nothing is wrong. Then the caller gets something it may change on some days and must not touch on others.

Equal, a different list, the same dicts inside. That is why you could afford to call it before every request. Hold on to the picture: today the record moves to a disk, and the view is still worked out in exactly this way.

import harness, lab

read = lab.call("read", {"path": "config.py"}, id="c1")
messages = [lab.user("Which port?"), lab.reply(read),
            lab.tool_result(read, "PORT = 9090\n"),
            lab.say("9090.")]
print("valid already:", lab.validate(messages) == [])

view = harness.repair_tool_history(messages)

print("view == messages:", view == messages)
print("view is messages:", view is messages)
print("same message objects inside:",
      all(a is b for a, b in zip(view, messages)))

valid already: True
view == messages: True
view is messages: False
same message objects inside: True

A teammate wants a frontend that beeps every time a tool starts, so they can look away from the screen. How many lines of run_agent have to change?

None. Since lesson 7 run_agent announces tool_execution_start and has no idea who is listening. The beeper is three lines in somebody's for event in h.prompt(...) loop. Keep that loop in mind. You will need it again today.

The cable

Your harness survives a tool that crashes, a model that will not stop, a provider that falls over and a user who walks away mid-call. Everything those defences protect is one Python list, inside one process. So here is the oldest failure there is: somebody trips over the cable.

The job is three small files, one per turn. The program below is the sensible first draft of saving, and main is all of it: load the session if there is one, run the job, write the list to the disk when the run ends. lab.power_cut() kills the process during the third model call. ws.reboot() hands you the disk as the dead process left it, and main runs again, as it would when you restart the program.

The model is scripted, but fairly: it decides what to write next only from the transcript it is sent. A file counts as done if the transcript shows a write call for it.

The power goes out during model call 3. By then the agent has asked for a.py and b.py, and your write tool has run twice. After the reboot, what is on the disk, and what does the restarted agent do?

The two files and a saved session of five messages. It writes only c.py

That has the save happening somewhere along the way. It is one line, and it sits after the loop. A power cut is not an exception your code gets to see: lesson 5's except never runs, no finally runs, and neither does the line after the loop.

The two files and no session. It writes all three again

The disk remembers what the tools did. Nothing remembers that they did it.

The two files and no session. But the model was there for the first two turns, so it carries on with c.py

That gives the model a memory of its own. It had the transcript you sent it, three times, and kept none of them. After the reboot you send an empty one.

A dead process runs no last lines, so "save when the run ends" saved nothing. The tools' work was never waiting to be saved: on this page's disk a file is there the moment write returns. So the disk says two files are done, the agent believes none are, and it does them again, on top of whatever is there, paying for the first two calls a second time. What had to be safe was never "the run" but each message, at the moment it was complete.

Where this comes from: Tau's note on how it saves sessions says it in one line, "message_end remains the durable-message boundary" (dev-notes/push-based-persistence.md:53).

import json
import harness, lab

def main(ws, model):
    """The whole program: load, run the job, save at the end."""
    saved = []
    if ws.exists("session.json"):
        saved = json.loads(ws.read_text("session.json"))
    print("loaded", len(saved), "messages")
    tools = [harness.make_write_tool(ws)]
    h = harness.Harness(model, harness.SYSTEM, tools,
                        messages=saved)
    for event in h.prompt("Write a.py, b.py and c.py."):
        pass
    # the run has ended: save
    ws.write_text("session.json", json.dumps(list(h.messages)))

FILES = {"a.py": "A = 1\n", "b.py": "B = 2\n", "c.py": "C = 3\n"}

def writer(request):
    """The model. It goes only by the transcript it is sent: a
    file is done if the transcript holds a write call for it."""
    done = [block["arguments"]["path"]
            for m in request.messages if m["role"] == "assistant"
            for block in m["content"]
            if block["type"] == "toolCall"]
    todo = [path for path in FILES if path not in done]
    if not todo:
        return lab.say("All three files are written.")
    arguments = {"path": todo[0], "content": FILES[todo[0]]}
    return lab.reply(lab.call("write", arguments,
                              id=f"w{len(done) + 1}"))

ws = lab.Workspace()
steps = [writer, writer, lab.power_cut()]   # the plug: call 3
try:
    main(ws, lab.ScriptedModel(steps))
except lab.PowerCut:        # caught by the stage, not by your code
    print("-- the power went out during model call 3 --")

disk = ws.reboot()      # the process died, the disk did not
print("on the disk:", disk.listdir())

main(disk, lab.ScriptedModel([lab.forever(writer)]))
print("written after the reboot:",
      [path for op, path, _ in disk.writes if path in FILES])

loaded 0 messages
-- the power went out during model call 3 --
on the disk: ['a.py', 'b.py']
loaded 0 messages
written after the reboot: ['a.py', 'b.py', 'c.py']

What goes on the disk

So save more often. How often? The list only ever changes in one way: it gains a message. Save at every change, then: after every message. Save what, though? The obvious thing is the thing you have. json.dumps(messages), written over the old file (which starts by emptying it), every time the list grows.

The other candidate looks clumsier. Never touch what is already in the file. For each new message, add one line at the end.

Both work on a good day. Below, the power goes out while message 4 is being saved, in both designs, at the same point: 60% of the bytes being written have reached the disk.

Messages 1 to 3 were saved without trouble a while ago. The cut comes while message 4 is being saved. How many messages can you get back from each file?

Three from each. Only the message being written can be hurt

That treats a file as a variable: the assignment happened or it did not. Writing a file over is not one step. It begins by emptying the file, and the power may go at any byte after that.

None from the one document. Three from the lines

The document was being written, all of it, so all of it was at risk. Of the lines, only the last was being written.

None from the document, and that is the better failure: an unreadable file is safer than a half-right one

All or nothing is a fine property when "nothing" means the previous version. Here it means nothing. You can buy the previous version back (write a second file, then rename it over the first), at the price of copying the whole session for every message.

A power cut leaves you the first part of whatever was being written, and nothing else of it. Rewrite the document and "whatever was being written" is everything you have; add a line and it is that one line, so with one writer the damage can only be at the tail, and message 400 costs one line, not 400. A file that only grows, one record per line, is an append-only log. One JSON object per line is common enough to have a name, JSON Lines, and an extension, .jsonl.

import json
import lab

write = lab.call("write", {"path": "a.py", "content": "A = 1\n"},
                 id="w1")
messages = [lab.user("Write a.py, b.py and c.py."),
            lab.reply(write),
            lab.tool_result(write, "Successfully wrote to a.py."),
            lab.say("a.py is done.")]

def cut_short(text):
    """A write the power cut interrupted: 60% of it got to disk."""
    return text[:len(text) * 6 // 10]

def readable(text):
    try:
        json.loads(text)
        return True
    except ValueError:
        return False

# Messages 1 to 3 were saved a while ago. The power goes out while
# message 4 is being saved.

# A: one JSON document, written afresh for every message.
#    Writing afresh starts from an empty file.
blob = cut_short(json.dumps(messages))

# B: one line per message, added at the end.
#    Lines 1 to 3 are not being written at all.
lines = "".join(json.dumps(m) + "\n" for m in messages[:3])
lines += cut_short(json.dumps(messages[3]) + "\n")

print("A ends:", repr(blob[-30:]))
print("A: messages you can get back:",
      len(json.loads(blob)) if readable(blob) else 0)
print()
print("B ends:", repr(lines[-30:]))
print("B: messages you can get back:",
      sum(readable(line) for line in lines.splitlines()),
      "of", len(lines.splitlines()), "lines")

A ends: 'ult", "tool_call_id": "w1", "t'
A: messages you can get back: 0

B ends: '": [{"type": "text", "text": "'
B: messages you can get back: 3 of 4 lines

Reading it back

You have given something up. With the one document, "the current messages" was a thing on the disk. Now the disk holds no current anything, only a history of additions, each wrapped in a small envelope with an id:

{"id": "e1", "type": "message", "message": {"role": "user", "content": "My name is Ada."}}

The envelope costs a few bytes and is there for next lesson: a type, because a log will come to hold things that are not messages, and an id, so that one line can point at another. Today every entry is a message and nothing points anywhere.

Parse every line and you have entries, a list of dicts like the one above, oldest first. Nobody ever edits a line. Write the one line of Python that gives you the transcript. Then: how much new code does "resume a session" need, and where would it go?

The file is in the order things happened. What order does a transcript want? And what has lesson 8's Harness always accepted?

messages = [entry["message"] for entry in entries]

Read the lines in order and keep the messages. The state is not stored anywhere; it is worked out from what happened, each time someone asks: a replay. And "resume" needs no code at all, because lesson 8's Harness has taken messages= since the day you wrote it. Put your line into the cell below and watch a new process answer.

Common answers, and what each one misses

"Read the last line: it holds the latest state." That is the one-document design wearing a disguise. A line holds one message. No line holds the state, which is why no cut can lose it.
"Keep a second file with the current list, for fast loading." Two copies, written at different moments, and a power cut between them. Which one is right? Lesson 2 had this argument about text and calls. A value you work out cannot go stale.
"Resume needs a loader that restores the harness: its turn count, its queues, whether it was running." The harness has nothing else worth restoring. The transcript is the only memory, and a queued message was never history.

A two-line session file, left behind by a process that no longer exists. The cell ships with a guess at the one line, and the guess loses Ada. Put your line in and run it again.

import json
import harness, lab

# A session file as the last process left it: one entry per line, oldest first.
FILE = (
    '{"id": "e1", "type": "message", "message": '
    '{"role": "user", "content": "My name is Ada."}}\n'
    '{"id": "e2", "type": "message", "message": '
    '{"role": "assistant", "stop_reason": "stop", '
    '"content": [{"type": "text", "text": "Noted."}]}}\n')

entries = [json.loads(line) for line in FILE.splitlines()]

# A guess: "the last line holds the state." Put your one line here.
messages = [entries[-1]["message"]]

print(lab.show(messages))
print()

# A new process, a new Harness.
model = lab.ScriptedModel([lab.forgetful()])
h = harness.Harness(model, harness.SYSTEM, [], messages=messages)
for event in h.prompt("What is my name?"):
    pass
print(lab.show(h.messages[-2:]))

assistant -> "Noted."

user -> "What is my name?"
assistant -> "I don't know your name."

With the right line in, the model "remembered" Ada across a reboot. It did nothing of the kind. The file did, you read it back, and you sent it again: lesson 1, with a disk in the middle.

Figure 11.1 The opening run again, this time with a log. (The figure numbers its calls c1 and c2; the cells on this page name them w1 and w2.)

Append: one line for each completed message. Lines e1 to e5 mirror the five messages in memory, and the slot for e6 shows where the next one goes. No earlier line ever changes.
power_cut(): the process died and the list died with it. The file is untouched.
replay(): after ws.reboot(), the same five messages, rebuilt from the lines in file order.

Who does the saving?

Suppose the log exists. You will write it in the first lab, and its surface is what you just derived: log.append_message(message) adds a line, log.replay() reads them all back. Somebody still has to call append_message at the right moments.

Since lesson 7 there is an obvious somebody. The loop that consumes a run already sees every message_end, so the terminal frontend can save as it prints. One line. Below, that design lives through a working week.

Monday you work in the terminal, whose loop prints and saves. Tuesday the same session is driven from a script through JsonRenderer, in a loop somebody else wrote. Wednesday, after a reboot, you ask the agent which files it has written. You need to read only two functions, print_frontend and json_frontend: one has a line the other lacks. Work out what Wednesday will print. The run will tell you whether you were right about that; it cannot tell you where the line belongs. That part is yours: where should that line live?

In every frontend. Add the same line to the JSON loop, and to each loop written after it

That makes saving a rule every consumer has to remember, and the cell is what it looks like when one forgets. The first cell on this page consumed its run with a bare for ... pass: no frontend at all. Who saves then?

Inside run_agent. Wherever it appends to the list, it appends to the log as well

Lesson 7 took print out of the loop so that the loop need not know what a screen is. This puts a disk in. run_agent would need a log, a path and an opinion about JSON, and next month somebody wants metrics.

Somewhere new: the harness keeps a list of functions it was handed, and calls each of them with every event

The harness is the one place every event of every run passes through, whoever consumes that run. And it need not know what the functions do with them: one saves, and next month one counts.

Wednesday's answer is a.py, with b.py sitting on the disk beside it. Tuesday's run happened and nobody wrote it down, because saving was a line in somebody else's loop. A for loop over a run is one consumer pulling events when it feels like it; saving needs the opposite, to be told, every time, whether or not anybody pulls. So the harness will keep a list of functions to call with each event, and the saver becomes one of them.

Where this failure comes from: Tau used to save from its consumer's loop. Pressing Esc in its terminal UI cancelled that consumer, and messages went missing from session files (dev-notes/push-based-persistence.md:17-26). Lesson 15 comes back to this.

import contextlib, io
import harness, lab

def print_frontend(h, log, text):
    for event in h.prompt(text):
        if event["type"] == "message_end":
            log.append_message(event["message"])   # the save
            print("  " + lab.show([event["message"]]))

def json_frontend(h, text):        # written later, by someone else
    renderer = harness.JsonRenderer()
    pipe = io.StringIO()            # its reader is another program
    with contextlib.redirect_stdout(pipe):
        for event in h.prompt(text):
            renderer.render(event)
    print("  (", len(pipe.getvalue().splitlines()),
          "JSON lines, piped to another program )")

def scribe(request):
    """The model. It writes the file the user names, and answers
    other questions from the write calls in the transcript it is
    sent. It has nothing else to go by."""
    asked = request.messages[-1]
    done = [block["arguments"]["path"]
            for m in request.messages if m["role"] == "assistant"
            for block in m["content"]
            if block["type"] == "toolCall"]
    if asked["role"] != "user":
        return lab.say("Done.")
    if asked["content"].startswith("Write "):
        path = asked["content"].removeprefix("Write ").rstrip(".")
        return lab.reply(lab.call(
            "write", {"path": path, "content": "X = 1\n"},
            id=f"w{len(done) + 1}"))
    return lab.say("So far I have written: " + ", ".join(done))

def session(ws):
    log = harness.SessionLog(ws, "session.jsonl")
    model = lab.ScriptedModel([lab.forever(scribe)])
    h = harness.Harness(model, harness.SYSTEM,
                        [harness.make_write_tool(ws)],
                        messages=log.replay())
    return h, log

ws = lab.Workspace()
h, log = session(ws)
print("Monday, in the terminal:")
print_frontend(h, log, "Write a.py.")
print("Tuesday, same session, from a script:")
json_frontend(h, "Write b.py.")

disk = ws.reboot()
h, log = session(disk)
print("Wednesday, after a reboot:")
print("  on the disk:", disk.listdir())
print_frontend(h, log, "Which files have you written?")

Monday, in the terminal:
  user -> "Write a.py."
  assistant -> toolCall w1 write({"path": "a.py", "content": "X = 1\n"})
  toolResult w1 -> "Successfully wrote to a.py."
  assistant -> "Done."
Tuesday, same session, from a script:
  ( 12 JSON lines, piped to another program )
Wednesday, after a reboot:
  on the disk: ['a.py', 'b.py', 'session.jsonl']
  user -> "Which files have you written?"
  assistant -> "So far I have written: a.py"

Figure 11.2 Push, then pull. At badge 1 the harness pushes the event to every function it was given, the saver among them. Only then, at badge 2, is the event yielded to the one consumer that is pulling. The log does not depend on anybody pulling.

Saving every message narrows the gap. It cannot close it: a tool can do its work and the process die before the result becomes a message. Here the model asks for pytest, that message reaches the log, and then the process is killed outright (kill -9: no warning, no last lines). No result line follows, ever. Next morning a new process replays the log as it is, builds a Harness on it and sends a prompt. What happens?

A 400. The log has to be mended first: append a result for c1, then resume

That takes the stored list and the sent list to be one list. You separated them last lesson, and a list that came from a file is still just the record.

It works. The model is sent a made-up result for c1, and the log keeps its hole

Look for the line in the output that nobody ever wrote to the disk.

It works only if replay() drops the dangling call. Half a turn is not worth keeping

Dropping it deletes the one line that says pytest may have run. Read the model's answer in the output: it is honest because that line is there.

You wrote the reason last lesson: context_for_model repairs the view before every request, and it does not care whether the record was built a second ago or read from a file written yesterday. Notice which half survived. An assistant message is announced, and so logged, before its tools run, so a cut can leave a call without a result and never a result without a call: the log may not know how it ended, but it always knows what was attempted. The record is not the view, and now the record is a file.

Where this comes from: a Tau session can reach its next prompt with a dangling call, "for example after a persist failure killed the previous run" (tests/test_agent_harness.py:317-321).

import harness, lab

# Yesterday: the model asked for pytest, the log got that
# message, and then: kill -9.
ws = lab.Workspace()
log = harness.SessionLog(ws, "session.jsonl")
pytest = lab.call("bash", {"command": "pytest"}, id="c1")
log.append_message(lab.user("Run the tests."))
log.append_message(lab.reply(lab.text("Running them."), pytest))
print("the log replays as:  ", lab.shape(log.replay()))

# Today: a new process, the same disk. Nothing is mended first.
disk = ws.reboot()
log = harness.SessionLog(disk, "session.jsonl")
model = lab.ScriptedModel([lab.say(
    "I can't tell whether the tests ran. Shall I run them?")])
bash = harness.make_bash_tool(lab.Shell(disk))
h = harness.Harness(model, harness.SYSTEM, [bash],
                    messages=log.replay())
for event in h.prompt("Are you still there?"):
    pass                # nobody saves yet: that is the second lab

sent = model.calls[0].messages
print("the model was sent:  ", lab.shape(sent))
print(lab.show(sent[2:3], clip=200))
print(lab.show(h.messages[-1:]))
print("the record is now:   ", lab.shape(h.messages))
print("the file still holds:", lab.shape(log.replay()),
      "in", len(log.entries()), "lines")

the log replays as:   U A[c1]
the model was sent:   U A[c1] R(c1) U
toolResult c1 -> "Tool call interrupted: no result was recorded. It may not have run, or may have run partly; check before repeating it."  [is_error]
assistant -> "I can't tell whether the tests ran. Shall I run them?"
the record is now:    U A[c1] U A
the file still holds: U A[c1] in 2 lines

Build: a session that outlives its process

The first lab is the file. Python you need, and nothing more: json.dumps(obj) gives one line of text, always: with its default settings it never produces a raw line break (a newline inside a string comes out as the two characters \n), so one message is exactly one line. json.loads(line) gives the object back and raises a ValueError when the line is not JSON, text.splitlines() cuts a text into lines, and enumerate(lines, 1) counts from 1. json is already imported at the top of your file: JsonRenderer needed it.

Spec only, as in lesson 10. At the bottom of harness.py a new region has arrived, == 6 environment ==, for what a session needs around the harness. Write SessionLog(ws, path) there. About 18 lines.

append_message(message) appends one JSON line with ws.append_text and returns its id: "e" plus the number of that line. Count by reading the file again. A SessionLog keeps nothing but ws and path, so two of them over one file, or one made after a reboot, can never disagree about the next id. It is the rule of this lesson in small: work it out, do not store it.
entries() returns every entry, as a list. With no file it returns [], and it never creates one.
rows() returns (id, message) pairs, and replay() the messages.
A line that is not JSON raises ValueError(f"{path}, line {n}: not a complete log entry"), with n counted from 1. It is never skipped, and append_message refuses in the same way, writing nothing. The last question of this lesson is about why.

Four methods. How many of them need to touch the disk? If entries() is right, what is rows() in terms of it, and replay() in terms of rows()? And if you know how many entries there are, what is the number of the next line?
Make entries() the only method that touches the disk, and build the other three on it. It takes the file's text (an empty text when ws.exists says there is no file, because reading must never create one), cuts it into lines and parses each, counting from 1. When a line will not parse, raise your own error with the path and that number in it: the JSON parser's own message talks about "line 1 column 17" of the line, which is the wrong number. Return a real list, not a generator: callers take len() of it. append_message asks entries() how many there are before it writes anything. That one call gives you the next id, and the refusal on a torn file.

Nearly the code:

entries():
    text = the file's text, or "" if there is no file
    for number, line in the lines, numbered from 1:
        try: parse the line, keep the result
        if it will not parse: raise ValueError,
            in the spec's words, with number
    return the list

append_message(message):
    n = how many entries there are now
        (this is what refuses a torn file)
    entry = id "e" and n + 1, type "message", the message
    append the entry as JSON text plus a newline
    return the id

rows():   an (id, message) pair for each entry
          whose type is "message"
replay(): the message out of each row

A session that outlives its process. Look at what you did not write: a loader, a "resume" function, a saved copy of the list. The last test resumed a session that died mid-call, and the only code involved was lesson 8's constructor and lesson 10's repair.

What changed since lesson 10

@@ -398,2 +398,32 @@
         print(text_of(self.last))
         return True
+
+
+# given in lesson 11
+# == 6 environment ==
+# What a session needs around the harness, starting with a disk. Regions 3 and 4 know none of it.
+
+# --- your code ---
+# Spec only. The tests import one name: SessionLog.
+#
+# SessionLog(ws, path) is a session on disk. `ws` is a workspace, the disk your tools already use
+# (ws.exists, ws.read_text, ws.append_text), and `path` names a file in it. The file holds one
+# JSON object per line, one line per message, and a line, once written, is never touched again:
+#     {"id": "e1", "type": "message", "message": {"role": "user", "content": "Hi"}}
+#
+#   .append_message(message) -> str   appends one line with ws.append_text and returns its id:
+#                     "e" + the number of that line, so "e1", "e2", ... The number is counted by
+#                     re-reading the file. A SessionLog keeps no state but `ws` and `path`, so two
+#                     of them over one file cannot disagree.
+#   .entries() -> list[dict]          every entry, oldest first; [] if there is no file yet
+#                     (reading never creates one). A line that is not JSON, the torn last line
+#                     of a power cut, raises
+#                         ValueError(f"{path}, line {n}: not a complete log entry")
+#                     with n counted from 1. It is never skipped, and append_message refuses in
+#                     the same way, writing nothing, while the log is in that state.
+#   .rows() -> list[tuple]            (entry id, message) for each message, in order
+#   .replay() -> list[dict]           the messages, in order: the transcript
+#
+# Nothing here is a special "load" or "resume". To resume a session is to write
+#     Harness(model, system, tools, messages=log.replay())
+# --- end ---

Four messages appended: the ids come back as e1, e2, e3, e4, and the file holds four lines, each a JSON object with "id", "type": "message" and "message".
ws.writes shows appends and nothing else, and the file after each message starts with the file as it was before.
replay() is the four messages, usage numbers and all; rows() pairs each with its id; and a new SessionLog on the rebooted disk says the same.
With no file on the disk, entries(), rows() and replay() are all [], and looking did not create the file.
Two SessionLog objects take turns appending to one file, and both are read after every append: the ids are e1, e2, e3, e4, and each of them always replays everything written so far, by either.
A two-line file in the format of the spec, written by hand: replay() gives its two messages and the next append is e3.
The power went out halfway through line 5. entries(), rows() and replay() each raise an error that says "line 5"; append_message raises too and leaves the file alone. Torn at line 2 of 4, the error says "line 2".
The log ends on a tool call with no result (kill -9). Harness(..., messages=log.replay()) on the rebooted disk answers the next prompt, and the model is shown the whole earlier session.

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. SYSTEM = "You are a helpful assistant." # standing instructions, sent with every call 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): """An assistant message that says why a run stopped: nothing in `content`, stop_reason "error", and the reason in `error_message`. A stop is written into the record, not raised.""" return {"role": "assistant", "content": [], "stop_reason": "error", "error_message": text} # == 2 tools == # A tool is a dict. "name", "description" and "parameters" are advertised to the model. # "execute" is a function, execute(arguments) -> str, that only this program ever runs. 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] def run_tool(tools, call): """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.""" 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 = 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} 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.""" def execute(arguments): 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, } def make_write_tool(ws): """The `write` tool, over the workspace `ws`: create a file, or replace all of it.""" def execute(arguments): 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, } 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.""" def execute(arguments): output, code = shell.run(str_arg(arguments, "command")) 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, } # == 3 loop == # The code that talks to the model. The model keeps nothing between calls. def run_agent(model, system, messages, tools, prompt, *, max_turns=None, get_steering=None, get_follow_ups=None): """Put `prompt` to the model and keep going until it stops asking for tools, a reply comes back failed (stop_reason "error"), or `max_turns` model calls have been made (None: no limit). A generator of events: plain 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}") else: reply = model.complete(system, context_for_model(messages), tool_specs(tools)) yield record(reply) # A failed reply: none of it is acted on and nobody is asked for more. failed = reply["stop_reason"] == "error" calls = [] if failed else tool_calls(reply) for call in calls: yield {"type": "tool_execution_start", "call": call} result = run_tool(tools, call) 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 and are empty are left out (a failed reply that got something out stays in), and 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") == "error")] 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 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 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.""" 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() @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 _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 [] 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: for event in run: yield event finally: # exhausted, closed by the consumer, or dead run.close() 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") == "error": 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 (stop_reason "error") 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"] == "error": print(f"Error: {self.last['error_message']}") return False print(text_of(self.last)) return True # given in lesson 11 # == 6 environment == # What a session needs around the harness, starting with a disk. Regions 3 and 4 know none of it. # --- your code --- # Spec only. The tests import one name: SessionLog. # # SessionLog(ws, path) is a session on disk. `ws` is a workspace, the disk your tools already use # (ws.exists, ws.read_text, ws.append_text), and `path` names a file in it. The file holds one # JSON object per line, one line per message, and a line, once written, is never touched again: # {"id": "e1", "type": "message", "message": {"role": "user", "content": "Hi"}} # # .append_message(message) -> str appends one line with ws.append_text and returns its id: # "e" + the number of that line, so "e1", "e2", ... The number is counted by # re-reading the file. A SessionLog keeps no state but `ws` and `path`, so two # of them over one file cannot disagree. # .entries() -> list[dict] every entry, oldest first; [] if there is no file yet # (reading never creates one). A line that is not JSON, the torn last line # of a power cut, raises # ValueError(f"{path}, line {n}: not a complete log entry") # with n counted from 1. It is never skipped, and append_message refuses in # the same way, writing nothing, while the log is in that state. # .rows() -> list[tuple] (entry id, message) for each message, in order # .replay() -> list[dict] the messages, in order: the transcript # # Nothing here is a special "load" or "resume". To resume a session is to write # Harness(model, system, tools, messages=log.replay()) # --- end ---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 11 tests.py

Build: nobody's loop

Now the hook. The harness gets a list of functions, called listeners, or subscribers. h.subscribe(listener) adds one and returns the function that takes it off again, so whoever subscribes holds their own undo. For every event of every run the harness calls each listener, and then yields the event to the consumer, if there is one. That order is the whole design, and lesson 10 is why. yield hands the event over and waits to be asked for the next one. A consumer that walks away never asks, so no line after that yield ever runs. Whatever must happen, happens before the yield. The saver is a listener five lines long: h.subscribe(persist_to(log)), and from then on nothing else in the program saves anything.

Figure 11.3 What the harness owns now: the list, the running flag, the two queues from lesson 9, and the listeners added by subscribe(). run_agent still owns nothing, and has not changed by a line.

Three places in Harness (region 4; the gap is marked), about 12 lines, and one new function at the very bottom of the file, about 5.

__init__: an empty list of listeners.
subscribe(self, listener) adds one and returns a function that removes it. Calling that function a second time does nothing.
_run: every listener is called with each event, of every type, and then the event is yielded. Put the calling in _notify(self, event). A listener may unsubscribe while it is being called, and its neighbours must still hear that event. A listener that raises ends the run, in front of whoever is consuming it. Catch nothing: a saver that cannot save must not be silent.
persist_to(log) returns a listener that appends the message of every message_end to the log and ignores every other event. It catches nothing either.

The tests pull the plug, walk away after every possible event, and consume the same job three ways.

In _run you hold an event and owe it to two parties: the listeners and the consumer. The consumer is the screen, and since lesson 10 you know it may close the run the instant it is handed an event. Who has to get the event first, so that the screen never shows a message the log does not have?
subscribe appends to self._listeners and returns an inner function that removes listener only if it is still in the list. _notify calls each listener with the event, walking list(self._listeners), a copy: a listener that unsubscribes mid-walk shortens the real list under the for, and the loop then skips the next listener, which may well be the saver. In _run, self._notify(event) goes on the line before yield event, with no try around it. persist_to defines an inner listener(event) and returns it.

Nearly the code:

__init__:  listeners = an empty list

subscribe(listener):
    add it to the list
    define undo(): if it is still on the list, take it off
    return undo

_notify(event):
    for each listener in a COPY of the list:
        call it with the event

_run:  for each event of the run: notify, then yield

persist_to(log):
    define listener(event):
        if the event's type is message_end,
            append its message to the log
    return listener

Record, log and consumer now agree at every event, and the tests tried every one of them. The order did it: change the list, tell the saver, then show the screen. No frontend saves anything, so no frontend can forget to.

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.

The three-file job with h.subscribe(persist_to(log)) and nothing else: the log holds one entry per message_end, in order, and log.replay() equals list(h.messages).
power_cut() fires during model call 3, after a.py and b.py were written. On the rebooted disk, Harness(..., messages=log.replay()) is sent all five completed messages, writes only c.py, and the log still equals the record.
The same job consumed three ways: by a bare `await lab.drive(h.prompt(...))`, by a loop that hands each event to JsonRenderer, and by one that hands each to FinalTextRenderer. The three session files are identical and complete.
The consumer closes the run after k events, for every k: the listener has heard exactly those k events, and at each message_end the message was already the last of h.messages.
Close-at-every-k with persist_to subscribed: list(h.messages) == log.replay() at every k, two further prompts are answered, and the log on the rebooted disk still equals the record.
Two listeners; the first unsubscribes itself at the first event it hears. The second still hears every event, that first one included.
Two listeners on one harness and the first is undone, twice: the second still hears the next run, the first hears nothing, and a listener on one harness hears nothing of another harness.
A listener raises OSError("disk full") at the first message_end: the error reaches whoever consumes the run, the harness is idle again, and the next prompt is accepted.
The session file ends in a torn line, so the log refuses every append. With h.subscribe(persist_to(log)), the first prompt ends with that error in front of whoever consumes the run, and the file is untouched.

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. SYSTEM = "You are a helpful assistant." # standing instructions, sent with every call 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): """An assistant message that says why a run stopped: nothing in `content`, stop_reason "error", and the reason in `error_message`. A stop is written into the record, not raised.""" return {"role": "assistant", "content": [], "stop_reason": "error", "error_message": text} # == 2 tools == # A tool is a dict. "name", "description" and "parameters" are advertised to the model. # "execute" is a function, execute(arguments) -> str, that only this program ever runs. 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] def run_tool(tools, call): """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.""" 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 = 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} 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.""" def execute(arguments): 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, } def make_write_tool(ws): """The `write` tool, over the workspace `ws`: create a file, or replace all of it.""" def execute(arguments): 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, } 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.""" def execute(arguments): output, code = shell.run(str_arg(arguments, "command")) 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, } # == 3 loop == # The code that talks to the model. The model keeps nothing between calls. def run_agent(model, system, messages, tools, prompt, *, max_turns=None, get_steering=None, get_follow_ups=None): """Put `prompt` to the model and keep going until it stops asking for tools, a reply comes back failed (stop_reason "error"), or `max_turns` model calls have been made (None: no limit). A generator of events: plain 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}") else: reply = model.complete(system, context_for_model(messages), tool_specs(tools)) yield record(reply) # A failed reply: none of it is acted on and nobody is asked for more. failed = reply["stop_reason"] == "error" calls = [] if failed else tool_calls(reply) for call in calls: yield {"type": "tool_execution_start", "call": call} result = run_tool(tools, call) 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 and are empty are left out (a failed reply that got something out stays in), and 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") == "error")] 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. # given in lesson 11: the docstring's words on subscribe 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 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.""" # --- your code --- # Below is lesson 9's Harness. About 12 new lines, in three places: # - __init__: a list of listeners, empty to start with # - subscribe(self, listener): as the docstring says. The function it returns takes this # listener off the list, and calling it a second time does nothing # - _run: every listener is called with each event, and then the event is yielded. Put the # calling in _notify(self, event). A listener may unsubscribe while it is being called # (itself, or another one), and its neighbours must still hear that event 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() @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 _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 [] 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: for event in run: yield event finally: # exhausted, closed by the consumer, or dead run.close() 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. 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") == "error": 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 (stop_reason "error") 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"] == "error": 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 rows(self): """(entry id, message) for every message of the transcript, in order.""" return [(entry["id"], entry["message"]) for entry in self.entries() if entry["type"] == "message"] def replay(self): """The transcript: the state is computed from what happened, never saved.""" return [message for _, message in self.rows()] # --- your code --- # Spec only. The tests import one name: persist_to. # # persist_to(log) -> listener. `log` is a SessionLog. The listener appends the message of every # message_end event to it, as the event arrives, and ignores every other event. If the log # raises (a torn file, a full disk), so does the listener: it catches nothing. Used as # h.subscribe(persist_to(log)) # and from then on nobody else in the program saves anything. # --- end ---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 11 tests.py

Five files this time, the plug pulled during model call 4, and a first frontend that renders nothing and saves nothing. Once lab 2 has passed, this runs against your code. Reboot, carry on, and count what gets written after the cut.

import harness, lab

NAMES = ("a.py", "b.py", "c.py", "d.py", "e.py")
FILES = {name: f"# {name}\n" for name in NAMES}
JOB = "Write a.py, b.py, c.py, d.py and e.py."

def writer(request):
    """Writes the files one turn at a time, going only by the transcript it gets."""
    done = [block["arguments"]["path"]
            for m in request.messages if m["role"] == "assistant"
            for block in m["content"] if block["type"] == "toolCall"]
    todo = [path for path in FILES if path not in done]
    if not todo:
        return lab.say("All five files are written: " + ", ".join(done) + ".")
    arguments = {"path": todo[0], "content": FILES[todo[0]]}
    return lab.reply(lab.call("write", arguments, id=f"w{len(done) + 1}"))

def session(ws, model):
    """The whole program: read the log, build the harness on it, subscribe a saver."""
    log = harness.SessionLog(ws, "session.jsonl")
    h = harness.Harness(model, harness.SYSTEM, [harness.make_write_tool(ws)],
                        messages=log.replay())
    h.subscribe(harness.persist_to(log))
    return h, log

def py_writes(ws):
    return [path for op, path, _ in ws.writes if path.endswith(".py")]

ws = lab.Workspace()
h, log = session(ws, lab.ScriptedModel([writer, writer, writer, lab.power_cut()]))
try:
    for event in h.prompt(JOB):
        pass                        # a frontend that shows nothing and saves nothing
except lab.PowerCut:
    print("-- the power went out during model call 4 --")
print("written before the cut:", py_writes(ws))

disk = ws.reboot()
model = lab.ScriptedModel([lab.forever(writer)])
h, log = session(disk, model)
print("the log replays as:    ", lab.shape(log.replay()))

frontend = harness.FinalTextRenderer()
for event in h.prompt("The power went out. Carry on."):
    frontend.render(event)
frontend.finish()

print("written after the cut: ", py_writes(disk))
print("log == record:", log.replay() == list(h.messages), "|",
      len(log.entries()), "lines | model calls after the reboot:", len(model.calls))

-- the power went out during model call 4 --
written before the cut: ['a.py', 'b.py', 'c.py']
the log replays as:     U A[w1] R(w1) A[w2] R(w2) A[w3] R(w3)
All five files are written: a.py, b.py, c.py, d.py, e.py.
written after the cut:  ['d.py', 'e.py']
log == record: True | 13 lines | model calls after the reboot: 3

Three model calls after the reboot, not six: d.py, e.py, and the sentence that says the job is done. The agent picked up at d.py because the file said a, b and c were done. Nothing else told it.

Notice an order you got for free. The model's request for write reaches the log before write runs. So the disk can hold a call with no result, which the kill -9 question showed is survivable, but never a side effect that no line admits to.

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.

In your own words, a sentence or two: where does the conversation live now? And what is the list in h.messages, compared with the file?

One of the two can be rebuilt from the other at any moment. Which way round?

The conversation lives in the file, as a history of things that happened. The list in memory is what you get by reading that history back: convenient, fast, and disposable. While the process lives the two agree at every event; that is what lab 2's tests checked, one event at a time. When it dies, only one of them is still there. You never save the state. You write down what happened, and work the state out.

Common answers, and what each one misses

"The transcript gets saved to a file." Nothing ever saves the transcript. Messages are added to a log one at a time, and the transcript is computed from it. That difference is the whole of what survived the plug.
"My program saves each message as it shows it." Whose loop? The saver hears every event whether or not anybody shows anything, and it hears it first.
"The agent now remembers across restarts." The file remembers. The harness reads it back, and the model is sent all of it again, and you are billed for it again ([general] at best at a cached rate), as in lesson 1.

The log is the truth. The transcript is just what you get when you read it back.

Tau's harness notifies its listeners and then yields, walking a copy of the listener list. Saving a session is one of those listeners.

for event in run:
    self._notify(event)
    yield event

def _notify(self, event):
    for listener in list(self._listeners):
        listener(event)

                await self._notify(event)
                yield event
...
    async def _notify(self, event: AgentEvent) -> None:
        for listener in list(self._listeners):
            result = listener(event)
            if isawaitable(result):
                await result

The same shapes.

subscribe returns its own undo, and undoing twice is harmless (src/tau_agent/harness.py:108-115).
Persistence is a subscriber, and the docstring says why: it subscribes "rather than running in the event consumer, which the TUI tears down on interrupt" (src/tau_coding/session.py:3472-3489).
Each completed message becomes one appended entry (src/tau_coding/session.py:3502-3516).
A line that will not parse is a loud error naming its line number (src/tau_agent/session/jsonl.py:44-50).
Replay is a loop over the entries of the active branch that collects (id, message) rows (src/tau_agent/session/memory.py:69-72).
Resume is: read every entry, replay, hand the messages to a new harness (src/tau_coding/session.py:492-513, src/tau_coding/session.py:731-740).

Note what that last citation passes beside the messages: system=system, built from today's configuration. The system prompt is not in the log. A resumed session is the saved conversation plus the current prompt, and lesson 13 builds on that.

Tau is async: until lesson 15, read await f(x) as f(x) and async def as def.

Tau's real reason for pushing. Not a second frontend. Its terminal UI cancels the task that consumes the run when you press Esc; the harness's finally then appends the results for the interrupted calls, and no consumer is left to save them. The regression test is tests/test_agent_harness.py:246-275. You cannot reproduce that yet: your runs cannot be cancelled from outside. Lesson 15.

What Tau adds. Durability you did not build: every append is forced to the disk with fsync, under a lock that other Tau processes take too (advisory: it stops only programs that ask for it). A batch is added the expensive way from earlier on this page: the old file plus the new entries go to a temporary file, which is then renamed over the original (src/tau_agent/session/storage.py:38-73). Old entry shapes are migrated as they are read (src/tau_agent/session/jsonl.py:63-69), and the first three entries of a new session are held back until the first real append, so opening Tau and quitting leaves no file (src/tau_coding/session.py:495-509). Ten kinds of entry, not one, and every entry has a random id and a parent_id, so one file is a tree of conversations (src/tau_agent/session/entries.py:25-32); that is side quest S2. Listeners may be async, which is what the isawaitable lines above are for. A write that fails is retried under the same entry id, so a line that did reach the disk is not written twice (src/tau_coding/session.py:3493-3499).

Where Tau differs from you. Tau does not leave the hole from the kill -9 question. At the start of the next run it appends a made-up result for every dangling call to its record and announces it as a message_end, so the saver writes it to the file (src/tau_agent/harness.py:168-173). That is still append-only. Yours repairs the view on every request and never writes the repair down. Tau yields an assistant's message_end before appending the message to its list (src/tau_agent/loop.py:135-146), so its saver cannot assume, as your tests do, that the message it hears about is already the last of harness.messages; its bookkeeping goes by message identity for that reason (dev-notes/push-based-persistence.md:58-60). Tau skips blank lines in a session file where yours refuses them (src/tau_agent/session/jsonl.py:45-46). During a normal run a listener that raises ends the run in Tau as in yours: listeners are trusted code. Only in the clean-up after a cancel are listener errors suppressed (src/tau_agent/harness.py:196-202).

Where yours is weaker. ws.append_text is a stage prop: it cannot tear and it cannot lie about having written. [general] A killed process loses only what it had not yet handed to the operating system. A power cut can also lose what the operating system had not yet put on the disk, which is what fsync asks for, and some drives lie even then. Two processes appending to your log at once can both count four lines and both write e5. Counting lines on every append is quadratic, which does not matter at this size and would at Tau's. splitlines() also breaks on a few rare characters (U+2028 among them) that json.dumps(..., ensure_ascii=False) would leave raw in a line; Tau splits on "\n" only (src/tau_agent/session/storage.py:90).

What both of you pay, and one tear both of you miss. Telling the saver first has a price: the run waits for every listener, at every event, and Tau awaits them too. And one torn file slips past both logs: a cut after the closing brace and before the newline. The line parses, the next append lands on the end of it, and the pair is unreadable. The fix is to check that the file ends in a newline before you append. Neither your log nor Tau's append does (src/tau_agent/session/storage.py:52-60).

src/tau_agent/harness.py:190-191,207-211 · pinned to commit 9fe6a71 · view on GitHub

Your spec said a torn line is never skipped, and never said why. A teammate finds that rude. Their QuietLog in the cell is your SessionLog with one change: a line it cannot read is passed over without a word. "Four good messages beat no session." The power cut tore line 5 in half; lines 1 to 4 are whole. The quiet log loads, and then one more message is appended. How many messages does the quiet log replay after that append? Pick, and give your reason in one line. The cell tries the quiet way first, then your SessionLog's way.

Five: the four good ones and the new one. Only the torn message is gone

That treats damage as something you can step round and leave behind. Lesson 10 stepped round damage too, but in the view, and the record lost nothing. Here the stepping round happens in the record's own file: append_text adds to the end of the file, and the end of this file is the middle of a line.

Four. The new message is lost as well

The new entry lands on the end of the torn half, the two make one line that is not JSON, and the quiet log skips that too. Look at the id the append returned.

None. One bad line and the file is no longer valid JSON

That is the one-document picture of a file again. Every line is parsed by itself, which is why lines 1 to 4 survive any cut. The damage is confined to the last line, and that is exactly where the next append goes.

Skipping does not lose one message. It loses two: the torn one, and the next one appended, which lands on the end of the wreck and makes one unreadable line that the quiet log skips as well. In an agent that second one is the user's next prompt, and append_message returned e5 for it as if all were well, then handed out e5 again. Your log refuses instead, by line number, and a human with a text editor deletes the torn tail once, knowing what was lost.

Where this comes from: Tau refuses a session line it cannot parse, and names the line (src/tau_agent/session/jsonl.py:44-50).

Why not mend the file on load?

Cut the file back to the last good line, then carry on: the tempting one, and close. [general] Databases often do trim a torn tail off their logs when they recover, and they say so loudly in their own logs. But each of their records carries a checksum, so they can tell a torn tail from corruption. Yours cannot. The lab's other torn file is torn at line 2 of 4. A power cut during an append does not do that, so something worse has happened, and "cut back to the last good line" would throw away two good messages to hide it, unattended, without a word. Without a checksum, refusing is the honest choice.

import json
import harness, lab

class QuietLog(harness.SessionLog):
    """Your SessionLog, except that a line it cannot read is
    skipped without a word."""
    def entries(self):
        entries = []
        for line in self.ws.read_text(self.path).splitlines():
            try:
                entries.append(json.loads(line))
            except ValueError:
                pass
        return entries

def line(n):
    entry = {"id": f"e{n}", "type": "message",
             "message": lab.user(f"message {n}")}
    return json.dumps(entry) + "\n"

whole = "".join(line(n) for n in range(1, 6))
torn = whole[:-35]      # the power went out during line 5

ws = lab.Workspace({"session.jsonl": torn})
quiet = QuietLog(ws, "session.jsonl")
print("quiet log, on load:        ", len(quiet.replay()),
      "messages, and not a word about a fifth")
new_id = quiet.append_message(lab.user("Carry on."))
print("quiet log, after an append:", len(quiet.replay()),
      "messages; the append returned", new_id)
print("the file's last line:")
last = ws.read_text("session.jsonl").splitlines()[-1]
print("  " + last[:60], "...")
again = quiet.append_message(lab.user("Hello?"))
print("quiet log, one more append:", len(quiet.replay()),
      "messages; the append returned", again)
print()

ws = lab.Workspace({"session.jsonl": torn})
try:
    log = harness.SessionLog(ws, "session.jsonl")
    log.append_message(lab.user("Carry on."))
except ValueError as error:
    print("your log: refused, naming line 5:",
          "line 5" in str(error))
print("file untouched:", ws.read_text("session.jsonl") == torn)

quiet log, on load:         4 messages, and not a word about a fifth
quiet log, after an append: 4 messages; the append returned e5
the file's last line:
  {"id": "e5", "type": "message", "message": {"role"{"id": "e5 ...
quiet log, one more append: 5 messages; the append returned e5

your log: refused, naming line 5: True
file untouched: True

You hit

an agent that woke up with two files on the disk and no idea it had written them

You built

SessionLog, Harness.subscribe() with notify-then-yield, and persist_to()

The principle

do not save the state; append what happened, and work the state out by reading it back

Your harness now

run_tool
run_agent
context_for_model
repair_tool_history
Harness
subscribe
FinalTextRenderer
JsonRenderer
SessionLog
persist_to

Still open

The log only grows, and every request is built from all of it. The model's window does not grow. Lesson 12.

One optional thing hangs off this lesson: a side quest about the session tree, where you go back to message three and try another approach without losing the first one. It is an hour, and nothing later needs it — lesson 12 carries on from the harness you have now, whether or not you take it.

Previous Checkpoint B Transcript doctor II Next 12 The wall The prompt is too long and the agent is mid-task. What do you throw away?