Reliable Shop Tools: 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 2. 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:

1. Explain the chapter's mechanism using a prediction and an observed intermediate result.
2. Construct the complete tool factory: reject duplicate product identities, isolate a copied stock snapshot, expose stock and supplier tools, and make draft quantity equal current need. Preserve strict Pydantic argument validation and USD integer-cents arithmetic.
3. Solve use one reservation rule in reporting and drafting using changed inputs and an independent expectation.
4. Retain your implementation, failed/corrected observations, causal explanation and limits.

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/ch02-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

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.

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.

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.

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.

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.

from pydantic import BaseModel, ConfigDict, Field, ValidationError

Pydantic from first principles · 20-minute guided introduction

No previous Pydantic experience is assumed. We start with familiar Python dictionaries, then build a small contract, inspect failures, and connect that contract to a tool call. Pydantic is a Python library for checking data against declared types and constraints. Here, model means a class describing data; it is not a neural network or language model.

By the end, you should be able to define a model, validate a dictionary, explain a validation error, reject unwanted conversions, and distinguish a schema from a business rule. This complete introduction supports either independent ninety-minute unit. Its small repair checkpoint is preparation for the full tool-factory and transfer exercises.

1. Why annotations alone are not a boundary · minutes 0–2

A caller sends a quantity as a string. A type annotation documents what we expect, but Python does not automatically check that expectation when the function runs. Predict: will this call raise an error, return twelve, or repeat the characters?

def annotated_double(quantity: int) -> int:
    return quantity * 2


incoming_quantity = "6"
observed = annotated_double(incoming_quantity)
print(repr(observed), type(observed).__name__)
assert observed == "66"

The result is '66', a string. The annotation did not become a runtime guard. A type checker can flag this example before execution, but external data still needs a runtime boundary. We will check it before passing it to calculations that assume validated input.

2. A model is a recipe; an instance holds data · minutes 2–5

Read class IntroLine(BaseModel) as “define our data class using Pydantic's validation machinery.” The indented annotations declare fields, the named pieces of data. sku and quantity have no defaults and must be supplied. note has a default, so callers can omit it. str and int are ordinary Python types. No decorators or inheritance theory are needed to use this pattern: declare the fields inside a class that inherits from BaseModel.

The class is the reusable recipe; line is one validated instance. The constructor accepts named arguments. model_validate accepts a dictionary, which is convenient when a tool caller has already produced one. Dot notation reads the instance's fields. Predict the missing note.

class IntroLine(BaseModel):
    sku: str
    quantity: int
    note: str = ""


line = IntroLine(sku="SKU-VANILLA", quantity=6)
incoming = {"sku": "SKU-VANILLA", "quantity": 6}
parsed_line = IntroLine.model_validate(incoming)
print(line.sku, line.quantity, repr(line.note))
print("Same values:", line == parsed_line)
assert line.note == "" and line == parsed_line

3. Read an error as a repair instruction · minutes 5–7

Now omit sku and supply a quantity that cannot be parsed as an integer. Pydantic raises ValidationError instead of returning a usable instance. Catch this expected exception so the notebook can show the failure and continue. Its errors() method returns structured details: loc identifies the field, type classifies the failure, and msg explains it. We hide the raw input in this display; we need the location and rule to understand the failure.

Predict: how many fields need repair? A missing required field is different from a field with an invalid value. Read every reported error, not just the first line.

bad_line = {"quantity": "six"}
try:
    IntroLine.model_validate(bad_line)
except ValidationError as error:
    details = error.errors(include_url=False, include_input=False)
    for detail in details:
        print(detail["loc"], detail["type"], detail["msg"])
    assert {item["loc"] for item in details} == {("sku",), ("quantity",)}
else:
    raise AssertionError("The invalid line was accepted")

Both fields need attention. An exception is useful feedback here, not a notebook setup failure. In the tool dispatcher later, this same exception becomes a refused observation. Validation reports the problem; it does not decide whether to ask the caller to correct it.

4. Accepted input can be converted input · minutes 7–10

By default, Pydantic may convert compatible input into the declared type. That is helpful for a form but can hide what a tool caller actually sent. For an integer field, compare 6, "6", True, 6.0, and "six". Write the two acceptance columns before running. The strict=True argument asks this validation call to reject these integer conversions. This table concerns Python inputs to an integer field; strict handling of JSON dates and other types has additional rules. We do not generalize this table to every type.

class IntroQuantity(BaseModel):
    quantity: int


def inspect_quantity(value, strict):
    try:
        result = IntroQuantity.model_validate({"quantity": value}, strict=strict)
    except ValidationError:
        return "REFUSED"
    return f"{result.quantity!r} ({type(result.quantity).__name__})"


quantity_inputs = [6, "6", True, 6.0, "six"]
for value in quantity_inputs:
    print(
        repr(value),
        "| default:",
        inspect_quantity(value, False),
        "| strict:",
        inspect_quantity(value, True),
    )
assert [inspect_quantity(value, True) != "REFUSED" for value in quantity_inputs] == [
    True,
    False,
    False,
    False,
    False,
]
boolean_quantity = True
print("Python considers bool an int subclass:", isinstance(boolean_quantity, int))
print("Its exact type is int:", type(boolean_quantity) is int)

Default validation converts the first four values to integers. The strict column accepts only 6. Python's boolean/integer relationship explains why a casual isinstance(value, int) check would miss one of these cases. Pydantic's strict integer validation rejects a boolean. Choose the conversion policy deliberately at the boundary.

5. Types, constraints, and model policy do different jobs · minutes 10–13

Our stock-reading contract has three layers:

ge means greater than or equal; gt means strictly greater. le means less than or equal. For strings, min_length and max_length constrain character count. Here Field(...) adds constraints; it does not supply a field value. model_config configures validation; it is not an input field. Pydantic otherwise ignores extra keys by default, so we explicitly forbid them when every accepted argument should belong to the declared contract.

Predict: zero tubs is valid stock. Should negative stock, a string count, or an extra approved key pass this contract? The stock example is different from an order, which will require a positive quantity in Exercise 1.

class IntroStockReading(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    sku: str = Field(min_length=1, max_length=100)
    on_hand: int = Field(ge=0, le=1000)


stock_examples = [
    ("empty shelf", {"sku": "SKU-VANILLA", "on_hand": 0}, True),
    ("negative count", {"sku": "SKU-VANILLA", "on_hand": -1}, False),
    ("string count", {"sku": "SKU-VANILLA", "on_hand": "2"}, False),
    ("extra key", {"sku": "SKU-VANILLA", "on_hand": 2, "approved": True}, False),
]
for label, payload, expected in stock_examples:
    try:
        validated = IntroStockReading.model_validate(payload)
        accepted = True
        print(label, "ACCEPTED", validated.model_dump())
    except ValidationError as error:
        accepted = False
        print(label, "REFUSED", [item["type"] for item in error.errors()])
    assert accepted == expected

6. Dictionary, instance, JSON, schema · minutes 13–16

These objects serve different purposes. A dictionary contains Python data. A model instance provides validated fields. JSON text is a portable string representation. A JSON Schema describes the expected shape so another program can discover the contract.

For these string/integer fields, the round trip preserves the values. Other field types can need different serialization choices. The schema below describes possible stock readings; it does not contain the current reading, validate a caller by itself, or execute a handler. Predict: which output contains the actual count, and which contains its minimum?

reading = IntroStockReading(sku="SKU-VANILLA", on_hand=2)
as_dict = reading.model_dump()
as_json = reading.model_dump_json()
schema_description = IntroStockReading.model_json_schema()
print("Dictionary:", as_dict)
print("JSON string:", repr(as_json))
print("Count description:", schema_description["properties"]["on_hand"])
print("Required fields:", schema_description["required"])
assert isinstance(as_dict, dict) and isinstance(as_json, str)
assert IntroStockReading.model_validate_json(as_json) == reading
assert schema_description["properties"]["on_hand"]["minimum"] == 0
assert schema_description["additionalProperties"] is False

7. Validation is a contract check, not a fact check · minutes 16–17

An invented count of 999 satisfies our type and range rules. Pydantic has no connection to Lucy's shelf and cannot know whether that count is true. We must compare it with an authoritative source separately. Similarly, validating an approved field in some other schema would not establish that a real person authorized a purchase.

The model also does not promise that ordinary assignments to an existing instance are revalidated: assignment validation is a separate configuration option. In this lesson we validate new input at the boundary and then use its fields; we do not use unvalidated updates.

plausible_but_unverified = IntroStockReading(sku="SKU-VANILLA", on_hand=999)
authoritative_count = 2
print("Passes the declared contract:", plausible_but_unverified.on_hand)
print("Matches the shop record:", plausible_but_unverified.on_hand == authoritative_count)
assert plausible_but_unverified.on_hand != authoritative_count

8. Your turn: repair the data without weakening the contract · minutes 17–20

The authoritative record says SKU SKU-VANILLA has 2 tubs. Repair repaired_stock in the next cell so it represents that record and passes IntroStockReading. Keep the model unchanged. Before running, name all three problems with the supplied dictionary. Explain why changing a type annotation to Any would hide a problem rather than repair this data.

Hint 1: inspect the required names and each value's Python type. Hint 2: an empty string is still a string, but the length constraint matters. Hint 3: the data contract has exactly two fields. A proposed approval is not stock data.

repaired_stock = {"sku": "", "on_hand": "2", "approved": True}

try:
    repaired_reading = IntroStockReading.model_validate(repaired_stock)
except ValidationError as error:
    pydantic_checkpoint = False
    for detail in error.errors(include_url=False, include_input=False):
        print(detail["loc"], detail["msg"])
else:
    pydantic_checkpoint = repaired_reading.model_dump() == {
        "sku": "SKU-VANILLA",
        "on_hand": 2,
    }
print("PASS" if pydantic_checkpoint else "NEEDS_WORK", "Pydantic data-repair checkpoint")

Retrieve the mechanism before continuing

Close the examples and explain: (1) what makes a field required, (2) why "6" can pass one integer model and fail another, (3) what loc tells you, (4) how a schema differs from an instance, and (5) why a valid count can still be false. Use your observed outputs as evidence.

Vocabulary: a field is one named data item; a constraint limits its allowed values; coercion converts input; validation checks the declared contract; serialization converts an instance to a representation for storage or transport. None of these grants permission.

Connection to our shop tools: NoArguments accepts an empty argument dictionary; ProductArguments checks a SKU; DraftArguments also checks a positive quantity. Each follows the same BaseModel pattern you have just used. A tool's parameters points to one of these classes. The dispatcher calls tool.parameters.model_validate(call.arguments) before handing the resulting instance to the handler. Later, trace that exact call in the code. The model owns structural checks; shop records and the handler own the business calculation; the allowlist and authority callback decide whether execution may proceed.

Official Pydantic 2 references for further reading: models, fields, strict mode, validation errors, serialization, and JSON Schema.

From a Python function to a shop tool

A tool is a program operation exposed through a named interface. In this chapter the stock tool reads a copied shop snapshot, the supplier tool reads a price, and the draft tool calculates an order proposal. The interface has a name, an argument contract and a handler. The handler is the Python function that performs the operation. Advertising a schema describes possible arguments; invoking a handler causes behavior.

Pydantic is introduced in full below. As you work through it, keep three independent questions in view: is the argument well formed, does it agree with the current business rule, and is this caller permitted to execute the operation? A positive integer can pass the first question and fail the second. A valid request can pass both and still be outside the caller's allowlist.

Calculate before designing an interface

Work with integer cents so multiplication does not introduce a floating-point rounding question. The SKU is a stable product identifier; the display name may change without changing identity. An empty catalog means no products. Two rows with the same SKU are ambiguous and must not silently overwrite each other during construction.

intro_shop = [
    {"sku": "MANGO", "on_hand": 1, "target": 5, "price_cents": 325},
    {"sku": "COCOA", "on_hand": 9, "target": 6, "price_cents": 300},
]
for intro_product in intro_shop:
    intro_needed = max(0, intro_product["target"] - intro_product["on_hand"])
    intro_total = intro_needed * intro_product["price_cents"]
    print(intro_product["sku"], intro_needed, intro_total)
assert 4 * 325 == 1300

Mango needs four tubs, costing 1300 cents. Cocoa needs none. A draft requesting three mango tubs is type-correct but inconsistent with our exact-need contract. The handler must obtain the authoritative price itself; accepting a caller's invented total would move the calculation's authority to untrusted input.

Observe why a snapshot is copied

A closure can retain access to a shop dictionary. Without a deep copy, another part of the program can mutate that dictionary after the tools are built and silently change what an existing tool sees. A copied snapshot makes this particular experiment repeatable. It does not promise that a real shop never changes; refreshing stale stock is a separate contract.

import copy

intro_original = {"MANGO": {"on_hand": 1, "target": 5}}
intro_snapshot = copy.deepcopy(intro_original)


def intro_stock_handler():
    return copy.deepcopy(intro_snapshot)


intro_original["MANGO"]["on_hand"] = 20
intro_observed_stock = intro_stock_handler()
assert intro_observed_stock["MANGO"]["on_hand"] == 1
intro_observed_stock["MANGO"]["on_hand"] = 99
assert intro_stock_handler()["MANGO"]["on_hand"] == 1
print("Neither external input mutation nor output mutation changed the retained snapshot.")

The two copies protect different directions: the first prevents mutation through the original input; the second prevents mutation through a returned result. Explain both before writing the factory. A factory is a function that constructs and returns configured objects.

Trace the three boundaries

ToolCall holds the requested name and arguments. ExecutableTool associates a name with a Pydantic parameter class and a handler. Dispatcher.invoke looks up the name, checks the allowlist, validates arguments and invokes the handler. The order matters: a refusal returned after the handler runs cannot undo a side effect. For consequential operations, the authority callback belongs before the handler. Our draft-only shop tools do not send purchases.

The construction exercise asks for the complete factory, not three constant answers. It must work with a changed shop, empty input and duplicate identities. The transfer adds reservations: sellable stock becomes physical stock minus reserved stock. Both reporting and draft validation must use the same revised rule, or the interface can contradict itself.

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

Design question: A draft is a calculation, not a purchase?

Construct the complete tool factory: reject duplicate product identities, isolate a copied stock snapshot, expose stock and supplier tools, and make draft quantity equal current need. Preserve strict Pydantic argument validation and USD integer-cents arithmetic.

You edit a complete function in a temporary copy of book/always_on/learner/ch02.py. The real callers, database and tool boundaries remain connected. The self-contained setup above supplies the frozen development runtime. The notebook runs reviewed local subprocesses; it is not a security sandbox. No live account is required.

Predict before running

Compare schema refusal, business-rule refusal and successful calculation. Connect build_tools to the same Dispatcher used by the Chapter 3 loop.

Write the expected result and a falsifying observation before running. Include one legal action, one refusal, and the exact-empty case where the interface permits it. Explain the consequence for Lucy if your prediction is wrong.

import json
import os
import runpy
from pathlib import Path

ROOT = COURSE_ROOT

SourceTask = runpy.run_path(str(ROOT / "book/always_on/exercises/source_tasks_v1.py"))["SourceTask"]
REFERENCE_LESSON = 2
HANDOFF = Path("ch02-unit-a-handoff-v1.json")

Construct the complete mechanism

Implement build_tools in the string below. Keep the named signature and existing helper interfaces. The starter is intentionally incomplete; its failed connection is reported separately from whether the notebook itself executed. A constant answer cannot stand in for the real mechanism.

Inspect the supplied caller around the function in book/always_on/learner/ch02.py. Draw the data path from the observed output through this function to its actual input or query. Then write your implementation from the contract.

implementation_source = """
def build_tools(shop):
    raise NotImplementedError("Construct this chapter mechanism")
"""

Hint 1: the design

An integer quantity can be schema-valid and still over-order.

Hint 2: the boundary

Inspect the parameters and the caller in book/always_on/learner/ch02.py. Identify validation, durable state and the first externally observable effect. Preserve the existing surrounding helper contracts.

Hint 3: the structure

Use rows keyed by identity, closures over a deep copy, the existing argument models, and three ExecutableTool registrations.

Connect to the cumulative runtime

The following installs your complete function into the copied runtime and executes the chapter probe against it. It saves your implementation and the observed connection for Unit B only after that connection succeeds.

def connect_build(source):
    task = SourceTask(ROOT, REFERENCE_LESSON)
    try:
        task.install(source)
        result = task.visible()
        if result["status"] == "PASS":
            task.save(HANDOFF, result)
        return result
    finally:
        task.close()


build_result = connect_build(implementation_source)
print("CONNECTION", build_result["status"])
print("OBSERVATION", build_result["observation"])

Challenge and transfer

Try a changed stock snapshot, an empty catalog, duplicated identities and a True quantity. Record which boundary refuses each.

Use a fresh SourceTask, install your implementation and edit only its copied probe to run the changed input. Keep the expected result in your prediction notes, independent of your implementation. Call task.run("MY_TRANSFER", expected=your_expected) and close the task in finally. Retain both a valid and a refused case so rejecting everything cannot pass.

The instructor runs additional cases with different identities and boundaries against the real source. Passing the visible connection alone is not the transfer verdict. Do not put instructor solutions or holdouts into a student submission.

Save and explain

After success, submit ch02-unit-a-handoff-v1.json, your source, prediction notes and changed-input observations. Unit B checks the chapter, runtime hash and exact implementation hash and re-executes your code.

Explain which input or state caused the output, which observation would refute the explanation and what remains outside the guarantee: No supplier send exists in these tools.

exercise_report = {
    "unit": "ch02-a",
    "attempted": 1,
    "completed": int(build_result["status"] == "PASS"),
    "failed": int(build_result["status"] != "PASS"),
    "skipped": 0,
    "connection": build_result["status"],
    "handoff": "WRITTEN" if build_result["status"] == "PASS" else "NOT_READY",
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))

Changed-constraint construction: Use one reservation rule in reporting and drafting

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(row). Return max(0, target - on_hand + reserved), treating an absent reserved field as zero. Inputs are validated nonnegative integers. The driver uses your result for both the stock report and the exact-quantity draft decision; changing your function must change both.

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.

def transfer_check(row):
    raise NotImplementedError("Calculate sellable stock before the deficit")

import copy
import json

TRANSFER_CASES = [
    ("reserved mango", [{"on_hand": 1, "target": 5, "reserved": 2}], 6),
    ("no reservations field", [{"on_hand": 1, "target": 5}], 4),
    ("surplus", [{"on_hand": 12, "target": 6, "reserved": 2}], 0),
    ("exact sellable target", [{"on_hand": 7, "target": 5, "reserved": 2}], 0),
]


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.

Connect the reservation rule to both callers

Predict the reported need and draft total. Both callers below invoke your function. A four-tub draft must refuse after two tubs are reserved.

if TRANSFER_PASSED:
    reservation_row = {"on_hand": 1, "target": 5, "reserved": 2}

    def reservation_stock():
        return {"needed": transfer_check(reservation_row)}

    def reservation_draft(quantity):
        if quantity != transfer_check(reservation_row):
            raise ValueError("quantity differs from current sellable-stock need")
        return {"quantity": quantity, "total_cents": quantity * 325, "status": "DRAFT"}

    assert reservation_stock()["needed"] == 6
    assert reservation_draft(6)["total_cents"] == 1950
    try:
        reservation_draft(4)
    except ValueError:
        print("Old four-tub request refused by the same shared rule.")
    else:
        raise AssertionError("The draft validator ignored the new rule")
    print(reservation_stock(), reservation_draft(6))
else:
    print("Finish the transfer function before connecting its two callers.")

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.

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": "ch02-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 / "ch02-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": "ch02-a",
            "transfer_passed": TRANSFER_PASSED,
            "starting_evidence": course_submission["starting_evidence"],
            "edition": "student",
        },
        sort_keys=True,
    )
)

Extension: put a language model behind the typed tools

Every tool call above was written by you, one line at a time. The reason the shop tools have a schema, a business rule and a Dispatcher in front of them is that in production the caller is not you: it is a model, and it can ask for anything. This closing section puts a real model behind the same three tools and shows how far the boundaries hold.

The mechanism is the loop you will build in Chapter 3, shipped in the embedded runtime as agent_loop.run_loop. It sends the conversation and the tool schemas to a model, reads back either an answer or tool calls, runs each call through the dispatcher, appends the observation under the same tool_call_id, and asks again, while counting every attempt against Limits. The model only ever proposes a call. The dispatcher checks the arguments against the Pydantic class; the handler applies the business rule; nothing the model says can purchase anything.

The runtime adds one file for this section, llm_lab.py. It implements the loop's Model protocol twice: ScriptedModel replays turns written down in the notebook, so this section runs identically for everyone, and OpenAIModel sends the same messages to a hosted model when a key is present. The loop cannot tell them apart; the record always says which one ran.

Prediction: the next cell builds the dispatcher from the supplied reference build_tools (not your Unit A copy, which the harness closed) and prints what the model will see. Write down which tool has no arguments, which argument is an integer, and what the loop's default limits are before you run it.

import contextlib
import json
import runpy

from sovereign_agent.agent_loop import Limits, run_loop
from sovereign_agent.llm_lab import ScriptedModel, describe, scripted_turn, tool_trace

# The learner module resolves the Chapter 1 checkpoint relative to the runtime root.
with contextlib.chdir(COURSE_ROOT):
    chapter_two = runpy.run_path(str(COURSE_ROOT / "book/always_on/learner/ch02.py"))
shop_dispatcher = chapter_two["build_tools"](chapter_two["SHOP"])
shop_need = {
    row["sku"]: max(0, row["reorder_point"] - row["on_hand"])
    for row in chapter_two["SHOP"]["products"]
}
for tool_schema in shop_dispatcher.schemas():
    function = tool_schema["function"]
    print(function["name"], "requires", function["parameters"].get("required", []))
    print("   ", function["description"])
print("Need per product:", shop_need)
print("Default limits:", Limits())
SHOP_MESSAGES = [
    {
        "role": "system",
        "content": (
            "You help Lucy run her ice cream shop. Check stock with the tools before drafting "
            "anything. A draft is a calculation, not a purchase; draft exactly the quantity "
            "the shop needs. Finish with a short plain-words summary of what you drafted."
        ),
    },
    {"role": "user", "content": "What needs ordering this morning?"},
]

The schemas are the same model_json_schema() output you inspected earlier, wrapped in the {"type": "function", ...} envelope the API expects; required comes from the fields without defaults. Limits() is the budget the loop enforces before every attempt: model calls, tool calls, seconds, context bytes and output tokens. Chapter 3 makes you construct that admission decision; here you watch it work.

Run the loop against a recorded transcript

The turns below are what a model replied, written down. Read them as a conversation. Turn one asks for stock. Turn two asks for a supplier price and two drafts, one of which asks for the wrong quantity. Turn three corrects that draft. Turn four gives the answer, which ends the loop.

Prediction: which draft is refused, and what does the model see when it is refused? Recall that Dispatcher.invoke flattens every handler exception to tool_failed so a raw error can never leak an argument; the model has to work out the right quantity from list_stock itself.

RECORDED_TURNS = [
    scripted_turn(calls=[{"name": "list_stock", "arguments": {}}]),
    scripted_turn(
        calls=[
            {"name": "supplier", "arguments": {"sku": "SKU-VANILLA"}},
            {"name": "draft_order", "arguments": {"sku": "SKU-VANILLA", "quantity": 6}},
            {"name": "draft_order", "arguments": {"sku": "SKU-STRAWBERRY", "quantity": 3}},
        ]
    ),
    scripted_turn(
        calls=[{"name": "draft_order", "arguments": {"sku": "SKU-STRAWBERRY", "quantity": 4}}]
    ),
    scripted_turn(
        "Drafted 6 vanilla at 250 cents each and 4 strawberry at 275 cents each. Chocolate is "
        "above its "
        "reorder point. The first strawberry draft was refused because 3 is not the need."
    ),
]
LOOP_LIMITS = Limits(model_calls=8, tool_calls=16, seconds=120, output_tokens=4096)

scripted_result = run_loop(
    ScriptedModel(RECORDED_TURNS), shop_dispatcher, SHOP_MESSAGES, limits=LOOP_LIMITS
)
print(describe(scripted_result, "scripted"))
scripted_trace = tool_trace(scripted_result)
assert scripted_result.status == "COMPLETED"
assert [step["ok"] for step in scripted_trace] == [True, True, True, False, True]
assert scripted_trace[3]["result"] == "tool_failed"
assert scripted_trace[4]["result"]["total_cents"] == 4 * 275
# Every observation in the transcript carries the identifier of the request it answers.
assert all(
    message.get("tool_call_id") for message in scripted_result.messages if message["role"] == "tool"
)

The refused draft appears in the transcript as an ordinary failed observation, and the loop carries on: refusing is not crashing. Count the tool calls in the record against the limits you predicted, and find the observation whose tool_call_id matches the refused request.

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 three schemas and the same system message. It may draft in a different order, ask the supplier first, or skip a product; what it cannot do is draft a quantity the shop does not need, because that rule lives in the handler and not in the prompt.

Prediction: write the drafts a careful assistant should produce from the stock you printed above. Then run the cell and compare the model's judgment, or the recorded transcript when the run is scripted, against your list.

from sovereign_agent.llm_lab import DEFAULT_MODEL, build_model, resolve_api_key

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, shop_dispatcher, SHOP_MESSAGES, limits=LOOP_LIMITS)
print(describe(live_result, chosen_model.source))
live_trace = tool_trace(live_result)
assert live_result.status in {"COMPLETED", "MODEL_CALL_LIMIT", "TOOL_LIMIT", "MODEL_FAILED"}
# Whatever the model did, every draft that succeeded matches the shop's own need.
for step in live_trace:
    if step["tool"] == "draft_order" and step["ok"]:
        assert step["arguments"]["quantity"] == shop_need[step["arguments"]["sku"]], step

Challenge: change the budget, then change the shop

The loop's admission rule is data. Rerun the recorded transcript with Limits(tool_calls=2) and predict, before running, which status ends the loop and how many observations the transcript holds when it stops. Then change one product's on_hand in a copy of SHOP, rebuild the dispatcher, and show that the recorded transcript's corrected strawberry draft is now refused too: the transcript did not change, the shop did. With a live key, rerun the live cell after the change and explain whether the model recovered by reading stock again.

In your notes, separate what the scripted run proves (the loop, the dispatcher, the handler rule, the transcript accounting) from what only a live run can show (the model's judgment). The saved record keeps the two apart.

tight_result = run_loop(
    ScriptedModel(RECORDED_TURNS),
    shop_dispatcher,
    SHOP_MESSAGES,
    limits=Limits(model_calls=8, tool_calls=2, seconds=120, output_tokens=4096),
)
print("Tight budget:", tight_result.status, "after", tight_result.tool_calls, "tool calls")
assert tight_result.status == "TOOL_LIMIT" and tight_result.tool_calls == 1

llm_lab_report = {
    "unit": "ch02-a",
    "scripted": {
        "status": scripted_result.status,
        "tool_calls": scripted_result.tool_calls,
        "refused": [step["tool"] for step in scripted_trace if not step["ok"]],
    },
    "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,
        "drafted": [
            step["arguments"] for step in live_trace if step["tool"] == "draft_order" and step["ok"]
        ],
        "answer": live_result.answer[:400],
    },
    "tight_budget": {"limits": {"tool_calls": 2}, "status": tight_result.status},
}
llm_report_path = COURSE_WORK / "ch02-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.