Worker Crash Recovery: Unit A Exercises

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

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

This is one of two practical units for Chapter 13. 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. Reconstruct the full worker assertion: verify paused state and external authority marker, compare database and claim epochs, match the full running lease identity and enforce delegation deadline and parent cancellation.
3. Solve explain admission across an authority replacement 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/ch13-a folder. Rerunning setup restores the frozen support files and keeps your saved work. Restarting a kernel clears variables, not saved submission files. Source basis: Sovereign Agent 444c5f6. Some tasks use reviewed local subprocesses; they are not an OS sandbox.

Supplied offline setup and teaching files

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

Commit to a prediction before the examples

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

Reading the Python vocabulary used in this notebook

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

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

import json

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

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

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

from dataclasses import dataclass


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


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

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

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


def intro_double(value):
    return value * 2


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

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

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

from pathlib import Path
from tempfile import TemporaryDirectory

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

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

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.

A live worker may no longer own the work

A worker process can pause for a long time and resume after another worker has taken over. Its memory still says “I own this job.” The database may disagree. Concurrency means more than one execution can interact with shared state. Correctness must therefore depend on current shared evidence rather than one process's stale local belief.

A claim identifies the owner of a work item. A lease limits that ownership in time. A generation distinguishes successive claims. An epoch distinguishes a larger authority era, such as before and after restoring a database. A fence is a check that refuses a stale claim before the protected operation. These terms name different fields because one matching field cannot compensate for another stale one.

Isolate a generation mismatch

Keep work identity, worker name and time valid. Change only generation. If the stale claimant can still write, the generation check is missing. If you also expire its lease, another check could hide that defect. This is why a good failure experiment varies one protection at a time.

intro_current_claim = {"work": "opening", "worker": "worker-1", "generation": 8, "epoch": "era-b"}
intro_old_claim = {**intro_current_claim, "generation": 7}
intro_effects = []


def intro_guarded_append(claim, value):
    if any(
        claim[key] != intro_current_claim[key] for key in ("work", "worker", "generation", "epoch")
    ):
        raise PermissionError("claim is stale")
    intro_effects.append(value)


try:
    intro_guarded_append(intro_old_claim, "stale observation")
except PermissionError:
    pass
intro_guarded_append(intro_current_claim, "current observation")
assert intro_effects == ["current observation"]
print(intro_effects)

The positive case is essential. A function that refuses every write would prevent stale writes but also prevent useful work. The actual exercise observes durable transcript rows after both admitted and refused attempts, not just the returned error strings.

Restoring state creates an authority boundary

A restored database may contain a claim whose worker name and generation match a surviving old process. An external authority marker and a fresh epoch distinguish the restored era from the old one. Comparing only the database to itself would not expose every stale process. The chapter checks the control epoch, claim epoch and retained marker before accepting a protected write.

intro_restored_control = {"epoch": "after-restore", "paused": False}
intro_marker = "after-restore"
intro_claim_before_restore = {"epoch": "before-restore", "lease_until": 200.0}
intro_now = 100.0
intro_authorized = (
    not intro_restored_control["paused"]
    and intro_marker == intro_restored_control["epoch"]
    and intro_claim_before_restore["epoch"] == intro_restored_control["epoch"]
    and intro_claim_before_restore["lease_until"] > intro_now
)
print("Unexpired but authorized:", intro_authorized)
assert intro_authorized is False

The lease is still in the future, but the epoch is stale. Explain which equality fails and why extending the lease cannot repair it. The construction task also checks pause, full running-claim identity, delegation deadline and parent cancellation. A delegated child's authority can end because its parent no longer permits the work even while the child's own process remains alive.

Recovery is not replaying every model call

If durable records already identify pending work and known effects, recovery should use them. Starting another model conversation can invent a new intent or duplicate cost. Recovery here means reconciling current authority and retained work so an admitted worker can proceed from evidence. It does not mean a failed process gets to reclaim its former rights automatically.

The actual assert_current helper sits in the write path. Your implementation must be invoked where observations or other protected changes occur. Calling it once during startup would leave a gap: ownership can change between startup and a later write. A fence also cannot recall an external request already admitted before ownership changed; that interval requires the operation identity and reconciliation reasoning from the order chapter.

Before the exercise, create a case matrix varying epoch, generation, lease, marker, pause and parent cancellation one at a time. Keep the remaining conditions valid. For every refused case, record the durable row count before and after. Then test a fresh claim. The changed-constraint task asks you to make the decision explainable without letting one stale field hide another.

Understand the supplied execution interface

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

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

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

Main practical: construct, connect and challenge

Design question: A surviving process may have lost authority?

Reconstruct the full worker assertion: verify paused state and external authority marker, compare database and claim epochs, match the full running lease identity and enforce delegation deadline and parent cancellation.

You edit a complete function in a temporary copy of src/sovereign_agent/assistant_work.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

Call observe through the real write path, then replace the epoch while keeping the same database and process.

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 = 10
HANDOFF = Path("ch13-unit-a-handoff-v1.json")

Construct the complete mechanism

Implement assert_current 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_work.py. Draw the data path from the observed output through this function to its actual input or query. Then write your implementation from the contract.

implementation_source = """
def assert_current(connection: sqlite3.Connection, work: Claim, now: float | None = None) -> None:
    raise NotImplementedError("Construct this chapter mechanism")
"""

Hint 1: the design

A matching worker name and lease generation cannot rescue an old restore epoch.

Hint 2: the boundary

Inspect the parameters and the caller in src/sovereign_agent/assistant_work.py. Identify validation, durable state and the first externally observable effect. Preserve the existing surrounding helper contracts.

Hint 3: the structure

Read the control row and marker, reject any mismatch, query the complete claim identity with an unexpired lease, then check research-parent constraints.

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

Keep a fresh claimant writable while refusing stale epoch, expired lease, missing marker and changed role. Count transcript rows after every attempt.

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 ch11-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: This local fence cannot recall an external request already admitted before replacement.

exercise_report = {
    "unit": "ch13-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: Explain admission across an authority replacement

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(current, claim, now). Current contains paused, epoch, generation, work and worker. Claim contains matching identity fields and lease_until. Return True only if current is not paused, all four identity fields match, and lease_until > now. These inputs are already shape-validated. Equality at expiry is refused.

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(current, claim, now):
    raise NotImplementedError("Bind the full identity and unexpired lease")

import copy
import json

TRANSFER_CASES = [
    (
        "current claim",
        [
            {"paused": False, "epoch": "new", "generation": 4, "work": "opening", "worker": "w1"},
            {"epoch": "new", "generation": 4, "work": "opening", "worker": "w1", "lease_until": 20},
            10,
        ],
        True,
    ),
    (
        "old epoch",
        [
            {"paused": False, "epoch": "new", "generation": 4, "work": "opening", "worker": "w1"},
            {"epoch": "old", "generation": 4, "work": "opening", "worker": "w1", "lease_until": 20},
            10,
        ],
        False,
    ),
    (
        "old generation",
        [
            {"paused": False, "epoch": "new", "generation": 4, "work": "opening", "worker": "w1"},
            {"epoch": "new", "generation": 3, "work": "opening", "worker": "w1", "lease_until": 20},
            10,
        ],
        False,
    ),
    (
        "exact expiry",
        [
            {"paused": False, "epoch": "new", "generation": 4, "work": "opening", "worker": "w1"},
            {"epoch": "new", "generation": 4, "work": "opening", "worker": "w1", "lease_until": 20},
            20,
        ],
        False,
    ),
]


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

Extension: put a language model behind a claim that can go stale

assert_current above answered one question before every durable write: does this process still own this work? It read the control row and the marker, compared epochs, matched the whole claim identity against an unexpired lease, and refused on any mismatch. In production the process holding the claim is running a model, and the model has no idea that a restore happened in another terminal while it was composing its next reply.

This closing section puts a model behind that fence. Reads go straight to the ledger; the one durable write, record_finding, appends a line to the work item's transcript through observe, which calls assert_current inside its transaction. The before_write gate is the same assert_current, called once more so the notebook keeps the runtime's own refusal sentence, which the dispatcher would otherwise flatten to tool_failed. A model can read after it has lost authority. It cannot write, and nothing it says can restore its claim: only a fresh claim under the current epoch can.

Prediction: the next cell builds the dispatcher and prints the schemas. Write down which tool is consequential, which fields assert_current matches against the assistant_work row, and what a stale worker will still be able to observe through the read tools.

import json
import time
from dataclasses import dataclass, field, replace
from pathlib import Path

from pydantic import Field

from reference_organizations.store.agent import NoArguments, seed_lucy, shop_dispatcher
from sovereign_agent.agent_loop import Limits, run_loop
from sovereign_agent.assistant_service import backup, restore
from sovereign_agent.assistant_work import assert_current, claim, enqueue, observe
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


class FindingArguments(NoArguments):
    text: str = Field(min_length=1, max_length=200)


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


@dataclass
class RecoveryLab:
    db: Database
    work: object
    backup_path: Path
    dispatcher: Dispatcher = None
    ledger: list = field(default_factory=list)

    def control(self):
        return dict(
            self.db.connection.execute(
                "SELECT epoch,paused FROM assistant_control WHERE id=1"
            ).fetchone()
        )

    def work_row(self):
        return dict(
            self.db.connection.execute(
                "SELECT generation,owner,status FROM assistant_work WHERE id=?", (self.work.id,)
            ).fetchone()
        )

    def findings(self):
        return [
            (row["generation"], json.loads(row["message"])["content"])
            for row in self.db.connection.execute(
                "SELECT generation,message FROM assistant_transcript WHERE work_id=? ORDER BY seq",
                (self.work.id,),
            )
        ]


def build_tools(lab, work):
    """Tools for one claim: two reads, one fenced durable write."""
    db = lab.db

    def authority_gate(call):
        try:
            assert_current(db.connection, work)
        except PermissionError as error:
            lab.ledger.append((work.generation, call.name, str(error)))
            raise
        lab.ledger.append((work.generation, call.name, "allowed"))

    def record_finding(args):
        observe(db, work, {"role": "finding", "content": args.text})
        return {"recorded": args.text, "generation": work.generation}

    def work_status(_):
        control = lab.control()
        return {
            "paused": bool(control["paused"]),
            "claim_epoch_current": control["epoch"] == work.epoch,
            "claim_generation": work.generation,
            "work": lab.work_row(),
            "transcript_rows": len(lab.findings()),
        }

    tools = [
        shop_dispatcher(db).tools["list_stock"],
        ExecutableTool(
            "work_status",
            "Read whether this worker's claim is still current and what the work row holds.",
            NoArguments,
            work_status,
        ),
        ExecutableTool(
            "record_finding",
            "Append one short finding to the work item's durable transcript.",
            FindingArguments,
            record_finding,
            consequential=True,
        ),
    ]
    return Dispatcher(
        tools, allowed=frozenset(tool.name for tool in tools), before_write=authority_gate
    )


def open_work(tag):
    """A fresh shop ledger with one work item claimed by this kernel."""
    db = Database(fresh_file(COURSE_WORK / f"ch13-{tag}-shop.sqlite"))
    seed_lucy(db)
    enqueue(db, f"ch13:{tag}", "lucy", "Check the shop this morning")
    work = claim(db, "notebook-worker", ttl=3600)
    lab = RecoveryLab(db, work, fresh_file(COURSE_WORK / f"ch13-{tag}-backup.sqlite"))
    lab.dispatcher = build_tools(lab, work)
    return lab


scripted_lab = open_work("scripted")
for tool_schema in scripted_lab.dispatcher.schemas():
    function = tool_schema["function"]
    print(function["name"], "requires", function["parameters"].get("required", []))
    print("   ", function["description"])
print("Claim:", scripted_lab.work.owner, "generation", scripted_lab.work.generation)
print("Control row:", scripted_lab.control()["paused"], "paused |", scripted_lab.work_row())
WORK_MESSAGES = [
    {
        "role": "system",
        "content": (
            "You help Lucy run her ice cream shop. Read stock, then record each product that "
            "needs attention as one short finding. If a finding is refused, do not claim it was "
            "recorded: read work_status and tell Lucy what it shows. Finish with a short summary."
        ),
    },
    {"role": "user", "content": "Check the shop and record what needs attention this morning."},
]

Three tools, one consequential. record_finding accepts a single bounded text field; extra="forbid" means any extra field is refused before the gate or the handler runs. The gate and the handler both call assert_current with the same claim; a current claim passes twice, a stale one fails at the gate and the ledger keeps the runtime's own sentence.

Run the recorded transcript twice: before and after a restore

Turn one reads stock. Turn two records a finding. Turn three records one with an extra urgent field. Turn four records the corrected finding. Turn five reads work_status. Turn six answers. The first run uses the fresh claim. Then Lucy takes a backup with the real backup, and restores it with the real restore: a new epoch, a new marker, the work row reset to READY with its generation raised, the control row paused. This kernel still holds the old Claim, and the same six turns run again through it.

Prediction: for the first run, which call is refused and does the gate ever see it? For the second run, which calls succeed, which are refused, what sentence does the gate record, and how many transcript rows exist afterwards? The recorded final answer is the same both times. Decide before running whether it is true both times.

RECORDED_TURNS = [
    scripted_turn(calls=[{"name": "list_stock", "arguments": {}}]),
    scripted_turn(
        calls=[{"name": "record_finding", "arguments": {"text": "Vanilla is at 2 tubs; 6 needed."}}]
    ),
    scripted_turn(
        calls=[
            {
                "name": "record_finding",
                "arguments": {"text": "Strawberry is at 1 tub; 4 needed.", "urgent": True},
            }
        ]
    ),
    scripted_turn(
        calls=[
            {"name": "record_finding", "arguments": {"text": "Strawberry is at 1 tub; 4 needed."}}
        ]
    ),
    scripted_turn(calls=[{"name": "work_status", "arguments": {}}]),
    scripted_turn(
        "Vanilla and strawberry are below their reorder points; chocolate is fine. I submitted "
        "two findings. The work_status observation above shows whether the ledger accepted them."
    ),
]
LOOP_LIMITS = Limits(model_calls=8, tool_calls=16, seconds=120, output_tokens=4096)

scripted_result = run_loop(
    ScriptedModel(RECORDED_TURNS), scripted_lab.dispatcher, WORK_MESSAGES, limits=LOOP_LIMITS
)
print(describe(scripted_result, "scripted"))
print("Gate ledger:", scripted_lab.ledger)
scripted_trace = tool_trace(scripted_result)
assert scripted_result.status == "COMPLETED"
assert [step["ok"] for step in scripted_trace] == [True, True, False, True, True]
assert scripted_trace[2]["result"] == "invalid_arguments"
# The schema refusal never reached the gate: two gate decisions, both allowed.
assert [decision for _, _, decision in scripted_lab.ledger] == ["allowed", "allowed"]
assert scripted_lab.findings() == [
    (1, "Vanilla is at 2 tubs; 6 needed."),
    (1, "Strawberry is at 1 tub; 4 needed."),
]
assert scripted_trace[4]["result"]["claim_epoch_current"] is True

# Lucy backs the ledger up, then restores it. This kernel still holds the old claim.
old_epoch = scripted_lab.control()["epoch"]
backup(scripted_lab.db, scripted_lab.backup_path)
restore(scripted_lab.db, scripted_lab.backup_path)
print(
    "Epoch changed:",
    scripted_lab.control()["epoch"] != old_epoch,
    "| work row:",
    scripted_lab.work_row(),
)
stale_result = run_loop(
    ScriptedModel(RECORDED_TURNS), scripted_lab.dispatcher, WORK_MESSAGES, limits=LOOP_LIMITS
)
print(describe(stale_result, "scripted, stale claim"))
print("Gate ledger:", scripted_lab.ledger[2:])
stale_trace = tool_trace(stale_result)
assert stale_result.status == "COMPLETED"
assert [step["ok"] for step in stale_trace] == [True, False, False, False, True]
assert stale_trace[1]["result"] == "tool_failed" and stale_trace[3]["result"] == "tool_failed"
assert [decision for _, _, decision in scripted_lab.ledger[2:]] == [
    "runtime paused or replaced by restore",
    "runtime paused or replaced by restore",
]
# The reads still worked; the writes did not; the transcript is exactly what the snapshot held.
assert stale_trace[4]["result"] == {
    "paused": True,
    "claim_epoch_current": False,
    "claim_generation": 1,
    "work": {"generation": 2, "owner": None, "status": "READY"},
    "transcript_rows": 2,
}
assert len(scripted_lab.findings()) == 2
assert stale_result.answer == scripted_result.answer  # same prose, different truth

Same transcript, same prose, different truth. The first run wrote two rows under generation 1; the schema refusal never reached the gate. After the restore the reads still answered, both writes were refused with the runtime's own sentence, and work_status showed why: paused, epoch no longer current, work row at generation 2 with no owner. The recorded final sentence defers to that observation, the only honest move for a model that cannot tell from inside whether its claim is still good.

Run the same loop against a live model

Nothing changes except the model. build_model looks for an OPENAI_API_KEY in Colab's Secrets pane (the key icon in the left sidebar: add a secret with that name and switch on notebook access for it), then in the process environment for a local kernel. Without a key it returns the scripted model again and the record says so; a missing key is a normal condition, never a silent substitution. The live model is gpt-5.1 through the openai library, which Colab ships preinstalled; on a local kernel run %pip install openai once.

The live run uses a fresh ledger and claim, so every finding is admitted. What it proves is the accounting: every transcript row carries this claim's generation, and there are exactly as many rows as successful record_finding calls. The wording of the findings is the model's judgment.

Prediction: write the findings a careful assistant should record from the stock above. Then run the cell and compare, or read the recorded transcript when the run is scripted.

from sovereign_agent.llm_lab import DEFAULT_MODEL, build_model, resolve_api_key

live_lab = open_work("live")
chosen_model = build_model(RECORDED_TURNS, model=DEFAULT_MODEL)
print("Model source:", chosen_model.source, "| key found:", resolve_api_key() is not None)
live_result = run_loop(chosen_model, live_lab.dispatcher, WORK_MESSAGES, limits=LOOP_LIMITS)
print(describe(live_result, chosen_model.source))
print("Transcript rows:", live_lab.findings())
live_trace = tool_trace(live_result)
assert live_result.status in {"COMPLETED", "MODEL_CALL_LIMIT", "TOOL_LIMIT", "MODEL_FAILED"}
# A current claim is never refused by the fence, whatever the model asks.
assert all(decision == "allowed" for _, _, decision in live_lab.ledger)
assert live_lab.control()["paused"] == 0
# One transcript row per admitted write, each stamped with this claim's generation.
recorded = [step for step in live_trace if step["tool"] == "record_finding" and step["ok"]]
assert len(live_lab.findings()) == len(recorded)
assert all(generation == live_lab.work.generation for generation, _ in live_lab.findings())
assert [text for _, text in live_lab.findings()] == [step["arguments"]["text"] for step in recorded]
assert all(
    message.get("tool_call_id") for message in live_result.messages if message["role"] == "tool"
)

Challenge: resume, reclaim, and keep the old claim out

The restored ledger is paused and owned by nobody. Recovery is not replaying the model: Lucy resumes the control row (the full recovery entry point, which also reconciles the supplier account, is a later chapter; here the operator step is one row), and a fresh claim takes the work at generation 3. The next cell runs the recorded transcript through that fresh claim and then tries the old claim, an expired lease, a missing marker and a changed role against assert_current directly, counting transcript rows after every attempt.

Predict before running: how many rows carry generation 1 and how many carry generation 3? Which sentence refuses the old claim now that the ledger is no longer paused, and is it the same sentence as before? Which refusals share a sentence, and why does the fence not say more?

In your notes, separate what the scripted runs prove (the fence, the generation stamps, the four refusals) from what only a live run can show (whether the model reads work_status when a write fails, or asserts that it recorded something it did not). The saved record keeps the two apart, and the surviving limit stays: this fence cannot recall a request already admitted before the replacement.

with scripted_lab.db.immediate() as connection:
    connection.execute("UPDATE assistant_control SET paused=0 WHERE id=1")
replacement = claim(scripted_lab.db, "replacement-worker", ttl=3600)
assert replacement is not None and replacement.generation == 3
fresh_dispatcher = build_tools(scripted_lab, replacement)
fresh_result = run_loop(
    ScriptedModel(RECORDED_TURNS), fresh_dispatcher, WORK_MESSAGES, limits=LOOP_LIMITS
)
print("Fresh claimant:", fresh_result.status, "| rows:", scripted_lab.findings())
assert [step["ok"] for step in tool_trace(fresh_result)] == [True, True, False, True, True]
assert [generation for generation, _ in scripted_lab.findings()] == [1, 1, 3, 3]

refusals = {}
marker = scripted_lab.db.path.with_suffix(".authority")
attempts = {
    "stale_epoch": lambda: assert_current(scripted_lab.db.connection, scripted_lab.work),
    "expired_lease": lambda: assert_current(
        scripted_lab.db.connection, replacement, now=time.time() + 3601
    ),
    "changed_role": lambda: assert_current(
        scripted_lab.db.connection, replace(replacement, role="research")
    ),
    "missing_marker": lambda: assert_current(scripted_lab.db.connection, replacement),
}
epoch_text = marker.read_text()
for name, attempt in attempts.items():
    if name == "missing_marker":
        marker.unlink()
    try:
        attempt()
    except PermissionError as error:
        refusals[name] = str(error)
    finally:
        if name == "missing_marker":
            marker.write_text(epoch_text)
    assert len(scripted_lab.findings()) == 4, name
print("Refusals:", json.dumps(refusals, indent=2))
assert refusals == {
    "stale_epoch": "runtime paused or replaced by restore",
    "expired_lease": "worker claim expired or superseded",
    "changed_role": "worker claim expired or superseded",
    "missing_marker": "authority marker is unavailable",
}
assert_current(scripted_lab.db.connection, replacement)  # the fresh claimant still passes

llm_lab_report = {
    "unit": "ch13-a",
    "scripted": {
        "status": scripted_result.status,
        "refused": [step["result"] for step in scripted_trace if not step["ok"]],
        "gate": [list(entry) for entry in scripted_lab.ledger[:2]],
        "rows_after": 2,
    },
    "stale_claim": {
        "status": stale_result.status,
        "gate": [list(entry) for entry in scripted_lab.ledger[2:4]],
        "work_status": stale_trace[4]["result"],
    },
    "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,
        "findings": [text for _, text in live_lab.findings()],
        "generation": live_lab.work.generation,
        "answer": live_result.answer[:400],
    },
    "recovery": {
        "replacement_generation": replacement.generation,
        "rows_by_generation": [generation for generation, _ in scripted_lab.findings()],
        "refusals": refusals,
    },
}
llm_report_path = COURSE_WORK / "ch13-a-llm-lab-report-v1.json"
llm_report_path.write_text(json.dumps(llm_lab_report, indent=2, sort_keys=True), encoding="utf-8")
print("Saved evidence:", llm_report_path)
print("LLM_LAB_REPORT=" + json.dumps(llm_lab_report["session"], sort_keys=True))
for open_lab in (scripted_lab, live_lab):
    open_lab.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.