Schedules and Stock Events: 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 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:

1. Explain the chapter's mechanism using a prediction and an observed intermediate result.
2. Repair the failure: Advancing by one interval replays every missed wakeup after downtime. Use schedule and tick against SQLite and inspect created work IDs, job time and events across repeated passes.
3. Solve coalesce while preserving a deferred due job using changed inputs and an independent expectation.
4. Retain your implementation, failed/corrected observations, causal explanation and limits.

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

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

Run the self-contained setup

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

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

Run setup on every fresh kernel. It writes scratch runtime files separately from your retained practical-work/ch10-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

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

Reading the Python vocabulary used in this notebook

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

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

import json

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

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

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

from dataclasses import dataclass


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


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

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

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


def intro_double(value):
    return value * 2


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

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

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

from pathlib import Path
from tempfile import TemporaryDirectory

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

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

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"] == 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.

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.

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.

Scheduling converts due times into bounded work

Lucy wants an opening check every hour. The computer may be asleep or unavailable for several hours. When it wakes, should the agent replay every missed check or prepare one current check? That is a business policy, not a detail that a loop should choose accidentally. This chapter uses coalescing: create at most one occurrence for each due job in this pass, record the missed intervals, and move its next due time beyond now.

A job stores a recurring request and its timing. A tick is one bounded scheduler pass. A queue holds admitted work for execution. Producing a queue item is different from executing a model or purchasing stock. When nothing is due, the scheduler should create no work and make no model calls. Scheduling itself is arithmetic and persistence, not language generation.

Derive catch-up arithmetic on a number line

Let the saved due time be 10, the interval 5, and now 22. Due boundaries were 10, 15 and 20. We emit one current occurrence, record two skipped intervals, and move next due to 25. The elapsed whole-interval count is (now - due) // interval. Add one interval beyond that count to reach the first future boundary. Predict the exact-boundary case at now 20.

intro_due = 10
intro_interval = 5
for intro_now in (9, 10, 19, 20, 22):
    if intro_now < intro_due:
        print(intro_now, "not due", intro_due)
    else:
        intro_skipped = int((intro_now - intro_due) // intro_interval)
        intro_next = intro_due + (intro_skipped + 1) * intro_interval
        print(intro_now, "one occurrence", "skipped", intro_skipped, "next", intro_next)
        assert intro_next > intro_now

At exactly 20 the occurrence is due, and the next boundary is 25. Advancing only from 10 to 15 would leave the job due; another same-time tick could emit another stale occurrence. Unit B injects that precise defect and observes repeated work. Merely returning “coalesced” in a report would not repair the stored next due time.

A repeatable clock is an input, not a sleep

The tests pass now explicitly so they can explore a year's downtime in milliseconds. A fake clock is an authored input, not a claim that a host woke unattended. Wall-clock time represents calendar time; a monotonic clock is useful for elapsed durations because it does not move backward with wall-clock adjustments. This lesson's persisted schedule uses its declared time convention and validates supplied values. Do not mix two clock domains in one comparison.

import math

for intro_candidate_time in (0.0, 12.5, float("inf"), float("nan")):
    print(repr(intro_candidate_time), "finite:", math.isfinite(intro_candidate_time))
intro_next_due = 10.0
intro_created = 0
for intro_now in (10.0, 10.0):
    if intro_next_due <= intro_now:
        intro_created += 1
        intro_next_due += 5.0
assert (intro_created, intro_next_due) == (1, 15.0)

Floating-point infinity and NaN are values Python can represent, but they are unsuitable schedule inputs here. A positive finite interval is a precondition for catch-up division. Fractional intervals deserve explicit cases rather than an assumption that every job runs on integer hours.

Queue capacity is a separate bound

A per-tick admission maximum limits new work in this pass. Total queue capacity limits retained pending work. They are not interchangeable. If the queue cannot accept a due job, the code must retain a meaningful deferral instead of advancing as though work was created. Paused control state and disabled jobs are additional reasons not to enqueue.

The actual tick function reads due jobs in a deterministic order, checks control state and capacity, creates work through the supplied enqueue path, updates timing and records events in the same SQLite transaction. This data path is the object of the construction exercise. The transfer uses repeated timestamps, a very long absence and an exact boundary to defeat a solution that only happens to work for one missed hour.

Before coding, draw columns for due, now, interval, skipped, created work and next due. Fill them for before-due, exact-due and long-absence cases. Then add “queue full” and explain which columns are allowed to change. A local tick passing does not certify a service manager, host uptime or calendar-time accuracy; those require the book's separate operational experiment.

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.

LEARNER_HANDOFF = None

Prepare and validate the supplied starting artifact

import json
import runpy
import shutil
import textwrap
from pathlib import Path

COURSE_INPUT = COURSE_WORK / "ch10-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:
    source_task_class = runpy.run_path(
        str(COURSE_ROOT / "book/always_on/exercises/source_tasks_v1.py")
    )["SourceTask"]
    reference_task = source_task_class(COURSE_ROOT, 7)
    try:
        reference_task.install(textwrap.dedent(reference_task.fragment))
        reference_observation = reference_task.visible("SUPPLIED_REFERENCE_START")
        if reference_observation["status"] != "PASS":
            raise RuntimeError("The supplied starting point did not pass its connection check")
        reference_task.save(COURSE_INPUT, reference_observation)
    finally:
        reference_task.close()
    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

Advancing by one interval replays every missed wakeup after downtime. This time you begin with your Unit A implementation and its saved evidence. A late wake creates repeated stale briefs instead of one current brief, consuming Lucy’s request allowance.

Verify the handoff

The starting-point cell has selected the Unit A artifact explicitly. A selected learner handoff must validate; the default reference start is labeled separately. Run the setup and keep the runtime and implementation hashes in your submission.

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 = 7
HANDOFF = Path("ch10-unit-a-handoff-v1.json")
handoff_status = "MISSING"
if HANDOFF.is_file():
    task = SourceTask(ROOT, REFERENCE_LESSON)
    try:
        handoff = task.load(HANDOFF)
        handoff_status = "VERIFIED"
        print("IMPLEMENTATION", handoff["implementation_sha256"])
    finally:
        task.close()
print("UNIT_A_HANDOFF", handoff_status)

Reproduce and diagnose

Predict the consequence of this injected boundary before executing it:

next_due = row["next_due"] + row["interval_seconds"]

The controlled mutation changes the same implementation you submitted. It refuses if the declared mutation boundary no longer occurs exactly once; inspect an alternative implementation with the instructor before adapting the experiment.

baseline = broken = None
if handoff_status == "VERIFIED":
    task = SourceTask(ROOT, REFERENCE_LESSON)
    try:
        task.load(HANDOFF)
        baseline = task.visible("YOUR_BASELINE")
        if baseline["status"] != "PASS":
            raise ValueError("Saved Unit A code no longer satisfies the visible contract")
        task.inject_failure()
        broken = task.run("INJECTED_FAILURE", expected=task.spec["expected_broken"])
        print("BEFORE", baseline["observation"])
        print("AFTER", broken["observation"])
    finally:
        task.close()
else:
    print("HANDOFF_REQUIRED: complete Unit A before performing Unit B")

State a diagnosis using those two observations. Name a test that would prove your diagnosis wrong. Use schedule and tick against SQLite and inspect created work IDs, job time and events across repeated passes.

Repair the boundary

Return the complete replacement for the injected fragment. Do not edit the oracle or print a desired observation. Repair the actual source. The starter keeps the defect so the learner outcome remains incomplete.

def repair_fragment():
    return 'next_due = row["next_due"] + row["interval_seconds"]'

Hint 1: the consequence

A late wake creates repeated stale briefs instead of one current brief, consuming Lucy’s request allowance.

Hint 2: the evidence

Compare the two observations, then trace the changed field to tick in src/sovereign_agent/assistant_work.py. Distinguish a schema refusal from a business-rule or authority refusal.

Hint 3: the design

Calculate elapsed whole intervals, enqueue one occurrence, then advance by skipped plus one intervals in the same transaction.

def connect_repair(fragment):
    task = SourceTask(ROOT, REFERENCE_LESSON)
    try:
        task.load(HANDOFF)
        task.inject_failure()
        task.repair(fragment)
        return task.visible("YOUR_REPAIR")
    finally:
        task.close()


repair_result = None
if handoff_status == "VERIFIED":
    repair_result = connect_repair(repair_fragment())
    print("REPAIR", repair_result["status"], repair_result["observation"])
else:
    print("REPAIR_NOT_ATTEMPTED: missing Unit A evidence")

Transfer under a changed constraint

Use exact due time, fractional intervals, a very long absence and repeated same-time ticks. Separate coalescing from total queue capacity.

Create a fresh task, load your handoff, inject the defect and apply your repair. Then change only the copied probe to exercise the new condition. Keep the actual observation and a prediction written beforehand. Explain why a visible-case lookup or a blanket refusal could pass the original example but fail this transfer.

The instructor's holdout applies your repair to a new copied runtime and checks both the positive case and the missing protection. An exact exception or changed state must cause a failure; no broad error is accepted as successful refusal.

Exit ticket

Submit the original handoff, baseline and broken observations, repair, transfer probe and results. State what Lucy would experience before and after the fix. Identify the guarantee that still requires separate evidence: This does not guarantee a running host or a wakeup at an exact wall-clock time.

passed = repair_result is not None and repair_result["status"] == "PASS"
exercise_report = {
    "unit": "ch10-b",
    "attempted": int(repair_result is not None),
    "completed": int(passed),
    "failed": int(repair_result is not None and not passed),
    "skipped": int(repair_result is None),
    "connection": "PASS" if passed else "NOT_READY",
    "handoff": handoff_status,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))

Changed-constraint construction: Coalesce while preserving a deferred due job

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(due, interval, now, capacity). Inputs are finite numbers with positive interval and nonnegative integer capacity. Return [created, next_due, skipped]. Before due or at zero capacity, return [0, due, 0]. Otherwise create one occurrence, count skipped whole intervals, and advance next_due strictly beyond now. Do not use sleep or the actual clock.

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(due, interval, now, capacity):
    raise NotImplementedError("Plan one bounded scheduler pass")

import copy
import json

TRANSFER_CASES = [
    ("before due", [10, 5, 9, 1], [0, 10, 0]),
    ("exact due", [10, 5, 10, 1], [1, 15, 0]),
    ("long absence", [10, 5, 22, 1], [1, 25, 2]),
    ("queue full retains due", [10, 5, 22, 0], [0, 10, 0]),
]


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


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


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

Design a counterexample and retrieve the mechanism

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

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

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

Extension: put a language model behind the scheduler

Every schedule and every tick above was called by you, with a clock you chose. In production the caller who decides what Lucy's shop should wake up for is a model, and the scheduler is the boundary that keeps that decision cheap: a schedule is a durable row, a pass converts due rows into bounded work, and a missed weekend coalesces into one wakeup rather than a backlog storm. This closing section puts a real model behind that boundary.

The mechanism is the loop from Chapter 3, shipped as agent_loop.run_loop. The model only ever proposes a tool call. The dispatcher checks arguments against a Pydantic class; the handler calls the same schedule, unschedule and tick from assistant_work.py that you traced; a before_write gate can refuse a write before the handler runs. The model cannot touch assistant_jobs directly, cannot invent a wall clock, and cannot make tick replay a missed wakeup more than once. What it can do is choose identities, intervals and the clock it hands to a pass, and those choices are what the transcript records.

The runtime adds one file for this section, llm_lab.py. ScriptedModel replays turns written down in the notebook so the section runs identically for everyone; OpenAIModel sends the same messages to a hosted model when a key is present. The record always says which one ran.

Prediction: the next cell builds four tools and prints their schemas. Write down which tools are consequential, which argument the gate inspects, and why the shop's minimum interval is thirty minutes when the daily intake allowance per session is fifty requests.

import json
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field

from sovereign_agent.agent_loop import Limits, run_loop
from sovereign_agent.assistant_work import schedule, tick, unschedule
from sovereign_agent.database import Database
from sovereign_agent.llm_lab import ScriptedModel, describe, scripted_turn, tool_trace
from sovereign_agent.tool_dispatch import Dispatcher, ExecutableTool

SCHEDULE_DB = COURSE_WORK / "ch10-llm-schedule.sqlite"
HOUR, DAY = 3600, 86400
SHOP_OPEN = 1_799_971_200  # Friday 00:00 UTC on the epoch clock; the clock is an input.
MONDAY_0900 = SHOP_OPEN + 3 * DAY + 9 * HOUR


def fresh_schedule_database(path):
    """Start from an empty ledger so reruns and replays observe the same rows."""
    for stale in (path, Path(f"{path}-wal"), Path(f"{path}-shm"), path.with_suffix(".authority")):
        stale.unlink(missing_ok=True)
    return Database(path)


class NoArguments(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)


class ScheduleArguments(NoArguments):
    identifier: str = Field(min_length=1, max_length=100)
    prompt: str = Field(min_length=1, max_length=500)
    first_due: int = Field(ge=0)
    interval_seconds: int = Field(ge=1)


class JobArguments(NoArguments):
    identifier: str = Field(min_length=1, max_length=100)


class PassArguments(NoArguments):
    now: int = Field(ge=0)


def build_schedule_tools(db, *, minimum_interval=HOUR // 2):
    """Four tools over the chapter's scheduler, plus a ledger of every refusal reason."""
    refusals = []

    def list_schedules(_):
        rows = db.connection.execute(
            "SELECT id,prompt,interval_seconds,next_due,enabled,deferred "
            "FROM assistant_jobs ORDER BY id"
        )
        return [dict(row) for row in rows]

    def schedule_job(args):
        try:
            schedule(
                db,
                args.identifier,
                "lucy",
                args.prompt,
                first_due=float(args.first_due),
                interval_seconds=args.interval_seconds,
            )
        except ValueError as error:
            refusals.append(("schedule_job", str(error)))
            raise
        return {"identifier": args.identifier, "next_due": args.first_due}

    def stop_job(args):
        try:
            unschedule(db, args.identifier)
        except ValueError as error:
            refusals.append(("stop_job", str(error)))
            raise
        return {"identifier": args.identifier, "enabled": 0}

    def run_pass(args):
        before = db.connection.execute("SELECT coalesce(max(seq),0) FROM events").fetchone()[0]
        created = tick(db, now=float(args.now))
        enqueued = db.connection.execute(
            "SELECT payload FROM events WHERE seq>? AND kind='assistant.job.enqueued' ORDER BY seq",
            (before,),
        ).fetchall()
        return {
            "now": args.now,
            "created": created,
            "coalesced": {
                json.loads(row["payload"])["job"]: json.loads(row["payload"])["coalesced"]
                for row in enqueued
            },
        }

    def before_write(call):
        interval = call.arguments.get("interval_seconds")
        if call.name == "schedule_job" and type(interval) is int and interval < minimum_interval:
            refusals.append(("schedule_job", "interval below the shop minimum"))
            raise PermissionError("interval below the shop minimum")

    tools = [
        ExecutableTool(
            "list_schedules",
            "Read every schedule row: identity, due time, enabled.",
            NoArguments,
            list_schedules,
        ),
        ExecutableTool(
            "schedule_job",
            "Create a repeating job. Identities are permanent; replace by stopping and creating.",
            ScheduleArguments,
            schedule_job,
            consequential=True,
        ),
        ExecutableTool(
            "stop_job",
            "Stop future ticks of a job; admitted work stays.",
            JobArguments,
            stop_job,
            consequential=True,
        ),
        ExecutableTool(
            "run_pass",
            "Run one scheduler pass at the given clock. Due jobs create one work item each.",
            PassArguments,
            run_pass,
            consequential=True,
        ),
    ]
    dispatcher = Dispatcher(
        tools, allowed=frozenset(tool.name for tool in tools), before_write=before_write
    )
    return dispatcher, refusals


schedule_db = fresh_schedule_database(SCHEDULE_DB)
schedule_dispatcher, schedule_refusals = build_schedule_tools(schedule_db)
for tool_schema in schedule_dispatcher.schemas():
    function = tool_schema["function"]
    print(function["name"], "requires", function["parameters"].get("required", []))
    print("   ", function["description"])
print("Consequential:", sorted(n for n, t in schedule_dispatcher.tools.items() if t.consequential))
SCHEDULE_MESSAGES = [
    {
        "role": "system",
        "content": (
            "You help Lucy run her ice cream shop's always-on assistant. Every time is an "
            f"integer of seconds on the shop's epoch clock; {SHOP_OPEN} is Friday 00:00 and "
            "the clock is an input you pass to a scheduler pass. Inspect the schedules before "
            "changing them. A schedule identity is permanent: to change a job, stop it and "
            "create a new identity. Finish with a short plain-words summary of what is "
            "scheduled and what the pass created."
        ),
    },
    {
        "role": "user",
        "content": (
            "Set up my morning brief at 07:00 every day and a stock check every hour. Then I "
            "changed my mind: make the brief 08:00, not 07:00. Finally show me what one "
            "scheduler pass creates if the laptop stays shut until Monday 09:00."
        ),
    },
]

The schemas are the model_json_schema() output you inspected earlier, wrapped in the {"type": "function", ...} envelope; strict=True means interval_seconds as the string "3600" is refused before any handler runs. Three tools are consequential, so the dispatcher would refuse them outright if no gate were attached. The gate reads one argument only: an interval under thirty minutes would consume the fifty-request daily allowance on its own (86400 / 1800 is 48), and _enqueue would start rejecting Lucy's real requests.

Run the loop against a recorded transcript

The turns below are what a model replied, written down. Turn one reads the empty schedule. Turn two creates the 07:00 brief and the hourly check. Turn three tries to move the brief by reusing its identity, which schedule refuses. Turn four stops the old job and creates morning-brief-0800. Turn five runs one pass at Monday 09:00, after a weekend with no passes at all. Turn six reads the schedules back, and turn seven answers.

Prediction: how many work items does the Monday pass create, and what does the record say was coalesced for each job? Do the number-line arithmetic from the chapter first: the brief was due Friday 08:00, the check Friday 01:00. Then write down where each next_due lands.

def job(identifier, prompt, first_due, interval_seconds):
    return {
        "name": "schedule_job",
        "arguments": {
            "identifier": identifier,
            "prompt": prompt,
            "first_due": first_due,
            "interval_seconds": interval_seconds,
        },
    }


RECORDED_TURNS = [
    scripted_turn(calls=[{"name": "list_schedules", "arguments": {}}]),
    scripted_turn(
        calls=[
            job("morning-brief", "Morning brief for Lucy", SHOP_OPEN + 7 * HOUR, DAY),
            job("stock-check", "Check stock against reorder points", SHOP_OPEN + HOUR, HOUR),
        ]
    ),
    scripted_turn(
        calls=[job("morning-brief", "Morning brief for Lucy", SHOP_OPEN + 8 * HOUR, DAY)]
    ),
    scripted_turn(
        calls=[
            {"name": "stop_job", "arguments": {"identifier": "morning-brief"}},
            job("morning-brief-0800", "Morning brief for Lucy", SHOP_OPEN + 8 * HOUR, DAY),
        ]
    ),
    scripted_turn(calls=[{"name": "run_pass", "arguments": {"now": MONDAY_0900}}]),
    scripted_turn(calls=[{"name": "list_schedules", "arguments": {}}]),
    scripted_turn(
        "Scheduled morning-brief-0800 daily at 08:00 and stock-check hourly; the 07:00 brief is "
        "stopped. The Monday 09:00 pass created two work items: one brief covering the four "
        "missed mornings and one stock check covering the missed hours. Next due: Tuesday "
        "08:00 and Monday 10:00."
    ),
]
LOOP_LIMITS = Limits(model_calls=10, tool_calls=16, seconds=120, output_tokens=4096)


def job_rows(db):
    rows = db.connection.execute(
        "SELECT id,next_due,enabled,interval_seconds FROM assistant_jobs ORDER BY id"
    )
    return {row["id"]: dict(row) for row in rows}


scripted_result = run_loop(
    ScriptedModel(RECORDED_TURNS), schedule_dispatcher, SCHEDULE_MESSAGES, limits=LOOP_LIMITS
)
print(describe(scripted_result, "scripted"))
scripted_trace = tool_trace(scripted_result)
scripted_jobs = job_rows(schedule_db)
print("Refusals:", schedule_refusals)
print("Jobs:", json.dumps(scripted_jobs, indent=1))
assert scripted_result.status == "COMPLETED"
assert [step["ok"] for step in scripted_trace] == [True, True, True, False, True, True, True, True]
assert scripted_trace[3]["result"] == "tool_failed"
assert schedule_refusals == [
    ("schedule_job", "schedule identity already exists; use a new identity for a replacement")
]
monday_pass = scripted_trace[6]["result"]
assert len(monday_pass["created"]) == 2
assert monday_pass["coalesced"] == {"morning-brief-0800": 3, "stock-check": 80}
assert scripted_jobs["morning-brief"]["enabled"] == 0
assert scripted_jobs["morning-brief"]["next_due"] == SHOP_OPEN + 7 * HOUR
assert scripted_jobs["morning-brief-0800"]["next_due"] == SHOP_OPEN + 4 * DAY + 8 * HOUR
assert scripted_jobs["stock-check"]["next_due"] == MONDAY_0900 + HOUR
ready_work = schedule_db.connection.execute(
    "SELECT origin FROM assistant_work WHERE status='READY' ORDER BY origin"
).fetchall()
assert len(ready_work) == 2 and all(row["origin"].startswith("job:") for row in ready_work)
# Every observation in the transcript carries the identifier of the request it answers.
assert all(m.get("tool_call_id") for m in scripted_result.messages if m["role"] == "tool")

The refused reschedule appears in the transcript as an ordinary failed observation and the loop carries on; the reason lives only in the notebook's own ledger, because invoke never copies an error message into what the model sees. Three missed mornings collapsed into one brief, and eighty missed hours into one check. The stopped 07:00 job kept its next_due untouched: unschedule stops future ticks without rewriting history.

Run the same loop against a live model

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

The live model is gpt-5.1 through the openai library, which Colab ships preinstalled; on a local kernel run %pip install openai once. One run costs a few thousand tokens. The model receives the same four schemas and system message. It may pick other identities or run the pass at a different clock; what it cannot do is create an interval under thirty minutes or make a pass advance a job by anything other than whole intervals, because those rules live in the gate and in tick.

Prediction: write the schedule a careful assistant should leave behind, then run the cell and compare the model's choices, or the recorded transcript when the run is scripted.

from sovereign_agent.llm_lab import DEFAULT_MODEL, build_model, resolve_api_key

schedule_db.close()
live_db = fresh_schedule_database(SCHEDULE_DB)
live_dispatcher, live_refusals = build_schedule_tools(live_db)
chosen_model = build_model(RECORDED_TURNS, model=DEFAULT_MODEL)
print("Model source:", chosen_model.source, "| key found:", resolve_api_key() is not None)
live_result = run_loop(chosen_model, live_dispatcher, SCHEDULE_MESSAGES, limits=LOOP_LIMITS)
print(describe(live_result, chosen_model.source))
print("Refusals:", live_refusals)
live_trace = tool_trace(live_result)
live_jobs = job_rows(live_db)
print("Jobs:", json.dumps(live_jobs, indent=1))
assert live_result.status in {"COMPLETED", "MODEL_CALL_LIMIT", "TOOL_LIMIT", "MODEL_FAILED"}
# Whatever the model did, every job row came from an accepted schedule_job and the gate held.
accepted = {
    step["arguments"]["identifier"]: step["arguments"]
    for step in live_trace
    if step["tool"] == "schedule_job" and step["ok"]
}
assert set(live_jobs) == set(accepted)
for identifier, row in live_jobs.items():
    first_due = accepted[identifier]["first_due"]
    interval = accepted[identifier]["interval_seconds"]
    assert interval >= HOUR // 2 and row["interval_seconds"] == interval
    # A pass only ever advances by whole intervals, never by "one interval per missed wakeup".
    assert row["next_due"] >= first_due and (row["next_due"] - first_due) % interval == 0
for step in live_trace:
    if step["tool"] == "run_pass" and step["ok"]:
        assert len(step["result"]["created"]) == len(step["result"]["coalesced"])
        assert all(count >= 0 for count in step["result"]["coalesced"].values())

Challenge: tighten the gate, then pause the runtime

Both rules are data. First rebuild the tools with minimum_interval=2 * HOUR and rerun the recorded transcript. Predict before running: which call is refused now, in which order do the two refusals appear in the ledger, and how many work items does the Monday pass create? The transcript did not change; the gate did.

Then rerun on a fresh ledger whose control row is paused, which is the state restore leaves behind in Chapter 19. tick honors that flag before it reads a single job. Predict what the pass creates and whether next_due moves. Notice that the script's final sentence is now wrong about what happened; the observations are right, and that is why the record keeps them apart. With a live key, rerun the live cell under either change and explain what the model did after its second job was refused.

In your notes, separate what the scripted run proves (the identity rule, the gate, coalescing arithmetic, the pause flag, transcript accounting) from what only a live run can show (the model's judgment about identities and clocks). The saved record keeps the two apart.

tight_db = fresh_schedule_database(COURSE_WORK / "ch10-llm-schedule-tight.sqlite")
tight_dispatcher, tight_refusals = build_schedule_tools(tight_db, minimum_interval=2 * HOUR)
tight_result = run_loop(
    ScriptedModel(RECORDED_TURNS), tight_dispatcher, SCHEDULE_MESSAGES, limits=LOOP_LIMITS
)
tight_trace = tool_trace(tight_result)
print("Tight gate:", tight_result.status, "refusals:", tight_refusals)
assert [step["ok"] for step in tight_trace] == [True, True, False, False, True, True, True, True]
assert [reason for _, reason in tight_refusals] == [
    "interval below the shop minimum",
    "schedule identity already exists; use a new identity for a replacement",
]
assert tight_trace[6]["result"]["coalesced"] == {"morning-brief-0800": 3}

paused_db = fresh_schedule_database(COURSE_WORK / "ch10-llm-schedule-paused.sqlite")
with paused_db.immediate() as connection:
    connection.execute("UPDATE assistant_control SET paused=1 WHERE id=1")
paused_dispatcher, paused_refusals = build_schedule_tools(paused_db)
paused_result = run_loop(
    ScriptedModel(RECORDED_TURNS), paused_dispatcher, SCHEDULE_MESSAGES, limits=LOOP_LIMITS
)
paused_trace = tool_trace(paused_result)
print("Paused runtime:", paused_result.status, "pass created:", paused_trace[6]["result"])
assert paused_trace[6]["result"] == {"now": MONDAY_0900, "created": [], "coalesced": {}}
assert job_rows(paused_db)["morning-brief-0800"]["next_due"] == SHOP_OPEN + 8 * HOUR

llm_lab_report = {
    "unit": "ch10-b",
    "scripted": {
        "status": scripted_result.status,
        "tool_calls": scripted_result.tool_calls,
        "refused": [reason for _, reason in schedule_refusals],
        "monday_pass": monday_pass["coalesced"],
        "next_due": {name: row["next_due"] for name, row in scripted_jobs.items()},
    },
    "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,
        "refused": [reason for _, reason in live_refusals],
        "jobs": live_jobs,
        "answer": live_result.answer[:400],
    },
    "tight_gate": {
        "minimum_interval": 2 * HOUR,
        "refused": [reason for _, reason in tight_refusals],
        "monday_pass": tight_trace[6]["result"]["coalesced"],
    },
    "paused": {"created": paused_trace[6]["result"]["created"]},
}
llm_report_path = COURSE_WORK / "ch10-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_db in (live_db, tight_db, paused_db):
    open_db.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.