Add continuum-proxy enrol to connect a gateway with a one-time code

Exchanges a short code from the Setup page for the farm's gateway
credentials and writes them into the config file (owner-only, atomic,
other keys kept). PC installs only; refused on a Proxy Box.

examples/bambu_mapping_check.rs

@@ -47,6 +47,10 @@ async fn main() -> anyhow::Result<()> {
     for f in continuum_proxy::printer::plate_filaments(&bytes, 1)? {
         println!("project filament {}: {} {} #{}", f.index, f.filament_type, f.tray_info_idx, f.color);
     }
+    if let Some(slot) = std::env::var("AMS_SLOT").ok().and_then(|v| v.parse::<u8>().ok()) {
+        println!("(with AMS_SLOT={slot} chosen explicitly)");
+        bambu.set_ams_slot_override(Some(slot));
+    }
     match bambu.filament_source_for(&bytes, 1).await {
         Ok(source) => println!("mapping the production path would use: {source:?}"),
         Err(err) => println!("the production path would REFUSE: {err}"),

src/completion.rs

@@ -105,7 +105,7 @@ pub async fn run(fleet: SharedFleet) {
                 if let Err(err) = printer.refresh_status().await {
                     warn!(printer = %printer.base().id, %err, "could not request a status refresh");
                 }
-                if let Some(state) = printer.print_status().await.and_then(|r| r.gcode_state) {
+                if let Some(state) = printer.status().await.and_then(|s| s.state) {
                     states.push((printer.base().id.clone(), state));
                 }
             }

src/enrol.rs

@@ -0,0 +1,188 @@
+//! `continuum-proxy enrol <cloud-url> <code> [name]`: connect this gateway to
+//! a farm with the one-time code the console shows, instead of copying a
+//! farm id and token between machines.
+//!
+//! The code is traded at `POST /auth/enrol` for the farm id and the gateway
+//! token, which are written into the proxy's config file (owner-only, since
+//! the token is a credential) next to whatever else it already holds. The
+//! next start reads them like any other setting.
+//!
+//! PC mode only. A Proxy Box is configured by a signed bundle, and writing
+//! its installed config from here would bypass exactly the check that exists
+//! to stop that.
+use std::path::Path;
+
+use anyhow::{bail, Context};
+use serde_json::Value;
+
+pub struct Enrolment {
+    pub farm_id: String,
+    pub gateway_token: String,
+}
+
+pub const USAGE: &str = "usage: continuum-proxy enrol <cloud-url> <code> [gateway-name]\n\
+    \n\
+    Connects this gateway to a farm using the one-time code shown in the console\n\
+    (valid for 15 minutes, usable once). <cloud-url> is the API's address, e.g.\n\
+    https://api.example.com. Settings are written to the proxy's config file.";
+
+/// The websocket address of the uplink for an API at `cloud`.
+pub fn uplink_url(cloud: &str) -> anyhow::Result<String> {
+    let cloud = cloud.trim().trim_end_matches('/');
+    let rest = if let Some(rest) = cloud.strip_prefix("https://") {
+        format!("wss://{rest}")
+    } else if let Some(rest) = cloud.strip_prefix("http://") {
+        format!("ws://{rest}")
+    } else {
+        bail!("the cloud address must start with http:// or https:// (got '{cloud}')");
+    };
+    Ok(format!("{rest}/ws/edge/v1"))
+}
+
+pub fn parse_response(body: &str) -> anyhow::Result<Enrolment> {
+    let value: Value = serde_json::from_str(body).context("the cloud's answer was not JSON")?;
+    let field = |name: &str| {
+        value
+            .get(name)
+            .and_then(Value::as_str)
+            .filter(|v| !v.is_empty())
+            .map(str::to_string)
+            .with_context(|| format!("the cloud's answer has no '{name}'"))
+    };
+    Ok(Enrolment { farm_id: field("farmId")?, gateway_token: field("gatewayToken")? })
+}
+
+/// The config file's new text: the four enrolment settings set, everything
+/// else the file already said kept. (Comments are not preserved - the file is
+/// parsed and written back.)
+pub fn merge_config(existing: Option<&str>, gateway_id: &str, enrolment: &Enrolment, uplink_url: &str) -> anyhow::Result<String> {
+    let mut table: toml::Table = match existing {
+        Some(text) => toml::from_str(text).context("the existing config file is not valid TOML; not touching it")?,
+        None => toml::Table::new(),
+    };
+    table.insert("gateway_id".into(), gateway_id.into());
+    table.insert("farm_id".into(), enrolment.farm_id.clone().into());
+    table.insert("gateway_token".into(), enrolment.gateway_token.clone().into());
+    table.insert("uplink_url".into(), uplink_url.into());
+    Ok(toml::to_string(&table)?)
+}
+
+/// Trades `code` for the farm's credentials.
+pub async fn exchange(cloud: &str, code: &str) -> anyhow::Result<Enrolment> {
+    let url = format!("{}/auth/enrol", cloud.trim().trim_end_matches('/'));
+    let body = serde_json::json!({ "code": code }).to_string();
+    let response = reqwest::Client::new()
+        .post(&url)
+        .header("Content-Type", "application/json")
+        .timeout(std::time::Duration::from_secs(20))
+        .body(body)
+        .send()
+        .await
+        .with_context(|| format!("could not reach {cloud}"))?;
+    let status = response.status();
+    let text = response.text().await.unwrap_or_default();
+    if !status.is_success() {
+        let message = serde_json::from_str::<Value>(&text)
+            .ok()
+            .and_then(|v| v.pointer("/error/message").and_then(Value::as_str).map(str::to_string))
+            .unwrap_or_else(|| format!("HTTP {}", status.as_u16()));
+        bail!("the cloud refused the code: {message}");
+    }
+    parse_response(&text)
+}
+
+/// Owner-only, and replaced atomically so a crash cannot leave half a file.
+fn write_private(path: &Path, text: &str) -> anyhow::Result<()> {
+    use std::io::Write;
+    let tmp = path.with_extension("tmp");
+    let mut options = std::fs::OpenOptions::new();
+    options.write(true).create(true).truncate(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    let mut file = options.open(&tmp).with_context(|| format!("writing {tmp:?}"))?;
+    file.write_all(text.as_bytes())?;
+    file.sync_all()?;
+    std::fs::rename(&tmp, path).with_context(|| format!("replacing {path:?}"))?;
+    Ok(())
+}
+
+fn default_gateway_name() -> String {
+    std::env::var("HOSTNAME")
+        .ok()
+        .or_else(|| std::fs::read_to_string("/etc/hostname").ok())
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+        .unwrap_or_else(|| "gateway".to_string())
+}
+
+/// The `enrol` subcommand. `args` are what follows the word `enrol`.
+pub async fn run(args: &[String], bootstrap: &crate::bootstrap::Bootstrap) -> anyhow::Result<()> {
+    if args.iter().any(|a| a == "--help" || a == "-h") {
+        println!("{USAGE}");
+        return Ok(());
+    }
+    let [cloud, code, rest @ ..] = args else {
+        bail!("{USAGE}");
+    };
+    if bootstrap.mode == crate::bootstrap::Mode::ProxyBox {
+        bail!("this is a Proxy Box: it is configured by a signed config bundle on a pendrive, not by enrol");
+    }
+    let uplink = uplink_url(cloud)?;
+    let enrolment = exchange(cloud, code).await?;
+
+    let path = &bootstrap.config_path;
+    let existing = match std::fs::read_to_string(path) {
+        Ok(text) => Some(text),
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => None,
+        Err(err) => bail!("could not read {path:?}: {err}"),
+    };
+    let name = rest.first().cloned().unwrap_or_else(default_gateway_name);
+    let text = merge_config(existing.as_deref(), &name, &enrolment, &uplink)?;
+    write_private(path, &text)?;
+    println!("Enrolled gateway '{name}' in farm {}. Settings written to {path:?}.", enrolment.farm_id);
+    println!("Start the proxy to connect it.");
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_uplink_follows_the_api_scheme() {
+        assert_eq!(uplink_url("https://api.example.com/").unwrap(), "wss://api.example.com/ws/edge/v1");
+        assert_eq!(uplink_url("http://localhost:3060").unwrap(), "ws://localhost:3060/ws/edge/v1");
+        assert!(uplink_url("api.example.com").is_err());
+    }
+
+    #[test]
+    fn the_answer_needs_both_fields() {
+        let ok = parse_response(r#"{"farmId":"f","gatewayToken":"t"}"#).unwrap();
+        assert_eq!((ok.farm_id.as_str(), ok.gateway_token.as_str()), ("f", "t"));
+        assert!(parse_response(r#"{"farmId":"f"}"#).is_err());
+        assert!(parse_response(r#"{"farmId":"f","gatewayToken":""}"#).is_err());
+        assert!(parse_response("not json").is_err());
+    }
+
+    #[test]
+    fn enrolling_keeps_the_rest_of_the_file_and_the_result_parses_as_a_config() {
+        let enrolment = Enrolment { farm_id: "farm-1".into(), gateway_token: "secret".into() };
+        let existing = "go2rtc_bin = \"/opt/go2rtc\"\ngateway_token = \"old\"\n[discovery]\nenabled = false\n";
+        let text = merge_config(Some(existing), "bench", &enrolment, "wss://x/ws/edge/v1").unwrap();
+        let parsed = crate::config::ProxyFile::parse(&text).unwrap();
+        assert_eq!(parsed.gateway_token.as_deref(), Some("secret"), "the new token replaces the old");
+        assert_eq!(parsed.farm_id.as_deref(), Some("farm-1"));
+        assert_eq!(parsed.gateway_id.as_deref(), Some("bench"));
+        assert_eq!(parsed.go2rtc_bin.as_deref(), Some("/opt/go2rtc"));
+        assert!(parsed.discovery.is_some());
+    }
+
+    #[test]
+    fn a_file_that_is_not_toml_is_left_alone() {
+        let enrolment = Enrolment { farm_id: "f".into(), gateway_token: "t".into() };
+        assert!(merge_config(Some("= nonsense ["), "g", &enrolment, "ws://x").is_err());
+    }
+}

src/jobs.rs

@@ -0,0 +1,185 @@
+//! Following the jobs the cloud has handed this gateway, so their outcomes
+//! reach the cloud even if it was unreachable when they happened.
+//!
+//! The cloud owns the queue; this only reports. When a `start_print` is
+//! accepted the job is recorded against its printer, and each tick the
+//! printer's state is compared with it. What the cloud should hear -
+//! started, finished, failed - goes into the store's outbox, and the uplink
+//! re-sends the outbox on every connection until the cloud acknowledges each
+//! event. An outcome reached during an outage, or before a restart, is
+//! therefore still delivered, instead of the cloud guessing from whatever
+//! telemetry it happens to see next.
+use std::time::Duration;
+
+use tracing::{info, warn};
+
+use crate::printer::GenericPrinter;
+use crate::store::{ActiveJob, Store};
+use crate::uplink::SharedFleet;
+
+const TICK: Duration = Duration::from_secs(5);
+/// A job whose printer never reported printing within this long is given up on.
+const NEVER_STARTED_AFTER_MS: i64 = 30 * 60 * 1000;
+
+pub fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+#[derive(Debug, PartialEq)]
+pub enum Step {
+    Nothing,
+    Started,
+    Finished,
+    Failed(String),
+}
+
+/// What a printer's reported state means for the job it was handed.
+///
+/// Nothing is concluded from a state until the job has been seen running: a
+/// printer that has just been told to start still reports the *previous*
+/// job's `FINISH` (or `FAILED`) for a while, and reading that as this job's
+/// outcome would report a print that has not begun as done.
+pub fn next_step(job: &ActiveJob, state: Option<&str>, now_ms: i64) -> Step {
+    let state = state.map(str::to_ascii_uppercase);
+    let running = matches!(state.as_deref(), Some("RUNNING" | "PAUSE" | "PREPARE" | "SLICING"));
+
+    if !job.started {
+        if running {
+            return Step::Started;
+        }
+        if now_ms - job.since_ms > NEVER_STARTED_AFTER_MS {
+            return Step::Failed("the printer never started printing it".into());
+        }
+        return Step::Nothing;
+    }
+
+    match state.as_deref() {
+        Some("FINISH") => Step::Finished,
+        Some("FAILED") => Step::Failed("the printer reported a fault".into()),
+        // Not a finish, not a fault, and no longer printing: stopped on the
+        // machine itself. Klipper and OctoPrint report a cancelled print as
+        // idle.
+        Some("IDLE") => Step::Failed("the printer stopped without finishing it".into()),
+        _ => Step::Nothing,
+    }
+}
+
+/// The gateway accepted a `start_print` for `job_id`.
+pub fn accepted(store: &Store, printer_id: &str, job_id: &str) {
+    if let Err(err) = store.active_job_set(printer_id, job_id, now_ms()) {
+        warn!(%err, %printer_id, %job_id, "could not record the job against its printer");
+    }
+}
+
+/// The accepted start then failed (download, transfer, start).
+pub fn start_failed(store: &Store, printer_id: &str, job_id: &str, reason: &str) {
+    if let Err(err) = store.outbox_push(job_id, "failed", Some(reason), now_ms()) {
+        warn!(%err, %job_id, "could not queue the failure for the cloud");
+    }
+    let _ = store.active_job_clear(printer_id);
+}
+
+pub async fn run(fleet: SharedFleet, store: std::sync::Arc<Store>) {
+    let mut ticker = tokio::time::interval(TICK);
+    loop {
+        ticker.tick().await;
+        let Ok(active) = store.active_jobs() else { continue };
+        if active.is_empty() {
+            continue;
+        }
+        let states: Vec<(ActiveJob, Option<String>)> = {
+            let fleet = fleet.lock().await;
+            let mut out = Vec::new();
+            for job in active {
+                let state = match fleet.iter().find(|p| p.base().id == job.printer_id) {
+                    Some(printer) => printer.status().await.and_then(|s| s.state),
+                    None => None,
+                };
+                out.push((job, state));
+            }
+            out
+        };
+        for (job, state) in states {
+            apply(&store, &job, next_step(&job, state.as_deref(), now_ms()));
+        }
+    }
+}
+
+fn apply(store: &Store, job: &ActiveJob, step: Step) {
+    let result = match &step {
+        Step::Nothing => return,
+        Step::Started => store
+            .outbox_push(&job.job_id, "started", None, now_ms())
+            .and_then(|_| store.active_job_started(&job.printer_id)),
+        Step::Finished => store
+            .outbox_push(&job.job_id, "finished", None, now_ms())
+            .and_then(|_| store.active_job_clear(&job.printer_id)),
+        Step::Failed(reason) => store
+            .outbox_push(&job.job_id, "failed", Some(reason), now_ms())
+            .and_then(|_| store.active_job_clear(&job.printer_id)),
+    };
+    match result {
+        Ok(_) => info!(printer = %job.printer_id, job = %job.job_id, ?step, "job event queued for the cloud"),
+        Err(err) => warn!(%err, job = %job.job_id, "could not queue a job event"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn job(started: bool, since_ms: i64) -> ActiveJob {
+        ActiveJob { printer_id: "p1".into(), job_id: "j1".into(), since_ms, started }
+    }
+
+    #[test]
+    fn a_stale_finish_from_the_previous_job_is_not_this_jobs_outcome() {
+        assert_eq!(next_step(&job(false, 0), Some("FINISH"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(false, 0), Some("FAILED"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(false, 0), Some("IDLE"), 10_000), Step::Nothing);
+    }
+
+    #[test]
+    fn running_marks_it_started_and_then_finish_finishes_it() {
+        assert_eq!(next_step(&job(false, 0), Some("RUNNING"), 10_000), Step::Started);
+        assert_eq!(next_step(&job(true, 0), Some("RUNNING"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(true, 0), Some("PAUSE"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(true, 0), Some("FINISH"), 10_000), Step::Finished);
+    }
+
+    #[test]
+    fn a_fault_or_a_stop_after_starting_is_a_failure() {
+        assert!(matches!(next_step(&job(true, 0), Some("FAILED"), 1), Step::Failed(_)));
+        assert!(matches!(next_step(&job(true, 0), Some("IDLE"), 1), Step::Failed(_)));
+    }
+
+    #[test]
+    fn a_printer_that_goes_quiet_leaves_the_job_alone() {
+        assert_eq!(next_step(&job(true, 0), None, 1), Step::Nothing);
+    }
+
+    #[test]
+    fn a_job_that_never_starts_is_eventually_given_up_on() {
+        assert_eq!(next_step(&job(false, 0), Some("IDLE"), NEVER_STARTED_AFTER_MS), Step::Nothing);
+        assert!(matches!(next_step(&job(false, 0), Some("IDLE"), NEVER_STARTED_AFTER_MS + 1), Step::Failed(_)));
+    }
+
+    #[test]
+    fn applying_steps_fills_the_outbox_and_clears_the_active_job() {
+        let dir = std::env::temp_dir().join(format!("continuum-jobs-test-{}", std::process::id()));
+        let store = Store::open(&dir.join("s.db")).unwrap();
+        accepted(&store, "p1", "j1");
+        let active = store.active_jobs().unwrap().remove(0);
+        apply(&store, &active, Step::Started);
+        let active = store.active_jobs().unwrap().remove(0);
+        assert!(active.started);
+        apply(&store, &active, Step::Finished);
+        assert!(store.active_jobs().unwrap().is_empty());
+        let events: Vec<_> = store.outbox_pending().unwrap().into_iter().map(|e| e.event).collect();
+        assert_eq!(events, ["started", "finished"]);
+        let _ = std::fs::remove_dir_all(dir);
+    }
+}

src/lib.rs

@@ -13,8 +13,10 @@ pub mod camera;
 pub mod completion;
 pub mod config;
 pub mod config_import;
+pub mod enrol;
 pub mod discovery;
 pub mod fleet;
+pub mod jobs;
 pub mod printer;
 pub mod provision;
 pub mod store;

src/main.rs

@@ -13,7 +13,7 @@ use continuum_proxy::bootstrap::{self, Bootstrap};
 use continuum_proxy::config::Config;
 use continuum_proxy::config_import;
 use continuum_proxy::printer::{GenericPrinter, PrinterHandle};
-use continuum_proxy::{camera, completion, fleet, uplink};
+use continuum_proxy::{camera, completion, fleet, jobs, uplink};
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -23,6 +23,13 @@ async fn main() -> anyhow::Result<()> {
 
     dotenvy::dotenv().ok();
 
+    // `continuum-proxy enrol ...` connects this gateway to a farm and exits;
+    // it needs the bootstrap (for where the config file is) but no settings.
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    if args.first().map(String::as_str) == Some("enrol") {
+        return continuum_proxy::enrol::run(&args[1..], &Bootstrap::discover()?).await;
+    }
+
     // Order matters: import first, so a stick present at boot is already
     // the installed config by the time it is read. A rejected stick is
     // logged inside and does not stop startup - the box runs on what it has.
@@ -177,6 +184,7 @@ async fn main() -> anyhow::Result<()> {
             restart_go2rtc.clone(),
         ) => {}
         _ = completion::run(fleet.clone()) => {}
+        _ = jobs::run(fleet.clone(), store.clone()) => {}
         // Returns only after installing a new config from a stick inserted
         // while running. Settings are read once at startup, so the process
         // ends here and relies on its supervisor (systemd `Restart=always`

src/printer/bambu.rs

@@ -246,6 +246,8 @@ pub struct BambuPrinter {
     /// alongside the protocol version it settled on, since the two are
     /// decided together and always used together - see `bambu_mqtt.rs`.
     connection: Option<(super::bambu_mqtt::BambuMqttSession, BambuProtocolVersion)>,
+    /// See `set_ams_slot_override`.
+    ams_slot_override: Option<u8>,
 }
 
 /// Which plate of a multi-plate project gets printed.
@@ -491,9 +493,17 @@ impl BambuPrinter {
             tls,
             preference,
             connection: None,
+            ams_slot_override: None,
         }
     }
 
+    /// The operator's chosen AMS slot for the next print (0-based across all
+    /// units), or `None` to map by the project's own filament. Set by the
+    /// uplink around one `start_print` and cleared after it.
+    pub fn set_ams_slot_override(&mut self, slot: Option<u8>) {
+        self.ams_slot_override = slot;
+    }
+
     pub fn base(&self) -> &PrinterBase {
         &self.base
     }
@@ -758,7 +768,8 @@ impl BambuPrinter {
         .map_err(|err| PrinterError(format!("failed to build pushall request: {err}")))?;
         body["push_target"] = serde_json::Value::from(1);
 
-        session.publish("pushing", &body).await
+        // Unacked: nothing waits on a reply to this, and see `publish_unacked`.
+        session.publish_unacked("pushing", &body).await
     }
 
     /// Just the FTPS transfer: puts `data` in the printer's `cache/`
@@ -874,7 +885,18 @@ impl BambuPrinter {
         // naming a filament that also happens to be on the spool should
         // still use the AMS, because that is what it was sliced for.
         if !trays.is_empty() {
-            match super::map_to_trays(&filaments, &trays) {
+            let mapped = match self.ams_slot_override {
+                Some(slot) => super::map_to_slot(&filaments, &trays, i32::from(slot)),
+                None => super::map_to_trays(&filaments, &trays),
+            };
+            // An explicit slot that cannot be honoured is an error, never a
+            // fallback to some other spool.
+            if self.ams_slot_override.is_some() {
+                if let Err(err) = &mapped {
+                    return Err(PrinterError(format!("{}: {err}", self.base.name)));
+                }
+            }
+            match mapped {
                 Ok(mapping) => {
                     tracing::info!(
                         printer = %self.base.name,

src/printer/bambu_commands/print.rs

@@ -160,7 +160,11 @@ impl PrintReport {
         self.envelope = newer.envelope;
         self.msg = newer.msg;
 
-        self.ams = newer.ams.or(self.ams.take());
+        // A partial `ams` object (no unit list) must not wipe the units already known.
+        self.ams = match newer.ams {
+            Some(partial) if partial.ams.is_empty() => self.ams.take(),
+            other => other.or(self.ams.take()),
+        };
         self.ams_exist_bits = newer.ams_exist_bits.or(self.ams_exist_bits.take());
         self.insert_flag = newer.insert_flag.or(self.insert_flag.take());
         self.power_on_flag = newer.power_on_flag.or(self.power_on_flag.take());
@@ -244,6 +248,10 @@ pub struct LightReport {
 /// actually sends.
 #[derive(serde::Deserialize, Debug, Clone)]
 pub struct AmsState {
+    /// Absent from the partial `ams` objects a P1P pushes mid-print (found on
+    /// the first real dispatch, 2026-10-29: every update was dropped as
+    /// "missing field `ams`" and the job could never complete).
+    #[serde(default)]
     pub ams: Vec<AmsUnit>,
 }
 

src/printer/bambu_mqtt.rs

@@ -76,6 +76,9 @@ use super::PrinterError;
 type PendingTable = Arc<Mutex<HashMap<String, oneshot::Sender<Value>>>>;
 
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// How long a publish may wait for room in the MQTT request queue.
+const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
 /// Both rumqttc clients default to a 10KB max packet size in *both*
 /// directions - fine for a request, nowhere near enough for a reply:
 /// confirmed live against a real H2C, whose `get_version` reply alone
@@ -359,18 +362,38 @@ impl BambuMqttSession {
     /// `publish` above and `GenericPrinter::publish` (a raw pass-through,
     /// no envelope opinion at all) are built on.
     pub async fn publish_raw(&self, payload: String) -> Result<(), PrinterError> {
+        self.publish_with_qos(payload, true).await
+    }
+
+    /// Like `publish`, but QoS 0: for requests that need no acknowledgement
+    /// (a status refresh). QoS 1 publishes the printer never acknowledges
+    /// fill rumqttc's in-flight window, after which every publish blocks
+    /// forever - found on the first real dispatch, where that froze the whole
+    /// proxy about 4.5 minutes in, because callers hold the fleet lock.
+    pub async fn publish_unacked(&self, category: &str, body: &Value) -> Result<(), PrinterError> {
+        self.publish_with_qos(serde_json::json!({ category: body }).to_string(), false).await
+    }
+
+    async fn publish_with_qos(&self, payload: String, acked: bool) -> Result<(), PrinterError> {
         let topic = format!("device/{}/request", self.sn);
 
+        // Bounded: a publish waits for room in rumqttc's request queue, and an
+        // unbounded wait here wedges every caller that holds the fleet lock.
+        let publishing = async {
             match &self.client {
                 BambuMqttClient::V1(client) => client
-                .publish(topic, QoS::AtLeastOnce, false, payload)
+                    .publish(topic, if acked { QoS::AtLeastOnce } else { QoS::AtMostOnce }, false, payload)
                     .await
                     .map_err(|err| PrinterError(format!("publish failed: {err}"))),
                 BambuMqttClient::V2(client) => client
-                .publish(topic, QoSV5::AtLeastOnce, false, payload)
+                    .publish(topic, if acked { QoSV5::AtLeastOnce } else { QoSV5::AtMostOnce }, false, payload)
                     .await
                     .map_err(|err| PrinterError(format!("publish failed: {err}"))),
             }
+        };
+        tokio::time::timeout(PUBLISH_TIMEOUT, publishing)
+            .await
+            .map_err(|_| PrinterError(format!("{} publish timed out after {PUBLISH_TIMEOUT:?} - the MQTT queue is not draining", self.sn)))?
     }
 
     /// The current in-memory `print.push_status` object, if at least one

src/printer/http.rs

@@ -0,0 +1,202 @@
+//! The HTTP plumbing the REST-speaking vendors (Klipper's Moonraker,
+//! OctoPrint, PrusaLink) share, so each driver is only its own endpoints.
+//!
+//! reqwest is built without its `json` and `multipart` features (see
+//! Cargo.toml), so bodies are made by hand here: JSON through `serde_json`
+//! and the one multipart upload as a few lines of bytes.
+use std::time::Duration;
+
+use reqwest::Method;
+use serde_json::Value;
+
+use super::PrinterError;
+
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+/// A project file over a Pi's Wi-Fi can be tens of megabytes.
+const UPLOAD_TIMEOUT: Duration = Duration::from_secs(600);
+
+#[derive(Clone)]
+pub struct Http {
+    client: reqwest::Client,
+    base: String,
+    api_key: Option<String>,
+}
+
+/// `host` as the operator wrote it: a bare `192.168.0.9`, `host:port`, or a
+/// full `http(s)://` URL. A bare host gets `default_port`.
+fn base_url(host: &str, default_port: u16) -> String {
+    let host = host.trim().trim_end_matches('/');
+    if host.starts_with("http://") || host.starts_with("https://") {
+        host.to_string()
+    } else if host.contains(':') {
+        format!("http://{host}")
+    } else {
+        format!("http://{host}:{default_port}")
+    }
+}
+
+impl Http {
+    pub fn new(host: &str, default_port: u16, api_key: Option<&str>) -> Self {
+        Self {
+            client: reqwest::Client::new(),
+            base: base_url(host, default_port),
+            api_key: api_key.filter(|k| !k.is_empty()).map(str::to_string),
+        }
+    }
+
+    async fn send(
+        &self,
+        method: Method,
+        path: &str,
+        content_type: Option<&str>,
+        extra_headers: &[(&str, &str)],
+        body: Option<Vec<u8>>,
+        timeout: Duration,
+    ) -> Result<Value, PrinterError> {
+        let url = format!("{}{path}", self.base);
+        let mut request = self.client.request(method.clone(), &url).timeout(timeout);
+        if let Some(key) = &self.api_key {
+            request = request.header("X-Api-Key", key);
+        }
+        if let Some(content_type) = content_type {
+            request = request.header("Content-Type", content_type);
+        }
+        for (name, value) in extra_headers {
+            request = request.header(*name, *value);
+        }
+        if let Some(body) = body {
+            request = request.body(body);
+        }
+
+        // The URL is left out of errors on purpose: it is the printer's
+        // address, which the caller already names.
+        let response = request.send().await.map_err(|err| PrinterError(format!("{method} {path}: {err}")))?;
+        let status = response.status();
+        let bytes = response.bytes().await.map_err(|err| PrinterError(format!("{method} {path}: {err}")))?;
+        if !status.is_success() {
+            let text = String::from_utf8_lossy(&bytes);
+            let text: String = text.trim().chars().take(200).collect();
+            return Err(PrinterError(format!("{method} {path}: HTTP {} {text}", status.as_u16())));
+        }
+        // An empty body (204, or a 201 with nothing) is a success, not a
+        // parse error; so is a non-JSON one, which only the callers that
+        // read the answer care about.
+        Ok(serde_json::from_slice(&bytes).unwrap_or(Value::Null))
+    }
+
+    pub async fn get_json(&self, path: &str) -> Result<Value, PrinterError> {
+        self.send(Method::GET, path, None, &[], None, REQUEST_TIMEOUT).await
+    }
+
+    pub async fn post_json(&self, path: &str, body: &Value) -> Result<Value, PrinterError> {
+        let body = body.to_string().into_bytes();
+        self.send(Method::POST, path, Some("application/json"), &[], Some(body), REQUEST_TIMEOUT).await
+    }
+
+    pub async fn post_empty(&self, path: &str) -> Result<Value, PrinterError> {
+        self.send(Method::POST, path, None, &[], None, REQUEST_TIMEOUT).await
+    }
+
+    pub async fn put_empty(&self, path: &str) -> Result<Value, PrinterError> {
+        self.send(Method::PUT, path, None, &[], None, REQUEST_TIMEOUT).await
+    }
+
+    pub async fn delete(&self, path: &str) -> Result<Value, PrinterError> {
+        self.send(Method::DELETE, path, None, &[], None, REQUEST_TIMEOUT).await
+    }
+
+    pub async fn put_bytes(&self, path: &str, headers: &[(&str, &str)], data: Vec<u8>) -> Result<Value, PrinterError> {
+        self.send(Method::PUT, path, Some("application/octet-stream"), headers, Some(data), UPLOAD_TIMEOUT).await
+    }
+
+    /// A `multipart/form-data` POST: the text `fields`, then `data` as the
+    /// part named `file_field`.
+    pub async fn post_file(
+        &self,
+        path: &str,
+        fields: &[(&str, &str)],
+        file_field: &str,
+        file_name: &str,
+        data: &[u8],
+    ) -> Result<Value, PrinterError> {
+        let boundary = format!(
+            "continuum-{:x}{:x}",
+            std::process::id(),
+            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0)
+        );
+        let body = multipart_body(&boundary, fields, file_field, file_name, data);
+        let content_type = format!("multipart/form-data; boundary={boundary}");
+        self.send(Method::POST, path, Some(&content_type), &[], Some(body), UPLOAD_TIMEOUT).await
+    }
+}
+
+fn multipart_body(boundary: &str, fields: &[(&str, &str)], file_field: &str, file_name: &str, data: &[u8]) -> Vec<u8> {
+    let mut body = Vec::with_capacity(data.len() + 512);
+    for (name, value) in fields {
+        body.extend_from_slice(
+            format!("--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n").as_bytes(),
+        );
+    }
+    body.extend_from_slice(
+        format!(
+            "--{boundary}\r\nContent-Disposition: form-data; name=\"{file_field}\"; filename=\"{file_name}\"\r\nContent-Type: application/octet-stream\r\n\r\n"
+        )
+        .as_bytes(),
+    );
+    body.extend_from_slice(data);
+    body.extend_from_slice(format!("\r\n--{boundary}--\r\n").as_bytes());
+    body
+}
+
+/// The file name to give the printer: the last path component, and nothing
+/// that could break out of a multipart header or a URL path.
+pub fn safe_file_name(file_name: &str) -> Result<String, PrinterError> {
+    let name = file_name.rsplit(['/', '\\']).next().unwrap_or("").trim();
+    if name.is_empty() || name.starts_with('.') || name.chars().any(|c| c.is_control() || c == '"') {
+        return Err(PrinterError(format!("'{file_name}' is not a usable file name")));
+    }
+    Ok(name.to_string())
+}
+
+/// Percent-encodes `name` for use as one URL path segment.
+pub fn encode_segment(name: &str) -> String {
+    let mut out = String::new();
+    for byte in name.bytes() {
+        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
+            out.push(byte as char);
+        } else {
+            out.push_str(&format!("%{byte:02X}"));
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_bare_host_gets_the_default_port() {
+        assert_eq!(base_url("192.168.0.9", 7125), "http://192.168.0.9:7125");
+        assert_eq!(base_url("printer.lan:80", 7125), "http://printer.lan:80");
+        assert_eq!(base_url("https://p.example/", 7125), "https://p.example");
+    }
+
+    #[test]
+    fn file_names_are_reduced_to_a_safe_last_component() {
+        assert_eq!(safe_file_name("jobs/../a b.gcode").unwrap(), "a b.gcode");
+        assert!(safe_file_name("..").is_err());
+        assert!(safe_file_name("x\".gcode").is_err());
+        assert!(safe_file_name("").is_err());
+        assert_eq!(encode_segment("a b+c.gcode"), "a%20b%2Bc.gcode");
+    }
+
+    #[test]
+    fn the_multipart_body_has_fields_then_the_file() {
+        let body = multipart_body("B", &[("print", "true")], "file", "x.gcode", b"G1");
+        let text = String::from_utf8(body).unwrap();
+        assert!(text.starts_with("--B\r\nContent-Disposition: form-data; name=\"print\"\r\n\r\ntrue\r\n"));
+        assert!(text.contains("name=\"file\"; filename=\"x.gcode\""));
+        assert!(text.ends_with("G1\r\n--B--\r\n"));
+    }
+}

src/printer/klipper.rs

@@ -1,75 +1,304 @@
-use super::{GenericPrinter, PrinterBase, PrinterError};
+//! Klipper, through Moonraker's HTTP API (default port 7125).
+//!
+//! Written from Moonraker's published API and exercised against a mock
+//! server in the tests below - no real Klipper machine has been on the
+//! bench, so treat the field names as unconfirmed until one has.
+use std::time::Duration;
+
+use serde_json::{json, Value};
+
+use super::http::{safe_file_name, Http};
+use super::polled::StatusFeed;
+use super::{GenericPrinter, PrintControl, PrinterBase, PrinterError, PrinterStatus};
+
+const MOONRAKER_PORT: u16 = 7125;
+const POLL_EVERY: Duration = Duration::from_secs(2);
+
+/// What one `objects/query` asks for - everything `parse_status` reads.
+const STATUS_QUERY: &str =
+    "/printer/objects/query?print_stats&display_status&extruder&heater_bed&virtual_sdcard";
 
 pub struct KlipperPrinter {
     base: PrinterBase,
+    http: Http,
+    feed: Option<StatusFeed>,
 }
 
 impl KlipperPrinter {
     pub fn new(id: &str, name: &str, host: &str) -> Self {
         Self {
             base: PrinterBase { id: id.into(), name: name.into(), host: host.into() },
+            http: Http::new(host, MOONRAKER_PORT, None),
+            feed: None,
+        }
+    }
+
+    pub fn status(&self) -> Option<PrinterStatus> {
+        self.feed.as_ref().and_then(StatusFeed::latest)
+    }
+}
+
+/// Moonraker's `print_stats.state` in the vocabulary the cloud already
+/// understands (the strings Bambu reports; see `mapStatus` in the backend).
+fn map_state(state: &str) -> Option<&'static str> {
+    match state {
+        "standby" => Some("IDLE"),
+        "printing" => Some("RUNNING"),
+        "paused" => Some("PAUSE"),
+        "complete" => Some("FINISH"),
+        // A cancelled print leaves nothing running; the machine is free.
+        "cancelled" => Some("IDLE"),
+        "error" => Some("FAILED"),
+        _ => None,
+    }
+}
+
+/// The `result.status` object of an `objects/query` answer.
+fn parse_status(answer: &Value) -> Option<PrinterStatus> {
+    let status = answer.get("result")?.get("status")?;
+    let stats = status.get("print_stats")?;
+    let state = stats.get("state")?.as_str()?;
+    let number = |v: Option<&Value>| v.and_then(Value::as_f64).map(|n| n as f32);
+
+    let progress = number(status.get("virtual_sdcard").and_then(|s| s.get("progress"))).unwrap_or(0.0);
+    let percent = match state {
+        "complete" => 100,
+        "printing" | "paused" => (progress * 100.0).round() as i32,
+        _ => 0,
+    };
+    let duration = number(stats.get("print_duration")).unwrap_or(0.0);
+    // Klipper has no estimate of its own; this is elapsed time scaled by
+    // progress, which is wrong early in a print and settles as it goes.
+    let remaining_minutes = (matches!(state, "printing" | "paused") && progress > 0.01)
+        .then(|| ((duration / progress - duration) / 60.0).round() as i32);
+    let layer_info = stats.get("info");
+    let int = |v: Option<&Value>| v.and_then(Value::as_i64).map(|n| n as i32);
+
+    Some(PrinterStatus {
+        state: map_state(state).map(str::to_string),
+        percent: Some(percent),
+        layer: int(layer_info.and_then(|i| i.get("current_layer"))),
+        total_layers: int(layer_info.and_then(|i| i.get("total_layer"))),
+        remaining_minutes,
+        job_name: stats
+            .get("filename")
+            .and_then(Value::as_str)
+            .filter(|f| !f.is_empty())
+            .map(str::to_string),
+        nozzle_temp: number(status.get("extruder").and_then(|e| e.get("temperature"))),
+        nozzle_target: number(status.get("extruder").and_then(|e| e.get("target"))),
+        bed_temp: number(status.get("heater_bed").and_then(|e| e.get("temperature"))),
+        bed_target: number(status.get("heater_bed").and_then(|e| e.get("target"))),
+        print_error: (state == "error").then_some(1),
+        ..PrinterStatus::default()
+    })
+}
+
+async fn poll(http: &Http) -> Option<PrinterStatus> {
+    match http.get_json(STATUS_QUERY).await {
+        Ok(answer) => parse_status(&answer),
+        // Moonraker answers 503 to object queries while Klipper itself is
+        // down, but still answers `/printer/info` - which is the
+        // difference between "off" and "faulted".
+        Err(_) => {
+            let info = http.get_json("/printer/info").await.ok()?;
+            let state = info.get("result")?.get("state")?.as_str()?;
+            matches!(state, "error" | "shutdown").then(|| PrinterStatus {
+                state: Some("FAILED".into()),
+                print_error: Some(1),
+                ..PrinterStatus::default()
+            })
         }
     }
 }
 
 impl GenericPrinter for KlipperPrinter {
-    /// OVERRIDE: real version is `POST /printer/print/pause`, `/resume` or
+    /// Note Klipper calls it "cancel" where Bambu calls it "stop".
-    /// `/cancel` on Moonraker - note Klipper calls it "cancel" where Bambu
+    async fn control_print(&mut self, action: PrintControl) -> Result<(), PrinterError> {
-    /// calls it "stop".
-    async fn control_print(&mut self, action: super::PrintControl) -> Result<(), PrinterError> {
         let endpoint = match action {
-            super::PrintControl::Pause => "pause",
+            PrintControl::Pause => "pause",
-            super::PrintControl::Resume => "resume",
+            PrintControl::Resume => "resume",
-            super::PrintControl::Stop => "cancel",
+            PrintControl::Stop => "cancel",
         };
-        println!("[{}] POST /printer/print/{endpoint}", self.base.name);
+        self.http.post_empty(&format!("/printer/print/{endpoint}")).await.map(|_| ())
-        Ok(())
     }
 
-
     fn base(&self) -> &PrinterBase {
         &self.base
     }
 
-    /// OVERRIDE: a third, again-different body - Moonraker's real feed is a
+    /// Checks Moonraker answers, then starts the status poller. Klipper
-    /// WebSocket (see `adapters::moonraker`). This also shows the `Err`
+    /// itself being down is not a connect failure: Moonraker is up and
-    /// path: an empty `host` fails immediately, same as any other vendor
+    /// will report it.
-    /// would if it couldn't reach its printer.
     async fn connect(&mut self) -> Result<(), PrinterError> {
-        if self.base.host.is_empty() {
+        if self.base.host.trim().is_empty() {
             return Err(PrinterError(format!("{} has no host configured", self.base.name)));
         }
-        println!("[{}] connecting over WebSocket (Moonraker)", self.base.name);
+        self.http
+            .get_json("/printer/info")
+            .await
+            .map_err(|err| PrinterError(format!("{} did not answer as Moonraker: {err}", self.base.name)))?;
+        let http = self.http.clone();
+        self.feed = Some(StatusFeed::start(POLL_EVERY, move || {
+            let http = http.clone();
+            async move { poll(&http).await }
+        }));
         Ok(())
     }
 
+    /// Nothing to publish to: Moonraker is request/response.
     async fn publish(&mut self, _payload: String) -> Result<(), PrinterError> {
-        // TODO: add publishing via HTTP
+        Err(PrinterError("Klipper has no publish channel".into()))
-        Ok(())
     }
 
-    /// OVERRIDE: Moonraker's own **hypothetical** native API, not gcode injection - no
+    /// `M106` works on Klipper for the part-cooling fan, same as elsewhere.
-    /// percent-to-PWM conversion needed since it already takes a
-    /// percentage directly. Real version:
-    /// `POST /printer/gcode/script?script=SET_FAN_SPEED SPEED=0.NN`.
     async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
-        println!("[{}] Moonraker: SET_FAN_SPEED SPEED={:.2}", self.base.name, percent as f32 / 100.0);
+        self.send_gcode_line(format!("M106 S{}", super::percent_to_pwm(percent))).await
-        Ok(())
     }
 
-    /// OVERRIDE: real version is a multipart `POST /server/files/upload`
+    /// Uploads to the printer's `gcodes` root and starts it (`print=true`),
-    /// with `data` as the file part - Moonraker stores it on the printer's
+    /// like Bambu's upload-then-`project_file`. Klipper runs plain G-code,
-    /// virtual SD card, same upload-then-start model as Bambu/PrusaLink.
+    /// so a slicer project (`.3mf`) is refused here rather than uploaded
+    /// and left to fail on the machine.
     async fn upload_project_file(&mut self, file_name: String, data: Vec<u8>) -> Result<(), PrinterError> {
-        println!("[{}] POST /server/files/upload {file_name} ({} bytes)", self.base.name, data.len());
+        let name = safe_file_name(&file_name)?;
+        let lower = name.to_lowercase();
+        if !(lower.ends_with(".gcode") || lower.ends_with(".gco") || lower.ends_with(".g")) {
+            return Err(PrinterError(format!(
+                "{name} is not G-code - Klipper needs a sliced .gcode file, not a project"
+            )));
+        }
+        if let Some(status) = self.status()
+            && matches!(status.state.as_deref(), Some("RUNNING" | "PAUSE"))
+        {
+            return Err(PrinterError(format!("{} is already printing", self.base.name)));
+        }
+        let answer = self
+            .http
+            .post_file("/server/files/upload", &[("root", "gcodes"), ("print", "true")], "file", &name, &data)
+            .await?;
+        if answer.get("print_started").and_then(Value::as_bool) == Some(false) {
+            return Err(PrinterError(format!("{name} was uploaded but Klipper did not start it")));
+        }
         Ok(())
     }
 
-    /// OVERRIDE: Moonraker's gcode-script endpoint, one line per call - same
-    /// idea as `set_fan_speed`'s `SET_FAN_SPEED` call above, just with an
-    /// arbitrary line instead of a hardcoded one.
     async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
-        println!("[{}] POST /printer/gcode/script?script={gcode_line}", self.base.name);
+        self.http.post_json("/printer/gcode/script", &json!({ "script": gcode_line })).await.map(|_| ())
-        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::printer::mock_http::MockServer;
+
+    const PRINTING: &str = r#"{"result":{"status":{
+        "print_stats":{"state":"printing","filename":"cube.gcode","print_duration":600.0,"info":{"current_layer":12,"total_layer":60}},
+        "virtual_sdcard":{"progress":0.25},
+        "extruder":{"temperature":214.8,"target":215.0},
+        "heater_bed":{"temperature":59.9,"target":60.0}}}}"#;
+
+    #[test]
+    fn a_printing_machine_maps_to_the_shared_vocabulary() {
+        let status = parse_status(&serde_json::from_str(PRINTING).unwrap()).unwrap();
+        assert_eq!(status.state.as_deref(), Some("RUNNING"));
+        assert_eq!(status.percent, Some(25));
+        assert_eq!((status.layer, status.total_layers), (Some(12), Some(60)));
+        // 600 s for a quarter: 1800 s more.
+        assert_eq!(status.remaining_minutes, Some(30));
+        assert_eq!(status.job_name.as_deref(), Some("cube.gcode"));
+        assert_eq!(status.nozzle_temp, Some(214.8));
+        assert_eq!(status.bed_target, Some(60.0));
+        assert_eq!(status.print_error, None);
+    }
+
+    #[test]
+    fn complete_is_100_percent_and_error_carries_a_code() {
+        let done = r#"{"result":{"status":{"print_stats":{"state":"complete","filename":"a.gcode"},"virtual_sdcard":{"progress":1.0}}}}"#;
+        let status = parse_status(&serde_json::from_str(done).unwrap()).unwrap();
+        assert_eq!((status.state.as_deref(), status.percent), (Some("FINISH"), Some(100)));
+
+        let failed = r#"{"result":{"status":{"print_stats":{"state":"error"}}}}"#;
+        let status = parse_status(&serde_json::from_str(failed).unwrap()).unwrap();
+        assert_eq!((status.state.as_deref(), status.print_error), (Some("FAILED"), Some(1)));
+    }
+
+    #[tokio::test]
+    async fn control_gcode_and_fan_go_to_the_right_endpoints() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#);
+        mock.answer("POST /printer/print/cancel", 200, r#"{"result":"ok"}"#);
+        mock.answer("POST /printer/gcode/script", 200, r#"{"result":"ok"}"#);
+        let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host());
+        printer.connect().await.unwrap();
+
+        printer.control_print(PrintControl::Stop).await.unwrap();
+        assert_eq!(mock.sent("POST", "/printer/print/cancel").len(), 1);
+
+        printer.set_fan_speed(50).await.unwrap();
+        let sent = mock.sent("POST", "/printer/gcode/script");
+        assert_eq!(sent[0].text(), r#"{"script":"M106 S127"}"#);
+    }
+
+    #[tokio::test]
+    async fn connect_fails_when_nothing_answers_as_moonraker() {
+        let mock = MockServer::start().await; // every route is a 404
+        let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host());
+        assert!(printer.connect().await.is_err());
+    }
+
+    #[tokio::test]
+    async fn the_poller_feeds_status_and_clears_it_when_the_printer_goes() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#);
+        mock.answer("GET /printer/objects/query", 200, PRINTING);
+        let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host());
+        printer.connect().await.unwrap();
+
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        assert_eq!(printer.status().unwrap().percent, Some(25));
+
+        // Klipper shuts down: queries 503, /printer/info says why.
+        mock.answer("GET /printer/objects/query", 503, "{}");
+        mock.answer("GET /printer/info", 200, r#"{"result":{"state":"shutdown"}}"#);
+        tokio::time::sleep(Duration::from_millis(2500)).await;
+        assert_eq!(printer.status().unwrap().state.as_deref(), Some("FAILED"));
+    }
+
+    #[tokio::test]
+    async fn upload_sends_the_file_with_print_true_and_refuses_a_project() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#);
+        mock.answer("POST /server/files/upload", 201, r#"{"print_started":true}"#);
+        let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host());
+        printer.connect().await.unwrap();
+
+        printer.upload_project_file("jobs/cube.gcode".into(), b"G28\nG1 X1\n".to_vec()).await.unwrap();
+        let sent = mock.sent("POST", "/server/files/upload");
+        assert_eq!(sent.len(), 1);
+        let body = sent[0].text();
+        assert!(body.contains("name=\"print\"\r\n\r\ntrue"));
+        assert!(body.contains("filename=\"cube.gcode\""));
+        assert!(body.contains("G28\nG1 X1\n"));
+        assert!(sent[0].header("content-type").unwrap().starts_with("multipart/form-data; boundary="));
+
+        let err = printer.upload_project_file("x.3mf".into(), vec![1]).await.unwrap_err();
+        assert!(err.0.contains("not G-code"), "{err}");
+        assert_eq!(mock.sent("POST", "/server/files/upload").len(), 1);
+    }
+
+    #[tokio::test]
+    async fn an_upload_is_refused_while_a_print_runs() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /printer/info", 200, r#"{"result":{"state":"ready"}}"#);
+        mock.answer("GET /printer/objects/query", 200, PRINTING);
+        let mut printer = KlipperPrinter::new("k1", "Voron", &mock.host());
+        printer.connect().await.unwrap();
+        tokio::time::sleep(Duration::from_millis(300)).await;
+
+        let err = printer.upload_project_file("b.gcode".into(), vec![1]).await.unwrap_err();
+        assert!(err.0.contains("already printing"), "{err}");
+        assert!(mock.sent("POST", "/server/files/upload").is_empty());
     }
 }

src/printer/mock_http.rs

@@ -0,0 +1,118 @@
+//! A tiny HTTP server for the REST drivers' tests: canned answers per
+//! `METHOD /path` (query string ignored), and a record of what was sent.
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+
+#[derive(Debug, Clone)]
+pub struct Recorded {
+    pub method: String,
+    pub path: String,
+    pub headers: Vec<(String, String)>,
+    pub body: Vec<u8>,
+}
+
+impl Recorded {
+    pub fn header(&self, name: &str) -> Option<&str> {
+        self.headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())
+    }
+    pub fn text(&self) -> String {
+        String::from_utf8_lossy(&self.body).into_owned()
+    }
+}
+
+#[derive(Clone)]
+pub struct MockServer {
+    pub addr: SocketAddr,
+    routes: Arc<Mutex<HashMap<String, (u16, String)>>>,
+    requests: Arc<Mutex<Vec<Recorded>>>,
+}
+
+impl MockServer {
+    pub async fn start() -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let server = Self {
+            addr: listener.local_addr().unwrap(),
+            routes: Arc::default(),
+            requests: Arc::default(),
+        };
+        let shared = server.clone();
+        tokio::spawn(async move {
+            loop {
+                let Ok((stream, _)) = listener.accept().await else { return };
+                let shared = shared.clone();
+                tokio::spawn(async move { shared.handle(stream).await });
+            }
+        });
+        server
+    }
+
+    pub fn host(&self) -> String {
+        self.addr.to_string()
+    }
+
+    /// `route` is e.g. `"GET /printer/info"`.
+    pub fn answer(&self, route: &str, status: u16, body: &str) {
+        self.routes.lock().unwrap().insert(route.to_string(), (status, body.to_string()));
+    }
+
+    pub fn requests(&self) -> Vec<Recorded> {
+        self.requests.lock().unwrap().clone()
+    }
+
+    pub fn sent(&self, method: &str, path: &str) -> Vec<Recorded> {
+        self.requests().into_iter().filter(|r| r.method == method && r.path == path).collect()
+    }
+
+    async fn handle(&self, mut stream: tokio::net::TcpStream) {
+        let mut buf = Vec::new();
+        let mut chunk = [0u8; 8192];
+        let head_end = loop {
+            if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
+                break i + 4;
+            }
+            match stream.read(&mut chunk).await {
+                Ok(0) | Err(_) => return,
+                Ok(n) => buf.extend_from_slice(&chunk[..n]),
+            }
+        };
+        let head = String::from_utf8_lossy(&buf[..head_end]).into_owned();
+        let mut lines = head.split("\r\n");
+        let mut first = lines.next().unwrap_or("").split(' ');
+        let method = first.next().unwrap_or("").to_string();
+        let target = first.next().unwrap_or("").to_string();
+        let headers: Vec<(String, String)> = lines
+            .filter_map(|l| l.split_once(':').map(|(k, v)| (k.trim().to_string(), v.trim().to_string())))
+            .collect();
+        let length: usize = headers
+            .iter()
+            .find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
+            .and_then(|(_, v)| v.parse().ok())
+            .unwrap_or(0);
+        while buf.len() < head_end + length {
+            match stream.read(&mut chunk).await {
+                Ok(0) | Err(_) => break,
+                Ok(n) => buf.extend_from_slice(&chunk[..n]),
+            }
+        }
+        let path = target.split('?').next().unwrap_or("").to_string();
+        let body = buf[head_end..].to_vec();
+        let (status, answer) = self
+            .routes
+            .lock()
+            .unwrap()
+            .get(&format!("{method} {path}"))
+            .cloned()
+            .unwrap_or((404, "{}".to_string()));
+        self.requests.lock().unwrap().push(Recorded { method, path, headers, body });
+        let response = format!(
+            "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{answer}",
+            answer.len()
+        );
+        let _ = stream.write_all(response.as_bytes()).await;
+        let _ = stream.shutdown().await;
+    }
+}

src/printer/mod.rs

@@ -24,23 +24,28 @@
 //! Real MQTT is built now - `bambu.rs`/`bambu_mqtt.rs` for Bambu's
 //! protocol, `bambu_ftps.rs` for getting files onto a Bambu printer,
 //! `marlin_serial.rs` for a real Marlin serial connection - see each
-//! module's own doc comment. PrusaLink/Klipper's HTTP clients are still
+//! module's own doc comment. Klipper (`klipper.rs`) and OctoPrint/PrusaLink
-//! stubs; that's the next piece to build past this module.
+//! (`octoprint.rs`) are real HTTP clients, tested against mock servers only.
 
 mod bambu;
 mod bambu_camera;
 mod bambu_commands;
 mod bambu_ftps;
 mod bambu_mqtt;
+mod http;
 mod klipper;
 mod marlin_serial;
+#[cfg(test)]
+pub(crate) mod mock_http;
+mod octoprint;
+mod polled;
 mod prusa;
 
 // Reading a `.3mf` well enough to know which filament a plate wants. Kept
 // free of printer types so it is testable without a printer.
 mod project_file;
 pub use project_file::{
-    map_to_trays, normalise_color, plate_filaments, AvailableTray, ProjectFilament,
+    map_to_slot, map_to_trays, normalise_color, plate_filaments, AvailableTray, ProjectFilament,
     ProjectFileError,
 };
 
@@ -54,7 +59,54 @@ pub use bambu_commands::{
     XcamSettings,
 };
 pub use klipper::KlipperPrinter;
-pub use prusa::{OctoPrintFlavour, OctoPrintPrinter, PrusaSerialPrinter};
+pub use octoprint::{OctoPrintFlavour, OctoPrintPrinter};
+pub use polled::StatusFeed;
+pub use prusa::PrusaSerialPrinter;
+
+/// What the cloud is told about a printer's job, whatever the vendor.
+///
+/// `state` uses the strings Bambu reports (`IDLE`, `RUNNING`, `PAUSE`,
+/// `FINISH`, `FAILED`) because the cloud's `mapStatus` and the completion
+/// watcher already speak them; the other vendors translate into them.
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct PrinterStatus {
+    pub state: Option<String>,
+    pub percent: Option<i32>,
+    pub stage: Option<i32>,
+    pub layer: Option<i32>,
+    pub total_layers: Option<i32>,
+    pub remaining_minutes: Option<i32>,
+    pub job_name: Option<String>,
+    pub nozzle_temp: Option<f32>,
+    pub nozzle_target: Option<f32>,
+    pub bed_temp: Option<f32>,
+    pub bed_target: Option<f32>,
+    pub print_error: Option<i32>,
+    pub active_alerts: Option<usize>,
+}
+
+impl From<&PrintReport> for PrinterStatus {
+    fn from(report: &PrintReport) -> Self {
+        Self {
+            state: report.gcode_state.clone(),
+            percent: report.mc_percent,
+            stage: report.stg_cur,
+            layer: report.layer_num,
+            total_layers: report.total_layer_num,
+            remaining_minutes: report.mc_remaining_time,
+            // An idle printer reports `""` rather than omitting the field;
+            // empty-string-as-absent is worth normalising here so the cloud
+            // doesn't have to know that.
+            job_name: report.subtask_name.clone().filter(|s| !s.is_empty()),
+            nozzle_temp: report.nozzle_temper,
+            nozzle_target: report.nozzle_target_temper,
+            bed_temp: report.bed_temper,
+            bed_target: report.bed_target_temper,
+            print_error: report.print_error,
+            active_alerts: report.hms.as_ref().map(|h| h.len()),
+        }
+    }
+}
 
 /// A tiny error type: just a message. `String` would work too - this exists
 /// mainly so `?` has something concrete to convert into everywhere.
@@ -204,6 +256,20 @@ impl PrinterHandle {
         }
     }
 
+    /// The latest vendor-neutral status, if the printer has one. Bambu's
+    /// comes from its MQTT cache, the HTTP vendors' from their pollers; the
+    /// serial printer has none.
+    pub async fn status(&self) -> Option<PrinterStatus> {
+        match self {
+            PrinterHandle::Bambu(p) => p.print_status().await.as_ref().map(PrinterStatus::from),
+            PrinterHandle::OctoPrint(p) => p.status(),
+            PrinterHandle::Klipper(p) => p.status(),
+            PrinterHandle::PrusaSerial(_) => None,
+        }
+    }
+
+    /// Bambu's full report, for what is Bambu-specific (camera, AMS).
+    ///
     /// The latest status this process has for the printer, if it has one.
     ///
     /// An inherent method rather than a `GenericPrinter` trait method,

src/printer/octoprint.rs

@@ -0,0 +1,417 @@
+//! OctoPrint and PrusaLink over HTTP (`X-Api-Key`).
+//!
+//! Written from the two projects' published REST APIs and exercised against
+//! a mock server in the tests below - no real OctoPrint or PrusaLink machine
+//! has been on the bench, so treat the field names as unconfirmed until one
+//! has.
+use std::time::Duration;
+
+use serde_json::{json, Value};
+
+use super::http::{encode_segment, safe_file_name, Http};
+use super::polled::StatusFeed;
+use super::{GenericPrinter, PrintControl, PrinterBase, PrinterError, PrinterStatus};
+
+const POLL_EVERY: Duration = Duration::from_secs(3);
+
+/// Which implementation of the OctoPrint REST API is on the other end.
+///
+/// PrusaLink deliberately implements OctoPrint's API, so one client serves
+/// both - but not identically. PrusaLink adds its own `/api/v1/*` (`info`,
+/// `status`, `job`) which is where the useful detail lives, while OctoPrint
+/// proper has the older `/api/printer` and `/api/job` and a plugin surface
+/// PrusaLink does not have.
+///
+/// This is recorded from config now, while the client is still a stub,
+/// because **no probe can settle it**: both answer `/api/version`, and every
+/// PrusaLink identity endpoint needs a credential, so from outside the two
+/// are indistinguishable (see `discovery::net::classify_octoprint_api`). It
+/// is the operator's answer to a question only they can answer, so it needs
+/// somewhere to live the moment they are asked - otherwise the real client
+/// would have to ask them a second time.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OctoPrintFlavour {
+    /// Prusa's own firmware-hosted implementation, on an MK4/XL/MINI.
+    PrusaLink,
+    /// OctoPrint itself, usually on a Raspberry Pi in front of any printer.
+    Generic,
+}
+
+impl OctoPrintFlavour {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            OctoPrintFlavour::PrusaLink => "PrusaLink",
+            OctoPrintFlavour::Generic => "OctoPrint",
+        }
+    }
+}
+
+
+pub struct OctoPrintPrinter {
+    base: PrinterBase,
+    pub flavour: OctoPrintFlavour,
+    http: Http,
+    feed: Option<StatusFeed>,
+}
+
+impl OctoPrintPrinter {
+    pub fn new(id: &str, name: &str, host: &str, api_key: &str, flavour: OctoPrintFlavour) -> Self {
+        Self {
+            base: PrinterBase { id: id.into(), name: name.into(), host: host.into() },
+            flavour,
+            // Port 80: OctoPrint's usual front (haproxy) and PrusaLink both
+            // answer there. An unusual port goes in `host`.
+            http: Http::new(host, 80, Some(api_key)),
+            feed: None,
+        }
+    }
+
+    pub fn status(&self) -> Option<PrinterStatus> {
+        self.feed.as_ref().and_then(StatusFeed::latest)
+    }
+}
+
+fn number(v: Option<&Value>) -> Option<f32> {
+    v.and_then(Value::as_f64).map(|n| n as f32)
+}
+
+/// OctoPrint's `GET /api/job`, plus `GET /api/printer` for temperatures
+/// (which is a 409 whenever the printer is not operational, hence `Option`).
+fn octoprint_status(job: &Value, printer: Option<&Value>) -> Option<PrinterStatus> {
+    let state = job.get("state")?.as_str()?;
+    let completion = number(job.get("progress").and_then(|p| p.get("completion")));
+    let left = number(job.get("progress").and_then(|p| p.get("printTimeLeft")));
+
+    let (mapped, percent) = if state.starts_with("Printing") || state == "Cancelling" {
+        ("RUNNING", completion.unwrap_or(0.0))
+    } else if state == "Pausing" || state == "Paused" {
+        ("PAUSE", completion.unwrap_or(0.0))
+    } else if state == "Operational" {
+        // OctoPrint has no "finished" state: the job simply stays at 100%.
+        match completion {
+            Some(c) if c >= 100.0 => ("FINISH", 100.0),
+            _ => ("IDLE", 0.0),
+        }
+    } else if state.contains("rror") && !state.starts_with("Offline") {
+        ("FAILED", 0.0)
+    } else {
+        // Offline, still opening the serial port, ...: not a state worth
+        // reporting as the printer's.
+        return None;
+    };
+
+    let tool = printer.and_then(|p| p.get("temperature")).and_then(|t| t.get("tool0"));
+    let bed = printer.and_then(|p| p.get("temperature")).and_then(|t| t.get("bed"));
+    Some(PrinterStatus {
+        state: Some(mapped.into()),
+        percent: Some(percent.round() as i32),
+        remaining_minutes: (mapped == "RUNNING" || mapped == "PAUSE").then_some(left).flatten().map(|s| (s / 60.0).round() as i32),
+        job_name: job
+            .get("job")
+            .and_then(|j| j.get("file"))
+            .and_then(|f| f.get("display").or_else(|| f.get("name")))
+            .and_then(Value::as_str)
+            .map(str::to_string),
+        nozzle_temp: number(tool.and_then(|t| t.get("actual"))),
+        nozzle_target: number(tool.and_then(|t| t.get("target"))),
+        bed_temp: number(bed.and_then(|t| t.get("actual"))),
+        bed_target: number(bed.and_then(|t| t.get("target"))),
+        print_error: (mapped == "FAILED").then_some(1),
+        ..PrinterStatus::default()
+    })
+}
+
+/// PrusaLink's `GET /api/v1/status`, with `GET /api/v1/job` for the file name
+/// when a job exists.
+fn prusalink_status(status: &Value, job: Option<&Value>) -> Option<PrinterStatus> {
+    let printer = status.get("printer")?;
+    let state = printer.get("state")?.as_str()?;
+    let mapped = match state {
+        "IDLE" | "STOPPED" => "IDLE",
+        "BUSY" => "PREPARE",
+        "PRINTING" => "RUNNING",
+        "PAUSED" | "ATTENTION" => "PAUSE",
+        "FINISHED" => "FINISH",
+        "ERROR" => "FAILED",
+        _ => return None,
+    };
+    let running = matches!(mapped, "RUNNING" | "PAUSE");
+    let progress = number(status.get("job").and_then(|j| j.get("progress")));
+    let left = number(status.get("job").and_then(|j| j.get("time_remaining")));
+    Some(PrinterStatus {
+        state: Some(mapped.into()),
+        percent: Some(match mapped {
+            "FINISH" => 100,
+            _ if running => progress.unwrap_or(0.0).round() as i32,
+            _ => 0,
+        }),
+        remaining_minutes: running.then_some(left).flatten().map(|s| (s / 60.0).round() as i32),
+        job_name: job
+            .and_then(|j| j.get("file"))
+            .and_then(|f| f.get("display_name").or_else(|| f.get("name")))
+            .and_then(Value::as_str)
+            .map(str::to_string),
+        nozzle_temp: number(printer.get("temp_nozzle")),
+        nozzle_target: number(printer.get("target_nozzle")),
+        bed_temp: number(printer.get("temp_bed")),
+        bed_target: number(printer.get("target_bed")),
+        print_error: (mapped == "FAILED").then_some(1),
+        ..PrinterStatus::default()
+    })
+}
+
+async fn poll(http: &Http, flavour: OctoPrintFlavour) -> Option<PrinterStatus> {
+    match flavour {
+        OctoPrintFlavour::Generic => {
+            let job = http.get_json("/api/job").await.ok()?;
+            let printer = http.get_json("/api/printer").await.ok();
+            octoprint_status(&job, printer.as_ref())
+        }
+        OctoPrintFlavour::PrusaLink => {
+            let status = http.get_json("/api/v1/status").await.ok()?;
+            let has_job = status.get("job").is_some_and(|j| !j.is_null());
+            let job = if has_job { http.get_json("/api/v1/job").await.ok() } else { None };
+            prusalink_status(&status, job.as_ref())
+        }
+    }
+}
+
+impl GenericPrinter for OctoPrintPrinter {
+    /// OctoPrint: `POST /api/job` with a command. PrusaLink addresses the
+    /// running job by id, so the current job is fetched first.
+    async fn control_print(&mut self, action: PrintControl) -> Result<(), PrinterError> {
+        match self.flavour {
+            OctoPrintFlavour::Generic => {
+                let body = match action {
+                    PrintControl::Pause => json!({ "command": "pause", "action": "pause" }),
+                    PrintControl::Resume => json!({ "command": "pause", "action": "resume" }),
+                    PrintControl::Stop => json!({ "command": "cancel" }),
+                };
+                self.http.post_json("/api/job", &body).await.map(|_| ())
+            }
+            OctoPrintFlavour::PrusaLink => {
+                let status = self.http.get_json("/api/v1/status").await?;
+                let id = status
+                    .get("job")
+                    .and_then(|j| j.get("id"))
+                    .and_then(Value::as_i64)
+                    .ok_or_else(|| PrinterError(format!("{} has no job to {}", self.base.name, action.as_str())))?;
+                match action {
+                    PrintControl::Pause => self.http.put_empty(&format!("/api/v1/job/{id}/pause")).await,
+                    PrintControl::Resume => self.http.put_empty(&format!("/api/v1/job/{id}/resume")).await,
+                    PrintControl::Stop => self.http.delete(&format!("/api/v1/job/{id}")).await,
+                }
+                .map(|_| ())
+            }
+        }
+    }
+
+    fn base(&self) -> &PrinterBase {
+        &self.base
+    }
+
+    /// Checks the printer answers with this key, then starts the poller.
+    async fn connect(&mut self) -> Result<(), PrinterError> {
+        if self.base.host.trim().is_empty() {
+            return Err(PrinterError(format!("{} has no host configured", self.base.name)));
+        }
+        // `/api/version` needs no permissions on either, so it would pass
+        // with a wrong key; these two are the first that check it.
+        let probe = match self.flavour {
+            OctoPrintFlavour::Generic => "/api/job",
+            OctoPrintFlavour::PrusaLink => "/api/v1/status",
+        };
+        self.http
+            .get_json(probe)
+            .await
+            .map_err(|err| PrinterError(format!("{} did not answer as {}: {err}", self.base.name, self.flavour.as_str())))?;
+        let (http, flavour) = (self.http.clone(), self.flavour);
+        self.feed = Some(StatusFeed::start(POLL_EVERY, move || {
+            let http = http.clone();
+            async move { poll(&http, flavour).await }
+        }));
+        Ok(())
+    }
+
+    async fn publish(&mut self, _payload: String) -> Result<(), PrinterError> {
+        Err(PrinterError(format!("{} has no publish channel", self.flavour.as_str())))
+    }
+
+    async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
+        self.send_gcode_line(format!("M106 S{}", super::percent_to_pwm(percent))).await
+    }
+
+    /// Uploads and starts the print. OctoPrint takes a multipart POST to
+    /// `/api/files/local`; PrusaLink a raw `PUT /api/v1/files/local/{name}`
+    /// with `Print-After-Upload`. Both run sliced G-code, not a project.
+    async fn upload_project_file(&mut self, file_name: String, data: Vec<u8>) -> Result<(), PrinterError> {
+        let name = safe_file_name(&file_name)?;
+        let lower = name.to_lowercase();
+        let printable = [".gcode", ".gco", ".g", ".bgcode"].iter().any(|ext| lower.ends_with(ext));
+        if !printable {
+            return Err(PrinterError(format!(
+                "{name} is not G-code - {} needs a sliced file, not a project",
+                self.flavour.as_str()
+            )));
+        }
+        if let Some(status) = self.status()
+            && matches!(status.state.as_deref(), Some("RUNNING" | "PAUSE" | "PREPARE"))
+        {
+            return Err(PrinterError(format!("{} is already printing", self.base.name)));
+        }
+        match self.flavour {
+            OctoPrintFlavour::Generic => self
+                .http
+                .post_file("/api/files/local", &[("select", "true"), ("print", "true")], "file", &name, &data)
+                .await
+                .map(|_| ()),
+            OctoPrintFlavour::PrusaLink => self
+                .http
+                .put_bytes(
+                    &format!("/api/v1/files/local/{}", encode_segment(&name)),
+                    &[("Print-After-Upload", "?1"), ("Overwrite", "?1")],
+                    data,
+                )
+                .await
+                .map(|_| ()),
+        }
+    }
+
+    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
+        match self.flavour {
+            OctoPrintFlavour::Generic => self
+                .http
+                .post_json("/api/printer/command", &json!({ "commands": [gcode_line] }))
+                .await
+                .map(|_| ()),
+            // PrusaLink's API has no G-code endpoint.
+            OctoPrintFlavour::PrusaLink => {
+                Err(PrinterError("PrusaLink cannot be sent G-code lines".into()))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::printer::mock_http::MockServer;
+
+    const JOB: &str = r#"{"job":{"file":{"name":"cube.gcode","display":"cube.gcode"}},
+        "progress":{"completion":40.0,"printTime":600,"printTimeLeft":900},"state":"Printing"}"#;
+    const TEMPS: &str =
+        r#"{"temperature":{"tool0":{"actual":209.5,"target":210.0},"bed":{"actual":60.0,"target":60.0}}}"#;
+
+    #[test]
+    fn octoprint_states_map_to_the_shared_vocabulary() {
+        let status = octoprint_status(&serde_json::from_str(JOB).unwrap(), serde_json::from_str(TEMPS).ok().as_ref()).unwrap();
+        assert_eq!(status.state.as_deref(), Some("RUNNING"));
+        assert_eq!(status.percent, Some(40));
+        assert_eq!(status.remaining_minutes, Some(15));
+        assert_eq!(status.job_name.as_deref(), Some("cube.gcode"));
+        assert_eq!(status.nozzle_target, Some(210.0));
+
+        // Finished: still "Operational", job stays at 100%. No temperatures
+        // (the printer endpoint was a 409) is fine.
+        let done = r#"{"state":"Operational","progress":{"completion":100.0}}"#;
+        let status = octoprint_status(&serde_json::from_str(done).unwrap(), None).unwrap();
+        assert_eq!((status.state.as_deref(), status.percent, status.nozzle_temp), (Some("FINISH"), Some(100), None));
+
+        let idle = r#"{"state":"Operational","progress":{"completion":null}}"#;
+        assert_eq!(octoprint_status(&serde_json::from_str(idle).unwrap(), None).unwrap().state.as_deref(), Some("IDLE"));
+
+        let offline = r#"{"state":"Offline","progress":{}}"#;
+        assert!(octoprint_status(&serde_json::from_str(offline).unwrap(), None).is_none());
+        let broken = r#"{"state":"Error","progress":{}}"#;
+        assert_eq!(octoprint_status(&serde_json::from_str(broken).unwrap(), None).unwrap().print_error, Some(1));
+    }
+
+    #[test]
+    fn prusalink_states_map_to_the_shared_vocabulary() {
+        let status = r#"{"printer":{"state":"PRINTING","temp_nozzle":215.1,"target_nozzle":215.0,"temp_bed":60.2,"target_bed":60.0},
+            "job":{"id":7,"progress":33.0,"time_remaining":1200,"time_printing":300}}"#;
+        let job = r#"{"id":7,"file":{"name":"CUBE~1.GCO","display_name":"cube.gcode"}}"#;
+        let parsed = prusalink_status(&serde_json::from_str(status).unwrap(), serde_json::from_str(job).ok().as_ref()).unwrap();
+        assert_eq!(parsed.state.as_deref(), Some("RUNNING"));
+        assert_eq!((parsed.percent, parsed.remaining_minutes), (Some(33), Some(20)));
+        assert_eq!(parsed.job_name.as_deref(), Some("cube.gcode"));
+        assert_eq!(parsed.bed_temp, Some(60.2));
+
+        let idle = r#"{"printer":{"state":"IDLE","temp_nozzle":25.0,"temp_bed":24.0}}"#;
+        let parsed = prusalink_status(&serde_json::from_str(idle).unwrap(), None).unwrap();
+        assert_eq!((parsed.state.as_deref(), parsed.percent), (Some("IDLE"), Some(0)));
+    }
+
+    #[tokio::test]
+    async fn octoprint_controls_upload_and_send_the_key() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /api/job", 200, r#"{"state":"Operational","progress":{}}"#);
+        mock.answer("POST /api/job", 204, "");
+        mock.answer("POST /api/printer/command", 204, "");
+        mock.answer("POST /api/files/local", 201, r#"{"done":true}"#);
+        let mut printer = OctoPrintPrinter::new("o1", "Ender", &mock.host(), "KEY123", OctoPrintFlavour::Generic);
+        printer.connect().await.unwrap();
+
+        printer.control_print(PrintControl::Resume).await.unwrap();
+        let sent = mock.sent("POST", "/api/job");
+        assert_eq!(sent[0].text(), r#"{"action":"resume","command":"pause"}"#);
+        assert_eq!(sent[0].header("x-api-key"), Some("KEY123"));
+        printer.control_print(PrintControl::Stop).await.unwrap();
+        assert_eq!(mock.sent("POST", "/api/job")[1].text(), r#"{"command":"cancel"}"#);
+
+        printer.set_fan_speed(100).await.unwrap();
+        assert_eq!(mock.sent("POST", "/api/printer/command")[0].text(), r#"{"commands":["M106 S255"]}"#);
+
+        printer.upload_project_file("a.gcode".into(), b"G28".to_vec()).await.unwrap();
+        let body = mock.sent("POST", "/api/files/local")[0].text();
+        assert!(body.contains("name=\"print\"\r\n\r\ntrue") && body.contains("filename=\"a.gcode\""));
+        assert!(printer.upload_project_file("a.3mf".into(), vec![1]).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn a_wrong_key_fails_connect() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /api/job", 403, r#"{"error":"forbidden"}"#);
+        let mut printer = OctoPrintPrinter::new("o1", "Ender", &mock.host(), "bad", OctoPrintFlavour::Generic);
+        assert!(printer.connect().await.unwrap_err().0.contains("403"));
+    }
+
+    #[tokio::test]
+    async fn prusalink_addresses_the_running_job_by_id() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /api/v1/status", 200, r#"{"printer":{"state":"PRINTING"},"job":{"id":7,"progress":10.0}}"#);
+        mock.answer("GET /api/v1/job", 200, r#"{"id":7,"file":{"display_name":"c.gcode"}}"#);
+        mock.answer("PUT /api/v1/job/7/pause", 204, "");
+        mock.answer("DELETE /api/v1/job/7", 204, "");
+        let mut printer = OctoPrintPrinter::new("p1", "MK4", &mock.host(), "K", OctoPrintFlavour::PrusaLink);
+        printer.connect().await.unwrap();
+
+        printer.control_print(PrintControl::Pause).await.unwrap();
+        printer.control_print(PrintControl::Stop).await.unwrap();
+        assert_eq!(mock.sent("PUT", "/api/v1/job/7/pause").len(), 1);
+        assert_eq!(mock.sent("DELETE", "/api/v1/job/7").len(), 1);
+        assert!(printer.send_gcode_line("M106".into()).await.is_err());
+
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        assert_eq!(printer.status().unwrap().job_name.as_deref(), Some("c.gcode"));
+        // A job is running, so an upload is refused before anything is sent.
+        let err = printer.upload_project_file("n.bgcode".into(), vec![1]).await.unwrap_err();
+        assert!(err.0.contains("already printing"), "{err}");
+        assert!(mock.sent("PUT", "/api/v1/files/local/n.bgcode").is_empty());
+    }
+
+    #[tokio::test]
+    async fn prusalink_upload_is_a_put_with_print_after_upload() {
+        let mock = MockServer::start().await;
+        mock.answer("GET /api/v1/status", 200, r#"{"printer":{"state":"IDLE"}}"#);
+        mock.answer("PUT /api/v1/files/local/my%20part.gcode", 201, "");
+        let mut printer = OctoPrintPrinter::new("p1", "MK4", &mock.host(), "K", OctoPrintFlavour::PrusaLink);
+        printer.connect().await.unwrap();
+        tokio::time::sleep(Duration::from_millis(300)).await;
+
+        printer.upload_project_file("my part.gcode".into(), b"G28".to_vec()).await.unwrap();
+        let sent = mock.sent("PUT", "/api/v1/files/local/my%20part.gcode");
+        assert_eq!(sent[0].header("print-after-upload"), Some("?1"));
+        assert_eq!(sent[0].body, b"G28");
+    }
+}

src/printer/polled.rs

@@ -0,0 +1,47 @@
+//! Status for vendors that have no push feed: a background task asks the
+//! printer on an interval and keeps the last answer for `status()` to read.
+//! (Bambu pushes over MQTT and keeps its own cache.)
+use std::future::Future;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use tokio::task::JoinHandle;
+
+use super::PrinterStatus;
+
+pub struct StatusFeed {
+    latest: Arc<Mutex<Option<PrinterStatus>>>,
+    task: JoinHandle<()>,
+}
+
+impl StatusFeed {
+    /// `fetch` returning `None` means "no usable answer" (printer
+    /// unreachable, or not ready), and clears the cache: a stale status
+    /// shown as current is worse than none.
+    pub fn start<F, Fut>(interval: Duration, fetch: F) -> Self
+    where
+        F: Fn() -> Fut + Send + 'static,
+        Fut: Future<Output = Option<PrinterStatus>> + Send,
+    {
+        let latest = Arc::new(Mutex::new(None));
+        let shared = latest.clone();
+        let task = tokio::spawn(async move {
+            loop {
+                let status = fetch().await;
+                *shared.lock().unwrap() = status;
+                tokio::time::sleep(interval).await;
+            }
+        });
+        Self { latest, task }
+    }
+
+    pub fn latest(&self) -> Option<PrinterStatus> {
+        self.latest.lock().unwrap().clone()
+    }
+}
+
+impl Drop for StatusFeed {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}

src/printer/project_file.rs

@@ -71,6 +71,8 @@ pub enum ProjectFileError {
     /// material or none.
     #[error("no loaded tray matches {wanted} (filament index {index}); loaded: {loaded}")]
     NoMatchingTray { index: usize, wanted: String, loaded: String },
+    #[error("cannot use AMS slot {0}: {1}")]
+    BadSlot(i32, String),
 }
 
 /// Normalises a colour to six uppercase hex digits.
@@ -271,6 +273,36 @@ pub fn map_to_trays(
 }
 
 /// Match quality, lower is better; `None` means no match.
+/// An explicit choice of slot instead of the colour-based pick: the operator
+/// said which spool to use. Deliberately narrow - a single-filament project
+/// only (which filament goes where is ambiguous otherwise), onto a slot that
+/// is loaded and holds the same *material type*, because the type decides
+/// the temperatures. A different colour is allowed: that is the point.
+pub fn map_to_slot(
+    filaments: &[ProjectFilament],
+    trays: &[AvailableTray],
+    slot: i32,
+) -> Result<Vec<i32>, ProjectFileError> {
+    let [filament] = filaments else {
+        return Err(ProjectFileError::BadSlot(
+            slot,
+            format!("the project uses {} filaments; a chosen slot is only for single-filament projects", filaments.len()),
+        ));
+    };
+    let tray = trays.iter().find(|t| t.slot == slot).ok_or_else(|| {
+        ProjectFileError::BadSlot(slot, "no filament is loaded in that slot".into())
+    })?;
+    if !tray.filament_type.eq_ignore_ascii_case(&filament.filament_type) {
+        return Err(ProjectFileError::BadSlot(
+            slot,
+            format!("it holds {} but the project was sliced for {}", tray.filament_type, filament.filament_type),
+        ));
+    }
+    let mut mapping = vec![-1; filament.index + 1];
+    mapping[filament.index] = slot;
+    Ok(mapping)
+}
+
 fn tier(filament: &ProjectFilament, tray: &AvailableTray) -> Option<u8> {
     let same_product = filament.tray_info_idx.eq_ignore_ascii_case(&tray.tray_info_idx);
     let same_color = filament.color == tray.color;
@@ -288,6 +320,18 @@ mod tests {
     use super::*;
     use std::io::Write;
 
+    #[test]
+    fn a_chosen_slot_overrides_the_colour_pick_but_not_the_material() {
+        let f = [filament(0, "GFG99", "PETG", "161616")];
+        let t = [tray(2, "GFG99", "PETG", "161616"), tray(3, "GFG99", "PETG", "2850E0"), tray(1, "GFL99", "PLA", "D3C5A3")];
+        assert_eq!(map_to_trays(&f, &t).unwrap(), vec![2]);
+        assert_eq!(map_to_slot(&f, &t, 3).unwrap(), vec![3]);
+        assert!(map_to_slot(&f, &t, 1).is_err(), "PLA slot for a PETG project");
+        assert!(map_to_slot(&f, &t, 4).is_err(), "empty slot");
+        let two = [filament(0, "GFG99", "PETG", "161616"), filament(1, "GFG99", "PETG", "2850E0")];
+        assert!(map_to_slot(&two, &t, 3).is_err(), "multi-filament project");
+    }
+
     const REAL: &str = r##"<?xml version="1.0" encoding="UTF-8"?>
 <config>
   <header>

src/printer/prusa.rs

@@ -1,110 +1,4 @@
 use super::{GenericPrinter, PrinterBase, PrinterError};
-
-/****************** OctoPrint API family ******************/
-
-/// Which implementation of the OctoPrint REST API is on the other end.
-///
-/// PrusaLink deliberately implements OctoPrint's API, so one client serves
-/// both - but not identically. PrusaLink adds its own `/api/v1/*` (`info`,
-/// `status`, `job`) which is where the useful detail lives, while OctoPrint
-/// proper has the older `/api/printer` and `/api/job` and a plugin surface
-/// PrusaLink does not have.
-///
-/// This is recorded from config now, while the client is still a stub,
-/// because **no probe can settle it**: both answer `/api/version`, and every
-/// PrusaLink identity endpoint needs a credential, so from outside the two
-/// are indistinguishable (see `discovery::net::classify_octoprint_api`). It
-/// is the operator's answer to a question only they can answer, so it needs
-/// somewhere to live the moment they are asked - otherwise the real client
-/// would have to ask them a second time.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum OctoPrintFlavour {
-    /// Prusa's own firmware-hosted implementation, on an MK4/XL/MINI.
-    PrusaLink,
-    /// OctoPrint itself, usually on a Raspberry Pi in front of any printer.
-    Generic,
-}
-
-impl OctoPrintFlavour {
-    pub fn as_str(self) -> &'static str {
-        match self {
-            OctoPrintFlavour::PrusaLink => "PrusaLink",
-            OctoPrintFlavour::Generic => "OctoPrint",
-        }
-    }
-}
-
-pub struct OctoPrintPrinter {
-    base: PrinterBase,
-    pub api_key: String,
-    pub flavour: OctoPrintFlavour,
-}
-
-impl OctoPrintPrinter {
-    pub fn new(id: &str, name: &str, host: &str, api_key: &str, flavour: OctoPrintFlavour) -> Self {
-        Self {
-            base: PrinterBase { id: id.into(), name: name.into(), host: host.into() },
-            api_key: api_key.into(),
-            flavour,
-        }
-    }
-}
-
-impl GenericPrinter for OctoPrintPrinter {
-    /// OVERRIDE: real version is `PUT /api/v1/job/{id}/pause`, `/resume`, or
-    /// `DELETE /api/v1/job/{id}` to cancel - PrusaLink addresses the running
-    /// job by id rather than the printer, so a real implementation has to
-    /// fetch the current job first.
-    async fn control_print(&mut self, action: super::PrintControl) -> Result<(), PrinterError> {
-        println!("[{}] {} job control: {}", self.base.name, self.flavour.as_str(), action.as_str());
-        Ok(())
-    }
-
-
-    fn base(&self) -> &PrinterBase {
-        &self.base
-    }
-
-    /// OVERRIDE: a completely different body from Bambu's - PrusaLink is a
-    /// plain REST API (real version: `GET /api/v1/status` with the API
-    /// key, see `adapters::prusalink`), no persistent session to hold.
-    async fn connect(&mut self) -> Result<(), PrinterError> {
-        println!("[{}] connecting over HTTP ({})", self.base.name, self.flavour.as_str());
-        Ok(())
-    }
-
-    async fn publish(&mut self, _payload: String) -> Result<(), PrinterError> {
-        // TODO: add publishing via HTTP
-        Ok(())
-    }
-
-    /// OVERRIDE: gcode via PrusaLink's command-injection endpoint - same
-    /// underlying M106 as Bambu, reusing the shared percent->PWM
-    /// conversion, but sent over HTTP instead of MQTT.
-    async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
-        let pwm = super::percent_to_pwm(percent);
-        println!("[{}] POST /api/printer/command {{\"commands\":[\"M106 S{pwm}\"]}}", self.base.name);
-        Ok(())
-    }
-
-    /// OVERRIDE: real version is `PUT /api/v1/files/local/{file_name}` with
-    /// `data` as the request body - PrusaLink stores it on the printer's own
-    /// SD card, same "upload to storage the firmware owns" model as Bambu.
-    async fn upload_project_file(&mut self, file_name: String, data: Vec<u8>) -> Result<(), PrinterError> {
-        println!("[{}] PUT /api/v1/files/local/{file_name} ({} bytes)", self.base.name, data.len());
-        Ok(())
-    }
-
-    /// OVERRIDE: PrusaLink's command-injection endpoint, one line per call -
-    /// same endpoint `set_fan_speed` above already uses for M106, just with
-    /// an arbitrary line instead of a hardcoded one.
-    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
-        println!("[{}] POST /api/printer/command {{\"commands\":[\"{gcode_line}\"]}}", self.base.name);
-        Ok(())
-    }
-}
-
-
 /****************** PrusaSerial ******************/
 
 pub struct PrusaSerialPrinter {

src/provision.rs

@@ -201,8 +201,8 @@ mod behaviour {
         let _ = std::fs::remove_file(path);
     }
 
-    /// The success path, using the OctoPrint client because its `connect` is
+    /// The success path, using the OctoPrint client against a mock server,
-    /// the one that succeeds without a real machine (the driver is a stub).
+    /// because no real machine is needed to connect to one.
     /// What this proves is the plumbing, not a vendor: a connected printer is
     /// saved, joins the running fleet, and is still there after a "restart"
     /// that reopens the store and rebuilds the fleet from it.
@@ -210,7 +210,10 @@ mod behaviour {
     async fn an_added_printer_joins_the_fleet_and_survives_a_restart() {
         let (store, path) = store();
         let fleet = empty_fleet();
+        let mock = crate::printer::mock_http::MockServer::start().await;
+        mock.answer("GET /api/job", 200, r#"{"state":"Operational","progress":{}}"#);
         let mut req = request("octo1");
+        req.host = Some(mock.host());
         req.vendor = "octoprint".into();
         req.sn = None;
         req.access_code = None;

src/store.rs

@@ -42,6 +42,22 @@ impl Store {
             "CREATE TABLE IF NOT EXISTS printers (
                 id   TEXT PRIMARY KEY,
                 json TEXT NOT NULL
+            );
+            -- What happened to jobs the cloud handed this gateway, until the
+            -- cloud has acknowledged it (see `jobs`). Survives a restart.
+            CREATE TABLE IF NOT EXISTS outbox (
+                id     INTEGER PRIMARY KEY AUTOINCREMENT,
+                job_id TEXT NOT NULL,
+                event  TEXT NOT NULL,
+                detail TEXT,
+                at_ms  INTEGER NOT NULL
+            );
+            -- The job each printer is working on, as far as this gateway knows.
+            CREATE TABLE IF NOT EXISTS active_jobs (
+                printer_id TEXT PRIMARY KEY,
+                job_id     TEXT NOT NULL,
+                since_ms   INTEGER NOT NULL,
+                started    INTEGER NOT NULL DEFAULT 0
             );",
         )?;
         Ok(Self { conn: Mutex::new(conn) })
@@ -78,6 +94,97 @@ impl Store {
     }
 }
 
+/// A job outcome waiting for the cloud's acknowledgement.
+#[derive(Debug, Clone, PartialEq)]
+pub struct OutboxEvent {
+    pub id: i64,
+    pub job_id: String,
+    pub event: String,
+    pub detail: Option<String>,
+    pub at_ms: i64,
+}
+
+/// A job a printer was handed and has not yet finished with.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ActiveJob {
+    pub printer_id: String,
+    pub job_id: String,
+    pub since_ms: i64,
+    pub started: bool,
+}
+
+impl Store {
+    /// Queues an outcome to tell the cloud. Returns its id.
+    pub fn outbox_push(&self, job_id: &str, event: &str, detail: Option<&str>, at_ms: i64) -> anyhow::Result<i64> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute(
+            "INSERT INTO outbox (job_id, event, detail, at_ms) VALUES (?1, ?2, ?3, ?4)",
+            params![job_id, event, detail, at_ms],
+        )?;
+        Ok(conn.last_insert_rowid())
+    }
+
+    /// Everything the cloud has not acknowledged, oldest first.
+    pub fn outbox_pending(&self) -> anyhow::Result<Vec<OutboxEvent>> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        let mut statement =
+            conn.prepare("SELECT id, job_id, event, detail, at_ms FROM outbox ORDER BY id LIMIT 500")?;
+        let rows = statement.query_map([], |row| {
+            Ok(OutboxEvent {
+                id: row.get(0)?,
+                job_id: row.get(1)?,
+                event: row.get(2)?,
+                detail: row.get(3)?,
+                at_ms: row.get(4)?,
+            })
+        })?;
+        Ok(rows.collect::<Result<_, _>>()?)
+    }
+
+    pub fn outbox_ack(&self, id: i64) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute("DELETE FROM outbox WHERE id = ?1", params![id])?;
+        Ok(())
+    }
+
+    /// Records that `printer_id` was handed `job_id`, replacing any earlier job.
+    pub fn active_job_set(&self, printer_id: &str, job_id: &str, now_ms: i64) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute(
+            "INSERT INTO active_jobs (printer_id, job_id, since_ms, started) VALUES (?1, ?2, ?3, 0)
+             ON CONFLICT(printer_id) DO UPDATE SET job_id = excluded.job_id, since_ms = excluded.since_ms, started = 0",
+            params![printer_id, job_id, now_ms],
+        )?;
+        Ok(())
+    }
+
+    pub fn active_job_started(&self, printer_id: &str) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute("UPDATE active_jobs SET started = 1 WHERE printer_id = ?1", params![printer_id])?;
+        Ok(())
+    }
+
+    pub fn active_job_clear(&self, printer_id: &str) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute("DELETE FROM active_jobs WHERE printer_id = ?1", params![printer_id])?;
+        Ok(())
+    }
+
+    pub fn active_jobs(&self) -> anyhow::Result<Vec<ActiveJob>> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        let mut statement = conn.prepare("SELECT printer_id, job_id, since_ms, started FROM active_jobs")?;
+        let rows = statement.query_map([], |row| {
+            Ok(ActiveJob {
+                printer_id: row.get(0)?,
+                job_id: row.get(1)?,
+                since_ms: row.get(2)?,
+                started: row.get::<_, i64>(3)? != 0,
+            })
+        })?;
+        Ok(rows.collect::<Result<_, _>>()?)
+    }
+}
+
 /// Creates the file empty and owner-only if it does not exist yet.
 #[cfg(unix)]
 fn create_private(path: &Path) -> anyhow::Result<()> {
@@ -196,4 +303,36 @@ mod tests {
         std::fs::write(&path, b"this is not a sqlite database at all, just text").unwrap();
         assert!(Store::open(&path).is_err());
     }
+
+    #[test]
+    fn an_unacknowledged_event_survives_a_restart_and_an_acknowledged_one_does_not() {
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        let first = store.outbox_push("job-1", "finished", None, 10).unwrap();
+        let second = store.outbox_push("job-2", "failed", Some("download failed"), 20).unwrap();
+        store.outbox_ack(first).unwrap();
+        drop(store);
+
+        let reopened = Store::open(&path).unwrap();
+        let pending = reopened.outbox_pending().unwrap();
+        assert_eq!(pending.len(), 1);
+        assert_eq!(pending[0].id, second);
+        assert_eq!(pending[0].detail.as_deref(), Some("download failed"));
+    }
+
+    #[test]
+    fn a_printer_has_one_active_job_and_a_new_one_replaces_it() {
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        store.active_job_set("p1", "job-1", 100).unwrap();
+        store.active_job_started("p1").unwrap();
+        assert!(store.active_jobs().unwrap()[0].started);
+
+        store.active_job_set("p1", "job-2", 200).unwrap();
+        let jobs = store.active_jobs().unwrap();
+        assert_eq!(jobs, vec![ActiveJob { printer_id: "p1".into(), job_id: "job-2".into(), since_ms: 200, started: false }]);
+
+        store.active_job_clear("p1").unwrap();
+        assert!(store.active_jobs().unwrap().is_empty());
+    }
 }

src/uplink/download.rs

@@ -26,6 +26,26 @@ pub const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(180);
 /// up the whole budget above.
 pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
 
+/// Dev-only escape hatch for a local object store (MinIO) that speaks plain
+/// HTTP: needs `CONTINUUM_ALLOW_INSECURE_LOCAL_DOWNLOADS=true` AND a loopback
+/// or private-network (RFC 1918) address literal, so a presigned URL can never
+/// cross a public network in the clear.
+fn insecure_local_allowed(url: &reqwest::Url) -> bool {
+    use std::net::IpAddr;
+    if url.scheme() != "http"
+        || !std::env::var("CONTINUUM_ALLOW_INSECURE_LOCAL_DOWNLOADS").is_ok_and(|v| v == "true")
+    {
+        return false;
+    }
+    let Some(ip) = url.host_str().and_then(|h| h.trim_matches(['[', ']']).parse::<IpAddr>().ok()) else {
+        return false;
+    };
+    match ip {
+        IpAddr::V4(v4) => v4.is_loopback() || v4.is_private(),
+        IpAddr::V6(v6) => v6.is_loopback(),
+    }
+}
+
 /// Downloads `url` fully into memory, or says why not.
 ///
 /// The error is a `String` naming the failure; the caller prefixes the stage.
@@ -39,7 +59,7 @@ pub async fn fetch_job_file_capped(url: &str, max_bytes: u64) -> Result<Vec<u8>,
     let parsed = reqwest::Url::parse(url).map_err(|err| format!("not a valid URL: {err}"))?;
     // A presigned URL carries its authority in the query string; sending it
     // over plain HTTP would expose it to anyone on the path.
-    if parsed.scheme() != "https" {
+    if parsed.scheme() != "https" && !insecure_local_allowed(&parsed) {
         return Err(format!("refusing non-HTTPS URL (scheme '{}')", parsed.scheme()));
     }
 

src/uplink/mod.rs

@@ -33,6 +33,7 @@ pub mod protocol;
 
 pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage};
 
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -82,6 +83,8 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
 /// a lot of traffic to describe a machine whose state changes on the scale of
 /// a layer, so this samples the cached report instead of chasing every push.
 const TELEMETRY_INTERVAL: Duration = Duration::from_secs(5);
+/// How often the job-event outbox is looked at for anything new to send.
+const OUTBOX_INTERVAL: Duration = Duration::from_secs(3);
 
 /// The printer fleet, shared between `main` (which owns it and keeps every
 /// connection alive) and this module (which reads status and dispatches
@@ -225,9 +228,34 @@ async fn serve(
     let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
     let mut telemetry = tokio::time::interval(TELEMETRY_INTERVAL);
     let mut seq: u64 = 0;
+    // Outbox events already written to THIS socket. Per connection on
+    // purpose: after a reconnect everything the cloud has not acknowledged
+    // goes out again, which is the point of keeping them.
+    let mut outbox_tick = tokio::time::interval(OUTBOX_INTERVAL);
+    let mut sent: HashSet<i64> = HashSet::new();
 
     loop {
         tokio::select! {
+            _ = outbox_tick.tick() => {
+                let Some(store) = store else { continue };
+                let pending = match store.outbox_pending() {
+                    Ok(pending) => pending,
+                    Err(err) => { warn!(%err, "could not read the job-event outbox"); continue }
+                };
+                let unsent: Vec<_> = pending.into_iter().filter(|e| !sent.contains(&e.id)).collect();
+                for event in unsent {
+                    let msg = ClientMessage::JobEvent {
+                        event_id: event.id,
+                        job_id: event.job_id,
+                        event: event.event,
+                        detail: event.detail,
+                        at_ms: event.at_ms,
+                    };
+                    write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
+                    sent.insert(event.id);
+                }
+            }
+
             _ = heartbeat.tick() => {
                 seq += 1;
                 let msg = ClientMessage::Heartbeat { seq };
@@ -275,6 +303,7 @@ async fn serve(
                                         command,
                                         command_id.clone(),
                                         outbound.clone(),
+                                        store.cloned(),
                                     );
                                     ClientMessage::CommandResult {
                                         command_id, ok: true, error: None,
@@ -328,6 +357,14 @@ async fn serve(
                             Ok(ServerMessage::ProvisionPrinter { request_id, printer }) => {
                                 spawn_provision(fleet.clone(), store.cloned(), request_id, Some(printer), outbound.clone());
                             }
+                            Ok(ServerMessage::JobEventAck { event_id }) => {
+                                sent.remove(&event_id);
+                                if let Some(store) = store
+                                    && let Err(err) = store.outbox_ack(event_id)
+                                {
+                                    warn!(%err, event_id, "could not clear an acknowledged job event");
+                                }
+                            }
                             Ok(ServerMessage::RemovePrinter { request_id, printer_id }) => {
                                 spawn_remove(fleet.clone(), store.cloned(), request_id, printer_id, outbound.clone());
                             }
@@ -374,29 +411,28 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
 
     for printer in fleet.iter() {
         let base = printer.base();
-        let Some(report) = printer.print_status().await else {
+        let Some(status) = printer.status().await else {
             continue;
         };
+        // Only Bambu has a camera report to derive these from.
+        let report = printer.print_status().await;
 
         out.push(PrinterTelemetry {
             printer_id: base.id.clone(),
             name: base.name.clone(),
-            state: report.gcode_state.clone(),
+            state: status.state,
-            percent: report.mc_percent,
+            percent: status.percent,
-            stage: report.stg_cur,
+            stage: status.stage,
-            layer: report.layer_num,
+            layer: status.layer,
-            total_layers: report.total_layer_num,
+            total_layers: status.total_layers,
-            remaining_minutes: report.mc_remaining_time,
+            remaining_minutes: status.remaining_minutes,
-            // An idle printer reports `""` rather than omitting the field;
+            job_name: status.job_name,
-            // empty-string-as-absent is worth normalising here so the cloud
+            nozzle_temp: status.nozzle_temp,
-            // doesn't have to know that.
+            nozzle_target: status.nozzle_target,
-            job_name: report.subtask_name.clone().filter(|s| !s.is_empty()),
+            bed_temp: status.bed_temp,
-            nozzle_temp: report.nozzle_temper,
+            bed_target: status.bed_target,
-            nozzle_target: report.nozzle_target_temper,
+            print_error: status.print_error,
-            bed_temp: report.bed_temper,
+            active_alerts: status.active_alerts,
-            bed_target: report.bed_target_temper,
-            print_error: report.print_error,
-            active_alerts: report.hms.as_ref().map(|h| h.len()),
             // Derived from the same reported URL `camera::sources` builds
             // the go2rtc config from, so a printer is advertised as
             // streamable here only if it is also actually being served
@@ -404,13 +440,13 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
             // the rule means the two cannot drift apart.
             camera_stream: crate::camera::source_for(
                 printer,
-                report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()),
+                report.as_ref().and_then(|r| r.ipcam.as_ref()).and_then(|c| c.rtsp_url.as_deref()),
             )
             .map(|source| source.stream_name),
             camera_snapshot: crate::camera::wants_snapshot(
                 printer,
-                report.ipcam.is_some(),
+                report.as_ref().is_some_and(|r| r.ipcam.is_some()),
-                report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()),
+                report.as_ref().and_then(|r| r.ipcam.as_ref()).and_then(|c| c.rtsp_url.as_deref()),
             )
             .then_some(true),
         });
@@ -461,10 +497,31 @@ fn spawn_long_command(
     command: PrinterCommand,
     command_id: String,
     outbound: tokio::sync::mpsc::Sender<ClientMessage>,
+    store: Option<Arc<Store>>,
 ) {
+    // The job is followed from here on, so its outcome is reported even if
+    // the cloud is unreachable when it happens (see `jobs`).
+    let job_id = match (&command, &store) {
+        (PrinterCommand::StartPrint { job_id, .. }, Some(store)) => {
+            crate::jobs::accepted(store, &printer_id, job_id);
+            Some(job_id.clone())
+        }
+        _ => None,
+    };
     tokio::spawn(async move {
         match dispatch(&fleet, &printer_id, command).await {
             Ok(()) => info!(%printer_id, "long-running command finished"),
+            Err(err) if job_id.is_some() => {
+                // Still logged locally; the cloud hears of it from the
+                // outbox, which survives it being down right now.
+                tracing::error!(
+                    %printer_id, %err,
+                    "long-running command failed after the cloud was already told it was accepted"
+                );
+                if let (Some(store), Some(job_id)) = (&store, &job_id) {
+                    crate::jobs::start_failed(store, &printer_id, job_id, &err);
+                }
+            }
             Err(err) => {
                 // Still logged locally, and still at error level: the queue
                 // only reaches a cloud that is up, and this is the copy that
@@ -652,7 +709,7 @@ async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand
         PrinterCommand::ControlPrint { control } => {
             printer.control_print(control).await.map_err(|err| err.to_string())
         }
-        PrinterCommand::StartPrint { job_id, file_name, .. } => {
+        PrinterCommand::StartPrint { job_id, file_name, ams_slot, .. } => {
             let bytes = downloaded.ok_or_else(|| "internal error: start_print reached dispatch with no downloaded file".to_string())?;
             let byte_count = bytes.len();
             info!(%printer_id, %job_id, %file_name, byte_count, "job file downloaded, sending to printer");
@@ -661,10 +718,19 @@ async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand
             // all as one error - so the stage here is "printer", not finer.
             // The printer's own message says which part (mapping, upload of
             // the file, or start) it was.
-            printer
+            if let PrinterHandle::Bambu(bambu) = &mut *printer {
+                bambu.set_ams_slot_override(ams_slot);
+            } else if ams_slot.is_some() {
+                return Err("this printer cannot be told which AMS slot to use".into());
+            }
+            let result = printer
                 .upload_project_file(file_name, bytes)
                 .await
-                .map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}"))
+                .map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}"));
+            if let PrinterHandle::Bambu(bambu) = &mut *printer {
+                bambu.set_ams_slot_override(None);
+            }
+            result
         }
     }
 }

src/uplink/protocol.rs

@@ -32,6 +32,17 @@ pub enum ClientMessage {
         stream_base_url: Option<String>,
     },
     Heartbeat { seq: u64 },
+    /// What happened to a job the cloud handed this gateway. Kept in the
+    /// store and re-sent on every connection until a `JobEventAck`.
+    JobEvent {
+        event_id: i64,
+        job_id: String,
+        /// `started`, `finished` or `failed`.
+        event: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        detail: Option<String>,
+        at_ms: i64,
+    },
     /// One snapshot per printer, on a timer. Sent as a batch rather than a
     /// message per printer: a farm has one uplink and many printers, and the
     /// cloud wants a coherent picture of the farm at an instant more than it
@@ -213,6 +224,8 @@ pub struct PrinterTelemetry {
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum ServerMessage {
     HelloAck { session_id: String },
+    /// The cloud has dealt with a `JobEvent`; the gateway can forget it.
+    JobEventAck { event_id: i64 },
     /// The handshake was rejected - a bad, expired or wrong-farm token. The
     /// backend closes the connection right after sending this.
     ///
@@ -329,7 +342,14 @@ pub enum PrinterCommand {
     /// hides it. `file_name` is what the file is called on the printer;
     /// `.3mf` is treated as a sliced project, anything else as bare gcode.
     /// `job_id` is only for lining gateway logs up against a cloud job.
-    StartPrint { job_id: String, file_name: String, url: String },
+    StartPrint {
+        job_id: String,
+        file_name: String,
+        url: String,
+        /// Use this AMS slot (0-based, all units) instead of the project's own pick.
+        #[serde(default)]
+        ams_slot: Option<u8>,
+    },
 }
 
 impl PrinterCommand {

tests/push_status_parsing.rs

@@ -144,3 +144,13 @@ fn merge_keeps_stage_and_spool_when_a_partial_update_omits_them() {
     assert_eq!(base.stg, Some(vec![54, 14, 13]));
     assert!(base.vt_tray.is_some());
 }
+
+/// A P1P mid-print pushes `ams` objects without the inner unit list. That
+/// used to fail the whole report ("missing field `ams`"), silencing telemetry
+/// for the rest of the print - found on the first real dispatch.
+#[test]
+fn a_partial_ams_object_parses_and_keeps_known_units() {
+    let partial = serde_json::json!({"command":"push_status","sequence_id":"1","msg":1,"gcode_state":"RUNNING","ams":{"tray_now":"2"}});
+    let report: PrintReport = serde_json::from_value(partial).expect("a partial ams object should parse");
+    assert!(report.ams.as_ref().is_some_and(|a| a.ams.is_empty()));
+}

tests/uplink_outbox.rs

@@ -0,0 +1,183 @@
+//! Job outcomes survive the cloud being away: an event in the gateway's store
+//! is sent on connect, sent again after a reconnect if nobody acknowledged it,
+//! and gone once acknowledged. A start that fails after being accepted is
+//! queued the same way instead of being sent once to whoever is listening.
+//!
+//! Real `uplink::run` against a real websocket server; no printer is touched
+//! (the fleet is empty, so the start fails on its lookup before any download).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use serde_json::{json, Value};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use tokio::sync::Mutex;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
+
+use continuum_proxy::config::Config;
+use continuum_proxy::store::Store;
+use continuum_proxy::uplink;
+
+/// Long enough for a loopback round trip, short enough that a regression
+/// fails the test instead of hanging the suite.
+const WAIT: Duration = Duration::from_secs(10);
+
+fn config(port: u16) -> Config {
+    Config {
+        gateway_id: "test-gw".into(),
+        farm_id: "test-farm".into(),
+        gateway_token: "test-token".into(),
+        uplink_url: format!("ws://127.0.0.1:{port}/ws/edge/v1"),
+        // Nothing here starts go2rtc or reads the fleet file - `uplink::run`
+        // is handed its fleet rather than loading one - but Config has no
+        // Default, so these have to say something.
+        go2rtc_bin: "/nonexistent-go2rtc".into(),
+        go2rtc_config: "/nonexistent/go2rtc.yaml".into(),
+        go2rtc_api_listen: "127.0.0.1:1984".into(),
+        stream_base_url: None,
+        printers_path: "/nonexistent/printers.toml".into(),
+        store_path: "/nonexistent/continuum.db".into(),
+        // Off, which is the default and also the only safe thing in a test:
+        // an enabled one would sweep whatever network the test host is on.
+        discovery: Default::default(),
+    }
+}
+
+/// The next frame as JSON. Heartbeats and telemetry share this socket, so
+/// callers filter; `interval` fires on its first tick, which means a
+/// heartbeat arrives immediately after the hello.
+async fn next_json<S>(ws: &mut WebSocketStream<S>) -> Value
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    loop {
+        let frame = tokio::time::timeout(WAIT, ws.next())
+            .await
+            .expect("timed out waiting for a frame from the gateway")
+            .expect("the gateway closed the connection")
+            .expect("websocket error");
+        if let Message::Text(text) = frame {
+            return serde_json::from_str(&text).expect("the gateway sent invalid JSON");
+        }
+    }
+}
+
+/// Plays the cloud's `/auth/edge-token`: reads one HTTP request, checks that
+/// the gateway sent its farm and long-lived token, and answers with a JWT.
+/// Returns the request text so the caller can assert on it.
+async fn answer_token_exchange(listener: &TcpListener) -> String {
+    let (mut stream, _) = listener.accept().await.unwrap();
+    let mut request = Vec::new();
+    let mut buf = [0u8; 4096];
+    loop {
+        let n = tokio::time::timeout(WAIT, stream.read(&mut buf)).await.unwrap().unwrap();
+        assert!(n > 0, "the gateway hung up before finishing its token request");
+        request.extend_from_slice(&buf[..n]);
+        let text = String::from_utf8_lossy(&request).into_owned();
+        if let Some((head, body)) = text.split_once("\r\n\r\n") {
+            let length = head
+                .lines()
+                .find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
+                .unwrap_or(0);
+            if body.len() >= length {
+                break;
+            }
+        }
+    }
+    let body = r#"{"token":"minted-edge-jwt","expiresInSeconds":3600}"#;
+    stream
+        .write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes())
+        .await
+        .unwrap();
+    String::from_utf8_lossy(&request).into_owned()
+}
+
+
+/// Answers the token exchange and accepts the websocket, returning it after
+/// the gateway's hello.
+async fn accept(listener: &TcpListener) -> WebSocketStream<tokio::net::TcpStream> {
+    answer_token_exchange(listener).await;
+    let (stream, _) = listener.accept().await.unwrap();
+    let mut ws = tokio_tungstenite::accept_hdr_async(stream, |_: &Request, response: Response| Ok(response)).await.unwrap();
+    let hello = next_json(&mut ws).await;
+    assert_eq!(hello["type"], "hello");
+    ws
+}
+
+/// Reads frames until `want` job events have arrived, returning them by job id.
+async fn job_events(ws: &mut WebSocketStream<tokio::net::TcpStream>, want: usize) -> Vec<Value> {
+    let mut events = Vec::new();
+    while events.len() < want {
+        let msg = next_json(ws).await;
+        if msg["type"] == "job_event" {
+            events.push(msg);
+        }
+    }
+    events.sort_by_key(|e| e["event_id"].as_i64());
+    events
+}
+
+#[tokio::test]
+async fn unacknowledged_job_events_are_resent_after_a_reconnect_and_stop_once_acknowledged() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+
+    let dir = std::env::temp_dir().join(format!("continuum-outbox-test-{}", std::process::id()));
+    let store = Arc::new(Store::open(&dir.join("s.db")).unwrap());
+    // An outcome reached while the cloud was away.
+    store.outbox_push("11111111-1111-4111-8111-111111111111", "finished", None, 1_000).unwrap();
+
+    let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new()));
+    let (outbound, inbox) = uplink::channel();
+    tokio::spawn(uplink::run(config(port), fleet, Some(store.clone()), outbound, inbox));
+
+    let mut ws = accept(&listener).await;
+    let first = job_events(&mut ws, 1).await;
+    assert_eq!(first[0]["job_id"], "11111111-1111-4111-8111-111111111111");
+    assert_eq!(first[0]["event"], "finished");
+    assert_eq!(first[0]["at_ms"], 1_000);
+
+    // A start that fails after being accepted goes through the same outbox.
+    ws.send(Message::Text(
+        json!({
+            "type": "command", "command_id": "cmd-1", "printer_id": "not-in-this-fleet",
+            "command": { "action": "start_print", "job_id": "22222222-2222-4222-8222-222222222222",
+                         "file_name": "cube.gcode", "url": "http://127.0.0.1:1/never-fetched" }
+        })
+        .to_string(),
+    ))
+    .await
+    .unwrap();
+    // Only the new one: the first was already sent on this connection.
+    let mut both = first.clone();
+    both.extend(job_events(&mut ws, 1).await);
+    assert_eq!(both[1]["job_id"], "22222222-2222-4222-8222-222222222222");
+    assert_eq!(both[1]["event"], "failed");
+    assert!(both[1]["detail"].as_str().unwrap().contains("not-in-this-fleet"), "{}", both[1]);
+
+    // The cloud never answers; the connection drops.
+    drop(ws);
+
+    let mut ws = accept(&listener).await;
+    let again = job_events(&mut ws, 2).await;
+    assert_eq!(
+        again.iter().map(|e| e["event_id"].clone()).collect::<Vec<_>>(),
+        both.iter().map(|e| e["event_id"].clone()).collect::<Vec<_>>(),
+        "after a reconnect everything unacknowledged is sent again"
+    );
+
+    ws.send(Message::Text(json!({ "type": "job_event_ack", "event_id": again[0]["event_id"] }).to_string()))
+        .await
+        .unwrap();
+    let deadline = tokio::time::Instant::now() + WAIT;
+    while store.outbox_pending().unwrap().len() != 1 {
+        assert!(tokio::time::Instant::now() < deadline, "the acknowledged event was not cleared");
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert_eq!(store.outbox_pending().unwrap()[0].job_id, "22222222-2222-4222-8222-222222222222");
+    let _ = std::fs::remove_dir_all(dir);
+}

tests/uplink_protocol.rs

@@ -106,7 +106,7 @@ fn a_vendor_specific_control_verb_is_rejected() {
 fn a_start_print_command_arrives_with_its_fields_intact() {
     let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"5b0e7c1a-0000-4000-8000-000000000000","file_name":"plate_1.3mf","url":"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"}}"#;
     match serde_json::from_str::<ServerMessage>(raw).expect("should parse") {
-        ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url }, .. } => {
+        ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url, .. }, .. } => {
             assert_eq!(job_id, "5b0e7c1a-0000-4000-8000-000000000000");
             assert_eq!(file_name, "plate_1.3mf");
             assert_eq!(