Uncertain Supplier Orders: Unit B 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 12. 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.
Repair a retry that invents an operation identity, then normalize another supplier's discovery response without changing intent.
Solve interpret discovery without inventing certainty 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	Reproduce, diagnose and repair	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/ch12-b 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.

SQLite from the first row to an atomic change

A dictionary disappears when its process ends. A database can retain records so a later process can resume from evidence. SQLite is an embedded database: Python's sqlite3 library opens a local database file without starting a separate database server. SQL is the language used to define, select and change its records. A table has named columns and rows. A primary key identifies a row; a query asks for rows satisfying a condition.

Start with a deliberately small preference table. Read the SQL as instructions: create the table, insert one named value, then select the value for one session. ? is a parameter placeholder; the values are passed separately so they are data, not SQL instructions. fetchone() returns one row or None; it does not guarantee that a matching row exists.

python

import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as intro_sql_folder:
    intro_db_path = Path(intro_sql_folder) / "example.sqlite"
    intro_db = sqlite3.connect(intro_db_path, autocommit=True)
    intro_db.execute("CREATE TABLE preference (id INTEGER PRIMARY KEY, session TEXT, value TEXT)")
    intro_db.execute("INSERT INTO preference (session,value) VALUES (?,?)", ("lucy", "09:00"))
    intro_row = intro_db.execute(
        "SELECT value FROM preference WHERE session=?", ("lucy",)
    ).fetchone()
    print("Matching row:", intro_row)
    assert intro_row == ("09:00",)
    assert (
        intro_db.execute(
            "SELECT value FROM preference WHERE session=?", ("another-session",)
        ).fetchone()
        is None
    )
    intro_db.close()
    intro_reopen = sqlite3.connect(intro_db_path, autocommit=True)
    assert intro_reopen.execute("SELECT count(*) FROM preference").fetchone()[0] == 1
    intro_reopen.close()
print("The row survived closing and reopening its connection.")

Index [0] selects the first column of a returned tuple. sqlite3.Row is an alternative row factory that also permits named-column access. dict(row) then produces an ordinary dictionary. The book's Database wrapper supplies that configuration and its schema; the wrapper is course code, while sqlite3 is the standard library. You will use the public connection and transaction methods explained at the exercise boundary, rather than needing to reconstruct the wrapper.

Now consider a budget. Moving five cents from reserved to spent requires two values to change together. A transaction makes a group of local changes commit together or roll back together. The example explicitly controls SQL transactions with autocommit=True and SQL statements. BEGIN IMMEDIATE starts a write transaction; COMMIT keeps its changes; ROLLBACK discards them. Predict the row after the deliberately raised exception. Catching an error alone would not undo the first update; the rollback is the operation that restores the prior state.

python

intro_ledger = sqlite3.connect(":memory:", autocommit=True)
intro_ledger.execute(
    "CREATE TABLE budget (id INTEGER PRIMARY KEY, reserved INTEGER, spent INTEGER)"
)
intro_ledger.execute("INSERT INTO budget VALUES (1,5,0)")
intro_ledger.execute("BEGIN IMMEDIATE")
try:
    intro_ledger.execute("UPDATE budget SET reserved=0 WHERE id=1")
    raise ValueError("injected failure before the matching spend update")
except ValueError:
    intro_ledger.execute("ROLLBACK")
assert intro_ledger.execute("SELECT reserved,spent FROM budget").fetchone() == (5, 0)
intro_ledger.execute("BEGIN IMMEDIATE")
intro_ledger.execute("UPDATE budget SET reserved=0,spent=5 WHERE id=1")
intro_ledger.execute("COMMIT")
print(
    "After a complete change:", intro_ledger.execute("SELECT reserved,spent FROM budget").fetchone()
)
intro_ledger.close()

The literal :memory: creates a temporary database inside this connection; it is useful for the small experiment, but the earlier file example establishes persistence. Neither example proves that a remote supplier rolls back when the local transaction rolls back. An external operation has its own state and evidence.

SQL you will meet later. UPDATE ... SET ... WHERE ... changes selected rows. AND combines conditions. ORDER BY makes an ordering explicit; absent that clause, do not rely on row order. count(*) counts rows; sum(amount) totals a column and can be NULL on an empty input; coalesce(sum(amount),0) uses zero for that empty aggregate. A UNIQUE constraint rejects duplicate identities. GROUP BY status computes one aggregate per status.

An invariant is a condition that must remain true across operations, such as nonnegative reserved money. A snapshot is a consistent view at one point; two separate reads can describe different moments unless their transaction contract binds them. In the book, with db.immediate() groups related writes. It is a course-defined context manager with the commit/rollback purpose you just observed. Do not assume that an arbitrary with connection has identical behavior under every SQLite autocommit setting.

Your prediction: two workers both read ten remaining cents outside a transaction and each approve seven. Why can both believe the next order fits? Explain what must be checked together with the write. Then change the example's initial reserved amount and repeat the failure. Reference: Python's SQLite tutorial and transaction control.

An ambiguous response leaves two systems with different knowledge

Lucy approves an order. The supplier stores it, then the network reply is lost. Locally we see a timeout; remotely an order exists. The central difficulty is not calculating the quantity again. It is deciding what evidence is available without creating another purchase. Distributed state means relevant records live in more than one independently changing system.

An intent is the durable local record of one exact operation before sending. An operation identity distinguishes that intent from another business operation. A receipt binds a supplier outcome to that identity and exact proposal. Reconciliation asks the external system for evidence and brings local state into agreement where the evidence is conclusive.

Separate the effect from the caller's observation

The function below deliberately stores a supplier record before raising. The exception describes what the caller observed; it does not reverse the stored effect. Predict both the local exception and the independent order count before running.

python

intro_supplier_orders = {}


def intro_accept_then_lose_reply(operation, proposal):
    if operation in intro_supplier_orders and intro_supplier_orders[operation] != proposal:
        raise ValueError("same identity used for different intent")
    intro_supplier_orders.setdefault(operation, dict(proposal))
    raise TimeoutError("reply lost after acceptance")


for intro_operation in ("opening-order", "opening-order"):
    try:
        intro_accept_then_lose_reply(intro_operation, {"sku": "MANGO", "quantity": 4})
    except TimeoutError:
        print("caller state UNKNOWN; supplier count", len(intro_supplier_orders))
assert len(intro_supplier_orders) == 1

This fixture's stable-identity contract prevents a second effect when the same operation is repeated. A real supplier must explicitly support the relevant behavior; a local ID string alone cannot force a remote service to deduplicate. The book's supplier fixture lets us inspect the independent records so the experiment does not grade its own final prose.

Why a new identity creates a new operation

Repeat the experiment with a new identity after the lost reply. Nothing tells the supplier that the second identity is intended as a retry of the first. There are now two valid-looking intents for the same products. Identical payloads are not enough to conclude that two operations are one.

python

try:
    intro_accept_then_lose_reply("opening-order-retry", {"sku": "MANGO", "quantity": 4})
except TimeoutError:
    pass
assert len(intro_supplier_orders) == 2
print("Stored operation identities:", sorted(intro_supplier_orders))

Unit B introduces this defect at the actual send boundary with a freshly generated UUID. A UUID is a generated identifier; uniqueness is useful for a new intent but harmful when it accidentally turns a retry into a new business operation. The repair reuses the recorded identity and proposal. The evidence is the supplier's order count, not a message claiming “retry handled.”

Read the local state machine

State	What the local record says	What it does not establish
APPROVED	Exact intent has bounded authority	Supplier acceptance
SENDING	Send was admitted and intent recorded	A conclusive response
UNKNOWN	Available evidence cannot settle the outcome	Failure or permission for a new order
CONFIRMED	Matching accepted receipt was retained	Physical delivery
REJECTED	Matching conclusive rejection was retained	That every transport failure is rejection

The reservation remains held while the outcome is uncertain. On a matching accepted receipt, money moves from reserved to spent exactly once. On a conclusive rejection, the reservation is released. A repeated identical receipt must not spend twice; a contradictory receipt must not silently overwrite the first outcome. Both identity and exact proposal need comparison.

Adapt vocabulary without weakening evidence

Another supplier may call its fields order_ref, payload and decision. An adapter maps these names to the internal receipt. It must preserve operation and proposal and reject unknown decisions. None from a lookup is not an accepted receipt. Nor is an unavailable lookup proof that no remote order exists. Keep absence, unavailability and conclusive rejection distinct.

The core code uses SQLite for local intent and a local supplier fixture for independent effects. A loopback server, where used, listens on this machine rather than contacting a real supplier. Its subprocess and temporary database are supplied infrastructure. Your work constructs the transition boundary, repairs identity handling and normalizes changed discovery data.

Before coding, draw two columns, local ledger and supplier ledger, at four instants: before send, after acceptance, after lost reply and after lookup. Put unknown values explicitly in the table. The transfer task will change the receipt vocabulary and failure schedule while requiring the same accounting invariant.

Choose an explicit starting point for this independent notebook

This Unit B runs without Unit A. By default it prepares a supplied reference starting point and labels its provenance. It is not evidence that you built Unit A. To investigate your own successful implementation, set LEARNER_HANDOFF to its saved path before running the cell. An invalid selected file refuses; it is never silently replaced with the reference.

SourceTask supplies copied-source execution and handoff validation; RuntimeLab supplies the controlled failure experiment. Their public operations are introduced beside the main exercise. The artifact stores identity and observations; no variables from another kernel are required.

python

LEARNER_HANDOFF = None

Prepare and validate the supplied starting artifact

python

import json
import runpy
import shutil
from pathlib import Path

COURSE_INPUT = COURSE_WORK / "ch12-unit-a-handoff-v1.json"
if LEARNER_HANDOFF is not None:
    learner_input = Path(LEARNER_HANDOFF).expanduser().resolve()
    if not learner_input.is_file():
        raise FileNotFoundError("The selected learner handoff does not exist")
    if learner_input != COURSE_INPUT.resolve():
        shutil.copy2(learner_input, COURSE_INPUT)
    HANDOFF_ORIGIN = "LEARNER_SELECTED"
else:
    reference_encoded = (
        "c-pmDZExJT5&o`U!PA$M5_x0i&_lC;8o0Kub(8EG>rK*wYY1Fv*2}Y&R7lG9USOa<qQ9`eq##8~)XUm="
        ">Fozwq{!j$nTIpO5!sfEOZ1yy)brMyZINKNz=2%T6kh|C6SxH`1tRr8T$VudRCjtFab}xmChQZhXt}cu"
        "@u?sZerlS#W|)L+xh4fPZL)=0lM>5KK~{F@b1XMTM-FS?Kr=`gzs8jOjwNAKOoU_{CR%ShO913)QDwXF"
        "-C-e!keJGp@f>)w&kMCpLa)zRL9$(>_RZNXM5yGfWCEU&cE|XQm6a8kY$F8pQw7zYnK_ntg$39-r>5<B"
        "SO};Qo7@5ygwd4HHA8=iQSeiQj$VeIx0uhDC`Kpq_lt`pRV~l+2+cw50;o(yRh9(-j^f%Z6JF6&;Ts6a"
        "13mo(@>h)AyuigB-iega7ug1~o01VK#fxln_-az_{4l)q5JJcwlkR(0o_9Wfdv@|GihB?SYtDWLDnJIG"
        "ePs{-A&ih@5*KNv7#6=w4-XGR&+~FvTfFnyECFXsM(A0DB<9y3V}DKVWyQgdkhrO0|KiR2gAd2Ev$t=L"
        "{Rn-kFqK5^;-AB*);3?ExPep)Q`@bPubJtGdSJtrAYauz#%e*nijbdea4CVmn!-Ksj0m-Uy_pF43n&Yq"
        "L~N8&)#^9MWGrr~T8!pHTvMSd>p;j#p=dN9>Y1!-TN$$H2}*+AZM(k)j{09MASc%#B#KeKnq-Vp$kf@G"
        "z&&J@gg}qc7Q9H_CMQcf{+Gr49U7d5kAF`V2_kv?+K*7s>K4^&g_AYNY{Mu7;pO3#?lYEa7s^*r^Xy1y"
        "-)jZhHqIpDv4OSeEj5ecnq6%-+?Rxdh?Uovs6Ro7o}-@+>k+k<9Aw2QYC&wZFl^JiP1#BSztyfpBi;@("
        "G~(>;%W)MF%SuGG32_l|$bgjp2cUJwmQmP!xiTEo)*!EsS|mz!Z1r<rA7M^ytI?4^u7ZK}bGul*Wq+!G"
        "y0opU)>ys;!=LJyF)0|%MW9DFudYI8f&p!0ynS;OBC<x^E(C=DG-DJxqFf04X$|3Jzu#5YfvFwucACKt"
        "HH$c8jOVH3m<rOcNKq~&xl5-A>G&BTqj9WMH5|Jma-*zDwp+*vmWt3=pw`IKM!%YlkD1@4z_TZrhmL2b"
        "R^0&)?TfGxJ|KiX;Mx|52o3+VqY1}E06GW0B|^oqgmcCNpOwJ%p+h+Xfhdz`1qGwmNHQchMChgc&Kep&"
        "6vxOvK7Dt#^t%gLLRbc(AJo9}^Tqr_a@rgFm}0cpAxxWCgj!9vc!ySiMC-Th?<sR=le|AaJzge5TU;d8"
        "Z^S=dB(u}A+24MA$~83R=(OBCt&TU%ViUdCgixupkI-jdclc9>p%1MzC^~q*pS_vQKhC}cakGPO06gCk"
        "DIr(*<_7DGZ2?vEhO#^QM?r4(_6?BtKw2ax$=Ue`Efu+mAXKT4L14~*gmRL}a34+|a8ZGQ$VL$kpDMy3"
        "??C}HV>#p~uKi_RZA&o(!w8vNgA7{78%9VlF4G&>iFnB?h>(C1b1WGbao|TP$(gEQxG30NN->Rp!G(ZF"
        "Xh_!G#hMf3hl(Ojvcq_z{0}keg@ms`rg0)$EVB)Qdz?vR$o{x;hP+{IC5k@#$0sMr`7$}x>0R=V<U~C`"
        "e2q4w)+J*!D=I;5VMK?{!O`oU%xAxxE#5u7Fj+qq7!{CpYdG_9wKGN~$BLE0gf-><DydFkZwF2)@MIeG"
        "_|_ZPt>cJVck}S&PNZ?tcen=Mx|5#bILn}v`mmU7JG$u5Vy*v&c$?QYBaB~n{LMgE;Lt&V*66LXoR%Ju"
        "d*+SoumgYe$VvUSZ8h1V+E_P}io4B&*O5OI=p<9I^Zfj^n|_W!qKSjItfJDticqIH#OUzRop(azU@7<Z"
        "(x}~!rRs>#j1-WMC_5HkBR5&@lkXwg8)d|3O68tl6CzQYh5DUz{pmBRbJ=T-b8u7H4AY$CSa$MMTf3Po"
        "L6t2q-<=W;>i)J1N)BssAN$$n=<uKwbZ|=qSru^L>%2C%<EQR@ad1yValay<szE_BJ8VIct}K%#3lyWv"
        "t3mIiHhsIdCSjlS6LynlhT(S!vEHmF#s)eRKW!d2EP5kB9mR@pO_=SG&bfd!7gL9dt;>7(u6^44r@7&L"
        "CvH#4tMSsV{IVNU){NH|5~>&NLS`?~>docB(bdRh(0SLn_;o!<U7~BXh|%R(5Q|VZvbT*<c}Be$1&0wj"
        "`jaY-b+jj<PH|$LJ~l3YbdXCMMUiLE9Q)gPl0QYSUUe-$I88cMe3i+p3cXDAUU_K8bq$v(u@K!k1OM0h"
        "z+;VBn+DibA#YBG4GO(Hz}7LGV{m?aaZ#H{^?~ZM#*L7#?P3c84_<FaiT<bZ4=fN2cQ&%zEJ;o;(vS1S"
        "8?~gK<4=*?l@RDgI7w5Aw~(goYHLDqQS^+XszHS+7^*U=akMBNI`cuLBJ@6-U7Y>(ZIaeoM=>(fdpVIo"
        "#w$6YGH?<OGdw~>*M#GjVQ2>VR`khywp<*aC{9>8DpvPauCp2klF;)^a-t8osw;SdY2Ol*8maqZkuDX-"
        "s26OS*LuCNO1Q2uYoUSdQbpnVNf+M3zU96o=+$L#Q}ngsCiM)t3!gr@D_AoQslHdoj})x-#=n6?<EtFO"
        "J;qo+IG8!Apf1AOjn$iW)#xnalne)Q%we2g?C~vH^|b_^7EFk*#G(n~!(~T3)P65zK4YMq<oc?Tb5<g`"
        "0kkGus9U}$85Lt!E=@i2Rn3-wgD%4q{Wxry+^3R$mu4z$)(EI(ll|Mep?)1a3+wV>eUI_S3jgkJ%__~X"
        "#*_tPH&q$zy9+iR?JF(OMj5^0P^jxUK98cXFIgHN>?BZHUzde@rnEHkrw-@#jMh@E^gUx5GaGA6JEj~?"
        "df0brTWxxW)-rtG#K4_7Oqx1!-Lxsx>5R%{B6PUdnHXd{8nwC7UY*Rep*mLHP?OQb%8}VUA~8Dh28=C6"
        "yqFV*Ra1u#6?0x5RFudAe6YbZXY2LB?a@TVP(Pg9aUvm=aPQnfo$%W5xDBxVa-I6{<^PNNQC&zQ^g8w{"
        "xjy)_?|ch0A^BuJo7t*CBl^;%>Xq{W#~o7h(|H?ut<&pDKbz0e*?gHUlH=1~(SQH_AH<==gbz=6q<kJ#"
        "pQaJM#)M8((6`@MQx5u*xIMtW!ce768c=L*+&-2n@T>H0ijGWQ#<pbvb)V9#n=_cIfWOB2=3Wm`UU+<%"
        "xFKb!#~n7}ZOz0<8dXZudL5I^SQM<X#{O#B$Hm!lnauXiv+pDC!6sD6Ka<7DnflV7oX;1_*hkM@uJnK#"
        "@Fe^n?Z7ER"
    )
    reference_code = zlib.decompress(base64.b85decode(reference_encoded)).decode()
    reference_namespace = {"COURSE_ROOT": COURSE_ROOT, "COURSE_WORK": COURSE_WORK}
    exec(compile(reference_code, "<supplied-unit-a-reference>", "exec"), reference_namespace)
    if not COURSE_INPUT.is_file():
        raise RuntimeError("The supplied reference did not produce its handoff")
    HANDOFF_ORIGIN = "SUPPLIED_REFERENCE"
print("Starting evidence:", HANDOFF_ORIGIN)
print("The core task below validates the selected artifact before using 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

Unit A preserved one exact intent and later settled it from supplier evidence. Now you will make a plausible retry bug create two remote purchases, repair the real order boundary, and adapt a second supplier's discovery shape without changing the meaning of its receipt.

This unit runs reviewed repository code in bounded local subprocesses and starts only in-memory or loopback supplier fixtures. A subprocess is not a security sandbox.

1. Verify the handoff and load the real experiment

python

import json
import os
import sys
import tempfile
import time
from pathlib import Path

assert sys.version_info >= (3, 12)

ROOT = COURSE_ROOT


HANDOFF_PATH = Path("ch12-unit-a-handoff-v1.json")
handoff_status = "MISSING"
if HANDOFF_PATH.is_file():
    handoff = json.loads(HANDOFF_PATH.read_text(encoding="utf-8"))
    valid_handoff = (
        handoff.get("after_loss", {}).get("status") == "UNKNOWN"
        and handoff.get("final", {}).get("status") == "CONFIRMED"
        and handoff.get("supplier_orders") == 1
    )
    handoff_status = "VERIFIED" if valid_handoff else "INVALID"
print("UNIT_A_HANDOFF", handoff_status)

support = runpy.run_path(str(ROOT / "book/always_on/educator/runtime_labs_v1.py"))
RuntimeLab = support["RuntimeLab"]

Predict the broken observation before running it. If a retry invents a new operation ID, what can lookup of the original ID discover? How many supplier rows remain?

2. Break the real source copy

python

lab = RuntimeLab(ROOT, 9)
baseline = lab.run("BASELINE", expected=lab.spec["expected_baseline"])
lab.break_source()
broken = lab.run("BROKEN", expected=lab.spec["expected_broken"])
print(json.dumps({"baseline": baseline["observation"], "broken": broken["observation"]}, indent=2))
assert baseline["status"] == "PASS"
assert broken["status"] == "PASS"

The fault replaces the stable operation ID at the actual supplier.order call. The supplier accepts two distinct effects while both local attempts remain UNKNOWN.

3. Construct the repair

Return the complete replacement for the one broken fragment. The starter preserves the unsafe fresh-ID send.

python

def repair_fragment():
    return 'receipt = supplier.order(uuid.uuid4().hex, json.loads(row["proposal"]))'

Hint 1: the decision

Retries for one intended purchase keep one operation identity. A new business purchase needs a new work item.

Hint 2: the evidence

Compare supplier_order_count and local_statuses in the baseline and broken records. Read the marked line in lab.source_excerpt().

Hint 3: the structure

Call supplier.order with the existing identifier and the stored proposal. Do not generate another UUID or change the expected data.

python

lab.repair(repair_fragment())
student_repair = lab.run("STUDENT_REPAIR", expected=lab.spec["expected_baseline"])
REPAIR_PASSED = student_repair["status"] == "PASS"
print(json.dumps(student_repair["observation"], indent=2))
print("REAL_SOURCE_REPAIR", "PASSED" if REPAIR_PASSED else "NEEDS_WORK")
lab.close()

State a diagnosis that this result could falsify. Printing the expected dictionary, changing expected data, or returning a canned tool result does not alter the independently observed supplier database.

4. Adapt another supplier's discovery evidence

The partner returns order_ref, payload, and decision. Implement the adapter boundary that converts None, accepted, or declined into the exact internal receipt shape. Reject unknown decisions rather than guessing.

python

def normalize_discovery(raw):
    return raw

Hint 1: the decision

Normalize vocabulary at the integration boundary. Keep the stable operation and exact proposal unchanged.

Hint 2: the evidence

Internal receipts use operation, proposal, and an uppercase conclusive status. An absent lookup stays None.

Hint 3: the structure

Return None unchanged. Map accepted to ACCEPTED and declined to REJECTED; raise ValueError for any other decision.

python

partner_proposal = {
    "sku": "SKU-VANILLA",
    "quantity": 6,
    "unit_cost_cents": 250,
    "supplier": "lucy-local",
    "currency": "USD",
}
expected_partner = {
    "operation": "partner-order-1",
    "proposal": partner_proposal,
    "status": "ACCEPTED",
}
try:
    VISIBLE_ADAPTER_PASSED = (
        normalize_discovery(None) is None
        and normalize_discovery(
            {
                "order_ref": "partner-order-1",
                "payload": partner_proposal,
                "decision": "accepted",
            }
        )
        == expected_partner
    )
except Exception:
    VISIBLE_ADAPTER_PASSED = False
print("VISIBLE_ADAPTER", "PASSED" if VISIBLE_ADAPTER_PASSED else "NEEDS_WORK")

5. Connect the adapter to the real order workflow

The partner makes its result discoverable only after order has been attempted. The second call to the real execute path must look up that evidence before considering another send.

python

from reference_organizations.store.agent import seed_lucy
from sovereign_agent.assistant_orders import SpendingPolicy, approve, execute, propose
from sovereign_agent.assistant_work import claim, enqueue
from sovereign_agent.database import Database


class PartnerSupplier:
    idempotent = True
    identity = "partner-v2"
    timeout = 1.0

    def __init__(self, normalizer, decision):
        self.normalizer = normalizer
        self.decision = decision
        self.receipts = {}
        self.events = []

    def order(self, operation, proposal):
        self.events.append(("order", operation))
        self.receipts.setdefault(
            operation,
            {"order_ref": operation, "payload": proposal, "decision": self.decision},
        )
        raise OSError("partner committed but reply was lost")

    def lookup(self, operation):
        self.events.append(("lookup", operation))
        return self.normalizer(self.receipts.get(operation))


def run_partner_fixture(normalizer, decision):
    with tempfile.TemporaryDirectory(prefix="ch10-partner-") as directory:
        db = Database(Path(directory) / "agent.sqlite")
        supplier = PartnerSupplier(normalizer, decision)
        policy = SpendingPolicy(frozenset({"lucy"}), total_cents=2_000)
        try:
            seed_lucy(db)
            enqueue(db, "chapter9:partner", "lucy", "Replenish", subject="SKU-VANILLA")
            work = claim(db, "chapter9-partner-worker")
            identifier = propose(db, work, "SKU-VANILLA", 6, target=supplier.identity)
            digest = db.connection.execute(
                "SELECT digest FROM assistant_orders WHERE id=?", (identifier,)
            ).fetchone()[0]
            approve(
                db,
                identifier,
                digest,
                actor="lucy",
                policy=policy,
                expires=time.time() + 60,
            )
            first = execute(db, work, identifier, supplier, policy=policy)
            second = execute(db, work, identifier, supplier, policy=policy)
            status = db.connection.execute(
                "SELECT status FROM assistant_orders WHERE id=?", (identifier,)
            ).fetchone()[0]
            money = tuple(
                db.connection.execute(
                    "SELECT reserved_cents,spent_cents FROM assistant_spending WHERE id=1"
                ).fetchone()
            )
            return {
                "statuses": [first["status"], second["status"]],
                "local_status": status,
                "events": [event for event, _ in supplier.events],
                "supplier_orders": len(supplier.receipts),
                "money": money,
            }
        finally:
            db.close()


partner_result = None
if REPAIR_PASSED and VISIBLE_ADAPTER_PASSED:
    partner_result = run_partner_fixture(normalize_discovery, "accepted")
    print(json.dumps(partner_result, indent=2))
    assert partner_result == {
        "statuses": ["UNKNOWN", "ACCEPTED"],
        "local_status": "CONFIRMED",
        "events": ["order", "lookup"],
        "supplier_orders": 1,
        "money": (0, 1500),
    }
else:
    print("TRANSFER_NOT_READY — repair both learner-owned mechanisms.")

The event order is evidence: no receipt existed until after order; reconciliation then used lookup; no second order began.

Exit ticket

Explain why a local transaction cannot guarantee exactly-once external effects. Name the supplier properties this result depends on, and state what the system must do if discovery is unavailable and retransmission is not idempotent.

python

exercise_report = {
    "unit": "ch12-b",
    "attempted": 2,
    "completed": int(REPAIR_PASSED) + int(VISIBLE_ADAPTER_PASSED),
    "failed": int(not REPAIR_PASSED) + int(not VISIBLE_ADAPTER_PASSED),
    "skipped": 0,
    "connection": "PASSED" if partner_result else "NOT_READY",
    "handoff": handoff_status,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))

Changed-constraint construction: Interpret discovery without inventing certainty

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(operation, proposal, receipt). None means UNKNOWN. Otherwise require a dictionary whose operation equals the intended operation and proposal equals the exact proposal. ACCEPTED maps to CONFIRMED; REJECTED maps to REJECTED. Any other receipt or decision raises ValueError. Do not alter the proposal or assign a new operation ID.

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(operation, proposal, receipt):
    raise NotImplementedError("Require matching conclusive supplier evidence")

python

import copy
import json

TRANSFER_CASES = [
    ("no evidence", ["op-a", {"quantity": 4}, None], "UNKNOWN"),
    (
        "accepted evidence",
        [
            "op-a",
            {"quantity": 4},
            {"operation": "op-a", "proposal": {"quantity": 4}, "status": "ACCEPTED"},
        ],
        "CONFIRMED",
    ),
    (
        "new identity",
        [
            "op-a",
            {"quantity": 4},
            {"operation": "op-b", "proposal": {"quantity": 4}, "status": "ACCEPTED"},
        ],
        {"raises": "ValueError"},
    ),
    (
        "changed quantity",
        [
            "op-a",
            {"quantity": 4},
            {"operation": "op-a", "proposal": {"quantity": 5}, "status": "ACCEPTED"},
        ],
        {"raises": "ValueError"},
    ),
]


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

Extension: put a language model behind the ambiguous order

Every transition above was driven by your own code. In production the caller that asks for an order is a model, and the reason SENDING, UNKNOWN and the exact receipt live in SQLite rather than in a prompt is that a model must not be able to invent certainty about money. This closing section puts a model behind the same rows and shows what it can and cannot do.

The mechanism is the Chapter 3 loop, agent_loop.run_loop, with tools that wrap the real assistant_orders functions. The model can propose an exact order and ask for it to be sent. It cannot choose the operation identity: propose derives one from the work item and the proposal digest, so asking twice yields the same identifier. It cannot settle anything: only a receipt whose operation and proposal match the stored row settles the reservation, and execute looks for that receipt before it would ever transmit again. It cannot release money: UNKNOWN keeps the reservation until conclusive evidence arrives. The supplier is an in-process copy of the loopback fixture's rules: it commits to its own SQLite file, then loses the reply.

Prediction: the next cell builds the dispatcher and prints the schemas. Write down which tools change durable state, which argument names the product rather than an order identifier, and why the model is never shown an argument for the operation identity.

python

import json
import sqlite3
from dataclasses import dataclass, field
from pathlib import Path

from pydantic import ValidationError

from reference_organizations.store.agent import (
    DraftArguments,
    NoArguments,
    StockArguments,
    shop_dispatcher,
)
from sovereign_agent.agent_loop import Limits, run_loop
from sovereign_agent.llm_lab import ScriptedModel, describe, scripted_turn, tool_trace
from sovereign_agent.tool_dispatch import Dispatcher, ExecutableTool

POLICY = SpendingPolicy(frozenset({"lucy"}), total_cents=2000)
# Lucy's standing instruction for this morning: one exact proposal, nothing else.
STANDING_APPROVAL = {"sku": "SKU-VANILLA", "quantity": 6, "ceiling_cents": 2000}


def fresh_file(path):
    """Remove a SQLite file and its sidecars so reruns and replays start from nothing."""
    for stale in (path, Path(f"{path}-wal"), Path(f"{path}-shm"), path.with_suffix(".authority")):
        stale.unlink(missing_ok=True)
    return path


class LostReplySupplier:
    """The loopback fixture's rules in one process: commit first, then lose the first reply."""

    idempotent = True
    identity = "lucy-local@notebook-fixture"
    timeout = 3.0

    def __init__(self, path, *, reject=False):
        self.connection = sqlite3.connect(fresh_file(path))
        self.connection.execute(
            "CREATE TABLE orders(operation TEXT PRIMARY KEY,proposal TEXT NOT NULL,"
            "receipt TEXT NOT NULL)"
        )
        self.reject = reject

    def lookup(self, operation):
        row = self.connection.execute(
            "SELECT receipt FROM orders WHERE operation=?", (operation,)
        ).fetchone()
        return None if row is None else json.loads(row[0])

    def order(self, operation, proposal):
        encoded = json.dumps(proposal, sort_keys=True)
        row = self.connection.execute(
            "SELECT proposal,receipt FROM orders WHERE operation=?", (operation,)
        ).fetchone()
        if row is not None:
            if row[0] != encoded:
                raise ValueError("identity conflict: same operation, different proposal")
            return json.loads(row[1])
        receipt = {
            "operation": operation,
            "proposal": proposal,
            "status": "REJECTED" if self.reject else "ACCEPTED",
        }
        with self.connection:
            self.connection.execute(
                "INSERT INTO orders VALUES (?,?,?)", (operation, encoded, json.dumps(receipt))
            )
        # The supplier has committed. Only its reply disappears.
        raise OSError("connection closed before the reply arrived")

    def remote_orders(self):
        return self.connection.execute("SELECT count(*) FROM orders").fetchone()[0]


@dataclass
class OrderLab:
    db: Database
    supplier: LostReplySupplier
    work: object
    dispatcher: Dispatcher
    ledger: list = field(default_factory=list)

    def spending(self):
        row = self.db.connection.execute(
            "SELECT reserved_cents,spent_cents FROM assistant_spending WHERE id=1"
        ).fetchone()
        return (0, 0) if row is None else tuple(row)

    def orders(self):
        return [
            dict(row)
            for row in self.db.connection.execute(
                "SELECT id,status,amount,proposal FROM assistant_orders WHERE work_id=? "
                "ORDER BY created,id",
                (self.work.id,),
            )
        ]

    def order_events(self):
        return [
            row[0]
            for row in self.db.connection.execute(
                "SELECT kind FROM events WHERE kind LIKE 'assistant.order.%' ORDER BY seq"
            )
        ]

    def close(self):
        self.db.close()
        self.supplier.connection.close()


def open_shop(tag, *, reject=False):
    """A fresh shop ledger, one claimed work item and a supplier that loses its first reply."""
    db = Database(fresh_file(COURSE_WORK / f"ch12-{tag}-shop.sqlite"))
    seed_lucy(db)
    enqueue(db, f"ch12:{tag}", "lucy", "Replenish stock this morning")
    work = claim(db, "notebook-worker", ttl=3600)
    supplier = LostReplySupplier(COURSE_WORK / f"ch12-{tag}-supplier.sqlite", reject=reject)
    lab = OrderLab(db, supplier, work, None)

    def current_order(sku):
        row = db.connection.execute(
            "SELECT * FROM assistant_orders WHERE work_id=? AND json_extract(proposal,'$.sku')=? "
            "AND revoked=0 ORDER BY created DESC LIMIT 1",
            (work.id, sku),
        ).fetchone()
        if row is None:
            raise ValueError("no proposal exists for that product")
        return row

    def lucy_gate(call):
        """Lucy's standing approval, checked at the boundary and recorded in the real ledger."""
        if call.name == "propose_order":
            wanted = (STANDING_APPROVAL["sku"], STANDING_APPROVAL["quantity"])
            if (call.arguments["sku"], call.arguments["quantity"]) != wanted:
                lab.ledger.append((call.name, call.arguments["sku"], "outside_standing_approval"))
                raise PermissionError("Lucy did not authorize that proposal")
            lab.ledger.append((call.name, call.arguments["sku"], "draft_allowed"))
            return
        try:
            row = current_order(call.arguments["sku"])
        except ValueError:
            lab.ledger.append((call.name, call.arguments["sku"], "no_proposal_to_send"))
            raise
        if row["status"] == "DRAFT" and row["amount"] <= STANDING_APPROVAL["ceiling_cents"]:
            approve(
                db, row["id"], row["digest"], actor="lucy", policy=POLICY, expires=time.time() + 600
            )
            lab.ledger.append((call.name, call.arguments["sku"], "approved_by_lucy"))
        elif row["status"] in {"APPROVED", "SENDING", "UNKNOWN"}:
            lab.ledger.append((call.name, call.arguments["sku"], "same_operation_retry"))
        else:
            lab.ledger.append((call.name, call.arguments["sku"], "already_settled"))

    def propose_order(args):
        identifier = propose(db, work, args.sku, args.quantity, target=supplier.identity)
        row = db.connection.execute(
            "SELECT status,amount FROM assistant_orders WHERE id=?", (identifier,)
        ).fetchone()
        return {"sku": args.sku, "quantity": args.quantity, "operation": identifier, **dict(row)}

    def send_order(args):
        return execute(db, work, current_order(args.sku)["id"], supplier, policy=POLICY)

    def order_status(_):
        reserved, spent = lab.spending()
        return {"orders": lab.orders(), "reserved_cents": reserved, "spent_cents": spent}

    tools = [
        shop_dispatcher(db).tools["list_stock"],
        ExecutableTool(
            "order_status",
            "Read this work item's orders and the money ledger.",
            NoArguments,
            order_status,
        ),
        ExecutableTool(
            "propose_order",
            "Record one exact order proposal for a product. Never purchases.",
            DraftArguments,
            propose_order,
            consequential=True,
        ),
        ExecutableTool(
            "send_order",
            "Send the recorded proposal for a product to the supplier, or read back its receipt.",
            StockArguments,
            send_order,
            consequential=True,
        ),
    ]
    lab.dispatcher = Dispatcher(
        tools, allowed=frozenset(tool.name for tool in tools), before_write=lucy_gate
    )
    return lab


scripted_lab = open_shop("scripted")
for tool_schema in scripted_lab.dispatcher.schemas():
    function = tool_schema["function"]
    print(function["name"], "requires", function["parameters"].get("required", []))
    print("   ", function["description"])
print(
    "Need per product:",
    {
        row["sku"]: row["needed"]
        for row in scripted_lab.dispatcher.tools["list_stock"].handler(NoArguments())
    },
)
try:
    StockArguments(sku="SKU-VANILLA", operation="made-up")
except ValidationError:
    print("An operation identity in the arguments is refused before any handler runs.")
ORDER_MESSAGES = [
    {
        "role": "system",
        "content": (
            "You help Lucy run her ice cream shop. Read stock first. Propose an order only for "
            "what the shop needs, then send it. A reply of UNKNOWN means the supplier's answer "
            "was lost, not that the order failed: send the same product again so the runtime "
            "can look up the receipt. Finish with a short plain summary of what the ledger shows."
        ),
    },
    {
        "role": "user",
        "content": "Order what we need this morning and tell me where the money stands.",
    },
]

Four tools, two of them consequential. propose_order writes a DRAFT row through the real propose; send_order runs the real execute, which admits SENDING, calls the supplier and records UNKNOWN when the reply is lost. Neither schema has an argument for the operation identity, and the strict StockArguments class refused one: the model names a product, the runtime derives the identity.

The gate is Lucy's standing approval. A proposal outside it is refused before the handler runs. A send_order for a DRAFT that fits is approved in the ledger with the real approve, under Lucy's name; a second send_order for the same product is a retry of the same operation, the retry Unit B repairs: same identity, no fresh UUID.

Run the loop against a recorded transcript

Turn one reads stock. Turn two proposes vanilla and strawberry. Turn three sends vanilla. Turn four reads the ledger. Turn five sends vanilla again. Turn six answers.

Prediction: which proposal is refused, and by whom? After turn three, what are the reserved and spent cents, and how many rows does the supplier hold? After turn five, one row or two, and which event settles the money? Write both ledgers down.

python

RECORDED_TURNS = [
    scripted_turn(calls=[{"name": "list_stock", "arguments": {}}]),
    scripted_turn(
        calls=[
            {"name": "propose_order", "arguments": {"sku": "SKU-VANILLA", "quantity": 6}},
            {"name": "propose_order", "arguments": {"sku": "SKU-STRAWBERRY", "quantity": 4}},
        ]
    ),
    scripted_turn(calls=[{"name": "send_order", "arguments": {"sku": "SKU-VANILLA"}}]),
    scripted_turn(calls=[{"name": "order_status", "arguments": {}}]),
    scripted_turn(calls=[{"name": "send_order", "arguments": {"sku": "SKU-VANILLA"}}]),
    scripted_turn(
        "Vanilla, 6 tubs at 250 cents, was sent once. The first reply was lost, so the ledger held "
        "1500 cents as reserved; the retry found the supplier's receipt and the order is "
        "confirmed. "
        "The strawberry proposal was refused because Lucy only authorized vanilla."
    ),
]
LOOP_LIMITS = Limits(model_calls=8, tool_calls=16, seconds=120, output_tokens=4096)

scripted_result = run_loop(
    ScriptedModel(RECORDED_TURNS), scripted_lab.dispatcher, ORDER_MESSAGES, limits=LOOP_LIMITS
)
print(describe(scripted_result, "scripted"))
print("Gate ledger:", scripted_lab.ledger)
print("Order events:", scripted_lab.order_events())
scripted_trace = tool_trace(scripted_result)
assert scripted_result.status == "COMPLETED"
assert [step["ok"] for step in scripted_trace] == [True, True, False, True, True, True]
assert scripted_trace[2]["result"] == "tool_failed"
assert scripted_lab.ledger[1] == ("propose_order", "SKU-STRAWBERRY", "outside_standing_approval")
# The lost reply left the intent durable and the money reserved, with one remote row.
assert scripted_trace[3]["result"]["status"] == "UNKNOWN"
assert scripted_trace[4]["result"]["reserved_cents"] == 1500
assert scripted_trace[4]["result"]["spent_cents"] == 0
# The retry kept the same operation identity and found the receipt instead of sending again.
assert scripted_trace[5]["result"]["status"] == "ACCEPTED"
assert scripted_trace[5]["result"]["operation"] == scripted_trace[3]["result"]["operation"]
assert scripted_lab.supplier.remote_orders() == 1
assert [row["status"] for row in scripted_lab.orders()] == ["CONFIRMED"]
assert scripted_lab.spending() == (0, 1500)
assert scripted_lab.order_events() == [
    "assistant.order.approved",
    "assistant.order.intent",
    "assistant.order.reconciled",
]
assert all(
    message.get("tool_call_id") for message in scripted_result.messages if message["role"] == "tool"
)

The refused strawberry proposal is an ordinary failed observation; the loop carries on. The interesting observation is the fourth: UNKNOWN, 1500 cents reserved, nothing spent, one supplier row already committed. The runtime knew less than the supplier, and said so. The retry sent nothing: execute saw UNKNOWN, called lookup, found the receipt and settled once. One intent event, one reconciled event, one remote row.

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. The model may read the ledger more often, retry sooner, or give up on the lost reply. It cannot create a second supplier row, spend money the ledger did not reserve, or send a product Lucy did not authorize: those rules live in propose, execute and the gate.

Prediction: write down what a careful assistant should do on seeing UNKNOWN. Then run the cell and compare the model's choice, or the recorded transcript when the run is scripted.

python

from sovereign_agent.llm_lab import DEFAULT_MODEL, build_model, resolve_api_key

live_lab = open_shop("live")
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_lab.dispatcher, ORDER_MESSAGES, limits=LOOP_LIMITS)
print(describe(live_result, chosen_model.source))
print("Gate ledger:", live_lab.ledger)
print("Orders:", [(row["status"], row["amount"]) for row in live_lab.orders()])
print(
    "Reserved and spent:",
    live_lab.spending(),
    "| supplier rows:",
    live_lab.supplier.remote_orders(),
)
live_trace = tool_trace(live_result)
assert live_result.status in {"COMPLETED", "MODEL_CALL_LIMIT", "TOOL_LIMIT", "MODEL_FAILED"}
# Whatever the model did: at most one remote effect, and it matches a stored exact proposal.
assert live_lab.supplier.remote_orders() <= 1
for operation, encoded in live_lab.supplier.connection.execute(
    "SELECT operation,proposal FROM orders"
):
    stored = live_lab.db.connection.execute(
        "SELECT proposal FROM assistant_orders WHERE id=?", (operation,)
    ).fetchone()
    assert stored is not None and json.loads(stored[0]) == json.loads(encoded)
# The money ledger equals the sum of the order rows, whatever their states.
held = sum(
    row["amount"]
    for row in live_lab.orders()
    if row["status"] in {"APPROVED", "SENDING", "UNKNOWN"}
)
spent = sum(
    row["amount"] for row in live_lab.orders() if row["status"] in {"CONFIRMED", "DELIVERED"}
)
assert live_lab.spending() == (held, spent)
# Only Lucy's standing proposal was ever approved through the gate.
assert all(
    sku == STANDING_APPROVAL["sku"]
    for _, sku, decision in live_lab.ledger
    if decision == "approved_by_lucy"
)
assert all(
    message.get("tool_call_id") for message in live_result.messages if message["role"] == "tool"
)

Challenge: stop the loop early, then let the supplier say no

Two changed conditions, same recorded transcript. First, Limits(tool_calls=4): predict which turn the loop refuses, what the order row holds when it stops, and whether the reservation is still there. Then settle that order with the real execute and no model at all, and say what that proves about recovery. Second, a supplier that commits REJECTED and still loses its reply: predict the first send_order result, the row after the retry, and the reserved and spent cents. Notice that the recorded final sentence still says "confirmed". The transcript is the model's claim; the rows are the evidence. Which one do you report to Lucy?

In your notes, separate what the scripted run proves (the transitions, the stable identity, the single remote row, the money accounting) from what only a live run can show (whether the model retries the same product, reads the ledger, or invents a story). The saved record keeps the two apart.

python

tight_lab = open_shop("tight")
tight_result = run_loop(
    ScriptedModel(RECORDED_TURNS),
    tight_lab.dispatcher,
    ORDER_MESSAGES,
    limits=Limits(model_calls=8, tool_calls=4, seconds=120, output_tokens=4096),
)
print("Tight budget:", tight_result.status, "after", tight_result.tool_calls, "tool calls")
print("Rows when it stopped:", [row["status"] for row in tight_lab.orders()], tight_lab.spending())
assert tight_result.status == "TOOL_LIMIT" and tight_result.tool_calls == 4
assert [row["status"] for row in tight_lab.orders()] == ["UNKNOWN"]
assert tight_lab.spending() == (1500, 0) and tight_lab.supplier.remote_orders() == 1
# Recovery reads the supplier's evidence; it does not ask a model anything.
recovered = execute(
    tight_lab.db, tight_lab.work, tight_lab.orders()[0]["id"], tight_lab.supplier, policy=POLICY
)
assert recovered["status"] == "ACCEPTED" and tight_lab.spending() == (0, 1500)
print("Settled without a model:", recovered["status"], tight_lab.spending())

reject_lab = open_shop("reject", reject=True)
reject_result = run_loop(
    ScriptedModel(RECORDED_TURNS), reject_lab.dispatcher, ORDER_MESSAGES, limits=LOOP_LIMITS
)
reject_trace = tool_trace(reject_result)
print("Rejecting supplier:", [row["status"] for row in reject_lab.orders()], reject_lab.spending())
assert reject_trace[3]["result"]["status"] == "UNKNOWN"
assert reject_trace[5]["result"]["status"] == "REJECTED"
assert [row["status"] for row in reject_lab.orders()] == ["REJECTED"]
assert reject_lab.spending() == (0, 0) and reject_lab.supplier.remote_orders() == 1
assert "confirmed" in reject_result.answer  # the recorded prose, not the evidence

llm_lab_report = {
    "unit": "ch12-b",
    "scripted": {
        "status": scripted_result.status,
        "tool_calls": scripted_result.tool_calls,
        "refused": [decision for _, _, decision in scripted_lab.ledger if "outside" in decision],
        "orders": [row["status"] for row in scripted_lab.orders()],
        "reserved_and_spent": list(scripted_lab.spending()),
        "supplier_rows": scripted_lab.supplier.remote_orders(),
    },
    "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,
        "gate": [list(entry) for entry in live_lab.ledger],
        "orders": [row["status"] for row in live_lab.orders()],
        "reserved_and_spent": list(live_lab.spending()),
        "supplier_rows": live_lab.supplier.remote_orders(),
        "answer": live_result.answer[:400],
    },
    "tight_budget": {
        "limits": {"tool_calls": 4},
        "status": tight_result.status,
        "recovered": recovered["status"],
    },
    "rejecting_supplier": {
        "orders": [row["status"] for row in reject_lab.orders()],
        "reserved_and_spent": list(reject_lab.spending()),
    },
}
llm_report_path = COURSE_WORK / "ch12-b-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))
for open_lab in (scripted_lab, live_lab, tight_lab, reject_lab):
    open_lab.close()

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.