Bounded Agent Loop: Unit A Exercises

Student edition · 90 minutes of dedicated work · 2026-09-09

Runs on Google Colab as it ships today, or on any local Python 3.12+ kernel.

This is one of two practical units for Chapter 3. Unit A constructs and connects the mechanism; Unit B investigates a controlled failure, repairs it and transfers the invariant. Each is a complete ninety-minute session, with its own setup and required conceptual introductions. Basic Python variables, conditions, loops, functions, lists and dictionaries are the starting knowledge. Libraries and specialized concepts used here are introduced below before the main task.

By the end you should be able to:

Explain the chapter's mechanism using a prediction and an observed intermediate result.
Implement next-call admission under call-count and cost-exposure limits; trace actual tool observations through the bounded loop.
Solve admit a variable-cost next model attempt using changed inputs and an independent expectation.
Retain your implementation, failed/corrected observations, causal explanation and limits.

Minutes	Dedicated activity	Evidence you produce
0–5	State the problem and make a prediction	Initial prediction in your own words
5–25	Foundations and library examples	Values, explanations, revised predictions
25–35	Trace setup and the main interface	Input → learner function → observation
35–60	Construct and connect	Source, visible checks and runtime evidence
60–80	Implement and challenge the transfer task	Function and a new counterexample
80–90	Retrieve, explain and save	Exit ticket and retained submission

Installation is preparation time. These are planning estimates, not measured completion times. Use the reference primers when a term is unfamiliar; in Unit B retrieve an explanation before re-reading it. Run All checks that the artifact executes. Unfinished student functions deliberately produce NEEDS_WORK. Keep your first attempt before opening answers.

This notebook belongs to the construction edition. Its supplied teaching runtime is embedded, so it can run without the textbook or another notebook. Where code uses REFERENCE_LESSON, that is the frozen runtime exercise identifier; the reader-facing chapter and saved unit identifiers use the current edition. Building against a supplied runtime is not proof that you have constructed all of its dependencies.

Run the self-contained setup

Open this notebook in Google Colab with the badge above, or use a local Python 3.12+ Jupyter kernel, with Pydantic 2. If needed, run %pip install "pydantic==2.13.4" once in a separate cell and restart the kernel. Package installation needs internet; the lesson itself needs no repository download, API key or prior notebook. The closing extension can use an OpenAI key from Colab's Secrets pane to put a live model behind the same tools; without a key it replays a recorded transcript and says so.

The collapsed cell contains 90 frozen teaching files. Base85 represents compressed bytes as text; zlib decompresses them; SHA-256 checks that the decoded files match this edition. These are supplied packaging operations, not learner algorithms. tempfile creates an isolated working copy; Path handles file locations; sys.path tells Python where the supplied modules live; a .pth file, the mechanism an editable install uses, tells reviewed local subprocesses the same. The code is available for inspection below and performs no package installation itself. The subsequent lesson teaches the libraries used by the mechanisms you will implement.

Run setup on every fresh kernel. It writes scratch runtime files separately from your retained practical-work/ch03-a folder. Rerunning setup restores the frozen support files and keeps your saved work. Restarting a kernel clears variables, not saved submission files. Source basis: Sovereign Agent 444c5f6. Some tasks use reviewed local subprocesses; they are not an OS sandbox.

Supplied offline setup and teaching files

The complete supplied setup cell is in the downloadable notebook. Run it there before attempting the cells below. This web edition omits only that compressed setup payload.

Commit to a prediction before the examples

python

prediction_notes = {
    "prediction": "Write the expected behavior before running the worked example.",
    "reason": "Name the input and rule behind that prediction.",
    "falsifier": "Name an observation that would prove the explanation wrong.",
    "revision": "After execution, explain what changed in your understanding.",
}

Reading the Python vocabulary used in this notebook

You need basic assignments, if, loops, functions, lists and dictionaries. The less familiar features used by the supplied code are introduced here. A library is reusable code that Python can import. The standard library ships with Python; Pydantic is an additional package. An import makes a name available, but does not mean that you have completed the exercise.

JSON is text for exchanging structured values. A Python dictionary is an in-memory object; the JSON representation is a string. Use json.dumps to encode and json.loads to decode. Decoding proves that text has valid JSON syntax, not that its fields match our business contract. Predict which of the following two decoded objects could describe a stock count.

python

import json

intro_data = {"sku": "MANGO", "count": 3}
intro_text = json.dumps(intro_data, sort_keys=True)
print(type(intro_data).__name__, type(intro_text).__name__, intro_text)
print(json.loads(intro_text))
print("Also valid JSON:", json.loads('["not", "a", "stock", "record"]'))
assert json.loads(intro_text) == intro_data

The first result is a dictionary; the second is a list. Before indexing a decoded object, check the shape that your function promises to accept. An exception interrupts the normal path. raise ValueError(...) refuses an invalid value; try/except lets a caller inspect that expected refusal. Catch the expected class, rather than turning every programming error into apparent success. finally runs cleanup even when an earlier operation raises.

An annotation, such as count: int, documents the expected type. It does not by itself enforce the type at runtime. A class defines a kind of object; an instance holds one object's data. @dataclass asks Python to generate routine construction and comparison methods from annotated fields. frozen=True prevents ordinary reassignment of the instance's fields; it does not make every object nested inside those fields immutable. A method is a function attached to a class; self refers to the instance receiving the call.

python

from dataclasses import dataclass


@dataclass(frozen=True)
class IntroObservation:
    operation: str
    count: int


intro_observation = IntroObservation("count-mango", 3)
print(intro_observation.operation, intro_observation.count)
assert intro_observation == IntroObservation("count-mango", 3)

A callback is a function passed to another function. This is how the classroom harness invokes your implementation. The argument candidate below is a function object; parentheses perform the call. Predict the two answers before execution, then trace the result to the callback.

python

def intro_apply(candidate, value):
    return {"input": value, "observed": candidate(value)}


def intro_double(value):
    return value * 2


print(intro_apply(intro_double, 3))
print(intro_apply(lambda value: value + 2, 3))
assert intro_apply(intro_double, 3)["observed"] == 6

lambda value: value + 2 is a small anonymous function. A closure is a function that retains access to values from its surrounding scope. It can bind a tool to a shop snapshot. A shallow copy duplicates only the outer container; copy.deepcopy also copies nested containers used in these fixtures. A set stores distinct values; required <= allowed asks whether every required item is allowed. frozenset is the corresponding immutable set. A tuple groups ordered values; (value,) is a one-item tuple, including the comma.

Paths and cleanup. Path represents a filesystem location. path / "file.json" constructs a child path; read_text and write_text read and write text. A context manager, used with with, manages entry and exit. A temporary-directory context removes its contents on exit. Save your submission outside temporary runtime directories. Reopening a file is different from reusing a Python variable: the former tests retained bytes, while the latter only tests this kernel.

python

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as intro_folder:
    intro_path = Path(intro_folder) / "observation.json"
    intro_path.write_text(json.dumps(intro_data), encoding="utf-8")
    intro_reopened = json.loads(intro_path.read_text(encoding="utf-8"))
    assert intro_reopened == intro_data
    print("Read from a file:", intro_reopened)

Retrieval check: explain JSON versus a dictionary, annotation versus validation, class versus instance, and defining a callback versus invoking it. Change the callback above so an incorrect implementation visibly changes the observed output. This distinction will matter when grading your connected work. Reference: Python's JSON, dataclasses, and pathlib documentation.

Pydantic: turn an input dictionary into a checked object

Pydantic is an additional Python library for validating data. Its model is a class describing fields, not a neural network. Inherit from BaseModel, declare annotated fields, then call model_validate on incoming data. A field without a default is required. A field with a default can be omitted. The result is an instance whose values you read with dot notation.

python

from pydantic import BaseModel, ConfigDict, Field, ValidationError


class IntroCourseRequest(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    name: str = Field(min_length=1)
    quantity: int = Field(gt=0, le=1000)
    note: str = ""


intro_request = IntroCourseRequest.model_validate({"name": "mango", "quantity": 4})
print(intro_request.name, intro_request.quantity, repr(intro_request.note))
assert intro_request.note == ""

The annotation says the field's type. Field supplies constraints: gt=0 means greater than zero, le=1000 means at most 1000, and min_length=1 excludes an empty name. ConfigDict sets model-wide behavior. strict=True rejects conversions for this integer field, including "4", 4.0 and True; extra="forbid" rejects undeclared keys. Pydantic can otherwise convert some compatible inputs, so choose this boundary deliberately rather than assuming every accepted input arrived in the expected type.

Predict which rule refuses each payload. ValidationError reports a failed contract. Its errors() entries contain loc, the field location, and type, the failure category. Catching that expected exception lets the notebook inspect the failure and continue.

python

intro_bad_requests = [
    {"name": "mango", "quantity": "4"},
    {"name": "mango", "quantity": True},
    {"name": "", "quantity": 4},
    {"name": "mango", "quantity": 0},
    {"name": "mango", "quantity": 4, "approved": True},
    {"quantity": 4},
]
for intro_bad_request in intro_bad_requests:
    try:
        IntroCourseRequest.model_validate(intro_bad_request)
    except ValidationError as intro_error:
        print([(item["loc"], item["type"]) for item in intro_error.errors(include_input=False)])
    else:
        raise AssertionError("An invalid input crossed the declared contract")

Use model_dump() for a Python dictionary, model_dump_json() for JSON text, and model_validate_json() to parse and validate JSON. model_json_schema() describes the contract; it is neither an instance's current values nor an invocation of the business handler.

python

intro_serialized = intro_request.model_dump_json()
intro_schema = IntroCourseRequest.model_json_schema()
assert IntroCourseRequest.model_validate_json(intro_serialized) == intro_request
print("Actual values:", intro_request.model_dump())
print("Quantity contract:", intro_schema["properties"]["quantity"])
assert intro_schema["properties"]["quantity"]["exclusiveMinimum"] == 0

Four is valid input to this schema even if the shop needs six. Pydantic checks the declared shape and constraints; the handler still needs authoritative stock, price and permission. Ordinary assignments to an existing instance are not automatically revalidated unless configured for assignment validation. This lesson validates new input at the boundary and uses the resulting values. Explain these limits before relying on a model object in a transaction or tool call.

Chapter 2's full introduction expands this pattern with a separate data-repair checkpoint. This notebook contains the required pattern here so prior Pydantic experience is not needed. References: models, fields, and strict mode.

A loop is a repeated decision with a stopping argument

One tool call cannot answer every request. Lucy's agent may first inspect stock, then ask for prices, then prepare drafts, and finally explain what it found. An agent loop repeatedly obtains a model turn, executes admitted tool requests, appends their observations and decides whether another turn is allowed. It is ordinary control flow surrounding a model interface.

A transcript is the ordered sequence of messages. A tool request carries a call identifier; the tool observation repeats it so a later reader can connect the result to the request. A state machine is a description of allowed states and transitions. Our loop starts running and eventually reaches a terminal reason such as completed, call limit or model failure. Terminal means this episode stops; it does not mean every requested business action succeeded.

Work one iteration by hand

The following replay has two tool requests and a final answer. iter creates an iterator; next consumes its next item. A real provider could return different turns, but the program's responsibility to retain observations and enforce limits is unchanged. Predict the role sequence.

python

intro_turns = [
    {"call_id": "stock-1", "tool": "stock"},
    {"call_id": "price-1", "tool": "price"},
    {"answer": "A draft is ready for review."},
]
intro_transcript = []
intro_tool_values = {"stock": 6, "price": 250}
for intro_turn in intro_turns:
    intro_transcript.append({"role": "assistant", "content": intro_turn})
    if "answer" in intro_turn:
        break
    intro_transcript.append(
        {
            "role": "tool",
            "tool_call_id": intro_turn["call_id"],
            "value": intro_tool_values[intro_turn["tool"]],
        }
    )
print([item["role"] for item in intro_transcript])
assert [item["role"] for item in intro_transcript] == [
    "assistant",
    "tool",
    "assistant",
    "tool",
    "assistant",
]

An assistant message requesting a tool is different from the tool observation. Do not replace the observation with an assistant's claim that a tool succeeded. The real exercise connects its admission function to the Chapter 2 dispatcher and retains actual handler results.

Bound the next attempt, not just the previous result

Suppose each model attempt has a configured exposure of two cents and the budget is six. With four already charged, one more is admitted; with five already charged, it is refused. Equality at the boundary is allowed. Exposure is an estimate used for admission, not an invoice. A failed admitted attempt still consumes one call and its configured exposure. Otherwise a repeatedly failing provider appears free and can evade the bound.

python

intro_spent = 0
intro_attempts = 0
intro_events = []
while intro_attempts < 3 and intro_spent + 2 <= 6:
    intro_attempts += 1
    intro_spent += 2
    try:
        raise RuntimeError("authored provider failure")
    except RuntimeError:
        intro_events.append((intro_attempts, intro_spent, "failed"))
print(intro_events)
assert intro_events == [(1, 2, "failed"), (2, 4, "failed"), (3, 6, "failed")]

Move charging after the raised error in a copy of this example and predict the consequence before trying it with a fixed three-iteration outer bound. Keep that outer bound so the experiment cannot become an accidental infinite loop. This is the failure investigated in Unit B: accounting belongs at admission, before the provider attempt can fail.

Multiple limits and deterministic precedence

A call-count limit bounds the number of provider attempts. A cost-exposure limit bounds their configured cumulative estimate. A tool-call limit bounds a different operation and must not be confused with model turns: one model turn may request multiple tools. If two refusal reasons apply, the implementation needs a declared precedence so the same input yields an explainable result. This lesson checks call count before cost exposure.

The dataclass ModelTurn packages a content string and a tuple of calls. ReplayModel.complete returns the next authored turn. These are test doubles implementing the same small interface that the loop expects from a provider. They are not trained models. Limits supplies configured bounds. Your function decides admission; the surrounding loop records counters and invokes it.

Before coding, draw a trace with columns for attempt number, exposure before, next cost, admission decision, provider outcome and exposure after. Include zero budget, exact fit and first-attempt failure. Then explain why a terminal status without counters is insufficient evidence for a bounded loop. The transfer changes costs and call identifiers, so memorizing the first transcript will not solve it.

Understand the supplied execution interface

The course runtime is provided so your implementation can be connected to real callers and storage. SourceTask(ROOT, chapter) makes a private copy. install(source) replaces only the declared function; visible() invokes the real chapter probe; save(path, result) retains a successful implementation and its evidence. load(path) checks the saved identities and hashes. inject_failure() changes the declared boundary; repair(fragment) replaces that broken fragment. close() removes the scratch copy after you retain evidence. These methods are supplied harness operations, not additional packages you must discover or install.

RuntimeLab provides the same copied-source failure experiment without the complete-function construction layer. Its run method records exit status, observations and the compared expectation. A subprocess log from an unfinished learner implementation is feedback about that implementation; it is not a successful connection. A syntax error in the notebook cell itself is a separate issue to fix. The task below names which interface it uses.

For direct-function units, the visible driver calls your callback without installing a source string. In either case, trace where your code is invoked. Supplied fixtures, database wrappers and replay models are labeled infrastructure; your own implementation and changed-case explanation are the evidence of learning.

Main practical: construct, connect and challenge

Lucy asks what needs ordering. One model turn requests stock, another requests two drafts, and a final turn explains the results. You will implement the admission decision that governs every model call, connect it to a real Chapter 2 tool dispatcher, and retain the transcript for Unit B.

The setup above supplies the required source; it runs on Google Colab or any Python 3.12+ kernel. It uses authored model turns and makes no network call or purchase.

1. Locate the cumulative code

python

import copy
import json
import os
import runpy
import sys
from dataclasses import dataclass
from pathlib import Path

assert sys.version_info >= (3, 12)

ROOT = COURSE_ROOT


def run_book(relative):
    previous = Path.cwd()
    try:
        os.chdir(ROOT)
        return runpy.run_path(str(ROOT / relative))
    finally:
        os.chdir(previous)


chapter2 = run_book("book/always_on/learner/ch02.py")
ToolCall = chapter2["ToolCall"]
dispatcher = chapter2["build_tools"](chapter2["SHOP"])
print("tools", [schema["function"]["name"] for schema in dispatcher.schemas()])

Predict the transcript roles for stock lookup, two draft calls and a final answer. A tool observation must retain the request's call identifier.

2. Represent authored model turns

python

class ModelError(RuntimeError):
    pass


@dataclass(frozen=True)
class ModelTurn:
    content: str = ""
    calls: tuple = ()

    def message(self):
        message = {"role": "assistant", "content": self.content}
        if self.calls:
            message["tool_calls"] = [
                {
                    "id": call.id,
                    "type": "function",
                    "function": {
                        "name": call.name,
                        "arguments": json.dumps(call.arguments, sort_keys=True),
                    },
                }
                for call in self.calls
            ]
        return message


class ReplayModel:
    def __init__(self, turns):
        self.turns = iter(turns)

    def complete(self, messages, tools):
        try:
            return next(self.turns)
        except StopIteration:
            raise ModelError("fixture exhausted") from None


OPENING_TURNS = [
    ModelTurn(calls=(ToolCall(id="stock-1", name="list_stock", arguments={}),)),
    ModelTurn(
        calls=(
            ToolCall(
                id="draft-v", name="draft_order", arguments={"sku": "SKU-VANILLA", "quantity": 6}
            ),
            ToolCall(
                id="draft-s", name="draft_order", arguments={"sku": "SKU-STRAWBERRY", "quantity": 4}
            ),
        )
    ),
    ModelTurn("Drafts total 2600 cents USD. No purchase was made."),
]

3. Construct model-call admission

The starter always admits another call. Repair it so call count and configured estimated exposure both stop the loop before another provider attempt. Check the call limit first. Inputs are already validated nonnegative integers.

python

def decide_admission(case):
    """Return CALL, MODEL_CALL_LIMIT, or MODEL_COST_LIMIT."""
    return "CALL"

Hint 1: the decision

Admission asks whether the next call may begin. Used attempts never disappear because the previous provider failed.

Hint 2: the evidence

Inspect used_calls, max_calls, spent, next_cost, and budget. Equality at the money boundary is allowed; exceeding it is not.

Hint 3: the structure

Return MODEL_CALL_LIMIT when used_calls >= max_calls; otherwise return MODEL_COST_LIMIT when spent + next_cost > budget; otherwise return CALL.

python

VISIBLE_CASES = [
    ({"used_calls": 0, "max_calls": 3, "spent": 0, "next_cost": 2, "budget": 6}, "CALL"),
    (
        {"used_calls": 3, "max_calls": 3, "spent": 0, "next_cost": 0, "budget": 6},
        "MODEL_CALL_LIMIT",
    ),
    (
        {"used_calls": 1, "max_calls": 3, "spent": 5, "next_cost": 2, "budget": 6},
        "MODEL_COST_LIMIT",
    ),
    ({"used_calls": 1, "max_calls": 3, "spent": 4, "next_cost": 2, "budget": 6}, "CALL"),
]


def grade_admission(candidate, cases):
    rows = []
    for number, (case, expected) in enumerate(cases, 1):
        supplied = copy.deepcopy(case)
        try:
            observed = candidate(supplied)
        except Exception as error:
            observed = type(error).__name__
        rows.append(
            {
                "case": number,
                "expected": expected,
                "observed": observed,
                "status": "PASS" if observed == expected and supplied == case else "FAIL",
            }
        )
    return rows


visible_results = grade_admission(decide_admission, VISIBLE_CASES)
VISIBLE_PASSED = all(row["status"] == "PASS" for row in visible_results)
print(json.dumps(visible_results, indent=2))
print("VISIBLE_CONTRACT", "PASSED" if VISIBLE_PASSED else "NEEDS_WORK")

4. Connect admission to the loop

The loop calls your decision before every model attempt. It increments calls and estimated exposure before invoking the provider, preserves assistant tool requests, invokes the real Chapter 2 dispatcher, and appends identified observations.

python

def run_loop(
    model, tool_dispatcher, initial_messages, admission, *, max_calls=4, call_cost=2, budget=8
):
    transcript = copy.deepcopy(initial_messages)
    model_calls = 0
    tool_calls = 0
    spent = 0
    seen = set()
    while True:
        decision = admission(
            {
                "used_calls": model_calls,
                "max_calls": max_calls,
                "spent": spent,
                "next_cost": call_cost,
                "budget": budget,
            }
        )
        if decision != "CALL":
            return {
                "status": decision,
                "messages": transcript,
                "model_calls": model_calls,
                "tool_calls": tool_calls,
                "estimated_cents": spent,
            }
        model_calls += 1
        spent += call_cost
        try:
            turn = model.complete(copy.deepcopy(transcript), tool_dispatcher.schemas())
        except ModelError:
            return {
                "status": "MODEL_FAILED",
                "messages": transcript,
                "model_calls": model_calls,
                "tool_calls": tool_calls,
                "estimated_cents": spent,
            }
        identifiers = [call.id for call in turn.calls]
        if len(identifiers) != len(set(identifiers)) or seen.intersection(identifiers):
            return {
                "status": "REPEATED_CALL_ID",
                "messages": transcript,
                "model_calls": model_calls,
                "tool_calls": tool_calls,
                "estimated_cents": spent,
            }
        seen.update(identifiers)
        transcript.append(turn.message())
        if not turn.calls:
            return {
                "status": "COMPLETED" if turn.content.strip() else "EMPTY_REPLY",
                "messages": transcript,
                "model_calls": model_calls,
                "tool_calls": tool_calls,
                "estimated_cents": spent,
                "answer": turn.content,
            }
        for call in turn.calls:
            tool_calls += 1
            result = tool_dispatcher.invoke(call)
            transcript.append(
                {
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": json.dumps(result, sort_keys=True),
                }
            )


INITIAL_MESSAGES = [
    {"role": "system", "content": "Use stock tools to prepare drafts. Never purchase."},
    {"role": "user", "content": "What needs ordering?"},
]
connected = None
if VISIBLE_PASSED:
    connected = run_loop(
        ReplayModel(OPENING_TURNS),
        dispatcher,
        INITIAL_MESSAGES,
        decide_admission,
    )
    print(connected["status"], connected["model_calls"], connected["tool_calls"])
    print([message["role"] for message in connected["messages"]])
    assert connected["status"] == "COMPLETED"
    assert (connected["model_calls"], connected["tool_calls"]) == (3, 3)
else:
    print("CONNECTION_NOT_READY — repair decide_admission, then run again.")

Trace one tool_call_id from assistant request to tool observation. Final prose alone does not prove the two draft tools returned valid results.

5. Save the bounded transcript

python

ARTIFACT_PATH = Path("ch03-unit-a-handoff-v1.json")
artifact_status = "NOT_WRITTEN"
if connected is not None:
    encoded = json.dumps(connected, indent=2, sort_keys=True) + "\n"
    ARTIFACT_PATH.write_text(encoded, encoding="utf-8")
    artifact_status = "WRITTEN"
    print(ARTIFACT_PATH)
else:
    print("HANDOFF_NOT_WRITTEN")

Exit ticket

Explain why failed provider attempts count, why call-limit precedence matters when two limits are exhausted, and which transcript observation proves each draft tool actually ran.

python

exercise_report = {
    "unit": "ch03-a",
    "attempted": 1,
    "completed": int(VISIBLE_PASSED),
    "failed": int(not VISIBLE_PASSED),
    "skipped": 0,
    "connection": "PASSED" if connected else "NOT_READY",
    "handoff": artifact_status,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))

Changed-constraint construction: Admit a variable-cost next model attempt

Allow twenty minutes. Spend three minutes predicting, ten implementing and tracing, five on a new case of your own, and two explaining the surviving limitation. This is dedicated work, not an invitation to run a supplied answer. Both units revisit the same invariant after different core experiences; in Unit B, attempt this task from memory before consulting Unit A.

Implement transfer_check(state). Fields calls, max_calls, spent, next_cost and budget are validated nonnegative integers. Return CALL_LIMIT first when calls >= max_calls, otherwise COST_LIMIT when spent + next_cost > budget, otherwise CALL. Equality at the money boundary is allowed. The next-cost estimate can differ on every attempt.

Write your expected values before running the table. Keep one accepted case and one refusal. Your function is passed directly into the driver below. The driver copies inputs and checks they remain unchanged; it does not replace your implementation with the reference answer.

Hint 1: identify the authoritative inputs Name the source field for each output value. Which input changes while the rule remains the same?

Hint 2: choose the boundary cases Start with exact empty, exact equality and one value on each side of the boundary where valid. Do not add a special case for a visible product name or operation identity.

python

def transfer_check(state):
    raise NotImplementedError("Decide before admitting the next attempt")

python

import json

TRANSFER_CASES = [
    ("exact fit", [{"calls": 1, "max_calls": 3, "spent": 4, "next_cost": 3, "budget": 7}], "CALL"),
    (
        "next attempt too costly",
        [{"calls": 1, "max_calls": 3, "spent": 4, "next_cost": 4, "budget": 7}],
        "COST_LIMIT",
    ),
    (
        "two exhausted limits",
        [{"calls": 3, "max_calls": 3, "spent": 7, "next_cost": 1, "budget": 7}],
        "CALL_LIMIT",
    ),
    (
        "zero allowed attempts",
        [{"calls": 0, "max_calls": 0, "spent": 0, "next_cost": 0, "budget": 0}],
        "CALL_LIMIT",
    ),
]


def same_transfer_value(actual, expected):
    if type(actual) is not type(expected):
        return False
    if isinstance(expected, dict):
        return actual.keys() == expected.keys() and all(
            same_transfer_value(actual[key], value) for key, value in expected.items()
        )
    if isinstance(expected, list):
        return len(actual) == len(expected) and all(
            same_transfer_value(a, e) for a, e in zip(actual, expected, strict=True)
        )
    return actual == expected


def run_transfer(candidate, cases):
    observations = []
    for label, arguments, expected in cases:
        supplied = copy.deepcopy(arguments)
        before = copy.deepcopy(supplied)
        raised = None
        try:
            actual = candidate(*supplied)
        except NotImplementedError:
            raised = "NotImplementedError"
            actual = {"unfinished": True}
        except Exception as error:
            raised = type(error).__name__
            actual = {"raises": raised}
        expects_error = isinstance(expected, dict) and set(expected) == {"raises"}
        correct = (
            raised == expected["raises"]
            if expects_error
            else (raised is None and same_transfer_value(actual, expected))
        )
        passed = correct and same_transfer_value(supplied, before)
        observations.append(
            {"case": label, "expected": expected, "observed": actual, "passed": passed}
        )
        print("PASS" if passed else "NEEDS_WORK", label, "expected", expected, "observed", actual)
    return observations


transfer_observations = run_transfer(transfer_check, TRANSFER_CASES)
TRANSFER_PASSED = all(row["passed"] for row in transfer_observations)
print("TRANSFER_STATUS", "PASS" if TRANSFER_PASSED else "NEEDS_WORK")

Design a counterexample and retrieve the mechanism

Add one new case with an independently calculated expected outcome to TRANSFER_CASES and rerun the driver. Change one condition at a time. Then deliberately replace your candidate with a constant answer in a temporary copy and show a case that rejects it. Restore your implementation. Explain why that counterexample is stronger than repeating the original example with a new name.

Without viewing the worked example, write the invariant in words and trace one observed value back to its input. Identify which part is a local fixture result and which claim would need a live provider, host or external-system observation. Keep a first attempt even if you used a hint.

Save your evidence and explain the result

Fill the prediction notes and your explanation before saving. Include the exact observed value, the input or retained row that caused it, your code's invocation point, one failed hypothesis, and the strongest claim the evidence still cannot support. A completed code cell alone does not earn explanation credit. Do not label reference-start behavior as your own Unit A construction.

Keep this edited notebook, the Markdown if used for notes, saved handoff files, and the JSON record below. Your work folder survives scratch cleanup and can be reopened in a new kernel. An instructor can ask for an unseen case after the visible checks; keep your implementation general.

python

explanation_notes = {
    "causal_trace": "Explain the input, learner invocation and observed result.",
    "failed_hypothesis": "Describe a prediction the evidence changed.",
    "remaining_limit": "Name the guarantee not established by this experiment.",
}
course_submission = {
    "unit": "ch03-a",
    "planned_minutes": 90,
    "starting_evidence": globals().get("HANDOFF_ORIGIN", "INDEPENDENT_UNIT_A"),
    "prediction": prediction_notes,
    "explanation": explanation_notes,
    "core_report": exercise_report,
    "transfer": transfer_observations,
    "explanation_review": "HUMAN_REVIEW_REQUIRED",
}
submission_path = COURSE_WORK / "ch03-a-submission-v1.json"
submission_path.write_text(
    json.dumps(course_submission, indent=2, sort_keys=True), encoding="utf-8"
)
print("Saved evidence:", submission_path)
print(
    "COURSE_REPORT="
    + json.dumps(
        {
            "unit": "ch03-a",
            "transfer_passed": TRANSFER_PASSED,
            "starting_evidence": course_submission["starting_evidence"],
            "edition": "student",
        },
        sort_keys=True,
    )
)

Extension: put a language model behind your bounded loop

Every turn the loop above consumed was authored: ReplayModel handed back turns you wrote, and the admission decision you constructed counted them against Limits. Admission runs before every attempt because, in production, the model is a metered service that keeps asking for tools for as long as you keep paying. This closing section puts a real model behind the same loop and makes the budget visible.

The loop is the runtime's agent_loop.run_loop, the same construction as yours: call limit first, then estimated cost exposure, then the attempt; a turn's tool calls are counted against tool_calls before any of them runs; every observation is appended under its tool_call_id. The tools are the Chapter 2 shop tools, rebuilt on the runtime's Dispatcher with one rule added in the chapter's own shape. Drafts are calculations, not purchases, but they add up, so draft_order is marked consequential and its before_write gate admits a draft only while reserved cents plus this draft's cost stays within a draft budget, equality allowed. That is your transfer task's decision, applied to money instead of model calls.

llm_lab.py supplies two models for the loop's Model protocol. ScriptedModel replays turns written down here, so the section runs identically for everyone; OpenAIModel calls a hosted model when a key is present. The loop cannot tell them apart; the record says which one ran.

Prediction: the next cell prints four schemas and the limits. Write down which tool is consequential, what the vanilla and strawberry drafts cost together from PRICES, and whether MODEL_COST_LIMIT can fire before MODEL_CALL_LIMIT at 2 cents per call against 20 cents.

python

import contextlib
import json
from runpy import run_path

from sovereign_agent.agent_loop import Limits, run_loop
from sovereign_agent.files import write_json
from sovereign_agent.llm_lab import ScriptedModel, describe, scripted_turn, tool_trace
from sovereign_agent.tool_dispatch import Dispatcher, ExecutableTool

# The learner module loads the Chapter 2 tools relative to the runtime root.
with contextlib.chdir(COURSE_ROOT):
    chapter_three = run_path(str(COURSE_ROOT / "book/always_on/learner/ch03.py"))
shop_tools = chapter_three["shop_tools"]
SHOP, PRICES = shop_tools["SHOP"], shop_tools["PRICES"]
DRAFT_BUDGET_CENTS = 2500
BASE_LIMITS = {
    "model_calls": 8,
    "tool_calls": 16,
    "seconds": 120,
    "output_tokens": 4096,
    "total_output_tokens": 16_384,
    "estimated_call_cents": 2,
    "model_budget_cents": 20,
}
LOOP_LIMITS = Limits(**BASE_LIMITS)


def fresh_drafts_path(label):
    """The drafts file is the only durable state here; start every run without one."""
    path = COURSE_WORK / f"ch03-a-llm-drafts-{label}.json"
    path.unlink(missing_ok=True)
    return path


def build_loop_tools(shop, drafts_path):
    """The Chapter 2 handlers, unchanged, behind a draft budget in the chapter's shape."""
    reference = shop_tools["build_tools"](shop)
    handlers = {name: tool.handler for name, tool in reference.tools.items()}
    ledger = {"reserved_cents": 0, "drafts": [], "refusals": []}

    def budget(_):
        reserved = ledger["reserved_cents"]
        return {
            "budget_cents": DRAFT_BUDGET_CENTS,
            "reserved_cents": reserved,
            "remaining_cents": DRAFT_BUDGET_CENTS - reserved,
        }

    def admit_draft(call):
        # Admission asks whether this draft may begin; equality at the boundary is allowed.
        cost = PRICES.get(call.arguments["sku"], 0) * call.arguments["quantity"]
        exposure = ledger["reserved_cents"] + cost
        if exposure > DRAFT_BUDGET_CENTS:
            ledger["refusals"].append(
                {"call": call.id, "reason": "DRAFT_COST_LIMIT", "exposure_cents": exposure}
            )
            raise ValueError("draft would exceed the draft budget")

    def draft(args):
        value = handlers["draft_order"](args)  # quantity must equal need, or ValueError
        ledger["reserved_cents"] += value["total_cents"]
        ledger["drafts"].append(value)
        write_json(drafts_path, budget(None) | {"drafts": ledger["drafts"]})
        return value

    no_arguments, product, draft_arguments = (
        shop_tools["NoArguments"],
        shop_tools["ProductArguments"],
        shop_tools["DraftArguments"],
    )
    registered = [
        ExecutableTool("list_stock", "Read stock and need.", no_arguments, handlers["list_stock"]),
        ExecutableTool("supplier", "Read supplier price in cents.", product, handlers["supplier"]),
        ExecutableTool("draft_budget", "Read the draft budget and reserved.", no_arguments, budget),
        ExecutableTool(
            "draft_order",
            "Calculate a draft against the draft budget; never purchases.",
            draft_arguments,
            draft,
            consequential=True,
        ),
    ]
    allowed = frozenset(tool.name for tool in registered)
    return Dispatcher(registered, allowed=allowed, before_write=admit_draft), ledger


scripted_drafts_path = fresh_drafts_path("scripted")
loop_dispatcher, loop_ledger = build_loop_tools(SHOP, scripted_drafts_path)
for tool_schema in loop_dispatcher.schemas():
    function = tool_schema["function"]
    print(function["name"], "requires", function["parameters"].get("required", []))
    print("   ", function["description"])
shop_need = {row["sku"]: max(0, row["reorder_point"] - row["on_hand"]) for row in SHOP["products"]}
print("Need per product:", shop_need, "| prices:", PRICES)
print("Draft budget:", DRAFT_BUDGET_CENTS, "cents | loop limits:", LOOP_LIMITS)
LOOP_MESSAGES = [
    {
        "role": "system",
        "content": (
            "Help Lucy prepare replenishment drafts. First call list_stock. For each product "
            "with needed > 0, call draft_order with exactly that quantity; do not draft products "
            "with needed = 0. Drafts count against a draft budget in USD cents. If a draft is "
            "refused, call draft_budget and say what could not be drafted. Drafts are proposals, "
            "never purchases. Finish with a short summary in USD cents."
        ),
    },
    {
        "role": "user",
        "content": "Prepare replenishment drafts from current stock. State USD amounts.",
    },
]

The four schemas are the Chapter 2 argument classes, strict=True and extra="forbid" included, in the {"type": "function", ...} envelope the API expects. LOOP_LIMITS is what your admission decision reads: eight model calls, sixteen tool calls, and 2 cents of estimated exposure per call against a 20 cents budget. Eight calls cost 16 cents, so with these numbers the call limit always wins; the challenge below changes that.

Run the loop against a recorded transcript

The turns below are what a model replied, written down. Turn one reads stock. Turn two asks for both supplier prices. Turn three proposes the vanilla draft (6 at 250 cents) and the strawberry draft (4 at 275 cents): 1500 plus 1100 is 2600, and the budget is 2500. Turn four reads the budget to find out why. Turn five answers, which ends the loop.

Prediction: write down the status, the model-call count, the tool-call count and the estimated exposure in cents. Which draft is refused, and what does the drafts file hold? Recall that Dispatcher.invoke flattens every refusal to tool_failed, so the gate keeps its own ledger with the exposure that was refused.

The cell then runs your Chapter 3 run_loop on the same script, with the same dispatcher and limits. The runtime's loop and yours must agree on every count and every observation; if they do not, the disagreement is a finding about one of the two loops, not about the model.

python

RECORDED_TURNS = [
    scripted_turn(calls=[{"name": "list_stock", "arguments": {}}]),
    scripted_turn(
        calls=[
            {"name": "supplier", "arguments": {"sku": "SKU-VANILLA"}},
            {"name": "supplier", "arguments": {"sku": "SKU-STRAWBERRY"}},
        ]
    ),
    scripted_turn(
        calls=[
            {"name": "draft_order", "arguments": {"sku": "SKU-VANILLA", "quantity": 6}},
            {"name": "draft_order", "arguments": {"sku": "SKU-STRAWBERRY", "quantity": 4}},
        ]
    ),
    scripted_turn(calls=[{"name": "draft_budget", "arguments": {}}]),
    scripted_turn(
        "Drafted 6 vanilla at 250 cents, 1500 cents in total. Strawberry needs 4 at 275 cents, "
        "1100 cents, which "
        "would take drafts to 2600 cents against a 2500 cents budget, so it was refused and needs "
        "Lucy's "
        "decision. Chocolate is above its reorder point. Nothing was purchased."
    ),
]

scripted_result = run_loop(
    ScriptedModel(RECORDED_TURNS), loop_dispatcher, LOOP_MESSAGES, limits=LOOP_LIMITS
)
print(describe(scripted_result, "scripted"))
print("Estimated exposure:", scripted_result.estimated_cost_cents, "cents")
print("Gate ledger:", loop_ledger["refusals"])
scripted_trace = tool_trace(scripted_result)
assert scripted_result.status == "COMPLETED"
assert (scripted_result.model_calls, scripted_result.tool_calls) == (5, 6)
assert scripted_result.estimated_cost_cents == 5 * 2
assert [step["ok"] for step in scripted_trace] == [True, True, True, True, False, True]
assert scripted_trace[4]["result"] == "tool_failed"
assert loop_ledger["refusals"] == [
    {"call": "scripted-3-call-2", "reason": "DRAFT_COST_LIMIT", "exposure_cents": 2600}
]
assert scripted_trace[5]["result"]["remaining_cents"] == DRAFT_BUDGET_CENTS - 1500
scripted_drafts = json.loads(scripted_drafts_path.read_text(encoding="utf-8"))
print("Drafts file:", scripted_drafts)
assert [d["sku"] for d in scripted_drafts["drafts"]] == ["SKU-VANILLA"]
assert scripted_drafts["reserved_cents"] == 1500
# Every observation in the transcript carries the identifier of the request it answers.
assert all(m.get("tool_call_id") for m in scripted_result.messages if m["role"] == "tool")

# Your own Chapter 3 loop, on the same script and a fresh dispatcher, must agree exactly.
own_dispatcher, own_ledger = build_loop_tools(SHOP, fresh_drafts_path("own-loop"))
own_result = chapter_three["run_loop"](
    ScriptedModel(RECORDED_TURNS),
    own_dispatcher,
    LOOP_MESSAGES,
    limits=chapter_three["Limits"](**BASE_LIMITS),
)
own_counts = (own_result.status, own_result.model_calls, own_result.tool_calls)
own_counts += (own_result.estimated_cost_cents,)
print("Your loop:", own_counts)
assert own_counts == ("COMPLETED", 5, 6, 10)
assert tool_trace(own_result) == scripted_trace
assert own_ledger["refusals"] == loop_ledger["refusals"]

The refused draft is an ordinary failed observation; the loop carries on, and the drafts file holds only what the gate admitted. Both loops produced the same six observations under the same identifiers: nothing in the runtime's loop is unexplained by your admission decision.

Run the same loop against a live model

Nothing changes except the model. build_model looks for an OPENAI_API_KEY in Colab's Secrets pane (the key icon in the left sidebar: add a secret with that name and switch on notebook access for it), then in the process environment for a local kernel. Without a key it returns the scripted model again and the record says so. A missing key is a normal condition, never a silent substitution.

The live model is gpt-5.1 through the openai library, which Colab ships preinstalled; on a local kernel run %pip install openai once. One run costs a few thousand tokens. The model receives the same four schemas and the same system message. It may draft strawberry first, skip the supplier, or never read the budget; what it cannot do is draft a quantity the shop does not need, or push reserved drafts past 2500 cents, because those rules live in the handler and the gate. The two drafts together cost 2600 cents, so at most one of them can succeed.

Prediction: write down which draft a careful assistant would make first, and what it should say about the other. Then run the cell and compare the model's judgment, or the recorded transcript when the run is scripted, with your prediction. The loop charges 2 cents for every attempt it admits, whether or not the model used it well.

python

from sovereign_agent.llm_lab import DEFAULT_MODEL, build_model, resolve_api_key

live_drafts_path = fresh_drafts_path("live")
live_dispatcher, live_ledger = build_loop_tools(SHOP, live_drafts_path)
chosen_model = build_model(RECORDED_TURNS, model=DEFAULT_MODEL)
print("Model source:", chosen_model.source, "| key found:", resolve_api_key() is not None)
live_result = run_loop(chosen_model, live_dispatcher, LOOP_MESSAGES, limits=LOOP_LIMITS)
print(describe(live_result, chosen_model.source))
print("Estimated exposure:", live_result.estimated_cost_cents, "cents")
print("Gate ledger:", live_ledger["refusals"])
live_trace = tool_trace(live_result)
assert live_result.status in {
    "COMPLETED",
    "MODEL_CALL_LIMIT",
    "MODEL_COST_LIMIT",
    "TOOL_LIMIT",
    "TOKEN_LIMIT",
    "MODEL_FAILED",
}
# The loop's accounting holds whatever the model did.
assert live_result.model_calls <= LOOP_LIMITS.model_calls
assert live_result.estimated_cost_cents == 2 * live_result.model_calls
assert live_result.estimated_cost_cents <= LOOP_LIMITS.model_budget_cents
assert live_result.tool_calls == sum(1 for m in live_result.messages if m["role"] == "tool")
# The handler's rule and the gate's rule hold whatever the model proposed.
live_drafted = [step for step in live_trace if step["tool"] == "draft_order" and step["ok"]]
for step in live_drafted:
    assert step["arguments"]["quantity"] == shop_need[step["arguments"]["sku"]], step
assert live_ledger["reserved_cents"] == sum(step["result"]["total_cents"] for step in live_drafted)
assert live_ledger["reserved_cents"] <= DRAFT_BUDGET_CENTS
assert all(entry["reason"] == "DRAFT_COST_LIMIT" for entry in live_ledger["refusals"])
assert live_drafts_path.exists() == bool(live_drafted)

Challenge: make each limit end the loop, then check precedence

The admission rule is data. The next cell replays the recorded transcript four times on fresh dispatchers, changing only Limits. Predict the status, model_calls, tool_calls and estimated_cost_cents for each before running, working turn by turn as you did in the Work one iteration by hand section:

model_calls=2: the loop admits two attempts, then stops before the third.
estimated_call_cents=30, model_budget_cents=100: 30, 60, 90 are admitted; is 120?
tool_calls=2: turn two proposes two calls when one is already spent.
model_calls=3 together with the 30 cents estimate: both rules would refuse the fourth attempt. Which status does your transfer task say is checked first?

Note what TOOL_LIMIT leaves behind: the turn that proposed the calls is not appended, so the transcript holds only the observations that were admitted. With a live key, rerun the live cell under model_calls=2 and explain what the model had done by the time it was cut off.

In your notes, separate what the scripted runs prove (the loop, the admission order, the gate, the transcript accounting) from what only a live run can show (the model's judgment). The saved record keeps the two apart.

python

def scripted_run(label, **overrides):
    dispatcher, _ = build_loop_tools(SHOP, fresh_drafts_path(label))
    limits = Limits(**{**BASE_LIMITS, **overrides})
    result = run_loop(ScriptedModel(RECORDED_TURNS), dispatcher, LOOP_MESSAGES, limits=limits)
    counts = (result.status, result.model_calls, result.tool_calls, result.estimated_cost_cents)
    print(label, overrides, "->", counts)
    return result


call_limited = scripted_run("call-limit", model_calls=2)
cost_limited = scripted_run("cost-limit", estimated_call_cents=30, model_budget_cents=100)
tool_limited = scripted_run("tool-limit", tool_calls=2)
precedence = scripted_run(
    "precedence", model_calls=3, estimated_call_cents=30, model_budget_cents=100
)
assert (call_limited.status, call_limited.model_calls, call_limited.tool_calls) == (
    "MODEL_CALL_LIMIT",
    2,
    3,
)
assert (cost_limited.status, cost_limited.model_calls) == ("MODEL_COST_LIMIT", 3)
assert (cost_limited.tool_calls, cost_limited.estimated_cost_cents) == (5, 90)
assert (tool_limited.status, tool_limited.model_calls, tool_limited.tool_calls) == (
    "TOOL_LIMIT",
    2,
    1,
)
assert sum(1 for m in tool_limited.messages if m["role"] == "tool") == 1
assert (precedence.status, precedence.model_calls, precedence.estimated_cost_cents) == (
    "MODEL_CALL_LIMIT",
    3,
    90,
)

llm_lab_report = {
    "unit": "ch03-a",
    "scripted": {
        "status": scripted_result.status,
        "model_calls": scripted_result.model_calls,
        "tool_calls": scripted_result.tool_calls,
        "estimated_cost_cents": scripted_result.estimated_cost_cents,
        "refused": loop_ledger["refusals"],
        "drafted": [d["sku"] for d in scripted_drafts["drafts"]],
        "own_loop_agrees": own_counts == ("COMPLETED", 5, 6, 10),
    },
    "session": {
        "source": chosen_model.source,
        "model": DEFAULT_MODEL if chosen_model.source == "live" else None,
        "status": live_result.status,
        "model_calls": live_result.model_calls,
        "tool_calls": live_result.tool_calls,
        "estimated_cost_cents": live_result.estimated_cost_cents,
        "drafted": [step["arguments"] for step in live_drafted],
        "refused": live_ledger["refusals"],
        "answer": live_result.answer[:400],
    },
    "limits": {
        "call_limit": call_limited.status,
        "cost_limit": cost_limited.status,
        "tool_limit": tool_limited.status,
        "precedence": precedence.status,
    },
}
llm_report_path = COURSE_WORK / "ch03-a-llm-lab-report-v1.json"
llm_report_path.write_text(json.dumps(llm_lab_report, indent=2, sort_keys=True), encoding="utf-8")
print("Saved evidence:", llm_report_path)
print("LLM_LAB_REPORT=" + json.dumps(llm_lab_report["session"], sort_keys=True))

Keep building with Prof Rod

Found this material through a colleague, classroom or shared download? Get the complete book at profrod.ai/book and join the Prof Rod learner community. Bring one result, one question or one failure you learned from. Share this resource with another learner and keep its source links with it so they can find the full course and future updates.