Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

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'
}
}

The first time a job's outcome lands, status is not yet 'done', so the UPDATE matches one row (rows_affected() == 1) and done_count bumps. On a redelivery — the same job re-dispatched after a worker died between finishing and acking — status is already 'done', the AND status != 'done' guard matches zero rows (rows_affected() == 0), the outcome is not rewritten and, crucially, done_count is not incremented again. That is what turns unsafe at-least-once delivery into exactly-once accounting.

Covered by: recording_the_same_job_twice_counts_once (asserts done_count == 1, not 2, after recording the same job id twice).

Run: cargo test -p control-store recording_the_same_job_twice_counts_once Expected: PASS.

usage_by_model + ModelUsage

Aggregate token usage per model across every finished job — the raw material for /stats. Aggregated in Rust from the stored JSON outcomes into a BTreeMap (so output is stable/sorted).

#![allow(unused)]
fn main() {
/// 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 {
    /// Total tokens per model across every finished job — the raw material for
    /// `/stats`. Aggregated in Rust from the stored outcomes.
    pub async fn usage_by_model(&self) -> Result<Vec<ModelUsage>, ControlError> {
        let rows = sqlx::query("SELECT outcome FROM jobs WHERE outcome IS NOT NULL")
            .fetch_all(&self.pool)
            .await
            .map_err(store_err)?;
        let mut by_model: std::collections::BTreeMap<String, ModelUsage> =
            std::collections::BTreeMap::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)?;
            for rec in outcome.records {
                let entry = by_model
                    .entry(rec.model.clone())
                    .or_insert_with(|| ModelUsage {
                        model: rec.model.clone(),
                        input_tokens: 0,
                        output_tokens: 0,
                        calls: 0,
                    });
                entry.input_tokens += u64::from(rec.usage.input_tokens);
                entry.output_tokens += u64::from(rec.usage.output_tokens);
                entry.calls += 1;
            }
        }
        Ok(by_model.into_values().collect())
    }
}
}

Covered by: usage_by_model_sums_across_jobs.

Run: cargo test -p control-store usage_by_model_sums_across_jobs Expected: PASS. (Run the whole crate with cargo test -p control-store — all 5 tests PASS.)

Part IV — panoptes-control: the API service

The coordinator's axum API accepts run submissions, persists them as queued runs, and reports status, results, and cost stats. The error taxonomy maps to status codes in exactly one place.

The router, AppState, and ApiError

app() builds the router; AppState is just a cheaply-cloned Store; ApiError wraps ControlError and its IntoResponse is the single place the taxonomy becomes HTTP status codes. This excerpt also shows the crate's module wiring and re-exports.

#![allow(unused)]
fn main() {
//! `panoptes-control` — the coordinator. The `axum` API is here (this lib
//! target); the `serve` wiring is in `main.rs`.
//!
//! The API accepts eval-run submissions, persists them as queued runs, and
//! reports their status and results. Actually *running* them is the scheduler's
//! job (a later arc); this layer is pure request → store.

pub mod remote;
pub mod scheduler;
pub mod worker;

pub use remote::{RemoteWorker, SharedPool, serve_workers};
pub use scheduler::{RetryPolicy, Scheduler, WorkerSource, plan_jobs};
pub use worker::LocalWorker;

use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use chrono::Utc;
use control_core::{ControlError, Run, RunId, RunStatus};
use control_store::{ModelUsage, Store};
use serde::Deserialize;
use serde_json::json;
use std::str::FromStr;
use tower_http::trace::TraceLayer;
use uuid::Uuid;

/// Shared handler state — just the store, cheaply cloned per request.
#[derive(Clone)]
pub struct AppState {
    pub store: Store,
}

/// Build the coordinator's router.
pub fn app(state: AppState) -> Router {
    Router::new()
        .route("/health", get(|| async { "ok" }))
        .route("/runs", post(create_run))
        .route("/runs/:id", get(get_run))
        .route("/runs/:id/results", get(get_results))
        .route("/stats", get(get_stats))
        .layer(TraceLayer::new_for_http())
        .with_state(state)
}

/// A `ControlError` that knows how to become an HTTP response. Handlers return
/// `Result<_, ApiError>` and use `?`; the taxonomy maps to status codes here,
/// in one place.
pub struct ApiError(ControlError);

impl From<ControlError> for ApiError {
    fn from(e: ControlError) -> Self {
        ApiError(e)
    }
}

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

Tests (whole lib module): post_run_returns_201_and_id, post_with_no_models_is_400, get_missing_run_is_404, get_run_after_post_returns_queued, stats_reports_tokens_and_cost_per_model.

Run: cargo test -p panoptes-control --lib Expected: PASS, 5 tests.

create_run handler

Validates the body (at least one model, epochs ≥ 1), computes job_count as models × epochs, inserts a queued run, and returns 201 Created with the new id.

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

async fn create_run(
    State(state): State<AppState>,
    Json(body): Json<CreateRun>,
) -> Result<Response, ApiError> {
    if body.models.is_empty() {
        return Err(ControlError::Invalid("at least one model is required".into()).into());
    }
    if body.epochs == 0 {
        return Err(ControlError::Invalid("epochs must be >= 1".into()).into());
    }
    let run = Run {
        id: RunId::new(),
        status: RunStatus::Queued,
        created_at: Utc::now(),
        manifest: body.manifest,
        // One job per (model × epoch); the scheduler fills them in later.
        job_count: body.models.len() as u32 * body.epochs,
        models: body.models,
        epochs: body.epochs,
        done_count: 0,
    };
    state.store.insert_run(&run).await?;
    Ok((
        StatusCode::CREATED,
        Json(json!({ "id": run.id.to_string() })),
    )
        .into_response())
}
}

Covered by: post_run_returns_201_and_id, post_with_no_models_is_400.

Run: cargo test -p panoptes-control post_run_returns_201_and_id Expected: PASS.

get_run + get_results handlers

Read paths, plus the shared parse_run_id helper. get_results returns 404 if the run itself doesn't exist rather than a bare empty list.

#![allow(unused)]
fn main() {
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())
}

async fn get_run(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<Run>, ApiError> {
    let run_id = parse_run_id(&id)?;
    state
        .store
        .get_run(run_id)
        .await?
        .map(Json)
        .ok_or_else(|| ControlError::NotFound(format!("run {id}")).into())
}

async fn get_results(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Response, ApiError> {
    let run_id = parse_run_id(&id)?;
    // 404 if the run itself doesn't exist, rather than a bare empty list.
    if state.store.get_run(run_id).await?.is_none() {
        return Err(ControlError::NotFound(format!("run {id}")).into());
    }
    let records = state.store.run_results(run_id).await?;
    Ok(Json(records).into_response())
}
}

Covered by: get_missing_run_is_404, get_run_after_post_returns_queued.

Run: cargo test -p panoptes-control get_run_after_post_returns_queued Expected: PASS.

Telemetry init + TraceLayer wiring

A tracing subscriber, idempotent so tests can call it freely (try_init). The request-span wiring is TraceLayer::new_for_http() layered onto the router in app() (shown under The router).

#![allow(unused)]
fn main() {
/// Telemetry setup — a `tracing` subscriber. Idempotent so tests can call it.
pub mod telemetry {
    use tracing_subscriber::EnvFilter;

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

The .layer(TraceLayer::new_for_http()) call in app() gives every request a span; main calls telemetry::init() once at startup.

Run: cargo build -p panoptes-control Expected: builds clean (telemetry is exercised implicitly by the lib tests, which construct the app).

get_stats + est_cost_usd

Turns usage_by_model into a per-model JSON report with an illustrative flat cost model ($3/1M input, $15/1M output) and a total.

#![allow(unused)]
fn main() {
/// An illustrative flat cost model: $3 per 1M input tokens, $15 per 1M output.
fn est_cost_usd(u: &ModelUsage) -> f64 {
    u.input_tokens as f64 / 1_000_000.0 * 3.0 + u.output_tokens as f64 / 1_000_000.0 * 15.0
}

async fn get_stats(State(state): State<AppState>) -> Result<Json<serde_json::Value>, ApiError> {
    let per_model = state.store.usage_by_model().await?;
    let models: Vec<_> = per_model
        .iter()
        .map(|m| {
            json!({
                "model": m.model,
                "input_tokens": m.input_tokens,
                "output_tokens": m.output_tokens,
                "calls": m.calls,
                "est_cost_usd": est_cost_usd(m),
            })
        })
        .collect();
    let total: f64 = per_model.iter().map(est_cost_usd).sum();
    Ok(Json(
        json!({ "models": models, "total_est_cost_usd": total }),
    ))
}
}

Covered by: stats_reports_tokens_and_cost_per_model (1M input @ $3/M + 1M output @ $15/M = $18).

Run: cargo test -p panoptes-control stats_reports_tokens_and_cost_per_model Expected: PASS.

Part V — the scheduler

The coordinator's engine: claim a queued run, split it into jobs (one per model × epoch), dispatch them across the WorkerHandle pool with bounded concurrency, and retry retryable failures on the next worker. It holds dyn WorkerHandle and so is oblivious to local vs remote.

plan_jobs + WorkerSource/StaticPool

plan_jobs fans a run out into one job (carrying the whole vignette batch) per model, per epoch. WorkerSource is the per-run pool snapshot abstraction; StaticPool is the fixed local case. This excerpt includes the module header and imports.

#![allow(unused)]
fn main() {
//! The scheduler — the coordinator's engine.
//!
//! It claims a queued run, splits it into jobs (one per model × epoch), and
//! dispatches them across the `WorkerHandle` pool with bounded concurrency,
//! retrying *retryable* failures on the next worker. It knows nothing about
//! local vs remote workers — it holds `dyn WorkerHandle` and that is the whole
//! seam that makes the distributed arc additive.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use control_core::{ControlError, EvalJob, Job, JobOutcome, Run, RunId, Vignette, WorkerHandle};
use control_eval::{append_records, load_manifest};
use control_store::{Store, new_job};
use futures::stream::{self, StreamExt};

/// Where the scheduler gets its workers. A snapshot per run lets the pool change
/// underneath — local workers are fixed, but *remote* workers join and leave as
/// connections come and go, and the scheduler shouldn't care which it holds.
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>>);

impl WorkerSource for StaticPool {
    fn snapshot(&self) -> Vec<Arc<dyn WorkerHandle>> {
        self.0.clone()
    }
}

/// Split a run into jobs: one job carrying the whole vignette batch, per model,
/// per epoch.
pub fn plan_jobs(run: &Run, vignettes: &[Vignette]) -> Vec<Job> {
    let mut jobs = Vec::new();
    for model in &run.models {
        for epoch in 0..run.epochs {
            jobs.push(new_job(
                run.id,
                EvalJob {
                    vignettes: vignettes.to_vec(),
                    model: model.clone(),
                    epoch,
                },
            ));
        }
    }
    jobs
}
}

Tests (whole scheduler module): 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.

Run: cargo test -p panoptes-control plan_jobs_is_one_per_model_epoch Expected: PASS.

run_job_with_retry + RetryPolicy

Dispatch one job, retrying retryable failures on the next worker in the pool (indexed by seed + attempt), up to max_attempts. Terminal errors return immediately.

#![allow(unused)]
fn main() {
/// How many times a job may be re-dispatched before it fails for good.
#[derive(Clone, Copy)]
pub struct RetryPolicy {
    pub max_attempts: u32,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self { max_attempts: 3 }
    }
}

/// Dispatch one job, retrying retryable failures on the next worker in the pool.
async fn run_job_with_retry(
    workers: &[Arc<dyn WorkerHandle>],
    policy: RetryPolicy,
    seed: usize,
    job: Job,
) -> Result<JobOutcome, ControlError> {
    let mut attempt = 0u32;
    loop {
        let worker = &workers[(seed + attempt as usize) % workers.len()];
        match worker.dispatch(job.clone()).await {
            Ok(outcome) => return Ok(outcome),
            Err(e) if e.is_retryable() && attempt + 1 < policy.max_attempts => {
                attempt += 1;
            }
            Err(e) => return Err(e),
        }
    }
}
}

Covered by: retryable_failure_is_redispatched_to_the_next_worker, terminal_failure_is_not_retried.

Run: cargo test -p panoptes-control retryable_failure_is_redispatched_to_the_next_worker Expected: PASS.

Scheduler::new/with_source/tick/process_run

Construction over a fixed pool or any WorkerSource, plus the per-run engine. tick claims one queued run and processes it; process_run snapshots the pool once, plans and inserts jobs, then dispatches with buffer_unordered(concurrency), appending records and recording each outcome as it lands.

#![allow(unused)]
fn main() {
pub struct Scheduler {
    store: Store,
    workers: Arc<dyn WorkerSource>,
    policy: RetryPolicy,
    concurrency: usize,
    out_dir: PathBuf,
}

impl Scheduler {
    /// Build a scheduler over a fixed local worker pool.
    pub fn new(
        store: Store,
        workers: Vec<Arc<dyn WorkerHandle>>,
        out_dir: impl Into<PathBuf>,
    ) -> Self {
        Self::with_source(store, Arc::new(StaticPool(workers)), out_dir)
    }

    /// Build a scheduler over any worker source — e.g. a live remote pool that
    /// gains and loses workers as connections open and close.
    pub fn with_source(
        store: Store,
        workers: Arc<dyn WorkerSource>,
        out_dir: impl Into<PathBuf>,
    ) -> Self {
        Self {
            store,
            workers,
            policy: RetryPolicy::default(),
            concurrency: 4,
            out_dir: out_dir.into(),
        }
    }

    /// Claim one queued run and process it to completion. `None` if nothing is
    /// queued.
    pub async fn tick(&self) -> Result<Option<RunId>, ControlError> {
        let Some(run) = self.store.claim_next_run().await? else {
            return Ok(None);
        };
        let run_id = run.id;
        self.process_run(run).await?;
        Ok(Some(run_id))
    }

    #[tracing::instrument(skip(self, run), fields(run_id = %run.id))]
    async fn process_run(&self, run: Run) -> Result<(), ControlError> {
        tokio::fs::create_dir_all(&self.out_dir).await?;
        // Snapshot the pool once for the whole run. A worker that dies mid-run
        // stays in the snapshot but fails its dispatches retryably, so its jobs
        // land on the surviving workers.
        let workers = self.workers.snapshot();
        if workers.is_empty() {
            return Err(ControlError::Worker("no workers available".into()));
        }
        let vignettes = load_manifest(&run.manifest).await?;
        let jobs = plan_jobs(&run, &vignettes);
        tracing::info!(
            jobs = jobs.len(),
            workers = workers.len(),
            "planned jobs for run"
        );
        self.store.insert_jobs(&jobs).await?;

        let run_id = run.id;
        let log_path = self.out_dir.join(format!("{run_id}.jsonl"));
        let policy = self.policy;
        let store = self.store.clone();

        let futures = jobs.into_iter().enumerate().map(|(i, job)| {
            let workers = workers.clone();
            let store = store.clone();
            let log_path = log_path.clone();
            async move {
                let outcome = run_job_with_retry(&workers, policy, i, job).await?;
                append_records(&log_path, &outcome.records).await?;
                store.record_job_outcome(run_id, &outcome).await?;
                Ok::<(), ControlError>(())
            }
        });

        let mut stream = stream::iter(futures).buffer_unordered(self.concurrency);
        while let Some(result) = stream.next().await {
            result?;
        }
        Ok(())
    }
}

Covered by: tick_processes_a_run_to_done.

Run: cargo test -p panoptes-control tick_processes_a_run_to_done Expected: PASS.

Scheduler::run_loop

Poll for queued runs until shutdown. A tick processes a whole run before re-checking shutdown, so shutdown drains the in-flight run rather than dropping its jobs — it just stops claiming new ones. (This method closes the impl Scheduler block opened above.)

#![allow(unused)]
fn main() {
    /// 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) {
        loop {
            if *shutdown.borrow() {
                break;
            }
            match self.tick().await {
                Ok(Some(_)) => continue, // a run was processed — try the next immediately
                Ok(None) => {}
                Err(e) => eprintln!("scheduler tick error: {e}"),
            }
            tokio::select! {
                _ = shutdown.changed() => {}
                _ = tokio::time::sleep(poll) => {}
            }
        }
    }
}
}

Run: cargo test -p panoptes-control scheduler Expected: PASS, 4 scheduler tests (run_loop is exercised via main's graceful-shutdown path and the distributed capstone).

LocalWorker

The in-process WorkerHandle the single-node coordinator uses: it just runs run_eval against a shared ModelClient. The distributed arc adds RemoteWorker behind the same trait, and the scheduler never learns which it holds.

#![allow(unused)]
fn main() {
//! `LocalWorker` — a `WorkerHandle` that runs the eval in-process.
//!
//! This is the concrete worker the single-node coordinator uses. The payoff arc
//! adds a `RemoteWorker` behind the *same* trait; the scheduler never learns
//! which it holds.

use std::sync::Arc;

use async_trait::async_trait;
use control_core::{ControlError, Job, JobOutcome, WorkerHandle};
use control_eval::{ModelClient, run_eval};

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

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

#[async_trait]
impl WorkerHandle for LocalWorker {
    fn id(&self) -> &str {
        &self.id
    }

    async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError> {
        let records = run_eval(self.client.as_ref(), &job.spec).await?;
        Ok(JobOutcome {
            job_id: job.id,
            records,
        })
    }
}
}

Tests: local_worker_runs_the_eval.

Run: cargo test -p panoptes-control local_worker_runs_the_eval Expected: PASS.

Part VI — the distributed layer

The payoff. Worker connections become WorkerHandles the scheduler can't distinguish from local ones. One connection actor per worker owns the socket and multiplexes it; when a connection dies, every in-flight job fails retryably, which is exactly what makes the scheduler redeliver it — and why the store's idempotent recording matters.

RemoteWorker + SharedPool

SharedPool is a live, mutable set of workers (add on register, remove on close) that implements WorkerSource. RemoteWorker is just a channel to a connection actor; its dispatch ships a job and awaits the matching Result, mapping a dead actor/channel to a retryable Worker error. This excerpt includes the module header, imports, and the heartbeat-timeout constant.

#![allow(unused)]
fn main() {
//! The coordinator's half of the network: worker connections become
//! [`WorkerHandle`]s the scheduler can't tell apart from local ones.
//!
//! The trick is one *connection actor* task per worker. It owns the socket and
//! multiplexes it: many jobs may be in flight over a single connection, results
//! come back tagged by job id, and heartbeats interleave with everything. A
//! [`RemoteWorker`] is just a channel to that actor; `dispatch` ships a job and
//! awaits the matching `Result` frame. When the connection dies, every in-flight
//! job fails *retryably* — which is exactly what makes the scheduler redeliver
//! it to a surviving worker (at-least-once), and why the store's idempotent
//! recording matters.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use async_trait::async_trait;
use control_core::{ControlError, Job, JobId, JobOutcome, Message, MessageStream, WorkerHandle};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{mpsc, oneshot};

use crate::scheduler::WorkerSource;

/// How long a connection may go silent — no result, no heartbeat — before the
/// coordinator declares the worker dead. This is what catches a *half-open*
/// connection: a worker whose process froze or whose network dropped without a
/// clean TCP close, so `recv` would otherwise block forever.
pub const DEFAULT_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(30);

/// A job handed to the connection actor plus the channel to answer it on.
type Dispatch = (Job, oneshot::Sender<Result<JobOutcome, ControlError>>);

/// A live, shared set of workers. The accept loop adds a worker when it
/// registers and removes it when its connection closes; the scheduler snapshots
/// it per run. Cheap to clone — it's an `Arc` inside.
#[derive(Clone, Default)]
pub struct SharedPool {
    workers: Arc<Mutex<Vec<Arc<dyn WorkerHandle>>>>,
}

impl SharedPool {
    pub fn add(&self, worker: Arc<dyn WorkerHandle>) {
        self.workers.lock().unwrap().push(worker);
    }

    pub fn remove(&self, id: &str) {
        self.workers.lock().unwrap().retain(|w| w.id() != id);
    }

    pub fn len(&self) -> usize {
        self.workers.lock().unwrap().len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl WorkerSource for SharedPool {
    fn snapshot(&self) -> Vec<Arc<dyn WorkerHandle>> {
        self.workers.lock().unwrap().clone()
    }
}

/// 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 {
        &self.id
    }

    async fn dispatch(&self, job: Job) -> Result<JobOutcome, ControlError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        // If the actor is gone, the connection is dead — retryable.
        self.tx
            .send((job, reply_tx))
            .await
            .map_err(|_| ControlError::Worker("worker connection closed".into()))?;
        // A dropped reply channel means the actor died with the job in flight.
        reply_rx
            .await
            .map_err(|_| ControlError::Worker("worker dropped the job".into()))?
    }
}
}

Tests (whole remote module): remote_worker_dispatches_over_the_wire, worker_death_makes_dispatch_retryable, a_silent_worker_is_reaped, heartbeats_keep_a_worker_alive.

Run: cargo test -p panoptes-control remote_worker_dispatches_over_the_wire Expected: PASS.

connection_actor

Owns one worker connection: forwards assigned jobs as Assign frames, matches returning Result frames to their waiters, honors capacity as real backpressure, resets an idle timer on any activity, and — crucially — fails everything still in flight when the socket closes or goes silent.

#![allow(unused)]
fn main() {
/// Own one worker connection: forward assigned jobs as `Assign` frames, match
/// returning `Result` frames back to their waiters, and — crucially — fail
/// everything still in flight when the socket closes.
///
/// `capacity` is real backpressure: while the worker already holds that many
/// jobs, the actor stops pulling new ones, so `RemoteWorker::dispatch` blocks at
/// its channel and the scheduler naturally throttles.
async fn connection_actor(
    mut conn: MessageStream,
    mut jobs: mpsc::Receiver<Dispatch>,
    capacity: usize,
    heartbeat_timeout: Duration,
) {
    let mut pending: HashMap<JobId, oneshot::Sender<Result<JobOutcome, ControlError>>> =
        HashMap::new();

    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! {
            // Only accept new work while below capacity.
            maybe = jobs.recv(), if pending.len() < capacity => {
                let Some((job, reply)) = maybe else {
                    break; // all RemoteWorker handles dropped
                };
                let job_id = job.id;
                if let Err(e) = conn.send(&Message::Assign { job }).await {
                    let _ = reply.send(Err(e));
                    break; // wire is broken
                }
                pending.insert(job_id, reply);
            }
            frame = conn.recv() => {
                match frame {
                    Ok(Some(Message::Result { outcome })) => {
                        if let Some(reply) = pending.remove(&outcome.job_id) {
                            let _ = reply.send(Ok(outcome));
                        }
                    }
                    Ok(Some(Message::Heartbeat { .. })) => { /* liveness — the read itself proves it */ }
                    Ok(Some(_)) => {} // Register/Assign inbound are protocol errors; ignore
                    Ok(None) | Err(_) => break, // closed or malformed — stop
                }
            }
            _ = idle => break, // silent past the timeout — presumed dead
        }
    }

    // The connection is finished. Fail every job still in flight so the
    // scheduler sees a retryable error and redelivers each to another worker.
    for (_, reply) in pending {
        let _ = reply.send(Err(ControlError::Worker("worker connection lost".into())));
    }
}
}

Covered by: worker_death_makes_dispatch_retryable, a_silent_worker_is_reaped, heartbeats_keep_a_worker_alive.

Run: cargo test -p panoptes-control a_silent_worker_is_reaped Expected: PASS.

serve_workers / serve_workers_with / handle_connection

The accept loop: every connection runs its own actor task; when it ends, the worker is removed from the pool. handle_connection enforces the handshake (first frame must be Register) before spinning up the actor. serve_workers uses DEFAULT_HEARTBEAT_TIMEOUT; tests inject a short one via serve_workers_with.

#![allow(unused)]
fn main() {
/// Accept worker connections forever, registering each into `pool`. Each
/// connection runs its own actor task; when it ends, the worker is removed.
pub async fn serve_workers(listener: TcpListener, pool: SharedPool) {
    serve_workers_with(listener, pool, DEFAULT_HEARTBEAT_TIMEOUT).await
}

/// [`serve_workers`] with an explicit inactivity timeout (tests use a short one).
pub async fn serve_workers_with(
    listener: TcpListener,
    pool: SharedPool,
    heartbeat_timeout: Duration,
) {
    loop {
        let Ok((sock, _addr)) = listener.accept().await else {
            continue;
        };
        let pool = pool.clone();
        tokio::spawn(async move {
            handle_connection(sock, pool, heartbeat_timeout).await;
        });
    }
}

/// Register one connection and run its actor until the socket closes or the
/// worker goes silent.
async fn handle_connection(sock: TcpStream, pool: SharedPool, heartbeat_timeout: Duration) {
    let mut conn = MessageStream::new(sock);
    // The handshake: the first frame must be a Register.
    let (worker_id, capacity) = match conn.recv().await {
        Ok(Some(Message::Register {
            worker_id,
            capacity,
        })) => (worker_id, (capacity as usize).max(1)),
        _ => return, // no valid registration — drop the connection
    };

    let (tx, jobs) = mpsc::channel(capacity);
    let worker: Arc<dyn WorkerHandle> = Arc::new(RemoteWorker {
        id: worker_id.clone(),
        tx,
    });
    pool.add(worker);

    connection_actor(conn, jobs, capacity, heartbeat_timeout).await;

    pool.remove(&worker_id);
}
}

Covered by: remote_worker_dispatches_over_the_wire (and the distributed capstone).

Run: cargo test -p panoptes-control --lib remote Expected: PASS, 4 tests.

panoptes-worker: the worker crate + binary

The far end of the wire, deliberately simple: register, then pull Assign frames, run run_eval, ship Result frames, and beat a heartbeat while idle. Supervision is just a reconnect loop. First the library (run_session / serve_worker / run_worker_forever):

#![allow(unused)]
fn main() {
//! `panoptes-worker` — the far end of the wire. A worker connects to the
//! coordinator, registers, and then does the one thing a worker does: pull
//! `Assign` frames, run the eval, and ship `Result` frames back. Everything
//! networking-shaped (framing, dispatch multiplexing, redelivery) lives on the
//! coordinator; a worker is deliberately simple.

use std::sync::Arc;
use std::time::Duration;

use control_core::{ControlError, JobOutcome, Message, MessageStream};
use control_eval::{ModelClient, run_eval};
use tokio::net::TcpStream;

/// How often a worker announces it is still alive while idle.
pub const DEFAULT_HEARTBEAT: Duration = Duration::from_secs(5);

/// Run one connection session to completion: register, then serve jobs until the
/// coordinator closes the connection (`Ok`) or something on the wire fails
/// (`Err`). An eval failure ends the session too — the coordinator will notice
/// the dropped connection and redeliver the job elsewhere.
pub async fn run_session(
    mut conn: MessageStream,
    worker_id: &str,
    capacity: u32,
    client: &dyn ModelClient,
    heartbeat: Duration,
) -> Result<(), ControlError> {
    conn.send(&Message::Register {
        worker_id: worker_id.to_string(),
        capacity,
    })
    .await?;

    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() => {
                match frame? {
                    Some(Message::Assign { job }) => {
                        let job_id = job.id;
                        // A failed eval bubbles up and ends the session; the
                        // coordinator redelivers. Success ships a Result back.
                        let records = run_eval(client, &job.spec).await?;
                        conn.send(&Message::Result { outcome: JobOutcome { job_id, records } }).await?;
                    }
                    Some(_) => {} // the coordinator only ever sends Assign
                    None => return Ok(()), // clean close
                }
            }
        }
    }
}

/// Connect to the coordinator and run a single session.
pub async fn serve_worker(
    coordinator: &str,
    worker_id: &str,
    capacity: u32,
    client: Arc<dyn ModelClient>,
    heartbeat: Duration,
) -> Result<(), ControlError> {
    let stream = TcpStream::connect(coordinator).await?;
    let conn = MessageStream::new(stream);
    run_session(conn, worker_id, capacity, &*client, heartbeat).await
}

/// Serve forever, reconnecting after any disconnect — workers are cattle, and
/// supervision is just a loop.
pub async fn run_worker_forever(
    coordinator: &str,
    worker_id: &str,
    capacity: u32,
    client: Arc<dyn ModelClient>,
    heartbeat: Duration,
    reconnect_delay: Duration,
) {
    loop {
        match serve_worker(coordinator, worker_id, capacity, client.clone(), heartbeat).await {
            Ok(()) => tracing::info!("coordinator closed the connection; reconnecting"),
            Err(e) => tracing::warn!(error = %e, "session ended; reconnecting"),
        }
        tokio::time::sleep(reconnect_delay).await;
    }
}
}

And the binary (panoptes-worker/src/main.rs) — parse flags, build one shared HttpModelClient, and reconnect forever:

//! The `panoptes-worker` binary — connect to a coordinator and serve eval jobs,
//! reconnecting forever. Runs one shared model client; scale by launching more
//! processes (on more machines), which is the whole point of pulling the worker
//! out of the coordinator.

use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use clap::Parser;
use control_eval::HttpModelClient;
use panoptes_worker::{DEFAULT_HEARTBEAT, run_worker_forever};
use tracing_subscriber::EnvFilter;

#[derive(Parser)]
#[command(name = "panoptes-worker", version, about = "A Panoptes eval worker")]
struct Cli {
    /// Coordinator worker-port address to connect to.
    #[arg(long, default_value = "127.0.0.1:8081")]
    coordinator: String,
    /// Base URL of the model API this worker calls.
    #[arg(long, default_value = "http://127.0.0.1:9000")]
    model_api: String,
    /// Model name to request.
    #[arg(long, default_value = "claude")]
    model: String,
    /// This worker's id (must be unique across the cluster).
    #[arg(long, default_value = "worker-1")]
    id: String,
    /// How many jobs this worker will hold at once.
    #[arg(long, default_value_t = 4)]
    capacity: u32,
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    tracing_subscriber::fmt().with_env_filter(filter).init();

    let client = Arc::new(HttpModelClient::new(cli.model_api, cli.model));
    tracing::info!(id = %cli.id, coordinator = %cli.coordinator, "worker starting");

    run_worker_forever(
        &cli.coordinator,
        &cli.id,
        cli.capacity,
        client,
        DEFAULT_HEARTBEAT,
        Duration::from_secs(1),
    )
    .await;
    Ok(())
}

Tests: worker_registers_runs_a_job_and_returns_a_result.

Run: cargo test -p panoptes-worker Expected: PASS, 1 test.

panoptes-control: the coordinator binary

Wires everything together under one graceful-shutdown signal: connect the store, assemble a SharedPool of in-process workers and/or a TCP port remote workers dial into, start the scheduler's run_loop, and serve the API. Ctrl-C drains the in-flight run before exit.

//! The `panoptes-control` binary — connect the store, assemble a worker pool
//! (in-process workers and/or a TCP port that remote workers dial into), start
//! the scheduler, and serve the API, all under one graceful-shutdown signal.

use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use clap::Parser;
use control_core::WorkerHandle;
use control_eval::HttpModelClient;
use control_store::Store;
use panoptes_control::{AppState, LocalWorker, Scheduler, SharedPool, app, serve_workers};

#[derive(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,
    /// Address remote workers connect to. Omit to run local-only.
    #[arg(long)]
    worker_addr: Option<String>,
    /// Base URL of the model API the *in-process* workers call.
    #[arg(long, default_value = "http://127.0.0.1:9000")]
    model_api: String,
    /// Number of in-process workers (0 to rely purely on remote 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() -> Result<()> {
    let cli = Cli::parse();
    panoptes_control::telemetry::init();
    let store = Store::connect(&cli.db).await?;

    // One pool holds every worker — local and remote look identical to the
    // scheduler.
    let pool = SharedPool::default();

    // In-process workers, all sharing one model client.
    if cli.workers > 0 {
        let client = Arc::new(HttpModelClient::new(cli.model_api, "claude"));
        for i in 0..cli.workers {
            let worker: Arc<dyn WorkerHandle> =
                Arc::new(LocalWorker::new(format!("local-{i}"), client.clone()));
            pool.add(worker);
        }
    }

    // A TCP port remote workers dial into, joining the same pool.
    if let Some(worker_addr) = &cli.worker_addr {
        let listener = tokio::net::TcpListener::bind(worker_addr).await?;
        eprintln!("panoptes-control accepting workers on {worker_addr}");
        let pool = pool.clone();
        tokio::spawn(async move { serve_workers(listener, pool).await });
    }

    let scheduler = Arc::new(Scheduler::with_source(
        store.clone(),
        Arc::new(pool.clone()),
        &cli.out_dir,
    ));

    // One shutdown signal drains the scheduler and stops the server together.
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
    let sched_task = {
        let scheduler = scheduler.clone();
        let rx = shutdown_rx.clone();
        tokio::spawn(async move { scheduler.run_loop(rx, Duration::from_millis(500)).await })
    };

    let listener = tokio::net::TcpListener::bind(&cli.addr).await?;
    eprintln!("panoptes-control listening on http://{}", cli.addr);
    axum::serve(listener, app(AppState { store }))
        .with_graceful_shutdown(async move {
            let _ = tokio::signal::ctrl_c().await;
            let _ = shutdown_tx.send(true);
        })
        .await?;

    // Let the scheduler finish its in-flight run before we exit.
    let _ = sched_task.await;
    Ok(())
}

Run: cargo build -p panoptes-control --bins Expected: builds clean (panoptes-control binary).

The capstone integration test

A full run dispatched across networked workers, one of which dies mid-job. The coordinator must lose no eval and double-count none — at-least-once delivery made safe by the idempotent record. This lives at crates/panoptes-control/tests/distributed.rs.

#![allow(unused)]
fn main() {
//! The payoff: a full run dispatched across *networked* workers, one of which
//! dies mid-job. The coordinator must lose no eval and double-count none —
//! at-least-once delivery made safe by idempotent recording.

use std::sync::Arc;
use std::time::Duration;

use control_core::{JobOutcome, Message, MessageStream, ResponseRecord, RunId, RunStatus, Usage};
use control_store::Store;
use panoptes_control::{Scheduler, SharedPool, serve_workers};
use tokio::net::{TcpListener, TcpStream};

/// Stand up a coordinator worker port + shared pool. Returns the address workers
/// dial and the pool the scheduler reads.
async fn coordinator() -> (std::net::SocketAddr, SharedPool) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let pool = SharedPool::default();
    {
        let pool = pool.clone();
        tokio::spawn(async move { serve_workers(listener, pool).await });
    }
    (addr, pool)
}

/// A worker that answers every Assign with one record per vignette, forever.
fn spawn_reliable_worker(addr: std::net::SocketAddr, id: &str) {
    let id = id.to_string();
    tokio::spawn(async move {
        let sock = TcpStream::connect(addr).await.unwrap();
        let mut conn = MessageStream::new(sock);
        conn.send(&Message::Register {
            worker_id: id,
            capacity: 4,
        })
        .await
        .unwrap();
        while let Ok(Some(Message::Assign { job })) = conn.recv().await {
            let records = job
                .spec
                .vignettes
                .iter()
                .map(|v| ResponseRecord {
                    vignette_id: v.id.clone(),
                    model: job.spec.model.clone(),
                    epoch: job.spec.epoch,
                    prompt: v.prompt.clone(),
                    response: "ok".into(),
                    usage: Usage {
                        input_tokens: 1,
                        output_tokens: 1,
                    },
                })
                .collect();
            conn.send(&Message::Result {
                outcome: JobOutcome {
                    job_id: job.id,
                    records,
                },
            })
            .await
            .unwrap();
        }
    });
}

/// A worker that registers, accepts exactly one job, then crashes without
/// answering — the failure the whole design exists to survive.
fn spawn_dying_worker(addr: std::net::SocketAddr, id: &str) {
    let id = id.to_string();
    tokio::spawn(async move {
        let sock = TcpStream::connect(addr).await.unwrap();
        let mut conn = MessageStream::new(sock);
        conn.send(&Message::Register {
            worker_id: id,
            capacity: 1,
        })
        .await
        .unwrap();
        let _ = conn.recv().await; // take one Assign, then drop the socket
    });
}

async fn wait_for_workers(pool: &SharedPool, n: usize) {
    tokio::time::timeout(Duration::from_secs(2), async {
        while pool.len() < n {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await
    .expect("workers never registered");
}

async fn write_manifest(vignettes: usize) -> std::path::PathBuf {
    let dir = std::env::temp_dir().join(format!("ctl-dist-{}", std::process::id()));
    tokio::fs::create_dir_all(&dir).await.unwrap();
    let path = dir.join("manifest.jsonl");
    let mut text = String::new();
    for i in 0..vignettes {
        text.push_str(&format!("{{\"id\":\"v{i}\",\"prompt\":\"p{i}\"}}\n"));
    }
    tokio::fs::write(&path, text).await.unwrap();
    path
}

#[tokio::test]
async fn a_dying_worker_loses_no_evals_and_double_counts_none() {
    let (addr, pool) = coordinator().await;
    spawn_reliable_worker(addr, "reliable");
    spawn_dying_worker(addr, "doomed");
    wait_for_workers(&pool, 2).await;

    // 2 models × 2 epochs = 4 jobs, each over 3 vignettes = 12 records expected.
    let manifest = write_manifest(3).await;
    let store = Store::in_memory().await.unwrap();
    let run = control_core::Run {
        id: RunId::new(),
        status: RunStatus::Queued,
        created_at: chrono::Utc::now(),
        manifest: manifest.to_string_lossy().into_owned(),
        models: vec!["claude".into(), "gpt".into()],
        epochs: 2,
        job_count: 4,
        done_count: 0,
    };
    store.insert_run(&run).await.unwrap();

    let out = std::env::temp_dir().join(format!("ctl-dist-out-{}", std::process::id()));
    let scheduler = Scheduler::with_source(store.clone(), Arc::new(pool.clone()), &out);

    // Process the whole run — redelivering the doomed worker's job(s) as it dies.
    let processed = scheduler.tick().await.unwrap();
    assert_eq!(processed, Some(run.id));

    let done = store.get_run(run.id).await.unwrap().unwrap();
    assert_eq!(done.status, RunStatus::Done);
    assert_eq!(done.done_count, 4, "every job must complete exactly once");

    // No eval lost, none double-counted: exactly 4 jobs × 3 vignettes.
    let records = store.run_results(run.id).await.unwrap();
    assert_eq!(records.len(), 12);
}
}

Run: cargo test -p panoptes-control --test distributed Expected: PASS, 1 test (a_dying_worker_loses_no_evals_and_double_counts_none).

Running everything

cargo test --workspace

Expected: PASS — 38 tests green across the five crates, cargo clippy --workspace clean. That is the whole control plane: a typed data model, a mockable workload, durable idempotent state, an HTTP API, a retrying scheduler, and a distributed worker fleet that survives a worker dying mid-job.