Wire fleet::load into main.rs; fix request() reply unwrapping

main.rs never actually loaded printers.toml or connected anything -
fleet/printer were only ever exercised by standalone examples. Now
main() loads the fleet (path configurable via CONTINUUM_PRINTERS_PATH,
default "printers.toml") and connects every printer concurrently at
startup, logging success/failure per printer rather than failing
startup over one bad printer.

That surfaced a real bug the instant it ran against a live printer:
complete_pending() was sending the whole {"info": {...}} wrapper through
the pending oneshot, not the unwrapped body - BambuPrinter::get_version
deserializing that into GetVersionResponse failed with "missing field
module" even though module was right there, just one level deeper than
expected. Fixed by unwrapping the single category key before completing
the pending request, symmetric with how a request gets wrapped on the
way out.

Verified live against a real H2C (MQTT v5/BambuMQTT v2): connect,
get_version, and the full pending-request correlation all worked
end-to-end, returning real firmware versions for every module (main
board 01.01.05.00, both AMS units, AMS HT, etc). The other configured
printer (P1P, forced v1) failed with a real, expected TLS error - it
doesn't chain to the bundled CA, matching the already-documented P1P
quirk; needs the ca_cert_path/fetch_bambu_cert workflow, not a code fix.

src/config.rs

@@ -10,6 +10,9 @@ pub struct Config {
     pub uplink_url: String,
     pub go2rtc_bin: String,
     pub go2rtc_config: String,
+    /// Path to the fleet config file - see `fleet.rs` and
+    /// `printers.example.toml` for the shape.
+    pub printers_path: String,
 
     /// Per-printer CA certificate overrides, keyed by printer ID. Most
     /// Bambu printers chain to the bundled shared CA (`certs/bambu_ca2.pem`)
@@ -34,6 +37,7 @@ impl Config {
             uplink_url: env_or("CONTINUUM_UPLINK_URL", "wss://api.continuum.local/ws/edge/v1"),
             go2rtc_bin: env_or("CONTINUUM_GO2RTC_BIN", "go2rtc"),
             go2rtc_config: env_or("CONTINUUM_GO2RTC_CONFIG", "./go2rtc.yaml"),
+            printers_path: env_or("CONTINUUM_PRINTERS_PATH", "printers.toml"),
 
             bambu_ca_cert_overrides: parse_cert_overrides(&env_or("CONTINUUM_BAMBU_CA_CERT_OVERRIDES", "")),
             bambu_require_valid_cert: env_or("CONTINUUM_BAMBU_REQUIRE_VALID_CERT", "true") == "true",

src/main.rs

@@ -1,4 +1,5 @@
 mod config;
+mod fleet;
 mod go2rtc;
 mod printer;
 mod uplink;
@@ -6,6 +7,7 @@ mod uplink;
 use tracing_subscriber::EnvFilter;
 
 use crate::config::Config;
+use crate::printer::{GenericPrinter, PrinterHandle};
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -18,12 +20,24 @@ async fn main() -> anyhow::Result<()> {
 
     tracing::info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy");
 
-    // Two things run at once for now: the cloud uplink, and the go2rtc
+    // Loaded and connected here, once, at startup - nothing yet re-reads
-    // camera-restreaming process. More will join them later - printer
+    // printers.toml on a running daemon (adding/removing a printer means
-    // discovery, the plate changer - as you build past src/printer/'s
+    // a restart), and nothing yet lets the uplink dispatch a command to
-    // stubs. `tokio::select!` runs both and stops as soon as either one
+    // one of these once connected - see PrinterHandle's own doc comment
-    // returns; both loop forever below, so in practice this runs until the
+    // for what's built and what's still a stub per vendor. Keeping
-    // process is killed.
+    // `fleet` alive as a local variable for the rest of `main` (which
+    // never returns, below) is what keeps every connection open; it's
+    // not otherwise touched yet.
+    let mut fleet = fleet::load(std::path::Path::new(&config.printers_path))?;
+    connect_fleet(&mut fleet).await;
+
+    // Three things run at once for now: the cloud uplink, the go2rtc
+    // camera-restreaming process, and (implicitly, via the connections
+    // opened above) every configured printer's live session. More will
+    // join them later - printer discovery, the plate changer - as you
+    // build past src/printer/'s stubs. `tokio::select!` runs both of the
+    // tasks below and stops as soon as either one returns; both loop
+    // forever, so in practice this runs until the process is killed.
     tokio::select! {
         _ = uplink::run(config.clone()) => {}
         _ = go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()) => {}
@@ -31,3 +45,25 @@ async fn main() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+/// Connects every configured printer concurrently (not one at a time -
+/// Bambu's auto-detect fallback alone can take several seconds per
+/// printer if it has to time out on MQTT v5 before trying v3.1.1;
+/// serially, a fleet of twenty would take twenty times as long to start).
+/// Logs success/failure per printer instead of failing startup over one -
+/// a farm shouldn't refuse to start because one printer is unplugged.
+async fn connect_fleet(fleet: &mut [PrinterHandle]) {
+    let results =
+        futures_util::future::join_all(fleet.iter_mut().map(|printer| async {
+            let name = printer.display_name();
+            (name, printer.connect().await)
+        }))
+        .await;
+
+    for (name, result) in results {
+        match result {
+            Ok(()) => tracing::info!(printer = %name, "connected"),
+            Err(err) => tracing::warn!(printer = %name, %err, "failed to connect"),
+        }
+    }
+}

src/printer/bambu_mqtt.rs

@@ -262,24 +262,28 @@ async fn poll_loop_v2(mut event_loop: rumqttc::v5::EventLoop, pending: PendingTa
 }
 
 /// Parses an incoming report and, if its `sequence_id` matches something
-/// `request` is waiting on, hands it the parsed JSON. A report with no
+/// `request` is waiting on, hands it the *unwrapped* body. A report with
-/// `sequence_id`, or one nothing's waiting on (an unsolicited status
+/// no `sequence_id`, or one nothing's waiting on (an unsolicited status
 /// push, not a reply to anything), is silently ignored - most Bambu
 /// reports aren't replies to a specific request at all.
+///
+/// A report is `{category: {"sequence_id": "...", ...}}` - one key, same
+/// shape a request is published as. `request` publishes `body` and wants
+/// `body`-shaped data back, not the wrapper around it, so this looks
+/// inside whichever single category key is present (rather than
+/// requiring the caller to already know it) and sends only that inner
+/// value through - confirmed against a real printer's `get_version`
+/// reply, which failed to parse as `GetVersionResponse` before this
+/// unwrapped it (real device, real bug: `serde_json::from_value` on the
+/// still-wrapped `{"info": {...}}` reported "missing field `module`",
+/// since `module` lives one level deeper than that).
 async fn complete_pending(pending: &PendingTable, payload: &[u8]) {
     let Ok(value) = serde_json::from_slice::<Value>(payload) else { return };
-    let Some(sequence_id) = find_sequence_id(&value) else { return };
+    let Some(object) = value.as_object() else { return };
+    let Some((_, body)) = object.iter().next() else { return };
+    let Some(sequence_id) = body.get("sequence_id").and_then(Value::as_str) else { return };
 
-    if let Some(sender) = pending.lock().await.remove(&sequence_id) {
+    if let Some(sender) = pending.lock().await.remove(sequence_id) {
-        let _ = sender.send(value);
+        let _ = sender.send(body.clone());
     }
 }
-
-/// A report is `{category: {"sequence_id": "...", ...}}` - one key, same
-/// shape as a request. Look inside whichever single category key is
-/// present rather than requiring the caller to already know it.
-fn find_sequence_id(report: &Value) -> Option<String> {
-    let object = report.as_object()?;
-    let (_, category_body) = object.iter().next()?;
-    category_body.get("sequence_id")?.as_str().map(str::to_string)
-}