Events at random: queues, utilization and scheduled work

Draft chapter. The textbook guide covers setup and the supplied code. Practice in Exercise Book 10, then check Solutions 10 once you have tried it.

Lucy has learned to ask for a replenishment brief from her phone. Now she wants the brief before she asks. A second request follows: if vanilla becomes scarce during the day, prepare another draft. Repeating yesterday's prompt in a terminal would work while the builder was present. Lucy needs the host to notice these conditions and retain the resulting work while everyone is elsewhere.

Part A first measures how long work waits when events arrive at random. Part B then builds two producers for the durable work queue from Chapter 9. One observes the clock. The other observes stock. Neither calls a model. The existing worker consumes their records, loads the current session context and runs the bounded loop. A standard Linux service manager keeps that worker available after the terminal closes.

There is also a less obvious failure to repair. During this chapter's live experiment, the model requested a correct seven-tub draft worth $17.50, then described it as $15.00. A queue can preserve a wrong report perfectly. We will use the structured tool results to render the quantities and amounts Lucy sees, while retaining the original model response for evaluation.

Learning objectives

Part A measures queues. After it you should be able to:

• compute a worker's utilization, and explain why the queue grows without bound at or above 1;
• derive the Pollaczek–Khinchine mean wait, and explain why variable job lengths lengthen it;
• check Little's law on a measured run;
• size the arrival rate one worker can take for a target wait, and add the delay of a periodic rescan.

Part B builds the producers. After it you should be able to:

• implement a bounded scheduler pass;
• choose and test a missed-run policy;
• turn a persistent stock condition into one scoped work item per observed episode;
• connect both producers to an unattended worker.

You can distinguish process liveness, work admission, successful execution and a correct displayed result. You will also know which parts of a fixed-interval timer do not implement a local-calendar appointment.

The deliverable is an unattended draft-producing agent built on your own code. The checkpoint runs your learner file as a separate process without a prompt argument and waits for a persisted result. Its optional live modes use the local HTTP model or Claude in that process too. Purchases remain unavailable: the tools can only read stock and calculate drafts.

Part A: how long does an event wait?

Schedules and stock events put work on a queue, and one worker takes it off. When events arrive faster, work waits longer, but not in proportion. This part derives how long work waits when events arrive at random, measures it with a real model doing the work, and turns the result into a rule for how busy Lucy's worker may be.

The functions live in the chapter's learner file.

import json
import runpy

q = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch10_queueing_learner.py")
measured = json.loads(open("docs/evidence/book-ch10/ch10-queueing-receipt-v1.json").read())

Utilization

Suppose jobs arrive at an average rate of $\lambda$ per second, and each takes the worker $S$ seconds, with mean $E[S]$. The worker is busy a fraction

$ \rho = \lambda, E[S] $

of the time: its utilization. If $\rho \ge 1$, work arrives faster than it can be done and the queue grows without bound. Below 1 the queue empties sometimes, but how long a job waits depends on more than $\rho$.

The Pollaczek–Khinchine formula

Stock events do not arrive on a timetable. A standard model for arrivals that are independent of one another is the Poisson process: the gaps between arrivals are independent and exponentially distributed with mean $1/\lambda$. Under Poisson arrivals, a job that arrives waits for two things: the rest of the job in progress, and every job already queued ahead of it.

The first part holds a surprise. An arrival at a random moment is more likely to land during a long job than a short one, because long jobs occupy more of the time. This is the inspection paradox. Working it through, the mean remaining service an arrival finds is $\lambda E[S^2]/2$, which depends on the second moment $E[S^2]$ and not only the mean.

The second part follows from Chapter 19's Little's law. If the mean wait is $W$, then on average $\lambda W$ jobs are queued ahead, each needing $E[S]$. So

$ W = \frac{\lambda E[S^2]}{2} + \lambda W, E[S] \quad\Longrightarrow\quad W = \frac{\lambda, E[S^2]}{2,(1 - \rho)}. $

This is the Pollaczek–Khinchine formula. Writing $E[S^2] = E[S]^2 (1 + c^2)$, where $c$ is the coefficient of variation of the service time (its standard deviation over its mean), gives the form that shows what matters:

$ W = E[S] \cdot \frac{\rho}{1 - \rho} \cdot \frac{1 + c^2}{2}. $

Two lessons are visible:

• The factor $\rho/(1 - \rho)$ explodes as the worker gets busy. At $\rho = 0.5$ it is 1, at 0.8 it is 4 and at 0.95 it is 19.
• Variable job lengths multiply the wait. With identical jobs ($c = 0$) the factor $(1 + c^2)/2$ is one half. The textbook exponential service time has $c = 1$, and the factor is 1.

for rho in (0.5, 0.8, 0.9, 0.95):
    identical = q["pk_wait"](rho, 1.0, 1.0)
    exponential = q["exponential_wait"](rho, 1.0)
    print(rho, round(identical, 2), round(exponential, 2))

0.5 0.5 1.0
0.8 2.0 4.0
0.9 4.5 9.0
0.95 9.5 19.0

Measured: a real worker under random arrivals

Each job in the experiment is one real model call: qwen2.5:0.5b writing Lucy a note about a stock event. There were two workloads:

• notes: every job is a short note, capped at 120 tokens;
• mixed: one job in five is instead a detailed weekly report, capped at 600 tokens.

A pilot of 60 jobs per workload measured the service-time moments. Then 400 jobs arrived at random for each workload, at rates that the pilot said would make the worker 50% and 80% busy, and one worker served them in order.

uv run python book/textbook/experiments/profrod_sovereign_agent_textbook_ch10_queueing_v1.py \
    --out ch10-queueing-receipt.json

for name, pilot in measured["pilots"].items():
    print(name, pilot["mean_service"], pilot["second_moment"], "c^2 =", pilot["squared_cv"])

notes 0.4706 0.2393 c^2 = 0.08
mixed 0.6901 0.9855 c^2 = 1.07

Short notes took 0.47 seconds on average with little variation ($c^2 = 0.08$): the 120-token cap makes them nearly identical. The mixed pilot measured 0.69 seconds with $c^2 = 1.07$.

But only 6 of the mixed pilot's 60 jobs, 10%, took longer than 1.5 seconds, although one job in five was meant to be a long report. In the runs, 63 and 64 of 400 jobs took that long, about 16%, and the mean service time was 0.91 seconds. The pilot had underestimated the mean by about a quarter, and that mistake matters below.

for row in measured["rows"]:
    from_run = row["pk_wait_from_this_run"]
    from_run = "no steady state" if from_run is None else f"{from_run} s"
    print(
        f"{row['workload']:5} rho {row['target_utilization']}: "
        f"measured {row['measured_mean_wait']} s, "
        f"Pollaczek-Khinchine {row['pk_wait_from_pilot']} s "
        f"(from this run {from_run}), "
        f"exponential {row['exponential_wait_from_pilot']} s"
    )

notes rho 0.5: measured 0.573 s, Pollaczek-Khinchine 0.254 s (from this run 0.5 s), exponential 0.471 s
notes rho 0.8: measured 1.214 s, Pollaczek-Khinchine 1.017 s (from this run 1.503 s), exponential 1.883 s
mixed rho 0.5: measured 2.198 s, Pollaczek-Khinchine 0.714 s (from this run 2.208 s), exponential 0.69 s
mixed rho 0.8: measured 38.862 s, Pollaczek-Khinchine 2.856 s (from this run no steady state), exponential 2.76 s

Compare each stable run with the prediction made from the service times that run actually saw. The pairs are 0.57 seconds measured against 0.50 predicted, 1.21 against 1.50, and 2.20 against 2.21. The formula tracks all three. The largest gap, 20%, is in the busiest run, where a queue's average varies most from one run to the next. The exponential column assumes $c = 1$; it is shown for comparison and is a model of neither workload.

The predictions from the pilot did worse, for two different reasons:

• Rare stalls: in the first notes run, 4 jobs of 400 stalled for up to 5.6 seconds. Rare slow jobs dominate $E[S^2]$, so they roughly doubled the predicted wait, and the measured wait followed.
• An unlucky pilot: for the mixed workload, the pilot's low mean meant that a rate planned for 50% utilization gave 69%, and one planned for 80% gave 111%. At 111% there is no steady state: the queue grew throughout the run. Jobs waited 38.9 seconds on average and up to 73.7 seconds, where the plan had predicted 2.9.

Mixing reports into the queue also lengthened every wait. At 69% utilization, the mixed workload waited 2.2 seconds, nearly twice as long as the notes at 85%.

View full-size diagram

xychart-beta
    title "Mean wait before service in the three stable runs"
    x-axis "Run" ["notes, 57% busy", "notes, 85% busy", "mixed, 69% busy"]
    y-axis "Mean wait (seconds)" 0 --> 2.5
    bar "Measured" [0.573, 1.214, 2.198]
    bar "Pollaczek-Khinchine prediction" [0.5, 1.503, 2.208]

for row in measured["rows"]:
    number, product = row["littles_law_number"], row["littles_law_rate_times_stay"]
    print(row["workload"], row["target_utilization"], number, product)

notes 0.5 1.195 1.195
notes 0.8 2.894 2.931
mixed 0.5 2.361 2.359
mixed 0.8 42.547 48.419

In the three stable runs the two sides agree to within about 1%: 1.195 and 1.195, 2.894 and 2.931, and 2.361 and 2.359. Little's law needs only that the system is stable. In the overloaded run they disagree, 42.5 against 48.4, because it never reached a steady state: the queue was still growing when the last job arrived, and draining it stretched the measurement window.

How many events can one worker take?

Turn the formula around to plan capacity. If Lucy wants stock notes to wait no more than $W^*$ seconds on average, the highest arrival rate one worker can take is found by solving the Pollaczek–Khinchine formula for $\lambda$:

$ \lambda^* = \frac{2 W^}{E[S^2] + 2 W^ E[S]}. $

for name, pilot in measured["pilots"].items():
    rate = q["highest_rate"](2.0, pilot["mean_service"], pilot["second_moment"])
    busy = rate * pilot["mean_service"]
    print(name, round(60 * rate, 1), "jobs per minute, utilization", round(busy, 2))

notes 113.1 jobs per minute, utilization 0.89
mixed 64.1 jobs per minute, utilization 0.74

By these figures, a worker doing only notes can take about 113 a minute, at 89% utilization, and a mixed worker about 64 a minute, at 74%. For the mixed workload the figure is optimistic, because it uses the pilot's mean, which was a quarter too low. That is exactly how the run planned for 80% ended at 111%. Plan capacity from enough jobs to include the rare long ones, and leave margin.

A periodic rescan adds its own delay

The chapter's stock producer rescans on a timer rather than reacting to every change. An event that happens at a random moment waits, on average, half the rescan period before it is noticed, and then its time in the queue. The two add. A rescan every 60 seconds, like the NanoClaw sweep described later in this chapter, adds 30 seconds on average: far more than the queue itself at modest utilization. Event hints can remove that half-period; the rescan remains as the recovery path when a hint is lost.

The decision

• Keep the worker well below full utilization. Waits scale with $\rho/(1-\rho)$, so a worker that looks only moderately busy can already be slow to answer.
• Measure the distribution of job lengths, not only the mean. The same mean with more variation means longer waits for everyone.
• Plan from a pilot large enough to contain the rare jobs, and re-measure in production. A quarter's error in the mean turned a planned 80% into an overloaded queue.
• Keep long jobs away from short, urgent ones. A weekly report in the same queue makes every stock note wait behind it. Separate queues, or a separate worker, keep the short jobs short.

Part B: wake up for schedules and stock events

Part A measured what a queue does to waiting work. This part builds the producers that fill Lucy's queue: a clock that survives restarts and a stock scanner that turns a shortage into one episode of work. Both are ordinary Python on the code you have already built. The queue is your Chapter 8 queue, with Chapter 9's routes. The stock is your Chapter 4 store, and each turn runs through your Chapter 7 context and Chapter 3 loop.

Separate a wake-up from the work it creates

A process can be alive while doing no useful work. It can also be temporarily absent while its work remains safely recorded. A liveness observation answers whether a process responds; it does not mean that Lucy needs another stock brief. Our scheduler and stock scanner answer different questions about business work. Keeping those questions explicit stops a frequent health check from becoming a frequent model call.

The clock producer asks whether a stored due time has passed. The stock producer asks whether a product has entered a positive replenishment-need episode. Both write through one admission function into the Chapter 8 queue. The worker later claims an admitted item, performs the bounded model loop and persists its report. That separation gives us somewhere to apply admission limits before intelligence becomes expensive.

The distinction also makes failures reproducible. We can pass an explicit clock value into a scheduler test without sleeping. We can record a sale and run one scanner pass without running a language model. Only the final integration experiment needs the producer, worker and model together. A failure there can be traced to the observation, admission, execution or reporting stage.

View full-size diagram

flowchart LR
    T[Stored due time] --> J[Bounded clock pass]
    S[Current stock] --> C[Bounded condition pass]
    J --> Q[Durable work intake]
    C --> Q
    Q --> W[Existing worker and loop]
    W --> R[Persisted draft report]
    H[Service manager] --> W

Keep the stock where the worker can read it

Until now the shop tools read a fixed copy of Chapter 1's stock. A worker that wakes up on its own needs the current stock, including a sale recorded after the process started. We already built the right place for it: Chapter 4's store keeps each product's tubs and the append-only events that changed them. This chapter adds one small table of its own, the products with their reorder points and prices, and its own line of migrations, owned by wakeups, beside those of the memory, skills, queue and channel.

The following listings execute cumulatively from the course folder. They load your Chapter 9 channel, which brings Chapters 3 to 8 with it, and construct the new functions here. The chapter's learner file is the completed comparison copy.

import json
import math
import re
import runpy
import sqlite3
import tempfile
import threading
import time
from pathlib import Path
from typing import Any

CHANNEL = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch09_messaging_learner.py")
SKILLS, MEMORY, LOOP = CHANNEL["SKILLS"], CHANNEL["MEMORY"], CHANNEL["LOOP"]
STORE, TOOLS = MEMORY["STORE"], LOOP["shop_tools"]


WAKE_MIGRATIONS = {
    1: (
        # Chapter 4's `stock` table holds each product's tubs; this is what a tub is worth.
        "CREATE TABLE products ("
        " sku TEXT PRIMARY KEY, name TEXT NOT NULL,"
        " reorder_point INTEGER NOT NULL CHECK (reorder_point >= 0),"
        " unit_cost_cents INTEGER NOT NULL CHECK (unit_cost_cents >= 0))",
        "CREATE TABLE wake_control (id INTEGER PRIMARY KEY CHECK (id = 1), paused INTEGER NOT NULL)",
        "INSERT INTO wake_control (id, paused) VALUES (1, 0)",
        "CREATE TABLE wake_jobs ("
        " id TEXT PRIMARY KEY, session TEXT NOT NULL, prompt TEXT NOT NULL,"
        " interval_seconds INTEGER NOT NULL, next_due REAL NOT NULL,"
        " channel TEXT NOT NULL, recipient TEXT NOT NULL,"
        " enabled INTEGER NOT NULL DEFAULT 1, deferred INTEGER NOT NULL DEFAULT 0)",
        "CREATE TABLE stock_conditions ("
        " id TEXT PRIMARY KEY, session TEXT NOT NULL, subject TEXT NOT NULL,"
        " channel TEXT NOT NULL, recipient TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1,"
        " armed INTEGER NOT NULL DEFAULT 1, generation INTEGER NOT NULL DEFAULT 0)",
        # One enabled condition per product, whoever writes the table.
        "CREATE UNIQUE INDEX stock_condition_subject ON stock_conditions (subject) WHERE enabled = 1",
        # The product a piece of work is scoped to. Work without a row here is whole-shop work.
        "CREATE TABLE work_subjects (work_id INTEGER PRIMARY KEY, subject TEXT NOT NULL)",
        # Each turn's raw transcript, kept for evaluation beside the report it produced.
        "CREATE TABLE work_transcripts ("
        " work_id INTEGER PRIMARY KEY, status TEXT NOT NULL, messages TEXT NOT NULL)",
    ),
}


def open_shop(path: str | Path, *, capacity: int = 10):
    """Chapter 9's store and queue, with Chapter 4's stock tables and this chapter's own."""
    db, queue = CHANNEL["open_channel"](path, capacity=capacity)
    db.initialize()
    db.migrate("wakeups", WAKE_MIGRATIONS)
    return db, queue


def seed_shop(db) -> None:
    """Lucy's three products from Chapter 1, at Chapter 2's prices, counted in by stock events.
    Seeding twice changes nothing: each opening count is an event with a fixed identity."""
    with db.immediate() as connection:
        for row in TOOLS["SHOP"]["products"]:
            connection.execute(
                "INSERT OR IGNORE INTO products VALUES (?, ?, ?, ?)",
                (row["sku"], row["name"], row["reorder_point"], TOOLS["PRICES"][row["sku"]]),
            )
    for row in TOOLS["SHOP"]["products"]:
        db.apply(STORE["StockEvent"](f"opening:{row['sku']}", row["sku"], row["on_hand"], "count"))


def adjust_stock(db, event_id: str, sku: str, delta: int, reason: str) -> str:
    """A delivery or a sale, as a Chapter 4 stock event: "applied" or "duplicate"."""
    return db.apply(STORE["StockEvent"](event_id, sku, delta, reason))


def stock_rows(connection, subject: str | None = None) -> list[dict[str, Any]]:
    """Current stock with its replenishment need, read from the database at call time."""
    rows = connection.execute(
        "SELECT p.sku, p.name, s.tubs, p.reorder_point FROM products p"
        " JOIN stock s ON s.sku = p.sku WHERE (? IS NULL OR p.sku = ?) ORDER BY p.sku",
        (subject, subject),
    ).fetchall()
    return [
        {
            "sku": sku,
            "name": name,
            "on_hand": tubs,
            "reorder_point": point,
            "needed": max(0, point - tubs),
        }
        for sku, name, tubs, point in rows
    ]


temporary = tempfile.TemporaryDirectory(prefix="lucy-ch10-")
db, queue = open_shop(Path(temporary.name) / "agent.sqlite")
seed_shop(db)
seed_shop(db)
for row in stock_rows(db.connection):
    print(row["sku"], row["on_hand"], "needed", row["needed"])

SKU-CHOCOLATE 12 needed 0
SKU-STRAWBERRY 1 needed 4
SKU-VANILLA 2 needed 6

Seeding twice counts the stock once, because each opening count is a Chapter 4 event with a fixed identity. Deliveries and sales below are events too, so the stock a worker reads always has a history that explains it. The need is Chapter 2's: the reorder point minus the tubs on hand, bounded below by zero.

Register a job that can survive a restart

Our job record holds an identity, a session, a prompt, an interval, the next due time and an output route. The identity is immutable. Reusing it for different content would make old work sources ambiguous. Disabling a job preserves its row and earlier work; replacing it requires a new identity. This is a modest configuration policy that we can explain and test without inventing a general calendar language.

Use epoch seconds for stored due times and integer seconds for intervals. A job every 86,400 seconds follows a fixed UTC interval. It does not mean 8 a.m. in Chicago across a daylight-saving transition. For this volume the main path uses the fixed interval and states it plainly. A reader who needs local-calendar behavior must introduce a zone-aware next-occurrence calculation and decide what a skipped or repeated local time means.

Before writing the row, validate the values that later intake depends on. A very long job identity might fit SQLite and then fail every time it becomes part of a work source. That defect appeared during construction: accepting the schedule was easy, but executing its first tick was impossible. Validate at the configuration boundary so the operator receives a useful refusal immediately.

A route says where a job's report belongs: local output, or a Telegram bot account and a numeric recipient, as in Chapter 9. Admission is Chapter 8's with one change. When the queue is full, Chapter 8 refused a request and Chapter 9 answered the sender. A producer has no sender to answer, so admit stores nothing and returns None, and the producer keeps its own state for a later pass.

def validate_route(channel: str, recipient: str) -> None:
    """Local output, or a Telegram bot account and a positive numeric recipient."""
    if (
        not isinstance(channel, str)
        or not isinstance(recipient, str)
        or len(channel) > 200
        or len(recipient) > 200
        or not (
            (channel == "local" and not recipient)
            or (
                re.fullmatch(r"telegram:[A-Za-z0-9_-]+", channel)
                and recipient.isascii()
                and recipient.isdigit()
                and int(recipient) > 0
            )
        )
    ):
        raise ValueError("local output or an explicit positive Telegram recipient required")


def admit(connection, queue, source_id, session_id, text, route, subject=None) -> int | None:
    """Chapter 8's admission, strict, inside the caller's transaction.

    Returns the work id, or None when the queue is full: nothing is stored, so a producer can
    leave its own state as it was and try again on a later pass. The route and the subject are
    stored with the work."""
    row = connection.execute(
        "SELECT work_id, session_id, text FROM work WHERE source_id = ?", (source_id,)
    ).fetchone()
    if row is not None:
        if (row[1], row[2]) != (session_id, text):
            raise ValueError("intake identity reused for different content")
        return row[0]
    (open_work,) = connection.execute(
        "SELECT COUNT(*) FROM work WHERE state IN ('pending', 'running')"
    ).fetchone()
    if open_work >= queue.capacity:
        return None
    work_id = connection.execute(
        "INSERT INTO work (source_id, session_id, text, state) VALUES (?, ?, ?, 'pending')",
        (source_id, session_id, text),
    ).lastrowid
    connection.execute("INSERT INTO channel_routes VALUES (?, ?, ?)", (work_id, *route))
    if subject is not None:
        connection.execute("INSERT INTO work_subjects VALUES (?, ?)", (work_id, subject))
    return work_id


validate_route("local", "")
validate_route("telegram:teaching", "123")
try:
    validate_route("telegram:teaching", "Lucy")
except ValueError:
    print("Named recipient refused")

Named recipient refused

A route describes where a future result belongs. It does not authorize the recipient by itself. Chapter 9's delivery still checks the numeric operator allowlist at send time. A local route has no remote recipient and makes no network request. Scheduled phone reports must use the same bot-account channel and numeric recipient that the adapter uses for ordinary conversations.

def schedule(
    db,
    identifier: str,
    session: str,
    prompt: str,
    *,
    first_due: float,
    interval_seconds: int,
    channel: str = "local",
    recipient: str = "",
) -> None:
    """Epoch UTC intervals; missed runs coalesce to one, rather than a backlog storm."""
    if type(interval_seconds) is not int or interval_seconds < 1 or not math.isfinite(first_due):
        raise ValueError("finite due time and positive integral interval required")
    if (
        any(
            not isinstance(value, str) or not value.strip() or len(value) > 100
            for value in (identifier, session)
        )
        or not isinstance(prompt, str)
        or not prompt.strip()
        or len(prompt.encode()) > 16_384
    ):
        raise ValueError("bounded job identity, session and prompt required")
    validate_route(channel, recipient)
    with db.immediate() as connection:
        if connection.execute("SELECT 1 FROM wake_jobs WHERE id=?", (identifier,)).fetchone():
            raise ValueError(
                "schedule identity already exists; use a new identity for a replacement"
            )
        connection.execute(
            "INSERT INTO wake_jobs(id,session,prompt,interval_seconds,next_due,channel,recipient) "
            "VALUES (?,?,?,?,?,?,?)",
            (identifier, session, prompt, interval_seconds, first_due, channel, recipient),
        )


def unschedule(db, identifier: str) -> None:
    """Stop future ticks without deleting work that a prior tick already admitted."""
    with db.immediate() as connection:
        if (
            connection.execute("UPDATE wake_jobs SET enabled=0 WHERE id=?", (identifier,)).rowcount
            != 1
        ):
            raise ValueError("unknown schedule")


schedule(
    db,
    "morning",
    "lucy",
    "Prepare the opening replenishment draft.",
    first_due=100.0,
    interval_seconds=10,
)
try:
    schedule(db, "morning", "lucy", "Different instructions", first_due=200.0, interval_seconds=10)
except ValueError:
    print("Replacement needs a new identity")
print(db.connection.execute("SELECT next_due FROM wake_jobs").fetchone()[0])

Replacement needs a new identity
100.0

Notice what registration does not do: it does not enqueue work or call the model. The first due time may already be in the past, but admission belongs to the next scheduler pass. That gives the runtime a consistent place to honor a pause and apply workload limits. It also lets a test inspect a configured job before execution has altered it.

Choose what a late scheduler means

Suppose the first due time is 100 seconds, the interval is ten seconds, and the host next observes time 139. The nominal due times were 100, 110, 120 and 130. Four separate morning briefs would repeat nearly the same work and multiply model cost. We choose one current brief and record that three additional occurrences were coalesced. The next due time becomes 140.

The calculation uses the stored due time as its anchor. Advancing from the observation time instead would slowly shift the schedule whenever a pass ran late. The number of additional elapsed intervals is the floor of (now - next_due) / interval. Add one to that count when advancing the next due time. At the exact due time, the floor is zero: one occurrence is due and none are skipped.

Coalescing is appropriate for “tell me the current stock position.” It would be wrong for some other jobs, such as processing each dated settlement record. Those need their own recovery policy and data source. We are making a business decision about this particular report, not discovering a universally correct timer setting. The recorded coalesced field makes that choice visible to the operator.

View full-size diagram

sequenceDiagram
    participant Clock
    participant Pass as Scheduler pass
    participant DB as SQLite transaction
    Clock->>Pass: Observe 139, stored due 100, interval 10
    Pass->>DB: Begin immediate
    Pass->>DB: Admit one current brief
    Pass->>DB: Store next due 140 and coalesced count 3
    Pass->>DB: Commit together
    Clock->>Pass: Observe 139 again
    Pass->>DB: No due job remains

def paused(connection) -> bool:
    return bool(connection.execute("SELECT paused FROM wake_control WHERE id=1").fetchone()[0])


def set_paused(db, value: bool) -> None:
    with db.immediate() as connection:
        connection.execute("UPDATE wake_control SET paused=? WHERE id=1", (int(value),))


def tick(db, queue, *, now: float | None = None, maximum: int = 100) -> list[int]:
    """One scheduler pass: admit each due job's current occurrence and advance it, together."""
    now = time.time() if now is None else now
    if not math.isfinite(now) or type(maximum) is not int or not 1 <= maximum <= 1000:
        raise ValueError("invalid scheduler pass")
    created = []
    with db.immediate() as connection:
        if paused(connection):
            return []
        rows = connection.execute(
            "SELECT * FROM wake_jobs WHERE enabled=1 AND next_due<=? ORDER BY next_due,id LIMIT ?",
            (now, maximum),
        ).fetchall()
        for row in rows:
            work = admit(
                connection,
                queue,
                f"job:{row['id']}:{row['next_due']!r}",
                row["session"],
                row["prompt"],
                (row["channel"], row["recipient"]),
            )
            if work is None:
                # The queue is full: keep the due time, and record the deferral once.
                if not row["deferred"]:
                    MEMORY["record_event"](connection, "wake.job.deferred", {"job": row["id"]})
                    connection.execute("UPDATE wake_jobs SET deferred=1 WHERE id=?", (row["id"],))
                continue
            created.append(work)
            skipped = math.floor((now - row["next_due"]) / row["interval_seconds"])
            next_due = row["next_due"] + (skipped + 1) * row["interval_seconds"]
            connection.execute(
                "UPDATE wake_jobs SET next_due=?,deferred=0 WHERE id=?", (next_due, row["id"])
            )
            MEMORY["record_event"](
                connection,
                "wake.job.enqueued",
                {"job": row["id"], "work": work, "coalesced": skipped},
            )
    return created


created = tick(db, queue, now=139.0)
print("Created:", len(created))
print("Repeated observation:", len(tick(db, queue, now=139.0)))
print("Next due:", db.connection.execute("SELECT next_due FROM wake_jobs").fetchone()[0])
event = json.loads(
    db.connection.execute(
        "SELECT payload FROM memory_events WHERE kind='wake.job.enqueued'"
    ).fetchone()[0]
)
print("Coalesced:", event["coalesced"])

Created: 1
Repeated observation: 0
Next due: 140.0
Coalesced: 3

There is no HTTP call inside this transaction. SQLite commits the work, the next due time and the event together; the event goes into Chapter 5's audit log, which names what changed. If the process dies before that commit, a later pass observes the old due time and tries again. If it dies after the commit, the work is already durable and the next pass sees the advanced time. The model worker does not need to have started for this handoff to be complete.

The work's source includes the job identity and the previous due time. Admission uses that source to find an existing item if the same occurrence is presented again. The transaction supplies the primary local consistency boundary; the stable source adds an explicit identity for diagnosis and duplicate handling. Neither mechanism claims exactly-once behavior at an external supplier. We will address that different problem when the agent can place orders.

Make pause and capacity observable

A global pause must leave a due job due. An earlier scheduler implementation advanced the job while the runtime was paused, consuming the scheduled occurrence even though the operator expected work to wait. The repair is the pause check inside the same transaction as the due-row query. A concurrent configuration change cannot slip between those two local steps.

set_paused(db, True)
print("Paused pass:", tick(db, queue, now=179.0))
print("Due retained:", db.connection.execute("SELECT next_due FROM wake_jobs").fetchone()[0])
set_paused(db, False)
print("Resumed work:", len(tick(db, queue, now=179.0)))
unschedule(db, "morning")
unschedule(db, "morning")
print("Disabled pass:", tick(db, queue, now=999.0))
print("Earlier work retained:", db.connection.execute("SELECT count(*) FROM work").fetchone()[0])

Paused pass: []
Due retained: 140.0
Resumed work: 1
Disabled pass: []
Earlier work retained: 2

Capacity is the Chapter 8 queue's: at most capacity pending or running items, ten by default. A saturated periodic tick leaves its due time unchanged and records one wake.job.deferred event. Repeated passes create neither rejected work nor repeated notices. When capacity returns, the scheduler admits one current report, coalesces missed occurrences and advances from the original due time.

full, full_queue = open_shop(Path(temporary.name) / "full.sqlite", capacity=1)
seed_shop(full)
schedule(full, "morning", "lucy", "Prepare the opening brief.", first_due=100.0, interval_seconds=10)
with full.immediate() as connection:
    admit(connection, full_queue, "manual:1", "lucy", "Count the freezer.", ("local", ""))
for now in (139.0, 150.0):
    print("Pass at", now, "created:", tick(full, full_queue, now=now))
print("Due retained:", full.connection.execute("SELECT next_due FROM wake_jobs").fetchone()[0])
print(
    "Deferral events:",
    full.connection.execute(
        "SELECT count(*) FROM memory_events WHERE kind='wake.job.deferred'"
    ).fetchone()[0],
)
print("Work items:", full.connection.execute("SELECT count(*) FROM work").fetchone()[0])
full.close()

Pass at 139.0 created: []
Pass at 150.0 created: []
Due retained: 100.0
Deferral events: 1
Work items: 1

Stock conditions use the same strict admission: a full queue leaves the condition armed for a later pass. Clock jobs retain their earliest due time, while stock conditions retain their observed need. Both preserve pending work without manufacturing a queue of rejection reports.

Turn a stock shortage into one episode

A stock condition has one immutable subject SKU, a session, a route, an armed bit and an episode generation. At most one enabled condition watches a product, and the database's unique index enforces that for any writer. The scanner reuses the need calculation from the stock rows rather than a second threshold rule.

When need is positive and the condition is armed, the scanner admits one scoped work item, increments the generation and disarms the condition. Further scans of the same positive need do nothing. An observation of zero need rearms it. A later positive observation creates a new episode with a new source. Restarting a process does not erase the armed state or generation because both are in SQLite.

This is an observation-based definition of an episode. If stock briefly reaches the reorder point and falls again entirely between scans, the scanner cannot know that the recovery occurred. It will treat the observations as one continuing shortage. That limit is acceptable for our periodic stock fixture. A business that must account for every intermediate transition needs to consume the stock events themselves, with a durable position in that log, not a faster guess.

View full-size diagram

stateDiagram-v2
    [*] --> Armed
    Armed --> Armed: No need or admission refused
    Armed --> Disarmed: Positive need and work admitted
    Disarmed --> Disarmed: Positive need persists
    Disarmed --> Armed: Zero need observed

def watch(
    db,
    identifier: str,
    session: str,
    sku: str,
    *,
    channel: str = "local",
    recipient: str = "",
) -> None:
    """Register a stock condition on one product. Its scope and route never change."""
    if any(
        not isinstance(value, str) or not value.strip() or len(value) > 100
        for value in (identifier, session, sku)
    ):
        raise ValueError("bounded condition identity, session and product required")
    validate_route(channel, recipient)
    with db.immediate() as connection:
        if not connection.execute(
            "SELECT 1 FROM products p JOIN stock s ON s.sku = p.sku WHERE p.sku=?", (sku,)
        ).fetchone():
            raise ValueError("condition requires a known product with stock")
        if connection.execute(
            "SELECT 1 FROM stock_conditions WHERE subject=? AND enabled=1 AND id!=?",
            (sku, identifier),
        ).fetchone():
            raise ValueError("one active stock condition per product is allowed")
        previous = connection.execute(
            "SELECT * FROM stock_conditions WHERE id=?", (identifier,)
        ).fetchone()
        if previous:
            if (
                previous["session"],
                previous["subject"],
                previous["channel"],
                previous["recipient"],
            ) != (session, sku, channel, recipient):
                raise ValueError("condition identity already binds another scope or route")
            if not previous["enabled"]:
                connection.execute(
                    "UPDATE stock_conditions SET enabled=1,armed=1 WHERE id=?", (identifier,)
                )
        else:
            if connection.execute("SELECT count(*) FROM stock_conditions").fetchone()[0] >= 100:
                raise ValueError("teaching implementation supports at most 100 stock conditions")
            connection.execute(
                "INSERT INTO stock_conditions(id,session,subject,channel,recipient) "
                "VALUES (?,?,?,?,?)",
                (identifier, session, sku, channel, recipient),
            )


def unwatch(db, identifier: str) -> None:
    """Stop a condition; the work its episodes already admitted remains."""
    with db.immediate() as connection:
        if (
            connection.execute(
                "UPDATE stock_conditions SET enabled=0 WHERE id=?", (identifier,)
            ).rowcount
            != 1
        ):
            raise ValueError("unknown stock condition")


# Use a new database so the prior clock examples cannot consume the next claim.
db.close()
db, queue = open_shop(Path(temporary.name) / "stock.sqlite")
seed_shop(db)
watch(db, "vanilla-low", "lucy", "SKU-VANILLA")
watch(db, "vanilla-low", "lucy", "SKU-VANILLA")
print("Conditions:", db.connection.execute("SELECT count(*) FROM stock_conditions").fetchone()[0])
try:
    watch(db, "vanilla-low", "lucy", "SKU-STRAWBERRY")
except ValueError:
    print("Changed subject refused")

Conditions: 1
Changed subject refused

Registering the same condition again is harmless when its binding is unchanged. A different SKU or recipient cannot borrow that identity. The hundred-condition ceiling also bounds the amount of stock inspection in one pass. These limits make the implementation suitable for a small teaching catalog; they are not a claim about a large retailer's event throughput.

def scan(db, queue, *, now: float | None = None, maximum: int = 100) -> list[int]:
    """One condition pass: admit one piece of work per observed shortage episode."""
    now = time.time() if now is None else now
    if not math.isfinite(now) or type(maximum) is not int or not 1 <= maximum <= 100:
        raise ValueError("finite observation time and bounded scan required")
    emitted = []
    with db.immediate() as connection:
        if paused(connection):
            return []
        for condition in connection.execute(
            "SELECT * FROM stock_conditions WHERE enabled=1 ORDER BY id"
        ).fetchall():
            rows = stock_rows(connection, condition["subject"])
            if not rows:
                raise ValueError("condition inventory observation is unavailable")
            if rows[0]["needed"] == 0:
                connection.execute(
                    "UPDATE stock_conditions SET armed=1 WHERE id=?", (condition["id"],)
                )
                continue
            if not condition["armed"]:
                continue
            generation = condition["generation"] + 1
            work = admit(
                connection,
                queue,
                f"stock-condition:{condition['id']}:{generation}",
                condition["session"],
                f"Prepare a replenishment draft for {condition['subject']} from current stock. "
                "State USD amounts.",
                (condition["channel"], condition["recipient"]),
                subject=condition["subject"],
            )
            if work is None:
                continue  # The queue is full; leave this episode armed for a later pass.
            connection.execute(
                "UPDATE stock_conditions SET armed=0,generation=? WHERE id=?",
                (generation, condition["id"]),
            )
            MEMORY["record_event"](
                connection,
                "stock.condition.triggered",
                {
                    "condition": condition["id"],
                    "subject": condition["subject"],
                    "generation": generation,
                    "work": work,
                },
            )
            emitted.append(work)
            if len(emitted) >= maximum:
                break
    return emitted


print("First scan:", len(scan(db, queue)))
print("Same shortage:", len(scan(db, queue)))
row = db.connection.execute(
    "SELECT w.source_id, s.subject FROM work w JOIN work_subjects s ON s.work_id = w.work_id"
).fetchone()
print(row["subject"], row["source_id"])

First scan: 1
Same shortage: 0
SKU-VANILLA stock-condition:vanilla-low:1

If the queue is full, admit returns None before storing anything; the scanner leaves the episode armed. Other failures remain errors: missing stock should not masquerade as a healthy level. Work, its subject, the episode state and the trigger event share the transaction, so they do not disagree after a local crash.

The maximum argument bounds newly emitted items, while the hundred-condition registration limit bounds observations. These are different quantities. A pass may inspect healthy or already-disarmed products without emitting anything. Giving each bound a name makes it easier to reason about the cost of an idle pass and the size of a sudden burst of shortages.

Carry the subject through the existing loop

The generated prompt names vanilla, but a prompt is not the scope boundary. The work's immutable subject reaches the tools. shop_tools builds Chapter 2's three tools, through Chapter 2's dispatcher, over the database's current stock. With a subject, stock results are filtered to it, and a quote or draft for another product is refused by the tool.

This matters if the model remembers a previous whole-shop brief. Strawberry may still be low, but a vanilla episode is not authority to act on strawberry. A tool request outside the subject is refused independently of the model's wording. The same principle will matter more when tools create external effects. Here we can inspect it safely with read-only data and draft calculations.

def shop_tools(db, subject: str | None = None):
    """Chapter 2's three tools over the database's current stock, through Chapter 2's dispatcher.

    With a subject, the tools see and draft only that product: a request for another product is
    refused by the tool, whatever the model was told or remembers."""

    def in_scope(sku: str) -> None:
        if subject is not None and sku != subject:
            raise PermissionError("this work is scoped to another product")

    def stock(_):
        return stock_rows(db.connection, subject)

    def quote(args):
        in_scope(args.sku)
        row = db.connection.execute(
            "SELECT unit_cost_cents FROM products WHERE sku = ?", (args.sku,)
        ).fetchone()
        if row is None:
            raise KeyError("unknown product")
        return {
            "sku": args.sku,
            "supplier": "lucy-local",
            "currency": "USD",
            "unit_cost_cents": row[0],
        }

    def draft(args):
        in_scope(args.sku)
        rows = stock_rows(db.connection, args.sku)
        if not rows or rows[0]["needed"] <= 0 or args.quantity != rows[0]["needed"]:
            raise ValueError("draft quantity must equal the current positive need")
        price = quote(args)
        return {
            **price,
            "quantity": args.quantity,
            "total_cents": args.quantity * price["unit_cost_cents"],
            "status": "DRAFT",
        }

    tool = TOOLS["ExecutableTool"]
    registered = [
        tool("list_stock", "Read stock and calculated need.", TOOLS["NoArguments"], stock),
        tool("supplier", "Read supplier price in USD cents.", TOOLS["ProductArguments"], quote),
        tool("draft_order", "Calculate a draft; never purchases.", TOOLS["DraftArguments"], draft),
    ]
    return TOOLS["Dispatcher"](registered, allowed=frozenset(t.name for t in registered))


ToolCall = LOOP["ToolCall"]
vanilla_only = shop_tools(db, "SKU-VANILLA")
listed = vanilla_only.invoke(ToolCall(id="s1", name="list_stock", arguments={}))
print("Visible:", [row["sku"] for row in listed["value"]])
arguments = {"sku": "SKU-STRAWBERRY", "quantity": 4}
print("Strawberry draft:", vanilla_only.invoke(ToolCall(id="s2", name="draft_order", arguments=arguments)))
print("Tools:", sorted(vanilla_only.tools))

Visible: ['SKU-VANILLA']
Strawberry draft: {'ok': False, 'error': 'tool_failed'}
Tools: ['draft_order', 'list_stock', 'supplier']

A draft reads the stock again when it is calculated, so its quantity must equal the need at that moment. None of the three tools can purchase anything; spending needs the permission boundary of the next chapter.

Failure experiment: repair a wrong explanation

The first live unattended run of this chapter, on a local model, reached the correct tool result for the second episode: seven vanilla tubs, 250 cents each, total 1,750 cents. Its final model sentence instead reported $15.00, the amount of the first episode still in the session's history. The old checkpoint checked the tool observation and would have accepted the run. Reading the final response exposed a gap between the numerical evidence and what Lucy would see.

We do not fix this by instructing the model to be more careful with arithmetic. The deterministic tool already did the arithmetic correctly. We need a presentation boundary that uses that result. For completed draft-producing turns, render the latest successful draft for each SKU. Keep the raw model transcript available for evaluation, but do not ask it to restate monetary facts in the delivered draft summary.

def draft_report(messages: list[dict[str, Any]]) -> str | None:
    """Render latest successful draft estimates; model narration is not arithmetic."""
    names = {
        call["id"]: call["function"]["name"]
        for message in messages
        for call in message.get("tool_calls", [])
    }
    latest: dict[str, tuple[int, int]] = {}
    for message in messages:
        if message["role"] != "tool" or names.get(message["tool_call_id"]) != "draft_order":
            continue
        observation = json.loads(message["content"])
        if observation.get("ok") is not True:
            continue
        value = observation["value"]
        sku, quantity, amount = value.get("sku"), value.get("quantity"), value.get("total_cents")
        if (
            not isinstance(sku, str)
            or not 1 <= len(sku) <= 100
            or type(quantity) is not int
            or quantity <= 0
            or type(amount) is not int
            or amount < 0
            or value.get("currency") != "USD"
        ):
            raise ValueError("draft observation lacks validated quantity or USD amount")
        # Recalculating a draft is not creating another purchase. Display only
        # the latest estimate for each SKU.
        latest[sku] = (quantity, amount)
    if not latest:
        return None
    lines = ["Draft estimates:"]
    for sku, (quantity, amount) in sorted(latest.items()):
        label = json.dumps(sku, ensure_ascii=False)
        lines.append(f"- {label}: {quantity} tubs, ${amount // 100}.{amount % 100:02d} USD.")
    total = sum(amount for _, amount in latest.values())
    lines.append(f"Total: ${total // 100}.{total % 100:02d} USD.")
    return "\n".join(lines)


assertion = [{"role": "assistant", "content": "I drafted seven tubs for $15.00."}]
print("Report from assertion:", draft_report(assertion))

Report from assertion: None

The function reads tool observations whose call names are draft_order; a model sentence claiming to have drafted something is not enough. It rejects malformed amounts, quantities and currency. Repeated calculations for one SKU replace that SKU's earlier estimate instead of being added as extra orders. JSON quoting keeps a product label containing a newline from inserting another report line. Integer cents remain integers until formatting.

The worker is Chapter 9's turn with three changes. It builds the tools from the work's subject; it takes the report from draft_report when the loop completed with drafts, and the model's answer otherwise; and it keeps the raw transcript beside the report. The claim is Chapter 9's, so two turns of one session never run at once, and the report is Chapter 8's, created in the same transaction that finishes the work.

def run_next(db, queue, model, *, worker_id: str = "wake-worker") -> dict[str, Any] | None:
    """Claim the next eligible work and run one turn: the session's context, Chapter 3's loop over
    the tools its subject allows, and a report built from the draft observations.

    The raw transcript is kept beside the report; the model's own final sentence is kept in the
    transcript and Chapter 5's history, never used as the amounts Lucy reads."""
    assignment = CHANNEL["claim"](db, worker_id)
    if assignment is None:
        return None
    row = db.connection.execute(
        "SELECT subject FROM work_subjects WHERE work_id = ?", (assignment.work_id,)
    ).fetchone()
    subject = row[0] if row else None
    dispatcher = shop_tools(db, subject)
    revision = MEMORY["memory_revision"](db, assignment.session_id)
    messages = SKILLS["context"](
        db, assignment.session_id, assignment.text, allowed=dispatcher.allowed
    )
    result = LOOP["run_loop"](model, dispatcher, messages)
    report = draft_report(result.messages) if result.status == "COMPLETED" else None
    answer = report or result.answer or "The agent stopped: " + result.status
    with db.immediate() as connection:
        connection.execute(
            "INSERT INTO work_transcripts VALUES (?, ?, ?)",
            (assignment.work_id, result.status, json.dumps(result.messages)),
        )
    report_id = queue.finish(assignment, answer)
    MEMORY["record_result"](db, assignment.session_id, assignment.text, answer, revision)
    return {
        "work": assignment.work_id,
        "subject": subject,
        "status": result.status,
        "report": report_id,
        "answer": answer,
        "messages": result.messages,
    }


first = run_next(db, queue, SKILLS["OfflineShopModel"]())
print("First work:", first["status"])
print(first["answer"])
adjust_stock(db, "delivery-1", "SKU-VANILLA", 6, "received")
print("Healthy scan:", scan(db, queue))
adjust_stock(db, "sales-1", "SKU-VANILLA", -7, "sold")
print("New episode:", len(scan(db, queue)))
print(
    "Episode generation:",
    db.connection.execute("SELECT generation FROM stock_conditions").fetchone()[0],
)

First work: COMPLETED
Draft estimates:
- "SKU-VANILLA": 6 tubs, $15.00 USD.
Total: $15.00 USD.
Healthy scan: []
New episode: 1
Episode generation: 2

The delivery and the sale are Chapter 4 stock events, not a receiving interface. They let us test the exact transition from two tubs to eight and then to one. A draft itself changes no stock and creates no supplier order. Keeping that distinction visible prevents a plausible replenishment report from being mistaken for a physical delivery.

Now replay the failure. The model below requests the right draft and then misstates the total, as the live run did.

class MisreportingModel(SKILLS["OfflineShopModel"]):
    """Requests the correct drafts, then misstates the total."""

    def complete(self, messages, tools, *, timeout, max_output_tokens):
        turn = super().complete(
            messages, tools, timeout=timeout, max_output_tokens=max_output_tokens
        )
        return turn if turn.calls else LOOP["ModelTurn"]("I drafted seven tubs for $15.00.")


second = run_next(db, queue, MisreportingModel())
print("Model said:", second["messages"][-1]["content"])
print(second["answer"])
stored = db.connection.execute(
    "SELECT body FROM reports WHERE report_id = ?", (second["report"],)
).fetchone()[0]
print("Stored report is the one shown:", stored == second["answer"])

Model said: I drafted seven tubs for $15.00.
Draft estimates:
- "SKU-VANILLA": 7 tubs, $17.50 USD.
Total: $17.50 USD.
Stored report is the one shown: True

The test follows the data to the boundary where the report would leave the process: the stored report is what Chapter 9's delivery sends. The wrong sentence is still in the transcript for evaluation. This repair covers structured draft facts, not every factual claim in free-form responses, and a draft remains an estimate from the observations made during that turn.

We then ran the checkpoint on Claude, with the parent and the unattended worker both using the model: Claude Haiku 4.5 and Claude Sonnet 5.5, once each, for $0.08 in all. Both passed every check, and in all six turns the model's own final answer stated the same total as the rendered report. Sonnet 5.5's unattended answer also mentioned the earlier $15.00 drafts from the session history, correctly calling them older. The history that misled the local model was there for Claude too. The rendered report does not depend on either model reading it correctly. The receipt records each turn's tools, answer, report, tokens and cost.

View full-size diagram

flowchart LR
    M[Model requests a draft] --> T[Validated deterministic tool]
    T --> O[Structured observation]
    O --> F[Draft report formatter]
    F --> D[Durable result and channel text]
    O --> N[Model narration]
    N --> E[Retained evaluation transcript]

Run the passes without a person prompting

serve is the bridge between the producers and the worker. On each pass it runs tick, runs scan, performs at most one work item and, when a bot is configured, delivers pending reports through Chapter 9's deliver_one. It then waits on a stop event, or not at all if it just ran something. No eligible work means no model call. A model turn can delay the next pass, so this is not a real-time scheduler.

The loop catches expected transport, validation and database errors and waits with bounded exponential backoff before another pass. Successful passes reset that failure counter. Sleeping does not prove that a failed task was repaired; the work stays in the queue either way. Before its first pass, serve finishes any turn that its own worker label left running when an earlier process stopped, with a report asking Lucy to ask again, using Chapter 9's release_interrupted. Without that, a killed turn would hold Lucy's session forever.

def serve(
    db,
    queue,
    model_factory,
    *,
    stop: threading.Event,
    interval: float = 1.0,
    worker_id: str = "wake-worker",
    bot=None,
    operators: frozenset[int] = frozenset(),
    log=print,
) -> None:
    """Tick, scan, run at most one item and deliver, until `stop` is set. Nobody prompts it.

    An idle pass calls no model. An expected failure waits with bounded exponential backoff and
    does not end the loop; work already admitted stays in the database either way. Work this
    worker label left running when a previous process stopped is finished with a report first."""
    for work_id in CHANNEL["release_interrupted"](db, queue, worker_id):
        log(f"work {work_id} interrupted by an earlier stop; reported")
    failures = 0
    while not stop.is_set():
        try:
            tick(db, queue)
            scan(db, queue)
            ran = run_next(db, queue, model_factory(), worker_id=worker_id)
            if ran is not None:
                log(f"work {ran['work']} finished: {ran['status']}")
            if bot is not None:
                while (outcome := CHANNEL["deliver_one"](db, bot, operators)) is not None:
                    log(f"delivery: {outcome}")
            failures = 0
        except (OSError, ValueError, sqlite3.Error, STORE["StoreError"]) as error:
            failures += 1
            ran = None
            log(f"pass failed ({type(error).__name__}); waiting before the next pass")
        stop.wait(min(60.0, interval * 2**failures) if failures else (0 if ran else interval))


schedule(
    db,
    "evening",
    "lucy",
    "Prepare the evening replenishment draft.",
    first_due=time.time(),
    interval_seconds=86_400,
)
stop = threading.Event()
threading.Timer(1.0, stop.set).start()
serve(db, queue, SKILLS["OfflineShopModel"], stop=stop, interval=0.05)
print(db.connection.execute("SELECT body FROM reports ORDER BY work_id DESC").fetchone()[0])
db.close()
temporary.cleanup()

work 3 finished: COMPLETED
Draft estimates:
- "SKU-STRAWBERRY": 4 tubs, $11.00 USD.
- "SKU-VANILLA": 7 tubs, $17.50 USD.
Total: $28.50 USD.

The learner file is also a program. python book/textbook/learner/profrod_sovereign_agent_ch10_wakeups_learner.py serve --root DIR opens DIR/agent.sqlite, installs handlers so that SIGTERM or Ctrl-C sets the stop event, and serves until stopped, printing STOPPED on a clean exit. It takes no prompt. With --live it uses your Chapter 1 local model; with --claude MODEL it uses Claude, with the key read from the environment or a .env file, never from an argument. If SOVEREIGN_AGENT_TELEGRAM_TOKEN and SOVEREIGN_AGENT_OPERATORS are set, it delivers reports routed to that bot.

Run the checkpoint from the course folder:

uv run python book/textbook/checkpoints/profrod_sovereign_agent_ch10_schedules_stock_events_checkpoint.py
uv run python book/textbook/checkpoints/profrod_sovereign_agent_ch10_schedules_stock_events_checkpoint.py --claude claude-haiku-4-5-20251001 --transcript

The default uses the deterministic model fixture. --live uses the local model from Chapter 1, and --claude uses Claude. The checkpoint first tests routes, identities, capacity, scope and the report renderer without a model. It then uses a fresh database with the evaluated opening skill: it tests coalescing and pause, runs the morning brief, completes the first stock episode, records a delivery and then a sale that lowers vanilla to one tub. It also schedules an evening job and leaves a claimed turn running, as a killed worker would. Crucially, the parent closes its database without scanning the second shortage.

The checkpoint then starts the learner file as a separate process with no prompt argument and no bot credential. Its tools can only read and draft. The child must release the interrupted turn, find the evening job and the stored condition, create generation two, run both turns and persist a seven-tub draft worth $17.50. The parent observes SQLite for a bounded time, checks the transcript and the stored reports, then stops the child and requires its clean exit. This proves the producer-to-worker connection across a process boundary.

Keep the worker available after the terminal closes

On Google Colab, the runtime itself stops when the session ends, so there is no service to keep: run serve in a cell with a stop event on a timer, as Chapter 9 ran its channel, and copy the database to Drive to keep it. A host that must keep working while you are away needs a service manager. Use one Linux user service for the teaching deployment. Chapter 19 expands maintenance, backups and upgrades; here we need a minimal operational path now.

Clone the course on a Linux host with a user systemd manager and run uv sync in the course folder, as in Chapter 1. Use an absolute state directory, separate from the checkout. The following prepares a new one with your own functions: it opens the store, counts the stock in, activates the evaluated opening skill, schedules a daily brief and watches vanilla. It refuses to reuse an existing directory.

mkdir -m 700 "$HOME/lucy-agent" &&
install -m 600 /dev/null "$HOME/lucy-agent/agent.env" &&
uv run python - "$HOME/lucy-agent" <<'EOF'
import runpy, sys, time
from pathlib import Path

wake = runpy.run_path("book/textbook/learner/profrod_sovereign_agent_ch10_wakeups_learner.py")
db, queue = wake["open_shop"](Path(sys.argv[1]) / "agent.sqlite")
wake["seed_shop"](db)
wake["CHANNEL"]["activate_opening_skill"](db)
wake["schedule"](db, "morning-v1", "lucy", "Prepare the opening replenishment draft.",
                 first_due=time.time(), interval_seconds=86_400)
wake["watch"](db, "vanilla-low", "lucy", "SKU-VANILLA")
db.close()
EOF

The first due time is now, so this creates a current brief and then repeats at a fixed UTC interval; it does not silently choose Lucy's local morning time. Leave agent.env empty for the offline model. For a live model, add the settings your successful interactive run used, and for Claude the key; add the bot credential and numeric allowlist only if you want reports on the phone. The environment file belongs to the operator and stays outside the repository.

Then write the unit yourself. It is short enough to read in full, and every line is a decision. Replace /home/lucy with your home directory:

# ~/.config/systemd/user/lucy-agent.service
[Unit]
Description=Lucy's agent: schedules and stock conditions

[Service]
WorkingDirectory=/home/lucy/profrodai-resources/courses/sovereign-agent-book
EnvironmentFile=/home/lucy/lucy-agent/agent.env
ExecStart=/home/lucy/profrodai-resources/courses/sovereign-agent-book/.venv/bin/python book/textbook/learner/profrod_sovereign_agent_ch10_wakeups_learner.py serve --root /home/lucy/lucy-agent
Restart=on-failure
RestartSec=10
TimeoutStopSec=90

[Install]
WantedBy=default.target

ExecStart names the course environment's Python and your learner file directly, so the service runs exactly the code you tested. Restart=on-failure restarts a crashed process after ten seconds, and it is safe to restart because the work, the schedule and the stock conditions are in SQLite. TimeoutStopSec gives a turn in progress ninety seconds to finish after SIGTERM before the manager kills it; a killed turn is released with a report on the next start. Enable and start it with systemctl --user daemon-reload and systemctl --user enable --now lucy-agent.service.

A user service must also have an appropriate user-manager lifetime to continue after logout and return at boot. Configure lingering for the dedicated teaching user through the host's administration path, then verify it with loginctl show-user "$USER" -p Linger. Close the original shell, reconnect, and inspect both systemctl --user show lucy-agent.service and the reports. Service activity alone is not evidence that the brief completed.

The upstream loginctl manual source, checked on 7 September 2026, documents that enabling lingering starts the user's manager at boot and retains it after logout. This is a host configuration step; writing a service unit alone does not establish it.

For the live deployment, confirm the model endpoint is reachable from the Linux host itself. A model on the builder's laptop is a dependency that may disappear when that laptop sleeps. The service manager restarts a failed agent process; it cannot make an unavailable model server, network or Telegram service available. That is the concrete availability boundary behind the book's “always-on” promise.

Clock and stock work have different sources, so the initial schedule and initial shortage can both create drafts. That is an intentional two-trigger demonstration, not duplicate delivery of one event. Disable the demonstration after inspection with unschedule(db, "morning-v1") and unwatch(db, "vanilla-low") on the same database. The already admitted work remains available for inspection.

Compare a periodic rescan with event hints

NanoClaw's src/host-sweep.ts, pinned to commit acc69a70962af6707aa8a6abba699bdaa7da95f8 and inspected on 7 September 2026, combines a periodic session sweep with event-driven enqueues. The source describes events as hints and the rescan as a way to recover when an event is lost. It queues active sessions and singleton duties, waits for the current queue to become idle, then schedules another sweep after a 60-second interval. See the pinned source.

That is the authors' documented recovery rationale. Our interpretation is that it favors rereading durable state over treating an in-memory event notification as the only evidence that something changed. We make a related, smaller choice: each stock pass rereads the database. We do not implement NanoClaw's session driver, keyed queue or event feeds. An experiment that could change our choice is a measured requirement for lower notification latency across a much larger catalog.

In a coding agent

Claude Code's in-session scheduler is a queue with one server, and this chapter's arithmetic applies to it. /loop and the cron tools check for due tasks every second and enqueue them at low priority. A scheduled prompt fires between turns, never mid-response. If Claude is busy when a task comes due, it waits until the current turn ends. Tasks fire only while Claude Code is running and idle, and recurring tasks expire after seven days.

So a busy session delays its own scheduled work, exactly as long jobs delayed Lucy's short ones. The scheduler also adds a deterministic offset to fire times so that sessions do not all call the API at the same moment: a recurring task can fire up to 30 minutes after its scheduled time. For work that must run whether or not a session is open, it points to cloud routines, desktop scheduled tasks and GitHub Actions.

Try it in Claude Code: schedule a one-minute loop, give Claude a long task, and measure how late the scheduled prompt runs.

Expected observations and learner verification

The offline checkpoint reports one coalesced morning task, one first stock episode, an interrupted turn reported, an unattended evening brief, one independently discovered second episode, zero duplicate episode work and zero purchases. It verifies that the second episode's stored report says $17.50 for seven tubs. It also observes the worker's clean shutdown. A missing child result, a wrong amount or an unexpected purchase fails the checkpoint rather than becoming an explanatory footnote.

During authoring, the first live run passed the tool-level checks but failed the explanation check when read. That is why the chapter asks you to inspect the stored report as well as the transcript. A successful tool call is useful evidence, but the builder's job includes delivering the right facts to Lucy.

The course gate runs the checkpoint with the supplied packages refused on import, so every function it reaches, in both processes, is yours. We also changed the learner file in 33 small ways, from ignoring a pause to showing the model's answer instead of the rendered report; every one of those changes fails the checkpoint. Run it yourself, inspect a transcript with --transcript, and try the Linux service exercise. Live model and phone observations remain separate from that deterministic gate.

Exercise 1: Change the missed-run contract

Create a job due at 100 with interval ten and observe it at 140. Predict the coalesced count and next due time before running the pass. Then propose a job for which replaying every missed occurrence would be necessary. Describe the durable input records and maximum replay batch that its implementation would need; do not simply remove the coalescing calculation and create an unbounded loop.

Exercise 2: Fill the intake queue

Open a shop whose queue has capacity one, admit one item, and run both a due clock job and an armed vanilla condition. Inspect the absence of a rejected clock record, its unchanged due time and the still-armed condition. Free capacity by claiming and finishing the waiting item, then rescan. Explain why one coalesced clock occurrence and one new condition episode should be admitted when enough capacity is available. Repeat the observation without freeing capacity and confirm that the scanner does not generate rejected rows on every pass.

Exercise 3: Hide an intermediate stock recovery

Start with a disarmed condition and positive need. Record a delivery that brings vanilla to its reorder point and a sale that takes it below again, without scanning between them. Predict the next result. Then repeat with a scan at the healthy value. Explain why these histories produce different episode counts, and explain how consuming Chapter 4's stock events, with a durable position in that log, would distinguish them even while the scanner is offline.

Exercise 4: Corrupt only the model's narration

Use a model fixture that requests the correct draft but ends with the wrong currency or total. Check the raw transcript, stored result and channel payload separately. Then corrupt the structured tool result instead and require a refusal. The report renderer should not turn arbitrary text into an authoritative amount, and an invalid structured amount must not be silently repaired by guessing what the tool intended.

Exercise 5: Give the reports their own queue

Split the mixed workload into two queues with two workers: one for short notes and one for weekly reports, with the same total arrival rate. Measure the mean wait of the short notes and of the reports, and compare them with the single shared queue. Then predict both waits with the Pollaczek–Khinchine formula before you run it.

Exercise 6: Find the knee

Rerun the notes workload at utilizations 0.3, 0.6, 0.9 and 0.95, with 400 jobs each. Plot the measured mean wait against $ ho$ with the prediction. Near 0.95, repeat the run with three different seeds. How much do the measured means vary, and what does that say about judging a queue from one busy afternoon?

Active recall

Without rereading: what is utilization, and what happens at 1? Derive the Pollaczek–Khinchine wait from the remaining service and Little's law. Why does a random arrival more often land in a long job? Why did the mixed workload wait longer than the notes even when it was less busy? How did the pilot's estimate turn a planned 80% into an overloaded queue?

What remains durable when the worker process is absent? Why do we advance a clock job from its old due time? What observation rearms a stock condition? How does a work subject differ from a SKU mentioned in a prompt? Which check caught the $15.00 explanation, and which narrower check had already passed? Why does a fixed 86,400-second interval not fully specify an 8 a.m. local appointment?

Vocabulary

Utilization is the share of time a worker is busy. A Poisson process has independent, exponentially distributed gaps between arrivals. The Pollaczek–Khinchine formula gives the mean wait for Poisson arrivals and any service-time distribution. The inspection paradox is that a random moment more often falls in a long interval. A scheduler pass is one bounded observation and admission step. Coalescing combines missed occurrences into one current task. A stock episode is the shortage period distinguished by the scanner's healthy and unhealthy observations. Armed means the next positive observation may create work. A generation distinguishes successive episodes. A subject is the immutable product scope carried with the task. Liveness describes the process; it does not certify the business outcome.

Summary

Waiting grows with $\rho/(1-\rho)$ and with the variability of job lengths. With a real model doing the work, the Pollaczek–Khinchine formula, fed the service times each run saw, predicted the mean wait of every stable run to within 20%. Mixing long reports into the queue made jobs wait longer even at lower utilization, and a pilot that happened to miss some long reports planned 80% utilization for a queue that ran at 111% and never caught up. A worker needs headroom, its plan needs enough measurements, and long jobs belong in their own queue.

You built durable clock and stock producers around the existing work queue, made pause and missed-run behavior explicit, and carried product scope into the owned agent loop. You connected those pieces to an unattended process and a standard Linux user service. You also repaired a reporting failure by taking draft quantities and USD amounts from structured tool observations. Lucy can now receive drafts while away; the next chapter establishes the permission boundary before any draft becomes spending.

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.