Instructor worked edition · 90 minutes of dedicated work · 2026-09-09
This is one of two practical units for Chapter 10. 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.
- Construct approval of one exact digest: validate operator, expiry, current work and proposal state; preserve uncertain outcomes; reserve only a new draft amount; enforce both installed and supplied account ceilings.
- Solve apply the stricter cumulative spending ceiling using changed inputs and an independent expectation.
- Retain your implementation, failed/corrected observations, causal explanation and limits.
| Minutes | Dedicated activity | Evidence you produce |
|---|---|---|
| 0–5 | State the problem and make a prediction | Initial prediction in your own words |
| 5–25 | Foundations and library examples | Values, explanations, revised predictions |
| 25–35 | Trace setup and the main interface | Input → learner function → observation |
| 35–60 | Construct and connect | Source, visible checks and runtime evidence |
| 60–80 | Implement and challenge the transfer task | Function and a new counterexample |
| 80–90 | Retrieve, explain and save | Exit ticket and retained submission |
Installation is preparation time. These are planning estimates, not measured completion times. Use the reference primers when a term is unfamiliar; in Unit B retrieve an explanation before re-reading it. Run All checks that the artifact executes. Unfinished student functions deliberately produce NEEDS_WORK. Keep your first attempt before opening answers.
This notebook belongs to the nineteen-chapter 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
Use a Python 3.14 Jupyter kernel and 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 complete source runtime uses Python 3.14, so this edition does not claim
compatibility with a hosted notebook service's default interpreter.
The collapsed cell contains 102 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. 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/ch10-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_dataThe 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"] == 6lambda 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.
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.
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.
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.
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"] == 0Four 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.
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 pence 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.
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 pence 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.
Approval binds an actor to one exact proposal and bounded money
“You may order stock” leaves too many questions unanswered. Which product, quantity, supplier, price and expiration did Lucy approve? This chapter treats approval as a decision over an exact proposal and a cumulative budget. A proposal is structured intended work. A digest names its serialized bytes. An approval records authority to proceed under stated conditions. A receipt is later evidence from the supplier; approval is not that receipt.
Make a change visible before any execution
Two dictionaries can express the same keys in different insertion orders. Canonical serialization uses a stable key order and separators before hashing. SHA-256 maps bytes to a fingerprint useful for detecting change. It is not encryption and does not identify the approving person by itself. Predict whether changing quantity changes the digest, and whether reordering keys does.
import hashlib
import json
def intro_digest(proposal):
encoded = json.dumps(proposal, sort_keys=True, separators=(",", ":"), allow_nan=False)
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
intro_proposal = {"sku": "MANGO", "quantity": 4, "unit_pence": 325}
intro_reordered = {"unit_pence": 325, "quantity": 4, "sku": "MANGO"}
intro_changed = {**intro_proposal, "quantity": 5}
assert intro_digest(intro_proposal) == intro_digest(intro_reordered)
assert intro_digest(intro_proposal) != intro_digest(intro_changed)
print("Reordered keys match; a changed quantity requires a new decision.")The ** syntax copies dictionary entries into a new dictionary before the explicit replacement.
allow_nan=False refuses nonstandard non-finite numbers in this representation. The actual order
proposal uses a validated schema and integer pence; hashing arbitrary malformed input would not
make it a meaningful proposal.
Reserved and spent money answer different questions
Reserved money is committed to admitted work whose final outcome may still be pending. Spent money is supported by confirmed accepted outcomes in the local ledger. Available authority is the ceiling minus both. If 700 pence is spent and 900 reserved under a 2000-pence ceiling, a 500-pence order is too large even though 500 is smaller than 2000.
intro_spent, intro_reserved, intro_ceiling = 700, 900, 2000
for intro_addition in (0, 400, 401, 500):
intro_exposure = intro_spent + intro_reserved + intro_addition
print(
intro_addition, "total exposure", intro_exposure, "allowed", intro_exposure <= intro_ceiling
)
assert intro_spent + intro_reserved + 400 == intro_ceilingFour hundred exactly fits; 401 does not. A repeated approval of the same already-reserved proposal must add zero, not reserve twice. This is idempotency at the local approval boundary: repeating the same admitted decision preserves its accounting effect. It does not mean that two distinct orders with the same price are the same order.
Check and reserve in one transaction
If two workers separately read the same available budget and then both reserve it, the combined
result can exceed the ceiling. Validation must be tied to the write within the transaction that
owns the current totals. The actual exercise uses db.immediate() for that local critical section.
The installed account ceiling and the supplied policy ceiling both constrain authority; the
effective ceiling is the smaller one. A caller cannot enlarge installed authority by sending a
more generous policy object.
Authority also depends on the actor, expiration, current work and proposal state. A revoked or expired decision must not become valid because its amount is small. An uncertain supplier outcome must retain its reservation rather than being reset to a fresh draft. Revocation withdraws future authority; it cannot recall an external request already admitted and sent.
Translate the contract into cases
Write a table for a fresh draft, repeated approval, changed digest, one-pence overflow, unknown actor, expiration and uncertain outcome. For each, predict both the returned decision and the reserved/spent row after the attempt. Refusal with changed money is still a defect.
Unit A implements the complete approval path. Unit B removes part of the cumulative calculation; the observable over-admission exposes the bug. In the changed-policy exercise, reduce the supplied ceiling below the installed ceiling and keep the exact-fit case. A blanket refusal cannot satisfy both the positive and negative cases. The final explanation must name which evidence grants authority and which future evidence would settle the supplier outcome.
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 labelled infrastructure; your own implementation and changed-case explanation are the evidence of learning.
Main practical: construct, connect and challenge
Design question: Small orders can exceed a cumulative ceiling?
Construct approval of one exact digest: validate operator, expiry, current work and proposal state; preserve uncertain outcomes; reserve only a new draft amount; enforce both installed and supplied account ceilings.
You edit a complete function in a temporary copy of src/sovereign_agent/assistant_orders.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
Propose real orders, approve them and inspect retained reservations. Reapproval must not reserve twice and refusal must leave the transaction unchanged.
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 = 8
HANDOFF = Path("ch10-unit-a-handoff-v1.json")Construct the complete mechanism
Implement approve 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 src/sovereign_agent/assistant_orders.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 = r"""
def approve(
db: Database,
identifier: str,
digest: str,
*,
actor: str,
policy: SpendingPolicy,
expires: float,
automatic: bool = False,
supplier: Supplier | None = None,
now: float | None = None,
) -> None:
now = time.time() if now is None else now
if type(automatic) is not bool:
raise ValueError("approval basis must be explicit")
if not math.isfinite(expires) or not now < expires <= now + 86400:
raise ValueError("approval must expire within one day")
if actor not in policy.operators:
raise PermissionError("operator is not allowlisted")
with db.immediate() as connection:
order = connection.execute(
"SELECT * FROM assistant_orders WHERE id=?", (identifier,)
).fetchone()
if (
order is None
or order["digest"] != digest
or order["status"] not in {"DRAFT", "APPROVED", "SENDING", "UNKNOWN"}
or order["revoked"]
):
raise PermissionError("approval does not match an eligible exact proposal")
_operator_state(db)
work = connection.execute(
"SELECT cancelled FROM assistant_work WHERE id=?", (order["work_id"],)
).fetchone()
if work["cancelled"]:
raise PermissionError("cancelled work cannot gain new approval")
if order["status"] in {"SENDING", "UNKNOWN"} and (
automatic
or supplier is None
or not supplier.idempotent
or supplier.identity != order["target"]
):
raise PermissionError(
"uncertain retry needs explicit approval and matching idempotent supplier"
)
if automatic and order["amount"] > policy.automatic_order_pence:
raise PermissionError("exact proposal needs operator approval")
connection.execute(
"INSERT OR IGNORE INTO assistant_spending(id,limit_pence) VALUES (1,?)",
(policy.total_pence,),
)
budget = connection.execute("SELECT * FROM assistant_spending WHERE id=1").fetchone()
assert budget
addition = order["amount"] if order["status"] == "DRAFT" else 0
# A supplied policy cannot silently raise the installed account ceiling.
if budget["spent_pence"] + budget["reserved_pence"] + addition > min(
budget["limit_pence"], policy.total_pence
):
raise PermissionError("cumulative spending ceiling reached")
connection.execute(
"UPDATE assistant_spending SET reserved_pence=reserved_pence+? WHERE id=1", (addition,)
)
connection.execute(
"UPDATE assistant_orders SET status=CASE WHEN status IN ('SENDING','UNKNOWN') "
"THEN status "
"ELSE 'APPROVED' END,approved_by=?,approved_until=?,"
"approval_basis=? WHERE id=?",
(actor, expires, "AUTOMATIC" if automatic else "OPERATOR", identifier),
)
append_event(
db,
"assistant.order.approved",
{"order": identifier, "digest": digest, "actor": actor, "automatic": automatic},
)
"""Hint 1 — the design
Comparing only the next amount ignores money already reserved or spent.
Hint 2 — the boundary
Inspect the parameters and the caller in src/sovereign_agent/assistant_orders.py. Identify validation, durable state and the first externally observable effect. Preserve the existing surrounding helper contracts.
Hint 3 — the structure
Keep all authority and accounting checks in db.immediate; compute addition from prior state; compare spent plus reserved plus addition to the smaller ceiling.
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
Test exact-limit acceptance, one-pence overflow, changed digest, stricter policy and repeated approval. Never call a real supplier.
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 ch08-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: An approval is bounded authority, not a supplier receipt.
exercise_report = {
"unit": "ch10-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: Apply the stricter cumulative spending ceiling
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(spent, reserved, addition, installed, supplied). All values must be exact nonnegative integers; otherwise raise ValueError. Return whether spent + reserved + addition is at most min(installed, supplied). A repeated already-reserved approval passes addition=0. Do not interpret this arithmetic result as actor approval or supplier evidence.
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(spent, reserved, addition, installed, supplied):
values = (spent, reserved, addition, installed, supplied)
if any(type(v) is not int or v < 0 for v in values):
raise ValueError("nonnegative integer pence required")
return spent + reserved + addition <= min(installed, supplied)import copy
import json
TRANSFER_CASES = [
("exact fit", [700, 900, 400, 2000, 2500], True),
("one penny too many", [700, 900, 401, 2000, 2500], False),
("stricter supplied policy", [700, 900, 100, 2000, 1600], False),
("no double reservation", [700, 900, 0, 2000, 1600], True),
]
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.
Instructor explanation and additional transfer cases
Check every monetary input before arithmetic. Reserved and spent both consume authority. The smaller ceiling prevents a caller from widening installed policy, while zero addition makes repeated approval accounting explicit.
Ask for the learner's first prediction and attempt before revealing this version. Passing these cases verifies behavior on these inputs; it does not establish independent student mastery. The original core holdouts also run against the connected implementation below.
INSTRUCTOR_TRANSFER_CASES = [
("negative addition", [0, 0, -1, 100, 100], {"raises": "ValueError"}),
("boolean money", [0, 0, True, 100, 100], {"raises": "ValueError"}),
("zero ceiling", [0, 0, 0, 0, 0], True),
]
instructor_transfer = run_transfer(transfer_check, INSTRUCTOR_TRANSFER_CASES)
assert TRANSFER_PASSED and all(row["passed"] for row in instructor_transfer)# Instructor holdout appended to a submitted Chapter 8 Unit A.
# ruff: noqa: F821
import json
task = SourceTask(ROOT, REFERENCE_LESSON)
try:
task.install(implementation_source)
outcome = task.transfer(ROOT / "book/always_on/exercises/ch08/holdouts/runtime-transfer-v1.py")
assert outcome["status"] == "PASS", outcome
finally:
task.close()
print("HOLDOUT_RESULT=" + json.dumps({"unit": "ch10-a", "status": "PASSED"}, sort_keys=True))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": "ch10-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 / "ch10-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": "ch10-a",
"transfer_passed": TRANSFER_PASSED,
"starting_evidence": course_submission["starting_evidence"],
"edition": "instructor",
},
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.