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.