Introduction

This is a course, not a manual. By the end you will have built panoptes-control — the coordinator that runs the EXECUTE stage of the Panoptes harness as a small distributed system — and, more importantly, you will understand why every piece is shaped the way it is. We are going to move the way a good pair-programming session moves: I frame a concept, you ask questions until it is solid, then we build the piece together and the compiler grades the work.

What we are building

panoptes-control is a control plane. A client submits an eval run — a manifest of vignettes, a list of models, a number of epochs — and the coordinator does four things:

1. Accepts the submission over an HTTP API and answers immediately.
2. Persists it as a queued run, durably, so a crash loses nothing.
3. Splits it into jobs — one per model × epoch — the unit that fans out.
4. Dispatches those jobs across a pool of workers, local first and networked later, collecting every model response back into the store.

Stages upstream (generating the vignettes) and downstream (coding responses, statistics) are other people's problems — some of them the Python side, some of them Course 1's harness. We are building the engine that takes a submitted run and executes it to completion across machines, correctly, even when machines fail.

The end state, held in view

Everything we build serves one payoff you should keep in mind from the very first chapter: under worker failure, no eval is lost and none is double-counted.

Those two properties pull in opposite directions, and that tension is the whole course. To lose nothing when a worker dies mid-job, the coordinator must be willing to hand that job to another worker — at-least-once delivery. But at-least-once means a job can genuinely run twice: the first worker was slow, not dead, and both results come back. The escape is idempotent recording — the store is written so that recording the same job's outcome twice is indistinguishable from recording it once. At-least-once delivery made safe by idempotent recording: that is the sentence the last arc pays off, and it is worth holding from here.

The architectural spine

There is one structural idea that makes the distributed half of this system additive rather than a rewrite, and you will meet it in the very next concept chapter: a single trait, WorkerHandle, behind which the scheduler dispatches every job.

The scheduler holds a pool of dyn WorkerHandle and calls dispatch(job). It never learns whether the handle on the other side runs the eval in an in-process task (a LocalWorker) or ships it over a TCP socket to another machine (a RemoteWorker). Because the scheduler depends only on the trait, the entire networked layer — Parts VI through VIII — arrives as a new implementation of an existing seam, not a change to the code that uses it. Get that seam right in Part II and the payoff arc has somewhere to plug in.

How the arcs are sequenced

The order is chosen for understanding, not for delivery speed. Each arc teaches one core idea and hands the next arc something to stand on:

• Part II — The Core Arc. The load-bearing seam (WorkerHandle), the Run/Job domain, the id newtypes, the error taxonomy that knows what is worth retrying, and the wire protocol. Everything downstream points here.
• Part III — The Eval Workload Arc. The ModelClient trait object and run_eval, tested against a mock HTTP server, plus the append-only file contract that logs every raw response.
• Part IV — The Persistence Arc. sqlx and SQLite: schema, migrations, and the atomic claim + transactional recording that make crashes and concurrent schedulers safe.
• Part V — The Service Arc. axum: handlers, shared state, extractors, and turning the error taxonomy into HTTP status codes in one place.
• Part VI — The Scheduler Arc. Bounded concurrency, retrying only the retryable, the LocalWorker, and graceful shutdown — the coordinator becomes a running program.
• Part VII — The Telemetry Arc. tracing spans and a /stats endpoint that accounts for tokens and cost per model.
• Part VIII — The Cluster Arc. The payoff: a framed TCP protocol, a connection actor multiplexing one socket, the RemoteWorker behind the same trait, and at-least-once redelivery with heartbeat reaping made safe by the store's idempotency.

A note on where you are starting from

You come to this course strong in two areas and new in a third, and it helps to be honest about all three. The domain — LLM evaluation, models, prompts, epochs, token accounting — is familiar ground; you have built systems like the thing this coordinator runs. And Rust itself is no longer new: from Courses 1 and 2 you already own ownership and borrowing, serde, traits and generics, async/await and tokio basics, wiremock, clap, and the thiserror/anyhow split. This course does not re-teach any of that.

What is new is the shape of a service: code that persists state to a database, answers HTTP requests, schedules concurrent work, and talks to other processes over a socket. That is the whole frontier here — axum, sqlx, and a hand-rolled network protocol — and it is exactly where the concept chapters spend their time. If a networking or persistence chapter feels hard, that is not a signal about your ability; it is a new class of failure (a dropped connection, a partial write, a race between two schedulers) that the type system is about to teach you to make impossible.

Turn the page for the architecture at a glance, then we start with Phase 0 and the seam that holds the whole thing together.

How to Use This Course

The two chapter types

Every arc alternates between two kinds of chapter, and they ask different things of you.

Concept chapters are the lecture. Read them without touching the keyboard. They argue why before how, they carry small runnable examples in a toy domain that mirrors the real build one-for-one, and they end with a short list of questions you should be able to answer for yourself. If any are fuzzy, that is the signal to stop and re-read before moving to the build. Do not proceed to a build chapter on a shaky concept; the whole point of frontloading concepts is that the build then feels obvious.

Build chapters are the pair-programming session. Here you write the code, test-first, in the same loop every time:

1. Write the failing test. You write it, from the behavior we described — not by copying an answer.
2. Predict the failure. Say out loud (or note down) what the compiler or test runner will report.
3. Run it and check your prediction. The gap between prediction and reality is the lesson.
4. Write the minimal implementation. Just enough to make the test pass. No more.
5. Run again, watch it pass.
6. Commit.

The predict-then-run loop

The prediction step is not a ritual; it is the mechanism. When you guess what cargo test will print — which test fails, on which line, with what left-and-right values, or which E-code the compiler will raise — you commit to a model of how the code behaves. Running it then either confirms that model or corrects it, and a corrected model sticks far better than one you were simply handed. In the networking and persistence arcs especially, the failures are new (a dropped connection surfacing as a retryable error, two schedulers racing for the same run), so the predict step is where those new failure modes become intuition.

Running the tests

The project is a Cargo workspace — one repository, several crates. That means you can test one crate in isolation or the whole thing at once:

cargo test -p control-core     # just the crate you are working in
cargo test                     # the whole workspace

Working on one arc, you will almost always want the -p form: it is faster and its output is scoped to the crate you are editing. Run the full cargo test before you commit, to confirm you have not broken a downstream crate that depends on the one you changed.

No arc needs a running database, a live model API, or a second machine. Every test runs from a plain cargo test — the persistence arc uses an in-memory SQLite database, the eval arc uses a mock HTTP server, and the cluster arc opens a real socket on the loopback interface. The next chapter explains that stance in full.

The concept-check quizzes

Each arc ends with one graded quiz. These are not busywork — they target the exact misconceptions that cause bugs three chapters later. Some questions are Tracing questions: a short program, and you decide whether it compiles and what it prints. Those programs are compiled by the real rustc when the book is built, so the answer is not a matter of opinion. Answer honestly before revealing; a wrong answer with a good explanation teaches more than a lucky guess. The score is for you, not for anyone else.

The answer key

There is a companion implementation plan (linked in the appendix) that contains the full, worked version of every task. Use it as an answer key, not a script. Try each build yourself first. Check the plan after. The distance between your version and the plan's version is precisely where the learning lives — if they match, you understood it; if they differ, the difference is worth investigating.

Working setup

You will want two things open side by side: this book, and a terminal in your project directory. A split screen or two monitors. The loop is fast — write test, run, read error, fix — and it only works if the feedback is immediate.

When you get stuck

Bring the exact error message, verbatim. Rust's compiler errors are unusually good; most of the time the fix is in the error itself, and learning to read them is half of learning the language. When we work through a stuck point together, we dig into what the compiler is actually objecting to rather than papering over it.

Phase 0: Workspace, Toolchain, the Loop

Before any coordinator code, we verify three things: that your toolchain is current, that you understand the workspace-per-arc structure the whole course grows inside, and that you can write a failing test, see it fail legibly, fix it, and see it pass. The last one is deliberately trivial — that is the point. We are testing the machinery, not your ability.

The toolchain

This course targets Rust edition 2024, which needs a stable toolchain ≥ 1.96. If you finished Courses 1 and 2 you already have rustup; bring it current and confirm the versions:

rustup update
rustc --version   # want 1.96.0 or newer
cargo --version   # want 1.96.0 or newer

Edition 2024 is set per crate in each Cargo.toml (edition = "2024"), not globally. It is what lets us use the current async ergonomics and the newest lint defaults without ceremony. If rustc --version reports something older than 1.96, rustup update fixes it before you go further.

The workspace, one crate per concern

panoptes-control is not one crate — it is a Cargo workspace of five, and the split is the architecture. A workspace is a single repository, a single target/ build directory, a single cargo test, but several independently-compiled member crates with explicit dependencies between them. Drawing those dependencies as crate boundaries is what stops, at compile time, a lower layer from reaching up into a higher one.

The root Cargo.toml is a [workspace], not a package — it lists members and pins shared dependency versions once:

[workspace]
resolver = "2"
members = [
    "crates/control-core",
    "crates/control-eval",
    "crates/control-store",
    "crates/panoptes-control",
    "crates/panoptes-worker",
]

[workspace.dependencies]
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }
# ... every shared version lives here; crates opt in with `{ workspace = true }`

The five crates, and what each one owns:

The dependency direction is the point. control-core knows about no one; everyone knows about control-core. That is what makes the WorkerHandle seam load-bearing: both the in-process LocalWorker (in panoptes-control) and the networked RemoteWorker implement a trait that lives in the crate at the bottom, so the scheduler can hold either without a dependency pointing the wrong way.

The no-infrastructure testing stance

A distributed system sounds like it needs infrastructure to test — a database server, a live model API, a second machine. This one needs none of that, and keeping it that way is a design constraint we hold from the first test to the last. Every arc runs from a plain cargo test:

• The store opens an in-memory SQLite database (sqlite::memory:). Each test gets a fresh, isolated schema in microseconds; nothing touches disk, and there is no server to start.
• The eval workload points its HTTP client at a wiremock mock server (you met this in Course 2) instead of a real model API — so responses are deterministic and no network leaves the machine.
• The cluster binds a real TCP listener to 127.0.0.1:0 — port zero, meaning "the OS picks a free port" — so a test spins up a genuine coordinator and worker over a loopback socket with no fixed port to collide.
• Time-dependent logic (heartbeats, timeouts) uses tokio::time::pause, so a test can fast-forward thirty seconds of "silence" instantly instead of actually waiting.

The payoff is that the feedback loop stays fast and every test is hermetic: it depends on nothing outside the process, so it cannot be flaky because of a busy port, a slow network, or leftover rows from a previous run.

The loop, on a workspace member

Let us confirm the machinery end to end. Suppose the very first control-core function is job_count — the split arithmetic, one job per model × epoch — with a test that asserts the wrong number on purpose:

#![allow(unused)]
fn main() {
pub fn job_count(models: u32, epochs: u32) -> u32 {
    models * epochs
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn one_job_per_model_times_epoch() {
        assert_eq!(job_count(2, 3), 5); // deliberately wrong
    }
}
}

Test one crate in isolation with -p:

cargo test -p control-core

running 1 test
test tests::one_job_per_model_times_epoch ... FAILED

failures:

---- tests::one_job_per_model_times_epoch stdout ----

thread 'tests::one_job_per_model_times_epoch' panicked at crates/control-core/src/lib.rs:11:9:
assertion `left == right` failed
  left: 6
 right: 5

Read that carefully, because you will read a hundred of these. It names the exact test, the exact file and line, and — critically — left: 6, right: 5: the computed value against the expected value. This is why we write assertions with the computed value on the left and the expected on the right; the output then reads naturally. Here the code is right and the test is wrong. Fix the expectation to 6:

#![allow(unused)]
fn main() {
        assert_eq!(job_count(2, 3), 6);
}

Run again:

running 1 test
test tests::one_job_per_model_times_epoch ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

That is the loop. -p control-core scopes the run to the crate you are editing; a bare cargo test runs the whole workspace and is what you do before every commit, to confirm no downstream crate broke:

cargo test          # the whole workspace, every crate

Done when

You have a current toolchain (cargo --version ≥ 1.96), you can name the five crates and which layer each occupies, and you can run cargo test -p <crate>, read a failure, and fix it without thinking about the mechanics. When that loop is automatic, turn the page: the next chapter is the one concept the entire distributed design rests on — the seam.

Concept: The Seam — dyn Trait Objects as Dependency Inversion

Kind: Concept (read, do not code). This is the chapter the whole distributed design rests on.

Here is the promise the introduction made: the networked half of this system will arrive as an addition, not a rewrite. Parts II through VII build a coordinator that dispatches every eval job to an in-process worker. Then Part VIII adds workers on other machines — and the scheduler, the code that actually hands out jobs, does not change by a single line. That is not luck or discipline. It is a structural property you install deliberately, in Part II, with one trait. This chapter is about why that works, before we build it.

You already know trait objects from Courses 1 and 2 — Box<dyn Trait>, dynamic dispatch, the vtable. So this chapter is not "what is a trait object." It is a sharper claim: a trait object placed at the right boundary is dependency inversion, and dependency inversion is precisely the tool that makes "local now, networked later" an additive change. We are going to make that claim concrete, argue why it holds, and only then map it onto the real seam.

The problem, stated as a dependency arrow

Think about who-depends-on-whom, because that is the whole game. The scheduler's job is to take a queued run, split it into jobs, and get each job executed somewhere. The naive way to write that is for the scheduler to know how execution happens:

• Version 1 (local): the scheduler calls run_eval(job) directly, in-process. Simple. But now the scheduler depends on the concrete way jobs run.
• Version 2 (networked): we want some jobs to run on other machines. If the scheduler called run_eval directly, we now have to go back into the scheduler and teach it about sockets, framing, retries, and reconnection. The scheduler — the piece we had working — gets rewritten to accommodate a lower-level detail. Every time execution changes, the scheduler changes.

That is a dependency pointing the wrong way: a high-level policy (how to schedule and retry a run) depending on a low-level mechanism (how one job physically executes). Dependency inversion is the fix, and its name is literal — you invert that arrow. Instead of the scheduler depending on a concrete worker, both the scheduler and every concrete worker depend on an abstraction in between: a trait.

The seam, in a toy that mirrors the real one

Let us build the smallest possible version of exactly this shape, in a domain with no networking to distract us. We want to notify someone that something happened. Some notifiers are in-process (append to a local log); some are remote (ship the message over the network). The code that decides what to notify about must not care how any notifier delivers.

The abstraction in the middle is a trait with one behavior:

#![allow(unused)]
fn main() {
/// The seam. `broadcast` will depend only on this.
trait Notifier {
    fn id(&self) -> &str;
    fn notify(&self, msg: &str) -> Result<(), String>;
}
}

Now two concrete implementations. The first is fully in-process — it does its whole job by writing locally, no network involved:

#![allow(unused)]
fn main() {
/// In-process: writes to the local log. No network at all.
struct LogNotifier {
    id: String,
}

impl Notifier for LogNotifier {
    fn id(&self) -> &str {
        &self.id
    }
    fn notify(&self, msg: &str) -> Result<(), String> {
        println!("[{}] {msg}", self.id);
        Ok(())
    }
}
}

The second stands in for the far side — the version that would open a socket and ship bytes to another machine. We are not writing that socket code here; the point of the toy is that the code holding the notifiers cannot tell the difference:

#![allow(unused)]
fn main() {
/// A sketch of the far side: this one would ship the message over a socket.
/// Here we only stand in for that — the point is that `broadcast` cannot tell.
struct RemoteNotifier {
    id: String,
    endpoint: String,
}

impl Notifier for RemoteNotifier {
    fn id(&self) -> &str {
        &self.id
    }
    fn notify(&self, msg: &str) -> Result<(), String> {
        // The real impl writes bytes to `self.endpoint`; we fake the send.
        println!("POST {} <- {msg}", self.endpoint);
        Ok(())
    }
}
}

And here is the payoff — the piece that is the high-level policy. It holds a pile of notifiers behind Box<dyn Notifier> and drives them. Read its type signature closely: &[Box<dyn Notifier>]. There is no LogNotifier, no RemoteNotifier, no enum of "kinds" anywhere in it. It literally cannot name the concrete types, so it cannot branch on them:

/// The payoff: this function holds a pile of notifiers and never learns which
/// kind any of them is. Add a third impl tomorrow — this code does not change.
fn broadcast(notifiers: &[Box<dyn Notifier>], msg: &str) {
    for n in notifiers {
        if let Err(e) = n.notify(msg) {
            println!("{} failed: {e}", n.id());
        }
    }
}

fn main() {
    let notifiers: Vec<Box<dyn Notifier>> = vec![
        Box::new(LogNotifier { id: "log-1".into() }),
        Box::new(RemoteNotifier {
            id: "remote-1".into(),
            endpoint: "10.0.0.5:9000".into(),
        }),
    ];
    broadcast(&notifiers, "run 7 started");
}

Predict the output before you run it — two lines, and which impl produces each.

[log-1] run 7 started
POST 10.0.0.5:9000 <- run 7 started

Two different mechanisms — a local log write and a (pretended) network send — driven by one loop that knows about neither. broadcast depends only on Notifier. LogNotifier and RemoteNotifier depend on Notifier. The arrow is inverted, and the consequence is the whole reason we are here: to add a third kind of notifier, you write a new impl Notifier and drop it in the vec. You do not touch broadcast. That is "additive, not a rewrite," in fourteen lines.

Why the dependency has to be a trait object here

You might ask: we know generics, why not broadcast<N: Notifier>(notifiers: &[N])? Because that would demand every element be the same concrete type N. Our vec is deliberately heterogeneous — a LogNotifier and a RemoteNotifier side by side — and the whole point is to hold a mix and decide the mix at runtime, when we discover which workers have actually connected. A generic is resolved at compile time to one type; dyn is resolved at runtime, per element, through the vtable. Distribution is inherently a runtime fact — you do not know at compile time how many workers will connect or of which kind — so the seam must be a trait object, not a generic. This is the case where dyn is not a stylistic choice but the only tool that fits.

The real seam is the same shape, asynchronously

The toy is synchronous so it runs anywhere. The real seam is identical in structure but its one method is async, because dispatching a job means awaiting a model call or a network round-trip. An async method in a trait you intend to use as dyn still, in this course's toolchain, wants the #[async_trait] macro — you used it in Course 2 — because it rewrites async fn into a method returning a boxed future, which is something a vtable can hold.

Here is the toy rewritten in the exact shape of the real WorkerHandle. It will not run on the playground (it needs the async-trait and tokio crates), so it is marked ignore:

// scratch Cargo.toml deps:
//   async-trait = "0.1"
//   tokio = { version = "1", features = ["full"] }
use async_trait::async_trait;
use std::sync::Arc;

#[async_trait]
trait Notifier: Send + Sync {
    fn id(&self) -> &str;
    async fn notify(&self, msg: &str) -> Result<(), String>;
}

struct LogNotifier {
    id: String,
}

#[async_trait]
impl Notifier for LogNotifier {
    fn id(&self) -> &str {
        &self.id
    }
    async fn notify(&self, msg: &str) -> Result<(), String> {
        println!("[{}] {msg}", self.id);
        Ok(())
    }
}

struct RemoteNotifier {
    id: String,
    endpoint: String,
}

#[async_trait]
impl Notifier for RemoteNotifier {
    fn id(&self) -> &str {
        &self.id
    }
    async fn notify(&self, msg: &str) -> Result<(), String> {
        // Real impl: open a socket to `self.endpoint`, write the frame, await ack.
        println!("POST {} <- {msg}", self.endpoint);
        Ok(())
    }
}

// The policy is unchanged in shape — it just `.await`s each call. Note `Arc`
// instead of `Box`: workers are shared across concurrent dispatch tasks.
async fn broadcast(notifiers: &[Arc<dyn Notifier>], msg: &str) {
    for n in notifiers {
        if let Err(e) = n.notify(msg).await {
            println!("{} failed: {e}", n.id());
        }
    }
}

Two details that carry straight into the real seam. First, the trait now requires Send + Sync: a dyn WorkerHandle will be moved between and shared across tokio tasks running on different threads, so the compiler must know it is safe to send and share. Second, the container is Arc<dyn Notifier>, not Box: a Box is a single owner, but a worker handle is dispatched to from several concurrent tasks at once, so it must be shared ownership. Those are the only differences between the toy and the production seam. The inversion — policy depends on trait, mechanisms depend on trait — is bit-for-bit the same.

One-for-one: the toy ↔ the real thing

Everything above maps onto the actual control-core seam you will build in Part II, one piece to one piece:

• Notifier (the trait, the seam) ↔ WorkerHandle — the trait the scheduler dispatches through. Its real method is async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError>.
• LogNotifier (in-process, no network) ↔ LocalWorker — a WorkerHandle that runs the eval in an in-process task. This is all the coordinator uses through Part VII.
• RemoteNotifier (the far-side sketch) ↔ RemoteWorker — a WorkerHandle that ships the job over a TCP socket to a worker process. This is the entire new surface Part VIII adds.
• broadcast(&[Box<dyn Notifier>]) (holds a mix, knows no concrete kind) ↔ the scheduler — it holds Vec<Arc<dyn WorkerHandle>> and calls dispatch, never learning local from remote.
• notify taking &str, not a generic ↔ dispatch taking a concrete Job — the seam's methods stay object-safe on purpose, so the trait can be dyn.

Hold that last row against the trap above: the day the compiler tells you a trait "is not dyn compatible," it is enforcing exactly the discipline that keeps this seam usable. And hold the mapping as a whole against the promise this chapter opened with. When Part VIII adds RemoteWorker, it adds a new impl WorkerHandle and a pool that hands the scheduler Arc<dyn WorkerHandle> values — and the scheduler, like broadcast, does not change. The distributed arc is additive because the arrow was inverted here, at the start.

Questions to lock

Genuinely stop on each. This is the foundation the last five arcs stand on.

1. What does "invert the dependency" mean concretely, in terms of which direction the arrow points before and after? Why does the inverted arrow make a new worker kind additive?
2. Why must the worker pool be dyn WorkerHandle rather than a generic Vec<W>? What runtime fact about distribution forces that choice?
3. Why does the real seam add Send + Sync and use Arc instead of Box? What would go wrong without each?
4. What makes a trait not object-safe (dyn-compatible), and what error do you get if you try to form &dyn of it?

Next: Part II, where we build this seam for real — the WorkerHandle trait, the domain it dispatches, and the error taxonomy that tells the scheduler which failures are worth retrying.

Concept: Newtypes, Transparent serde, and the Domain

Kind: Concept. New crate: uuid — this chapter shows it working before you build with it.

This is the Core arc, and it is the arc every other one leans on. control-core holds the vocabulary — the ids, the domain structs, the status enums — that the store, the service, the scheduler, and the cluster all speak. Get the types right here and the rest of the course is filling in behavior around a fixed shape. Get them loose and every later arc inherits the looseness.

You already know serde, enums, and structs cold from the first two courses. So this chapter is not about how to derive Serialize. It is about two smaller decisions that look cosmetic and are not: wrapping a Uuid in a newtype, and telling serde to serialize that newtype transparently. Both are cheap. Both buy you a class of bug the compiler catches for free.

The problem: two ids that are the same shape and must never be confused

The control plane has two id-shaped things. A run is a submitted eval sweep — the top-level unit a client creates. A job is a chunk of a run that one worker executes. Both are identified by a UUID. Structurally, a run id and a job id are identical: 128 random bits.

So the lazy move is type RunId = Uuid; type JobId = Uuid; — or worse, just pass Uuid around everywhere. The problem shows up the first time you write a function that takes both:

// Uses the `uuid` crate — add it in your scratch project (below) to run this.
fn cancel_job(run: Uuid, job: Uuid) { /* ... */ }

Now every call site is a coin flip. cancel_job(job, run) compiles perfectly and is wrong at runtime, and you find out when a job in run A gets cancelled against run B's ledger. This is exactly the stringly-typed disease from Course 1, wearing a Uuid costume: the type carries no information about which kind of id it is, so the compiler cannot help you.

The fix: a newtype per id

A newtype is a one-field tuple struct that wraps another type to give it a name and an identity:

use uuid::Uuid;

pub struct RunId(pub Uuid);
pub struct JobId(pub Uuid);

RunId and JobId now wrap the same Uuid, but they are different types. And that is the whole point: the compiler will not let you pass one where the other is expected. Predict what happens here before you read the output:

use uuid::Uuid;

struct RunId(Uuid);
struct JobId(Uuid);

fn cancel_run(run: RunId) { /* ... */ }

fn main() {
    let job = JobId(Uuid::new_v4());
    cancel_run(job); // wrong id kind
}

error[E0308]: mismatched types
  --> src/main.rs:10:16
   |
10 |     cancel_run(job); // wrong id kind
   |     ---------- ^^^ expected `RunId`, found `JobId`
   |     |
   |     arguments to this function are incorrect
   |
note: function defined here

E0308, at compile time. The program never runs. Compare this to type RunId = Uuid — a plain type alias is not a new type, it is a second name for the same one, so cancel_run(job) would compile and the bug would ship. The newtype is what turns "two things that happen to be UUIDs" into "two things the compiler keeps apart."

The catch: a newtype changes the wire format — unless you say otherwise

Here is the trap, and it is the reason this chapter exists rather than a one-line footnote. serde serializes a newtype struct the way it serializes any struct — and a tuple struct with one field serializes as a one-element sequence or a wrapper, not as the bare inner value. A RunId would land on the wire looking like ["…uuid…"] or {"0":"…"} depending on format, when what you want in your JSONL logs and your HTTP bodies is just the bare quoted UUID a human (or jq, or a database column) can read.

The fix is one attribute — #[serde(transparent)] — which tells serde: serialize this struct as if it were its single field. The wrapper vanishes on the wire; the type safety stays in the code. Here is the whole toy, runnable:

// Scratch deps: serde = { version = "1", features = ["derive"] },
//               serde_json = "1", uuid = { version = "1", features = ["v4", "serde"] }
use serde::{Deserialize, Serialize};
use uuid::Uuid;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
struct OrderId(Uuid);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
struct LineId(Uuid);

impl OrderId {
    fn new() -> Self { OrderId(Uuid::new_v4()) }
}
impl LineId {
    fn new() -> Self { LineId(Uuid::new_v4()) }
}

fn main() {
    let oid = OrderId::new();

    // #[serde(transparent)]: the wire form is the bare uuid, no wrapper.
    let json = serde_json::to_string(&oid).unwrap();
    println!("OrderId on the wire: {json}");

    let back: OrderId = serde_json::from_str(&json).unwrap();
    println!("round-tripped equal: {}", back == oid);

    let lid = LineId::new();
    println!("OrderId and LineId are different types, both wrapping a Uuid");
    println!("  order = {oid:?}");
    println!("  line  = {lid:?}");
}

OrderId on the wire: "5aad6fdf-d682-4161-ac42-9d4e6f2ff948"
round-tripped equal: true
OrderId and LineId are different types, both wrapping a Uuid
  order = OrderId(5aad6fdf-d682-4161-ac42-9d4e6f2ff948)
  line  = LineId(20ec3bd8-3883-42ae-903b-342315d51967)

(The UUIDs are random, so your run prints different ones.) Read the first line: "5aad6fdf-…" — a bare quoted string, not ["5aad6fdf-…"]. That is transparent doing its job. Drop the attribute and the same struct serializes with a wrapper, which would silently change every id column in every log file. This is why a round-trip test that pins the wire form — not just "encode then decode is equal" — is worth writing: it freezes the byte-level contract so a future refactor can't quietly reshape it.

Display, so an id can become a string when you need one

A newtype hides the inner Uuid, which means println!("{run_id}") won't work until you say how it prints — the wrapper doesn't inherit Display. You give it one line:

use std::fmt;
use uuid::Uuid;

struct RunId(Uuid);

impl fmt::Display for RunId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0) // delegate straight to the inner uuid
    }
}

Now run_id.to_string() equals the inner UUID's string, so log lines, URL path segments (/runs/{id}), and error messages all read as the bare id. transparent governs the serde wire form; Display governs the {}/to_string() form. They are two different codecs and you want both to say the same bare-UUID thing — which is why the build has a test for each.

The rest of the vocabulary: status enums and a total

The domain is mostly plain structs you could write in your sleep by now — Vignette, EvalJob, ResponseRecord, Job, Run. Two pieces are worth calling out because the build tests target them.

Status enums serialize snake_case. RunStatus is Queued | Running | Done | Failed; JobStatus is Pending | Assigned | Done | Failed. On the wire they must be lowercase ("queued", "assigned") because that is what a database column and a JSON API want — so both carry #[serde(rename_all = "snake_case")]. This is the same rename_all you used for the codebook enums, just a different casing.

JobOutcome knows its own token total. A JobOutcome holds the ResponseRecords a job produced; each record carries a Usage { input_tokens, output_tokens }. The outcome offers total_usage() that folds every record's usage into one sum — the accounting the /stats endpoint will report in Part VII. Here is that shape as a toy, runnable on serde alone:

use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
struct Charge {
    goods: u32,
    shipping: u32,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct Line {
    sku: String,
    charge: Charge,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct Order {
    lines: Vec<Line>,
}

impl Order {
    fn total(&self) -> Charge {
        self.lines.iter().fold(Charge::default(), |acc, l| Charge {
            goods: acc.goods + l.charge.goods,
            shipping: acc.shipping + l.charge.shipping,
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum OrderStatus {
    Pending,
    Shipped,
    Delivered,
    Cancelled,
}

fn main() {
    let order = Order {
        lines: vec![
            Line { sku: "A".into(), charge: Charge { goods: 100, shipping: 20 } },
            Line { sku: "B".into(), charge: Charge { goods: 50, shipping: 10 } },
        ],
    };
    println!("total: {:?}", order.total());
    println!("status: {}", serde_json::to_string(&OrderStatus::Shipped).unwrap());
    println!("status: {}", serde_json::to_string(&OrderStatus::Cancelled).unwrap());
}

total: Charge { goods: 150, shipping: 30 }
status: "shipped"
status: "cancelled"

The fold sums each field independently — 100 + 50 goods, 20 + 10 shipping — which is exactly how total_usage() sums input and output tokens across records. And OrderStatus::Shipped lands on the wire as "shipped", the snake_case convention the real status enums use.

One-for-one: the toy ↔ the build

Everything above maps straight onto control-core:

The build is the same three ideas — a transparent newtype per id, snake_case status enums, a fold that totals a Copy accounting struct — pointed at run/job/usage instead of order/line/charge.

Questions to lock

1. Why does a type RunId = Uuid; alias fail to prevent the run-id/job-id mixup, while a struct RunId(Uuid); newtype prevents it? What is the mechanical difference the compiler sees?
2. What does #[serde(transparent)] change about a newtype's wire form, and what breaks in your log files if you forget it?
3. transparent and Display both make a RunId "look like its inner UUID." Which one governs serde_json::to_string, and which governs format!("{run_id}")? Why do you want a test for each?

Next chapter is the first build of the arc: ids.rs and domain.rs, starting from a failing test.

Build: control-core — Ids and the Domain

Maps to: Phase 0 (control-core). Kind: Build.

Objective

Stand up control-core, the crate that holds every type the rest of the control plane speaks, and fill in its two most foundational files: ids.rs (the RunId/JobId newtypes) and domain.rs (the status enums, the accounting structs, and the Run/Job aggregates). Prove with tests that ids are unique, serialize as bare UUID strings, and print as their inner UUID; that statuses serialize snake_case; that a JobOutcome totals its usage; and that a Job round-trips through JSON.

Scaffold

Create:

• crates/control-core/Cargo.toml — the crate manifest. [dependencies]: serde (with derive), uuid (with v4, serde), chrono (with serde). [dev-dependencies]: serde_json (the round-trip and serialization tests encode to JSON), pretty_assertions (readable assert_eq! diffs).
• crates/control-core/src/lib.rs — module declarations + re-exports (pub mod ids; pub mod domain; and pub use the types you want callers to reach without the module path).
• crates/control-core/src/ids.rs — RunId, JobId, and their tests.
• crates/control-core/src/domain.rs — the domain types and their tests.

Why these deps: uuid gives you the 128-bit id and its v4 random constructor; the serde feature lets a Uuid serialize on its own so your transparent newtype has something to delegate to. chrono supplies DateTime<Utc> for Run::created_at, and its serde feature makes that field serialize as an RFC 3339 string.

Expected result: cargo test -p control-core → 6 tests pass:

• from ids.rs: ids_are_unique, run_id_serializes_as_bare_uuid_string, display_matches_inner_uuid
• from domain.rs: total_usage_sums_every_record, statuses_serialize_snake_case, job_roundtrips_through_json

The spec (givens)

ids.rs — the id newtypes. Two tuple structs, each wrapping a uuid::Uuid:

• RunId(pub Uuid) and JobId(pub Uuid).
• Derive stack on both: Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize.
• Attribute on both: #[serde(transparent)] — the wire form is the bare quoted UUID, no wrapper.
• Each gets an inherent new() -> Self returning a fresh Uuid::new_v4(), a Default impl that calls new(), and a Display impl that writes the inner UUID (write!(f, "{}", self.0)).

[→ Answer key](../appendix-answer-key.md#core-ids)

domain.rs — the vocabulary.

• RunStatus: variants Queued, Running, Done, Failed. JobStatus: variants Pending, Assigned, Done, Failed. Both derive Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize and carry #[serde(rename_all = "snake_case")].
• Usage: fields input_tokens: u32, output_tokens: u32. Derives Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize (the Default is what lets a fold start from zero).
• Vignette: fields id: String, prompt: String. Derives Debug, Clone, PartialEq, Eq, Serialize, Deserialize.
• EvalJob: fields vignettes: Vec<Vignette>, model: String, epoch: u32. Same derive stack as Vignette (no Copy — it owns a Vec).
• ResponseRecord: fields vignette_id: String, model: String, epoch: u32, prompt: String, response: String, usage: Usage. Same derive stack.
• Job: fields id: JobId, run_id: RunId, status: JobStatus, attempt: u32, spec: EvalJob. Same derive stack.
• JobOutcome: fields job_id: JobId, records: Vec<ResponseRecord>. Same derive stack. Plus an inherent method total_usage(&self) -> Usage whose rule is: sum input_tokens and output_tokens across every record, independently, starting from Usage::default().
• Run: fields id: RunId, status: RunStatus, created_at: DateTime<Utc>, manifest: String, models: Vec<String>, epochs: u32, job_count: u32, done_count: u32. Same derive stack as the other aggregates.

[→ Answer key](../appendix-answer-key.md#core-domain)

Concepts exercised

• Newtype-over-Uuid for id safety; #[serde(transparent)] for bare-value serialization.
• A hand-written Display that delegates to an inner field.
• #[serde(rename_all = "snake_case")] on unit-variant enums.
• Folding a Copy, Default accounting struct across a Vec.
• Which aggregates can derive Copy (all fields Copy) and which cannot (owns a Vec or String).

The build loop (you drive)

Write each test first, predict its failure mode, run, then implement the minimum to green.

1. ids_are_unique — asserts RunId::new() != RunId::new() and the same for JobId. Predict: does this fail to compile (the type does not exist yet) or fail an assertion? Run, confirm it is the former, then define the newtypes with new().
2. run_id_serializes_as_bare_uuid_string — build a RunId, serde_json::to_string it, and assert the JSON equals format!("\"{}\"", id.0) (a bare quoted UUID, no [...] or {...}); then deserialize back and assert equality. This is the test that pins the wire contract — add #[serde(transparent)] to pass it and watch what removing the attribute does.
3. display_matches_inner_uuid — assert id.to_string() == id.0.to_string(). Add the Display impl.
4. statuses_serialize_snake_case — assert serde_json::to_string(&RunStatus::Queued) is "\"queued\"" and &JobStatus::Assigned is "\"assigned\"". Add the enums with rename_all.
5. total_usage_sums_every_record — build a JobOutcome with two records (say 100/20 and 50/10 tokens) and assert total_usage() equals Usage { input_tokens: 150, output_tokens: 30 }. Implement the fold.
6. job_roundtrips_through_json — build a full Job (with a nested EvalJob and a Vignette), encode to JSON, decode back, assert equal. This exercises every derive at once.

Done when

cargo test -p control-core shows the six tests green, run_id_serializes_as_bare_uuid_string passes with #[serde(transparent)] present (and you have seen it fail with the attribute removed), and you can say which of the domain types derive Copy and why the rest cannot. Commit.

Concept: An Error Taxonomy That Knows What to Retry

Kind: Concept.

You already know thiserror — how to derive Error, how #[error("...")] writes the Display message, how #[from] wires up a conversion. This chapter is not about that machinery. It is about one design decision the machinery lets you make: putting the retry decision in the error type itself, so the whole system asks the type "should I try again?" and gets one authoritative answer.

The problem: a distributed system fails constantly, and only some failures are worth retrying

The control plane dispatches jobs to workers over a network. In that world failure is not exceptional — it is Tuesday. A worker process dies mid-job. A TCP connection drops between the coordinator and a worker. A client sends a malformed request. A run id that never existed gets looked up. A disk write fails.

Those failures are not equal, and the difference is the single most important thing the scheduler needs to know:

• A worker that died or a connection that dropped is transient. Re-dispatch the same job to a different worker and it may well succeed. Retrying is correct.
• A malformed request or a missing run is terminal. Retrying sends the exact same bad input and fails identically — retrying just wastes time and buries the real problem. Surfacing it is correct.

So every place that handles an error has to answer: is this one worth retrying? The question is: where does that answer live?

The wrong shape: the decision scattered at call sites

The tempting-but-wrong version spreads the retry logic across every caller:

// The anti-pattern — do NOT build this.
match dispatch(job).await {
    Err(e) if e.to_string().contains("worker") => retry(job),   // string-matching!
    Err(e) if e.to_string().contains("timed out") => retry(job),
    Err(e) => return Err(e),
    Ok(o) => o,
}

Every call site re-derives the retry policy, usually by sniffing the Display string — which is brittle (reword a message and the retry breaks silently) and duplicated (add a new retryable case and you must find every match and update it). The policy has no single home, so it drifts.

The right shape: the decision is a method on the type

Encode the taxonomy in the enum, and give it one method that answers the question. Every caller asks err.is_retryable(); the type answers. Here is the whole idea as a toy — a FetchError for a hypothetical HTTP fetcher — runnable in a scratch project:

// Scratch dep: thiserror = "1"
use thiserror::Error;

#[derive(Debug, Error)]
enum FetchError {
    /// The URL points at nothing. Terminal — retrying fetches the same 404.
    #[error("not found: {0}")]
    NotFound(String),

    /// The request is malformed. Terminal — it fails identically next time.
    #[error("bad request: {0}")]
    Invalid(String),

    /// The upstream took too long. Retryable — it may answer on the next try.
    #[error("upstream timed out: {0}")]
    Timeout(String),

    /// The connection dropped mid-flight. Retryable — reconnect and retry.
    #[error("connection dropped: {0}")]
    Connection(String),

    /// A local I/O failure, converted from std::io::Error by `#[from]`.
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
}

impl FetchError {
    /// The retry decision lives here, decided once, for the whole system.
    fn is_retryable(&self) -> bool {
        matches!(self, FetchError::Timeout(_) | FetchError::Connection(_))
    }
}

/// `#[from]` generated `From<std::io::Error>`, so `?` converts for free.
fn load_cache(path: &str) -> Result<String, FetchError> {
    let body = std::fs::read_to_string(path)?; // io::Error -> FetchError::Io
    Ok(body)
}

fn main() {
    let errs = [
        FetchError::NotFound("GET /x".into()),
        FetchError::Invalid("empty body".into()),
        FetchError::Timeout("30s elapsed".into()),
        FetchError::Connection("peer reset".into()),
    ];
    for e in &errs {
        // The caller asks the type; it does not re-decide per call site.
        let verb = if e.is_retryable() { "retry" } else { "surface" };
        println!("{verb:8} <- {e}");
    }

    // `?` funnels a missing file through #[from] into FetchError::Io.
    match load_cache("does-not-exist.json") {
        Err(e) => println!("{:8} <- {e}  (retryable={})", "io", e.is_retryable()),
        Ok(_) => unreachable!(),
    }
}

surface  <- not found: GET /x
surface  <- bad request: empty body
retry    <- upstream timed out: 30s elapsed
retry    <- connection dropped: peer reset
io       <- io error: No such file or directory (os error 2)  (retryable=false)

Read the output as a policy table. NotFound and Invalid surface; Timeout and Connection retry; and Io — a real failure, produced by ? converting a genuine std::io::Error from the missing file — surfaces too. The retry decision appears in exactly one place, is_retryable, and every caller defers to it.

Why #[from] for I/O — and only I/O

Notice that Io is the one variant with #[from]. That is deliberate. #[from] generates impl From<std::io::Error> for FetchError, which is what lets the ? operator automatically lift a bare I/O error into your error type. You want that for I/O because I/O errors bubble up from std functions you call constantly (read_to_string, socket reads, file writes) and threading them through by hand would be noise.

You do not want #[from] on the domain variants (NotFound, Invalid, …), because those carry a String message you construct deliberately at the point you detect the problem — there is no single source type to convert from, and an automatic conversion there would blur where the error was actually raised. The rule of thumb: #[from] for errors that arrive from libraries you call; explicit construction for errors your own logic decides to raise.

The trap: forgetting #[from] and reaching for ?

Here is the exact failure the build hits if the I/O variant is declared without #[from]. Predict the outcome before reading it.

// Scratch dep: thiserror = "1"
use thiserror::Error;

#[derive(Debug, Error)]
enum FetchError {
    #[error("io error: {0}")]
    Io(std::io::Error), // note: no #[from]
}

fn load_cache(path: &str) -> Result<String, FetchError> {
    let body = std::fs::read_to_string(path)?; // ? needs From<io::Error>
    Ok(body)
}

fn main() {
    let _ = load_cache("x");
}

error[E0277]: `?` couldn't convert the error to `FetchError`
  --> src/main.rs:11:45
   |
10 | fn load_cache(path: &str) -> Result<String, FetchError> {
   |                              -------------------------- expected `FetchError` because of this
11 |     let body = std::fs::read_to_string(path)?; // ? needs From<io::Error>
   |                -----------------------------^ the trait `From<std::io::Error>` is not implemented for `FetchError`
   |
note: `FetchError` needs to implement `From<std::io::Error>`

E0277, a missing-trait error — not a type mismatch. The ? operator desugars to "on Err, convert via From and return," and without #[from] there is no From<std::io::Error> to convert through. The fix is one attribute. This is the theme again: the compiler refuses to build the program until the conversion is spelled out, so a "we forgot to handle I/O errors here" bug cannot ship.

One-for-one: the toy ↔ the build

FetchError is ControlError with the labels changed:

The build adds a Store variant (persistence failures, terminal) that the toy omits, but the shape is identical: a closed taxonomy, #[from] on exactly the I/O variant, and a single is_retryable that names the two retryable cases and lets everything else fall through to terminal.

Questions to lock

1. Why is is_retryable as a method on the error type strictly better than each caller deciding retry-worthiness itself? Name two concrete failure modes of the scattered version.
2. Why does #[from] belong on the Io variant but not on Worker/Invalid/NotFound? What does #[from] actually generate, and what uses it?
3. If you add a new retryable variant to the taxonomy, how many lines of retry logic must change — and why is that number what makes this design worth it?

Next: the build that turns this taxonomy into error.rs, plus the WorkerHandle seam and the wire Message enum in worker.rs and proto.rs.

Build: ControlError, the WorkerHandle Seam, and the Wire

Maps to: Phase 0 (control-core). Kind: Build.

Objective

Finish control-core by adding the three files the rest of the course dispatches through: error.rs (the ControlError taxonomy whose type answers "should I retry?"), worker.rs (the WorkerHandle trait — the seam the concept chapter promised, now in code), and proto.rs (the wire Message enum). Prove with tests that Worker/Protocol errors are retryable and the rest are terminal; that a concrete worker can be stored as Box<dyn WorkerHandle> and dispatched through; and that a Message serializes with a self-describing "type" tag and round-trips.

Scaffold

You already have the crate from the previous build — ids.rs and domain.rs are green. Here you extend it.

Create:

• crates/control-core/src/error.rs — ControlError and its two tests.
• crates/control-core/src/worker.rs — the WorkerHandle trait and its one async test.
• crates/control-core/src/proto.rs — the Message enum and its two tests.

Edit:

• crates/control-core/src/lib.rs — add pub mod error; pub mod worker; pub mod proto; and re-export ControlError, WorkerHandle, and Message alongside the ids and domain types.
• crates/control-core/Cargo.toml — three new dependencies.

New deps and why:

• thiserror — derives Error and writes each variant's Display message from its #[error("…")], and generates the From<std::io::Error> that #[from] needs. ([dependencies].)
• async-trait — rewrites the trait's async fn into a method returning a boxed future, which is what a vtable can hold; without it an async fn in a trait cannot be used behind dyn. ([dependencies].)
• tokio (with the macros and rt features) — supplies #[tokio::test], since the worker test must .await a dispatch. ([dev-dependencies].)

serde, serde_json, and pretty_assertions are already in the manifest from the domain build; proto.rs reuses them.

Expected result: cargo test -p control-core → 11 tests pass (the 6 from the domain build plus the 5 you add here):

• from error.rs: worker_and_protocol_are_retryable, invalid_and_not_found_are_terminal
• from worker.rs: a_worker_handle_can_be_boxed_as_dyn
• from proto.rs: register_is_tagged_and_roundtrips, assign_carries_a_full_job

The spec (givens)

error.rs — the error taxonomy. One enum, deriving Debug and thiserror::Error. Six variants, each carrying a String payload except the last, and each with an #[error("…")] message:

• Exactly the Io variant carries #[from]; the domain variants are constructed by hand where the problem is detected, so they get no #[from].
• Inherent method is_retryable(&self) -> bool. The rule: it returns true for Worker and Protocol and false for everything else — write it as one total matches! over the two retryable variants so a future terminal variant falls through to false automatically.

[→ Answer key](../appendix-answer-key.md#core-error)

worker.rs — the seam. One trait, annotated #[async_trait]:

• pub trait WorkerHandle: Send + Sync — the Send + Sync bound is required because a dyn WorkerHandle is moved between and shared across tokio tasks on different threads.
• fn id(&self) -> &str; — a stable identifier for logging and accounting. Note it is not async and returns a borrowed &str.
• async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError>; — run one job to completion. Takes Job and JobOutcome from domain, and the error from error. The method takes a concrete Job, not a generic — that is what keeps the trait object-safe (dyn-compatible).

[→ Answer key](../appendix-answer-key.md#core-worker)

proto.rs — the wire protocol. One enum:

• pub enum Message, deriving Debug, Clone, PartialEq, Eq, Serialize, Deserialize.
• Attribute: #[serde(tag = "type", rename_all = "snake_case")] — an internally tagged enum. Each variant serializes as a JSON object carrying a "type" discriminator ("register", "assign", "result", "heartbeat") alongside its fields, so a receiver can route a frame without a side channel.
• Struct-style variants:
  • Register { worker_id: String, capacity: u32 } — a worker announcing itself and how many jobs it can hold at once.
  • Assign { job: Job } — the coordinator handing a job to a worker.
  • Result { outcome: JobOutcome } — a worker returning a finished job.
  • Heartbeat { worker_id: String } — a liveness ping.

[→ Answer key](../appendix-answer-key.md#core-proto)

Concepts exercised

• Encoding the retry decision in the error type via a single is_retryable method, so no call site re-derives the policy.
• #[from] on exactly the I/O variant; explicit construction for the domain variants.
• An #[async_trait] trait with Send + Sync used as Box<dyn WorkerHandle> — the concrete seam behind the whole distributed design.
• Object-safety: why dispatch takes a concrete Job rather than a generic.
• serde's internally tagged enum (tag = "type") for a self-describing wire frame, and rename_all for snake_case discriminators.

The build loop (you drive)

Write each test first, predict its failure, run to see the red you predicted, then implement the minimum to green.

1. worker_and_protocol_are_retryable — asserts ControlError::Worker("died".into()).is_retryable() and the same for Protocol. Predict: with the enum not yet written, does this fail to compile or fail an assertion? Run, confirm it is a compile failure, then define the enum and is_retryable.

2. invalid_and_not_found_are_terminal — asserts !ControlError::Invalid(…).is_retryable() and !ControlError::NotFound(…).is_retryable(). Once your matches! names only Worker | Protocol, both of these should pass without touching the method.

3. a_worker_handle_can_be_boxed_as_dyn — a #[tokio::test]. Define a tiny EchoWorker inside the test module whose id() returns "echo" and whose dispatch returns a JobOutcome with the job's id and no records. Then store it as let worker: Box<dyn WorkerHandle> = Box::new(EchoWorker);, assert worker.id() == "echo", and assert the outcome of worker.dispatch(job).await carries the same job_id you sent. Predict: what does the compiler say if you forget #[async_trait] on the impl block?

4. register_is_tagged_and_roundtrips — build a Message::Register { worker_id: "w1".into(), capacity: 4 }, serde_json::to_string it, assert the JSON contains("\"type\":\"register\""), then from_str back and assert equal. This pins the discriminator: it is the tag = "type" + rename_all doing their job. Predict what the "type" value would be if you dropped rename_all and the variant were RegisterWorker instead.

5. assign_carries_a_full_job — build a Message::Assign { job } around a full Job (nested EvalJob), encode, assert contains("\"type\":\"assign\""), decode, assert equal. This proves a whole Job — ids, status, spec — travels inside one tagged frame and comes back byte-identical.

Done when

cargo test -p control-core shows 11 green (6 from the domain build, 5 from this one); you have watched worker_and_protocol_are_retryable fail to compile before the enum exists; you can say why dispatch takes a concrete Job rather than a generic, and what error[E0038] would mean if it did not; and a Message::Register serializes with "type":"register". Commit. The Core arc is complete — every seam the rest of the course leans on now exists and is tested.

Concept-Check: The Core

Kind: Quiz. One pass over the whole arc before you move on.

The Core arc installed the vocabulary every other arc speaks: id newtypes that the compiler keeps apart, status enums that serialize snake_case, an error taxonomy whose type answers "should I retry?", the WorkerHandle seam the distributed layer plugs into, and the tagged wire Message. This check mixes compiler-verified tracing questions with judgment questions across all of it.

If a question stings, the fix is upstream: Newtypes and Transparent serde for the id and status questions, and An Error Taxonomy That Knows What to Retry for the retry and #[from] questions. Re-read the section, then come back — the last five arcs stand on these types.

Next: Part III, the Eval Workload arc — a ModelClient trait object tested against a mock HTTP server, and the append-only file contract that logs every raw response.

Concept: Trait-Object Clients and Mocking the Network

Kind: Concept. New crate: async-trait — lets an async fn live inside a trait that you can still use behind dyn.

The whole workload is a network round-trip

Step back and look at what a job in this system actually does. It takes a batch of vignettes, and for each one it POSTs a prompt to a model endpoint and waits for the reply. That is the entire workload. Strip away the scheduling, the storage, the telemetry, and what remains is a loop around one HTTP call.

That single fact drives most of the architecture that follows. Because the work is a network round-trip — almost all of it spent waiting on a remote server — a worker is barely using its CPU while a job runs. One machine could keep hundreds of these in flight at once, and the work parallelizes trivially across machines because each call is independent. The workload is worth distributing precisely because it is I/O, not computation. Hold onto that; it is why Parts VI and VIII exist.

For this arc, the consequence is narrower: the one thing a job depends on is a way to turn a prompt into a response over the network. Name that dependency, put it behind a seam, and everything downstream — run_eval, the local worker, the remote worker — can be written and tested without ever touching a real model.

The seam: ModelClient

You already know the trait-object seam from Part I: depend on a trait, not a concrete type, so the concrete type can be swapped. Here the trait is the model call itself:

#[async_trait]
pub trait ModelClient: Send + Sync {
    fn model_name(&self) -> &str;
    async fn generate(&self, prompt: &str) -> Result<ModelResponse, ControlError>;
}

Two methods, and only two. model_name reports the pinned model string; generate takes the entire prompt (single-turn — the prompt is the whole input) and returns a ModelResponse or a ControlError. run_eval will accept &dyn ModelClient and never know whether it is talking to a real HTTP client or a stub. That is the whole payoff: the workload logic is provider-agnostic and testable, and the transport is a detail chosen at the edge.

The Send + Sync bound is not decoration. A future produced by generate will be moved onto a runtime and, later in the course, handed between worker tasks on different threads. Send + Sync is the compiler's promise that doing so is safe. You met these bounds in the async arc; here is where they earn their keep.

Why async fn in a trait needs #[async_trait]

Try the obvious thing — a bare async fn in the trait, used behind dyn — and the compiler stops you cold:

#![allow(unused)]
fn main() {
trait QuoteClient {
    async fn quote(&self, topic: &str) -> String;
}

fn take_seam(_c: &dyn QuoteClient) {}
}

error[E0038]: the trait `QuoteClient` is not dyn compatible
 --> src/lib.rs:6:19
  |
6 | fn take_seam(_c: &dyn QuoteClient) {}
  |                   ^^^^^^^^^^^^^^^ `QuoteClient` is not dyn compatible
  |
note: for a trait to be dyn compatible it needs to allow building a vtable
 --> src/lib.rs:3:14
  |
2 | trait QuoteClient {
  |       ----------- this trait is not dyn compatible...
3 |     async fn quote(&self, topic: &str) -> String;
  |              ^^^^^ ...because method `quote` is `async`

Read the reason on the last line: the method is async. An async fn desugars to a function returning impl Future, and every call site can produce a different concrete future type of a different size. A dyn object needs one fixed vtable with one fixed return layout — and "some future, size unknown" is not that. So the trait is not dyn compatible, and &dyn QuoteClient will not compile.

#[async_trait] is the fix. It rewrites each async fn in the trait to return Pin<Box<dyn Future + Send>> — a heap-allocated, fixed-size future behind a pointer. Now every method has one uniform return type, the vtable can be built, and &dyn ModelClient works. The cost is one allocation per call, which against a network round-trip is free. (Rust is steadily lifting the language-level restriction, but for a dyn-dispatched trait like this one, #[async_trait] is still the standard tool.)

A server error is a retryable Worker error

Recall the taxonomy from Part II: ControlError splits into retryable (Worker, Protocol) and terminal (Invalid, NotFound). The model call is where that distinction gets its first real workout, and the rule is deliberate:

• The HTTP call fails to complete, or the server answers with a non-2xx status → ControlError::Worker, which is_retryable() reports as true. A 503 means the model backend hiccuped; the same request to another worker may well succeed. That is exactly what retryable means.
• The server answers 2xx but the body is not the shape we expect → ControlError::Invalid, terminal. A malformed reply will be malformed no matter who asks; retrying only wastes time.

That mapping is a policy decision, and it is worth making it concrete before you wire it to HTTP. Here it is on a toy error type — a non-2xx becomes a retryable server fault, an unparseable body becomes a terminal one:

/// A toy of the harness's error taxonomy: retryable vs terminal.
enum QuoteError {
    /// The server answered, but with a 5xx — a transient fault. Retryable.
    Server(String),
    /// The reply body was not the shape we expected. Terminal.
    Bad(String),
}

impl QuoteError {
    fn is_retryable(&self) -> bool {
        matches!(self, QuoteError::Server(_))
    }
    fn message(&self) -> &str {
        match self {
            QuoteError::Server(m) | QuoteError::Bad(m) => m,
        }
    }
}

/// Turn a raw HTTP outcome into our error taxonomy, exactly as `quote` would:
/// a non-2xx status is a transient Server fault; a body we cannot parse is Bad.
fn classify(status: u16, parses: bool) -> Result<&'static str, QuoteError> {
    if !(200..300).contains(&status) {
        return Err(QuoteError::Server(format!("quote status {status}")));
    }
    if !parses {
        return Err(QuoteError::Bad("missing `text` field".into()));
    }
    Ok("a witty maxim")
}

fn main() {
    for (status, parses) in [(200, true), (503, true), (200, false)] {
        match classify(status, parses) {
            Ok(text) => println!("{status} parses={parses:<5} -> Ok({text:?})"),
            Err(e) => println!(
                "{status} parses={parses:<5} -> Err({:?})  retryable={}",
                e.message(),
                e.is_retryable()
            ),
        }
    }
}

200 parses=true  -> Ok("a witty maxim")
503 parses=true  -> Err("quote status 503")  retryable=true
200 parses=false -> Err("missing `text` field")  retryable=false

The toy, end to end: a QuoteClient

Here is the exact shape the build asks of you, on a service that cannot be mistaken for the answer key: a quote API. POST {base}/quote with a JSON body; the reply is { "text": ..., "credits": ... }; the client parses it into a clean public type and maps failures into the taxonomy above. It is behind an #[async_trait] trait, so a stub and the real HTTP client are interchangeable.

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

/// A witticism plus the credits the call cost. Flat, typed, ours.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Quote {
    pub text: String,
    pub credits: u32,
}

/// The seam: anything that turns a topic into a quote. Mockable.
#[async_trait]
pub trait QuoteClient: Send + Sync {
    fn source_name(&self) -> &str;
    async fn quote(&self, topic: &str) -> Result<Quote, QuoteError>;
}

/// A concrete client over HTTP with an injectable base URL, so a test can point
/// it at a mock instead of a real provider.
pub struct HttpQuoteClient {
    base_url: String,
    source: String,
    http: reqwest::Client,
}

impl HttpQuoteClient {
    pub fn new(base_url: impl Into<String>, source: impl Into<String>) -> Self {
        Self { base_url: base_url.into(), source: source.into(), http: reqwest::Client::new() }
    }
}

#[derive(Serialize)]
struct QuoteRequest<'a> { source: &'a str, topic: &'a str }

#[derive(Deserialize)]
struct QuoteReply { text: String, credits: u32 }

#[async_trait]
impl QuoteClient for HttpQuoteClient {
    fn source_name(&self) -> &str {
        &self.source
    }

    async fn quote(&self, topic: &str) -> Result<Quote, QuoteError> {
        let resp = self.http
            .post(format!("{}/quote", self.base_url.trim_end_matches('/')))
            .json(&QuoteRequest { source: &self.source, topic })
            .send().await
            // A dropped or refused connection is worth another attempt.
            .map_err(|e| QuoteError::Server(e.to_string()))?;
        if !resp.status().is_success() {
            // A non-2xx is a transient server fault → retryable.
            return Err(QuoteError::Server(format!("quote status {}", resp.status())));
        }
        // A 2xx we cannot parse is terminal → Bad.
        let reply: QuoteReply = resp.json().await
            .map_err(|e| QuoteError::Bad(e.to_string()))?;
        Ok(Quote { text: reply.text, credits: reply.credits })
    }
}

The Raw* split is the same trick from the mocking chapter: QuoteReply models only the fields we consume, and serde ignores the rest. Quote is the flat, owned type callers actually want.

Mock it — briefly, since you know the drill

You met wiremock in Course 1: it starts a real HTTP server on a random local port and hands you server.uri(), which you pass in as the client's base_url. The client makes a genuine request; the server returns exactly the canned response you configured. No key, no spend, no flake, and the full request-building and response-parsing path runs for real. The only new wrinkle here is that the thing under test is an #[async_trait] implementation.

#[cfg(test)]
mod tests {
    use super::*;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    #[tokio::test]
    async fn quote_client_posts_topic_and_parses_reply() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/quote"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "text": "Fortune favors the bold.", "credits": 1
            })))
            .expect(1) // guard against the silent-404 trap
            .mount(&server)
            .await;

        let client = HttpQuoteClient::new(server.uri(), "seneca");
        let q = client.quote("courage").await.unwrap();
        assert_eq!(q, Quote { text: "Fortune favors the bold.".into(), credits: 1 });
    }

    #[tokio::test]
    async fn server_error_is_retryable() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(503))
            .mount(&server)
            .await;
        let client = HttpQuoteClient::new(server.uri(), "seneca");
        let err = client.quote("x").await.unwrap_err();
        assert!(err.is_retryable()); // the 503 became a retryable Server fault
    }
}

Both tests pass — one real HTTP round trip each, zero network beyond loopback. The second is the important one: it pins the policy, not just the plumbing. It asserts that a 503 from the endpoint surfaces as a retryable error, which is the promise the scheduler will rely on.

To run these two examples yourself, make a scratch crate with serde/serde_json, tokio (features ["full"]), async-trait, and reqwest (features ["json"]) as dependencies, and wiremock = "0.6" under [dev-dependencies] — the same set control-eval declares. async-trait, reqwest, and wiremock are not on the Rust playground, so there is no play button.

One-for-one with the build

The toy maps onto control-eval exactly:

• QuoteClient ↔ ModelClient (the #[async_trait] seam, Send + Sync)
• HttpQuoteClient ↔ HttpModelClient (injectable base_url, a reqwest::Client inside)
• POST /quote ↔ POST /v1/generate
• quote(topic) ↔ generate(prompt)
• QuoteRequest/QuoteReply ↔ the request/response structs serde maps to the wire
• Quote ↔ ModelResponse (flat, owned, { text, usage })
• QuoteError::Server on a non-2xx ↔ ControlError::Worker, retryable
• QuoteError::Bad on an unparseable body ↔ ControlError::Invalid, terminal

Same shape, different domain. Build the model client and you have built the quote client with the labels changed.

Questions to lock

1. Why is the eval workload described as "a network round-trip," and why does that make it worth distributing across workers?
2. What exact compiler error do you get from a bare async fn in a trait used behind dyn, and what does #[async_trait] change to fix it?
3. A model endpoint returns 503. Which ControlError variant should generate produce, is it retryable, and why is that the right call?
4. The mock test aims the client at server.uri() instead of a real provider. What design property of the client makes that possible, and where else does that same seam pay off?

Build: ModelClient + run_eval

Maps to: Phase 1 (control-eval). Kind: Build.

Objective

Create the control-eval crate. Define the ModelClient seam (with #[async_trait]), one concrete HttpModelClient over reqwest, and run_eval — the pure workload a job performs. By the end you have the thing every worker in this system ultimately runs, tested against a mock with zero network spend.

Scaffold

Create (new crate — add crates/control-eval to the workspace members):

• crates/control-eval/Cargo.toml
  • [dependencies]: control-core = { path = "../control-core" } (the domain and error taxonomy live there), plus serde, serde_json, async-trait, reqwest, tokio — all { workspace = true }.
    • reqwest (features ["json"]) is the HTTP client; .json() on requests and responses is why the feature matters.
    • async-trait is what lets generate be an async fn in a dyn-compatible trait (see the concept chapter for the E0038 you get without it).
  • [dev-dependencies]: wiremock, pretty_assertions, tokio — all { workspace = true }.
• crates/control-eval/src/lib.rs — pub mod client; and pub mod eval;, re-exporting the public types.
• crates/control-eval/src/client.rs — the trait, ModelResponse, and HttpModelClient.
• crates/control-eval/src/eval.rs — run_eval.

Dependencies this chapter exercises: async-trait (async method behind dyn), reqwest (the round trip), serde/serde_json (wire ↔ types), wiremock (dev — the mock server).

Expected result: cargo test -p control-eval → 3 tests pass (model_client_posts_prompt_and_parses_reply, server_error_is_a_retryable_worker_error, run_eval_produces_one_record_per_vignette).

The spec (givens)

ModelResponse and the ModelClient trait

/// A model's reply plus its token accounting.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModelResponse {
    pub text: String,
    pub usage: Usage, // from control-core
}

#[async_trait]
pub trait ModelClient: Send + Sync {
    fn model_name(&self) -> &str;
    async fn generate(&self, prompt: &str) -> Result<ModelResponse, ControlError>;
}

Exactly two methods. model_name returns the pinned model string; generate is single-turn — the prompt is the entire input. The Send + Sync bound is required so futures can move across worker threads later.

→ Answer key

HttpModelClient and the wire shapes

HttpModelClient holds a base_url, a model string, and a reqwest::Client. Construct it with new(base_url, model) where both accept impl Into<String>. The base_url is injectable — a test passes server.uri(), production passes the real endpoint.

generate POSTs to {base_url}/v1/generate (trim a trailing / from the base first). The request and response bodies:

// Request body → POST {base_url}/v1/generate
{ "model": "claude", "prompt": "the vignette prompt" }

// Success response body (HTTP 2xx)
{ "text": "the model's reply", "usage": { "input_tokens": 120, "output_tokens": 8 } }

Define two private structs — a Serialize request borrowing &str fields, and a Deserialize response modelling only text and usage. serde ignores any other fields the endpoint sends.

→ Answer key

Error mapping (the policy the test pins)

send() fails to complete            → ControlError::Worker   (retryable)
response status is not 2xx          → ControlError::Worker   (retryable)   e.g. "model status 503"
2xx body fails to deserialize       → ControlError::Invalid  (terminal)

A non-2xx or a dropped connection is a transient fault worth re-dispatching; a malformed 2xx body will fail identically on retry and must surface. This is the Part II taxonomy applied to the network.

→ Answer key

run_eval

pub async fn run_eval(
    client: &dyn ModelClient,
    spec: &EvalJob,
) -> Result<Vec<ResponseRecord>, ControlError>;

Pose every vignette in spec.vignettes to the client, in order, and collect one ResponseRecord per vignette. Each record carries vignette_id and prompt from the vignette, model and epoch from the job spec (not from the client), and response/usage from the ModelResponse. A single failing call ?-propagates and fails the whole job — the scheduler decides later whether to retry, based on is_retryable.

→ Answer key

Concepts exercised

• An #[async_trait] trait as a mockable seam (&dyn ModelClient).
• Mapping HTTP outcomes onto a retryable-vs-terminal error taxonomy.
• Request/response modelling with serde — declaring only the fields you consume.
• Testing an async client against a wiremock server with an injected base URL.
• A pure workload function that depends on the seam, not the concrete client.

The build loop (you drive)

Test 1 — model_client_posts_prompt_and_parses_reply (in client.rs, #[tokio::test])

1. Write the failing test. Start a MockServer, mount a Mock matching method("POST") and path("/v1/generate") that responds 200 with the success body above. Build HttpModelClient::new(server.uri(), "claude"), call generate("decide"), and assert resp.text == "maneuver" and resp.usage == Usage { input_tokens: 120, output_tokens: 8 }.
2. Predict: before you implement generate, what status will the client see if you POST to /v1/generte (typo) instead? Recall the silent-404 trap.
3. Run — it fails to compile (no generate yet).
4. Implement ModelResponse, the trait, the request/response structs, and generate — POST, check status, parse. What it does is specified above; how you arrange it is yours.
5. Run green, commit.

Test 2 — server_error_is_a_retryable_worker_error (in client.rs, #[tokio::test])

1. Write the failing test. Mount a mock that responds 503 to any POST. Build the client, call generate("x"), take unwrap_err(), and assert err.is_retryable().
2. Predict: if generate treated every error the same and returned ControlError::Invalid, would this test pass? What would is_retryable() report, and what would the scheduler then do with a transient 503?
3. Run, confirm the assertion, adjust the status-check branch so a non-2xx maps to Worker.
4. Run green, commit.

Test 3 — run_eval_produces_one_record_per_vignette (in eval.rs, #[tokio::test])

1. Write the failing test. Define a StubClient that implements ModelClient with no network — it counts calls in an AtomicUsize and returns ModelResponse { text: format!("re: {prompt}"), .. }. Build an EvalJob with two vignettes ("a"/"one", "b"/"two"), model: "claude", epoch: 3. Call run_eval(&client, &spec) and assert: records.len() == 2, the client saw exactly 2 calls, records[0].model == "claude", records[0].epoch == 3, records[1].vignette_id == "b", records[1].response == "re: two".
2. Predict: the stub reports model_name() == "stub", but the job's model is "claude". Which one lands in each record's model field — and why is taking it from the spec rather than the client the correct choice for a blinded eval?
3. Run — fails to compile (no run_eval).
4. Implement run_eval: loop the vignettes, .await each generate, build one record each from the vignette + spec + response.
5. Run green, commit.

Done when

cargo test -p control-eval shows 3 passing tests, the 503 case is proven retryable, and run_eval produces exactly one record per vignette with model/epoch drawn from the job spec.

Concept: Append-Only Logs and the File Contract

Kind: Concept. No new crate — this is std::fs and serde_json doing one disciplined thing.

The log is the instrument's memory

The client chapter gave you the act of an eval: pose a prompt, get a response. This chapter is about what happens to that response after the call returns. It has to land somewhere durable, because the whole point of running an eval is to have a record you can go back to — re-score, audit, reproduce, compare against last month's run. A model response that lives only in memory is not evidence; it is a rumour.

So the coordinator writes every response to a file, and the shape of that file is a contract — the same JSON Lines contract the rest of Panoptes already speaks. The harness upstream generates a manifest of vignettes (one JSON object per line) and drops it where the coordinator can read it. The coordinator runs the eval and writes a response log (again, one JSON object per line) that the coding stage downstream reads. Neither side owns a shared database or an API; they hand each other files. That is deliberate — a file on disk is the most boring, most portable, most debuggable interface two stages can share, and jq can read it at 3 a.m. when nothing else works.

Two functions carry this contract, and they are almost aggressively small:

• read a manifest: parse JSONL text into Vec<Vignette>.
• append records: serialize each ResponseRecord to one line and add it to the log.

The reading half is unremarkable — you have parsed JSONL before. The writing half hides the one decision this chapter exists to make: the log is append-only.

Why append-only is a contract, not a convenience

Append-only means exactly one thing: a write may add lines to the end of the file, and it may never touch a byte that is already there. No rewrite, no truncate, no in-place edit. History only grows.

That sounds like a small implementation detail — .append(true) versus .write(true) on an OpenOptions. It is not. It is the property the entire instrument's trustworthiness rests on, for three reasons that compound:

• Durability. responses.jsonl is the dataset of record. If a run could truncate it, a single buggy re-run — pointed at the wrong path, restarted after a crash — could erase months of collected responses in one open(). Append-only makes that class of accident structurally impossible: the file-open mode simply cannot overwrite. The safety is in the mode flag, not in remembering to be careful.
• Reproducibility. Each response is one immutable line, written once and never edited. That means the log is the run — a faithful, ordered transcript of what the model actually said, byte for byte. Re-score it a year later and you are scoring the same responses, not a mutated copy where some later pass "fixed" a few. An eval you cannot reproduce is not a measurement; it is an anecdote.
• Concurrency and crash-safety. Because a write only ever extends the file, two things fall out for free. A crash mid-run leaves a log that is shorter than intended but never corrupt — every complete line before the crash is still a valid record. And append is the one file operation that composes safely across separate runs: a second run appends its lines after the first run's, and the first run's records are exactly where they were.

Hold onto the middle one especially — each eval response is one immutable line — because it is the sentence that turns "a log file" into "an eval instrument." The rest of this chapter is that sentence made mechanical.

The toy, end to end: an append-only event log

Here is the exact shape the build asks of you, on a domain that cannot be mistaken for the answer key: a tiny event log. Each event is one immutable line — a logical tick and a reading. One function parses the log back into events; one function appends events, append-only. It runs on std and serde_json alone, so there is a play button.

use serde::{Deserialize, Serialize};
use std::fs::OpenOptions;
use std::io::Write;

/// One immutable event: when it happened, and what was observed.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct Event {
    at: u64, // a logical tick here; a wall-clock timestamp in the real instrument
    reading: String,
}

/// Parse a JSONL log into events — one line, one immutable record.
fn parse_log(text: &str) -> Vec<Event> {
    text.lines()
        .filter(|l| !l.trim().is_empty())
        .map(|l| serde_json::from_str(l).expect("each line is one JSON event"))
        .collect()
}

/// Append events to the log as JSONL. Append-only: `.append(true)` never
/// truncates what is already there — the write starts at the end of the file.
fn append_events(path: &std::path::Path, events: &[Event]) -> std::io::Result<()> {
    let mut file = OpenOptions::new().create(true).append(true).open(path)?;
    let mut buf = String::new();
    for e in events {
        buf.push_str(&serde_json::to_string(e).unwrap());
        buf.push('\n');
    }
    file.write_all(buf.as_bytes())
}

fn main() -> std::io::Result<()> {
    let path = std::env::temp_dir().join("p3fin-eventlog.jsonl");
    let _ = std::fs::remove_file(&path); // fresh start so the demo is repeatable

    // Two separate runs write to the same log — the second must not erase the first.
    append_events(
        &path,
        &[
            Event { at: 1, reading: "boot".into() },
            Event { at: 2, reading: "warm".into() },
        ],
    )?;
    append_events(&path, &[Event { at: 3, reading: "hot".into() }])?;

    // Read the whole history back.
    let text = std::fs::read_to_string(&path)?;
    let events = parse_log(&text);

    println!("lines on disk: {}", text.lines().count());
    println!("events parsed: {}", events.len());
    for e in &events {
        println!("  at={} {}", e.at, e.reading);
    }
    // The first run's record is still the first record — history is intact.
    println!(
        "first record still first: {}",
        events[0] == Event { at: 1, reading: "boot".into() }
    );
    Ok(())
}

lines on disk: 3
events parsed: 3
  at=1 boot
  at=2 warm
  at=3 hot
first record still first: true

Two things earn their keep here. The at: 1 event survives the second write untouched — first record still first: true is the append-only property, proven, not asserted. And the two writes came from two separate append_events calls, which is the crash-safety story in miniature: imagine the process had died between them, and you would be left with exactly the first two lines — short, but every one of them a whole, valid record.

The trap: File::create truncates, silently

The prediction matters because the wrong choice does not error, warn, or misbehave visibly — it quietly destroys data and returns Ok. std::fs::File::create opens for writing and truncates the file to zero length first. Point a second run at a log opened that way and every earlier record is gone before your first new byte lands. Here is that mistake, runnable, so you see the damage with your own eyes:

use std::fs::{File, OpenOptions};
use std::io::Write;

fn main() -> std::io::Result<()> {
    let path = std::env::temp_dir().join("p3fin-trap.jsonl");
    let _ = std::fs::remove_file(&path);

    // First "run" appends two records — the honest way.
    let mut f = OpenOptions::new().create(true).append(true).open(&path)?;
    writeln!(f, "{{\"at\":1,\"reading\":\"boot\"}}")?;
    writeln!(f, "{{\"at\":2,\"reading\":\"warm\"}}")?;
    drop(f);

    // Second "run" opens with File::create — the trap. This TRUNCATES to empty
    // before the first byte is written, so the earlier history is gone.
    let mut f = File::create(&path)?;
    writeln!(f, "{{\"at\":3,\"reading\":\"hot\"}}")?;
    drop(f);

    let text = std::fs::read_to_string(&path)?;
    println!("lines after the truncating write: {}", text.lines().count());
    print!("{text}");
    Ok(())
}

lines after the truncating write: 1
{"at":3,"reading":"hot"}

One line. The boot and warm records are gone, and nothing anywhere said so — no panic, no Err, no log line. This is why the append-only property lives in a test in the build and not in a comment: "remember to open in append mode" is exactly the kind of instruction a tired future maintainer forgets, and the failure is invisible until someone goes looking for data that no longer exists. A test that appends twice and asserts the line count grew is cheap, permanent insurance against silently shredding the dataset of record.

Reading is the easy half — but note who owns the failure

The reading side has no such trap, but it makes one small policy choice worth naming. A manifest line that is not valid JSON — a truncated file, a hand-edit gone wrong — is a terminal failure, not a retryable one. There is no server to blame and no point trying again: the same bad bytes will fail to parse identically on the next attempt. So the parse maps a serde error to ControlError::Invalid, the terminal variant from the Part II taxonomy — the same call you make when a model returns a 2xx body you cannot deserialize. Bad input is bad input, whether it arrives over a socket or off a disk.

enum ControlError {
    Worker(String),  // transient — worth retrying
    Invalid(String), // terminal  — a retry changes nothing
}

/// Parse a JSONL manifest into (id, prompt) pairs. A malformed line is terminal.
fn parse_manifest(text: &str) -> Result<Vec<(String, String)>, ControlError> {
    let mut out = Vec::new();
    for line in text.lines().filter(|l| !l.trim().is_empty()) {
        // Stand-in for `serde_json::from_str`: split "id|prompt".
        match line.split_once('|') {
            Some((id, prompt)) => out.push((id.to_string(), prompt.to_string())),
            None => return Err(ControlError::Invalid(format!("bad manifest line: {line:?}"))),
        }
    }
    Ok(out)
}

fn main() {
    let good = "ca_geo-030|decide\nca_geo-060|choose\n";
    match parse_manifest(good) {
        Ok(v) => println!("good -> Ok, {} vignettes", v.len()),
        Err(_) => println!("good -> unexpected error"),
    }

    let bad = "ca_geo-030|decide\nthis line has no delimiter\n";
    match parse_manifest(bad) {
        Ok(_) => println!("bad -> unexpectedly ok"),
        Err(ControlError::Invalid(m)) => println!("bad -> Invalid (terminal): {m}"),
        Err(ControlError::Worker(m)) => println!("bad -> Worker (retryable): {m}"),
    }
}

good -> Ok, 2 vignettes
bad -> Invalid (terminal): bad manifest line: "this line has no delimiter"

The real parse_manifest uses serde_json::from_str in place of that split_once, and maps its error to ControlError::Invalid the same way — a malformed manifest is a terminal fault the scheduler will not waste a retry on.

One-for-one with the build

The toy maps onto control-eval's contract.rs exactly:

• parse_log (JSONL text → Vec<Event>) ↔ parse_manifest (JSONL text → Vec<Vignette>)
• reading the file then parsing ↔ load_manifest (tokio::fs::read_to_string then parse_manifest)
• append_events with .create(true).append(true) ↔ append_records, append-only
• Event { at, reading }, one per line ↔ ResponseRecord, one immutable line per response
• a malformed line → terminal ↔ serde error → ControlError::Invalid
• "first record still first" after a second write ↔ the append_records_is_append_only test's line-count assertion

Same shape, different domain. The real thing swaps std::fs for tokio::fs (the coordinator is async) and the logical at for a real ResponseRecord, but the contract — parse in, append-only out, each response one immutable line — is identical.

Questions to lock

1. What exactly does "append-only" forbid, and which single OpenOptions flag enforces it? What does File::create do instead, and why is that dangerous for a response log?
2. Give the three properties append-only buys the eval instrument (durability, reproducibility, crash-safety) and one sentence on why each follows from "history only grows."
3. A manifest line fails to parse. Is that a retryable Worker error or a terminal Invalid one, and why does retrying change nothing?
4. Why is the append-only property pinned by a test that appends twice, rather than trusted to a comment telling the maintainer to open in append mode?

Build: The Manifest and the JSONL Record Log

Maps to: Phase 1 (control-eval). Kind: Build.

Objective

Add the file contract to control-eval: the functions that read a vignette manifest and append response records to the log. This is the boundary where the coordinator meets the rest of Panoptes — a manifest comes in as JSONL, a response log goes out as JSONL, and the log is append-only. By the end you have parse_manifest, load_manifest, and append_records, with a test that pins the property the whole instrument depends on: a second append never overwrites the first.

Scaffold

Create — one new module in the crate you built in the previous chapter:

• crates/control-eval/src/contract.rs — parse_manifest, load_manifest, append_records.
• In crates/control-eval/src/lib.rs, add pub mod contract; and re-export the three functions.

No new dependencies. You already have everything this chapter needs:

• serde_json — one record per line, in and out.
• tokio (fs, io) — tokio::fs::read_to_string, tokio::fs::OpenOptions, AsyncWriteExt::write_all. The coordinator is async, so file I/O is async too.
• control-core — Vignette and ResponseRecord are the wire shapes; ControlError is the failure type.
• pretty_assertions (dev) — already declared for the crate's tests.

Expected result: cargo test -p control-eval contract → 2 tests pass (parse_manifest_reads_jsonl, append_records_is_append_only).

The spec (givens)

parse_manifest and load_manifest

pub fn parse_manifest(text: &str) -> Result<Vec<Vignette>, ControlError>;
pub async fn load_manifest(path: impl AsRef<Path>) -> Result<Vec<Vignette>, ControlError>;

parse_manifest takes the whole manifest text and returns one Vignette per non-blank line. Each line is a JSON object { "id": ..., "prompt": ... } — serde_json::from_str straight into a Vignette. Skip blank lines (a trailing newline must not become an empty-line parse error). A line that fails to parse is a terminal fault: map the serde error to ControlError::Invalid, not Worker — a malformed manifest fails identically on retry, so there is nothing to retry.

load_manifest is the thin async wrapper: tokio::fs::read_to_string(path).await, then hand the text to parse_manifest. Let the ? on the read turn an I/O error into ControlError via the From impl control-core already provides.

→ Answer key

append_records

pub async fn append_records(
    path: impl AsRef<Path>,
    records: &[ResponseRecord],
) -> Result<usize, ControlError>;

Open the log with tokio::fs::OpenOptions::new().create(true).append(true) — create it if missing, and append, never truncate. Serialize each record to one JSON line (serde_json::to_string, then push a '\n'), write the batch, and return the count of records written. A serialization failure maps to ControlError::Invalid.

The mode flags are the whole point of the chapter. .append(true) is what makes the log the durable dataset of record: a write extends the file and can never overwrite a byte already on disk. .create(true) makes the first-ever write to a fresh path succeed instead of erroring on a missing file.

→ Answer key

Concepts exercised

• Append-only file semantics via OpenOptions — .create(true).append(true), and why anything that truncates is wrong here.
• JSONL as a stage-to-stage contract: one JSON object per line, in and out.
• Mapping a parse or serialize failure to the terminal ControlError::Invalid, not the retryable Worker.
• Async file I/O with tokio::fs and AsyncWriteExt.

The build loop (you drive)

Test 1 — parse_manifest_reads_jsonl (in contract.rs, plain #[test])

1. Write the failing test. Build a two-line JSONL string — {"id":"ca_geo-030","prompt":"decide"} and {"id":"ca_geo-060","prompt":"choose"}, each followed by \n. Call parse_manifest(text).unwrap() and assert vs.len() == 2, vs[0].id == "ca_geo-030", vs[1].prompt == "choose".
2. Predict: str::lines() is kind about a trailing \n — "a\nb\n".lines() yields just ["a", "b"], no empty tail. But a blank line between records ("a\n\nb\n") does yield an empty "" in the middle. If a manifest ever carried such a blank line and you skipped the filter, what would serde_json::from_str("") return — Ok, or an Err you then map to Invalid? That is why the filter is there even though the trailing newline is harmless.
3. Run — it fails to compile (no parse_manifest yet).
4. Implement parse_manifest: iterate text.lines(), filter out blank lines, parse each with serde_json::from_str, mapping the error to ControlError::Invalid. How you arrange it is yours.
5. Run green, commit.

Test 2 — append_records_is_append_only (in contract.rs, #[tokio::test])

This is the load-bearing test of the chapter. It exists to prove the log only ever grows.

1. Write the failing test. Pick a temp path under std::env::temp_dir() (join a name that includes std::process::id() so parallel test runs don't collide), and remove_file it first so the test starts clean. Then call append_records twice — once with one record, once with two — and afterwards read the file back with tokio::fs::read_to_string and assert text.lines().count() == 3.
2. Predict: this is the whole point of the exercise. Suppose you implemented append_records with .write(true).truncate(true) (or File::create) instead of .append(true). After the two calls, how many lines does the file hold — 3, or 1? Which call's records survive? Say it out loud before you write a line of the implementation; the test is built to fail loudly on exactly that mistake.
3. Run — it fails to compile (no append_records yet).
4. Implement append_records: open with .create(true).append(true), serialize each record to a line, write_all the batch, return the count.
5. Run green, commit. Then, to feel the test working, temporarily swap .append(true) for .truncate(true).write(true) and watch the assertion fail with 2 != 3 — then put it back. That failing run is the proof the test is guarding what you think it guards.

Done when

cargo test -p control-eval contract shows 2 passing tests, parse_manifest turns a malformed line into a terminal ControlError::Invalid, and append_records_is_append_only proves the second append left the first append's records exactly where they were.

Concept-Check: The Workload

The eval workload is the beating heart of the coordinator: one network round-trip per vignette, a taxonomy that knows a transient 503 from a terminal bad body, and an append-only log that turns those responses into evidence. If these are solid, the persistence and scheduler arcs have a workload worth scheduling.

Concept: sqlx, SQLite, and Migrations

Kind: Concept. New crate: sqlx — this chapter shows it working before you build with it.

This is the persistence arc. Everything before it kept state in memory or in an append-only file. Now the control plane needs a place where a run's status survives a process restart, where two schedulers can both look at the same queue, and where "this job finished" and "the run's done-count went up" are recorded as one indivisible fact. That place is a SQLite database, and the crate that gets Rust talking to it is sqlx.

This chapter is about the crate, not the schema — the schema is the next chapter's build. Here we answer three questions that trip people up the first time: what sqlx is (and the one big decision we make about how to use it), how you open a database and keep it alive, and how the table definitions get created. We do all of it on a toy notes table so nothing here can be mistaken for the answer key.

What sqlx is, and the decision that shapes the whole course

sqlx is an async SQL toolkit. You hand it SQL as text, it sends that text to the database, and it hands you back rows you pull columns out of. It is not an ORM — there is no Note.save(), no query builder that writes SQL for you out of method chains. You write SQL; sqlx runs it asynchronously and maps the results.

Here is the fork in the road, and it is worth understanding before you type anything, because it explains an oddity you would otherwise trip over. sqlx offers two ways to run a query:

• The compile-time-checked macros (sqlx::query!, sqlx::query_as!). These connect to a real database at compile time, run your SQL against it, and verify the column names and types match the Rust you wrote. A typo in a column name becomes a build error.
• The runtime query API (sqlx::query, sqlx::query_as — no !). These treat the SQL as an ordinary string that is checked only when it runs.

The macros sound strictly better — who would not want their SQL checked at compile time? But they carry a price: to check your SQL at build time, the compiler needs a live database to check against, which sqlx locates through a DATABASE_URL environment variable. Watch what happens when it is not set:

error: set `DATABASE_URL` to use query macros online, or run `cargo sqlx prepare` to update the query cache
 --> examples/compile_time.rs:5:15
  |
5 |     let row = sqlx::query!("SELECT 1 as one").fetch_one(&pool).await?;
  |               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

That is a build failure. It means anyone who clones this repo and types cargo test needs a database provisioned and an environment variable pointing at it, or the workspace does not compile. For a teaching codebase — and for CI — that is a heavy tax.

Say this back to yourself once: runtime API, because there is no DATABASE_URL at build time, so SQL errors surface in tests rather than at compile time. That sentence is the answer to the first quiz question and the reason the store code looks the way it does.

Connecting: the pool, and two options that matter

You do not hand sqlx a bare connection; you hand it a pool. A pool is a small set of reusable connections. Even when the set has size one, the pool is what owns the connection's lifetime. Here is the toy store's connect, which is the exact shape you will write for the real one:

use std::str::FromStr;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::SqlitePool;

async fn connect(url: &str) -> anyhow::Result<SqlitePool> {
    let opts = SqliteConnectOptions::from_str(url)?
        .create_if_missing(true);        // make the file if it is not there yet
    let pool = SqlitePoolOptions::new()
        .max_connections(1)              // one writer; keeps :memory: alive
        .connect_with(opts)
        .await?;
    sqlx::migrate!().run(&pool).await?;  // create the tables (next section)
    Ok(pool)
}

Two of those lines are load-bearing, and both have a why worth holding onto:

• create_if_missing(true) — by default, opening sqlite://data.db on a path that does not exist is an error. This flag says "if the file is not there, create it." Without it, the very first run of a fresh checkout fails because the database file does not exist yet.

• max_connections(1) — this looks like a performance knob and is really a correctness one, for two reasons. First, SQLite is a single-writer database: only one connection may write at a time, so a larger pool buys you contention, not throughput. Second, and this is the subtle one, an in-memory database (sqlite::memory:) is owned by its connection — when that connection closes, the database and everything in it vanishes. A pool of one keeps exactly one connection alive for the pool's whole life, so the in-memory database stays alive too. A pool of five in-memory connections would be five different, empty databases. Pin it to one.

Migrations: how the tables come to exist

A fresh database has no tables. Migrations are the ordered SQL scripts that build the schema up from nothing. sqlx's migrate!() macro embeds every .sql file from a migrations/ directory into your binary at compile time, and .run(&pool) executes any that have not been applied yet, tracking which ran in a bookkeeping table it manages for you.

For the toy, migrations/0001_init.sql is one statement:

CREATE TABLE notes (
    id         TEXT PRIMARY KEY,
    body       TEXT NOT NULL,
    created_at TEXT NOT NULL
);

The file name matters: sqlx applies migrations in lexical order, so the numeric prefix (0001_, 0002_, …) is the version sequence. Run migrate!() against a fresh database and this table appears; run it again and sqlx sees the migration is already applied and does nothing. That idempotence is why calling connect at the start of every test is safe — the second, hundredth, thousandth call all converge on the same schema.

Note also that migrate!() embeds the SQL at compile time from a fixed directory — which is a different, and cheaper, use of a macro than query!. It reads local files; it needs no live database. That is why we can keep migrations as a macro while rejecting the query macros.

The whole toy, working

Here is notes end to end — connect, insert, get, and a miss — the same three-method shape (connect/in_memory, insert, get) the real store will have. It uses the runtime API throughout: SQL as strings, ? placeholders filled by .bind(...), columns pulled out by name with try_get.

use std::str::FromStr;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow};
use sqlx::{Row, SqlitePool};

#[derive(Debug, PartialEq, Eq)]
struct Note {
    id: String,
    body: String,
    created_at: String,
}

#[derive(Clone)]
struct NoteStore {
    pool: SqlitePool,
}

impl NoteStore {
    async fn connect(url: &str) -> anyhow::Result<Self> {
        let opts = SqliteConnectOptions::from_str(url)?.create_if_missing(true);
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(opts)
            .await?;
        sqlx::migrate!().run(&pool).await?;
        Ok(Self { pool })
    }

    // One fresh, private in-memory database — the per-test entry point.
    async fn in_memory() -> anyhow::Result<Self> {
        Self::connect("sqlite::memory:").await
    }

    async fn insert_note(&self, note: &Note) -> anyhow::Result<()> {
        sqlx::query("INSERT INTO notes (id, body, created_at) VALUES (?, ?, ?)")
            .bind(&note.id)
            .bind(&note.body)
            .bind(&note.created_at)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    async fn get_note(&self, id: &str) -> anyhow::Result<Option<Note>> {
        let row = sqlx::query("SELECT * FROM notes WHERE id = ?")
            .bind(id)
            .fetch_optional(&self.pool)   // Option: the row may not exist
            .await?;
        Ok(row.map(row_to_note).transpose()?)
    }
}

// Turn a raw row into our type. This hand-mapping is the price of the
// runtime API — the macros would generate it, but would need DATABASE_URL.
fn row_to_note(row: SqliteRow) -> anyhow::Result<Note> {
    Ok(Note {
        id: row.try_get("id")?,
        body: row.try_get("body")?,
        created_at: row.try_get("created_at")?,
    })
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let store = NoteStore::in_memory().await?;
    let note = Note {
        id: "n1".into(),
        body: "buy milk".into(),
        created_at: "2026-07-21T10:00:00Z".into(),
    };
    store.insert_note(&note).await?;

    let fetched = store.get_note("n1").await?;
    println!("fetched: {fetched:?}");
    println!("round-trips equal: {}", fetched.as_ref() == Some(&note));

    let missing = store.get_note("nope").await?;
    println!("missing id -> {missing:?}");
    Ok(())
}

Run it and you see the round trip, plus the deliberate miss:

fetched: Some(Note { id: "n1", body: "buy milk", created_at: "2026-07-21T10:00:00Z" })
round-trips equal: true
missing id -> None

Three details to lock, because each recurs in the build:

• .bind(...) for every ?. You never format values into the SQL string yourself. The placeholders keep the value and the query text separate — which is both how sqlx knows the value's type and why SQL injection is a non-issue: a bound value can never be read as SQL.
• fetch_optional returns Option<Row>. A lookup by id might find nothing; the type says so. That is why get_note returns Option<Note> and the miss prints None. (Its siblings: fetch_one when a row must exist, fetch_all for many.)
• try_get("column") pulls a typed value out by name. This is the manual mapping the runtime API costs you — the exact code the query_as! macro would have generated, written by hand instead, in exchange for not needing a build-time database.

These examples use sqlx, which is not on the Rust playground, so there is no play button. To run them yourself: cargo new notes-scratch, add sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "macros", "migrate"] }, tokio = { version = "1", features = ["full"] }, and anyhow; put the CREATE TABLE notes script in migrations/0001_init.sql; paste the code into main.rs. That is the same crate set — and the same runtime-API style — the store crate declares.

Why this is the right foundation for the arc

The next chapter builds the real store, and the mapping is one-for-one:

• notes table ↔ runs table (plus a jobs table alongside it)
• NoteStore::connect / in_memory ↔ Store::connect / in_memory — identical, down to create_if_missing(true) and max_connections(1)
• insert_note ↔ insert_run (bind fields, execute)
• get_note ↔ get_run (fetch_optional, try_get, return Option)
• migrations/0001_init.sql (one CREATE TABLE) ↔ migrations/0001_init.sql (two tables plus indexes)

The domain changes from grocery notes to eval runs; the shape does not move. Get the shape solid here and the store chapter is mostly translation. What it adds — the atomic claim and the transactional record — is the concept two chapters from now, and it is where SQL stops being a filing cabinet and starts being the thing that keeps two schedulers from stepping on each other.

Questions to lock

Genuinely pause on each. If one is fuzzy, that is the signal to re-read.

1. Why does this course use the runtime query API instead of the compile-time query! macros — and what concretely would break for someone running cargo test if we used the macros?
2. What are create_if_missing(true) and max_connections(1) each protecting against? (For the second one, why does an in-memory database make it correctness, not just performance?)
3. What does migrate!() do, why is calling it at the start of every test safe, and why can it stay a macro when query! cannot?

Next chapter is the first build of this arc: the runs and jobs schema, and Store::connect/in_memory/insert_run/get_run/insert_jobs.

Build: The Store — runs, jobs, insert and get

Maps to: Phase 2 (control-store). Kind: Build.

Objective

Stand up control-store, the crate that gives a run somewhere durable to live. In this chapter you write the schema — the runs and jobs tables plus their indexes — and the four methods that get data in and out without any concurrency yet: Store::connect, Store::in_memory, insert_run, get_run, and insert_jobs. One test, insert_then_get_run, proves a Run survives a round trip through SQLite unchanged. The atomic claim and transactional recording — the reason this crate exists — are the next build; here we lay the table they act on.

Everything here is the concept-sqlx toy with the domain swapped in: notes becomes runs (with jobs alongside), NoteStore becomes Store, and the connect/insert/get shape does not move. If the toy compiled and ran for you, this chapter is mostly translation — the only genuinely new thing is a second table and a couple of JSON-encoded columns.

Scaffold

Create (new crate — add crates/control-store to the workspace members):

• crates/control-store/Cargo.toml
  • [dependencies]: control-core = { path = "../control-core" } (the Run/Job types and ControlError live there), plus sqlx, serde_json, chrono, uuid — all { workspace = true }.
    • sqlx (features ["runtime-tokio", "sqlite", "macros", "migrate"]) — the async SQL toolkit. macros is for migrate!() (which embeds local .sql files, needing no build-time database); migrate pulls in the migrator that runs them. We do not use the query! macros, so there is no DATABASE_URL at build time — that decision was the whole first chapter.
    • serde_json — three columns (models, spec, outcome) hold JSON that serde encodes and decodes; the database sees a TEXT string, your Rust sees a Vec/struct.
    • chrono — Run::created_at is a DateTime<Utc>, stored as an RFC 3339 string.
    • uuid — RunId/JobId wrap a Uuid, stored as its string form.
  • [dev-dependencies]: tokio (features ["macros", "rt-multi-thread"] — the tests are #[tokio::test]), pretty_assertions — both { workspace = true }.
• crates/control-store/migrations/0001_init.sql — the schema (given below). migrate!() reads this directory at compile time.
• crates/control-store/src/lib.rs — the Store struct (a single pool: SqlitePool field), the five methods, a private row_to_run mapper, and the test module.

Expected result: cargo test -p control-store → 1 test passes: insert_then_get_run.

The spec (givens)

The schema — migrations/0001_init.sql

Two tables and two indexes. Every column is TEXT or INTEGER — SQLite has no native UUID, timestamp, boolean, or array type, so ids and timestamps are stringified and collections are JSON in a TEXT column. The status columns are plain TEXT holding the snake_case string your Part II enums already serialize to.

CREATE TABLE runs (
    id         TEXT    PRIMARY KEY,
    status     TEXT    NOT NULL,
    created_at TEXT    NOT NULL,
    manifest   TEXT    NOT NULL,
    models     TEXT    NOT NULL,           -- JSON array of model names
    epochs     INTEGER NOT NULL,
    job_count  INTEGER NOT NULL DEFAULT 0,
    done_count INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE jobs (
    id      TEXT    PRIMARY KEY,
    run_id  TEXT    NOT NULL REFERENCES runs (id),
    status  TEXT    NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 0,
    spec    TEXT    NOT NULL,              -- JSON EvalJob
    outcome TEXT                           -- JSON JobOutcome, NULL until done
);

CREATE INDEX idx_runs_status ON runs (status, created_at);
CREATE INDEX idx_jobs_run ON jobs (run_id);

Four choices are load-bearing, and each pays off in the next chapter — note them now so the claim SQL reads naturally later:

• job_count / done_count on runs. The run carries its own progress counters. "How far along is this run?" is a single-row read, not a COUNT(*) over jobs. The transactional record bumps done_count; when it reaches job_count, the run is finished.
• outcome is nullable. A job with outcome IS NULL has not reported back yet. Everything else is NOT NULL; this one column is deliberately nullable because "no result yet" is a real, expected state.
• idx_runs_status ON (status, created_at). The claim query filters WHERE status = 'queued' and orders by created_at. This composite index is exactly that access pattern — the scheduler asks "oldest queued run?" on a hot loop, and the index makes it a cheap seek rather than a table scan.
• jobs.run_id REFERENCES runs (id). A job belongs to a run; the foreign key documents that, and idx_jobs_run makes "all jobs for this run" fast — which run_results will need. (SQLite does not enforce foreign keys unless a pragma is set, so treat this as documentation plus an index, not a guardrail.)

[→ Answer key](../appendix-answer-key.md#store-schema)

The runtime-query decision, restated in code

Every method here uses the runtime query API — sqlx::query("..."), values supplied with .bind(...), columns pulled out by name with try_get. No query!, no DATABASE_URL, no build-time database. The trade you accepted in the concept chapter is now concrete: a SQL typo surfaces when insert_then_get_run runs, not when the crate compiles. That is why the test exists.

connect is the toy's connect/in_memory, unchanged down to the two load-bearing options:

• SqliteConnectOptions::from_str(url)?.create_if_missing(true) — create the file on first run rather than erroring.
• SqlitePoolOptions::new().max_connections(1) — one writer (SQLite's model), and the one connection that keeps a sqlite::memory: database alive for the pool's life.
• sqlx::migrate!().run(&pool).await? — apply 0001_init.sql; idempotent, so calling in_memory() at the top of every test rebuilds the same fresh schema.

insert_run binds every column in order (RunId, RunStatus, and DateTime each stringified; models via serde_json::to_string) and .executes it. get_run is SELECT * FROM runs WHERE id = ? with .fetch_optional (a lookup by id may miss → Option<Run>), mapped back through row_to_run. insert_jobs inserts a slice of Jobs — and here is the first transaction in the codebase: wrap the inserts in pool.begin() … tx.commit() so a run's jobs land all-or-nothing rather than half a fan-out surviving a mid-loop failure.

[→ Answer key](../appendix-answer-key.md#store-crud)

Concepts exercised

• Modelling a Rust domain in SQLite's storage classes: UUIDs and timestamps as TEXT, Vec/structs as JSON-in-TEXT, u32 as INTEGER read back through i64.
• The runtime query API end to end: query + bind + execute for writes, fetch_optional + try_get for reads.
• A nullable column (outcome) encoding an expected "not yet" state.
• A composite index (status, created_at) shaped to a query you have not written yet.
• The first transaction: begin/commit around a batch insert for all-or-nothing.
• Denormalized progress counters (job_count/done_count) on the parent row.

The build loop (you drive)

Write the test first, predict its failure mode, run, implement to green.

Test — insert_then_get_run (in lib.rs, #[tokio::test])

1. Write the failing test. Build a Store::in_memory().await and a small Run fixture — a helper a_run(job_count: u32) -> Run with a fresh RunId::new(), RunStatus::Queued, Utc::now(), one model, and done_count: 0 is worth writing now, because the next chapter reuses it. insert_run(&run).await, then assert get_run(run.id).await.unwrap() == Some(run).
2. Predict: the round trip encodes created_at with to_rfc3339() and parses it back with parse_from_rfc3339. If you instead stored it via to_string() (a different textual form), would insert fail, or would get fail to parse it back? Name which side of the round trip a format mismatch surfaces on.
3. Run — it fails to compile (Store and its methods do not exist yet).
4. Implement the schema file, then connect/in_memory, insert_run, get_run, and row_to_run. What each does is specified above and pinned by the test; how you arrange the bind order and the mapper is yours. Build insert_jobs here too — the next chapter's tests need jobs to exist, and a clean compile is your only check on it until then.
5. Run green, commit.

Done when

cargo test -p control-store shows insert_then_get_run green, a Run round-trips through SQLite unchanged, and you can point to the two lines in connect that (a) create the file if missing and (b) keep the in-memory database alive. The schema is in place and the counters are sitting at zero — waiting for the claim and the record, which is the next chapter.

Concept: The Atomic Claim and Transactions

Kind: Concept. Builds on the store you just wrote — no new crate, one new idea (two, really, and they are cousins).

The last build gave a run a durable home. It did not give it a safe one. Every method so far — insert_run, get_run, insert_jobs — assumed exactly one actor touching the database at a time, and for a single-process coordinator that assumption holds right up until the moment it doesn't: the moment there are two schedulers, or one scheduler that restarted while the old one was still draining, or two worker results landing at the same instant. This chapter is about the two SQL tools that make those moments safe. Both are one idea wearing two hats — this happens as one indivisible step, or it does not happen at all.

We build them on a toy tasks queue so nothing here is the answer key, then map it one-for-one onto the real store in the next chapter.

Why this matters the instant there is more than one worker

Picture the store as it stands, and two schedulers both looking for work. Each wants "the oldest queued run." With the tools you have, the honest way to grab one is two steps:

scheduler A                         scheduler B
-----------                         -----------
SELECT id FROM runs
  WHERE status='queued'
  ORDER BY created_at LIMIT 1   ->  run r7
                                    SELECT id FROM runs
                                      WHERE status='queued'
                                      ORDER BY created_at LIMIT 1  -> run r7
UPDATE runs SET status='running'
  WHERE id = r7
                                    UPDATE runs SET status='running'
                                      WHERE id = r7

Both read r7 as queued — the read happened before either write — so both claim r7 and run it. Every vignette in that run gets evaluated twice, tokens spent twice, and the run's counters are now a lie. Nothing here is a bug in either scheduler; each did the obvious correct thing. The bug is the gap between the read and the write, a gap another actor slipped into. That gap is called a race, and it is the defining hazard of the whole distributed half of this course.

The atomic claim: UPDATE … RETURNING

The race exists because "find the oldest queued run" and "mark it running" were two statements. SQL lets you fuse them. A single UPDATE can do the selecting and the writing, and RETURNING hands you back the row it just changed:

UPDATE tasks SET status = 'running'
WHERE id = (SELECT id FROM tasks WHERE status = 'queued' ORDER BY created_at LIMIT 1)
RETURNING *;

Read it as one motion: among the queued tasks, take the oldest, flip it to running, and give it back to me. The subquery picks the target; the UPDATE writes it; RETURNING * yields the post-update row so the caller learns which task it got — all inside one statement the database runs to completion before any other connection sees the table. There is no gap for a second scheduler to read the same row as still queued, because by the time anyone else looks, it is already running.

Two properties fall out, and both are the point:

• No double-claim. Because the flip is part of the same statement as the selection, a concurrent claim either runs before this one (and picks a different oldest row) or after (and sees this row as no longer queued). It can never land in the middle. Two schedulers hammering claim_next divide the queue between them; they never split a row.
• The empty queue answers honestly. When nothing is queued, the subquery returns no id, the UPDATE matches no row, and RETURNING yields nothing — which sqlx surfaces as fetch_optional returning None. "Nothing to claim" and "here is your claim" come back through the same call, distinguished by Some/None.

Transactions: all-or-nothing across several statements

The claim fuses a read and a write. The other hazard is the opposite shape: several writes that must all land together. When a job finishes, the store does two things — mark the job done, and bump its run's done_count. Do them as two independent statements and a crash (or a rolled-back connection, or a failed second statement) between them leaves the database in a state that can never be true: a job marked done that the run's counter never counted, or a counter that ran ahead of a job still marked pending. The run's progress is now wrong forever.

A transaction is the fix. You open one with pool.begin(), run as many statements as you like against the returned handle, and finish with commit(). Until you commit, none of it is visible to anyone else; and if you never commit — you call rollback(), or the connection drops, or the process dies — every statement in the transaction is undone as a unit. All of them, or none of them. There is no half.

let mut tx = pool.begin().await?;
sqlx::query("UPDATE tasks SET status = 'done' WHERE id = ?")
    .bind("t1")
    .execute(&mut *tx)            // note: &mut *tx, not &pool
    .await?;
sqlx::query("UPDATE counters SET value = value + 1 WHERE name = 'done'")
    .execute(&mut *tx)
    .await?;
tx.commit().await?;              // both land now, atomically — or neither did

The one syntactic thing to notice: inside a transaction you execute against &mut *tx, not &pool. That is how sqlx routes both statements down the same connection and holds them open as one unit; hand either statement &pool and it would run on its own connection, outside the transaction, defeating the whole point. (You saw this already in insert_jobs last chapter — the batch insert was a transaction so a run's jobs land all-or-nothing. Same tool, now load-bearing.)

The whole toy, working

Here is tasks end to end: two queued rows, the atomic claim called three times, a committed two-statement transaction, and the same transaction rolled back to show all-or-nothing from the other side. It is the exact shape the real store will have — claim_next is claim_next_run, the commit is the happy path of record_job_outcome, the rollback is what a crash mid-record does for you.

use std::str::FromStr;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow};
use sqlx::{Row, SqlitePool};

#[derive(Debug)]
struct Task { id: String, status: String, created_at: String }

fn row_to_task(row: SqliteRow) -> anyhow::Result<Task> {
    Ok(Task {
        id: row.try_get("id")?,
        status: row.try_get("status")?,
        created_at: row.try_get("created_at")?,
    })
}

async fn connect() -> anyhow::Result<SqlitePool> {
    let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.create_if_missing(true);
    let pool = SqlitePoolOptions::new().max_connections(1).connect_with(opts).await?;
    sqlx::migrate!().run(&pool).await?;
    Ok(pool)
}

// The atomic claim: select the oldest queued row and flip it to running in one
// statement. RETURNING hands back the row it just changed — or nothing.
async fn claim_next(pool: &SqlitePool) -> anyhow::Result<Option<Task>> {
    let row = sqlx::query(
        "UPDATE tasks SET status = 'running' \
         WHERE id = (SELECT id FROM tasks WHERE status = 'queued' ORDER BY created_at LIMIT 1) \
         RETURNING *",
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.map(row_to_task).transpose()?)
}

async fn status(pool: &SqlitePool, id: &str) -> anyhow::Result<String> {
    let row = sqlx::query("SELECT status FROM tasks WHERE id = ?").bind(id)
        .fetch_one(pool).await?;
    Ok(row.try_get("status")?)
}
async fn counter(pool: &SqlitePool) -> anyhow::Result<i64> {
    let row = sqlx::query("SELECT value FROM counters WHERE name = 'done'")
        .fetch_one(pool).await?;
    Ok(row.try_get("value")?)
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pool = connect().await?;
    // Seed: two queued tasks (t1 older than t2) and a counter at zero.
    for (id, at) in [("t1", "2026-07-21T10:00:00Z"), ("t2", "2026-07-21T10:01:00Z")] {
        sqlx::query("INSERT INTO tasks (id, status, created_at) VALUES (?, 'queued', ?)")
            .bind(id).bind(at).execute(&pool).await?;
    }
    sqlx::query("INSERT INTO counters (name, value) VALUES ('done', 0)")
        .execute(&pool).await?;

    // --- The atomic claim, three times over a two-item queue ---
    println!("claim 1 -> {:?}", claim_next(&pool).await?.map(|t| t.id));
    println!("claim 2 -> {:?}", claim_next(&pool).await?.map(|t| t.id));
    println!("claim 3 -> {:?}", claim_next(&pool).await?.map(|t| t.id));

    // --- Two statements, committed together ---
    let mut tx = pool.begin().await?;
    sqlx::query("UPDATE tasks SET status = 'done' WHERE id = ?").bind("t1")
        .execute(&mut *tx).await?;
    sqlx::query("UPDATE counters SET value = value + 1 WHERE name = 'done'")
        .execute(&mut *tx).await?;
    tx.commit().await?;
    println!("after commit:   t1={}, counter={}", status(&pool, "t1").await?, counter(&pool).await?);

    // --- The same two statements, rolled back (what a crash mid-record does) ---
    let mut tx = pool.begin().await?;
    sqlx::query("UPDATE tasks SET status = 'done' WHERE id = ?").bind("t2")
        .execute(&mut *tx).await?;
    sqlx::query("UPDATE counters SET value = value + 1 WHERE name = 'done'")
        .execute(&mut *tx).await?;
    tx.rollback().await?;
    println!("after rollback: t2={}, counter={}", status(&pool, "t2").await?, counter(&pool).await?);
    Ok(())
}

The migration is two small tables:

CREATE TABLE tasks (
    id         TEXT PRIMARY KEY,
    status     TEXT NOT NULL,   -- 'queued' | 'running' | 'done'
    created_at TEXT NOT NULL
);
CREATE TABLE counters (
    name  TEXT PRIMARY KEY,
    value INTEGER NOT NULL
);

Run it, and the two ideas print their signatures:

claim 1 -> Some("t1")
claim 2 -> Some("t2")
claim 3 -> None
after commit:   t1=done, counter=1
after rollback: t2=running, counter=1

Read every line against the concept:

• claim 1 -> Some("t1"), claim 2 -> Some("t2"), claim 3 -> None. Each claim flips its target to running inside the one statement, so the next claim never sees it as queued. The queue drains, oldest first; the empty queue answers None. Two schedulers alternating these calls would split the two tasks between them and never collide — that is the atomic claim earning its name.
• after commit: t1=done, counter=1. Both statements landed together. The job is done and the count moved — the pair that must always agree, agreeing.
• after rollback: t2=running, counter=1. This is all-or-nothing from the failure side. The transaction set t2 to done and bumped the counter, then rolled back — and the database undid both. t2 is back to running (its pre-transaction state, since it had been claimed) and the counter is untouched at 1. A crash between the two writes would have done exactly this for you: no orphaned half-write, no counter running ahead of reality.

This toy uses sqlx, so there is no playground button. To run it: cargo new claim-scratch, add sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "macros", "migrate"] }, tokio = { version = "1", features = ["full"] }, and anyhow; put the two CREATE TABLEs in migrations/0001_init.sql; paste the code into main.rs. Same crate set and same runtime-API style as the store crate.

The seam to the build: one-for-one

The next chapter builds the two real store methods, and the toy maps straight across:

• toy tasks queue ↔ the runs table with its status/created_at columns (and the idx_runs_status index you already built for exactly this claim)
• claim_next (the UPDATE … RETURNING) ↔ claim_next_run — same statement, tasks becomes runs, Task becomes Run
• the committed two-statement transaction (mark task done + bump counter) ↔ record_job_outcome — mark the job done, then bump the run's done_count, in one begin/commit
• the rollback demo ↔ what a crash mid-record does: the job stays pending and the counter stays put, so redelivery can safely try again

The domain changes from anonymous tasks to eval runs and jobs; the two moves — fuse the read and write into one statement, fuse the several writes into one transaction — do not. The real record_job_outcome adds one twist the toy omits: a guard that makes recording the same job twice count once, so at-least-once redelivery is safe. That guard is the next build's centerpiece, and why it has to exist is the payoff waiting in Part VIII.

Questions to lock

Pause on each. A fuzzy answer is the signal to re-read the matching section.

1. Two schedulers each run SELECT oldest queued then UPDATE it to running as separate statements. Walk the interleaving that makes them both claim the same run. Then explain precisely how folding both into one UPDATE … RETURNING closes the gap.
2. Inside a transaction you .execute(&mut *tx) instead of .execute(&pool). What goes wrong if you accidentally pass &pool for one of the two statements in record_job_outcome?
3. record_job_outcome marks a job done and bumps its run's done_count in one transaction. If the process crashes between those two writes, what state is the database left in — and why is that the state that makes redelivery safe rather than corrupt?

Next chapter is the build: claim_next_run and record_job_outcome, the transactional-and-idempotent record that is the reason this whole crate exists.

Build: claim_next_run + Transactional Recording

Maps to: Phase 2 (control-store). Kind: Build.

Objective

Add the two methods this crate exists for: claim_next_run, the atomic claim that lets any number of schedulers pull work without ever grabbing the same run twice, and record_job_outcome, the transactional-and-idempotent record that advances a run's progress exactly once per job even when a result is delivered twice. Three tests pin the behavior — the claim drains the queue and then answers None, a full run's outcomes advance done_count and flip the run to done on the last job, and the same job recorded twice counts once. When these are green, the store is safe under concurrency and under redelivery, which is everything the scheduler and cluster arcs will lean on.

The atomic claim and the two-statement transaction are the concept toy, translated: tasks becomes runs, claim_next becomes claim_next_run, the committed transfer becomes the outcome record. The one genuinely new piece is the idempotency guard on the record — a single AND status != 'done' clause, plus a check on how many rows it touched — and it is worth understanding before you type it.

Scaffold

Add to crates/control-store/src/lib.rs (no new files, no new deps): two methods on Store — claim_next_run and record_job_outcome — and three tests. You will also lean on insert_jobs from the last chapter, which now gets its first real exercise.

Expected result: cargo test -p control-store → 4 tests pass (the previous insert_then_get_run plus the three below):

• claim_moves_run_to_running_then_none
• record_outcome_advances_done_count_and_finishes_run
• recording_the_same_job_twice_counts_once

The spec (givens)

claim_next_run — the atomic claim

Signature: async fn claim_next_run(&self) -> Result<Option<Run>, ControlError>. One statement, run with fetch_optional, mapped through the same row_to_run you already wrote:

UPDATE runs SET status = 'running'
WHERE id = (SELECT id FROM runs WHERE status = 'queued' ORDER BY created_at LIMIT 1)
RETURNING *;

The subquery picks the oldest queued run (the idx_runs_status index makes this a seek); the UPDATE flips it to running; RETURNING * yields the post-update row so the caller gets the Run it claimed. No ? placeholders here — there is nothing to bind. fetch_optional gives None when the queue is empty (the UPDATE matched no row), which is how "nothing to claim" comes back through the same call as a successful claim. This is the concept toy's claim_next with tasks → runs.

[→ Answer key](../appendix-answer-key.md#store-claim)

record_job_outcome — transactional and idempotent

Signature: async fn record_job_outcome(&self, run_id: RunId, outcome: &JobOutcome) -> Result<(), ControlError>. One transaction, two statements — but the second runs only conditionally. The full logic:

-- statement 1: mark the job done, but only if it is not already done
UPDATE jobs SET status = 'done', outcome = ?
WHERE id = ? AND status != 'done';

-- statement 2: run this ONLY when statement 1 changed exactly one row
UPDATE runs SET done_count = done_count + 1,
    status = CASE WHEN done_count + 1 >= job_count THEN 'done' ELSE status END
WHERE id = ?;

Wrap both in pool.begin() … tx.commit(), executing each against &mut *tx. Between the two, branch on res.rows_affected() from statement 1: run statement 2 only if rows_affected() == 1. Bind the JSON-encoded outcome and outcome.job_id into statement 1, and run_id into statement 2.

Three clauses carry the whole method, and each is deliberate:

• AND status != 'done' — the guard. A job already marked done is not matched by statement 1, so the UPDATE changes zero rows. A first-time job is matched, so it changes one. This one clause is the entire difference between "recorded once" and "recorded twice."
• if res.rows_affected() == 1 — the gate. rows_affected() reports how many rows statement 1 actually changed: 1 the first time a job lands, 0 on any redelivery of an already-done job. The run's done_count advances only when a job newly transitions to done. A redelivered outcome rewrites the (identical) outcome JSON on... nothing — it matched no row — and leaves done_count exactly where it was.
• CASE WHEN done_count + 1 >= job_count THEN 'done' ELSE status END — the finish. In the same statement that bumps the counter, check whether this landing is the last one; if the incremented count reaches job_count, flip the run to done. (done_count + 1 because the arithmetic in SET reads the old value; the whole SET sees the pre-update row.) No separate "is the run finished?" query, no second round trip — the completion check rides along with the increment, inside the transaction.

[→ Answer key](../appendix-answer-key.md#store-record)

Why idempotency, in one paragraph (full story: Part VIII)

The guard exists because the coordinator will deliver jobs at-least-once: if a worker finishes a job but dies before its Result is acknowledged, the scheduler cannot tell "finished but unacked" from "never finished," so it re-dispatches — and the job genuinely runs twice, or its outcome arrives twice. At-least-once is the only delivery guarantee that loses nothing under worker failure, but it makes duplicate recording a normal event rather than a bug. Idempotent recording is the escape: written this way, recording the same job's outcome twice is indistinguishable from recording it once, so at-least-once delivery becomes safe. The full argument — heartbeats, reaping a dead worker's lease, why exactly-once is a fiction — is the capstone of Part VIII; here you build the store-side half that makes it all sound, and prove it with recording_the_same_job_twice_counts_once.

Concepts exercised

• The atomic claim (UPDATE … RETURNING) as concurrency-safe work pulling — the concept toy, on runs.
• A multi-statement transaction whose second statement is conditional on the first's rows_affected().
• Idempotency by a WHERE … AND status != 'done' guard plus a rows_affected() == 1 gate.
• A CASE expression folding a completion check into the same UPDATE that advances the counter.
• Reading rows_affected() to distinguish "newly changed" from "already in that state."

The build loop (you drive)

Write each test first, predict, run red, implement to green.

Test 1 — claim_moves_run_to_running_then_none (#[tokio::test])

1. Write it. Store::in_memory(), insert one a_run(0) (queued). Call claim_next_run().await.unwrap().unwrap(); assert the claimed run's id matches and its status == RunStatus::Running. Then call claim_next_run() again and assert it is_none().
2. Predict: the second claim runs the identical statement against the identical table — nothing was inserted or deleted between the calls. Why does it return None rather than re-claiming the same run? Point at the exact word in the SQL that changed the answer.
3. Run red (no claim_next_run yet), implement the one statement, run green, commit.

Test 2 — record_outcome_advances_done_count_and_finishes_run (#[tokio::test])

1. Write it. Insert a_run(2) (two jobs), insert_jobs two jobs j1/j2 for it, and claim_next_run() to move the run to running (a run reports outcomes while it is running). Record j1's outcome; get_run and assert done_count == 1, status == Running. Record j2's outcome; get_run and assert done_count == 2, status == Done.
2. Predict: which record call flips the run to Done, and what makes the CASE fire on that call and not the first? (Compute done_count + 1 versus job_count for each.)
3. Run red, implement the transaction + conditional second statement, run green, commit.

Test 3 — recording_the_same_job_twice_counts_once (#[tokio::test])

1. Write it. Insert a_run(2), insert_jobs one job j1, claim_next_run(). Record j1's outcome, then record the same outcome again — same job_id. get_run and assert done_count == 1 (not 2) and status == Running.
2. Predict: on the second call, rows_affected() from statement 1 is what number — and therefore does statement 2 run at all? This is the at-least-once safety property in one assertion; name what would break without the guard.
3. Run red — and note that with a naive record (no guard) this test is the one that fails while tests 1 and 2 still pass, which is exactly why it exists. Implement the guard + gate, run green, commit.

Done when

cargo test -p control-store shows all four tests green; claim_next_run drains the queue and then answers None; record_job_outcome advances done_count once per job and finishes the run on the last one; and recording_the_same_job_twice_counts_once proves a redelivered outcome does not double-count. You can state, in one sentence, why the rows_affected() == 1 gate and the AND status != 'done' guard are a pair — and you know that the reason the coordinator needs this at all is the at-least-once delivery that Part VIII builds on top of it. Commit; the persistence arc is done.

Concept-Check: Persistence

This is where SQL stopped being a filing cabinet and started being the thing that keeps the coordinator honest under concurrency. A run now has a durable home; the atomic claim guarantees two schedulers never grab the same run; the transaction makes "job done" and "run counter moved" one indivisible fact; and the idempotency guard makes recording the same outcome twice count once — the store-side half of the at-least-once safety the cluster arc pays off. If these are solid, the service and scheduler arcs have a persistence layer they can trust when machines fail.

Concept: axum — Handlers, State, Extractors, IntoResponse

Kind: Concept. New crate: axum — this chapter shows it working before you build with it.

This is the Service arc. Everything you have built so far is a library: the domain types, the ModelClient seam, the store. None of it has a front door. This arc bolts one on — an HTTP API a client can POST a run to and GET a run's status back from — and the crate that does it is axum.

You already know async, tokio, serde, and reqwest cold from the first two courses and the arcs before this one. So this chapter is not about what an HTTP request is or how JSON serializes. It is about the four moving parts axum gives you and how they compose: a handler (an async fn), the extractors that feed it typed pieces of the request, the IntoResponse trait that turns your return value back into bytes on the wire, and the Router that wires paths to handlers. Learn those four and axum is a small crate; miss how they fit and the compiler errors read like hieroglyphics.

We do all of it on a toy notes API — POST /notes, GET /notes/:id, backed by an in-memory store — so nothing here can be mistaken for the answer key. It is the coordinator's runs API with the domain filed off.

What axum is, and the one idea under all of it

axum is an HTTP framework built on tokio and hyper. You describe your API as a set of routes, each pointing at a handler function, and axum runs the server that accepts connections, parses requests, calls the right handler, and writes the response back.

The one idea that makes axum feel different from frameworks in other languages is this: a handler is just an async fn, and its argument types and return type do the work. There is no Request req, Response res pair you reach into. Instead, each parameter is an extractor — a type that knows how to pull one piece out of the incoming request (the JSON body, a path segment, the shared state) — and the return type is anything that implements IntoResponse. axum uses the types in the signature to decide what to parse and how to reply. Your job is to pick the right types; axum does the plumbing.

That is why this chapter is mostly about four types, not four hundred lines of API. Get the types right and the handler bodies are trivial.

Handlers: an async fn that returns something responseful

The simplest handler takes nothing and returns a &'static str:

// Scratch deps: axum = "0.7", tokio = { version = "1", features = ["full"] }
async fn health() -> &'static str {
    "ok"
}

&'static str implements IntoResponse — axum turns it into a 200 OK with a text/plain body. So does String, StatusCode, Json<T>, a tuple like (StatusCode, Json<T>), and Result<T, E> when both T and E are responseful. The handler never touches a response object; it returns a value whose type says how to become one. Hold that — it is the whole trick, and the error mapping later is just one more IntoResponse impl.

The Router: paths in, handlers out

The Router maps a path and method to a handler:

use axum::routing::{get, post};
use axum::Router;

fn app() -> Router {
    Router::new()
        .route("/notes", post(create_note))
        .route("/notes/:id", get(get_note))
}

post(create_note) says "a POST to this path is handled by create_note." The :id in /notes/:id is a path parameter — a wildcard segment whose value a handler can extract. .route(...) chains, so the whole API is one expression. This is the exact shape the build's app() has, one route per endpoint.

Extractors: typed pieces of the request

An extractor is a parameter type that implements axum's FromRequestParts (or FromRequest) trait. You will use three, and they are the three the build uses:

• State<T> — hands the handler a clone of the shared application state (the store).
• Path<T> — pulls the wildcard path segment(s) and parses them into T.
• Json<T> — reads the request body and deserializes it into T with serde.

You destructure them right in the parameter list. Here is create_note, taking shared state and a JSON body:

use axum::extract::State;
use axum::Json;

async fn create_note(
    State(state): State<AppState>,
    Json(body): Json<CreateNote>,
) -> Result<Response, ApiError> {
    // `state` is the shared AppState; `body` is the parsed CreateNote.
    // ...
}

State(state) and Json(body) are pattern matches: the extractor type is State<AppState>, and State(state) binds the inner AppState to state. axum sees State<AppState> in the signature and injects the state; it sees Json<CreateNote> and deserializes the body. No manual parsing anywhere.

IntoResponse: your types, back on the wire

The return type is where the handler decides its reply. The rich shape is a tuple: (StatusCode, Json<serde_json::Value>) says "this status, with this JSON body." create_note returns a 201 Created with the new id:

use axum::http::StatusCode;
use serde_json::json;

Ok((StatusCode::CREATED, Json(json!({ "id": id }))).into_response())

And get_note returns Json<Note> directly — a 200 OK with the serialized note — or an error. Because the return type is Result<Json<Note>, ApiError>, a handler can ?-propagate: a None from the store becomes an ApiError, and axum turns that into a response. Which brings us to the piece that makes the whole build click.

The error type: one IntoResponse impl, one place the mapping lives

This is the theme of the arc, so slow down here. Your handlers all return Result<_, ApiError>. ApiError is a newtype wrapping your domain error, and it implements IntoResponse once — that single impl is the only place the domain's error taxonomy becomes an HTTP status code:

use axum::response::{IntoResponse, Response};
use axum::http::StatusCode;
use axum::Json;
use serde_json::json;

struct ApiError(NoteError);

// A `?` on a NoteError produces an ApiError for free.
impl From<NoteError> for ApiError {
    fn from(e: NoteError) -> Self {
        ApiError(e)
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let status = match &self.0 {
            NoteError::NotFound(_) => StatusCode::NOT_FOUND,   // 404
            NoteError::Invalid(_) => StatusCode::BAD_REQUEST,  // 400
        };
        (status, Json(json!({ "error": self.0.to_string() }))).into_response()
    }
}

Read what this buys. A handler that does state.store.get(&id).ok_or_else(|| NoteError::NotFound(...))? never mentions 404. The type carries the outcome; the match in into_response is the single dispatch table from error kind to status. Add a new error variant and there is exactly one place the compiler makes you decide its status. This is the same "make the type the source of truth" move from the error-taxonomy arc, now pointed at HTTP: the taxonomy the store speaks and the status codes the client sees are joined in one match, not scattered across every handler.

The whole toy, working

Here is the notes API end to end — state, the error mapping, two handlers, the router — plus a main that spawns it on a random port and hits it with reqwest, which is also how you will test it. Read it once top to bottom; every piece above is in here.

// Scratch deps: axum = "0.7", tokio = { version = "1", features = ["full"] },
//   serde = { version = "1", features = ["derive"] }, serde_json = "1",
//   reqwest = { version = "0.12", features = ["json"] }, uuid = { version = "1", features = ["v4"] }
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use uuid::Uuid;

#[derive(Debug)]
enum NoteError {
    NotFound(String),
    Invalid(String),
}
impl std::fmt::Display for NoteError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            NoteError::NotFound(s) => write!(f, "not found: {s}"),
            NoteError::Invalid(s) => write!(f, "invalid request: {s}"),
        }
    }
}

// The store is the source of truth. Cheap to clone (an Arc), shared per request.
#[derive(Clone, Default)]
struct NoteStore {
    inner: Arc<Mutex<HashMap<String, Note>>>,
}
#[derive(Clone, Serialize)]
struct Note {
    id: String,
    body: String,
    status: NoteStatus,
}
#[derive(Clone, Copy, Serialize)]
#[serde(rename_all = "snake_case")]
enum NoteStatus {
    Draft,
}
impl NoteStore {
    fn insert(&self, note: Note) {
        self.inner.lock().unwrap().insert(note.id.clone(), note);
    }
    fn get(&self, id: &str) -> Option<Note> {
        self.inner.lock().unwrap().get(id).cloned()
    }
}

#[derive(Clone)]
struct AppState {
    store: NoteStore,
}

struct ApiError(NoteError);
impl From<NoteError> for ApiError {
    fn from(e: NoteError) -> Self {
        ApiError(e)
    }
}
impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let status = match &self.0 {
            NoteError::NotFound(_) => StatusCode::NOT_FOUND,
            NoteError::Invalid(_) => StatusCode::BAD_REQUEST,
        };
        (status, Json(json!({ "error": self.0.to_string() }))).into_response()
    }
}

#[derive(Deserialize)]
struct CreateNote {
    body: String,
}

// State first, body-consuming Json last.
async fn create_note(
    State(state): State<AppState>,
    Json(body): Json<CreateNote>,
) -> Result<Response, ApiError> {
    if body.body.trim().is_empty() {
        return Err(NoteError::Invalid("body must not be empty".into()).into());
    }
    let note = Note {
        id: Uuid::new_v4().to_string(),
        body: body.body,
        status: NoteStatus::Draft,
    };
    let id = note.id.clone();
    state.store.insert(note);
    Ok((StatusCode::CREATED, Json(json!({ "id": id }))).into_response())
}

async fn get_note(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<Note>, ApiError> {
    state
        .store
        .get(&id)
        .map(Json)
        .ok_or_else(|| NoteError::NotFound(format!("note {id}")).into())
}

fn app(state: AppState) -> Router {
    Router::new()
        .route("/notes", post(create_note))
        .route("/notes/:id", get(get_note))
        .with_state(state)
}

#[tokio::main]
async fn main() {
    // Spawn the app on a random free port (port 0 = "OS, pick one").
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap(); // the real port the OS chose
    let app = app(AppState { store: NoteStore::default() });
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });

    let base = format!("http://{addr}");
    let client = reqwest::Client::new();

    let resp = client.post(format!("{base}/notes"))
        .json(&json!({ "body": "buy milk" })).send().await.unwrap();
    println!("POST /notes status: {}", resp.status().as_u16());
    let id = resp.json::<serde_json::Value>().await.unwrap()["id"].as_str().unwrap().to_string();

    let resp = client.get(format!("{base}/notes/{id}")).send().await.unwrap();
    println!("GET /notes/:id status: {}", resp.status().as_u16());
    let note: serde_json::Value = resp.json().await.unwrap();
    println!("body: {} status: {}", note["body"], note["status"]);

    let resp = client.post(format!("{base}/notes"))
        .json(&json!({ "body": "" })).send().await.unwrap();
    println!("POST empty body status: {}", resp.status().as_u16());

    let resp = client.get(format!("{base}/notes/{}", Uuid::new_v4())).send().await.unwrap();
    println!("GET missing status: {}", resp.status().as_u16());
    println!("error body: {}", resp.json::<serde_json::Value>().await.unwrap()["error"]);
}

Run it and every part reports in:

POST /notes status: 201
GET /notes/:id status: 200
body: "buy milk" status: "draft"
POST empty body status: 400
GET missing status: 404
error body: "not found: note 475a5139-ccf8-426b-a6d0-fd7ea2f51496"

(The UUID is random, so your run prints a different one.) Read the last four lines against the IntoResponse impl: the empty body hit the Invalid branch → 400; the missing id hit NotFound → 404, with the body coming from NoteError's Display. The status codes were never typed into a handler — the one match decided all of them.

Testing: spawn on port 0, hit it with reqwest

Look again at that main — it is the test harness. This is the pattern the build uses for every test, and it is worth naming because it is the whole reason the API is testable without mocks:

1. Bind to 127.0.0.1:0. Port 0 tells the OS "give me any free port." You read the actual port back with listener.local_addr(). This means tests never collide on a fixed port and never need cleanup — each test gets its own ephemeral port.
2. tokio::spawn the server so it runs in the background while the test body drives it. The server task and the client run concurrently on the same runtime.
3. Hit it with a real reqwest client against http://{addr}. The request travels the genuine HTTP path — routing, extraction, your handler, IntoResponse — exactly as production would. Nothing is stubbed.

Because the state is injected via AppState, a test can seed the store before spawning and then assert what the API returns. The build's helper is a three-line spawn() that returns (base_url, store) so a test can do exactly that.

The trap: a body-consuming extractor that is not last

Here is the error that eats an afternoon the first time. The HTTP body is a stream you can read exactly once, so at most one extractor may consume it — Json, Form, Bytes, String — and it must be the last parameter. Every extractor before it must be a non-consuming one (State, Path, headers). Put Json first and watch:

// WRONG: Json consumes the body, so it cannot come before State.
async fn create_note(
    Json(body): Json<CreateNote>,
    State(state): State<AppState>,
) -> Result<Response, ApiError> { /* ... */ }

The compiler does not say "move Json last." It says the whole function is not a handler:

error[E0277]: the trait bound `fn(Json<CreateNote>, State<AppState>) -> ... {create_note}: Handler<_, _>` is not satisfied
   --> src/bin/trap.rs:24:31
    |
24  |         .route("/notes", post(create_note))
    |                          ---- ^^^^^^^^^^^ unsatisfied trait bound
    |                          |
    |                          required by a bound introduced by this call
    |
    = help: the trait `Handler<_, _>` is not implemented for fn item `fn(Json<CreateNote>, State<AppState>) -> ...`
    = note: Consider using `#[axum::debug_handler]` to improve the error message

That first line is the tell: when a handler "is not a Handler," suspect extractor order before anything else. The mechanical reason is that the body-consuming extractor implements FromRequest (which takes the whole request, body included) while the others implement FromRequestParts (which take only the head) — and axum's blanket Handler impls require every parameter but the last to be FromRequestParts. Move Json to the end and the impl is satisfied. Note the compiler's own hint: slap #[axum::debug_handler] on the function and the error turns from this vtable riddle into a plain-English sentence pointing at the offending argument — the first thing to reach for when a handler signature is rejected.

One-for-one: the toy ↔ the build

Everything above maps straight onto the coordinator API:

The domain changes from notes to eval runs; the four moving parts — handler, extractors, IntoResponse, Router — do not move. The build adds one branch to the error match (a catch-all _ => 500 for the store/worker errors) and one more route (/runs/:id/results), and that is the whole delta.

Questions to lock

1. A handler is async fn, its parameters are extractors, its return type implements IntoResponse. For create_run, name which extractor supplies the store, which supplies the request body, and why the body one must be the last parameter.
2. The ApiError IntoResponse impl is "the one place the taxonomy becomes a status code." What concretely goes wrong if, instead, each handler picked its own status code inline — and what does adding a new ControlError variant force you to do under the one-match design?
3. The test binds to 127.0.0.1:0 rather than a fixed port like :8080. What two problems does port 0 solve, and why does hitting the spawned server with reqwest test more than calling the handler function directly?

Next chapter is the build: the coordinator's Router, AppState, ApiError, and the create_run / get_run / get_results handlers — starting, as always, from a failing test.

Build: The Coordinator API

Maps to: Phase 3 (panoptes-control API). Kind: Build.

Objective

Give the coordinator a front door. Add the panoptes-control library target: an axum Router, an AppState carrying the store, an ApiError that maps the ControlError taxonomy to HTTP status codes in one place, and three handlers — create_run, get_run, get_results. By the end a client can POST a run, get a 201 with its id, and GET the run back as queued — all persisted through the store you built in Part IV, tested by spawning the real server on a random port and hitting it with reqwest.

Scaffold

Create (new crate — add crates/panoptes-control to the workspace members):

• crates/panoptes-control/Cargo.toml
  • [dependencies]: control-core = { path = "../control-core" } (the domain, ids, and ControlError taxonomy) and control-store = { path = "../control-store" } (the Store), plus axum, tokio, serde, serde_json, chrono, uuid — all { workspace = true }.
    • axum (0.7) is the HTTP framework: the Router, the extractors (State, Path, Json), and the IntoResponse trait all come from here.
    • serde_json supplies the json! macro for the response bodies ({ "id": … }, { "error": … }).
    • uuid (feature v4) is needed to parse a path id string back into a RunId and to reject a malformed one as 400.
  • [dev-dependencies]: reqwest (features ["json"]), pretty_assertions, tokio — all { workspace = true }. reqwest is the test client that drives the spawned server.
• crates/panoptes-control/src/lib.rs — the whole API surface: AppState, app(), ApiError, CreateRun, and the three handlers, plus the #[cfg(test)] module.

Dependencies this chapter exercises: axum (router, extractors, IntoResponse), control-store (the source of truth), control-core (the domain + error taxonomy), serde_json (the json! bodies), reqwest (dev — the test client hitting a spawned server).

Expected result: cargo test -p panoptes-control → 4 tests pass (post_run_returns_201_and_id, post_with_no_models_is_400, get_missing_run_is_404, get_run_after_post_returns_queued).

The spec (givens)

The Router, AppState, and app()

AppState holds the store and nothing else. It must be Clone — axum clones it into each request — and the store is cheap to clone (it wraps a connection pool behind an Arc).

#[derive(Clone)]
pub struct AppState {
    pub store: Store,
}

pub fn app(state: AppState) -> Router {
    Router::new()
        .route("/runs", post(create_run))
        .route("/runs/:id", get(get_run))
        .route("/runs/:id/results", get(get_results))
        .with_state(state)
}

Three routes, one handler each. :id is the path parameter get_run and get_results extract. .with_state(state) is what makes State<AppState> available to every handler; forget it and the router will not type-check against handlers that take State.

→ Answer key

CreateRun, validation, and create_run

The request body deserializes into a private CreateRun:

// Request body → POST /runs
{ "manifest": "vignettes.jsonl", "models": ["claude", "gpt-4"], "epochs": 3 }

#[derive(Deserialize)]
struct CreateRun {
    manifest: String,
    models: Vec<String>,
    epochs: u32,
}

create_run takes State(state): State<AppState> and Json(body): Json<CreateRun> — Json last, because it consumes the body (the concept chapter's trap). It validates, builds a Run, persists it, and returns 201:

• Validate. If body.models is empty, return ControlError::Invalid("at least one model is required"). If body.epochs == 0, return ControlError::Invalid("epochs must be >= 1"). Each becomes a 400 through ApiError.
• Build the Run. id: RunId::new(), status: RunStatus::Queued, created_at: Utc::now(), manifest from the body, and — the one computed field — job_count = body.models.len() as u32 * body.epochs (one job per model × epoch). done_count: 0. Move models and epochs in from the body.
• Persist, then answer. state.store.insert_run(&run).await?, then return (StatusCode::CREATED, Json(json!({ "id": run.id.to_string() }))).into_response().

The response body is just the new id: { "id": "…uuid…" }. The ? on insert_run is where a store failure would turn into a 500 — through the same ApiError you are about to write.

→ Answer key

ApiError — the taxonomy → status mapping, in one place

ApiError wraps a ControlError and implements IntoResponse once. This match is the only place in the crate a ControlError becomes a status code:

ControlError::NotFound(_)  → 404 NOT_FOUND
ControlError::Invalid(_)   → 400 BAD_REQUEST
_ (Worker/Protocol/Store/Io) → 500 INTERNAL_SERVER_ERROR

Provide impl From<ControlError> for ApiError so handlers can ?-propagate a ControlError straight into an ApiError. The response body on every error is Json(json!({ "error": self.0.to_string() })) — the error's Display, which the taxonomy already gives you.

get_run and get_results

Both take State(state) and Path(id): Path<String>. The id arrives as a String; parse it into a RunId first, and a malformed id is a 400, not a 404:

fn parse_run_id(id: &str) -> Result<RunId, ApiError> {
    Uuid::from_str(id)
        .map(RunId)
        .map_err(|_| ControlError::Invalid(format!("bad run id {id:?}")).into())
}

• get_run returns Result<Json<Run>, ApiError>. Parse the id, state.store.get_run(run_id).await?, and map the Option: Some(run) → Json(run) (200), None → ControlError::NotFound(format!("run {id}")) (404).
• get_results parses the id, then checks the run exists first (get_run(...).await?.is_none() → NotFound) so a missing run is a 404 rather than a bare empty list, then returns Json(state.store.run_results(run_id).await?).

→ Answer key

Concepts exercised

• A handler as an async fn whose extractor parameters and IntoResponse return type do the parsing and replying.
• State<AppState> as injected shared state, wired by .with_state(...).
• Path<String> and Json<CreateRun> extraction — and Json placed last because it consumes the body.
• One IntoResponse impl as the single ControlError → status-code dispatch table.
• Testing the real request path by spawning the server on 127.0.0.1:0 and driving it with reqwest.

The build loop (you drive)

The test module needs a helper. Write it once; all four tests use it:

// Spawn the app on a random port; return its base URL and the store, so a
// test can seed state before hitting the API.
async fn spawn() -> (String, Store) {
    let store = Store::in_memory().await.unwrap();
    let app = app(AppState { store: store.clone() });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
    (format!("http://{addr}"), store)
}

Test 1 — post_run_returns_201_and_id (#[tokio::test])

1. Write the failing test. spawn(), POST /runs with json!({ "manifest": "m.jsonl", "models": ["claude"], "epochs": 2 }). Assert resp.status() == 201 and the returned body["id"].as_str() is longer than 10 chars.
2. Predict: before implementing create_run, if you accidentally wrote the handler as async fn create_run(Json(body): Json<CreateRun>, State(state): State<AppState>) — Json first — what does the compiler say, and does it name the real problem? (Recall the concept chapter's Handler<_, _> trap.)
3. Run — it fails to compile (no app/handlers yet).
4. Implement AppState, app, ApiError, CreateRun, and create_run. What each does is specified above; how you arrange it is yours.
5. Run green, commit.

Test 2 — post_with_no_models_is_400 (#[tokio::test])

1. Write the failing test. POST /runs with "models": [] and "epochs": 1. Assert resp.status() == 400.
2. Predict: the empty-models check returns ControlError::Invalid. If your ApiError match were missing the Invalid arm and fell through to the _ => 500 catch-all, what status would this test see, and would create_run itself need changing to fix it — or only the one match?
3. Run, confirm the Invalid → 400 mapping, adjust the ApiError match if needed.
4. Run green, commit.

Test 3 — get_missing_run_is_404 (#[tokio::test])

1. Write the failing test. GET /runs/{} for a fresh Uuid::new_v4() that was never posted. Assert resp.status() == 404.
2. Predict: the id here is a valid UUID that simply is not in the store, so it reaches get_run(run_id).await? and comes back None. Which ControlError variant does the None branch produce, and which status does it map to — versus what a malformed id like /runs/not-a-uuid would produce instead?
3. Run — fails (no get_run yet, or the None branch missing).
4. Implement parse_run_id and get_run: parse, get_run, Some → Json, None → NotFound.
5. Run green, commit.

Test 4 — get_run_after_post_returns_queued (#[tokio::test])

1. Write the failing test. POST /runs with "models": ["a", "b"], "epochs": 3; pull id out of the response. Then GET /runs/{id} and assert run["status"] == "queued" and run["job_count"] == 6.
2. Predict: why is job_count 6 and not 5 or 3? Trace models.len() * epochs. And why does status serialize as the lowercase string "queued" rather than "Queued" — which attribute on RunStatus decided that, back in the Core arc?
3. Run — confirm the round trip: a run posted, persisted, and read back through the store with its computed job_count and snake_case status intact.
4. Run green, commit.

Done when

cargo test -p panoptes-control shows 4 passing tests: a posted run comes back 201 with an id, empty models is 400, an unknown but valid id is 404, and a posted run reads back queued with job_count == models.len() * epochs. The ControlError → status mapping lives in exactly one match, and every test drove the real server over a real socket on a random port.

Concept-Check: The Service

Kind: Quiz. One pass over the whole arc before you move on.

The Service arc gave the coordinator a front door: an axum Router mapping paths to handlers, extractors (State, Path, Json) that hand each handler typed pieces of the request, a single ApiError IntoResponse impl where the ControlError taxonomy becomes an HTTP status code, and a test style that spawns the real server on 127.0.0.1:0 and drives it with reqwest. This check mixes a compiler-verified tracing question with judgment questions across all of it.

If a question stings, the fix is upstream: re-read axum — Handlers, State, Extractors, IntoResponse for the extractor-order and IntoResponse questions, and the Build: The Coordinator API spec for the validation and status-mapping details. The theme to carry forward: the type taxonomy maps to HTTP status in one place, and the store is the source of truth.

Next: Part VI, the Scheduler arc — bounded concurrency, retrying only the retryable, the LocalWorker, and graceful shutdown, where the coordinator stops being a request/store layer and becomes a running program that actually executes the queued runs.

Concept: Bounded Concurrency and Retrying the Retryable

Kind: Concept (read, do not code). This is where the coordinator stops being a library and starts being an engine.

Everything before this arc was a part waiting to be driven. control-core gave us the domain and the WorkerHandle seam; control-eval gave us the workload one worker runs; control-store gave us durable runs and jobs; control-service gave us the API that queues them. The scheduler is the piece that takes a queued run and executes it to completion — and doing that well is entirely about two questions:

1. How many jobs run at once? Not one at a time (we would waste the whole async foundation), and not all at once (that way lies exhaustion). Some bounded number.
2. What happens when a job fails? Some failures are worth retrying on another worker; some are not. The scheduler must tell them apart and act — which is exactly what the error taxonomy from Part II was built to let it do.

This chapter is about the two tools that answer those questions: futures::stream::buffer_unordered for bounded concurrency, and a small retry loop that walks the worker pool. Both are tiny. The reasons they are shaped the way they are are the whole lesson.

Why bounded, and not the two obvious extremes

Picture a run that fans out to 200 jobs — say 10 models × 20 epochs. You already know from the async arc that awaiting them one after another is the wrong move: each job spends almost all its wall-clock waiting on a model call, and doing them sequentially means the machine sits idle 199/200 of the time. Async exists precisely so those waits overlap.

So the naive fix is "launch all 200 at once" — futures::future::join_all(jobs) or a tokio::spawn per job. That overlaps every wait, and for a toy it looks great. In a real coordinator it is a foot-gun:

• You open 200 concurrent connections to the model API. Most providers rate-limit hard; you get a wall of 429s, and your "concurrency" turns into a wall of retries.
• You hold 200 jobs' worth of vignette batches and in-flight buffers in memory at once. For large runs that is real pressure, and it scales with the run size — the one thing you do not control.
• You have no back-pressure. If workers are slower than you launch, the pile of started-but-unfinished work grows without limit.

The honest answer is a ceiling: run at most N jobs concurrently, and as each one finishes, start the next. N waits overlap — you get most of the throughput of "all at once" — but you never have more than N in flight, so memory, connections, and rate-limit pressure are all bounded by a number you choose, independent of how big the run is. That is bounded concurrency, and buffer_unordered is the one-line way to get it.

buffer_unordered, and proof the ceiling holds

buffer_unordered(n) is an adapter on a stream of futures. You hand it a Stream whose items are Futures; it polls up to n of them at once, yields each result as it resolves, and starts another future each time a slot frees. You drive the whole thing by pulling the resulting stream with .next().await (or .collect()).

Here is the shape, with an instrument bolted on to prove the ceiling is real. Twelve jobs, each a 50ms stand-in for a model call; a buffer_unordered(4); and two atomics that track how many are in flight and the high-water mark. Predict the peak before you run it.

use futures::stream::{self, StreamExt};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let inflight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));

    // Twelve jobs, each a stand-in for a model call that mostly waits.
    let tasks = (0..12u32).map(|job| {
        let inflight = inflight.clone();
        let peak = peak.clone();
        async move {
            let now = inflight.fetch_add(1, Ordering::SeqCst) + 1;
            peak.fetch_max(now, Ordering::SeqCst);
            sleep(Duration::from_millis(50)).await; // waiting on the "network"
            inflight.fetch_sub(1, Ordering::SeqCst);
            job * 2
        }
    });

    // buffer_unordered(4): at most four futures in flight at once.
    let mut stream = stream::iter(tasks).buffer_unordered(4);
    let mut sum = 0u32;
    while let Some(doubled) = stream.next().await {
        sum += doubled;
    }
    println!("sum = {sum}, peak concurrency = {}", peak.load(Ordering::SeqCst));
}

sum = 132, peak concurrency = 4

The sum is a sanity check that every one of the twelve jobs ran and was collected — (0+1+…+11)·2 = 132. The number that matters is peak concurrency = 4: no matter that we defined twelve jobs, never more than four were in flight at any instant. Raise the 4 to 8 and the peak becomes 8; the run finishes faster but leans harder on whatever is downstream. That single argument is the knob — the whole point is that it is your number, not the run's.

error[E0599]: no method named `buffer_unordered` found for struct
              `futures::stream::Iter<I>` in the current scope
   |
 5 |     let mut stream = stream::iter(tasks).buffer_unordered(4);
   |                                          ^^^^^^^^^^^^^^^^
   |
   = help: items from traits can only be used if the trait is in scope
help: trait `StreamExt` which provides `buffer_unordered` is implemented but
      not in scope; perhaps you want to import it
   |
 1 + use futures::StreamExt;
   |
help: there is a method `buffered` with a similar name

The second question: which failures are worth retrying

Bounded concurrency decides how many jobs run. The other half of the scheduler's job is deciding what to do when one fails — and this is where the error taxonomy you built in Part II finally does the work it was designed for.

Recall the one distinction that taxonomy encodes: ControlError::is_retryable() is true for Worker and Protocol (a worker died, a connection dropped) and false for Invalid and NotFound (a malformed job, a missing run). The scheduler is the code that reads that bit and acts on it:

• A retryable failure means "this exact job, handed to a different worker, might well succeed." A worker crashed mid-job; the network hiccuped. Re-dispatch it.
• A terminal failure means "this job will fail identically no matter who runs it." The spec is malformed. Retrying only burns time and other workers' attention. Surface it.

The mechanism is a loop that, on a retryable error, moves to the next worker in the pool and tries again — up to a cap so a genuinely broken job cannot retry forever. Here is that loop as a runnable toy, with a Worker whose behaviour we can pin, a RetryPolicy with max_attempts, and the exact index arithmetic the build uses (workers[(seed + attempt) % len]):

#[derive(Debug)]
enum DispatchError {
    Retryable(&'static str),
    Terminal(&'static str),
}
impl DispatchError {
    fn is_retryable(&self) -> bool {
        matches!(self, DispatchError::Retryable(_))
    }
    fn reason(&self) -> &str {
        match self {
            DispatchError::Retryable(m) | DispatchError::Terminal(m) => m,
        }
    }
}

enum Behavior {
    Ok,
    Retryable,
    Terminal,
}

struct Worker {
    id: &'static str,
    behavior: Behavior,
}
impl Worker {
    async fn dispatch(&self, job: u32) -> Result<String, DispatchError> {
        match self.behavior {
            Behavior::Ok => Ok(format!("{} handled job {job}", self.id)),
            Behavior::Retryable => Err(DispatchError::Retryable("connection reset")),
            Behavior::Terminal => Err(DispatchError::Terminal("malformed job spec")),
        }
    }
}

struct RetryPolicy {
    max_attempts: u32,
}

async fn run_job_with_retry(
    workers: &[Worker],
    policy: &RetryPolicy,
    seed: usize,
    job: u32,
) -> Result<String, DispatchError> {
    let mut attempt = 0u32;
    loop {
        let worker = &workers[(seed + attempt as usize) % workers.len()];
        println!("  attempt {attempt} -> worker {}", worker.id);
        match worker.dispatch(job).await {
            Ok(out) => return Ok(out),
            Err(e) if e.is_retryable() && attempt + 1 < policy.max_attempts => {
                println!("    retryable ({}) — moving to the next worker", e.reason());
                attempt += 1;
            }
            Err(e) => return Err(e),
        }
    }
}

#[tokio::main]
async fn main() {
    let policy = RetryPolicy { max_attempts: 3 };

    println!("scenario 1: flaky worker, then a good one");
    let workers = vec![
        Worker { id: "w0", behavior: Behavior::Retryable },
        Worker { id: "w1", behavior: Behavior::Ok },
    ];
    println!("  => {:?}\n", run_job_with_retry(&workers, &policy, 0, 42).await);

    println!("scenario 2: a terminal failure is not retried");
    let workers = vec![
        Worker { id: "w0", behavior: Behavior::Terminal },
        Worker { id: "w1", behavior: Behavior::Ok },
    ];
    println!("  => {:?}\n", run_job_with_retry(&workers, &policy, 0, 42).await);

    println!("scenario 3: every worker retryable — attempts exhaust");
    let workers = vec![
        Worker { id: "w0", behavior: Behavior::Retryable },
        Worker { id: "w1", behavior: Behavior::Retryable },
    ];
    println!("  => {:?}", run_job_with_retry(&workers, &policy, 0, 42).await);
}

Predict all three results before you run — which worker each attempt lands on, and whether it ends Ok or Err.

scenario 1: flaky worker, then a good one
  attempt 0 -> worker w0
    retryable (connection reset) — moving to the next worker
  attempt 1 -> worker w1
  => Ok("w1 handled job 42")

scenario 2: a terminal failure is not retried
  attempt 0 -> worker w0
  => Err(Terminal("malformed job spec"))

scenario 3: every worker retryable — attempts exhaust
  attempt 0 -> worker w0
    retryable (connection reset) — moving to the next worker
  attempt 1 -> worker w1
    retryable (connection reset) — moving to the next worker
  attempt 2 -> worker w0
  => Err(Retryable("connection reset"))

Read all three against the theme:

• Scenario 1 is the payoff. w0 fails retryably, the loop advances attempt and computes (0+1) % 2 = 1 → w1, which succeeds. A flaky worker cost one attempt, not the job. This is "lose nothing under worker failure" in miniature.
• Scenario 2 is the guard rail. w0 fails terminally, and the Err(e) => arm returns immediately — w1 is never tried. A malformed job does not get to march through every worker in the pool, one after another, failing identically each time.
• Scenario 3 is the cap earning its keep. Both workers are flaky, so no attempt ever succeeds; max_attempts = 3 stops the loop after three tries (indices 0,1,0 — the modulo wraps back to w0) and surfaces the last error instead of spinning forever.

One-for-one: the toy ↔ the real thing

Everything here maps straight onto the Scheduler you build next, piece for piece:

• buffer_unordered(4) over a stream of job-futures ↔ Scheduler::process_run — it builds one future per job and drives them with stream::iter(futures).buffer_unordered(self.concurrency), concurrency defaulting to 4.
• run_job_with_retry(workers, policy, seed, job) ↔ the real function of the same name — same workers[(seed + attempt) % workers.len()] walk, same RetryPolicy { max_attempts: 3 }, same "retryable-and-under-cap" guard.
• the toy Worker with an inherent async fn dispatch ↔ Arc<dyn WorkerHandle> — in the build the workers are trait objects behind the seam from Part I, so the same retry loop drives a LocalWorker today and a RemoteWorker in Part VIII without changing a line.
• DispatchError::is_retryable() ↔ ControlError::is_retryable() — the real taxonomy, doing the exact job it was designed for: telling the scheduler which failures to re-dispatch.
• the toy's per-job closure returning job * 2 ↔ the real closure that runs the retry loop, then appends the outcome's records to the log and records the outcome in the store — same bounded-concurrency skeleton, real side effects inside.

Hold the last point against the course's spine. The scheduler holds Arc<dyn WorkerHandle> and never learns local from remote; run_job_with_retry walks that pool on retryable failures. When Part VIII drops RemoteWorker into the pool, this loop is what transparently starts re-dispatching failed jobs across machines — bounded concurrency and retry-the-retryable, unchanged, now distributed.

Questions to lock

1. Why is bounded concurrency the right choice over both "one at a time" and "all at once"? What does the bound protect, and why does the right bound depend on downstream limits rather than on the run's size?
2. What does buffer_unordered(n) guarantee about how many futures are in flight, and why does it not require its futures to be 'static the way tokio::spawn would?
3. In run_job_with_retry, why are both clauses of the retry condition load-bearing? Describe the failure mode you get by dropping each one.
4. A Worker failure and a Protocol failure are both retryable; an Invalid failure is terminal. Trace what the scheduler does with each, and why "retry on the next worker" is the correct response to the retryable ones.

Next: we build it — plan_jobs, run_job_with_retry, Scheduler::process_run/tick, the LocalWorker behind the seam, and the WorkerSource that hands the scheduler its pool.

Build: LocalWorker + the Scheduler

Maps to: Phase 4 (scheduler + LocalWorker). Kind: Build.

Objective

Turn the parts into an engine. In the panoptes-control crate (the binary crate Part V started for the API), add two modules: worker.rs with LocalWorker — the first concrete WorkerHandle, running the eval in-process — and scheduler.rs with plan_jobs, run_job_with_retry, the WorkerSource seam, and the Scheduler that claims a queued run, splits it into jobs, and drives them with bounded concurrency and retry-the-retryable. By the end, tick() takes a queued run all the way to Done, with its records in the log and the store — and it does so through dyn WorkerHandle, never learning local from remote.

Scaffold

Create (two new modules in the existing crates/panoptes-control):

• crates/panoptes-control/src/worker.rs — LocalWorker.
• crates/panoptes-control/src/scheduler.rs — RetryPolicy, WorkerSource, plan_jobs, run_job_with_retry, Scheduler.

Modify:

• crates/panoptes-control/src/lib.rs — add pub mod scheduler; and pub mod worker;, and re-export the public surface: pub use scheduler::{RetryPolicy, Scheduler, WorkerSource, plan_jobs}; and pub use worker::LocalWorker;.
• crates/panoptes-control/Cargo.toml — ensure [dependencies] has futures = { workspace = true } (for buffer_unordered) and async-trait = { workspace = true } (for the WorkerHandle impl); control-core, control-eval, control-store, tokio, and tracing are already there from Part V. [dev-dependencies] needs pretty_assertions.

Dependencies this chapter exercises: futures (the buffer_unordered bounded-concurrency adapter — see the concept chapter), async-trait (the LocalWorker: WorkerHandle impl behind dyn), control-store (claim_next_run, insert_jobs, record_job_outcome, new_job), control-eval (run_eval, load_manifest, append_records), tokio::fs (creating the out-dir).

Expected result: cargo test -p panoptes-control → 5 new tests pass — four in scheduler.rs (plan_jobs_is_one_per_model_epoch, tick_processes_a_run_to_done, retryable_failure_is_redispatched_to_the_next_worker, terminal_failure_is_not_retried) and one in worker.rs (local_worker_runs_the_eval), alongside the API tests already passing.

The spec (givens)

plan_jobs — split a run into the unit that fans out

/// One job carrying the whole vignette batch, per model, per epoch.
pub fn plan_jobs(run: &Run, vignettes: &[Vignette]) -> Vec<Job>;

For every model in run.models, for every epoch in 0..run.epochs, produce one Job carrying the full vignettes batch, that model, and that epoch. Order is model-outer, epoch-inner. Build each job with control_store::new_job(run.id, EvalJob { vignettes: vignettes.to_vec(), model: model.clone(), epoch }), which stamps a fresh JobId and JobStatus::Pending. A run with 2 models × 3 epochs yields 6 jobs — this is the model × epoch fan-out unit the whole coordinator is built around.

→ Answer key

RetryPolicy and run_job_with_retry — retry only the retryable, on the next worker

#[derive(Clone, Copy)]
pub struct RetryPolicy {
    pub max_attempts: u32, // Default: 3
}

async fn run_job_with_retry(
    workers: &[Arc<dyn WorkerHandle>],
    policy: RetryPolicy,
    seed: usize,
    job: Job,
) -> Result<JobOutcome, ControlError>;

RetryPolicy derives Default with max_attempts: 3. run_job_with_retry loops: on attempt n, dispatch to workers[(seed + n) % workers.len()]. On Ok, return it. On an error that is retryable and is under the attempt cap (e.is_retryable() && attempt + 1 < policy.max_attempts), bump attempt and try the next worker. On any other error — terminal, or the cap reached — return the error. The seed is the job's index in the run, so different jobs start on different workers and the pool spreads evenly. This is the concept chapter's loop, now over Arc<dyn WorkerHandle>.

→ Answer key

WorkerSource and StaticPool — where the scheduler gets its pool

/// Where the scheduler gets its workers — snapshotted once per run.
pub trait WorkerSource: Send + Sync {
    fn snapshot(&self) -> Vec<Arc<dyn WorkerHandle>>;
}

/// A fixed set of workers — the local, single-process case.
struct StaticPool(Vec<Arc<dyn WorkerHandle>>);

The scheduler does not hold a Vec of workers directly; it holds an Arc<dyn WorkerSource> and calls snapshot() once at the start of each run. StaticPool is the trivial source — it clones its fixed vec. Why the indirection now, when the pool is fixed? Because in Part VIII the remote pool gains and loses workers as connections open and close, and snapshot() is the seam that lets the pool change underneath without the scheduler noticing. Introduce it here so the payoff arc has somewhere to plug in.

→ Answer key

Scheduler — fields, tick, and process_run

pub struct Scheduler {
    store: Store,
    workers: Arc<dyn WorkerSource>,
    policy: RetryPolicy,
    concurrency: usize, // 4
    out_dir: PathBuf,
}

impl Scheduler {
    /// Over a fixed local worker pool.
    pub fn new(store: Store, workers: Vec<Arc<dyn WorkerHandle>>, out_dir: impl Into<PathBuf>) -> Self;
    /// Over any worker source (the Part VIII hook).
    pub fn with_source(store: Store, workers: Arc<dyn WorkerSource>, out_dir: impl Into<PathBuf>) -> Self;

    /// Claim one queued run and process it to completion. `None` if nothing is queued.
    pub async fn tick(&self) -> Result<Option<RunId>, ControlError>;
}

new wraps the vec in a StaticPool and delegates to with_source; both set policy to RetryPolicy::default() and concurrency to 4.

tick calls store.claim_next_run().await?; if that is None, return Ok(None); otherwise remember the run id, run the (private) process_run(run), and return Ok(Some(run_id)).

process_run(run) is the heart:

1. tokio::fs::create_dir_all(&self.out_dir).await?.
2. let workers = self.workers.snapshot(); — snapshot the pool once. If it is empty, return ControlError::Worker("no workers available").
3. load_manifest(&run.manifest).await? → the vignettes; plan_jobs(&run, &vignettes) → the jobs; self.store.insert_jobs(&jobs).await?.
4. Build one future per job, .enumerate()d so each job's index is its retry seed. Inside each: run_job_with_retry(&workers, policy, i, job).await?, then append_records(&log_path, &outcome.records).await?, then store.record_job_outcome(run_id, &outcome).await?. The log path is out_dir.join(format!("{run_id}.jsonl")).
5. Drive them: stream::iter(futures).buffer_unordered(self.concurrency), pulling with .next().await and ?-propagating each result.

record_job_outcome (from Part IV) is what advances done_count and flips the run to Done when the last job lands — so process_run never sets run status itself; it just records outcomes and lets the store's transactional bookkeeping close the run out.

→ Answer key

LocalWorker — the first concrete WorkerHandle

pub struct LocalWorker {
    id: String,
    client: Arc<dyn ModelClient>,
}

impl LocalWorker {
    pub fn new(id: impl Into<String>, client: Arc<dyn ModelClient>) -> Self;
}

#[async_trait]
impl WorkerHandle for LocalWorker {
    fn id(&self) -> &str;
    async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError>;
}

dispatch is three lines of glue: run_eval(self.client.as_ref(), &job.spec).await? for the records, then wrap them in JobOutcome { job_id: job.id, records }. That is the entire difference between "runs here" and "runs on another machine" — the LocalWorker runs run_eval in-process; the RemoteWorker of Part VIII ships job over a socket. Both are just a WorkerHandle, and the scheduler holds them as Arc<dyn WorkerHandle> without knowing which.

→ Answer key

Concepts exercised

• Bounded concurrency with futures::stream::buffer_unordered over a stream of per-job futures.
• A retry loop that re-dispatches retryable failures across a pool and lets terminal ones surface.
• The dyn WorkerHandle seam paying off: one LocalWorker impl, driven by a scheduler that names no concrete worker type.
• A WorkerSource indirection that lets the pool be fixed now and live later.
• Composing the store's atomic claim and transactional recording (Part IV) with the eval workload (Part III) into one drive loop.

The build loop (you drive)

Test 1 — plan_jobs_is_one_per_model_epoch (in scheduler.rs)

1. Write the failing test. Build a Run with models = ["a", "b"], epochs = 3, one vignette. Call plan_jobs(&run, &[vignette]). Assert jobs.len() == 6 and jobs.iter().all(|j| j.status == JobStatus::Pending).
2. Predict: if you looped epoch-outer, model-inner instead, would len() change? Would any test in this chapter notice? (What is pinned about the order?)
3. Run — fails to compile (no plan_jobs).
4. Implement plan_jobs.
5. Run green, commit.

Test 2 — local_worker_runs_the_eval (in worker.rs, #[tokio::test])

1. Write the failing test. Define a StubClient: ModelClient with no network — generate(prompt) returns ModelResponse { text: format!("re:{prompt}"), .. }. Build LocalWorker::new("w1", Arc::new(StubClient)) and a Job with two vignettes ("a"/"x", "b"/"y"), model "claude", epoch 0. dispatch(job.clone()).await.unwrap(). Assert worker.id() == "w1", outcome.job_id == job.id, outcome.records.len() == 2, outcome.records[0].response == "re:x".
2. Predict: the stub reports model_name() == "stub", but the job's model is "claude". Which lands in each record's model field, and why is that run_eval's doing rather than the worker's?
3. Run — fails to compile (no LocalWorker).
4. Implement LocalWorker + the WorkerHandle impl.
5. Run green, commit.

Test 3 — retryable_failure_is_redispatched_to_the_next_worker (in scheduler.rs, #[tokio::test])

1. Write the failing test. Define a RecordingWorker (one record per vignette, no network) and a FlakyWorker that always returns ControlError::Worker("boom") and counts calls in an Arc<AtomicUsize>. Pool = [FlakyWorker, RecordingWorker]. Build a one-vignette Job. Call run_job_with_retry(&workers, RetryPolicy::default(), 0, job).await.unwrap(). Assert the flaky worker was called exactly once and outcome.records.len() == 1.
2. Predict: with seed = 0, attempt 0 hits index 0 (flaky) and attempt 1 hits (0+1) % 2 = 1 (recording). If you passed seed = 1 instead, which worker would attempt 0 hit — and would the test still pass?
3. Run — fails to compile (no run_job_with_retry).
4. Implement RetryPolicy + run_job_with_retry.
5. Run green, commit.

Test 4 — terminal_failure_is_not_retried (in scheduler.rs, #[tokio::test])

1. Write the failing test. Define a BadInput worker that returns ControlError::Invalid("nope"). Pool = [BadInput]. Call run_job_with_retry and unwrap_err(). Assert !err.is_retryable().
2. Predict: Invalid is terminal, so the loop returns on the first Err(e) => arm. If dispatch had returned ControlError::Worker instead, how many times would BadInput be called before the loop gave up, and why? (Hint: one worker, max_attempts = 3, modulo wraps.)
3. Run, check, implement if the arm order was wrong (terminal must fall through the retryable guard).
4. Run green, commit.

Test 5 — tick_processes_a_run_to_done (in scheduler.rs, #[tokio::test])

1. Write the failing test. Store::in_memory(). Write a 2-vignette manifest to a temp file. Insert a Queued run over it (models = ["claude"], epochs = 1). Build Scheduler::new(store.clone(), vec![Arc::new(RecordingWorker { .. })], &out_dir). Assert sched.tick().await.unwrap() == Some(run.id). Then read the run back: status == Done, done_count == 1, and store.run_results(run.id).await.unwrap().len() == 2 (1 job × 2 vignettes). Finally assert a second tick() returns None — nothing queued left.
2. Predict: process_run never sets the run's status to Done itself. What does, and when? (Recall record_job_outcome from Part IV.)
3. Run — fails to compile (no Scheduler).
4. Implement WorkerSource, StaticPool, Scheduler::new/with_source, tick, process_run.
5. Run green, commit.

Done when

cargo test -p panoptes-control is green with the five new tests: plan_jobs yields one job per model×epoch, LocalWorker runs the eval through the seam, a retryable failure lands on the next worker while a terminal one surfaces at once, and tick drives a queued run all the way to Done with its records in both the log and the store.

Concept: Graceful Shutdown with watch + select!

Kind: Concept (read, do not code).

The scheduler from the last build runs one tick and returns. To make the coordinator a program, we wrap tick in a loop that polls for queued runs forever. But "forever" is a problem: a service has to be able to stop — on Ctrl-C, on a deploy, on a SIGTERM from the orchestrator. And here the course's opening promise comes due. Under worker failure, no eval is lost — but a careless shutdown is itself a kind of failure. Stop the scheduler the wrong way, mid-run, and you drop jobs that were in flight: evals lost not to a crash but to your own exit path.

This chapter is about stopping correctly: a shutdown that drains the in-flight run to completion and only then exits, rather than cancelling it. The two tools are tokio::sync::watch (to carry the signal) and tokio::select! (to react to it without blocking). But the real lesson is about where you check the signal, because that placement is the entire difference between draining and dropping.

Why dropping a future is the danger

Recall from the async arc what a future is: an inert description of work that only advances when it is polled. The runtime polls it under .await. The flip side, the part that bites here: if you stop polling a future — if you drop it — its work simply stops, wherever it was. No unwind, no error, no chance to finish. A half-done eval future that gets dropped never writes its records. The model call already spent is gone; the job is neither done nor recorded.

Now consider the tempting, wrong way to build a stoppable loop. You have a shutdown signal and a run to process, so you race them:

// WRONG: races the run against shutdown
tokio::select! {
    _ = self.tick() => {}          // process the run
    _ = shutdown.changed() => {}   // ...or bail the instant shutdown fires
}

select! polls both futures and, the moment either is ready, runs that branch and drops the other. So if shutdown fires while tick() is halfway through a run, select! drops the tick() future — cancelling the in-flight run, dropping every job still running inside it. That is exactly the eval-loss we swore off. The run was in progress, and we threw it away to exit a few hundred milliseconds sooner.

The fix: check between ticks, drain within one

The correct shape flips the relationship. The unit of work — one tick, which processes one whole run — runs to completion outside any select!. The shutdown signal is only consulted at the top of the loop (should I claim another run?) and to cut the idle wait short (if I'm sleeping between polls, wake up now). Shutdown never races the run itself.

Concretely: check the flag at the top; if set, stop claiming and break. Otherwise process a whole run. Then, for the idle interval between polls, select! the sleep against the signal — so a shutdown during the idle wait wakes us immediately, but a shutdown during a run is simply observed at the next top-of-loop check, after the run has drained.

tokio::sync::watch is the right channel for this. It is a single-value broadcast latch: one sender sets a value, any number of receivers can cheaply read the latest with borrow() or await a change with changed(). That "many receivers, one latched value" shape is exactly what a shutdown flag is — the scheduler loop and the HTTP server both hold a receiver and both see the one flip. (An mpsc would be consumed by whoever read it first; a watch lets everyone see the same latched true.)

Here is the whole pattern as a runnable toy. process_one stands in for tick — an indivisible unit of work that, once started, runs to completion. Watch where the signal is checked.

use std::time::Duration;
use tokio::sync::watch;
use tokio::time::sleep;

async fn process_one(n: u32) {
    // Stand-in for "process a whole run": once started, it runs to completion.
    println!("  processing unit {n} (takes 100ms)...");
    sleep(Duration::from_millis(100)).await;
    println!("  unit {n} done");
}

async fn poll_loop(mut shutdown: watch::Receiver<bool>, poll: Duration) {
    let mut next = 0u32;
    loop {
        if *shutdown.borrow() {
            println!("shutdown observed at top of loop — stop claiming");
            break;
        }
        // A whole unit of work is drained before we re-check shutdown.
        process_one(next).await;
        next += 1;

        tokio::select! {
            _ = shutdown.changed() => println!("woken early by shutdown signal"),
            _ = sleep(poll) => println!("poll interval elapsed"),
        }
    }
    println!("loop exited cleanly, drained {next} units");
}

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel(false);
    let worker = tokio::spawn(poll_loop(rx, Duration::from_millis(30)));

    // Fire shutdown 180ms in — squarely in the middle of unit 1.
    sleep(Duration::from_millis(180)).await;
    println!("main: sending shutdown");
    tx.send(true).unwrap();

    worker.await.unwrap();
}

Predict the output before running — in particular, when shutdown fires at 180ms (unit 1 is mid-flight, from ~130ms to ~230ms), does unit 1 finish or get dropped?

  processing unit 0 (takes 100ms)...
  unit 0 done
poll interval elapsed
  processing unit 1 (takes 100ms)...
main: sending shutdown
  unit 1 done
woken early by shutdown signal
shutdown observed at top of loop — stop claiming
loop exited cleanly, drained 2 units

Read the last four lines closely, because they are the guarantee:

• main: sending shutdown prints while unit 1 is still running. The signal is now true.
• unit 1 done prints anyway. The in-flight unit drained — it was not racing the signal, so it ran to completion. That is the eval that would have been lost under the select!-around-the-work design.
• woken early by shutdown signal — back at the bottom of the loop, changed() had already fired, so the select! returned immediately instead of sleeping out the 30ms. Shutdown is prompt: we don't dawdle in the idle wait.
• shutdown observed at top of loop — the next iteration's top-of-loop check sees true and breaks. No new unit is claimed.

So the two behaviours we want fall out of one placement decision: drain (the running unit finishes because it is not on a select! arm) and prompt (the loop wakes from its idle sleep the instant the signal flips). Shutdown stops us claiming new work immediately; it never throws away work already in flight.

error[E0596]: cannot borrow `shutdown` as mutable, as it is not declared as mutable
 --> src/main.rs:8:17
  |
8 |             _ = shutdown.changed() => {}
  |                 ^^^^^^^^ cannot borrow as mutable
  |
help: consider changing this to be mutable
  |
4 | async fn poll_loop(mut shutdown: watch::Receiver<bool>, poll: Duration) {
  |                    +++

One-for-one: the toy ↔ the real thing

This maps directly onto Scheduler::run_loop, which you build next:

• poll_loop(mut shutdown: watch::Receiver<bool>, poll: Duration) ↔ Scheduler::run_loop(mut shutdown: watch::Receiver<bool>, poll: Duration) — same signature, same watch channel.
• process_one(n) — an indivisible unit run to completion ↔ self.tick() — which processes one whole run (all its jobs, at buffer_unordered concurrency) before returning. Draining "one unit" means draining an entire run's fan-out.
• if *shutdown.borrow() { break } at the top ↔ the identical top-of-loop check — stop claiming new runs once shutdown is set.
• select! { changed() / sleep(poll) } at the bottom ↔ the identical idle-wait race — wake promptly on shutdown, otherwise sleep the poll interval. (The real loop also continues immediately when a tick did process a run, so back-to-back queued runs don't wait out a poll interval.)
• the watch::Sender in main ↔ the coordinator binary's shutdown_tx, fired from tokio::signal::ctrl_c() inside axum's with_graceful_shutdown — one signal that stops the server accepting requests and tells the scheduler to drain and stop, together.

Hold this against the spine one more time. The scheduler drives dyn WorkerHandle; in Part VIII some of those are RemoteWorkers on other machines. When that day comes, a graceful shutdown here will drain an in-flight run whose jobs are executing across the cluster — and because the drain is a property of where we check the signal, not of where the jobs run, it keeps working unchanged. Shutdown drains; it never drops.

Questions to lock

1. What happens to a future when it is dropped, and why does that make "race the run against shutdown in select!" lose evals?
2. Where in the loop is the shutdown flag checked, and how does that placement produce both draining (finish the in-flight run) and promptness (don't sleep out the idle interval)?
3. Why is tokio::sync::watch the right channel for a shutdown signal shared by the scheduler loop and the HTTP server, rather than an mpsc?
4. Why is the top-of-loop check borrow() and not borrow_and_update(), and why must the receiver be mut?

Next: we build run_loop and wire up the panoptes-control binary — clap flags, the store, the worker pool, the scheduler, and axum::serve with graceful shutdown, all under one signal. The coordinator becomes a program you can run.

Build: The Run Loop and the Coordinator Binary

Maps to: Phase 4 (run_loop + main). Kind: Build.

Objective

Close the arc: wrap tick in Scheduler::run_loop — the poll-forever loop that drains its in-flight run and stops on a watch signal — and then write main.rs, the panoptes-control binary that connects the store, assembles a LocalWorker pool, spawns the run loop, and serves the axum API, all under one graceful-shutdown signal. There are no new unit tests here; the payoff is different — the binary compiles and runs. panoptes-control becomes a program you can start, POST a run to, and Ctrl-C without losing the run in flight.

Scaffold

Modify:

• crates/panoptes-control/src/scheduler.rs — add run_loop to the impl Scheduler.
• crates/panoptes-control/src/main.rs — the binary (Part V may have left a stub; this is its real body).
• crates/panoptes-control/Cargo.toml — ensure [dependencies] has clap = { workspace = true } (features ["derive"]) and anyhow = { workspace = true }; axum, tokio, control-store, control-eval are already present.

Dependencies this chapter exercises: tokio::sync::watch and tokio::select! (the drain loop — see the shutdown concept), tokio::signal::ctrl_c (the OS signal), axum::serve with with_graceful_shutdown (Part V's app), clap derive (the CLI), anyhow (the binary's Result).

Expected result: cargo build -p panoptes-control succeeds and produces the panoptes-control binary; cargo test -p panoptes-control stays green (the five tests from the last chapter, unchanged — run_loop is exercised by running the binary, not by a new unit test).

The spec (givens)

Scheduler::run_loop — poll forever, drain on shutdown

/// Poll for queued runs until shutdown. A tick processes a whole run before
/// the loop re-checks shutdown, so shutdown *drains* the in-flight run rather
/// than dropping its jobs; it just stops claiming new ones.
pub async fn run_loop(
    &self,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
    poll: Duration,
);

The loop, exactly as the concept chapter argued it:

1. Top check. if *shutdown.borrow() { break; } — once shutdown is set, claim nothing more and exit.
2. Tick. match self.tick().await:
   • Ok(Some(_)) → a run was processed; continue immediately, so back-to-back queued runs don't wait out a poll interval.
   • Ok(None) → nothing queued; fall through to the idle wait.
   • Err(e) → log it (eprintln! is fine) and fall through — a tick error must not kill the loop.
3. Idle wait. tokio::select! { _ = shutdown.changed() => {}, _ = tokio::time::sleep(poll) => {} } — sleep the poll interval, but wake early the instant shutdown flips.

The unit of work — tick, one whole run — runs outside the select!. That placement is the entire drain guarantee: shutdown during a run is observed only at the next top-of-loop check, after the run has finished; shutdown during the idle wait wakes us at once. Note the receiver is mut (because changed() takes &mut self).

→ Answer key

main.rs — the coordinator binary

#[derive(clap::Parser)]
#[command(name = "panoptes-control", version, about = "The Panoptes eval control plane")]
struct Cli {
    /// SQLite database URL.
    #[arg(long, default_value = "sqlite://control.db?mode=rwc")]
    db: String,
    /// Address to bind the API to.
    #[arg(long, default_value = "127.0.0.1:8080")]
    addr: String,
    /// Number of in-process workers.
    #[arg(long, default_value_t = 4)]
    workers: usize,
    /// Directory for response logs.
    #[arg(long, default_value = "data")]
    out_dir: String,
}

#[tokio::main] async fn main() -> anyhow::Result<()> assembles the program in this order:

1. let cli = Cli::parse(); and panoptes_control::telemetry::init(); (from Part V).
2. let store = Store::connect(&cli.db).await?;.
3. Build the worker pool. Construct one shared Arc::new(HttpModelClient::new(base_url, "claude")) and cli.workers LocalWorkers over it, collected into a Vec<Arc<dyn WorkerHandle>> (format!("local-{i}") ids). The model endpoint base_url is read from the environment (e.g. PANOPTES_MODEL_API, defaulting to http://127.0.0.1:9000) — the four CLI flags stay as specified above; the model API location is deployment config, not a run parameter.
4. let scheduler = Arc::new(Scheduler::new(store.clone(), workers, &cli.out_dir));.
5. One shutdown signal. let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);. Spawn the loop: tokio::spawn({ let s = scheduler.clone(); let rx = shutdown_rx.clone(); async move { s.run_loop(rx, Duration::from_millis(500)).await } }) — hold the JoinHandle.
6. Serve. Bind cli.addr, then:

   axum::serve(listener, app(AppState { store }))
       .with_graceful_shutdown(async move {
           let _ = tokio::signal::ctrl_c().await;
           let _ = shutdown_tx.send(true);
       })
       .await?;

   Ctrl-C makes axum stop accepting new connections and fires shutdown_tx.send(true), which the scheduler's run_loop sees — one signal, both halves.
7. Join the drain. let _ = sched_task.await; — wait for the scheduler to finish its in-flight run before the process exits.

→ Answer key

Concepts exercised

• A poll-forever service loop that drains in-flight work on shutdown (watch + select!, work outside the race).
• One watch signal shared by two subsystems — the HTTP server (with_graceful_shutdown) and the scheduler (run_loop).
• tokio::spawn for a long-lived background task, held by its JoinHandle so the main task can await its clean exit.
• clap derive turning a struct into the binary's CLI.
• Assembling every crate built so far — control-store, control-eval, the API from Part V, and this arc's scheduler — into one running program.

The build loop (you drive)

There is no red-green test cycle here; the compiler and a manual smoke test are the graders.

1. Add run_loop to scheduler.rs. Re-run cargo test -p panoptes-control — the five existing tests must still pass (you changed no behaviour they touch).
2. Predict: if you had written the loop as select! { _ = self.tick() => {}, _ = shutdown.changed() => {} } instead — the run inside the race — what would a Ctrl-C during a large run do to that run's un-recorded jobs? (Re-read the shutdown concept's WRONG example.)
3. Write main.rs. cargo build -p panoptes-control — fix until it compiles. A binary that builds is the milestone this chapter promised.
4. Smoke-test the drain (optional but worth it). Start a stub model endpoint on :9000 (a wiremock server, or the mock from Part III), run panoptes-control --workers 2, POST a run to /runs, watch it process, then Ctrl-C. Observe: the server stops, the scheduler finishes the run it was on, data/<run-id>.jsonl is complete, and the process exits. Shutdown drained; it did not drop.

Done when

cargo build -p panoptes-control produces the binary, cargo test -p panoptes-control is green, and you can start the coordinator, submit a run, and Ctrl-C it mid-run without the in-flight run's records going missing — the shutdown drains the run, then the process exits.

Concept-Check: The Scheduler

This is the arc where the coordinator became an engine. It claims a queued run, splits it into one job per model×epoch, and drives those jobs at a bounded concurrency it chooses — retrying the retryable failures on the next worker in the pool and letting the terminal ones surface. Then it wraps that in a loop that can be stopped without loss: a shutdown that drains the in-flight run rather than dropping its jobs. And every worker it touches is a dyn WorkerHandle, so the whole thing is ready for remote workers to plug in behind the same seam. If the pieces below are solid — the bound, the retry condition, and where the shutdown check goes — the cluster arc has an engine worth distributing.

Concept: tracing — Spans, instrument, and the Trace Layer

Kind: Concept. New crate: tracing — this chapter shows it working before you build with it.

Observability is not an afterthought

By the end of Part VI your coordinator is a running program: it claims a run, splits it into jobs, and dispatches them across a worker pool with bounded concurrency. That last phrase — bounded concurrency — is exactly what makes it hard to see what it is doing. At any instant four jobs are in flight, from different models and epochs, their await points interleaved on the runtime. When something goes wrong at two in the morning during a real eval, "add a println!" produces this:

processing order
processing order
reserved
processing order
reserved
reserved
order complete

Which run did the third reserved belong to? You cannot tell. The lines from independent concurrent tasks are shuffled together, none of them carries the id of the work it came from, and there is no way to filter to just the run you care about. A println! is a string with no structure and no context — and concurrency is precisely the setting where a string with no context is useless.

tracing exists to fix this. It is the observability layer the whole coordinator threads through, and the design goal is one sentence: every line of output should know which run produced it, and you should be able to ask for exactly the lines you want without recompiling. This chapter builds that machinery on a toy so that when you wire it into process_run, the run id follows every job through the entire dispatch — for free, because the span carries it.

Events and spans: the two primitives

tracing has exactly two things you emit, and the distinction is the whole idea.

An event is a single moment — a structured println!. Instead of formatting a string, you name a message and attach typed key/value fields:

tracing::info!(user = "ada", "logged in");

A span is a period of time with a name and its own fields. While a span is entered, every event — and every nested span — inherits its context automatically. You do not pass the context down by hand; entering the span is what attaches it.

Here are both, in a program that needs nothing but tracing and tracing-subscriber:

use tracing::{info, info_span};

fn main() {
    tracing_subscriber::fmt().with_target(false).init();

    // An EVENT: a single moment, like a structured println.
    info!(user = "ada", "logged in");

    // A SPAN: a period of time whose fields every event inside inherits.
    let span = info_span!("checkout", cart_id = 7);
    let _guard = span.enter();
    info!(item = "book", "added"); // inherits cart_id=7
    info!(item = "pen", "added");  // inherits cart_id=7
    // `_guard` drops here → the span closes
}

Run it and read the output closely (timestamps will differ each run):

2026-07-21T19:18:45.992143Z  INFO logged in user="ada"
2026-07-21T19:18:45.992312Z  INFO checkout{cart_id=7}: added item="book"
2026-07-21T19:18:45.992334Z  INFO checkout{cart_id=7}: added item="pen"

The first event has no span, so it prints bare. The two events inside the span are each prefixed with checkout{cart_id=7}: — the span's name and fields, stamped onto every line beneath it, without either info! mentioning cart_id. That prefix is the thing a println! can never give you: a way to know, from the line alone, what larger unit of work it belongs to. A span is entered when the guard is created and closed when the guard drops — the span's lifetime is a real Rust scope, which is why nesting Just Works.

#[tracing::instrument]: the span you don't write by hand

Entering a span manually is fine, but the common case is "wrap this whole function in a span named after it, with these arguments as fields." That is a macro: #[tracing::instrument]. Put it on a function and every call gets its own span, opened on entry and closed on return, with the function's arguments recorded as fields.

This is the toy that mirrors the build one-for-one. process_order takes an order, does some nested work per item, and we want the order id stamped on every line the call produces — including lines from the functions it calls:

use tracing::info;
use tracing_subscriber::EnvFilter;

#[derive(Debug)]
struct Order {
    id: u32,
    items: Vec<&'static str>,
}

#[tracing::instrument(skip(order), fields(order_id = order.id))]
fn process_order(order: &Order) {
    info!(items = order.items.len(), "processing order");
    for item in &order.items {
        reserve_item(item);
    }
    info!("order complete");
}

#[tracing::instrument]
fn reserve_item(item: &str) {
    info!("reserved");
}

fn main() {
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    tracing_subscriber::fmt()
        .with_env_filter(filter)
        .with_target(false) // trims the module-path column for readability
        .init();

    process_order(&Order { id: 42, items: vec!["widget", "gadget"] });
}

The real, captured output:

2026-07-21T19:20:40.618206Z  INFO process_order{order_id=42}: processing order items=2
2026-07-21T19:20:40.618404Z  INFO process_order{order_id=42}:reserve_item{item="widget"}: reserved
2026-07-21T19:20:40.618456Z  INFO process_order{order_id=42}:reserve_item{item="gadget"}: reserved
2026-07-21T19:20:40.618485Z  INFO process_order{order_id=42}: order complete

Read what the annotation bought us:

• order_id=42 is on every single line — including the two from reserve_item, which never saw the order. The nested spans render as process_order{order_id=42}:reserve_item{item="widget"}:, so a line tells you both the order it belongs to and the item being reserved inside it. That is the context a println! throws away.
• skip(order) tells the macro not to record the whole Order as a field. By default #[instrument] records every argument via its Debug, which for a big struct is noise (and for a secret would be a leak). skip opts a field out.
• fields(order_id = order.id) then adds back exactly the one piece we want — the id — pulled out of the skipped argument. skip the noisy whole, fields the useful part: that pairing is the idiom.

Now hold this next to the build. In process_run, the annotation is:

#[tracing::instrument(skip(self, run), fields(run_id = %run.id))]
async fn process_run(&self, run: Run) -> Result<(), ControlError> { ... }

Identical shape. skip(self, run) drops the two big arguments; fields(run_id = %run.id) records just the run id. From that point on, every log the scheduler emits while processing that run — planning jobs, dispatching, retrying, recording outcomes — is stamped with run_id, because all of it happens inside the process_run span. The % is one small new thing: it means "record this field using its Display (to_string) rather than its Debug." A RunId displays as the bare uuid, which is what you want in a log; %run.id gets you run_id=3f2a… instead of run_id=RunId(3f2a…). (instrument also handles async fn correctly — it re-enters the span every time the future is polled, so the context survives every .await.)

EnvFilter: turning the firehose up and down without recompiling

Structured lines are only half the win. The other half is choosing which lines you get — at startup, from the environment, without touching the code. That is EnvFilter.

An EnvFilter reads a directive string (conventionally from the RUST_LOG environment variable) and decides, per event, whether it passes. The directive can be as blunt as a level (info, warn) or as surgical as per-module (control_store=debug,info — debug for the store, info for everything else). The subscriber consults the filter for every event; anything below the threshold is dropped before it is ever formatted.

The construction you will use in the build is exactly the one in the toy above:

let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
tracing_subscriber::fmt().with_env_filter(filter).try_init();

try_from_default_env() reads RUST_LOG; if it is unset, the unwrap_or_else supplies a sensible default of "info". So the coordinator is talkative-enough out of the box, and an operator who wants more detail sets RUST_LOG=debug and reruns — no recompile. Run the process_order example with RUST_LOG=warn and all four info lines vanish, because none of them clears the warn bar; run it with nothing set and you get the full output above.

Two more details worth pinning now, because they surprise people:

• fmt() writes to stdout by default, not stderr. (Verified: redirecting stdout to /dev/null silences the logs; redirecting stderr does not.) That is fine for this course, but in a real service you often want logs on stderr so stdout stays clean for actual output — a one-liner, .with_writer(std::io::stderr), if you ever need it.
• try_init is idempotent-friendly. It returns a Result instead of panicking if a global subscriber is already set. That is why telemetry::init can be called from every test's setup without the second call blowing up — the first install wins and the rest quietly no-op. The panicking init() would make the second test crash.

One layer for every request: tower-http's TraceLayer

The API tier gets its telemetry almost for free. axum is built on tower, and tower middleware is composable Layers wrapped around your router. tower-http ships a ready-made one, TraceLayer, that opens a span for every incoming HTTP request and logs when it completes — method, path, status, latency — with no per-handler code at all.

use tower_http::trace::TraceLayer;

pub fn app(state: AppState) -> Router {
    Router::new()
        .route("/runs", post(create_run))
        .route("/stats", get(get_stats))
        // ... other routes ...
        .layer(TraceLayer::new_for_http()) // one line: every request now traced
        .with_state(state)
}

new_for_http() is a preset tuned for HTTP: it knows to pull method and path into the request span and to log the status and duration on the way out. Because it is a layer, it wraps all the routes uniformly — you do not, and must not, sprinkle logging into each handler. That is the same "one place" discipline you already applied to error mapping in Part V (IntoResponse turned the taxonomy into status codes in a single spot); here one layer turns every request into a traced span in a single spot.

tower-http and axum are not on the Rust playground, so the block above is marked ignore — there is no run button. To exercise it, add axum, tower-http (features ["trace"]), and tracing to a scratch crate, or just read it in place; you will build the real thing against the router in the next chapter.

One-for-one with the build

The toy maps onto the coordinator's telemetry exactly:

• #[instrument] on process_order ↔ #[instrument] on process_run (the span that wraps a whole unit of work)
• skip(order) + fields(order_id = order.id) ↔ skip(self, run) + fields(run_id = %run.id) (drop the big args, keep the id)
• order_id stamped on nested reserve_item lines ↔ run_id stamped on every dispatch/retry/record line inside the run
• info_span!("checkout", cart_id = 7) ↔ the request span TraceLayer opens per HTTP call
• EnvFilter from RUST_LOG, default "info" ↔ telemetry::init — the identical try_from_default_env/try_init pattern

Same two primitives, same one annotation, same one filter. Build the order tracer and you have built the run tracer with the labels changed.

Questions to lock

1. In a coordinator running four jobs concurrently, why does an unadorned println! fail to tell you which run a line came from — and what does a span give every line beneath it that fixes this?
2. What is the difference between an event and a span, and what does entering a span do to the events emitted inside it?
3. #[tracing::instrument(skip(self, run), fields(run_id = %run.id))] — say what each of skip(...), fields(...), and the % is doing, and why you would skip an argument only to add one field back.
4. You set RUST_LOG=warn but your info! lines still print. What is the single most likely cause, given how EnvFilter gets attached to a subscriber?
5. TraceLayer::new_for_http() is one line on the router. Why is a layer the right place for request logging rather than a log statement inside each handler?

Next chapter is the build: telemetry::init, the TraceLayer on the router, the run span on process_run, and the /stats endpoint that turns the tokens your workers already report into a cost per model.

Build: Run Spans and /stats Token Accounting

Maps to: Phase 5 (telemetry + /stats). Kind: Build.

Objective

Make the coordinator observable and accountable. You will add five things, all small, all leaning on the same principle: the telemetry is not new bookkeeping bolted on the side — it reads the tokens the workers already report and the run ids the domain already carries.

1. telemetry::init — a tracing subscriber, filtered by RUST_LOG, safe to call from every test.
2. TraceLayer::new_for_http() on the router — one line, and every request is a span.
3. #[tracing::instrument] on process_run — the run id follows every job through dispatch.
4. Store::usage_by_model() returning Vec<ModelUsage> — token totals per model, aggregated from the stored outcomes.
5. GET /stats — turns those totals into tokens and an estimated cost per model, plus a grand total.

By the end, cargo run prints run-stamped logs you can filter, and GET /stats answers "how many tokens has each model burned, and what would it cost?"

Scaffold

Edit crates/control-store/Cargo.toml — nothing new is required; serde is already a dependency (ModelUsage derives Serialize).

Edit crates/panoptes-control/Cargo.toml — add, all { workspace = true }:

• tracing — the #[instrument] macro and the info!/warn! macros.
• tracing-subscriber (features ["env-filter"]) — the fmt subscriber and EnvFilter. The feature is what turns RUST_LOG into a filter; without it, EnvFilter does not exist.
• tower-http (features ["trace"]) — supplies TraceLayer. The trace feature is why the layer is in scope.

tracing and tracing-subscriber are already available to control-store if you choose to log there, but the aggregation itself needs no logging.

Files you touch:

• crates/control-store/src/lib.rs — add ModelUsage and Store::usage_by_model.
• crates/panoptes-control/src/lib.rs — add the telemetry module, the TraceLayer on app, est_cost_usd, and the get_stats handler wired to GET /stats.
• crates/panoptes-control/src/scheduler.rs — add the #[instrument] attribute to process_run.

Expected result:

• cargo test -p control-store → the store suite grows by 1: usage_by_model_sums_across_jobs.
• cargo test -p panoptes-control → the API suite grows by 1: stats_reports_tokens_and_cost_per_model.

The spec (givens)

telemetry::init

A module with one function that installs the global subscriber:

pub mod telemetry {
    pub fn init() { /* you write this */ }
}

The contract — three properties the rest of the system depends on:

• Filtered by RUST_LOG, defaulting to "info". Build an EnvFilter from the default env; if the variable is unset, fall back to EnvFilter::new("info"). This is the exact try_from_default_env().unwrap_or_else(...) shape from the concept chapter.
• Idempotent. Use try_init, not init. It returns a Result instead of panicking when a subscriber is already installed, so every test can call telemetry::init() in its setup and only the first call wins. Discard the Result (let _ = ...).
• fmt subscriber. A plain tracing_subscriber::fmt() with the filter attached is all this course needs.

main.rs calls telemetry::init() once at startup, before serve.

→ Answer key

TraceLayer on the router

app gains exactly one line — a layer, applied after the routes and before .with_state:

.layer(TraceLayer::new_for_http())

Every request now opens a span carrying method and path and logs its status and latency on completion. No handler changes. (Layer order matters in general; for this single layer, placing it after the routes is correct.)

→ Answer key

The run span on process_run

Add this attribute — and nothing else — to process_run in scheduler.rs:

#[tracing::instrument(skip(self, run), fields(run_id = %run.id))]
async fn process_run(&self, run: Run) -> Result<(), ControlError> { ... }

skip(self, run) keeps the two large arguments out of the span's fields; fields(run_id = %run.id) records just the id, via Display (that is the %). From here every info!/warn! the scheduler emits inside the run — including the existing "planned jobs for run" line — is stamped with run_id. The body does not change.

→ Answer key

Store::usage_by_model and ModelUsage

The aggregate type, in control-store:

/// Aggregate token usage for one model, across every recorded job.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct ModelUsage {
    pub model: String,
    pub input_tokens: u64,
    pub output_tokens: u64,
    pub calls: u64,
}

impl Store {
    pub async fn usage_by_model(&self) -> Result<Vec<ModelUsage>, ControlError>;
}

The contract:

• Read every job that has an outcome (the same rows run_results reads), deserialize each JobOutcome, and walk its records.
• Group by record.model. For each record, add usage.input_tokens to the model's input_tokens, usage.output_tokens to its output_tokens, and 1 to calls — calls counts records (one model call each), not jobs.
• Widen the per-record u32 token counts to u64 as you sum, so a long run cannot overflow the total.
• Return one ModelUsage per distinct model. A BTreeMap<String, ModelUsage> keyed by model name gives you grouping and a stable, sorted order for free — worth it so the output and the tests are deterministic.

→ Answer key

GET /stats, est_cost_usd, and the pricing

The cost helper is an illustrative flat model — $3 per 1,000,000 input tokens, $15 per 1,000,000 output tokens:

fn est_cost_usd(u: &ModelUsage) -> f64; // input/1e6 * 3.0 + output/1e6 * 15.0

get_stats calls store.usage_by_model(), maps each ModelUsage to a JSON object, sums the per-model costs into a grand total, and answers this exact shape:

// GET /stats  → 200
{
  "models": [
    {
      "model": "claude",
      "input_tokens": 1000000,
      "output_tokens": 1000000,
      "calls": 1,
      "est_cost_usd": 18.0
    }
  ],
  "total_est_cost_usd": 18.0
}

The worked number pins the arithmetic: 1,000,000 input @ $3/M = $3, plus 1,000,000 output @ $15/M = $15, is $18.00. Wire it as .route("/stats", get(get_stats)); the handler returns Result<Json<serde_json::Value>, ApiError> and uses ? on the store call, exactly like the Part V handlers.

→ Answer key

Concepts exercised

• #[tracing::instrument] with skip/fields to carry a run_id through an async call and its nested work (from Concept: tracing).
• EnvFilter + try_init for a RUST_LOG-controlled, idempotent subscriber.
• TraceLayer as a single tower layer that traces every request uniformly.
• Aggregating stored outcomes in Rust with a BTreeMap for grouped, deterministic results.
• Computing a derived value (cost) from data the system already records, and shaping it into a JSON response with the Part V handler pattern.

The build loop (you drive)

Test 1 — usage_by_model_sums_across_jobs (in control-store/src/lib.rs, #[tokio::test])

1. Write the failing test. Insert a run with job_count = 2 and two jobs. Record an outcome for job 1 with one record usage { input: 100, output: 10 }, and an outcome for job 2 with two records, each usage { input: 50, output: 5 } — all model: "claude". Call store.usage_by_model() and assert: stats.len() == 1, stats[0].model == "claude", stats[0].input_tokens == 200, stats[0].output_tokens == 20, stats[0].calls == 3.
2. Predict: three records land across two jobs. Before you implement the loop — is calls going to be 2 or 3? Say which, and what that tells you about whether calls counts jobs or model calls.
3. Run — it fails to compile (no usage_by_model yet).
4. Implement ModelUsage and usage_by_model per the contract above. What it computes is specified; how you group is yours (a BTreeMap is the clean way).
5. Run green, commit.

Test 2 — stats_reports_tokens_and_cost_per_model (in panoptes-control/src/lib.rs, #[tokio::test])

1. Write the failing test. Spawn the app on a random port with a shared in-memory Store (the existing spawn() helper returns both). Seed a Running run, insert one job for it, and record_job_outcome with a single record whose usage is { input: 1_000_000, output: 1_000_000 }, model: "claude". Then GET /stats, parse the JSON, and assert: stats["models"][0]["model"] == "claude", stats["models"][0]["input_tokens"] == 1_000_000, and stats["total_est_cost_usd"] == 18.0.
2. Predict: with input and output both at 1,000,000, work the pricing by hand — $3/M in, $15/M out. What is est_cost_usd for this model, and what is total_est_cost_usd? (If you get $18.0, your mental model of the formula matches the test.)
3. Run — it fails (no /stats route, no get_stats, no est_cost_usd).
4. Implement est_cost_usd, get_stats, and the .route("/stats", get(get_stats)) wiring. Add the TraceLayer line and telemetry::init while you are in this file.
5. Run green, commit.

Done when

cargo test -p control-store and cargo test -p panoptes-control are both green with the two new tests passing; usage_by_model sums to 200 / 20 / 3 across the two jobs; GET /stats reports input_tokens: 1000000 and total_est_cost_usd: 18.0 for the seeded run; cargo run emits run_id-stamped log lines that RUST_LOG=warn can quiet and RUST_LOG=debug can widen.

Concept-Check: Telemetry

Kind: Quiz. One pass over the whole arc before you move on.

The Telemetry arc gave the coordinator eyes and a ledger. Spans stamp the run_id onto every line a run produces, so concurrent jobs stop being an indistinguishable pile of strings; #[tracing::instrument] opens that span for you and carries it across every .await; EnvFilter lets an operator dial the detail up or down from RUST_LOG without a recompile; and /stats accounts for tokens and cost per model — derived at read time from the same usage the workers already report, never stored twice. This check mixes compiler-verified aggregation questions with judgment about spans, filtering, and where derived values belong.

If a question stings, the fix is upstream: Concept: tracing for spans, instrument, EnvFilter, and the TraceLayer, and Build: Run Spans and /stats for the aggregation and cost math. Re-read the section, then come back.

Next: Part VIII, the Cluster arc — a framed TCP protocol, a connection actor multiplexing one socket, the RemoteWorker behind the same WorkerHandle trait, and at-least-once redelivery made safe by the store's idempotency. The payoff the whole course was built toward.

Concept: TCP Is a Byte Stream — Framing with LengthDelimitedCodec

Kind: Concept. New crate: tokio-util — this chapter shows it working before you build with it.

For seven arcs the coordinator has run every job in-process. A LocalWorker is a WorkerHandle that awaits run_eval on the same machine; the scheduler dispatches to it and gets a JobOutcome back. That is the whole system so far, and it is complete — it schedules, retries, persists, and reports. What it cannot do is spread work across machines. The last arc adds that, and it adds it, as promised in Part I, as a new impl WorkerHandle — the scheduler never changes. But a RemoteWorker has to actually get a Job to another process and a JobOutcome back, and between here and there is a TCP socket. This chapter is about the single most surprising thing about that socket, and the crate that tames it.

The surprising thing: TCP has no messages

You think of the wire in terms of messages: the coordinator sends an Assign, the worker sends back a Result. That is the mental model, and it is the model the rest of the arc is built on. But TCP does not have that model. TCP is a stream of bytes. It guarantees that the bytes you write come out the other end in order and without gaps — and that is all it guarantees. It does not remember where one write ended and the next began. It is free to glue two of your writes into one read, or split one of your writes across two reads, however the kernel and the network happen to buffer things.

That is not a bug or an edge case; it is the definition of a stream protocol, and it bites the moment you send two things in a row. Let us watch it bite. Here is a client that does two entirely separate write_all calls — five bytes, then five more — and a server that reads until the connection closes and reports what it got:

// scratch Cargo.toml deps:
//   tokio = { version = "1", features = ["full"] }
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    // Server: read every byte until the peer closes, then report what arrived.
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        let mut buf = Vec::new();
        sock.read_to_end(&mut buf).await.unwrap();
        buf
    });

    // Client: two *separate* writes, each a whole "message" as far as we mean it.
    let mut client = TcpStream::connect(addr).await.unwrap();
    client.write_all(b"hello").await.unwrap();
    client.write_all(b"world").await.unwrap();
    client.shutdown().await.unwrap(); // close so read_to_end returns

    let got = server.await.unwrap();
    println!("two writes of 5 bytes each; the reader saw {} bytes:", got.len());
    println!("{:?}", String::from_utf8(got).unwrap());
}

Predict the output before you read on. You wrote "hello" and "world" as two calls — how many reads does the other side see, and what is in them?

two writes of 5 bytes each; the reader saw 10 bytes:
"helloworld"

One blob. "helloworld", ten bytes, no seam. The reader has no way to know you meant two messages — the boundary between them existed only in your source code, and TCP threw it away. If those had been two JSON objects, {"a":1}{"b":2}, they would have arrived as {"a":1}{"b":2} with nothing marking where the first ends, and a naive serde_json::from_slice over the whole buffer would choke on the trailing content after the first object. Worse, on a busy network the split can land inside an object — {"a": in one read, 1}{"b":2} in the next — and now you are hand-writing a buffering parser that stitches partial reads back together. That job, "turn a byte stream back into the messages someone meant," is called framing, and you do not want to write it by hand.

The fix: prefix each frame with its length

There are a few classic ways to put boundaries back into a byte stream. You could pick a delimiter byte (like a newline) and agree that it never appears inside a message — but JSON contains newlines, so you would have to escape them, and now you have two encodings. The robust, standard answer is simpler: before each message, write its length. The reader reads a fixed-size length header first — say, four bytes — learns "the next N bytes are one message," reads exactly N more bytes (buffering across as many TCP reads as that takes), and hands you those N bytes as one complete frame. Then it does it again for the next message. Length-prefixing turns an undelimited stream back into a sequence of discrete, whole messages.

You will not write that reader. tokio-util ships it as LengthDelimitedCodec. A codec is a small object that knows how to turn bytes into frames and frames into bytes; LengthDelimitedCodec is the one that does exactly the length-prefix scheme just described. You wrap it around a TcpStream with Framed, and the pair gives you two superpowers: Framed is both a Stream of incoming frames (.next().await yields the next complete Bytes blob) and a Sink for outgoing frames (.send(bytes).await length-prefixes and writes one blob). The four-byte length header is added on send and stripped on receive; you never see it.

The toy: a message channel over a real socket

Here is the whole scheme working, in a domain with none of the coordinator's machinery. We define a tiny two-variant enum, Note, and a wrapper called NoteStream that owns a Framed<TcpStream, LengthDelimitedCodec> and offers exactly two methods: send(&Note) and recv() -> Option<Note>. send serializes the note to JSON bytes and pushes one frame; recv pulls one frame and deserializes it. This is a scaled-down MessageStream — the very type you build in the next chapter — and it is deliberately the same shape, method for method.

It needs tokio-util, futures, bytes, and serde, none of which are on the playground, so it is marked ignore. It is a real, runnable program — the output below is from actually running it against a loopback socket.

// scratch Cargo.toml deps:
//   tokio       = { version = "1", features = ["full"] }
//   tokio-util  = { version = "0.7", features = ["codec"] }
//   futures     = "0.3"
//   bytes       = "1"
//   serde       = { version = "1", features = ["derive"] }
//   serde_json  = "1"
use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::net::{TcpListener, TcpStream};
use tokio_util::codec::{Framed, LengthDelimitedCodec};

/// The toy message — a two-variant enum, self-describing on the wire via
/// serde's internal `"type"` tag (exactly how the real `Message` is tagged).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum Note {
    Hello { who: String },
    Ping { seq: u32 },
}

/// A bidirectional channel of `Note`s over one TCP connection. This is a
/// scaled-down `MessageStream`: `Framed` + `LengthDelimitedCodec` do the framing,
/// serde_json does the encoding, one method each way.
struct NoteStream {
    framed: Framed<TcpStream, LengthDelimitedCodec>,
}

impl NoteStream {
    fn new(stream: TcpStream) -> Self {
        Self {
            framed: Framed::new(stream, LengthDelimitedCodec::new()),
        }
    }

    async fn send(&mut self, note: &Note) -> std::io::Result<()> {
        let bytes = serde_json::to_vec(note).expect("serialize");
        self.framed.send(Bytes::from(bytes)).await
    }

    /// `None` means the peer closed the connection cleanly.
    async fn recv(&mut self) -> Option<Note> {
        match self.framed.next().await {
            Some(Ok(frame)) => Some(serde_json::from_slice(&frame).expect("deserialize")),
            Some(Err(e)) => panic!("frame error: {e}"),
            None => None,
        }
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    // Echo server: read each framed message, send it straight back.
    tokio::spawn(async move {
        let (sock, _) = listener.accept().await.unwrap();
        let mut server = NoteStream::new(sock);
        while let Some(note) = server.recv().await {
            server.send(&note).await.unwrap();
        }
    });

    let sock = TcpStream::connect(addr).await.unwrap();
    let mut client = NoteStream::new(sock);

    // Two messages, sent back-to-back with no gap between them on the wire.
    let a = Note::Hello { who: "w1".into() };
    let b = Note::Ping { seq: 7 };
    client.send(&a).await.unwrap();
    client.send(&b).await.unwrap();

    // The codec still hands us exactly two whole messages, in order.
    let ra = client.recv().await.unwrap();
    let rb = client.recv().await.unwrap();
    println!("sent:    {a:?}");
    println!("sent:    {b:?}");
    println!("got back: {ra:?}");
    println!("got back: {rb:?}");
    assert_eq!(a, ra);
    assert_eq!(b, rb);
}

The two sends go out back-to-back — exactly the situation that produced one glued blob in the first program. Predict what recv returns this time, and how many times.

sent:    Hello { who: "w1" }
sent:    Ping { seq: 7 }
got back: Hello { who: "w1" }
got back: Ping { seq: 7 }

Two calls to recv, two whole messages, in order, each a fully-typed Note — even though the same two writes over a bare socket arrived as "helloworld". Nothing in this program buffers partial reads or hunts for boundaries; LengthDelimitedCodec does all of it under Framed. The length prefix it wrote before each JSON blob is what let the reader carve the stream back into Hello { who: "w1" } and Ping { seq: 7 }. That is the entire idea of the arc's wire: frame on the outside, JSON on the inside, one typed message per recv.

Where the errors go

In the toy, recv panics on a malformed frame — fine for a demo. The real MessageStream cannot panic a connection task, so it does the disciplined thing: both a serde failure and a framing/IO failure map to ControlError::Protocol(...), the retryable protocol-error variant you built in Part II. That choice is deliberate, and it is the same reasoning as the whole error taxonomy: a garbled frame or a dropped connection is a transient fault about this worker's link, not a claim that the job is bad — so it should be retryable, and the scheduler should get the chance to hand the job to someone else. A clean end-of-stream is different: recv returns Ok(None), meaning "the peer closed, no error," which the connection actor in the next concept chapter reads as "this worker is gone." Malformed is an error; closed is a None. Hold that distinction — the actor leans on it.

One-for-one: the toy ↔ the real thing

Everything here maps straight onto the control-core codec you build in the very next chapter, piece for piece:

• NoteStream (owns a Framed, offers send/recv) ↔ MessageStream — the real wrapper, Framed<TcpStream, LengthDelimitedCodec> inside, send(&Message) and recv() -> Option<Message> outside.
• Note (a small tagged enum) ↔ Message — the Register/Assign/Result/Heartbeat enum you already defined in Part II, tagged with the same #[serde(tag = "type")], finally traveling over a socket instead of only round-tripping in a unit test.
• serde_json::to_vec + from_slice per frame ↔ the same calls in MessageStream — one JSON document per length-delimited frame, both directions.
• the toy's recv panic on bad bytes ↔ ControlError::Protocol — the real code turns a serde or framing failure into the retryable protocol error instead of panicking.
• recv returning None on clean close ↔ Ok(None) from MessageStream::recv — the "peer closed cleanly" signal the connection actor watches for.

That last row is the seam between this chapter and the next: framing gives you whole messages, and "no more messages" (None) is how the socket tells you the worker left. The next build chapter turns NoteStream into MessageStream and proves it with a Message round-trip over a real loopback socket — the twelfth control-core test, and the first one that touches the network.

Questions to lock

Stop on each; the whole arc's wire rests here.

1. TCP guarantees your bytes arrive in order and intact. What does it specifically not guarantee, and why does that force you to add framing before you can send two messages in a row?
2. What are the two separate jobs that LengthDelimitedCodec and serde_json each do, and why is it worth keeping them as two layers rather than folding them into one?
3. Which two futures traits must you import to call .send() and .next() on a Framed, and what does the compiler error look like when you forget?
4. MessageStream::recv maps a malformed frame to ControlError::Protocol but returns Ok(None) on a clean close. Why is one an error and the other not — and what will the connection actor do with each?

Next: build MessageStream for real, and watch a Message survive a round trip over an actual socket.

Build: The MessageStream Codec

Maps to: Phase 6 (cluster). Kind: Build.

Objective

Add the wire to control-core. You already have the Message enum from Part II — defined, tested, but never yet sent anywhere. Here you build MessageStream: a thin wrapper over Framed<TcpStream, LengthDelimitedCodec> that offers send(&Message) and recv() -> Option<Message>, and prove it with a single test that ships a Message across a real loopback socket and gets it back byte-identical. This is the first code in the entire course that touches the network — and it is small on purpose, because LengthDelimitedCodec does the hard part.

Scaffold

Create:

• crates/control-core/src/codec.rs — the MessageStream struct, its new/send/recv, and its one test.

Edit:

• crates/control-core/src/lib.rs — add pub mod codec; and re-export MessageStream alongside the other public types.
• crates/control-core/Cargo.toml — four new dependencies.

New deps and why:

• tokio-util (features ["codec"]) — supplies Framed and LengthDelimitedCodec, the framing layer. This is the new crate the concept chapter demonstrated. ([dependencies].)
• futures — supplies the SinkExt/StreamExt extension traits whose .send() and .next() you call on a Framed. ([dependencies].)
• bytes — supplies Bytes, the type the codec's Sink accepts for an outgoing frame. ([dependencies].)
• tokio — add the net feature (for TcpStream/TcpListener) to the existing entry; macros/rt are already there from the seam build. ([dependencies], plus #[tokio::test] from dev.)

serde_json (encode/decode each frame) and pretty_assertions (the test's assert_eq!) are already in the manifest from earlier builds.

Expected result: cargo test -p control-core → 12 tests pass. The eleven from the Core arc are untouched; the new one is a_message_roundtrips_over_a_real_socket in codec.rs. The Core-arc build told you a twelfth test would arrive "when you build the framed codec that actually ships these frames" — this is that test, and the count reaches 12 now.

The spec (givens)

MessageStream

/// A bidirectional channel of `Message`s over one TCP connection.
pub struct MessageStream {
    framed: Framed<TcpStream, LengthDelimitedCodec>,
}

impl MessageStream {
    pub fn new(stream: TcpStream) -> Self { /* wrap in Framed + LengthDelimitedCodec */ }

    /// Encode a message as one length-prefixed JSON frame and send it.
    pub async fn send(&mut self, msg: &Message) -> Result<(), ControlError>;

    /// Read the next frame and decode it. `Ok(None)` means the peer closed cleanly.
    pub async fn recv(&mut self) -> Result<Option<Message>, ControlError>;
}

• new wraps the TcpStream with Framed::new(stream, LengthDelimitedCodec::new()). Nothing else — the codec is the whole configuration.
• send serializes msg with serde_json::to_vec, wraps the Vec<u8> in Bytes::from(...), and pushes it with self.framed.send(...).await. Both the serialize step and the sink step map any error to ControlError::Protocol(e.to_string()).
• recv pulls self.framed.next().await and matches three arms: Some(Ok(frame)) deserializes the frame bytes with serde_json::from_slice, mapping a serde error to ControlError::Protocol and returning Ok(Some(msg)); Some(Err(e)) (a framing/IO error) maps to Err(ControlError::Protocol(...)); None (clean end of stream) returns Ok(None).

The error mapping is the whole policy: a garbled frame or a broken link is retryable (Protocol), and a clean close is not an error at all (Ok(None)). Hold the two apart — the connection actor in the next chapter branches on exactly this difference.

→ Answer key

Concepts exercised

• Framed + LengthDelimitedCodec as the framing layer over a raw TcpStream.
• The futures::SinkExt/StreamExt extension traits behind .send()/.next().
• One JSON document per length-delimited frame via serde_json::to_vec/from_slice.
• Mapping both serde and framing failures onto the retryable ControlError::Protocol, and a clean close onto Ok(None).
• Driving two MessageStreams over a loopback TcpListener bound to 127.0.0.1:0.

The build loop (you drive)

Test — a_message_roundtrips_over_a_real_socket (in codec.rs, #[tokio::test])

1. Write the failing test. Bind a TcpListener to 127.0.0.1:0 (the OS picks a free port; read it back with local_addr()). tokio::spawn a server task that accept()s one connection, wraps it in a MessageStream, and echoes: recv().await one message and send it straight back. In the main task, TcpStream::connect(addr), wrap it in a MessageStream, send a Message::Register { worker_id: "w1".into(), capacity: 4 }, then recv and assert the echoed message equals the one you sent.
2. Predict: before you implement recv, what does the server task's recv().await return the instant the client task ends and its socket drops — Ok(Some(..)), Ok(None), or an Err? Tie your answer to the three match arms in the spec.
3. Run — it fails to compile (no MessageStream yet).
4. Implement MessageStream — new, send, recv — exactly as specified. Do not forget use futures::{SinkExt, StreamExt};, or the compiler will claim Framed has no send/next method.
5. Run green. Then run the whole suite — cargo test -p control-core — and confirm you are at 12 with the eleven Core-arc tests still green.
6. Commit.

Done when

cargo test -p control-core shows 12 green (the eleven Core-arc tests plus a_message_roundtrips_over_a_real_socket); a Message::Register survives the trip across a loopback socket and compares equal; you can say why send wraps its bytes in Bytes and why recv's clean-close arm returns Ok(None) rather than an error. Commit. control-core now has a wire — the next chapter puts a worker on the other end of it.

Concept: The Connection Actor — Multiplexing One Socket

Kind: Concept.

The last chapter gave you a MessageStream: a typed, framed channel over one socket. That is the pipe. This chapter is about what sits on top of the pipe, and it is the piece that makes the whole cluster work — one small task that owns the socket and turns it into something the scheduler can treat exactly like a local worker.

Start from the problem, because the shape of the solution falls straight out of it. A worker advertises a capacity — say four — meaning it can run four jobs at once. So the coordinator wants to have four jobs in flight over one connection simultaneously: send Assign for job A, then B, then C, then D, without waiting for A's Result to come back first. The four results will then arrive in whatever order the jobs happen to finish — maybe C, then A, then D, then B. That is the entire difficulty in one sentence: many jobs share one socket, and their answers come back interleaved and out of order. Whoever sent job A has to be handed A's result and not C's.

Why one task must own the socket

The instinct is to let each dispatch call just use the socket directly — write the Assign, then read until the matching Result shows up. That falls apart immediately, and it is worth seeing why, because the failure is the reason the actor pattern exists. A TcpStream (and the MessageStream around it) is not something two tasks can safely share for reading. If job A's task and job C's task both call recv on the same socket, the frames race: A's task might read C's result and C's task might read A's, or the length-prefix reads from the two tasks interleave and corrupt the stream entirely. Sharing a socket across concurrent readers is a data race dressed up as a logic bug.

The fix is a discipline: exactly one task touches the socket. Everyone else talks to that task instead of to the socket. That one task is the connection actor — an actor in the plain sense, a task that owns some state (here, the socket and a table of who is waiting) and is the only thing allowed to mutate it. Other tasks send it requests over a channel; it serializes all socket access through its own single loop. No sharing, no races, and — the payoff — the actor is the one place that sees every frame come back, so it is the natural place to match each result to its waiter.

The three moving parts

The actor needs three things, and each answers one piece of the problem.

An mpsc receiver — the inbox of jobs to send. Many dispatch callers, one actor: that is a multi-producer, single-consumer channel by definition. Each caller sends the actor a job plus a private oneshot::Sender to answer on. The actor pulls jobs from this receiver and writes each as an Assign frame.

A oneshot per job — the private answer line. A oneshot channel carries exactly one value, once. When a caller dispatches a job, it makes a oneshot, keeps the receiver, and hands the sender to the actor. The caller then simply .awaits its receiver — it is parked until its answer arrives, and it is structurally impossible for it to be woken by someone else's result. This is the mechanism that solves out-of-order delivery: the ordering does not matter because each waiter has its own dedicated channel.

A HashMap<JobId, oneshot::Sender> — the table of who is waiting. When the actor sends Assign for a job, it stores that job's oneshot sender under the job's id. When a Result frame comes back, the actor looks up outcome.job_id in the map, removes the sender, and fires the result down it — waking exactly the right waiter. The map is the multiplexer: id in, correct waiter out.

And the loop that ties them together is a select! over two events: a new job arriving on the mpsc, or a frame arriving on the socket. One subtlety makes select! do real work here — a guard on the job arm:

tokio::select! {
    // Only pull new work while we are below the worker's capacity.
    maybe = jobs.recv(), if pending.len() < capacity => { /* send Assign, insert into map */ }
    frame = conn.recv() => { /* match Result back to its waiter, or handle close */ }
}

The if pending.len() < capacity is backpressure, for free. While the worker already holds capacity jobs, the actor stops selecting the job arm entirely — it will not pull another job off the mpsc. The mpsc then fills to its bound and dispatch's send blocks, which means the scheduler blocks trying to hand out more work to this worker. The worker's advertised capacity becomes a real, enforced limit, propagated all the way back to the scheduler through nothing but a select! guard and a bounded channel. No counter to decrement by hand, no semaphore — the pending.len() map and the guard are the whole mechanism.

The toy: a mini actor that matches replies by id

Here is exactly that shape, with the socket replaced by a pair of in-process channels so it runs anywhere. jobs is the mpsc inbox; conn_tx/conn_rx stand in for MessageStream's send and recv; pending is the id→waiter map; and the select! has the capacity guard. Two scenarios run: three jobs whose replies deliberately come back out of order (the worker answers job 2 before job 1), and a worker that dies mid-job so you can watch the cleanup path fire.

This is plain tokio — mpsc, oneshot, select! — so it runs as-is. The output below is real.

use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};

type JobId = u64;

/// The far side's frames, mirrored down to the one that matters here.
enum Frame {
    Result { job_id: JobId, answer: String },
}

/// A job handed to the actor plus the one-shot to answer it on. This pair is the
/// shape of the real `Dispatch = (Job, oneshot::Sender<Result<JobOutcome, _>>)`.
type Dispatch = (JobId, String, oneshot::Sender<Result<String, String>>);

/// One task owns the "connection" and multiplexes many in-flight jobs. `conn_rx`
/// stands in for `MessageStream::recv`; `conn_tx` for `conn.send(Assign)`.
async fn connection_actor(
    mut jobs: mpsc::Receiver<Dispatch>,
    mut conn_rx: mpsc::Receiver<Frame>,
    conn_tx: mpsc::Sender<(JobId, String)>,
    capacity: usize,
) {
    let mut pending: HashMap<JobId, oneshot::Sender<Result<String, String>>> = HashMap::new();

    loop {
        tokio::select! {
            // Only pull new work while below capacity — this is the backpressure.
            maybe = jobs.recv(), if pending.len() < capacity => {
                let Some((job_id, text, reply)) = maybe else { break };
                if conn_tx.send((job_id, text)).await.is_err() {
                    let _ = reply.send(Err("connection lost".into()));
                    break;
                }
                pending.insert(job_id, reply);
            }
            frame = conn_rx.recv() => {
                match frame {
                    Some(Frame::Result { job_id, answer }) => {
                        if let Some(reply) = pending.remove(&job_id) {
                            let _ = reply.send(Ok(answer)); // wake exactly this waiter
                        }
                    }
                    None => break, // the connection closed
                }
            }
        }
    }

    // Connection finished: fail every job still in flight, RETRYABLY, so the
    // caller re-dispatches it elsewhere. This is the crucial line.
    for (job_id, reply) in pending {
        println!("actor: failing in-flight job {job_id} retryably");
        let _ = reply.send(Err("worker connection lost".into()));
    }
}

#[tokio::main]
async fn main() {
    // --- scenario 1: three jobs, replies matched by id (out of order) ---
    {
        let (jobs_tx, jobs_rx) = mpsc::channel::<Dispatch>(8);
        let (assign_tx, mut assign_rx) = mpsc::channel::<(JobId, String)>(8);
        let (result_tx, result_rx) = mpsc::channel::<Frame>(8);

        // A fake worker: collect all three assigns, then answer 2, 1, 3 — proving
        // replies are matched by id, not by the order they come back.
        tokio::spawn(async move {
            let mut seen = Vec::new();
            while let Some((id, text)) = assign_rx.recv().await {
                seen.push((id, text));
                if seen.len() == 3 {
                    for &(id, ref text) in [&seen[1], &seen[0], &seen[2]] {
                        result_tx
                            .send(Frame::Result { job_id: id, answer: format!("re: {text}") })
                            .await
                            .unwrap();
                    }
                }
            }
        });

        tokio::spawn(connection_actor(jobs_rx, result_rx, assign_tx, 4));

        let mut waiters = Vec::new();
        for (id, text) in [(1u64, "alpha"), (2, "beta"), (3, "gamma")] {
            let (reply_tx, reply_rx) = oneshot::channel();
            jobs_tx.send((id, text.into(), reply_tx)).await.unwrap();
            waiters.push((id, reply_rx));
        }
        for (id, rx) in waiters {
            println!("job {id} -> {:?}", rx.await.unwrap());
        }
    }

    println!("---");

    // --- scenario 2: worker dies mid-job -> the pending job fails retryably ---
    {
        let (jobs_tx, jobs_rx) = mpsc::channel::<Dispatch>(8);
        let (assign_tx, mut assign_rx) = mpsc::channel::<(JobId, String)>(8);
        let (result_tx, result_rx) = mpsc::channel::<Frame>(8);

        // Worker takes the assign, then vanishes without answering (a crash).
        tokio::spawn(async move {
            let _ = assign_rx.recv().await;
        });
        drop(result_tx); // the actor's conn_rx now sees the close

        tokio::spawn(connection_actor(jobs_rx, result_rx, assign_tx, 4));

        let (reply_tx, reply_rx) = oneshot::channel();
        jobs_tx.send((99, "doomed".into(), reply_tx)).await.unwrap();
        println!("job 99 -> {:?}", reply_rx.await.unwrap());
    }
}

Predict two things before you read the output. First: in scenario 1, does job 1 print re: alpha even though the worker answered job 2 first? Second: in scenario 2, what does job 99 receive when the worker dies without answering?

job 1 -> Ok("re: alpha")
job 2 -> Ok("re: beta")
job 3 -> Ok("re: gamma")
---
actor: failing in-flight job 99 retryably
job 99 -> Err("worker connection lost")

Every waiter got its own answer despite the scrambled reply order — because each waited on a private oneshot and the actor routed by id through the map. And the doomed job did not hang forever: when conn_rx returned None, the loop broke, and the cleanup pass fired an error into every still-pending waiter. job 99 unblocked with an error rather than deadlocking.

The most important line: failing pending jobs on close

That cleanup loop is the pivot of the entire cluster arc, so slow down on it. When the socket closes — the worker crashed, the network dropped, the process was killed — the actor's recv returns None (a clean close) or an Err (a broken frame), the loop breaks, and control reaches the final for over pending. Every job the worker accepted but never answered is sitting in that map. The actor sends each one an error.

The kind of error is everything. In the real RemoteWorker, that error is ControlError::Worker(...) — a retryable error. Recall from Part II that dispatch returning a retryable error is precisely the signal the scheduler acts on: it hands the job to a different worker. So "the socket closed" becomes, through this one loop, "redeliver every in-flight job elsewhere." A worker dying mid-job does not lose the job — it releases it back to the scheduler. That is at-least-once delivery, and it is born right here, in the difference between failing a pending job retryably and letting it hang.

One-for-one: the toy ↔ the real thing

Every part of the toy maps onto the real connection_actor and RemoteWorker you build next:

• connection_actor (toy) ↔ connection_actor (real) — same loop, same select! with the capacity guard; the real one selects over a MessageStream instead of a stand-in mpsc, and also grows a heartbeat-timeout arm (second half of the arc).
• Dispatch = (JobId, String, oneshot::Sender<...>) ↔ Dispatch = (Job, oneshot::Sender<Result<JobOutcome, ControlError>>) — the job plus its private answer line, posted to the actor's inbox.
• the mpsc job inbox ↔ RemoteWorker's tx: mpsc::Sender<Dispatch> — RemoteWorker::dispatch makes a oneshot, sends (job, sender) to the actor, and awaits the receiver. The whole round-trip hides behind one WorkerHandle::dispatch.
• pending: HashMap<JobId, oneshot::Sender> ↔ the identical map in the real actor — id in, correct waiter out; the multiplexer itself.
• conn_tx.send(Assign) / conn_rx.recv() ↔ conn.send(&Message::Assign { job }) / conn.recv() — the real socket, via MessageStream.
• if pending.len() < capacity ↔ the same guard — backpressure that throttles the scheduler through the bounded mpsc.
• the final for failing pending jobs ↔ the real cleanup firing ControlError::Worker into each pending oneshot — retryable, so the scheduler redelivers. The at-least-once seam.

Hold that last row against the promise the whole course opened with. The scheduler still just calls dispatch on a dyn WorkerHandle and gets a Result — it never learns there is a socket, an actor, a map, or a redelivery underneath. The connection actor is the machine that makes a remote worker indistinguishable from a local one, and it fits behind the same seam because the seam was built for exactly this.

Questions to lock

Stop on each; the next build assembles precisely this.

1. Why can't two dispatch calls just share the socket and each read until their own result shows up? What specifically goes wrong, and how does routing everything through one actor task fix it?
2. What job does the per-job oneshot do that the shared mpsc inbox cannot? Why does giving each waiter its own channel make out-of-order results a non-problem?
3. Trace the backpressure: how does if pending.len() < capacity on the select! job arm end up throttling the scheduler? Name every link in the chain.
4. When the socket closes, the actor fails every pending job with a retryable error. Why retryable specifically, and what does the scheduler do as a result? What property of the whole system is that the beginning of?

Next: build it for real — RemoteWorker, the connection_actor over a MessageStream, the SharedPool, the accept loop, and the worker binary on the far end.

Build: RemoteWorker and the Worker Binary

Maps to: Phase 6 (cluster). Kind: Build.

Objective

Close the seam. You have a wire (MessageStream) and, from the concept chapter, the shape of the machine that drives it. Here you build both ends: on the coordinator, a RemoteWorker that is a WorkerHandle like any other, a connection_actor that multiplexes one socket, a SharedPool the scheduler snapshots, and an accept loop that turns each incoming connection into a registered worker. On the far side, the panoptes-worker binary that connects, registers, and serves eval jobs. When this compiles green, the scheduler can dispatch a job to a process on another machine and never know the difference — the promise Part I made, paid in full.

This is the largest build in the course, and it spans two crates. Take it one test at a time; each named test locks one piece.

Scaffold

Create — coordinator side (panoptes-control):

• crates/panoptes-control/src/remote.rs — RemoteWorker, SharedPool, connection_actor, serve_workers, handle_connection, and two tests.

Edit:

• crates/panoptes-control/src/lib.rs — add pub mod remote; and re-export SharedPool, serve_workers.
• crates/panoptes-control/Cargo.toml — no new external deps; tokio (with net), async-trait, and control-core are already present. The tokio::sync::{mpsc, oneshot} primitives are part of tokio's sync/full features you already enabled.

Create — worker side (new crate panoptes-worker; add crates/panoptes-worker to the workspace members):

• crates/panoptes-worker/Cargo.toml
  • [dependencies]: control-core, control-eval (both { path = ... }), plus tokio (features ["full"]), async-trait, anyhow, clap (features ["derive"]), tracing, tracing-subscriber (features ["env-filter"]) — all { workspace = true } where the workspace pins them.
    • control-eval gives you ModelClient, HttpModelClient, and run_eval — the workload the worker runs.
    • clap parses the binary's flags; tracing-subscriber sets up logging in main.
  • [dev-dependencies]: async-trait, tokio (for the StubClient and #[tokio::test]).
• crates/panoptes-worker/src/lib.rs — run_session, serve_worker, run_worker_forever, DEFAULT_HEARTBEAT, and one test.
• crates/panoptes-worker/src/main.rs — the binary: Cli, #[tokio::main], wire up run_worker_forever.

Expected result:

• cargo test -p panoptes-control → the scheduler-arc tests still pass, plus 2 new: remote_worker_dispatches_over_the_wire, worker_death_makes_dispatch_retryable.
• cargo test -p panoptes-worker → 1 test: worker_registers_runs_a_job_and_returns_a_result.

The spec (givens)

RemoteWorker and SharedPool

/// A job handed to the connection actor plus the channel to answer it on.
type Dispatch = (Job, oneshot::Sender<Result<JobOutcome, ControlError>>);

/// Coordinator-side handle to a worker across a TCP connection. To the scheduler
/// it is an ordinary `WorkerHandle`; `dispatch` hides the round-trip.
pub struct RemoteWorker {
    id: String,
    tx: mpsc::Sender<Dispatch>,
}

#[async_trait]
impl WorkerHandle for RemoteWorker {
    fn id(&self) -> &str;
    async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError>;
}

/// A live, shared set of workers. The accept loop adds one on register and
/// removes it on disconnect; the scheduler snapshots it per run. Cheap to clone.
#[derive(Clone, Default)]
pub struct SharedPool {
    workers: Arc<Mutex<Vec<Arc<dyn WorkerHandle>>>>,
}

• RemoteWorker::dispatch is the whole illusion: make a oneshot, self.tx.send((job, reply_tx)).await, then reply_rx.await. Both failure paths map to ControlError::Worker(...) — a send error means the actor (and thus the connection) is already gone; a dropped reply_rx means the actor died with the job in flight. Both are retryable, which is exactly what makes the scheduler redeliver. Note the double ?/unwrap shape: reply_rx.await yields Result<Result<JobOutcome, ControlError>, RecvError> — the outer error is the dropped channel, the inner is the actor's own answer.
• SharedPool needs add(Arc<dyn WorkerHandle>), remove(&str) (retain by id()), len(), is_empty(), and an impl WorkerSource whose snapshot(&self) -> Vec<Arc<dyn WorkerHandle>> clones the inner vec. WorkerSource is the scheduler-arc trait — this is the second implementation of it, the remote counterpart to the local pool, and the scheduler cannot tell them apart.

→ Answer key

connection_actor

async fn connection_actor(
    mut conn: MessageStream,
    mut jobs: mpsc::Receiver<Dispatch>,
    capacity: usize,
);

Own the socket; multiplex it. Keep a HashMap<JobId, oneshot::Sender<Result<JobOutcome, ControlError>>> of in-flight jobs. Loop over a tokio::select! of two arms:

• maybe = jobs.recv(), if pending.len() < capacity — the guarded arm. On Some((job, reply)): capture job.id, send Message::Assign { job } over conn; if that send fails, answer reply with the error and break (the wire is broken); otherwise insert reply into pending under the job id. On None (all RemoteWorker handles dropped), break.
• frame = conn.recv() — the socket arm. Match: Ok(Some(Message::Result { outcome })) → remove outcome.job_id from pending and send Ok(outcome) to its waiter; Ok(Some(Message::Heartbeat { .. })) → ignore (liveness only); Ok(Some(_)) → ignore (an inbound Register/Assign is a protocol slip, not fatal); Ok(None) | Err(_) → break (closed or malformed).

After the loop, the crucial cleanup: drain pending and send each waiter Err(ControlError::Worker("worker connection lost".into())). Retryable, so every job the dead worker was holding gets redelivered. This is the line the concept chapter called the pivot of the arc.

→ Answer key

serve_workers and handle_connection

/// Accept worker connections forever, registering each into `pool`.
pub async fn serve_workers(listener: TcpListener, pool: SharedPool);

/// Register one connection and run its actor until the socket closes.
async fn handle_connection(sock: TcpStream, pool: SharedPool);

• serve_workers loops on listener.accept(); for each socket, clone the pool and tokio::spawn(handle_connection(sock, pool)). An accept error should not kill the loop — continue.
• handle_connection runs the handshake: wrap the socket in a MessageStream, recv the first frame, and require it to be Message::Register { worker_id, capacity }. Anything else (or a closed/errored socket) → return, dropping the connection. On a valid register: make mpsc::channel(capacity) (clamp capacity to at least 1), build Arc::new(RemoteWorker { id: worker_id, tx }), pool.add(...), run connection_actor(conn, jobs, capacity).await, and when it returns, pool.remove(&worker_id). Add-then-actor-then-remove is the worker's whole lifecycle in the pool.

→ Answer key

The worker binary — run_session, serve_worker, run_worker_forever, main

pub const DEFAULT_HEARTBEAT: Duration = Duration::from_secs(5);

/// Register, then serve jobs until the coordinator closes (`Ok`) or the wire fails (`Err`).
pub async fn run_session(
    mut conn: MessageStream,
    worker_id: &str,
    capacity: u32,
    client: &dyn ModelClient,
    heartbeat: Duration,
) -> Result<(), ControlError>;

/// Connect to the coordinator and run one session.
pub async fn serve_worker(
    coordinator: &str,
    worker_id: &str,
    capacity: u32,
    client: Arc<dyn ModelClient>,
    heartbeat: Duration,
) -> Result<(), ControlError>;

/// Serve forever, reconnecting after any disconnect.
pub async fn run_worker_forever(
    coordinator: &str,
    worker_id: &str,
    capacity: u32,
    client: Arc<dyn ModelClient>,
    heartbeat: Duration,
    reconnect_delay: Duration,
);

• run_session first sends Message::Register { worker_id, capacity }. Then it sets up a tokio::time::interval(heartbeat) and skips the immediate first tick. Then it loops a select! of two arms: beat.tick() → send Message::Heartbeat { worker_id }; conn.recv() → match Some(Message::Assign { job }) (capture job.id, run_eval(client, &job.spec).await?, then send Message::Result { outcome: JobOutcome { job_id, records } }), Some(_) (ignore — the coordinator only ever sends Assign), None (clean close → return Ok(())). A failed eval ?-propagates and ends the session; the coordinator will notice the dropped connection and redeliver. The worker is deliberately simple: no map, no multiplexing — that all lives on the coordinator.
• serve_worker connects a TcpStream, wraps it in MessageStream::new, and calls run_session.
• run_worker_forever loops serve_worker forever, logging the outcome (Ok → coordinator closed; Err → session failed) and sleeping reconnect_delay between attempts. Workers are cattle; supervision is just a loop.
• main is #[tokio::main]: parse a clap Cli (a --coordinator address, --model-api base URL, --model, --id, --capacity), init a tracing_subscriber with an EnvFilter (default info), build Arc::new(HttpModelClient::new(model_api, model)), log a startup line, and call run_worker_forever(..., DEFAULT_HEARTBEAT, Duration::from_secs(1)).

→ Answer key

Concepts exercised

• A second impl WorkerHandle (RemoteWorker) and a second impl WorkerSource (SharedPool) — the distributed layer arriving as new implementations of existing seams, the scheduler untouched.
• The connection actor: select! with a capacity guard, an id→waiter HashMap, and retryable failure of all pending jobs on close.
• A TCP accept loop that spawns one task per connection and a register handshake that gates the connection.
• The worker session loop: interleaving heartbeats with job service over one MessageStream via select!.
• A supervised reconnect loop and a clap binary that composes the whole worker.

The build loop (you drive)

Test 1 — remote_worker_dispatches_over_the_wire (in remote.rs, #[tokio::test])

1. Write the failing test. Stand up a coordinator: bind 127.0.0.1:0, make a SharedPool, spawn serve_workers(listener, pool.clone()). Spawn a fake worker task: connect, send Register { worker_id: "w1", capacity: 2 }, then loop recv and for each Assign { job } build one ResponseRecord per vignette and send back Result { outcome: JobOutcome { job_id: job.id, records } }. Poll the pool until it holds 1 worker, snapshot().pop() it, dispatch(a_job()).await.unwrap(), and assert the outcome has one record.
2. Predict: the fake worker answers on the same socket the actor is reading. Which task inserts into pending, and which removes from it? Trace the one job's oneshot from dispatch to the assert.
3. Run — fails to compile (nothing built yet).
4. Implement RemoteWorker, SharedPool, connection_actor, serve_workers, handle_connection.
5. Run green, commit.

Test 2 — worker_death_makes_dispatch_retryable (in remote.rs, #[tokio::test])

1. Write the failing test. Same coordinator. This fake worker registers (capacity: 1), receives its one Assign, then drops the socket without answering — a crash mid-job. Poll for 1 worker, snapshot-pop it, dispatch(a_job()).await.unwrap_err(), and assert err.is_retryable().
2. Predict: the worker never sends a Result, so the pending entry is never removed by the socket arm. What removes it, and what error does the waiter's oneshot receive? Which cleanup path in connection_actor fires?
3. Run, confirm the assertion pins the retryable error.
4. Run green, commit.

Test 3 — worker_registers_runs_a_job_and_returns_a_result (in panoptes-worker, #[tokio::test])

1. Write the failing test. This time the coordinator is faked and the worker is real. Define a StubClient: ModelClient (no network — generate returns ModelResponse { text: format!("re: {prompt}"), usage: .. }). Bind 127.0.0.1:0; spawn a fake coordinator that accepts, asserts the first frame is Register, sends one Assign { job } (a job with one vignette "a"/"hi"), then loops recv ignoring heartbeats until it sees Result { outcome } and asserts outcome.job_id matches, one record, records[0].response == "re: hi". In the main task, run serve_worker(addr, "w1", 1, Arc::new(StubClient), Duration::from_secs(30)) under a tokio::time::timeout (it serves forever, so time it out).
2. Predict: the coordinator sends one Assign and then never closes. Why does wrapping serve_worker in a timeout matter — what would happen without it, and is that a bug or expected?
3. Run — fails to compile (worker crate not built).
4. Implement run_session, serve_worker, run_worker_forever, and main.rs.
5. Run green, commit.

Done when

cargo test -p panoptes-control shows the scheduler-arc tests plus remote_worker_dispatches_over_the_wire and worker_death_makes_dispatch_retryable green; cargo test -p panoptes-worker shows worker_registers_runs_a_job_and_returns_a_result green; a whole-workspace cargo test is green; and you can trace one job from RemoteWorker::dispatch, through the actor's Assign, across the socket to the worker's run_eval, back as a Result, and out the job's oneshot — and say why a worker's death turns that same path into a retryable error. Commit. The scheduler now dispatches across machines through the exact seam it has always used.

The cluster is wired but not yet safe: a frozen worker that never closes its socket would stall a job forever, and a redelivered job could be recorded twice. Those are the capstone — heartbeat reaping and at-least-once delivery made safe by the store's idempotency — where the whole course pays off.

Concept: At-Least-Once, Heartbeats, and Idempotency

Kind: Concept (read, do not code). This is the chapter that arms the payoff — the last idea before the capstone.

You have built almost the whole cluster. The framed codec turns a byte stream into whole messages; the connection actor multiplexes one socket, matching Result frames back to the jobs that are waiting on them; the RemoteWorker hides all of that behind the same WorkerHandle the scheduler has dispatched through since Part VI. On a good day, a job goes out as an Assign frame and its answer comes back as a Result frame, and nobody upstream can tell the worker was on another machine.

This chapter is about the bad day — the one the entire course was designed around. A worker dies mid-job. Hold the sentence from the introduction in view: under worker failure, no eval is lost and none is double-counted. Those two properties pull against each other, and this chapter is where the tension finally resolves. To lose nothing, the coordinator must be willing to run a job again somewhere else. But running it again means it can genuinely run twice. The escape is that the store was built — back in Part IV — so that recording the same job's outcome twice is indistinguishable from recording it once. This chapter earns that claim.

We proceed in the order the ideas depend on each other: first how you even notice a worker died (it is harder than it sounds), then what you do about it (redeliver), then why that is safe (idempotency).

Why a clean close is not the only failure

Start with the failure you already handle. When a worker process exits — a panic, a kill, a clean shutdown — its TCP socket closes. On the coordinator side, conn.recv() returns Ok(None): end of stream. The connection actor sees that, breaks its loop, and fails every job still in flight with a retryable error. You built this in the previous chapter, and its test (worker_death_makes_dispatch_retryable) is green. A clean close is the easy death, because the operating system tells you about it.

Now the hard death. A worker's process freezes — it deadlocks, its event loop wedges, the box swaps itself to a standstill, or the network path silently drops every packet without ever delivering a FIN. The socket is still open from the coordinator's point of view. Nothing closes. conn.recv() does not return Ok(None); it does not return anything at all. It just blocks, forever, waiting for a frame that will never come. This is a half-open connection: one side considers the link alive and is waiting to read, while the other side is gone or catatonic and will never write.

If the connection actor only ever broke on Ok(None), a frozen worker would pin its in-flight jobs forever. Those jobs would never fail, so the scheduler would never redeliver them, so the run would never finish. "No eval is lost" would be violated in the most insidious way possible: nothing errors, the system just quietly stops making progress. A silent hang is worse than a loud crash.

The fix: define silence, then time it out

Since the transport won't tell you the worker is dead, you decide what "dead" means, and the honest definition is behavioural: a worker that has said nothing for too long is presumed dead. Not "disconnected" — silent. You pick a window — call it the heartbeat timeout — and if a whole window passes with no frame of any kind arriving on that connection, you reap it: break the loop, fail the in-flight jobs retryably, exactly as if it had closed cleanly. The half-open hang becomes an ordinary retryable failure, and the redelivery machinery you already have takes over.

In the connection actor this is one more branch in the select!. Alongside "a new job to dispatch" and "a frame arrived from the worker," you add "the inactivity timer fired." The subtle, load-bearing detail is that the timer is created fresh at the top of every loop iteration, so any activity — a job going out, a result coming back, a heartbeat — resets the clock by starting the next iteration with a brand-new timer. Only genuine silence for the entire window lets the timer win the select!. That is the whole reaper, and you will meet its toy below.

Heartbeats: filling the silence on purpose

There is an obvious problem with "silent for a window means dead": a healthy worker can be silent. If it is sitting idle with no jobs assigned, it has nothing to say, and it would be absurd to reap a perfectly good worker just because the coordinator handed it no work. Worse, even a busy worker on a long job might not produce a Result frame for a while.

The fix is that the worker takes responsibility for its own liveness. When it has nothing else to send, it periodically sends a heartbeat — a tiny frame whose only purpose is to prove "still here." The coordinator does not need to do anything with a heartbeat's contents; the mere fact that a frame arrived resets the inactivity timer, which is the entire point. A heartbeat is silence-insurance.

This is a rate relationship, and getting it right is the whole game. The worker's heartbeat interval must be comfortably shorter than the coordinator's inactivity timeout — several beats should fit inside one window. If a worker beats every 10 seconds against a 30-second timeout, then losing one or two beats to a slow network is harmless; it takes a sustained, three-window silence to trip the reaper. Make the interval too close to the timeout and you get false reaps — you kill live workers over ordinary jitter. Make the timeout enormous and you are slow to notice real deaths. The gap between "how often I promise to speak" and "how long you'll wait before giving up on me" is your tolerance for jitter.

At-least-once: what reaping does

Reaping is only worth anything because of what happens to the reaped worker's jobs. When the connection actor breaks — whether from a clean close, a malformed frame, or the inactivity timeout — it does one crucial thing on the way out: it drains its pending map and sends every waiting job a retryable ControlError::Worker. Each RemoteWorker::dispatch that was blocked awaiting a reply now wakes up with that error.

And a retryable error is precisely the signal the scheduler was built to act on, back in Part VI. The scheduler dispatched the job; the dispatch returned Err(e) with e.is_retryable() == true; so the scheduler puts the job back and hands it to another worker from the pool. The job that the dying worker never finished gets redelivered to a survivor. That is at-least-once delivery: the coordinator guarantees each job runs at least once by being willing to re-run any job whose worker did not confirm completion.

Notice the honesty in "at least." The coordinator cannot know why a dispatch failed. Maybe the worker died before touching the job — then redelivery is the only way the job ever runs. But maybe the worker actually finished the job and died (or was reaped) before its Result frame made it back across the wire. Or maybe it was merely slow, got reaped as "silent," and is still grinding away — and now a survivor is running the same job in parallel. From the coordinator's side these are indistinguishable: all it saw was a dispatch that did not return a confirmed outcome. So it must assume the job might not have run — and redeliver. The unavoidable cost of never losing a job is that a job can genuinely execute twice.

Idempotency: what makes the duplicate harmless

So a job can arrive at the store twice. The property that makes this a non-event is idempotency: an operation is idempotent when applying it twice has the same effect as applying it once. Recording a job outcome must be idempotent — the second recording of the same job must not advance the run's progress a second time.

You already built this, in Part IV, and it is worth revisiting why it was shaped the way it was — because this chapter is the reason. The store's record_job_outcome does not blindly bump the run's done-count. It first tries to move this specific job from not-done to done, and it advances the run only when that transition actually happened for the first time. A redelivered outcome finds the job already marked done, the transition is a no-op, and the done-count is left untouched. The same records land, but the run's progress moves exactly once. (Revisit the guard and its test at the idempotent-recording build step.)

This is the sentence the whole course has been walking toward, and now every clause is load-bearing: at-least-once delivery, made safe by idempotent recording. At-least-once (redeliver on any unconfirmed failure) guarantees no eval is lost. Idempotent recording (advance only on a job's first landing) guarantees none is double-counted. Neither property alone is enough; the pair is the payoff. Heartbeats and the reaper are what turn a silent half-open death into the "unconfirmed failure" that triggers redelivery in the first place — without them, the whole chain never starts.

Toy (a): a select! inactivity-timeout loop that reaps a silent peer

Here is the reaper, distilled to channels and timers — no sockets, so it runs anywhere. A watch task loops on a select! between "a beat arrived" and "the inactivity timer fired." The timer is rebuilt at the top of every iteration, so each beat resets the clock; only a full window of silence lets the timer win. This is the connection actor's heartbeat branch with the networking stripped away.

use std::time::Duration;
use tokio::sync::mpsc;

async fn watch(mut beats: mpsc::Receiver<&'static str>, timeout: Duration) -> &'static str {
    loop {
        // A fresh timer each iteration: any frame resets the clock, so only
        // genuine silence for the whole window trips it.
        let idle = tokio::time::sleep(timeout);
        tokio::select! {
            frame = beats.recv() => {
                match frame {
                    Some(b) => println!("saw {b} — clock reset"),
                    None => { println!("channel closed"); return "closed"; }
                }
            }
            _ = idle => {
                println!("silent past {timeout:?} — reaped");
                return "reaped";
            }
        }
    }
}

#[tokio::main]
async fn main() {
    // Peer A beats twice inside the window, then goes silent → reaped.
    let (tx, rx) = mpsc::channel(4);
    tokio::spawn(async move {
        for b in ["beat-1", "beat-2"] {
            tokio::time::sleep(Duration::from_millis(30)).await;
            let _ = tx.send(b).await;
        }
        // then freeze: hold tx, never send again
        tokio::time::sleep(Duration::from_secs(5)).await;
        drop(tx);
    });
    let verdict = watch(rx, Duration::from_millis(80)).await;
    println!("verdict A = {verdict}");

    // Peer B beats every 30ms, comfortably inside 80ms → stays alive.
    let (tx, rx) = mpsc::channel(4);
    let h = tokio::spawn(async move {
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_millis(30)).await;
            if tx.send("beat").await.is_err() {
                return;
            }
        }
    });
    // Watch only long enough to prove it survives well past one window.
    let verdict =
        tokio::time::timeout(Duration::from_millis(200), watch(rx, Duration::from_millis(80)))
            .await
            .unwrap_or("still-alive");
    h.abort();
    println!("verdict B = {verdict}");
}

Predict the two verdicts before you read on. Peer A sends two beats 30ms apart — each well inside the 80ms window, each resetting the clock — then freezes; 80ms of silence later it is reaped. Peer B keeps beating every 30ms, so the timer never wins and watch never returns; the outer timeout gives up first and we call it still-alive. Real output:

saw beat-1 — clock reset
saw beat-2 — clock reset
silent past 80ms — reaped
verdict A = reaped
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
verdict B = still-alive

The two branches of the select! are exactly the coordinator's two questions about a worker: did you send me something? (reset the clock) and have you been silent too long? (reap). Heartbeats exist only to keep the answer to the first question "yes."

Toy (b): an idempotent counter that advances only on a key's first arrival

Now the other half — the thing that makes redelivery safe. This counter records outcomes keyed by job id and advances its progress only the first time a given id lands. A HashSet::insert returns true only when the key was new, which is the in-memory twin of the store's UPDATE ... WHERE status != 'done' affecting exactly one row.

use std::collections::HashSet;

struct Ledger {
    seen: HashSet<u64>, // which job ids have already landed
    done_count: u64,    // the run's progress
}

impl Ledger {
    fn new() -> Self {
        Ledger { seen: HashSet::new(), done_count: 0 }
    }

    /// Record an outcome. Advance the count ONLY the first time this id lands;
    /// a redelivered duplicate is recorded but does not double-count.
    fn record(&mut self, job_id: u64) {
        // `insert` returns true only if the id was NOT already present — the
        // in-memory twin of `UPDATE ... WHERE status != 'done'` affecting 1 row.
        if self.seen.insert(job_id) {
            self.done_count += 1;
            println!("job {job_id}: first landing — done_count = {}", self.done_count);
        } else {
            println!("job {job_id}: duplicate — ignored, done_count = {}", self.done_count);
        }
    }
}

fn main() {
    let mut ledger = Ledger::new();
    // Jobs 1 and 2 land once each; job 1 is REDELIVERED twice (a slow worker was
    // not dead, or a survivor re-ran it after a reap). At-least-once, made safe.
    ledger.record(1);
    ledger.record(2);
    ledger.record(1); // redelivery
    ledger.record(1); // another redelivery
    assert_eq!(ledger.done_count, 2, "two distinct jobs, counted once each");
    println!("final done_count = {} (not 4)", ledger.done_count);
}

Predict the final count: four record calls, but only two distinct ids, so the run advances exactly twice. Real output:

job 1: first landing — done_count = 1
job 2: first landing — done_count = 2
job 1: duplicate — ignored, done_count = 2
job 1: duplicate — ignored, done_count = 2
final done_count = 2 (not 4)

Four recordings, done_count == 2. That is idempotency in one method: the operation is safe to apply as many times as redelivery demands, because every application after the first is a no-op on the thing that matters. The real store enforces this transactionally in SQLite rather than with a HashSet, but the shape — decide firstness, act only on firstness — is identical.

One-for-one: the toy ↔ the real thing

• The watch loop's fresh sleep(timeout) per iteration ↔ connection_actor's let idle = tokio::time::sleep(heartbeat_timeout) at the top of the select! loop — a new timer each pass, so any frame resets the clock.
• The idle branch returning "reaped" ↔ the actor's _ = idle => break arm — genuine silence for the whole window ends the connection; the reap.
• The channel closing → "closed" ↔ conn.recv() returning Ok(None) → break — the clean death, distinct from the silent one but funnelled to the same exit.
• Peer B's beat loop ↔ the worker's heartbeat interval in run_session — periodic tiny frames whose only job is to reset the coordinator's clock.
• watch returning a verdict its caller observes ↔ the actor draining pending and sending each waiter a retryable ControlError::Worker — reaping produces the retryable error that triggers redelivery.
• Ledger::record advancing only when seen.insert is true ↔ record_job_outcome advancing done_count only when the WHERE status != 'done' update affected one row — advance on first landing only.
• Four record(1) calls leaving done_count == 2 ↔ recording_the_same_job_twice_counts_once — redelivery cannot double-count.

Hold the whole mapping against the payoff sentence. The reaper (toy a) converts a half-open freeze into a retryable failure; the scheduler turns that failure into a redelivery; the idempotent ledger (toy b) makes the redelivery harmless. At-least-once, made safe by idempotent recording — and next chapter you build both halves for real and prove them together.

Questions to lock

Genuinely stop on each. The capstone assumes you can answer all four cold.

1. Why isn't conn.recv() returning Ok(None) enough to detect every dead worker? Describe the failure it misses and what you add to catch it.
2. Why must the inactivity timer be recreated at the top of each loop iteration rather than once before the loop? What breaks if you build it once?
3. What is the required relationship between the worker's heartbeat interval and the coordinator's inactivity timeout, and what goes wrong at each extreme (interval too close to timeout; timeout enormous)?
4. State the payoff sentence and defend both halves: which mechanism guarantees "no eval is lost," which guarantees "none is double-counted," and why is neither sufficient alone?

Next: the capstone build — the heartbeat timeout in connection_actor, the worker's own heartbeat, and the integration test that kills a worker mid-run and proves no eval is lost and none double-counted.

Build: Redelivery and Heartbeat Reaping — the Capstone

Maps to: Phase 6 (cluster — the capstone). Kind: Build.

This is the climax. Every arc of this course has been laying track toward one sentence — under worker failure, no eval is lost and none is double-counted — and this chapter is where you lay the last rail and run a train over it. You already have a RemoteWorker, a connection actor that multiplexes one socket, and a worker binary that registers and answers Assign frames. What is missing is the machinery that survives a worker freezing rather than closing cleanly, and — the whole point — a test that kills a worker mid-run and proves the two properties hold. You add three things: an inactivity timeout in the connection actor, a heartbeat from the worker, and the capstone integration test. When it goes green, the system does the thing it was built to do.

Objective

Add half-open failure detection to the cluster and prove the payoff end to end. Concretely: give connection_actor a fresh inactivity timer per loop iteration that any frame resets and whose expiry reaps the connection; add DEFAULT_HEARTBEAT_TIMEOUT and the serve_workers_with seam that injects a short timeout for tests; make the worker binary emit a periodic Heartbeat; then write the integration test that dispatches a full run across two networked workers, kills one mid-job, and asserts every job completed exactly once. By the end the workspace is at 38 tests, clippy is clean, and both binaries build.

Scaffold

Modify (no new crates this chapter):

• crates/panoptes-control/src/remote.rs — add the DEFAULT_HEARTBEAT_TIMEOUT constant, thread a heartbeat_timeout: Duration through serve_workers_with → handle_connection → connection_actor, and add the inactivity branch to the actor's select!. serve_workers stays as the public entry point that calls serve_workers_with(listener, pool, DEFAULT_HEARTBEAT_TIMEOUT). Two new unit tests go in the existing #[cfg(test)] mod tests.
• crates/panoptes-worker/src/lib.rs — add a heartbeat: Duration parameter to run_session and a tokio::time::interval branch in its select! that sends a Heartbeat frame each tick. Add a DEFAULT_HEARTBEAT constant (the worker's beat interval — several times shorter than the coordinator's timeout).
• crates/panoptes-control/tests/distributed.rs — new integration test file holding the capstone a_dying_worker_loses_no_evals_and_double_counts_none.

Dependencies this chapter exercises: tokio::time (sleep, interval, timeout) for the timers; tokio::select! for the reaper branch; the Scheduler and its WorkerSource from Part VI; the Store's idempotent record_job_outcome from Part IV. No new crate dependencies.

Expected result:

• cargo test -p panoptes-control remote:: → the two new unit tests pass (a_silent_worker_is_reaped, heartbeats_keep_a_worker_alive) alongside the existing remote tests.
• cargo test -p panoptes-control --test distributed → 1 test passes (a_dying_worker_loses_no_evals_and_double_counts_none).
• cargo test across the workspace → 38 tests pass.

The spec (givens)

The inactivity timeout in connection_actor

The actor already select!s over two branches: pull a new Dispatch while below capacity, and receive a frame from the socket. Add a third branch — the reaper. The exact shape matters:

loop {
    // A fresh timer each iteration: any frame or dispatch resets the clock,
    // so only genuine silence for the whole window trips it.
    let idle = tokio::time::sleep(heartbeat_timeout);
    tokio::select! {
        maybe = jobs.recv(), if pending.len() < capacity => { /* ... existing ... */ }
        frame = conn.recv() => { /* ... existing; a Heartbeat is a no-op ... */ }
        _ = idle => break, // silent past the timeout — presumed dead
    }
}

Three things are load-bearing and easy to get wrong:

1. The timer is created inside the loop, at the top of each iteration. Because select! drops the losing futures, every time a job branch or a frame branch wins, the next iteration builds a brand-new sleep. That is what makes any activity reset the clock. Build the timer once above the loop and it will fire on wall-clock schedule regardless of traffic — reaping busy, healthy workers. (This is exactly toy (a) from the concept chapter.)
2. A Heartbeat frame is handled but does nothing. Its arrival already reset the clock by driving another loop iteration; there is no state to update. The match arm exists only so the frame is consumed rather than treated as a protocol error.
3. On the idle branch you break — you fall out of the loop into the same cleanup the clean-close path uses: drain pending and send each waiter a retryable ControlError::Worker. Reaping and clean close funnel to one exit, so a frozen worker becomes an ordinary redelivery.

→ Answer key

DEFAULT_HEARTBEAT_TIMEOUT and serve_workers_with

/// How long a connection may go silent — no result, no heartbeat — before the
/// coordinator declares the worker dead. Catches a *half-open* connection.
pub const DEFAULT_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(30);

pub async fn serve_workers(listener: TcpListener, pool: SharedPool) {
    serve_workers_with(listener, pool, DEFAULT_HEARTBEAT_TIMEOUT).await
}

pub async fn serve_workers_with(
    listener: TcpListener,
    pool: SharedPool,
    heartbeat_timeout: Duration,
);

serve_workers is what production calls — 30 seconds is generous, tuned so a worker that beats every few seconds tolerates ordinary jitter. serve_workers_with exists for testability: a test cannot wait 30 real seconds to watch a reap, so it passes something like 80ms. This is the same injection trick as the base_url seam in Part III — a production default with a test-time override — applied to time instead of a URL. Thread heartbeat_timeout down through handle_connection into connection_actor unchanged.

→ Answer key

The worker's heartbeat in run_session

Give run_session a heartbeat: Duration and add a beat branch to its select!:

let mut beat = tokio::time::interval(heartbeat);
beat.tick().await; // the first tick is immediate — skip it

loop {
    tokio::select! {
        _ = beat.tick() => {
            conn.send(&Message::Heartbeat { worker_id: worker_id.to_string() }).await?;
        }
        frame = conn.recv() => { /* ... existing Assign handling ... */ }
    }
}

Two givens. First, skip the immediate first tick — tokio::time::interval fires once right away, and you do not want a beat before the session is even doing anything; beat.tick().await once before the loop discards it. Second, DEFAULT_HEARTBEAT (the worker's interval) must be several times shorter than DEFAULT_HEARTBEAT_TIMEOUT (the coordinator's window): 5 seconds against 30 means six beats fit in a window, so losing a few to a slow network never causes a false reap. The rate relationship is the correctness argument from the concept chapter — set it wrong and you either reap live workers or notice deaths slowly.

→ Answer key

Why redelivery is safe — the idempotency you already built

When the reaped worker's in-flight jobs fail retryably, the scheduler redelivers each to a survivor. But a redelivered job can also have been run by the worker that died — a worker that finished, shipped its Result, and was reaped before a later beat, or one that was merely slow. So the same job's outcome can reach the store twice. That is harmless only because record_job_outcome is idempotent: it advances the run's done_count only when the job transitions to done for the first time (UPDATE jobs SET status='done' ... WHERE status != 'done', then bump the run only if that affected one row). The second recording is a no-op on progress. You do not build anything new here — you depend on the guard you built in Part IV. If that test (recording_the_same_job_twice_counts_once) were not green, the capstone below could not pass. Re-read the idempotent-recording step if the guard is fuzzy.

→ Answer key

The capstone test wiring

a_dying_worker_loses_no_evals_and_double_counts_none stands up the whole system and breaks it on purpose. The pieces:

• A coordinator: bind a TcpListener on 127.0.0.1:0, make a SharedPool, and tokio::spawn(serve_workers(listener, pool)).
• A reliable worker helper: connect, Register with capacity 4, then loop forever answering each Assign with one ResponseRecord per vignette.
• A dying worker helper: connect, Register with capacity 1, receive exactly one Assign, then drop the socket without answering — a crash mid-job.
• Spawn one of each, wait_for_workers(&pool, 2), then submit a real run: a manifest of 3 vignettes, models = ["claude", "gpt"], epochs = 2 → 4 jobs × 3 vignettes = 12 records expected. Insert it into an in-memory Store, build a Scheduler::with_source(store, Arc::new(pool), &out), and drive it with scheduler.tick().await.
• Assert: the run ends RunStatus::Done, done_count == 4 ("every job must complete exactly once"), and store.run_results(run.id) returns exactly 12 records.

The dying worker takes a job and vanishes; that job fails retryably (via the socket close here — the reaper covers the frozen variant, proven separately by the unit test); the scheduler redelivers it to the reliable worker. If redelivery double-counted, done_count would exceed 4 or run_results would exceed 12. It does neither. That is the sentence, executed.

→ Answer key

Concepts exercised

• A tokio::select! inactivity-timeout branch with a per-iteration timer (the reaper).
• A production default plus a test-time override for a Duration — the timing analogue of the base_url seam.
• A periodic tokio::time::interval heartbeat, with the immediate first tick skipped.
• The heartbeat-interval-vs-timeout rate relationship as a correctness property.
• At-least-once redelivery driven entirely by the retryable error taxonomy from Part II.
• Idempotent recording (Part IV) as the safety net that makes at-least-once non-destructive.
• A full-stack integration test: real sockets, a real scheduler, a real store, an induced failure.

The build loop (you drive)

Test 1 — a_silent_worker_is_reaped (in remote.rs, #[tokio::test])

1. Write the failing test. Stand up a coordinator with a short timeout via serve_workers_with(listener, pool, Duration::from_millis(80)). Spawn a fake worker that Registers, then sleeps for 5 seconds holding the socket open — silent, never a beat, never a close. wait_for_workers(&pool, 1). Then assert the pool empties: timeout(2s, ...) a loop that waits while !pool.is_empty().
2. Predict: with the reaper branch not yet added, what does the actor's select! do while the worker sits silent — and does the pool ever empty? Why would this test hang rather than fail cleanly?
3. Run — it hangs/times out (no reaper yet).
4. Implement the third select! branch and thread heartbeat_timeout through. How you arrange it is yours; the givens above pin what.
5. Run green, commit.

Test 2 — heartbeats_keep_a_worker_alive (in remote.rs, #[tokio::test])

1. Write the failing test. Coordinator with Duration::from_millis(120). Spawn a worker that Registers and then beats every 30ms (send a Heartbeat frame, sleep(30ms), repeat ~10 times). wait_for_workers(&pool, 1), then sleep(200ms) — well past a single 120ms window — and assert pool.len() == 1.
2. Predict: 30ms beats against a 120ms window — how many beats land per window, and why does no single stretch of silence reach 120ms? What would you change to make this same worker get falsely reaped?
3. Run — confirm it passes once the reaper resets its clock on each frame. If it fails, your timer is not being rebuilt per iteration (given #1).
4. Run green, commit.

Test 3 — a_dying_worker_loses_no_evals_and_double_counts_none (new file tests/distributed.rs, #[tokio::test])

1. Write the failing test. Wire it exactly as the capstone givens describe: coordinator, one reliable worker, one dying worker, a 4-job run over 3 vignettes, drive with scheduler.tick(), assert Done / done_count == 4 / 12 records.
2. Predict: the dying worker takes one job and vanishes. Trace the redelivered job's path: which error does its dispatch return, what does the scheduler do with a retryable error, and which store property stops the redelivery from pushing done_count to 5? Name the test in Part IV that guarantees that property.
3. Run — it fails only if a wire is loose (the worker binary must beat, the scheduler must redeliver, the store must be idempotent). If done_count comes back as 5 or records as more than 12, the idempotency guard is the thing to inspect.
4. Run green, commit. This is the payoff — take a beat.

Done when

cargo test across the workspace shows 38 passing tests, including a_silent_worker_is_reaped, heartbeats_keep_a_worker_alive, and the integration capstone a_dying_worker_loses_no_evals_and_double_counts_none. cargo clippy --workspace --all-targets is clean, and both binaries — the coordinator and the worker — build. When that capstone line goes green, the coordinator does, for real, the thing the introduction promised on page one: it runs an eval run to completion across machines, and when a machine dies mid-job, it loses no eval and double-counts none.

Concept-Check: The Cluster

Kind: Quiz. One pass over the whole arc — and, really, over the whole course.

The Cluster arc turned the coordinator into a distributed system without touching the scheduler. You framed a byte stream into whole messages with a length-delimited codec; you built a connection actor that multiplexes one socket, tagging results back to the jobs awaiting them; you wrapped it in a RemoteWorker that the scheduler dispatches through exactly like a LocalWorker; and you closed the loop with heartbeat reaping and at-least-once redelivery, made safe by the store's idempotent recording. This check mixes compiler-verified tracing questions with judgment questions spanning framing, the actor, the seam, heartbeats, redelivery, and idempotency — the sentence the whole course pays off.

If a question stings, the fix is upstream: TCP Is a Byte Stream for framing, The Connection Actor for multiplexing and reply-matching, and At-Least-Once, Heartbeats, and Idempotency for reaping and redelivery. And the idempotency half reaches all the way back to the atomic claim and transactional recording in Part IV — that guard is what this arc leans on. Re-read the section, then come back.

Next: Part IX, the wrap-up — where this coordinator plugs into the wider Panoptes harness, and what a production version would add next (real brokers, mTLS, sharding).

Where This Plugs Into Panoptes

Kind: Wrap-up.

You have built panoptes-control. Before we talk about where it goes next, let us place it — precisely — inside the harness it serves. You did not build a toy distributed system that happens to shuttle eval jobs. You built the control plane for the EXECUTE stage of the Panoptes harness, and every design choice you made was in service of one job in a larger pipeline.

The stage you now own

The harness has stages on either side of you, and they hand work across file contracts, not function calls. Upstream, the generation stage produces a manifest of vignettes — decision-scenario prompts, parameterized and pinned. That manifest, plus a set of models and a number of epochs, is the submission your coordinator accepts. Downstream, the coding and analysis stages consume a log of raw model responses. That log — one JSONL record per response, prompt and raw output and model version and token usage, append-only — is the evidentiary record the rest of Panoptes is built on. It is the dataset of record for the thesis, the benchmark, and the client deliverable alike.

Your coordinator is the engine that turns the first artifact into the second. Submitted as a Run, split into Jobs — one per model × epoch — scheduled across workers, each response logged as it lands, the run marked done when every job has. That is the EXECUTE stage, promoted from a batch loop that waits on the network into a long-lived service that schedules the waiting across machines.

The artifacts you built, and how they compose

Five crates, each one thing, the dependency arrow always pointing inward at the core:

• control-core — the seams. The WorkerHandle trait the scheduler dispatches through, the Run/Job/JobOutcome domain, the id newtypes, the ControlError taxonomy that knows what is worth retrying, and the wire Message enum with its framed codec. Nothing in the workspace that matters is not defined here first.
• control-eval — the workload. A mockable ModelClient, run_eval that drives a job's prompts through it, and the append-only JSONL contract that writes the evidentiary record. This is the code that actually produces what Panoptes consumes.
• control-store — the coordination point. sqlx/SQLite, the atomic claim_next_run, and the transactional, idempotent outcome recording. Every fact about a run's progress lives here, which is exactly why it can survive a crash and referee two schedulers.
• panoptes-control — the coordinator binary. The axum API that accepts submissions, the scheduler that fans jobs out with bounded concurrency and retries only the retryable, the /stats endpoint that accounts for tokens and cost per model, and the pool that fronts remote workers.
• panoptes-worker — the networked worker binary. Connect, register, run run_eval, heartbeat, reconnect.

Read that list as a pipeline and a spine at once. The coordinator accepts a run and persists it (store); the scheduler claims it and splits it (store + core); each job is dispatched through a WorkerHandle (core); the worker runs the eval (eval) and appends to the log (eval); the outcome is recorded idempotently (store); /stats reads the tally back out (store). The same run_eval runs whether the job executed in-process or across a socket — only the transport differed.

The spine, one more time

Here is the sentence the whole course was built to earn. The scheduler holds Vec<Arc<dyn WorkerHandle>> and calls dispatch. It never learns whether the handle on the other side is a LocalWorker running the eval in an in-process task or a RemoteWorker shipping it over framed TCP to another machine.

That one seam is why the same coordinator binary runs two deployments without a line of scheduler code changing: a laptop's in-process pool for a quick sweep, and a rack of networked workers for a real run. Parts II through VII built the coordinator against LocalWorker alone. Part VIII added RemoteWorker as a new impl WorkerHandle and a pool to hand it to the scheduler — additive, not a rewrite, exactly as the orientation chapter promised. The distributed system did not require you to rebuild the control plane. It required you to plug a new mechanism into a seam you had already installed. Dependency inversion is not a slogan here; it is the load-bearing wall you can now see holding up two roofs.

Where the verified build lives

Everything here has a worked, compiling reference behind it — the same discipline as courses 1 and 2. The full panoptes-control build, all five crates and 38 tests (including the capstone that kills a worker mid-run and proves no eval is lost or double-counted), lives at github.com/tbar4/panoptes_control. When your version diverges from the answer key, that repo is the arbiter.

Turn the page for the honest accounting: what the hand-rolled protocol costs, and the production paths that pick up where this course deliberately stops.

What's Next: Real Brokers, mTLS, Sharding

Kind: Wrap-up.

Let us be honest about what you built. The protocol at the heart of Part VIII — serde_json frames over tokio-util's LengthDelimitedCodec, a hand-rolled handshake, heartbeats, and at-least-once redelivery — is a pedagogical protocol. It was the right thing to hand-roll, because hand-rolling it is how framing, backpressure, and death detection stop being words and become code you wrote. But a hand-rolled protocol is also a liability you now maintain. This page is the map of where a production system goes from here, and the good news running through all of it: you already have the seam for most of it. The work ahead is filling in mechanisms behind boundaries you have already drawn, not tearing anything out.

(a) A real broker or gRPC, instead of the hand-rolled wire

The framed-TCP protocol is readable and debuggable, and it taught the lesson. In production you would likely reach for tonic (gRPC) or a message broker (NATS, RabbitMQ, Redis Streams) instead.

• What you gain. Backpressure and flow control you did not have to write. Reconnection and retry semantics as library concerns, not your bug surface. Schema evolution — gRPC's protobuf and a broker's typed subjects both give you a disciplined way to add a field without breaking every worker at once, where today a change to the Message enum is a change every peer must ship simultaneously.
• What you lose. A dependency, and — for a broker — an operational thing to run, monitor, and keep alive. gRPC keeps you broker-free but hands you protobuf codegen and a heavier toolchain. The hand-rolled protocol's one virtue is that it has no moving parts you did not build.

You already have the seam for this: the transport lives entirely behind RemoteWorker, which is one impl WorkerHandle. Swapping framed TCP for a tonic client is a rewrite of that one type. The scheduler — which knows only the trait — does not notice.

(b) Security: mTLS and auth on the worker connection

Right now, any process that can reach the coordinator's port can send Register and start receiving jobs. That is fine on loopback and inside a trusted network; it is unacceptable across one.

The production path is mTLS — mutual TLS, where the coordinator and each worker both present certificates, so the coordinator knows a connecting worker is one it issued a cert to, and the worker knows it is talking to the real coordinator. Layer an auth token in the Register handshake on top, and registration becomes a gate instead of an open door.

The seam is here too: TLS wraps the TcpStream before your MessageStream codec ever sees it. The framing and protocol code is unchanged; you are inserting a layer beneath it.

(c) Horizontal scale: sharding across coordinators

One coordinator with many workers takes you a long way. The next ceiling is the coordinator itself — its scheduler, its store. Sharding runs across multiple coordinators, each owning a slice of the run space.

Here is the part worth savoring: the hard primitive for this is already built. The atomic claim_next_run — the UPDATE … RETURNING that hands a queued run to exactly one claimant and no other — is exactly the coordination point two coordinators need to not step on each other. It was built so a crashed scheduler and its replacement could not both run the same run. That same atomicity means two live coordinators pointed at one store can safely race for work: whoever wins the claim owns the run, and the loser moves on. Sharding on a shared store is nearly free because you already paid for the atomic claim. Sharding on separate stores is a bigger step — now you need a router deciding which coordinator owns which run — but the per-coordinator engine does not change.

(d) Durability beyond at-least-once

You built at-least-once delivery made safe by idempotent recording: a dead worker's in-flight job fails retryably, gets redelivered, and the store refuses to double-count it. That is the correct foundation. Production hardens it further:

• A dead-letter path. A job that fails its retry budget should not vanish or wedge its run — it should land somewhere inspectable, so an operator can see why and decide what to do. Today a permanently failing job is a gap you would have to go looking for.
• The "exactly-once" illusion. There is no true exactly-once delivery over an unreliable network — the honest framing is at-least-once delivery plus idempotent processing, which is the combination you already have. Naming it correctly matters, because it tells you where to spend: not on a mythical exactly-once transport, but on making every side effect idempotent, the way the store's recording already is.

The store is the coordination point for all of it. It is already the single source of truth about what happened; dead-lettering is one more terminal state in a state machine you already own.

Where to look when you build these

Two artifacts close the loop. The architecture diagram shows every seam named on this page — RemoteWorker behind WorkerHandle, the store as the coordination point, the transport as a replaceable layer — so you can see exactly where each production path plugs in. And the answer key is the full worked task plan: when you extend the reference build toward any of these, it is the ground truth for the shape you are extending.

You set out to close a distributed-systems gap. You now have a running control plane, a payoff test that proves its hardest property, and a clear-eyed map of the four roads out. That is not a course that ended. That is a foundation with seams cut for what comes next.

Appendix: Task-to-Chapter Map

The course is generated from a single verified build plan. Every build chapter maps to exactly one phase of that plan, and every phase's code lives, in full, in the Answer Key. This page is the index between the three.

The invariants

Four rules hold across the whole course. If you ever find one broken, it is a bug in the course, not in your understanding:

1. Every build chapter maps to exactly one phase of the reference build (panoptes_control), and that phase is green (its tests pass) before the chapter ships.
2. No build chapter uses machinery no earlier chapter demonstrated. Each new crate or API gets a concept chapter first. If a build needs something with no lead-in, that is a missing concept chapter.
3. Specs are givens; implementations are the exercise. A build chapter hands you the design decisions you cannot derive (field lists, wire formats, SQL, constants) and the test names — never the function bodies. The bodies are in the Answer Key, to check against, not to copy from before you try.
4. Every runnable example and every Tracing quiz is compiled before publish. Claimed output is captured from a real run.

The map

Concept chapters and the seams they unlock

Every concept chapter exists to make a later build derivable. The through-line: one seam per arc.

• The seam itself (Part I) — dyn WorkerHandle. Unlocks everything: the scheduler dispatches to it without knowing local from remote.
• The error taxonomy (Part II) — is_retryable(). Unlocks retry (Part VI) and redelivery (Part VIII).
• The client seam (Part III) — dyn ModelClient. Unlocks mock-tested evals with no live network.
• The atomic claim (Part IV) — UPDATE … RETURNING. Unlocks safe concurrent scheduling.
• Bounded concurrency + graceful shutdown (Part VI) — buffer_unordered, watch + select!. Unlocks the dispatch engine.
• Framing + the connection actor (Part VIII) — LengthDelimitedCodec, one actor per socket. Unlocks the networked worker, and the whole payoff falls out: kill a worker mid-run, lose no eval, double-count none.

The Answer Key (Full Task Plan)

This is the one page in Course 3 that shows the full implementations. The build chapters give you specs, tests, and signatures and ask you to write the code yourself; this appendix is where you check your work — or unblock yourself when a task fights back.

Every line below is reproduced verbatim from a verified, compiling workspace: cargo test runs green across all 38 tests and cargo clippy is clean. The reference source lives at https://github.com/tbar4/panoptes_control. The workspace has five crates, built in dependency order:

• control-core — the shared data model (ids, domain types, errors, the worker trait, the wire protocol, and its codec). No I/O beyond the socket in codec.
• control-eval — the workload: a ModelClient and the pure run_eval, plus the JSONL file contract.
• control-store — durable run/job state in SQLite via sqlx, including the atomic claim and the idempotent outcome record.
• panoptes-control — the coordinator: an axum API, the scheduler, the local worker, telemetry, and the distributed (remote-worker) layer.
• panoptes-worker — the far end of the wire: a standalone worker binary.

Use it as an answer key, not a script. The learning is in writing each test and its implementation first. But the wire formats, status-code mappings, SQL, and type shapes are givens — look them up here freely. Test bodies are elided (their names are listed) since you write those in the build chapters; the non-test code is complete.

Part I — control-core: the shared data model

Everything downstream depends on these types, so core is built first. The load-bearing idea throughout is that structurally-identical-but-semantically-distinct values (a run id vs a job id) get distinct types, and that the retryable-vs-terminal split lives in the error enum.

Run and job identifiers

Newtypes over Uuid so a RunId can never be passed where a JobId is expected. #[serde(transparent)] makes the wire form a bare quoted UUID — no wrapper object.

#![allow(unused)]
fn main() {
use std::fmt;

use serde::{Deserialize, Serialize};
use uuid::Uuid;

/// The id of a submitted eval run.
///
/// A newtype over `Uuid` so a run id can never be confused with a job id — the
/// two are structurally identical but semantically distinct, and mixing them is
/// a real bug the type system should reject.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct RunId(pub Uuid);

impl RunId {
    /// A fresh, random run id.
    pub fn new() -> Self {
        RunId(Uuid::new_v4())
    }
}

impl Default for RunId {
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Display for RunId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// The id of a single job (a chunk of a run a worker executes).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct JobId(pub Uuid);

impl JobId {
    pub fn new() -> Self {
        JobId(Uuid::new_v4())
    }
}

impl Default for JobId {
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Display for JobId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}
}

Tests (in #[cfg(test)] mod tests): ids_are_unique, run_id_serializes_as_bare_uuid_string, display_matches_inner_uuid.

Run: cargo test -p control-core ids Expected: PASS, 3 tests.

Domain types: runs, jobs, records

The nouns of the system. RunStatus/JobStatus serialize snake_case; Usage is Default so it can be folded; JobOutcome::total_usage sums the tokens across a job's records.

#![allow(unused)]
fn main() {
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

use crate::ids::{JobId, RunId};

/// The lifecycle of a submitted eval run.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunStatus {
    Queued,
    Running,
    Done,
    Failed,
}

/// The lifecycle of a single job.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum JobStatus {
    Pending,
    Assigned,
    Done,
    Failed,
}

/// Token accounting for one model call.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct Usage {
    pub input_tokens: u32,
    pub output_tokens: u32,
}

/// One vignette in a job's batch — an id and the prompt to pose.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Vignette {
    pub id: String,
    pub prompt: String,
}

/// The workload a single job performs: pose a batch of vignettes to one model
/// at one epoch. This is the unit that fans out across workers.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EvalJob {
    pub vignettes: Vec<Vignette>,
    pub model: String,
    pub epoch: u32,
}

/// One logged model response.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResponseRecord {
    pub vignette_id: String,
    pub model: String,
    pub epoch: u32,
    pub prompt: String,
    pub response: String,
    pub usage: Usage,
}

/// A job: a chunk of a run assigned to a worker.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Job {
    pub id: JobId,
    pub run_id: RunId,
    pub status: JobStatus,
    pub attempt: u32,
    pub spec: EvalJob,
}

/// The result of running one job: the records it produced.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobOutcome {
    pub job_id: JobId,
    pub records: Vec<ResponseRecord>,
}

impl JobOutcome {
    /// Total tokens across every record in this outcome.
    pub fn total_usage(&self) -> Usage {
        self.records.iter().fold(Usage::default(), |acc, r| Usage {
            input_tokens: acc.input_tokens + r.usage.input_tokens,
            output_tokens: acc.output_tokens + r.usage.output_tokens,
        })
    }
}

/// A submitted eval sweep — the top-level unit clients create via the API.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Run {
    pub id: RunId,
    pub status: RunStatus,
    pub created_at: DateTime<Utc>,
    /// The vignette manifest to draw from (the file contract with the harness).
    pub manifest: String,
    pub models: Vec<String>,
    pub epochs: u32,
    pub job_count: u32,
    pub done_count: u32,
}
}

Tests: total_usage_sums_every_record, statuses_serialize_snake_case, job_roundtrips_through_json.

Run: cargo test -p control-core domain Expected: PASS, 3 tests.

The error taxonomy (retryable vs terminal)

One error type for the whole control plane. The distinction that drives the scheduler is is_retryable: Worker and Protocol failures are worth re-dispatching; everything else surfaces.

#![allow(unused)]
fn main() {
use thiserror::Error;

/// The single error type the control plane's stages return.
///
/// As with the ETL, the load-bearing distinction is **retryable vs terminal**:
/// a job whose worker died (`Worker`) or whose connection dropped (`Protocol`)
/// is worth re-dispatching to another worker; a malformed request (`Invalid`)
/// or a missing run (`NotFound`) will fail identically and must surface.
#[derive(Debug, Error)]
pub enum ControlError {
    /// A run or job id that does not exist. Terminal.
    #[error("not found: {0}")]
    NotFound(String),

    /// A request the server cannot honor as written. Terminal.
    #[error("invalid request: {0}")]
    Invalid(String),

    /// A worker failed or vanished mid-job. Retryable on another worker.
    #[error("worker error: {0}")]
    Worker(String),

    /// A wire-protocol failure — a dropped connection, a bad frame. Retryable.
    #[error("protocol error: {0}")]
    Protocol(String),

    /// A persistence failure.
    #[error("store error: {0}")]
    Store(String),

    /// A local I/O failure.
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
}

impl ControlError {
    /// Whether re-dispatching the work could plausibly succeed.
    pub fn is_retryable(&self) -> bool {
        matches!(self, ControlError::Worker(_) | ControlError::Protocol(_))
    }
}
}

Tests: worker_and_protocol_are_retryable, invalid_and_not_found_are_terminal.

Run: cargo test -p control-core error Expected: PASS, 2 tests.

The WorkerHandle trait (the load-bearing seam)

The scheduler holds Box<dyn WorkerHandle>/Arc<dyn WorkerHandle> and dispatches jobs without knowing whether a handle runs work in-process or ships it over the wire. This is why the distributed layer is additive rather than a rewrite.

#![allow(unused)]
fn main() {
use async_trait::async_trait;

use crate::domain::{Job, JobOutcome};
use crate::error::ControlError;

/// The load-bearing seam of the whole control plane.
///
/// The scheduler holds a pool of `Box<dyn WorkerHandle>` and dispatches jobs
/// across it, knowing nothing about *how* a handle runs the work. A
/// `LocalWorker` runs it in an in-process task; a `RemoteWorker` ships it over
/// the wire to a worker process. Because the scheduler depends only on this
/// trait, the distributed layer is additive — a new handle, not a rewrite.
///
/// `Send + Sync` and object-safe (via `async_trait`) so it can be stored as
/// `dyn`.
#[async_trait]
pub trait WorkerHandle: Send + Sync {
    /// A stable identifier for this worker, for logging and accounting.
    fn id(&self) -> &str;

    /// Run one job to completion and return its outcome.
    async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError>;
}
}

Tests: a_worker_handle_can_be_boxed_as_dyn.

Run: cargo test -p control-core worker Expected: PASS, 1 test.

The wire protocol

One self-describing message enum for the coordinator↔worker conversation. #[serde(tag = "type")] gives every JSON frame a discriminator so the receiver routes it without a side channel.

#![allow(unused)]
fn main() {
use serde::{Deserialize, Serialize};

use crate::domain::{Job, JobOutcome};

/// One message on the coordinator↔worker wire.
///
/// Serialized as JSON and shipped as a length-delimited frame in Part VII. The
/// `#[serde(tag = "type")]` gives every frame a self-describing discriminator,
/// so the receiver can route it without a side channel.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum Message {
    /// A worker announcing itself and how many jobs it can hold at once.
    Register { worker_id: String, capacity: u32 },
    /// The coordinator handing a job to a worker.
    Assign { job: Job },
    /// A worker returning a finished job.
    Result { outcome: JobOutcome },
    /// A liveness ping from a worker.
    Heartbeat { worker_id: String },
}
}

Tests: register_is_tagged_and_roundtrips, assign_carries_a_full_job.

Run: cargo test -p control-core proto Expected: PASS, 2 tests.

The codec: length-delimited JSON frames over TCP

LengthDelimitedCodec prefixes each frame with its byte length so the reader always knows where a message ends despite TCP being a boundary-less byte stream. On top of that framing, each Message is JSON. recv returning Ok(None) means a clean peer close.

#![allow(unused)]
fn main() {
//! The wire: length-delimited, JSON-encoded [`Message`] frames over TCP.
//!
//! `LengthDelimitedCodec` handles framing — it prefixes each frame with its
//! byte length so the reader always knows where a message ends, even though TCP
//! itself is just a stream of bytes with no message boundaries. On top of that
//! framing we serialize each `Message` as JSON: readable and debuggable.

use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use tokio::net::TcpStream;
use tokio_util::codec::{Framed, LengthDelimitedCodec};

use crate::error::ControlError;
use crate::proto::Message;

/// A bidirectional channel of [`Message`]s over one TCP connection.
pub struct MessageStream {
    framed: Framed<TcpStream, LengthDelimitedCodec>,
}

impl MessageStream {
    pub fn new(stream: TcpStream) -> Self {
        Self {
            framed: Framed::new(stream, LengthDelimitedCodec::new()),
        }
    }

    /// Encode a message as a single length-prefixed JSON frame and send it.
    pub async fn send(&mut self, msg: &Message) -> Result<(), ControlError> {
        let bytes = serde_json::to_vec(msg).map_err(|e| ControlError::Protocol(e.to_string()))?;
        self.framed
            .send(Bytes::from(bytes))
            .await
            .map_err(|e| ControlError::Protocol(e.to_string()))
    }

    /// Read the next frame and decode it. `Ok(None)` means the peer closed the
    /// connection cleanly.
    pub async fn recv(&mut self) -> Result<Option<Message>, ControlError> {
        match self.framed.next().await {
            Some(Ok(frame)) => Ok(Some(
                serde_json::from_slice(&frame)
                    .map_err(|e| ControlError::Protocol(e.to_string()))?,
            )),
            Some(Err(e)) => Err(ControlError::Protocol(e.to_string())),
            None => Ok(None),
        }
    }
}
}

Tests: a_message_roundtrips_over_a_real_socket.

Run: cargo test -p control-core codec Expected: PASS, 1 test.

Part II — control-eval: the workload

A job is mostly a network round-trip to a model. Isolating that behind ModelClient keeps the scheduler provider-agnostic and lets everything be tested against a wiremock server instead of a real API.

ModelClient + an HTTP implementation

A mockable trait and one concrete client with an injectable base_url. A transport failure or non-success status becomes a retryable Worker error; a malformed body is a terminal Invalid error.

#![allow(unused)]
fn main() {
//! The model client — the network call a job is mostly made of.
//!
//! A trait so the scheduler and `run_eval` are provider-agnostic and testable
//! against a mock, plus one concrete HTTP implementation. This mirrors the
//! harness's `ModelClient`; the point here is that the whole workload is a
//! network round-trip, which is what makes distributing it worthwhile.

use async_trait::async_trait;
use control_core::{ControlError, Usage};
use serde::{Deserialize, Serialize};

/// A model's reply plus its token accounting.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModelResponse {
    pub text: String,
    pub usage: Usage,
}

/// Something that turns a prompt into a response. Mockable.
#[async_trait]
pub trait ModelClient: Send + Sync {
    fn model_name(&self) -> &str;
    async fn generate(&self, prompt: &str) -> Result<ModelResponse, ControlError>;
}

/// A concrete client over HTTP with an injectable base URL (so tests point it
/// at a `wiremock` server instead of a real provider).
pub struct HttpModelClient {
    base_url: String,
    model: String,
    http: reqwest::Client,
}

impl HttpModelClient {
    pub fn new(base_url: impl Into<String>, model: impl Into<String>) -> Self {
        Self {
            base_url: base_url.into(),
            model: model.into(),
            http: reqwest::Client::new(),
        }
    }
}

#[derive(Serialize)]
struct GenRequest<'a> {
    model: &'a str,
    prompt: &'a str,
}

#[derive(Deserialize)]
struct GenResponse {
    text: String,
    usage: Usage,
}

#[async_trait]
impl ModelClient for HttpModelClient {
    fn model_name(&self) -> &str {
        &self.model
    }

    async fn generate(&self, prompt: &str) -> Result<ModelResponse, ControlError> {
        let resp = self
            .http
            .post(format!(
                "{}/v1/generate",
                self.base_url.trim_end_matches('/')
            ))
            .json(&GenRequest {
                model: &self.model,
                prompt,
            })
            .send()
            .await
            // A failed model call is worth retrying on another attempt.
            .map_err(|e| ControlError::Worker(e.to_string()))?;
        if !resp.status().is_success() {
            return Err(ControlError::Worker(format!(
                "model status {}",
                resp.status()
            )));
        }
        let parsed: GenResponse = resp
            .json()
            .await
            .map_err(|e| ControlError::Invalid(e.to_string()))?;
        Ok(ModelResponse {
            text: parsed.text,
            usage: parsed.usage,
        })
    }
}
}

Tests: model_client_posts_prompt_and_parses_reply, server_error_is_a_retryable_worker_error.

Run: cargo test -p control-eval client Expected: PASS, 2 tests.

run_eval: the pure workload

The one function both local and remote workers run — only the transport around it differs. It poses every vignette and collects one record each; a single failing call fails the whole job, leaving the retry decision to the scheduler.

#![allow(unused)]
fn main() {
//! `run_eval` — the pure workload a job performs, shared by local and remote
//! workers alike. Only the transport around it differs.

use control_core::{ControlError, EvalJob, ResponseRecord};

use crate::client::ModelClient;

/// Pose every vignette in the job to the model and collect one record each.
///
/// A single failing call fails the whole job (the scheduler decides whether to
/// retry, based on `ControlError::is_retryable`).
pub async fn run_eval(
    client: &dyn ModelClient,
    spec: &EvalJob,
) -> Result<Vec<ResponseRecord>, ControlError> {
    let mut records = Vec::with_capacity(spec.vignettes.len());
    for vignette in &spec.vignettes {
        let resp = client.generate(&vignette.prompt).await?;
        records.push(ResponseRecord {
            vignette_id: vignette.id.clone(),
            model: spec.model.clone(),
            epoch: spec.epoch,
            prompt: vignette.prompt.clone(),
            response: resp.text,
            usage: resp.usage,
        });
    }
    Ok(records)
}
}

Tests: run_eval_produces_one_record_per_vignette.

Run: cargo test -p control-eval eval Expected: PASS, 1 test.

The file contract: manifest in, records out

Read a JSONL vignette manifest (the harness's generation output) and append ResponseRecords the coding stage reads. Append-only — a run never truncates prior output.

#![allow(unused)]
fn main() {
//! The file contract with the rest of Panoptes: read a vignette manifest
//! (the harness's generation output) and append response records the coding
//! stage reads. JSON Lines, append-only — the same shape the other courses use.

use std::path::Path;

use control_core::{ControlError, ResponseRecord, Vignette};
use tokio::fs::OpenOptions;
use tokio::io::AsyncWriteExt;

/// Parse a JSONL manifest of `{ "id": ..., "prompt": ... }` into vignettes.
pub fn parse_manifest(text: &str) -> Result<Vec<Vignette>, ControlError> {
    let mut out = Vec::new();
    for line in text.lines().filter(|l| !l.trim().is_empty()) {
        out.push(serde_json::from_str(line).map_err(|e| ControlError::Invalid(e.to_string()))?);
    }
    Ok(out)
}

/// Read and parse a manifest file.
pub async fn load_manifest(path: impl AsRef<Path>) -> Result<Vec<Vignette>, ControlError> {
    let text = tokio::fs::read_to_string(path).await?;
    parse_manifest(&text)
}

/// Append records to the response log as JSONL. Append-only — a run never
/// truncates prior output.
pub async fn append_records(
    path: impl AsRef<Path>,
    records: &[ResponseRecord],
) -> Result<usize, ControlError> {
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .await?;
    let mut buf = String::new();
    for rec in records {
        buf.push_str(
            &serde_json::to_string(rec).map_err(|e| ControlError::Invalid(e.to_string()))?,
        );
        buf.push('\n');
    }
    file.write_all(buf.as_bytes()).await?;
    Ok(records.len())
}
}

Tests: parse_manifest_reads_jsonl, append_records_is_append_only.

Run: cargo test -p control-eval contract Expected: PASS, 2 tests.

Part III — control-store: durable run/job state

SQLite via sqlx's runtime query API (no compile-time DATABASE_URL, so the whole course stays a plain cargo test). The two load-bearing operations are the atomic claim and the idempotent, transactional outcome record.

The schema

Two tables — runs and jobs. models is a JSON array; a job's outcome is JSON, NULL until the job finishes. Indexes support the claim (oldest queued) and per-run gathering.

CREATE TABLE runs (
    id         TEXT    PRIMARY KEY,
    status     TEXT    NOT NULL,
    created_at TEXT    NOT NULL,
    manifest   TEXT    NOT NULL,
    models     TEXT    NOT NULL,           -- JSON array of model names
    epochs     INTEGER NOT NULL,
    job_count  INTEGER NOT NULL DEFAULT 0,
    done_count INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE jobs (
    id      TEXT    PRIMARY KEY,
    run_id  TEXT    NOT NULL REFERENCES runs (id),
    status  TEXT    NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 0,
    spec    TEXT    NOT NULL,              -- JSON EvalJob
    outcome TEXT                            -- JSON JobOutcome, NULL until done
);

CREATE INDEX idx_runs_status ON runs (status, created_at);
CREATE INDEX idx_jobs_run ON jobs (run_id);

The migration is embedded and run by Store::connect via sqlx::migrate!().

Store: connect, in-memory, and basic CRUD

The module header, status↔string helpers, the Store struct, and the create/read paths. max_connections(1) keeps an in-memory database alive for the pool's life and matches SQLite's single-writer model. new_job (the job-builder helper) and row_to_run are shown here too since the CRUD paths use them.

#![allow(unused)]
fn main() {
//! `control-store` — durable run/job state in SQLite via `sqlx`.
//!
//! Uses the runtime query API (no compile-time `DATABASE_URL` needed, so the
//! whole course stays a plain `cargo test`). The two load-bearing operations
//! are the **atomic claim** (`UPDATE … RETURNING`, so two schedulers never grab
//! the same run) and the **transactional outcome record** (job + run counts
//! move together or not at all).

use std::str::FromStr;

use chrono::{DateTime, Utc};
use control_core::{
    ControlError, EvalJob, Job, JobOutcome, JobStatus, ResponseRecord, Run, RunStatus,
};
use control_core::{JobId, RunId};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow};
use sqlx::{Row, SqlitePool};
use uuid::Uuid;

fn store_err(e: impl std::fmt::Display) -> ControlError {
    ControlError::Store(e.to_string())
}

fn run_status_str(s: RunStatus) -> &'static str {
    match s {
        RunStatus::Queued => "queued",
        RunStatus::Running => "running",
        RunStatus::Done => "done",
        RunStatus::Failed => "failed",
    }
}

fn parse_run_status(s: &str) -> Result<RunStatus, ControlError> {
    match s {
        "queued" => Ok(RunStatus::Queued),
        "running" => Ok(RunStatus::Running),
        "done" => Ok(RunStatus::Done),
        "failed" => Ok(RunStatus::Failed),
        other => Err(ControlError::Store(format!("bad run status {other:?}"))),
    }
}

fn job_status_str(s: JobStatus) -> &'static str {
    match s {
        JobStatus::Pending => "pending",
        JobStatus::Assigned => "assigned",
        JobStatus::Done => "done",
        JobStatus::Failed => "failed",
    }
}

/// The persistence layer. Cheap to clone (holds a pool handle).
#[derive(Clone)]
pub struct Store {
    pool: SqlitePool,
}

impl Store {
    /// Connect (creating the file if missing), run migrations. `max_connections(1)`
    /// keeps an in-memory database alive for the life of the pool and matches
    /// SQLite's single-writer model.
    pub async fn connect(url: &str) -> Result<Self, ControlError> {
        let opts = SqliteConnectOptions::from_str(url)
            .map_err(store_err)?
            .create_if_missing(true);
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(opts)
            .await
            .map_err(store_err)?;
        sqlx::migrate!().run(&pool).await.map_err(store_err)?;
        Ok(Self { pool })
    }

    /// A fresh in-memory database — one per test.
    pub async fn in_memory() -> Result<Self, ControlError> {
        Self::connect("sqlite::memory:").await
    }

    /// Insert a new run.
    pub async fn insert_run(&self, run: &Run) -> Result<(), ControlError> {
        sqlx::query(
            "INSERT INTO runs (id, status, created_at, manifest, models, epochs, job_count, done_count) \
             VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(run.id.to_string())
        .bind(run_status_str(run.status))
        .bind(run.created_at.to_rfc3339())
        .bind(&run.manifest)
        .bind(serde_json::to_string(&run.models).map_err(store_err)?)
        .bind(run.epochs)
        .bind(run.job_count)
        .bind(run.done_count)
        .execute(&self.pool)
        .await
        .map_err(store_err)?;
        Ok(())
    }

    /// Fetch a run by id.
    pub async fn get_run(&self, id: RunId) -> Result<Option<Run>, ControlError> {
        let row = sqlx::query("SELECT * FROM runs WHERE id = ?")
            .bind(id.to_string())
            .fetch_optional(&self.pool)
            .await
            .map_err(store_err)?;
        row.map(row_to_run).transpose()
    }

    /// Insert a run's jobs.
    pub async fn insert_jobs(&self, jobs: &[Job]) -> Result<(), ControlError> {
        let mut tx = self.pool.begin().await.map_err(store_err)?;
        for job in jobs {
            sqlx::query(
                "INSERT INTO jobs (id, run_id, status, attempt, spec, outcome) VALUES (?, ?, ?, ?, ?, NULL)",
            )
            .bind(job.id.to_string())
            .bind(job.run_id.to_string())
            .bind(job_status_str(job.status))
            .bind(job.attempt)
            .bind(serde_json::to_string(&job.spec).map_err(store_err)?)
            .execute(&mut *tx)
            .await
            .map_err(store_err)?;
        }
        tx.commit().await.map_err(store_err)?;
        Ok(())
    }

    /// All response records produced by a run, gathered from its jobs' outcomes.
    pub async fn run_results(&self, run_id: RunId) -> Result<Vec<ResponseRecord>, ControlError> {
        let rows = sqlx::query("SELECT outcome FROM jobs WHERE run_id = ? AND outcome IS NOT NULL")
            .bind(run_id.to_string())
            .fetch_all(&self.pool)
            .await
            .map_err(store_err)?;
        let mut out = Vec::new();
        for row in rows {
            let json: String = row.try_get("outcome").map_err(store_err)?;
            let outcome: JobOutcome = serde_json::from_str(&json).map_err(store_err)?;
            out.extend(outcome.records);
        }
        Ok(out)
    }
}

fn row_to_run(row: SqliteRow) -> Result<Run, ControlError> {
    let id: String = row.try_get("id").map_err(store_err)?;
    let status: String = row.try_get("status").map_err(store_err)?;
    let created_at: String = row.try_get("created_at").map_err(store_err)?;
    let models: String = row.try_get("models").map_err(store_err)?;
    Ok(Run {
        id: RunId(Uuid::from_str(&id).map_err(store_err)?),
        status: parse_run_status(&status)?,
        created_at: DateTime::parse_from_rfc3339(&created_at)
            .map_err(store_err)?
            .with_timezone(&Utc),
        manifest: row.try_get("manifest").map_err(store_err)?,
        models: serde_json::from_str(&models).map_err(store_err)?,
        epochs: row.try_get::<i64, _>("epochs").map_err(store_err)? as u32,
        job_count: row.try_get::<i64, _>("job_count").map_err(store_err)? as u32,
        done_count: row.try_get::<i64, _>("done_count").map_err(store_err)? as u32,
    })
}

/// Build a job (helper shared by callers that split a run into jobs).
pub fn new_job(run_id: RunId, spec: EvalJob) -> Job {
    Job {
        id: JobId::new(),
        run_id,
        status: JobStatus::Pending,
        attempt: 0,
        spec,
    }
}
}

Tests (whole store module): insert_then_get_run, claim_moves_run_to_running_then_none, record_outcome_advances_done_count_and_finishes_run, recording_the_same_job_twice_counts_once, usage_by_model_sums_across_jobs.

Run: cargo test -p control-store insert_then_get_run Expected: PASS.

claim_next_run: the atomic claim

UPDATE … RETURNING makes reading the oldest queued run and flipping it to running one indivisible step, so two schedulers can never grab the same run. None means nothing is queued.

#![allow(unused)]
fn main() {
    /// Atomically claim the oldest queued run, moving it to `running`. Returns
    /// `None` if nothing is queued. `UPDATE … RETURNING` makes the read and the
    /// state change one indivisible step.
    pub async fn claim_next_run(&self) -> Result<Option<Run>, ControlError> {
        let row = sqlx::query(
            "UPDATE runs SET status = 'running' \
             WHERE id = (SELECT id FROM runs WHERE status = 'queued' ORDER BY created_at LIMIT 1) \
             RETURNING *",
        )
        .fetch_optional(&self.pool)
        .await
        .map_err(store_err)?;
        row.map(row_to_run).transpose()
    }
}

Covered by: claim_moves_run_to_running_then_none.

Run: cargo test -p control-store claim_moves_run_to_running_then_none Expected: PASS.

record_job_outcome: the transactional record

Record a finished job and advance its run's counters in one transaction; when the last job lands, the run flips to done. The job UPDATE and the run counter UPDATE commit together or not at all.

#![allow(unused)]
fn main() {
    /// Record a finished job and advance its run's counters, in one transaction.
    /// When the last job lands, the run flips to `done`.
    ///
    /// **Idempotent by job id.** The jobs `UPDATE` is guarded by
    /// `status != 'done'`, so a redelivered outcome (the same job run twice under
    /// at-least-once) rewrites the outcome but only bumps `done_count` on the
    /// *first* landing — `rows_affected()` tells us whether the job newly
    /// completed. Without this, a worker that dies after finishing but before its
    /// `Result` is acked would get re-dispatched and double-count the run.
    pub async fn record_job_outcome(
        &self,
        run_id: RunId,
        outcome: &JobOutcome,
    ) -> Result<(), ControlError> {
        let mut tx = self.pool.begin().await.map_err(store_err)?;
        let res = sqlx::query(
            "UPDATE jobs SET status = 'done', outcome = ? WHERE id = ? AND status != 'done'",
        )
        .bind(serde_json::to_string(outcome).map_err(store_err)?)
        .bind(outcome.job_id.to_string())
        .execute(&mut *tx)
        .await
        .map_err(store_err)?;
        // Only advance the run when this job transitioned to done just now.
        if res.rows_affected() == 1 {
            sqlx::query(
                "UPDATE runs SET done_count = done_count + 1, \
                 status = CASE WHEN done_count + 1 >= job_count THEN 'done' ELSE status END \
                 WHERE id = ?",
            )
            .bind(run_id.to_string())
            .execute(&mut *tx)
            .await
            .map_err(store_err)?;
        }
        tx.commit().await.map_err(store_err)?;
        Ok(())
    }
}

Covered by: record_outcome_advances_done_count_and_finishes_run.

Run: cargo test -p control-store record_outcome_advances_done_count_and_finishes_run Expected: PASS.

The idempotency guard (why the same job can't double-count)

The heart of at-least-once safety is two lines inside record_job_outcome above. The jobs UPDATE carries a guard:

#![allow(unused)]
fn main() {
"UPDATE jobs SET status = 'done', outcome = ? WHERE id = ? AND status != 'done'"
}

and the run counter only advances when that update actually changed a row:

#![allow(unused)]
fn main() {
if res.rows_affected() == 1 {
    // bump done_count and possibly flip the run to 'done'
}
}