Capstone: out of the browser

Your lesson 15 harness, byte for byte, driven by a six-line class whose replies are plain dicts written out in the cell. No lab.ScriptedModel. Look at what the harness asked of it.

import harness, lab


class Postcard:
    """Not lab.ScriptedModel, and not a real client either: six lines of code, one of them the
    method your loop calls. The replies are plain dicts, written out in full, in your harness's
    vocabulary."""

    def __init__(self, replies):
        self._replies, self._asked = list(replies), []

    async def acomplete(self, system, messages, tools=(), signal=None):
        self._asked.append(len(messages))
        return self._replies.pop(0)


model = Postcard([
    {"role": "assistant", "stop_reason": "toolUse",
     "content": [{"type": "text", "text": "Let me look."},
                 {"type": "toolCall", "id": "toolu_01", "name": "read",
                  "arguments": {"path": "README.md"}}]},
    {"role": "assistant", "stop_reason": "stop",
     "content": [{"type": "text", "text": "The port is 9090."}]},
])

ws = lab.Workspace({"README.md": "The port is 9090.\n"})
h = harness.Harness(model, "You are a careful coding agent.", [harness.make_read_tool(ws)])
await lab.drive(h.prompt("What port does the server use?"))

print("all there is on the model object:", [n for n in dir(model) if not n.startswith("_")])
print("messages in each request:       ", model._asked)
print("the file it read:               ", ws.reads)
print()
print(lab.show(list(h.messages), stop_reason=True))
print()
print("the run's answer:", harness.text_of(h.messages[-1]))
print("a provider would take that transcript:", lab.validate(list(h.messages)) == [])

all there is on the model object: ['acomplete']
messages in each request:        [1, 3]
the file it read:                ['README.md']

user -> "What port does the server use?"
assistant -> "Let me look." + toolCall toolu_01 read({"path": "README.md"})  [stop_reason=toolUse]
toolResult toolu_01 -> "The port is 9090.\n"
assistant -> "The port is 9090."  [stop_reason=stop]

the run's answer: The port is 9090.
a provider would take that transcript: True

The same stream, through the reference parse_sse, with the number of the line that produced each event. Once lab 2 has passed, this runs on yours.

import lab_fixtures
# A cell always loads the file you are editing under the name `harness`; here that file is
# adapter.py, so this is your parse_sse (the reference one until lab 2 passes).
from harness import parse_sse

seen = [0]


def tape(lines):
    """The stream from the page, counting the data: lines as they are handed over."""
    for line in lines:
        seen[0] += line.startswith("data:")
        yield line


def short(event):
    if "delta" in event:
        return repr(event["delta"])
    if "call" in event:
        return f"{event['call']['name']} {event['call']['arguments']}"
    if "message" in event:
        return f"stop_reason={event['message']['stop_reason']!r}"
    return ""


print("data line        event")
for event in parse_sse(tape(lab_fixtures.anthropic_tool_call)):
    print(f"{seen[0]:>9}        {event['type']:<14} {short(event)}")

data line        event
        1        start          
        3        text_delta     "I'll read"
        4        text_delta     ' the README.'
        6        toolcall_start read {}
        7        toolcall_delta ''
        8        toolcall_delta '{"pa'
        9        toolcall_delta 'th": "READ'
       10        toolcall_delta 'ME.md"}'
       11        toolcall_end   read {'path': 'README.md'}
       13        done           stop_reason='toolUse'

Both endings, through the reference parse_sse: the events, how many of them are terminal, and the message each one carries. Watch what is inside the message in each case.

import lab_fixtures
from harness import parse_sse                  # a cell loads adapter.py under the name `harness`

for name in ("anthropic_overloaded", "anthropic_cut"):
    stream = getattr(lab_fixtures, name)
    last = [line for line in stream if line][-1]
    print(f"{name}: {len(stream)} lines, ending {last[:72]}")
    events = list(parse_sse(iter(stream)))
    print("  events:   ", [event["type"] for event in events])
    print("  terminals:", sum(event["type"] in ("done", "error") for event in events))
    message = events[-1]["message"]
    print("  message:  ", {key: message[key] for key in ("content", "stop_reason",
                                                         "error_message")})
    print()

anthropic_overloaded: 12 lines, ending data: {"type": "error", "error": {"type": "overloaded_error", "message":
  events:    ['start', 'text_delta', 'error']
  terminals: 1
  message:   {'content': [{'type': 'text', 'text': 'Let me'}], 'stop_reason': 'error', 'error_message': 'Overloaded'}

anthropic_cut: 12 lines, ending data: {"type": "content_block_delta", "index": 0, "delta": {"type": "tex
  events:    ['start', 'text_delta', 'text_delta', 'error']
  terminals: 1
  message:   {'content': [{'type': 'text', 'text': 'The answer is'}], 'stop_reason': 'error', 'error_message': 'stream ended without a terminal event'}

That transcript through the reference to_anthropic: the roles that go out, every wire message in full, and the rest of the body. Once lab 1 has passed, this runs on yours.

import json
import lab, lab_fixtures
from harness import to_anthropic              # a cell loads adapter.py under the name `harness`

case = lab_fixtures.request_two_calls_and_steer
body = to_anthropic(case["system"], case["messages"], case["tools"])

print("the transcript:", lab.shape(case["messages"]))
print("wire roles:    ", [message["role"] for message in body["messages"]])
print()
for index, message in enumerate(body["messages"]):
    print(index, json.dumps(message))
print()
print("system:", json.dumps(body["system"]))
print("tools: ", json.dumps(body["tools"]))
print("and the rest of the body:", {k: v for k, v in body.items()
                                    if k not in ("messages", "system", "tools")})

the transcript: U A[toolu_02,toolu_03] R(toolu_02) R(toolu_03) U
wire roles:     ['user', 'assistant', 'user', 'user', 'user']

0 {"role": "user", "content": "Compare a.py and b.py."}
1 {"role": "assistant", "content": [{"type": "text", "text": "Reading both."}, {"type": "tool_use", "id": "toolu_02", "name": "read", "input": {"path": "a.py"}}, {"type": "tool_use", "id": "toolu_03", "name": "read", "input": {"path": "b.py"}}]}
2 {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_02", "content": "A = 1\n", "is_error": false}]}
3 {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_03", "content": "File not found: b.py", "is_error": true}]}
4 {"role": "user", "content": "Forget b.py, it was deleted. Just describe a.py."}

system: "You are a careful coding agent."
tools:  [{"name": "read", "description": "Read a text file from the workspace.", "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}]
and the rest of the body: {'model': 'claude-opus-5', 'max_tokens': 16000, 'stream': True}

A provider whose network is two of those streams, built out of your two functions and nothing else. The crank is turned by hand here, as it was in lesson 3, because a cell can only load one of your two files at a time — the version that drives your real loop is the last hidden test above, and page 2 runs it on your machine. No tests: read the two replies, the usage that came down the wire with them, and the bytes of the second request.

import json
import lab, lab_fixtures, lab_wire
from harness import parse_sse, to_anthropic   # a cell loads adapter.py under the name `harness`

READ = {"name": "read", "description": "Read a text file from the workspace.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}},
                       "required": ["path"]}}
SYSTEM = "You are a careful coding agent."
ws = lab.Workspace({"README.md": "The port is 9090.\n"})

# A provider whose network is two of the page's streams, through YOUR two functions and nothing else.
model = lab_wire.ReplayModel([lab_fixtures.anthropic_tool_call, lab_fixtures.anthropic_text],
                             to_anthropic, parse_sse)

messages = [lab.user("What port does the server use?")]
reply = await model.acomplete(SYSTEM, messages, [READ])
messages.append(reply)
print(lab.show([reply], stop_reason=True, usage=True))

call = reply["content"][-1]                   # turning the crank by hand, as in lesson 3
messages.append(lab.tool_result(call, ws.read_text(call["arguments"]["path"])))
messages.append(await model.acomplete(SYSTEM, messages, [READ]))
print(lab.show(messages[-2:], stop_reason=True, usage=True))
print()
print("what went out on the wire the second time:")
for message in model.requests[1]["messages"]:
    print("   ", json.dumps(message))
print()
print("transcript:", lab.shape(messages), " valid:", lab.validate(messages) == [])

assistant -> "I'll read the README." + toolCall toolu_01 read({"path": "README.md"})  [stop_reason=toolUse]  [usage: {"input": 412, "output": 41, "cache_read": 0}]
toolResult toolu_01 -> "The port is 9090.\n"
assistant -> "The port is 9090."  [stop_reason=stop]  [usage: {"input": 25, "output": 9, "cache_read": 0}]

what went out on the wire the second time:
    {"role": "user", "content": "What port does the server use?"}
    {"role": "assistant", "content": [{"type": "text", "text": "I'll read the README."}, {"type": "tool_use", "id": "toolu_01", "name": "read", "input": {"path": "README.md"}}]}
    {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_01", "content": "The port is 9090.\n", "is_error": false}]}

transcript: U A[toolu_01] R(toolu_01) A  valid: True