Streaming, interrupts and sessions

def repair(messages):
    last = messages[-1]
    if last["role"] != "assistant":
        return messages
    open_calls = [b["id"] for b in last["content"] if b["type"] == "tool_use"]
    if open_calls:
        messages.append({"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": i, "content": "Interrupted by the user.", "is_error": True}
            for i in open_calls]})
    return messages

messages = [
    {"role": "user", "content": [{"type": "text", "text": "Run the tests"}]},
    {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "bash"}]},
]
repair(messages)
print(messages[-1])

Sessions on disk

Save each session as an append-only log, like a ship's log written one line at a time: one JSON object per line (JSONL), written as each message is added. A crash loses at most the line being written. Resuming means reading the lines back into a message list.

• Resume: load the log and carry on.
• Fork: copy the first n messages into a new session, to try a different approach from the same point. It's like photocopying a draft up to page 10, then trying another ending.
• Metadata: session ID, start time, working folder, model. Never the API key.

The world moves on. After a vacation, you'd check what changed before picking up old work. Between sessions, files change and branches move. On resume, tell the model: "Session resumed. Files may have changed since; read them again before editing." The stale-edit check from the file tools lesson is your safety net.

import json

lines = [
    json.dumps({"role": "user", "content": "Fix the bug"}),
    json.dumps({"role": "assistant", "content": "On it"}),
    '{"role": "user", "cont',   # the crash happened here
]

def load(lines):
    messages = []
    for n, line in enumerate(lines):
        try:
            messages.append(json.loads(line))
        except ValueError:
            if n != len(lines) - 1:
                raise        # damage in the middle isn't a crash: stop and look
    return messages

print(load(lines))

import json

pieces = ['{"pa', 'th": "src/', 'app.py", "li', 'mit": 50}']
for p in pieces:
    try:
        json.loads(p)
        print("parsed alone:", p)
    except json.JSONDecodeError:
        print("not JSON alone:", repr(p))
print(json.loads("".join(pieces)))

Assembling a stream correctly

Don't parse tool input until it's finished. Tool input arrives as pieces of JSON text, and each piece on its own is usually invalid.

import json

pieces = ['{"path": "sr', 'c/app.py", ', '"limit": 10}']

try:
    json.loads(pieces[0])
except ValueError as err:
    print("too soon:", err.msg)

print(json.loads("".join(pieces)))

Collect the pieces by the block's index, and parse once at content_block_stop. If you want to show progress before then, display the raw text.

Save as you go

If you only save the transcript when a session ends, then a crash loses everything, including what the agent did to your files. So append each message when it happens. Disk writes cost almost nothing next to a model call.

Resuming isn't simply reloading

A person might come back to a session hours or days later. In that time, files can change, branches can move, and a test that was failing may have been fixed. The transcript still describes the old world, and the model will trust it. So when resuming, tell the model that time has passed, and what's changed if you can tell. For example, add a note: "This session resumed after 3 days. Files may have changed, so read them again before editing." It's a small message, and it prevents the agent from confidently editing code that's no longer there.

Stop means stop

The emergency stop on a machine doesn't wait for the current cycle to finish. When a person presses Stop, they may have spotted the agent doing something wrong. So stopping must take effect at once: close the stream, kill running commands, and don't start anything new. A Stop that waits for the current tool to finish isn't really a Stop. Test it on purpose, with a long-running command, since this is the kind of path that nobody exercises until the day it matters, and then it really has to work.