Update docs and prune examples superseded by real functionality

README/certs/README/.env.example/printers.example.toml were all still
describing this as a learning-stage stub with no real printer protocol
clients - rewritten to reflect the real Bambu MQTT client, cert
auto-pinning, and marlin serial connection that now exist.

- Remove examples/fetch_bambu_cert.rs: its manual fetch-and-save workflow
  is now what BambuPrinter::connect()'s ensure_trusted does automatically
  on every connect, so it has no remaining use case.
- Fix examples/printer_polymorphism.rs, which no longer compiled -
  dispatch_gcode was replaced by real methods earlier this session;
  swapped in print_status() as the vendor-only-extension example.
- Remove Config::bambu_ca_cert_overrides/bambu_require_valid_cert and the
  .env vars documenting them - confirmed dead (nothing read them),
  fully superseded by printers.toml's per-printer ca_cert_path/
  require_valid_cert plus automatic trust-on-first-connect pinning.
- Fix stale doc comments in printer/mod.rs and printer/bambu.rs that
  described certificate trust and MQTT as not built yet, or referencing
  the now-removed example.

No behavior change in the daemon itself - verified with cargo build
--examples and a live run of printer_polymorphism and auto_connect_bambu
against the real P1P/H2C printers.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.env.example

@@ -10,11 +10,10 @@ CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1
 CONTINUUM_GO2RTC_BIN=/usr/local/bin/go2rtc
 CONTINUUM_GO2RTC_CONFIG=./go2rtc.yaml
 
-# Bambu MQTTS certificate trust. Most printers verify fine against the
-# bundled shared CA (certs/bambu_ca2.pem) and need nothing here. For a
-# printer that doesn't chain to it (P1P, at least), add "printer_id=path"
-# entries, comma-separated: CONTINUUM_BAMBU_CA_CERT_OVERRIDES=p1=./certs/p1p.pem
-CONTINUUM_BAMBU_CA_CERT_OVERRIDES=
-CONTINUUM_BAMBU_REQUIRE_VALID_CERT=true
+# Fleet config - see printers.example.toml. Bambu MQTTS certificate trust
+# (per-printer ca_cert_path/require_valid_cert, and automatic
+# trust-on-first-connect pinning otherwise) is configured per printer in
+# that file, not here.
+CONTINUUM_PRINTERS_PATH=printers.toml
 
 RUST_LOG=info,continuum_proxy=debug

Cargo.toml

@@ -21,10 +21,6 @@ path = "examples/printer_polymorphism.rs"
 name = "test_bambu_certs"
 path = "examples/test_bambu_certs.rs"
 
-[[example]]
-name = "fetch_bambu_cert"
-path = "examples/fetch_bambu_cert.rs"
-
 [[example]]
 name = "auto_connect_bambu"
 path = "examples/auto_connect_bambu.rs"
@@ -34,6 +30,18 @@ name = "bambu_commands_demo"
 path = "examples/bambu_commands_demo.rs"
 
 [dependencies]
+# Real printer protocol clients (MQTT for Bambu, HTTP for PrusaLink/
+# Moonraker, FTPS for gcode transfer) go here once you're ready to build
+# past the stubs in src/printer/ - see that module's doc comment.
+continuum-rust-types = { path = "../continuum-schemas/packages/rust-types" }
+# FOR NON-LOCAL DEV
+# continuum-rust-types = { git = "https://git.octoturge.com/Continuum/continuum-schemas.git" }
+
+# Builds a JSON instance straight from the schema files in
+# continuum-schemas/schemas/ - used for Bambu's raw MQTT JSON messages
+# instead of hand-written serde structs (see bambu.rs).
+continuum-schema-instance = { path = "../continuum-common/packages/rust-schema-instance" }
+
 tokio = { version = "1.40", features = ["full"] }
 tokio-tungstenite = { version = "0.24", features = ["native-tls"] }
 futures-util = "0.3"
@@ -47,10 +55,5 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 anyhow = "1"
 thiserror = "1"
 dotenvy = "0.15"
-
-# Real printer protocol clients (MQTT for Bambu, HTTP for PrusaLink/
-# Moonraker, FTPS for gcode transfer) go here once you're ready to build
-# past the stubs in src/printer/ - see that module's doc comment.
-continuum-rust-types = { path = "../continuum-schemas/packages/rust-types" }
-# FOR NON-LOCAL DEV
-# continuum-rust-types = { git = "https://git.octoturge.com/Continuum/continuum-schemas.git" }
+tokio-serial = { version = "5.5.0", default-features = false }
+rumqttc = { version = "0.25.1", default-features = false, features = ["use-native-tls"] }

README.md

@@ -4,62 +4,90 @@ Edge gateway daemon for the Continuum print farm platform. Runs on a Linux
 SBC on-site, talks to printers over the LAN, and keeps a connection to the
 cloud control plane (`continuum-backend`).
 
-## This is a learning-stage boilerplate
+## What's real right now
 
-This repo is deliberately minimal right now - real printer protocol clients
-(Bambu MQTT+FTPS, PrusaLink REST, Klipper/Moonraker WebSocket) are **not**
-implemented yet. Each vendor is a stub that just prints what it would do
-(`src/printer/bambu.rs`, `prusa.rs`, `klipper.rs`). The idea is to learn
-Rust's polymorphism pattern (trait + enum, since Rust has no class
-inheritance) on something simple before adding real networking on top.
+- **Bambu MQTT** - a real client for both protocol generations (MQTT
+  v3.1.1+BambuMQTT v1 on older firmware, MQTT v5+BambuMQTT v2 on current
+  firmware), auto-detected per printer on connect (a firmware update can
+  move the same physical printer between them, so this isn't something
+  config can hard-code once and forget - see `printer/bambu.rs`'s
+  `BambuProtocolPreference::Auto`). Request/reply correlation via
+  `sequence_id`, an in-memory `print.push_status` cache that merges v1's
+  partial updates and replaces on v2's full ones, and TLS trust that
+  auto-pins a printer's certificate via trust-on-first-connect the first
+  time the bundled CA doesn't verify (`printer/bambu.rs`'s
+  `ensure_trusted`) - see `printer/bambu_mqtt.rs`'s doc comment for the
+  full protocol write-up.
+- **Marlin/RepRap serial** - a real held-open serial connection that paces
+  gcode lines by Marlin's own "ok" acknowledgment (`printer/marlin_serial.rs`).
+- **Wire shapes from schemas, not hand duplication** - Bambu's raw MQTT
+  JSON is built straight from `continuum-schemas`' JSON Schema files via
+  `continuum-schema-instance`, instead of a second hand-written copy of
+  every request shape drifting out of sync with the schema.
+- **Cloud uplink** - a WebSocket client to `continuum-backend` that
+  connects, sends a heartbeat on a timer, and reconnects on drop
+  (`src/uplink/`).
+- **go2rtc watchdog** - restarts the camera-restreaming subprocess if it
+  dies (`src/go2rtc.rs`).
+
+## What's still a stub
+
+PrusaLink (REST) and Klipper/Moonraker (WebSocket) - `printer/prusa.rs`
+and `printer/klipper.rs` just `println!` what they'd send. Also not built
+yet: dispatching a cloud-issued command to a connected printer (the uplink
+and the printer fleet run side by side in `main.rs`, but nothing routes
+between them yet), local buffering for telemetry across connectivity gaps,
+LAN printer discovery (SSDP/mDNS), and the mechanical plate-changer
+interface.
 
 ## Getting started
 
 ```bash
 cp .env.example .env
-cargo run                                  # the daemon: cloud uplink + go2rtc watchdog
-cargo run --example printer_polymorphism   # standalone demo, no network/env needed
+cargo run                                  # the daemon: connects every configured printer, then runs the cloud uplink + go2rtc watchdog
 
 cp printers.example.toml printers.toml     # fill in your real printers (gitignored)
-cargo run --example test_bambu_certs       # real TLS handshake test against each Bambu printer
+cargo run --example auto_connect_bambu     # checks/auto-pins each Bambu printer's certificate, no MQTT needed
+cargo run --example test_bambu_certs       # narrower: just the raw TLS handshake, no auto-pinning
+cargo run --example printer_polymorphism   # standalone trait+enum demo, fake data, no network/env needed
+cargo run --example bambu_commands_demo    # round-trips hand-written structs against real captured Bambu JSON
 ```
 
 ## Structure
 
 ```
 src/
-  main.rs      Runs the uplink and the go2rtc watchdog side by side
+  main.rs               Loads+connects the printer fleet, then runs the uplink and go2rtc watchdog side by side
   config.rs             Loads settings from environment variables
   fleet.rs              Loads printers.toml into ready-to-use PrinterHandles
   uplink/                WebSocket client to continuum-backend: connect, heartbeat, reconnect on drop
-  printer/     GenericPrinter trait + PrinterBase + PrinterHandle enum + one stub per vendor
   go2rtc.rs             Restarts the go2rtc camera-restreaming process if it dies
+  printer/
+    mod.rs               GenericPrinter trait + PrinterBase + PrinterHandle enum - the trait+enum
+                         "inheritance" pattern this project uses instead of class inheritance
+    bambu.rs             The real Bambu printer type: cert trust, protocol version auto-detect, commands
+    bambu_mqtt.rs        The real MQTT wiring underneath bambu.rs - see its doc comment for the protocol details
+    bambu_commands/      Hand-written serde structs mirroring continuum-schemas' bambulab/ shapes
+    marlin_serial.rs     A real held-open Marlin/RepRap serial connection
+    prusa.rs, klipper.rs Still stubs - see "What's still a stub" above
 examples/
-  printer_polymorphism.rs   Runs all five printer stubs through one `connect()` call site
-  test_bambu_certs.rs       Loads printers.toml, does a real TLS handshake to each Bambu printer
+  auto_connect_bambu.rs   Checks/auto-pins every Bambu printer's certificate (no MQTT)
+  test_bambu_certs.rs     Narrower: just a raw TLS handshake per Bambu printer
+  printer_polymorphism.rs Standalone trait+enum demo, five fake printers, no network needed
+  bambu_commands_demo.rs  Round-trips bambu_commands' structs against real captured Bambu JSON
 ```
 
-`src/printer/` is where the "inheritance" question lives - see that
-module's doc comment for the trait+enum pattern this project uses instead
-of class inheritance, and run `printer_polymorphism` to see it work.
-
 `printers.toml` (gitignored - copy from `printers.example.toml`) holds real
-per-printer connection details: host, access code, and - since not every
-Bambu printer trusts the same certificate (see `certs/README.md`) - an
-optional per-printer CA override. `src/fleet.rs` loads it; nothing in
-`main.rs` uses it yet, but `test_bambu_certs` does, as a real (if narrow -
-just the TLS handshake, no MQTT) way to check a printer's certificate
-without needing the full MQTT client built yet.
+per-printer connection details: host, access code/API key/serial port
+depending on vendor, and - for Bambu - an optional per-printer CA
+certificate override (see `certs/README.md`; most printers don't need one,
+since `connect()` auto-pins automatically).
 
-## What's not here yet (on purpose)
+## Learning Rust through this repo
 
-- Real MQTT/FTPS/HTTP/WebSocket printer clients - `src/printer/*.rs` has a
-  `println!` where each of these will go.
-- Local SQLite buffering for telemetry across connectivity gaps.
-- LAN printer discovery (SSDP/mDNS).
-- The mechanical plate-changer interface (serial/GPIO).
-
-Add these back in one at a time as you get comfortable with the Rust
-underneath them - each is its own small lesson (async I/O, a new crate's
-API, error handling for a real protocol) rather than something to absorb
-all at once.
+`src/printer/mod.rs`'s doc comment is where the "how do I do inheritance
+without classes" question lives - trait (shared behavior) + struct
+(shared fields) + enum (the closed set of concrete types), instead of
+`GenericPrinter -> BambuPrinter/PrusaLinkPrinter/...` class inheritance.
+Run `printer_polymorphism` to see it work end to end without needing any
+real printer on the network.

certs/README.md

@@ -28,3 +28,16 @@ certificates at the same time.
 If a printer's connection ever fails certificate verification against this
 file, re-fetch from the source URL above (BambuStudio ships whatever the
 current fleet needs) and update the fingerprint here.
+
+## `pinned/`
+
+Not checked in (gitignored) - where `BambuPrinter::ensure_trusted` saves a
+certificate it auto-pinned via trust-on-first-connect, one file per
+printer, named `<serial number>.pem`. This only happens for a printer
+whose certificate doesn't verify against `bambu_ca2.pem` above (P1P, at
+least) and that doesn't already have an explicit `ca_cert_path` set in
+`printers.toml` - see that method's doc comment in `src/printer/bambu.rs`
+for the full trust-on-first-connect flow, and the SSH-host-key analogy it's
+built on. Delete a printer's file here to force it to re-pin (a real
+error, not silent, if the certificate it captures the second time still
+doesn't verify) the next time it connects.

examples/auto_connect_bambu.rs

@@ -1,12 +1,13 @@
 //! Run with: cargo run --example auto_connect_bambu
 //!
-//! The "smooth, no manual steps" version of test_bambu_certs +
-//! fetch_bambu_cert combined: for each Bambu printer in printers.toml,
-//! makes sure it has a working certificate, auto-pinning one via
-//! trust-on-first-connect if the bundled CA doesn't verify. Safe to run
-//! repeatedly - a printer that's already pinned (certs/pinned/<serial
-//! number>.pem) just gets re-verified against its pin, no network trust
-//! decision is made again.
+//! For each Bambu printer in printers.toml, makes sure it has a working
+//! certificate, auto-pinning one via trust-on-first-connect if the
+//! bundled CA doesn't verify - the exact same `ensure_trusted` step
+//! `BambuPrinter::connect()` runs before opening MQTT, isolated here so
+//! you can check/establish trust without needing a full MQTT handshake to
+//! succeed too. Safe to run repeatedly - a printer that's already pinned
+//! (certs/pinned/<serial number>.pem) just gets re-verified against its
+//! pin, no network trust decision is made again.
 
 use continuum_proxy::fleet;
 use continuum_proxy::printer::PrinterHandle;
@@ -17,8 +18,7 @@ fn main() -> anyhow::Result<()> {
 
     for printer in &mut printers {
         match printer {
-            PrinterHandle::BambuV1(bambu) => report(&bambu.ensure_trusted(cert_dir)),
-            PrinterHandle::BambuV2(bambu) => report(&bambu.ensure_trusted(cert_dir)),
+            PrinterHandle::Bambu(bambu) => report(&bambu.ensure_trusted(cert_dir)),
             _ => {}
         }
     }

examples/bambu_commands_demo.rs

@@ -36,8 +36,14 @@ fn main() {
     // --- print report: arrives under "print", NOT "pushing" (pushall
     // triggers this stream, it isn't a direct reply to it) - includes the
     // full ams/xcam example, with one empty tray and one loaded tray, to
-    // exercise the untagged Empty/Loaded split.
+    // exercise the untagged Empty/Loaded split. sequence_id/command added
+    // by hand here (the original real capture this was built from was
+    // itself a truncated example that happened to omit them, along with
+    // dozens of other real fields the schema defines - not proof the real
+    // message lacks them, see this struct's own doc comment).
     let real_print_report = r#"{
+        "sequence_id": "1",
+        "command": "push_status",
         "ams": {"ams": [{"humidity":"4","id":"0","temp":"22.7","tray":[
             {"id":"0"},
             {"bed_temp":"0","bed_temp_type":"0","cols":["000000FF"],"drying_temp":"0","drying_time":"0","id":"1","nozzle_temp_max":"240","nozzle_temp_min":"190","remain":0,"tag_uid":"0000000000000000","tray_color":"000000FF","tray_diameter":"0.00","tray_id_name":"","tray_info_idx":"GFA00","tray_sub_brands":"","tray_type":"PLA","tray_uuid":"00000000000000000000000000000000","tray_weight":"0","xcam_info":"000000000000000000000000"}

examples/fetch_bambu_cert.rs

@@ -1,37 +0,0 @@
-//! Run with: cargo run --example fetch_bambu_cert -- <printer_id> <output_path>
-//!
-//! "Trust on first connect": connects to one printer from printers.toml
-//! with certificate verification disabled, captures whatever certificate
-//! it presents, and saves it as PEM. Meant for a printer like P1P that
-//! doesn't chain to the bundled CA - after running this, point that
-//! printer's `ca_cert_path` in printers.toml at the saved file and
-//! `test_bambu_certs` should report it as verified from then on.
-//!
-//! This trusts whoever answers on the network *right now* - only run it on
-//! a network you trust, ideally right after unboxing the printer.
-
-use continuum_proxy::fleet;
-use continuum_proxy::printer::{GenericPrinter, PrinterHandle};
-
-fn main() -> anyhow::Result<()> {
-    let mut args = std::env::args().skip(1);
-    let printer_id = args.next().ok_or_else(|| anyhow::anyhow!("usage: fetch_bambu_cert <printer_id> <output_path>"))?;
-    let output_path = args.next().ok_or_else(|| anyhow::anyhow!("usage: fetch_bambu_cert <printer_id> <output_path>"))?;
-
-    let printers = fleet::load(std::path::Path::new("printers.toml"))?;
-
-    let pem = printers
-        .iter()
-        .find_map(|printer| match printer {
-            PrinterHandle::BambuV1(bambu) if bambu.base().id == printer_id => Some(bambu.fetch_certificate()),
-            PrinterHandle::BambuV2(bambu) if bambu.base().id == printer_id => Some(bambu.fetch_certificate()),
-            _ => None,
-        })
-        .ok_or_else(|| anyhow::anyhow!("no Bambu printer with id '{printer_id}' in printers.toml"))??;
-
-    std::fs::write(&output_path, &pem)?;
-    println!("saved {printer_id}'s certificate to {output_path}");
-    println!("now set ca_cert_path = \"{output_path}\" for {printer_id} in printers.toml");
-
-    Ok(())
-}

examples/printer_polymorphism.rs

@@ -4,36 +4,40 @@
 //! - `connect()` is called the same way regardless of vendor (polymorphism),
 //!   but each variant's `impl GenericPrinter` runs completely different
 //!   code (override).
-//! - `dispatch_gcode()` only exists on the two Bambu structs (extension) -
-//!   you have to `match` the enum back down to the concrete type to reach
-//!   it, which is the trade-off for not having implicit downcasting.
+//! - `dispatch_gcode()` only exists on the Bambu struct (extension) - you
+//!   have to `match` the enum back down to the concrete type to reach it,
+//!   which is the trade-off for not having implicit downcasting.
 
 use continuum_proxy::printer::{
-    BambuTls, BambuV1Printer, BambuV2Printer, GenericPrinter, KlipperPrinter, PrinterHandle, PrusaLinkPrinter,
-    PrusaSerialPrinter,
+    BambuPrinter, BambuProtocolPreference, BambuTls, GenericPrinter, KlipperPrinter, PrinterHandle,
+    PrusaLinkPrinter, PrusaSerialPrinter,
 };
 
 #[tokio::main]
 async fn main() {
     let mut fleet: Vec<PrinterHandle> = vec![
-        PrinterHandle::BambuV1(BambuV1Printer::new(
+        // Forced: a known-old printer.
+        PrinterHandle::Bambu(BambuPrinter::new(
             "p1",
             "P1S",
             "192.168.1.49",
             "87654321",
             "01P00A000000000",
             BambuTls::BundledCa,
+            BambuProtocolPreference::ForceV1,
         )),
-        PrinterHandle::BambuV2(BambuV2Printer::new(
+        // Auto: the recommended default - see bambu.rs's doc comment.
+        PrinterHandle::Bambu(BambuPrinter::new(
             "p2",
             "X1 Carbon",
             "192.168.1.50",
             "12345678",
             "00M00A000000000",
             BambuTls::BundledCa,
+            BambuProtocolPreference::Auto,
         )),
         PrinterHandle::PrusaLink(PrusaLinkPrinter::new("p3", "MK4", "192.168.1.51", "prusa-api-key")),
-        PrinterHandle::PrusaSerial(PrusaSerialPrinter::new("p4", "MK3S+", 0)),
+        PrinterHandle::PrusaSerial(PrusaSerialPrinter::new("p4", "MK3S+", "/dev/ttyACM0", 115_200)),
         PrinterHandle::Klipper(KlipperPrinter::new("p5", "Voron 2.4", "192.168.1.52")),
         // No host configured - this one will hit the Err path.
         PrinterHandle::Klipper(KlipperPrinter::new("p6", "Broken Voron", "")),
@@ -50,15 +54,15 @@ async fn main() {
     }
 
     // Vendor-only extension: reachable only after matching the concrete
-    // variant back out of the enum. BambuV1Printer and BambuV2Printer are
-    // separate types now (each *has* a BambuGenericPrinter rather than
-    // being the same struct), so - unlike before the split - this needs
-    // two arms instead of one `A(x) | B(x)` pattern: an or-pattern requires
-    // every alternative to bind the same type, and these no longer do.
+    // variant back out of the enum - one arm now that Bambu is a single
+    // type regardless of which protocol version `connect()` settled on.
+    // `print_status()` is real Bambu-only behavior with no equivalent on
+    // `GenericPrinter` (no other vendor here has an in-memory report
+    // cache) - none of these connected above (fake IPs), so it's always
+    // `None`, but the point is the downcast, not a live reading.
     for printer in &fleet {
         match printer {
-            PrinterHandle::BambuV1(bambu) => bambu.dispatch_gcode("part.gcode.3mf"),
-            PrinterHandle::BambuV2(bambu) => bambu.dispatch_gcode("part.gcode.3mf"),
+            PrinterHandle::Bambu(bambu) => println!("  {}: cached print_status = {:?}", bambu.sn, bambu.print_status().await),
             _ => {}
         }
     }

examples/test_bambu_certs.rs

@@ -12,13 +12,9 @@ use continuum_proxy::printer::{GenericPrinter, PrinterHandle};
 fn main() -> anyhow::Result<()> {
     let printers = fleet::load(std::path::Path::new("printers.toml"))?;
 
-    // BambuV1Printer and BambuV2Printer are separate types (each *has* a
-    // BambuGenericPrinter rather than being the same struct), so this
-    // needs one arm per variant rather than a single `A(x) | B(x)` pattern.
     for printer in &printers {
         match printer {
-            PrinterHandle::BambuV1(bambu) => report(bambu.base().name.as_str(), bambu.test_tls_handshake()),
-            PrinterHandle::BambuV2(bambu) => report(bambu.base().name.as_str(), bambu.test_tls_handshake()),
+            PrinterHandle::Bambu(bambu) => report(bambu.base().name.as_str(), bambu.test_tls_handshake()),
             _ => {}
         }
     }

printers.example.toml

@@ -5,20 +5,26 @@
 [[printer]]
 id = "p1p"
 name = "P1P"
-vendor = "bambu_v1"                    # bambu_v1 | bambu_v2 | prusa_link | prusa_serial | klipper
+# Forced instead of "bambu" here since this one's a known-old printer -
+# see fleet.rs's PrinterEntry doc comment for the full vendor list.
+vendor = "bambu_v1"
 host = "192.168.1.50"
 access_code = "REPLACE_WITH_ACCESS_CODE"
 sn = "REPLACE_WITH_SERIAL_NUMBER"      # printer screen > Settings > Device (used for MQTT topics later)
-# P1P doesn't chain to the bundled CA - point this at a cert you've
-# downloaded from the printer itself. Leave unset to use the bundled CA.
-# `cargo run --example fetch_bambu_cert -- p1p ./certs/p1p.pem` can fetch it
-# for you (trust-on-first-connect - see that example's doc comment).
-ca_cert_path = "./certs/p1p.pem"
+# No ca_cert_path needed even though P1P doesn't chain to the bundled CA -
+# `connect()` auto-pins a certificate via trust-on-first-connect the first
+# time it can't verify against the bundled one, and saves it under
+# certs/pinned/<sn>.pem for every run after that (see BambuPrinter::
+# ensure_trusted's doc comment). Only set ca_cert_path explicitly if you
+# want to point at a certificate you've captured/vetted some other way.
 
 [[printer]]
 id = "h2c"
 name = "H2C"
-vendor = "bambu_v2"
+# "bambu" auto-detects MQTT v5-vs-v3.1.1 (BambuMQTT v2-vs-v1) on connect -
+# the recommended default, since a firmware update can change which one a
+# given printer speaks without you knowing ahead of time.
+vendor = "bambu"
 host = "192.168.1.51"
 access_code = "REPLACE_WITH_ACCESS_CODE"
 sn = "REPLACE_WITH_SERIAL_NUMBER"
@@ -37,3 +43,10 @@ sn = "REPLACE_WITH_SERIAL_NUMBER"
 # vendor = "prusa_link"
 # host = "192.168.1.53"
 # api_key = "REPLACE_WITH_API_KEY"
+
+# [[printer]]
+# id = "mk3s"
+# name = "MK3S+"
+# vendor = "prusa_serial"
+# port = "/dev/ttyACM0"
+# baud_rate = 115200            # optional, defaults to 115200

src/config.rs

@@ -1,6 +1,4 @@
-use std::collections::HashMap;
 use std::env;
-use std::path::PathBuf;
 
 #[derive(Debug, Clone)]
 pub struct Config {
@@ -10,19 +8,13 @@ pub struct Config {
     pub uplink_url: String,
     pub go2rtc_bin: String,
     pub go2rtc_config: String,
-
-    /// Per-printer CA certificate overrides, keyed by printer ID. Most
-    /// Bambu printers chain to the bundled shared CA (`certs/bambu_ca2.pem`)
-    /// and don't need an entry here - but some models (P1P, at least) ship
-    /// their own certificate instead of one signed by that shared CA, with
-    /// no single file covering all of them. Look up a printer's own entry
-    /// (if any) when constructing it; there's no fleet-wide "the" override.
-    pub bambu_ca_cert_overrides: HashMap<String, PathBuf>,
-    /// When `false`, Bambu MQTTS connections skip certificate verification
-    /// entirely instead of checking against a CA. `true` (the default) is
-    /// the secure choice; only flip this fleet-wide policy if you have
-    /// printers you can't get to verify any other way.
-    pub bambu_require_valid_cert: bool,
+    /// Path to the fleet config file - see `fleet.rs` and
+    /// `printers.example.toml` for the shape. Bambu certificate trust
+    /// (per-printer `ca_cert_path`/`require_valid_cert`, and automatic
+    /// trust-on-first-connect pinning otherwise - see
+    /// `BambuPrinter::ensure_trusted`) lives entirely in that file now,
+    /// not as a separate fleet-wide env-configured policy here.
+    pub printers_path: String,
 }
 
 impl Config {
@@ -34,22 +26,11 @@ 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"),
-
-            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",
+            printers_path: env_or("CONTINUUM_PRINTERS_PATH", "printers.toml"),
         })
     }
 }
 
-/// Parses `"printer_id=path,printer_id2=path2"` into a lookup map. Empty
-/// input (the common case - no overrides needed) yields an empty map.
-fn parse_cert_overrides(raw: &str) -> HashMap<String, PathBuf> {
-    raw.split(',')
-        .filter_map(|entry| entry.split_once('='))
-        .map(|(id, path)| (id.trim().to_string(), PathBuf::from(path.trim())))
-        .collect()
-}
-
 fn require(key: &str) -> anyhow::Result<String> {
     env::var(key).map_err(|_| anyhow::anyhow!("missing required env var {key}"))
 }

src/fleet.rs

@@ -7,7 +7,10 @@ use std::path::Path;
 
 use serde::Deserialize;
 
-use crate::printer::{BambuTls, BambuV1Printer, BambuV2Printer, KlipperPrinter, PrinterHandle, PrusaLinkPrinter, PrusaSerialPrinter};
+use crate::printer::{
+    BambuPrinter, BambuProtocolPreference, BambuTls, KlipperPrinter, PrinterHandle, PrusaLinkPrinter,
+    PrusaSerialPrinter,
+};
 
 #[derive(Debug, Deserialize)]
 struct FleetFile {
@@ -19,7 +22,10 @@ struct FleetFile {
 struct PrinterEntry {
     id: String,
     name: String,
-    /// One of: "bambu_v1", "bambu_v2", "prusa_link", "prusa_serial", "klipper".
+    /// One of: "bambu" (auto-detects MQTT v5-vs-v3.1.1/BambuMQTT v2-vs-v1 -
+    /// the default, since a firmware update can move the same physical
+    /// printer between them), "bambu_v1"/"bambu_v2" (force a specific one),
+    /// "prusa_link", "prusa_serial", "klipper".
     vendor: String,
     host: Option<String>,
     access_code: Option<String>,
@@ -27,7 +33,10 @@ struct PrinterEntry {
     /// by serial (`device/{sn}/report`), not by our own `id`.
     sn: Option<String>,
     api_key: Option<String>,
-    com_port: Option<i16>,
+    /// Serial device path, e.g. `/dev/ttyACM0` on Linux.
+    port: Option<String>,
+    #[serde(default = "default_baud_rate")]
+    baud_rate: u32,
     /// Path to a printer-specific CA certificate. Leave unset to use the
     /// bundled Bambu CA - only needed for a printer that doesn't chain to
     /// it (P1P, at least; see certs/README.md).
@@ -40,6 +49,11 @@ fn default_true() -> bool {
     true
 }
 
+/// Marlin's usual default on most current Prusa firmware.
+fn default_baud_rate() -> u32 {
+    115_200
+}
+
 /// Reads and parses a printers file into ready-to-use `PrinterHandle`s.
 pub fn load(path: &Path) -> anyhow::Result<Vec<PrinterHandle>> {
     let text = std::fs::read_to_string(path).map_err(|err| {
@@ -54,28 +68,17 @@ pub fn load(path: &Path) -> anyhow::Result<Vec<PrinterHandle>> {
 
 fn build_printer(entry: PrinterEntry) -> anyhow::Result<PrinterHandle> {
     match entry.vendor.as_str() {
-        "bambu_v1" => {
-            let host = require_field(&entry.id, "host", entry.host)?;
-            let access_code = require_field(&entry.id, "access_code", entry.access_code)?;
-            let sn = require_field(&entry.id, "sn", entry.sn)?;
-            let tls = BambuTls::resolve(entry.ca_cert_path.as_deref().map(Path::new), entry.require_valid_cert)?;
-            Ok(PrinterHandle::BambuV1(BambuV1Printer::new(&entry.id, &entry.name, &host, &access_code, &sn, tls)))
-        }
-        "bambu_v2" => {
-            let host = require_field(&entry.id, "host", entry.host)?;
-            let access_code = require_field(&entry.id, "access_code", entry.access_code)?;
-            let sn = require_field(&entry.id, "sn", entry.sn)?;
-            let tls = BambuTls::resolve(entry.ca_cert_path.as_deref().map(Path::new), entry.require_valid_cert)?;
-            Ok(PrinterHandle::BambuV2(BambuV2Printer::new(&entry.id, &entry.name, &host, &access_code, &sn, tls)))
-        }
+        "bambu" => build_bambu(entry, BambuProtocolPreference::Auto),
+        "bambu_v1" => build_bambu(entry, BambuProtocolPreference::ForceV1),
+        "bambu_v2" => build_bambu(entry, BambuProtocolPreference::ForceV2),
         "prusa_link" => {
             let host = require_field(&entry.id, "host", entry.host)?;
             let api_key = require_field(&entry.id, "api_key", entry.api_key)?;
             Ok(PrinterHandle::PrusaLink(PrusaLinkPrinter::new(&entry.id, &entry.name, &host, &api_key)))
         }
         "prusa_serial" => {
-            let com_port = entry.com_port.ok_or_else(|| anyhow::anyhow!("{}: prusa_serial needs com_port", entry.id))?;
-            Ok(PrinterHandle::PrusaSerial(PrusaSerialPrinter::new(&entry.id, &entry.name, com_port)))
+            let port = require_field(&entry.id, "port", entry.port)?;
+            Ok(PrinterHandle::PrusaSerial(PrusaSerialPrinter::new(&entry.id, &entry.name, &port, entry.baud_rate)))
         }
         "klipper" => {
             let host = require_field(&entry.id, "host", entry.host)?;
@@ -85,6 +88,14 @@ fn build_printer(entry: PrinterEntry) -> anyhow::Result<PrinterHandle> {
     }
 }
 
+fn build_bambu(entry: PrinterEntry, preference: BambuProtocolPreference) -> anyhow::Result<PrinterHandle> {
+    let host = require_field(&entry.id, "host", entry.host)?;
+    let access_code = require_field(&entry.id, "access_code", entry.access_code)?;
+    let sn = require_field(&entry.id, "sn", entry.sn)?;
+    let tls = BambuTls::resolve(entry.ca_cert_path.as_deref().map(Path::new), entry.require_valid_cert)?;
+    Ok(PrinterHandle::Bambu(BambuPrinter::new(&entry.id, &entry.name, &host, &access_code, &sn, tls, preference)))
+}
+
 fn require_field(printer_id: &str, field: &str, value: Option<String>) -> anyhow::Result<String> {
     value.ok_or_else(|| anyhow::anyhow!("{printer_id}: missing required field '{field}'"))
 }

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
-    // camera-restreaming process. More will join them later - printer
-    // discovery, the plate changer - as you build past src/printer/'s
-    // stubs. `tokio::select!` runs both and stops as soon as either one
-    // returns; both loop forever below, so in practice this runs until the
-    // process is killed.
+    // Loaded and connected here, once, at startup - nothing yet re-reads
+    // printers.toml on a running daemon (adding/removing a printer means
+    // a restart), and nothing yet lets the uplink dispatch a command to
+    // one of these once connected - see PrinterHandle's own doc comment
+    // for what's built and what's still a stub per vendor. Keeping
+    // `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.rs

@@ -1,15 +1,40 @@
 use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::time::Duration;
 
 use super::{GenericPrinter, PrinterBase, PrinterError};
 
+// region: Certificates
+
+/// The canonical shape for every Bambu MQTT message lives in
+/// continuum-schemas, not here - see `send_gcode_line` below, which builds
+/// an instance of it directly via `continuum-schema-instance` instead of a
+/// hand-written serde struct. `v1`/`v2` underneath this directory are
+/// BambuMQTT's two message schemas, not to be confused with the MQTT
+/// *transport* version (v3.1.1 vs v5) below - they happen to move
+/// together on real Bambu firmware, but they're two different things.
+const BAMBU_SCHEMA_DIR: &str = concat!(
+    env!("CARGO_MANIFEST_DIR"),
+    "/../continuum-schemas/schemas/bambulab"
+);
+
+fn get_schema_for_command(version: BambuProtocolVersion, root: String, command: String) -> PathBuf {
+    Path::new(BAMBU_SCHEMA_DIR)
+        .join(version.schema_dir())
+        .join(format!("request/{root}/{command}.schema.json"))
+}
+
 /// Bambu's shared LAN-mode root CA ("BBL CA2 RSA") - see certs/README.md
 /// for how this was obtained and verified. `include_bytes!` embeds it into
 /// the compiled binary at build time, so there's no file to ship alongside
 /// the executable.
 const BUNDLED_CA: &[u8] = include_bytes!("../../certs/bambu_ca2.pem");
 
+/// Where auto-pinned certificates get saved - same directory
+/// `examples/auto_connect_bambu.rs` already used by hand. Relative to
+/// wherever the daemon is run from, same convention as `printers.toml`.
+const CERT_DIR: &str = "certs/pinned";
+
 /// Which certificate to trust when connecting to a Bambu printer's MQTTS
 /// port (8883 - Bambu's LAN mode is TLS-only, there's no unencrypted
 /// fallback).
@@ -26,11 +51,13 @@ pub enum BambuTls {
 
 impl BambuTls {
     /// Resolves the trust mode for *one* printer. Call this per printer,
-    /// passing `config.bambu_ca_cert_overrides.get(printer_id)` - there's
-    /// no single override for the whole fleet, because not every Bambu
-    /// model chains to the same CA. `require_valid_cert = false` always
-    /// means `Insecure`, regardless of whether that printer also has an
-    /// override entry.
+    /// passing that printer's own `ca_cert_path`/`require_valid_cert` from
+    /// `printers.toml` (see `fleet.rs::build_bambu`) - there's no single
+    /// override for the whole fleet, because not every Bambu model chains
+    /// to the same CA, and most printers don't need this at all (see
+    /// `BambuPrinter::ensure_trusted` for the automatic path).
+    /// `require_valid_cert = false` always means `Insecure`, regardless of
+    /// whether that printer also has a `cert_path`.
     pub fn resolve(cert_path: Option<&Path>, require_valid_cert: bool) -> anyhow::Result<Self> {
         if !require_valid_cert {
             return Ok(BambuTls::Insecure);
@@ -53,7 +80,11 @@ impl BambuTls {
 
     /// Builds a `native_tls` connector configured for this trust mode:
     /// trusting only `ca_bytes()`, or accepting anything for `Insecure`.
-    fn build_connector(&self) -> anyhow::Result<native_tls::TlsConnector> {
+    /// `pub(crate)`, not private: `bambu_mqtt.rs` (a sibling module, not a
+    /// descendant of this one) reuses this exact connector to configure
+    /// rumqttc's TLS transport, instead of every consumer's TLS trust
+    /// decisions living in two separate places.
+    pub(crate) fn build_connector(&self) -> anyhow::Result<native_tls::TlsConnector> {
         let mut builder = native_tls::TlsConnector::builder();
 
         // Bambu printer certificates don't carry their LAN IP as a Subject
@@ -62,22 +93,52 @@ impl BambuTls {
         // to be off regardless of trust mode: what matters is whether the
         // certificate chains to a CA we trust, not whether its SAN matches
         // the specific IP we happened to dial today. Chain verification
-        // itself stays fully enforced below for BundledCa/Custom - this
-        // narrows *what* gets checked, it doesn't turn checking off.
+        // itself stays fully enforced below for BundledCa - this narrows
+        // *what* gets checked, it doesn't turn checking off.
         builder.danger_accept_invalid_hostnames(true);
 
-        if matches!(self, BambuTls::Insecure) {
-            builder.danger_accept_invalid_certs(true);
-        } else {
+        match self {
+            BambuTls::BundledCa => {
+                // A real CA root - chain validation is exactly the right
+                // check here.
                 builder.add_root_certificate(native_tls::Certificate::from_pem(self.ca_bytes())?);
             }
+            BambuTls::Custom(_) => {
+                // `ensure_trusted` pins a *leaf* certificate (the printer's
+                // own, CN = its serial number, issued by Bambu's own CA -
+                // not a root), for byte-exact fingerprint comparison, not
+                // chain validation. Handing that same leaf to
+                // `add_root_certificate` here would ask OpenSSL to treat it
+                // as its own trust anchor - exactly the bug
+                // `test_tls_handshake`'s doc comment already describes
+                // ("self-signed certificate in certificate chain"), just
+                // not fixed at this second call site too until now.
+                // native_tls has no hook for a custom per-certificate
+                // verification callback (unlike rustls, which would let us
+                // do the real fingerprint check here) - so this accepts
+                // any certificate at the TLS layer instead, relying on
+                // `ensure_trusted` already having fingerprint-verified this
+                // exact printer moments before `connect()` opens this
+                // transport. Weaker than continuous fingerprint pinning on
+                // every reconnect, but not a fresh regression: pinning a
+                // leaf as a root here never actually worked in the first
+                // place.
+                builder.danger_accept_invalid_certs(true);
+            }
+            BambuTls::Insecure => {
+                builder.danger_accept_invalid_certs(true);
+            }
+        }
 
         Ok(builder.build()?)
     }
 }
 
 fn resolve_addr(host: &str) -> anyhow::Result<SocketAddr> {
-    (host, 8883).to_socket_addrs()?.next().ok_or_else(|| anyhow::anyhow!("could not resolve {host}"))
+    (host, 8883)
+        .to_socket_addrs()?
+        .next()
+        .ok_or_else(|| anyhow::anyhow!("could not resolve {host}"))
 }
 
 /// Minimal DER -> PEM encoder (base64, 64-char lines, standard
@@ -102,28 +163,90 @@ fn pem_to_der(pem: &[u8]) -> anyhow::Result<Vec<u8>> {
     Ok(native_tls::Certificate::from_pem(pem)?.to_der()?)
 }
 
-/// Fields and behavior every Bambu printer shares, regardless of protocol
-/// generation - the layer between `GenericPrinter` and the version-specific
-/// `BambuV1Printer`/`BambuV2Printer` leaves below. Those two *have* one of
-/// these (composition) rather than a deeper inheritance chain; because
-/// Rust's field/method privacy is scoped to the *module* (this whole file),
-/// not the struct, `BambuV1Printer`/`BambuV2Printer` can freely use
-/// `BambuGenericPrinter`'s private fields even though they're separate
-/// types.
-pub struct BambuGenericPrinter {
+// endregion: Certificates
+
+// region: BambuMQTTProtocol
+/// Which BambuMQTT message schema (and, on real firmware, which MQTT
+/// transport version) a connection ended up using. This is an *outcome*
+/// `connect()` settles on, not something you can assume from the printer
+/// model - a firmware update can move the same physical X1C from v1 to
+/// v2. Every later message this printer sends is built against
+/// `schemas/bambulab/{v1,v2}/` depending on which one this is.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BambuProtocolVersion {
+    /// MQTT v3.1.1, BambuMQTT v1 - older firmware (P1P/P1S, original X1 launch firmware).
+    V1,
+    /// MQTT v5, BambuMQTT v2 - current firmware.
+    V2,
+}
+
+impl BambuProtocolVersion {
+    fn schema_dir(self) -> &'static str {
+        match self {
+            BambuProtocolVersion::V1 => "v1",
+            BambuProtocolVersion::V2 => "v2",
+        }
+    }
+}
+
+/// How to pick the protocol generation for a given printer - config
+/// *input*, as opposed to `BambuProtocolVersion` (the outcome `connect()`
+/// settles on). `Auto` is the default for exactly the reason
+/// `BambuProtocolVersion` isn't derivable from config alone: you can't
+/// reliably know which one a specific printer speaks today without
+/// asking it. `ForceV1`/`ForceV2` stay available as an override - useful
+/// if auto-detection ever misbehaves, or to test one path deliberately.
+#[derive(Debug, Clone, Copy)]
+pub enum BambuProtocolPreference {
+    /// Try MQTT v5 first; if the broker's CONNACK rejects that protocol
+    /// version (older firmware only understands v3.1.1), fall back and
+    /// reconnect with v3.1.1 instead.
+    Auto,
+    ForceV1,
+    ForceV2,
+}
+// endregion: BambuMQTTProtocol
+
+/// A Bambu printer. Only one struct now, not two: the old
+/// `BambuGenericPrinter` (shared fields/behavior) + `BambuV1Printer`/
+/// `BambuV2Printer` (thin per-version wrappers) split existed to let two
+/// *separate types* share a base - but since which protocol version a
+/// printer uses is now something `connect()` discovers at runtime rather
+/// than something config picks between two types for, there's only ever
+/// one type to begin with, so the extra layer stopped earning its keep.
+pub struct BambuPrinter {
     base: PrinterBase,
     pub access_code: String,
     pub sn: String,
     tls: BambuTls,
+    preference: BambuProtocolPreference,
+    /// `None` until `connect()` succeeds. Carries the live MQTT session
+    /// 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)>,
 }
 
-impl BambuGenericPrinter {
-    pub fn new(id: &str, name: &str, host: &str, access_code: &str, sn: &str, tls: BambuTls) -> Self {
+impl BambuPrinter {
+    pub fn new(
+        id: &str,
+        name: &str,
+        host: &str,
+        access_code: &str,
+        sn: &str,
+        tls: BambuTls,
+        preference: BambuProtocolPreference,
+    ) -> Self {
         Self {
-            base: PrinterBase { id: id.into(), name: name.into(), host: host.into() },
+            base: PrinterBase {
+                id: id.into(),
+                name: name.into(),
+                host: host.into(),
+            },
             access_code: access_code.into(),
             sn: sn.into(),
             tls,
+            preference,
+            connection: None,
         }
     }
 
@@ -131,13 +254,7 @@ impl BambuGenericPrinter {
         &self.base
     }
 
-    /// Not part of the `GenericPrinter` trait - Prusa/Klipper have no
-    /// equivalent, since FTPS gcode transfer is a Bambu-specific thing.
-    pub fn dispatch_gcode(&self, file_name: &str) {
-        println!("[{}] uploading {file_name} over FTPS", self.base.name);
-    }
-
-    /// A real (blocking, one-shot) TLS handshake to port 8883 - no MQTT
+    /* A real (blocking, one-shot) TLS handshake to port 8883 - no MQTT
     /// protocol, just "does the certificate verify" against this printer's
     /// configured trust mode.
     ///
@@ -155,13 +272,15 @@ impl BambuGenericPrinter {
     /// the very next connection). So instead: connect with verification
     /// off, then compare the certificate actually presented against the
     /// exact bytes pinned, byte for byte. A mismatch is a hard failure -
-    /// no PKI judgment call, just "is this the same certificate as before."
+    /// no PKI judgment call, just "is this the same certificate as before." */
     pub fn test_tls_handshake(&self) -> anyhow::Result<()> {
         let addr = resolve_addr(&self.base.host)?;
         let stream = TcpStream::connect_timeout(&addr, Duration::from_secs(5))?;
 
         if let BambuTls::Custom(pinned_pem) = &self.tls {
-            let tls_stream = BambuTls::Insecure.build_connector()?.connect(&self.base.host, stream)?;
+            let tls_stream = BambuTls::Insecure
+                .build_connector()?
+                .connect(&self.base.host, stream)?;
             let presented = tls_stream
                 .peer_certificate()?
                 .ok_or_else(|| anyhow::anyhow!("{} presented no certificate", self.base.name))?
@@ -170,32 +289,39 @@ impl BambuGenericPrinter {
             if presented != pem_to_der(pinned_pem)? {
                 anyhow::bail!(
                     "{} presented a certificate different from the one pinned - could be a legitimate \
-                     certificate rotation, could be something worse; re-run fetch_bambu_cert deliberately \
-                     if you're sure it's the former",
+                     certificate rotation, could be something worse; delete its file under certs/pinned/ \
+                     and reconnect to re-pin deliberately if you're sure it's the former",
                     self.base.name
                 );
             }
             return Ok(());
         }
 
-        self.tls.build_connector()?.connect(&self.base.host, stream)?;
+        self.tls
+            .build_connector()?
+            .connect(&self.base.host, stream)?;
         Ok(())
     }
 
-    /// "Trust on first connect": connects once with certificate
+    /* "Trust on first connect": connects once with certificate
     /// verification disabled, captures whatever certificate the printer
     /// presents, and returns it as PEM bytes.
     ///
     /// This is meaningfully weaker than verifying against a CA you already
-    /// trust - anyone on the network at the exact moment you run this gets
-    /// trusted forever after. That's why it's a method you call once, by
-    /// hand (see `examples/fetch_bambu_cert.rs`), rather than something
-    /// `connect()` falls back to silently. Save the result to a file and
-    /// point that printer's `ca_cert_path` in `printers.toml` at it.
+    /// trust - anyone on the network at the exact moment this runs gets
+    /// trusted forever after. `ensure_trusted` below is the only caller -
+    /// it only reaches this path when the bundled CA didn't verify, saves
+    /// the result under `certs/pinned/`, and every connection after this
+    /// one re-verifies against that pinned file instead of trusting the
+    /// network again. Not exposed as a manual step for the caller to run
+    /// ahead of time - there's no longer a reason to, now that `connect()`
+    /// does this automatically the first time a printer needs it. */
     pub fn fetch_certificate(&self) -> anyhow::Result<Vec<u8>> {
         let addr = resolve_addr(&self.base.host)?;
         let stream = TcpStream::connect_timeout(&addr, Duration::from_secs(5))?;
-        let tls_stream = BambuTls::Insecure.build_connector()?.connect(&self.base.host, stream)?;
+        let tls_stream = BambuTls::Insecure
+            .build_connector()?
+            .connect(&self.base.host, stream)?;
 
         let cert = tls_stream
             .peer_certificate()?
@@ -204,10 +330,9 @@ impl BambuGenericPrinter {
         Ok(der_to_pem(&cert.to_der()?))
     }
 
-    /// Makes sure this printer has a *working* trust configuration,
+    /* Makes sure this printer has a *working* trust configuration,
     /// auto-pinning a certificate via trust-on-first-connect if needed -
-    /// the "smooth, no manual steps" version of the fetch_bambu_cert
-    /// workflow, built the way SSH handles host keys:
+    /// no manual step required, built the way SSH handles host keys:
     ///
     /// - A pinned certificate already on disk (`cert_dir/<sn>.pem`) is used
     ///   directly. No new trust decision gets made on every run - that
@@ -230,7 +355,7 @@ impl BambuGenericPrinter {
     ///   in printers.toml) never auto-pins over it - a verification
     ///   failure there is a real error, not something to paper over,
     ///   since that's exactly the signal pinning exists to give you (the
-    ///   cert rotated, or something worse).
+    ///   cert rotated, or something worse). */
     pub fn ensure_trusted(&mut self, cert_dir: &Path) -> anyhow::Result<()> {
         let pinned_path = cert_dir.join(format!("{}.pem", self.sn));
 
@@ -241,7 +366,10 @@ impl BambuGenericPrinter {
         match self.test_tls_handshake() {
             Ok(()) => Ok(()),
             Err(err) if matches!(self.tls, BambuTls::BundledCa) => {
-                eprintln!("{}: bundled CA didn't verify ({err}); auto-pinning on first connect", self.base.name);
+                eprintln!(
+                    "{}: bundled CA didn't verify ({err}); auto-pinning on first connect",
+                    self.base.name
+                );
                 let pem = self.fetch_certificate()?;
                 std::fs::create_dir_all(cert_dir)?;
                 std::fs::write(&pinned_path, &pem)?;
@@ -252,112 +380,220 @@ impl BambuGenericPrinter {
         }
     }
 
-    /// Shared by both `BambuV1Printer` and `BambuV2Printer` - same MQTT
-    /// gcode-injection mechanism regardless of protocol generation. Real
-    /// version would publish
-    /// `{"print":{"command":"gcode_line","param":"M106 P1 S<pwm>"}}` to
-    /// `device/{sn}/request` over the MQTTS session.
-    async fn set_fan_speed_impl(&mut self, percent: u8) -> Result<(), PrinterError> {
+    /// Bambu-only - no `GenericPrinter` equivalent, since every vendor's
+    /// way of reporting firmware info (if it has one at all) is shaped
+    /// completely differently. Same "vendor-only extension" pattern as
+    /// `test_tls_handshake`/`fetch_certificate`/`ensure_trusted` above:
+    /// reached by matching `PrinterHandle::Bambu` back out of the enum,
+    /// not through the shared trait.
+    ///
+    /// The first real use of `BambuMqttSession::request` -
+    /// `info.get_version` is a genuine one-shot, correlated reply (unlike
+    /// `pushing.pushall`, see `request`'s doc comment), so this is exactly
+    /// the case it's for.
+    pub async fn get_version(&self) -> Result<super::GetVersionResponse, PrinterError> {
+        let (session, version) =
+            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
+
+        let body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
+            *version,
+            "info".into(),
+            "get_version".into(),
+        ))
+        .map_err(|err| PrinterError(format!("failed to build get_version request: {err}")))?;
+
+        let response = session.request("info", body).await?;
+        serde_json::from_value(response)
+            .map_err(|err| PrinterError(format!("failed to parse get_version response: {err}")))
+    }
+
+    /// Bambu-only, same reasoning as `get_version` above. Sends
+    /// `print.calibration` and waits for the correlated `print.calibration`
+    /// *report* it triggers - confirmed live on both protocol versions,
+    /// which respond identically here (an earlier, incorrect assumption
+    /// had v1 folding this into `push_status` instead of replying
+    /// directly; fixed in continuum-schemas). The printer also keeps
+    /// sending ordinary `push_status` updates reflecting progress while
+    /// calibration runs, same as any other job - read those via
+    /// `print_status()` if you want progress, this call just confirms the
+    /// calibration sequence started and got acknowledged.
+    ///
+    /// `option` is the bitmask this command's own schema defines:
+    /// `lidarCalibration = 1`, `bedLevelling = 1 << 1`,
+    /// `vibrationCompensation = 1 << 2`, `motorCancellation = 1 << 3` (OR
+    /// together whichever steps you want). Deliberately not defaulted to
+    /// "everything" here - this starts a real, physical calibration
+    /// sequence on the printer, so the caller states exactly what they
+    /// mean to trigger rather than this method guessing.
+    pub async fn get_calibration(&self, option: i64) -> Result<serde_json::Value, PrinterError> {
+        let (session, version) =
+            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
+
+        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
+            *version,
+            "print".into(),
+            "calibration".into(),
+        ))
+        .map_err(|err| PrinterError(format!("failed to build calibration request: {err}")))?;
+        body["option"] = serde_json::Value::from(option);
+
+        session.request("print", body).await
+    }
+
+    /// Bambu-only, same reasoning as `get_version` above. Publishes
+    /// `pushing.pushall`, which - unlike `get_version` - doesn't reply
+    /// under its own category: it triggers/refreshes the ongoing
+    /// `print.push_status` stream instead. Fire-and-forget, same as
+    /// `send_gcode_line`: there's nothing to wait for here, since the
+    /// forced full update this triggers self-describes as such via its
+    /// own `msg` field once it arrives (see `PrintStatusCache::apply`),
+    /// whenever that turns out to be - waiting on it added two real bugs
+    /// (a race, and matching the wrong "print" message) for no benefit
+    /// once that was understood.
+    ///
+    /// Call this periodically for v1 printers specifically - v2 has no
+    /// `pushing` category at all (checked: `schemas/bambulab/v2/
+    /// request.schema.json` only wires camera/info/print/system/upgrade/
+    /// xcam). Whether v2 needs some *other* trigger to get
+    /// `print.push_status` flowing at all is still an open question (a
+    /// live test meant to check that lost its connection entirely, an
+    /// unrelated failure, not real evidence either way) - if it does,
+    /// that's a different method to add once identified, not a variant
+    /// of this one. A v2 connection makes this call a no-op rather than
+    /// an error, since `pushing.pushall` specifically is confirmed not to
+    /// exist for it regardless.
+    pub async fn pushall(&self) -> Result<(), PrinterError> {
+        let (session, version) =
+            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
+
+        if matches!(version, BambuProtocolVersion::V2) {
+            return Ok(());
+        }
+
+        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
+            *version,
+            "pushing".into(),
+            "pushall".into(),
+        ))
+        .map_err(|err| PrinterError(format!("failed to build pushall request: {err}")))?;
+        body["push_target"] = serde_json::Value::from(1);
+
+        session.publish("pushing", &body).await
+    }
+
+    /// The current in-memory `print.push_status` object - `None` until at
+    /// least one has arrived (which, on v1, requires having called
+    /// `pushall` at least once since connecting; v2 pushes on its own).
+    pub async fn print_status(&self) -> Option<super::PrintReport> {
+        match self.connection.as_ref() {
+            Some((session, _)) => session.print_status().await,
+            None => None,
+        }
+    }
+}
+
+impl GenericPrinter for BambuPrinter {
+    fn base(&self) -> &PrinterBase {
+        &self.base
+    }
+
+    /// Real, not a stub: opens the MQTT session (see `bambu_mqtt.rs` for
+    /// `preference`'s `Auto`-tries-v5-then-v3.1.1 fallback) and keeps it -
+    /// same reasoning as `PrusaSerialPrinter::connect` holding its port
+    /// open, just an MQTT session instead of a serial port.
+    ///
+    /// `ensure_trusted` runs first so a newly-added printer with no
+    /// `ca_cert_path` configured just works on the first real connect
+    /// attempt: bundled CA if it verifies, auto-pin via trust-on-first-
+    /// connect if it doesn't (P1P, at least) - the same thing
+    /// `examples/auto_connect_bambu.rs` already did by hand, now
+    /// automatic. A printer with an explicit `ca_cert_path` (already
+    /// resolved to `Custom` in `fleet.rs`) is untouched by this - see
+    /// `ensure_trusted`'s own doc comment for why it never auto-pins over
+    /// an operator's explicit choice.
+    ///
+    /// `ensure_trusted` is a blocking call (a raw, synchronous TCP+TLS
+    /// handshake) run here inside an `async fn` - tolerable for a couple
+    /// of printers connecting concurrently on tokio's multi-thread
+    /// runtime, but at real farm scale this should move to
+    /// `tokio::task::spawn_blocking` so a slow/unreachable printer's
+    /// trust check can't tie up a worker thread other printers' connects
+    /// need.
+    async fn connect(&mut self) -> Result<(), PrinterError> {
+        self.ensure_trusted(Path::new(CERT_DIR))
+            .map_err(|err| PrinterError(format!("{} certificate trust failed: {err}", self.base.name)))?;
+
+        let (session, version) = super::bambu_mqtt::BambuMqttSession::open(
+            &self.base.host,
+            &self.access_code,
+            &self.sn,
+            &self.tls,
+            self.preference,
+        )
+        .await?;
+
+        println!("[{}] connected over MQTT ({version:?})", self.base.name);
+        self.connection = Some((session, version));
+        Ok(())
+    }
+
+    /// Real: a raw pass-through publish to `device/{sn}/request`, no
+    /// category envelope opinion - `send_gcode_line` below builds and
+    /// sends its own properly-wrapped request instead of going through
+    /// this.
+    async fn publish(&mut self, payload: String) -> Result<(), PrinterError> {
+        let (session, _) =
+            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
+        session.publish_raw(payload).await
+    }
+
+    /// Same MQTT gcode-injection mechanism regardless of protocol
+    /// generation - just routed through `send_gcode_line` below instead of
+    /// its own separate stub, since that's genuinely the same operation.
+    async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
         let pwm = super::percent_to_pwm(percent);
-        println!("[{}] sending gcode: M106 P1 S{pwm}", self.base.name);
+        self.send_gcode_line(format!("M106 P1 S{pwm}")).await
+    }
+
+    /// Real version FTPS-`PUT`s `data` to the printer's own `/cache/`
+    /// directory (that's storage the printer's firmware owns, same idea
+    /// as PrusaLink's SD card), then publishes a `project_file` MQTT
+    /// command referencing it by name so the printer actually starts
+    /// printing it - two separate steps, both stubbed here as one line
+    /// for now.
+    async fn upload_project_file(
+        &mut self,
+        file_name: String,
+        data: Vec<u8>,
+    ) -> Result<(), PrinterError> {
+        println!(
+            "[{}] FTPS upload {file_name} ({} bytes) to /cache/",
+            self.base.name,
+            data.len()
+        );
         Ok(())
     }
 
-    /// Shared "connect" logic both V1 and V2 delegate to - the real
-    /// version would open the MQTTS session here. Still a stub, but no
-    /// longer duplicated across two structs.
-    async fn connect_impl(&mut self) -> Result<(), PrinterError> {
-        let mode = match &self.tls {
-            BambuTls::BundledCa => "bundled Bambu CA",
-            BambuTls::Custom(_) => "custom CA cert",
-            BambuTls::Insecure => "no certificate verification",
-        };
-        println!("[{}] connecting over MQTTS:8883 ({mode})", self.base.name);
-        Ok(())
-    }
-}
-
-/// Older Bambu MQTT report schema (P1P, P1S, original X1 firmware).
-/// No V1-only fields yet - add them here as you learn what differs.
-pub struct BambuV1Printer {
-    generic: BambuGenericPrinter,
-}
-
-impl BambuV1Printer {
-    pub fn new(id: &str, name: &str, host: &str, access_code: &str, sn: &str, tls: BambuTls) -> Self {
-        Self { generic: BambuGenericPrinter::new(id, name, host, access_code, sn, tls) }
-    }
-
-    pub fn dispatch_gcode(&self, file_name: &str) {
-        self.generic.dispatch_gcode(file_name)
-    }
-
-    pub fn test_tls_handshake(&self) -> anyhow::Result<()> {
-        self.generic.test_tls_handshake()
-    }
-
-    pub fn fetch_certificate(&self) -> anyhow::Result<Vec<u8>> {
-        self.generic.fetch_certificate()
-    }
-
-    pub fn ensure_trusted(&mut self, cert_dir: &Path) -> anyhow::Result<()> {
-        self.generic.ensure_trusted(cert_dir)
-    }
-}
-
-impl GenericPrinter for BambuV1Printer {
-    fn base(&self) -> &PrinterBase {
-        self.generic.base()
-    }
-
-    async fn connect(&mut self) -> Result<(), PrinterError> {
-        self.generic.connect_impl().await
-    }
-
-    async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
-        self.generic.set_fan_speed_impl(percent).await
-    }
-}
-
-/// Current-generation Bambu MQTT report schema (X1 Carbon/X1E, H2 series).
-/// No V2-only fields yet (e.g. AMS slot state) - add them here as you build
-/// out real report parsing.
-pub struct BambuV2Printer {
-    generic: BambuGenericPrinter,
-}
-
-impl BambuV2Printer {
-    pub fn new(id: &str, name: &str, host: &str, access_code: &str, sn: &str, tls: BambuTls) -> Self {
-        Self { generic: BambuGenericPrinter::new(id, name, host, access_code, sn, tls) }
-    }
-
-    pub fn dispatch_gcode(&self, file_name: &str) {
-        self.generic.dispatch_gcode(file_name)
-    }
-
-    pub fn test_tls_handshake(&self) -> anyhow::Result<()> {
-        self.generic.test_tls_handshake()
-    }
-
-    pub fn fetch_certificate(&self) -> anyhow::Result<Vec<u8>> {
-        self.generic.fetch_certificate()
-    }
-
-    pub fn ensure_trusted(&mut self, cert_dir: &Path) -> anyhow::Result<()> {
-        self.generic.ensure_trusted(cert_dir)
-    }
-}
-
-impl GenericPrinter for BambuV2Printer {
-    fn base(&self) -> &PrinterBase {
-        self.generic.base()
-    }
-
-    async fn connect(&mut self) -> Result<(), PrinterError> {
-        self.generic.connect_impl().await
-    }
-
-    async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
-        self.generic.set_fan_speed_impl(percent).await
+    /// Builds the real JSON payload straight from continuum-schemas'
+    /// `schemas/bambulab/{v1,v2}/request/print/gcode_line.schema.json` -
+    /// which one depends on `self.version`, settled by `connect()` above -
+    /// instead of a hand-written serde struct: `sequence_id` and the
+    /// `command` const get filled in by the schema itself, `param` is the
+    /// one field we actually set. Real, end to end: publishes over the
+    /// live MQTT session `connect()` opened, no waiting for a reply -
+    /// `gcode_line` is fire-and-forget, the printer just executes it.
+    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
+        let (session, version) =
+            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
+
+        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
+            *version,
+            "print".into(),
+            "gcode_line".into(),
+        ))
+        .map_err(|err| PrinterError(format!("failed to build gcode_line request: {err}")))?;
+        body["param"] = serde_json::Value::String(gcode_line);
+
+        session.publish("print", &body).await
     }
 }

src/printer/bambu_commands/mod.rs

@@ -49,6 +49,12 @@ pub struct CommandEnvelope {
 /// category can have more than one possible request shape - if that ever
 /// happens here, that variant's payload becomes its own small
 /// `#[serde(tag = "command")]` enum instead of a single struct.
+///
+/// The "print" category isn't a variant here at all: unlike `info`/
+/// `pushing`, it has many possible request shapes (`gcode_line`, `pause`,
+/// `resume`, ...) in continuum-schemas' own `oneOf`, so
+/// `bambu.rs`'s `send_gcode_line_impl` builds that one directly via
+/// `continuum-schema-instance` instead of adding it here.
 #[derive(serde::Serialize, Debug)]
 #[serde(rename_all = "snake_case")]
 pub enum BambuRequest {

src/printer/bambu_commands/print.rs

@@ -1,11 +1,29 @@
-//! The ongoing print-state push (root key "print"). Doesn't carry a
-//! `CommandEnvelope` - no `sequence_id`/`command` seen on this one in the
-//! real capture, and it may genuinely not have them (an async push, not a
-//! reply to a specific request) - revisit if a real capture proves
-//! otherwise.
+//! The ongoing print-state push (root key "print"). The one real capture
+//! this struct was originally built from didn't show `sequence_id`/
+//! `command` - turned out to be because that capture was itself a
+//! truncated example (it's also missing dozens of other fields the real
+//! schema defines), not because the message genuinely lacks them:
+//! `push_status.schema.json` composes the shared envelope schema like
+//! every other Bambu message, so it does carry `sequence_id` - which
+//! matters a lot, see `BambuMqttSession`'s doc comment on how it's used
+//! both to correlate a `pushall` reply and to detect a gap in the
+//! ongoing organic stream.
+use super::CommandEnvelope;
 
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 pub struct PrintReport {
+    #[serde(flatten)]
+    pub envelope: CommandEnvelope,
+    /// BambuMQTT v1 only: `1` on an automatic/organic push (a partial
+    /// update - only the fields that changed), `0` on a forced one (a
+    /// `pushing.pushall`-triggered reply - the full object). This is the
+    /// actual, reliable signal for which kind a given message is -
+    /// confirmed against a real printer, unlike trying to infer it from
+    /// request/reply timing (see `BambuMqttSession`'s `PrintStatusCache`,
+    /// which used to try that and got it wrong twice). Not in
+    /// `push_status.schema.json` yet - add it there too when you're back
+    /// in continuum-schemas.
+    pub msg: Option<i32>,
     pub ams: Option<AmsState>,
     pub ams_exist_bits: Option<String>,
     pub insert_flag: Option<bool>,
@@ -16,19 +34,57 @@ pub struct PrintReport {
     pub xcam_status: Option<String>,
     // The real payload has a lot more fields than shown here (this
     // example was truncated) - add them the same way, one field per line,
-    // `Option<T>` for anything that isn't always present.
+    // `Option<T>` for anything that isn't always present. Remember to
+    // also add each new field to `merge_from` below.
+}
+
+impl PrintReport {
+    /// Merges `newer`'s fields into `self`, in place - BambuMQTT v1's
+    /// organic `push_status` pushes only carry the fields that changed,
+    /// so any field `newer` doesn't have (`None`) leaves `self`'s
+    /// existing value alone instead of clobbering it. Not used for v2
+    /// (always a full object) or for a `pushall`-triggered reply on
+    /// either version (also guaranteed full) - see `BambuMqttSession`.
+    pub fn merge_from(&mut self, newer: PrintReport) {
+        self.envelope = newer.envelope;
+        self.msg = newer.msg;
+        if newer.ams.is_some() {
+            self.ams = newer.ams;
+        }
+        if newer.ams_exist_bits.is_some() {
+            self.ams_exist_bits = newer.ams_exist_bits;
+        }
+        if newer.insert_flag.is_some() {
+            self.insert_flag = newer.insert_flag;
+        }
+        if newer.power_on_flag.is_some() {
+            self.power_on_flag = newer.power_on_flag;
+        }
+        if newer.tray_exist_bits.is_some() {
+            self.tray_exist_bits = newer.tray_exist_bits;
+        }
+        if newer.tray_is_bbl_bits.is_some() {
+            self.tray_is_bbl_bits = newer.tray_is_bbl_bits;
+        }
+        if newer.xcam.is_some() {
+            self.xcam = newer.xcam;
+        }
+        if newer.xcam_status.is_some() {
+            self.xcam_status = newer.xcam_status;
+        }
+    }
 }
 
 /// Bambu's own naming has the outer container and the inner array both
 /// called `ams` - `print.ams.ams`, not a typo here, just what the printer
 /// actually sends.
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 pub struct AmsState {
     pub ams: Vec<AmsUnit>,
 }
 
 /// One physical AMS unit (Bambu's numbering starts at 0 via `id`).
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 pub struct AmsUnit {
     pub humidity: String,
     pub id: String,
@@ -44,20 +100,20 @@ pub struct AmsUnit {
 /// every field `Empty` allows, plus more) would incorrectly match `Empty`
 /// first, since a plain struct doesn't reject extra fields by default.
 /// Verified this the hard way - see git history.
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 #[serde(untagged)]
 pub enum AmsTray {
     Empty(EmptyTray),
     Loaded(LoadedTray),
 }
 
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
 pub struct EmptyTray {
     pub id: String,
 }
 
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 pub struct LoadedTray {
     pub id: String,
     pub bed_temp: String,
@@ -82,7 +138,7 @@ pub struct LoadedTray {
 
 /// Camera-based AI print monitoring settings (spaghetti detection,
 /// first-layer inspection, etc.).
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 pub struct XcamSettings {
     pub allow_skip_parts: bool,
     pub buildplate_marker_detector: bool,

src/printer/bambu_mqtt.rs

@@ -0,0 +1,470 @@
+//! The real MQTT wiring for Bambu printers - the piece `connect`/
+//! `send_gcode_line` in `bambu.rs` used to just stub out with a
+//! `println!`. Three things make this more than "call `publish`, get a
+//! `String` back":
+//!
+//! 1. MQTT v3.1.1 (BambuMQTT v1) and MQTT v5 (BambuMQTT v2) are two
+//!    entirely separate rumqttc client types (`rumqttc::AsyncClient` vs
+//!    `rumqttc::v5::AsyncClient`) with different event/error types -
+//!    `BambuMqttClient` is the enum that hides which one a connection
+//!    ended up using, same pattern as `PrinterHandle` hides which vendor
+//!    struct you're holding.
+//! 2. A response to a published request doesn't come back from that same
+//!    call - it arrives later, asynchronously, on a *subscription* to
+//!    `device/{sn}/report`. That needs a background task continuously
+//!    polling the connection (rumqttc doesn't do this for you - nothing
+//!    happens until you poll the `EventLoop` yourself, repeatedly,
+//!    forever) - this is the first real use of `tokio::spawn` in this
+//!    crate, flagged as deferred since `printer/mod.rs`'s very first doc
+//!    comment, and this is why: it only makes sense once there's a
+//!    long-running background job that has to run independently of
+//!    whatever call site is waiting on a response.
+//!
+//!    Correlating a reply to its request isn't as uniform as it looks -
+//!    confirmed live, not assumed. `info.get_version`'s reply echoes back
+//!    the `sequence_id` you sent; `request` relies on that. A tempting
+//!    generalization - add your own extra field instead of `sequence_id`,
+//!    since most replies are supposed to echo back whatever extra fields
+//!    a request carried - turned out false for this exact command: a real
+//!    H2C's `get_version` reply had precisely `["command", "module",
+//!    "sequence_id"]`, nothing else, so a homemade correlation field never
+//!    survived and every request timed out. `sequence_id` - proven to
+//!    round-trip, not just assumed to - is what's actually used.
+//!    `print.push_status` doesn't echo `sequence_id` either (the printer
+//!    stamps that with its own free-running counter regardless of what
+//!    triggered it) - but it was never a request/reply case to begin with
+//!    (`pushing.pushall` triggers an ongoing *stream*, not a single
+//!    reply) - see `PrintStatusCache` for how that's handled instead.
+//! 3. `print.push_status` also needs to be kept as running in-memory
+//!    state, not just handed to whoever happens to be waiting at the
+//!    moment it arrives - see `PrintStatusCache` below for why (short
+//!    version: v1 only ever sends partial updates, so something has to
+//!    remember what the fields *were* to produce a coherent whole
+//!    object).
+//!
+//! Unverified against a live printer (no way to force it in a test): the
+//! exact `ConnectReturnCode`/error shape a real Bambu printer's embedded
+//! broker gives when it doesn't understand MQTT v5's CONNECT at all
+//! (rather than a clean version-rejection CONNACK) - `open()` below
+//! treats *any* failure of the v5 attempt as "fall back to v3.1.1", not
+//! just a precisely-matched version-rejection code, since a broker that
+//! can't parse a v5 CONNECT packet may just drop the connection instead
+//! of replying with one.
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use rumqttc::v5::mqttbytes::v5::Packet as PacketV5;
+use rumqttc::v5::mqttbytes::QoS as QoSV5;
+use rumqttc::{Event, Packet, QoS, Transport, TlsConfiguration};
+use serde_json::Value;
+use tokio::sync::{oneshot, Mutex};
+
+use super::bambu::BambuTls;
+use super::bambu_commands::PrintReport;
+use super::BambuProtocolVersion;
+use super::PrinterError;
+
+/// Requests waiting on a report with a matching `sequence_id` - only
+/// valid for a command empirically confirmed to echo it back, like
+/// `get_version` (see this module's doc comment - a homemade correlation
+/// field was tried instead and confirmed *not* to survive on this
+/// command's reply). Shared between whichever call registered one (in
+/// `request`) and the background poll task that completes it.
+type PendingTable = Arc<Mutex<HashMap<String, oneshot::Sender<Value>>>>;
+
+const REQUEST_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
+/// (25625 bytes - it reports a lot of modules: dual AMS, AMS HT, six WTM
+/// units, ...) blew straight through that limit and killed the whole
+/// connection outright (a real MQTT v5 protocol error, not a timeout).
+/// `print.push_status` reports (full AMS/xcam state) can get large too.
+/// 1MB has real headroom for that.
+const MAX_PACKET_SIZE: usize = 1024 * 1024;
+
+/// Every `sequence_id` this crate generates gets this prefix - a printer
+/// runs its own independent counter for unsolicited messages (confirmed
+/// live: `print.push_status`'s was already in the 500s at the start of a
+/// session, climbing steadily), and that counter has only ever been
+/// observed as a plain unprefixed integer string. An earlier version of
+/// this just started our own counter from a large offset instead of
+/// prefixing it, to dodge the same range a printer's counter climbs
+/// through - confirmed live (with an ISO-timestamp-prefixed sequence_id)
+/// that `sequence_id` isn't required to be numeric at all, so a
+/// non-numeric prefix rules out collision outright instead of merely
+/// making it improbable, and the counter itself can go back to starting
+/// at 1 each session.
+const SEQUENCE_ID_PREFIX: &str = "cxproxy-";
+
+enum BambuMqttClient {
+    V1(rumqttc::AsyncClient),
+    V2(rumqttc::v5::AsyncClient),
+}
+
+/// The running in-memory `print.push_status` object, plus enough state to
+/// know how to fold the next update into it:
+///
+/// - BambuMQTT v2 sends the *entire* object every time - always replace.
+/// - BambuMQTT v1 only sends the fields that changed since last time -
+///   merge into the existing object, or the picture goes incomplete the
+///   instant anything not present in the latest push gets treated as
+///   "unset". The exception is a `pushing.pushall`-triggered push, which
+///   is always the full object - and v1 self-describes which kind a given
+///   message is via its own `msg` field (`1` automatic/partial, `0`
+///   forced/full), confirmed against a real printer. `push_status` also
+///   doesn't echo `sequence_id` (see this module's doc comment), so it
+///   couldn't be identified via `request`'s correlation even if the
+///   timing worked out; `msg` needs none of that: it's on the message
+///   itself, correct regardless of which client (or nothing at all)
+///   triggered it, and regardless of timing.
+/// - `last_sequence_id` exists to notice when v1's organic stream skipped
+///   an update (network hiccup, a message this client wasn't fast enough
+///   to read): if the incoming `sequence_id` isn't exactly one more than
+///   the last one seen, the merged object is missing whatever changed in
+///   the gap. There's no way to know what that baseline number even is
+///   ahead of time - the printer runs its own counter, not one this
+///   client started - so the very first `push_status` received after
+///   connecting just seeds `last_sequence_id` rather than being checked
+///   against anything.
+struct PrintStatusCache {
+    report: Option<PrintReport>,
+    last_sequence_id: Option<u64>,
+}
+
+impl PrintStatusCache {
+    fn new() -> Self {
+        Self { report: None, last_sequence_id: None }
+    }
+
+    fn apply(&mut self, incoming: PrintReport, version: BambuProtocolVersion) {
+        let incoming_seq = incoming.envelope.sequence_id.parse::<u64>().ok();
+        // A forced/full update isn't missing anything by definition, so a
+        // sequence gap right after one (or on v2, where every message is
+        // full) isn't a real problem worth warning about.
+        let is_full_object = matches!(version, BambuProtocolVersion::V2) || incoming.msg != Some(1);
+
+        if let (Some(last), Some(incoming_seq), false) = (self.last_sequence_id, incoming_seq, is_full_object) {
+            if incoming_seq != last.wrapping_add(1) {
+                eprintln!(
+                    "print.push_status sequence gap: expected {}, got {incoming_seq} - the in-memory \
+                     object may be missing whatever changed in between",
+                    last.wrapping_add(1)
+                );
+            }
+        }
+        if let Some(incoming_seq) = incoming_seq {
+            self.last_sequence_id = Some(incoming_seq);
+        }
+
+        match (&mut self.report, is_full_object) {
+            (_, true) => self.report = Some(incoming),
+            (Some(existing), false) => existing.merge_from(incoming),
+            (None, false) => self.report = Some(incoming),
+        }
+    }
+}
+
+/// One live MQTT connection to a Bambu printer - whichever protocol
+/// version `open()` settled on.
+pub struct BambuMqttSession {
+    client: BambuMqttClient,
+    pending: PendingTable,
+    print_status: Arc<Mutex<PrintStatusCache>>,
+    sn: String,
+    next_sequence_id: AtomicU64,
+}
+
+impl BambuMqttSession {
+    /// `ForceV1`/`ForceV2` connect with that protocol directly. `Auto`
+    /// tries v5 first and falls back to v3.1.1 if that attempt fails -
+    /// see this module's doc comment for why "any failure", not just a
+    /// precisely-matched version-rejection code.
+    pub async fn open(
+        host: &str,
+        access_code: &str,
+        sn: &str,
+        tls: &BambuTls,
+        preference: super::bambu::BambuProtocolPreference,
+    ) -> Result<(Self, BambuProtocolVersion), PrinterError> {
+        use super::bambu::BambuProtocolPreference as Pref;
+
+        match preference {
+            Pref::ForceV1 => {
+                let session = Self::open_v1(host, access_code, sn, tls).await?;
+                Ok((session, BambuProtocolVersion::V1))
+            }
+            Pref::ForceV2 => {
+                let session = Self::open_v2(host, access_code, sn, tls).await?;
+                Ok((session, BambuProtocolVersion::V2))
+            }
+            Pref::Auto => match Self::open_v2(host, access_code, sn, tls).await {
+                Ok(session) => Ok((session, BambuProtocolVersion::V2)),
+                Err(_) => {
+                    let session = Self::open_v1(host, access_code, sn, tls).await?;
+                    Ok((session, BambuProtocolVersion::V1))
+                }
+            },
+        }
+    }
+
+    async fn open_v1(host: &str, access_code: &str, sn: &str, tls: &BambuTls) -> Result<Self, PrinterError> {
+        let connector = tls.build_connector().map_err(|err| PrinterError(format!("TLS setup failed: {err}")))?;
+
+        let mut options = rumqttc::MqttOptions::new(format!("continuum-{sn}"), host, 8883);
+        options.set_credentials("bblp", access_code);
+        options.set_transport(Transport::tls_with_config(TlsConfiguration::NativeConnector(connector)));
+        options.set_keep_alive(Duration::from_secs(30));
+        options.set_max_packet_size(MAX_PACKET_SIZE, MAX_PACKET_SIZE);
+
+        let (client, mut event_loop) = rumqttc::AsyncClient::new(options, 10);
+        client
+            .subscribe(format!("device/{sn}/report"), QoS::AtMostOnce)
+            .await
+            .map_err(|err| PrinterError(format!("subscribe failed: {err}")))?;
+
+        // Nothing actually happens on the wire until the eventloop is
+        // polled - this first poll is what makes the TCP+TLS+CONNECT
+        // handshake happen, and its result is the only way to know
+        // whether it worked.
+        match event_loop.poll().await {
+            Ok(_) => {}
+            Err(err) => return Err(PrinterError(format!("MQTT v3.1.1 connect failed: {err}"))),
+        }
+
+        let pending: PendingTable = Arc::new(Mutex::new(HashMap::new()));
+        let print_status = Arc::new(Mutex::new(PrintStatusCache::new()));
+        let sn_owned = sn.to_string();
+        tokio::spawn(poll_loop_v1(event_loop, pending.clone(), print_status.clone(), BambuProtocolVersion::V1, sn_owned));
+
+        Ok(Self {
+            client: BambuMqttClient::V1(client),
+            pending,
+            print_status,
+            sn: sn.to_string(),
+            next_sequence_id: AtomicU64::new(1),
+        })
+    }
+
+    async fn open_v2(host: &str, access_code: &str, sn: &str, tls: &BambuTls) -> Result<Self, PrinterError> {
+        let connector = tls.build_connector().map_err(|err| PrinterError(format!("TLS setup failed: {err}")))?;
+
+        let mut options = rumqttc::v5::MqttOptions::new(format!("continuum-{sn}"), host, 8883);
+        options.set_credentials("bblp", access_code);
+        options.set_transport(Transport::tls_with_config(TlsConfiguration::NativeConnector(connector)));
+        options.set_keep_alive(Duration::from_secs(30));
+        options.set_max_packet_size(Some(MAX_PACKET_SIZE as u32));
+
+        let (client, mut event_loop) = rumqttc::v5::AsyncClient::new(options, 10);
+        client
+            .subscribe(format!("device/{sn}/report"), QoSV5::AtMostOnce)
+            .await
+            .map_err(|err| PrinterError(format!("subscribe failed: {err}")))?;
+
+        match event_loop.poll().await {
+            Ok(_) => {}
+            Err(err) => return Err(PrinterError(format!("MQTT v5 connect failed: {err}"))),
+        }
+
+        let pending: PendingTable = Arc::new(Mutex::new(HashMap::new()));
+        let print_status = Arc::new(Mutex::new(PrintStatusCache::new()));
+        let sn_owned = sn.to_string();
+        tokio::spawn(poll_loop_v2(event_loop, pending.clone(), print_status.clone(), BambuProtocolVersion::V2, sn_owned));
+
+        Ok(Self {
+            client: BambuMqttClient::V2(client),
+            pending,
+            print_status,
+            sn: sn.to_string(),
+            next_sequence_id: AtomicU64::new(1),
+        })
+    }
+
+    /// Publishes `body` under `device/{sn}/request`, wrapped as
+    /// `{category: body}` (Bambu's top-level key names the *category*,
+    /// e.g. "print"/"info" - `body`'s own `command` field says which
+    /// specific one), and waits up to `REQUEST_TIMEOUT` for a report
+    /// whose `sequence_id` matches the one this call sent.
+    ///
+    /// This crate tried adding its own correlation field
+    /// (`continuum_correlation_id`) instead of relying on `sequence_id`,
+    /// on the understanding that extra fields survive into most replies.
+    /// Confirmed live that this doesn't hold for `get_version` specifically
+    /// - a real H2C's reply had exactly `["command", "module",
+    /// "sequence_id"]`, nothing extra at all, so the correlation field
+    /// never came back and every `get_version` call just timed out.
+    /// `sequence_id` is what's actually proven to round-trip for it
+    /// (confirmed earlier, live, on both protocol versions), so that's
+    /// what this uses. Whether a custom field survives on some *other*
+    /// command (`calibration`, say) is unverified, not confirmed false -
+    /// this was only ever checked for `get_version`.
+    ///
+    /// Not for `print.push_status` - that's not a single reply, see
+    /// `PrintStatusCache`. Not for a command with no reply at all either,
+    /// like `gcode_line` - use `publish` (fire-and-forget) for those.
+    pub async fn request(&self, category: &str, mut body: Value) -> Result<Value, PrinterError> {
+        let sequence_id = format!(
+            "{SEQUENCE_ID_PREFIX}{}",
+            self.next_sequence_id.fetch_add(1, Ordering::Relaxed)
+        );
+        body["sequence_id"] = Value::String(sequence_id.clone());
+
+        let (tx, rx) = oneshot::channel();
+        self.pending.lock().await.insert(sequence_id.clone(), tx);
+
+        if let Err(err) = self.publish(category, &body).await {
+            self.pending.lock().await.remove(&sequence_id);
+            return Err(err);
+        }
+
+        match tokio::time::timeout(REQUEST_TIMEOUT, rx).await {
+            Ok(Ok(response)) => Ok(response),
+            Ok(Err(_)) => Err(PrinterError(format!("{} closed before replying", self.sn))),
+            Err(_) => {
+                self.pending.lock().await.remove(&sequence_id);
+                Err(PrinterError(format!("{} did not reply within {REQUEST_TIMEOUT:?}", self.sn)))
+            }
+        }
+    }
+
+    /// Publishes `body` under `device/{sn}/request`, wrapped as
+    /// `{category: body}`, without waiting for a reply - for commands
+    /// like `gcode_line` and `pushing.pushall` that don't get a direct
+    /// correlated reply (an implicit ack the printer just executes, or an
+    /// ongoing stream instead of a single answer, respectively).
+    pub async fn publish(&self, category: &str, body: &Value) -> Result<(), PrinterError> {
+        self.publish_raw(serde_json::json!({ category: body }).to_string()).await
+    }
+
+    /// Publishes an already-serialized payload under `device/{sn}/request`
+    /// as-is, with no category wrapping - the low-level primitive both
+    /// `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> {
+        let topic = format!("device/{}/request", self.sn);
+
+        match &self.client {
+            BambuMqttClient::V1(client) => client
+                .publish(topic, QoS::AtLeastOnce, false, payload)
+                .await
+                .map_err(|err| PrinterError(format!("publish failed: {err}"))),
+            BambuMqttClient::V2(client) => client
+                .publish(topic, QoSV5::AtLeastOnce, false, payload)
+                .await
+                .map_err(|err| PrinterError(format!("publish failed: {err}"))),
+        }
+    }
+
+    /// The current in-memory `print.push_status` object, if at least one
+    /// has arrived since connecting - `None` until then. Cloned out from
+    /// behind the lock rather than handing out a guard, so callers aren't
+    /// holding the poll loop's own lock while they use it.
+    pub async fn print_status(&self) -> Option<PrintReport> {
+        self.print_status.lock().await.report.clone()
+    }
+}
+
+/// Runs forever, polling the v3.1.1 eventloop and routing incoming
+/// reports (pending-request correlation, `print.push_status` caching).
+/// Exits - loudly, not silently like before (that silence is exactly
+/// what made a real dropped connection to a live H2C look identical to
+/// "the printer just isn't sending anything") - once the connection is
+/// gone for good. Dropping the `BambuMqttSession` drops its `AsyncClient`,
+/// which is what eventually makes that happen during normal shutdown;
+/// this also fires on an unexpected disconnect, which is the case this
+/// logging is actually for. No reconnect logic yet - that's the next
+/// thing to add here, not something to fake by pretending the loop
+/// didn't end.
+async fn poll_loop_v1(
+    mut event_loop: rumqttc::EventLoop,
+    pending: PendingTable,
+    print_status: Arc<Mutex<PrintStatusCache>>,
+    version: BambuProtocolVersion,
+    sn: String,
+) {
+    loop {
+        match event_loop.poll().await {
+            Ok(Event::Incoming(Packet::Publish(publish))) => {
+                handle_incoming_report(&pending, &print_status, version, &publish.payload).await;
+            }
+            Ok(_) => {}
+            Err(err) => {
+                eprintln!("[{sn}] MQTT v3.1.1 connection lost, no longer receiving reports: {err}");
+                return;
+            }
+        }
+    }
+}
+
+async fn poll_loop_v2(
+    mut event_loop: rumqttc::v5::EventLoop,
+    pending: PendingTable,
+    print_status: Arc<Mutex<PrintStatusCache>>,
+    version: BambuProtocolVersion,
+    sn: String,
+) {
+    loop {
+        match event_loop.poll().await {
+            Ok(rumqttc::v5::Event::Incoming(PacketV5::Publish(publish))) => {
+                handle_incoming_report(&pending, &print_status, version, &publish.payload).await;
+            }
+            Ok(_) => {}
+            Err(err) => {
+                eprintln!("[{sn}] MQTT v5 connection lost, no longer receiving reports: {err}");
+                return;
+            }
+        }
+    }
+}
+
+/// Parses an incoming report, unwraps its single category key (a report
+/// is `{category: {"sequence_id": "...", ...}}`, same shape a request is
+/// published as - `request` publishes `body` and wants `body`-shaped data
+/// back, not the wrapper around it), and:
+///
+/// 1. If its `sequence_id` matches something `request` is waiting on,
+///    hands it the unwrapped body - but doesn't wake that waiter yet
+///    (see 3 for why).
+/// 2. If the unwrapped body is a `print.push_status`, folds it into the
+///    running `PrintStatusCache` (which reads the message's own `msg`
+///    field to decide full-vs-merge, not anything to do with step 1 - see
+///    `PrintStatusCache::apply`'s doc comment for why not).
+/// 3. Only now - once the cache is guaranteed up to date - actually sends
+///    on whichever oneshot matched in step 1, waking `request`. The order
+///    of 2 and 3 is load-bearing, caught live on an earlier version of
+///    this mechanism: sending on the oneshot first let a waiter's task
+///    resume and check `print_status()` before this function had gotten
+///    around to updating it. Nothing here awaits between the cache write
+///    and the send, so that race stays closed.
+///
+/// A report matching nothing pending isn't an error - most Bambu reports
+/// (including every organic `push_status` push) aren't a reply to
+/// anything specific.
+async fn handle_incoming_report(
+    pending: &PendingTable,
+    print_status: &Arc<Mutex<PrintStatusCache>>,
+    version: BambuProtocolVersion,
+    payload: &[u8],
+) {
+    let Ok(value) = serde_json::from_slice::<Value>(payload) else { return };
+    let Some(object) = value.as_object() else { return };
+    let Some((category, body)) = object.iter().next() else { return };
+
+    let request_sender = match body.get("sequence_id").and_then(Value::as_str) {
+        Some(sequence_id) => pending.lock().await.remove(sequence_id),
+        None => None,
+    };
+
+    if category == "print" && body.get("command").and_then(Value::as_str) == Some("push_status") {
+        if let Ok(report) = serde_json::from_value::<PrintReport>(body.clone()) {
+            print_status.lock().await.apply(report, version);
+        }
+    }
+
+    if let Some(sender) = request_sender {
+        let _ = sender.send(body.clone());
+    }
+}

src/printer/klipper.rs

@@ -29,6 +29,11 @@ impl GenericPrinter for KlipperPrinter {
         Ok(())
     }
 
+    async fn publish(&mut self, payload: String) -> Result<(), PrinterError> {
+        // TODO: add publishing via HTTP
+        Ok(())
+    }
+
     /// OVERRIDE: Moonraker's own **hypothetical** native API, not gcode injection - no
     /// percent-to-PWM conversion needed since it already takes a
     /// percentage directly. Real version:
@@ -37,4 +42,20 @@ impl GenericPrinter for KlipperPrinter {
         println!("[{}] Moonraker: SET_FAN_SPEED SPEED={:.2}", self.base.name, percent as f32 / 100.0);
         Ok(())
     }
+
+    /// OVERRIDE: real version is a multipart `POST /server/files/upload`
+    /// with `data` as the file part - Moonraker stores it on the printer's
+    /// virtual SD card, same upload-then-start model as Bambu/PrusaLink.
+    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());
+        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);
+        Ok(())
+    }
 }

src/printer/marlin_serial.rs

@@ -0,0 +1,69 @@
+//! A held-open serial connection to Marlin/RepRap-family firmware - usable
+//! by any printer connected over a raw serial port, since there's no
+//! printer-side storage to bulk-upload to the way there is for Bambu/
+//! PrusaLink/Klipper (see `prusa.rs`'s `PrusaSerialPrinter`).
+//!
+//! Marlin only ever buffers a handful of commands at once, so lines have
+//! to be paced by its own "ok" acknowledgment - send faster than it acks
+//! and you overrun that buffer, you don't just queue up harmlessly. That
+//! write-a-line/wait-for-"ok" pattern is the one new idea here. Splitting
+//! the connection into a read half and a write half so both can be used at
+//! once isn't new - `uplink/mod.rs` already does the same thing
+//! (`ws_stream.split()`) for the WebSocket to `continuum-backend`.
+//!
+//! The connection has to stay open across many `send_line` calls, not
+//! reopen per line: opening/closing per line would be wasteful, and on
+//! Arduino-based boards toggling the serial line's DTR signal (which
+//! opening the port does) resets the printer.
+
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines, ReadHalf, WriteHalf};
+use tokio_serial::{SerialPortBuilderExt, SerialStream};
+
+use super::PrinterError;
+
+pub struct MarlinSerialConnection {
+    write_half: WriteHalf<SerialStream>,
+    lines_in: Lines<BufReader<ReadHalf<SerialStream>>>,
+}
+
+impl MarlinSerialConnection {
+    /// Opens `port` (a serial device path - e.g. `/dev/ttyACM0` on Linux) at
+    /// `baud_rate`.
+    pub async fn open(port: &str, baud_rate: u32) -> Result<Self, PrinterError> {
+        let stream = tokio_serial::new(port, baud_rate)
+            .open_native_async()
+            .map_err(|err| PrinterError(format!("failed to open {port}: {err}")))?;
+
+        // Same shape as `ws_stream.split()` in uplink/mod.rs: one
+        // connection, used for writing gcode out and reading
+        // acknowledgments back at the same time.
+        let (read_half, write_half) = tokio::io::split(stream);
+        let lines_in = BufReader::new(read_half).lines();
+
+        Ok(Self { write_half, lines_in })
+    }
+
+    /// Sends one gcode line and waits for Marlin's "ok" before returning.
+    pub async fn send_line(&mut self, line: &str) -> Result<(), PrinterError> {
+        self.write_half
+            .write_all(format!("{line}\n").as_bytes())
+            .await
+            .map_err(|err| PrinterError(format!("write failed: {err}")))?;
+
+        // Marlin also sends unsolicited lines between commands
+        // (temperature reports, "echo:" messages) - keep reading until the
+        // real "ok" shows up instead of treating the first response as it.
+        loop {
+            let response = self
+                .lines_in
+                .next_line()
+                .await
+                .map_err(|err| PrinterError(format!("read failed: {err}")))?
+                .ok_or_else(|| PrinterError("serial connection closed".into()))?;
+
+            if response.starts_with("ok") {
+                return Ok(());
+            }
+        }
+    }
+}

src/printer/mod.rs

@@ -2,12 +2,10 @@
 //! `GenericPrinter -> BambuV1/BambuV2/PrusaLink/PrusaSerial/Klipper`-style
 //! inheritance, since Rust structs can't `extend` each other.
 //!
-//! Bambu goes one level deeper than the others: `bambu.rs` has
-//! `BambuGenericPrinter` (fields + behavior every Bambu printer shares -
-//! access code, CA trust, the TLS handshake test/fetch) which
-//! `BambuV1Printer`/`BambuV2Printer` each *have* one of, mirroring the
-//! `GenericPrinter -> BambuGenericPrinter -> BambuV1/V2` chain this project
-//! actually wanted from the start - see that file's doc comment.
+//! Bambu is a single `BambuPrinter` type, not a per-version split: which
+//! MQTT/BambuMQTT protocol generation (v1 vs v2) a given printer uses is
+//! something `connect()` discovers (or is told to force), not something
+//! config picks a *type* for - see `bambu.rs`'s doc comment.
 //!
 //! Three pieces, each doing one job an OOP base class would normally do:
 //!
@@ -23,17 +21,19 @@
 //!
 //! Run `cargo run --example printer_polymorphism` to see it in action.
 //!
-//! Deliberately left out of this module for now, to keep it focused on one
-//! idea at a time: real MQTT/HTTP/FTPS calls (that's `src/adapters/`) and
-//! `Send`-bound futures (matters once this is wired into a multi-threaded
-//! `tokio::spawn`, not yet).
+//! Real MQTT is built now - `bambu.rs`/`bambu_mqtt.rs` for Bambu's
+//! protocol, `marlin_serial.rs` for a real Marlin serial connection - see
+//! each module's own doc comment. PrusaLink/Klipper's HTTP clients are
+//! still stubs; that's the next piece to build past this module.
 
 mod bambu;
 mod bambu_commands;
+mod bambu_mqtt;
 mod klipper;
+mod marlin_serial;
 mod prusa;
 
-pub use bambu::{BambuGenericPrinter, BambuTls, BambuV1Printer, BambuV2Printer};
+pub use bambu::{BambuPrinter, BambuProtocolPreference, BambuProtocolVersion, BambuTls};
 pub use bambu_commands::{
     AmsState, AmsTray, AmsUnit, BambuReport, BambuRequest, CommandEnvelope, EmptyTray, GetVersionRequest,
     GetVersionResponse, LoadedTray, PrintReport, PushallRequest, XcamSettings,
@@ -72,12 +72,33 @@ pub trait GenericPrinter {
     /// into the multi-threaded daemon - safe to ignore for now.
     async fn connect(&mut self) -> Result<(), PrinterError>;
 
+
+    async fn publish(&mut self, payload: String) -> Result<(), PrinterError>;
+
     /// REQUIRED, no default body: `percent` is 0-100, but how it actually
     /// gets sent - an MQTT gcode-injection command, a REST call, raw bytes
     /// on a serial port - is different for every vendor, same as
     /// `connect()`.
     async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError>;
 
+    /// REQUIRED, no default body: takes the *whole file* already in memory
+    /// (`data`), not a path - whatever called this already had to download
+    /// it from the S3 URL the cloud handed over, so there'd be nothing on
+    /// disk to point a path at anyway. Each vendor does something quite
+    /// different with `data`: Bambu/PrusaLink/Klipper each do one bulk
+    /// transfer to storage the printer's own firmware owns; PrusaSerial
+    /// has no such storage at all - see that impl's doc comment.
+    async fn upload_project_file(&mut self, file_name: String, data: Vec<u8>) -> Result<(), PrinterError>;
+
+    /// REQUIRED, no default body: sends exactly *one* gcode line, using
+    /// whatever single-command mechanism this vendor has (an MQTT
+    /// `gcode_line` publish for Bambu, an HTTP command-injection endpoint
+    /// for PrusaLink/Klipper, a raw write over the port for PrusaSerial).
+    /// `upload_gcode` below is the generic driver that calls this once per
+    /// line for a whole file - this method is where each vendor's actual
+    /// difference lives, same division of labor as every other method here.
+    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError>;
+
     /// OPTIONAL: a default body every printer gets for free unless it
     /// writes its own. None of ours do below, so all three vendors share
     /// this exact implementation - the trait-method equivalent of
@@ -100,8 +121,7 @@ fn percent_to_pwm(percent: u8) -> u8 {
 
 /// The closed set of printer kinds this fleet can talk to.
 pub enum PrinterHandle {
-    BambuV1(BambuV1Printer),
-    BambuV2(BambuV2Printer),
+    Bambu(BambuPrinter),
     PrusaLink(PrusaLinkPrinter),
     PrusaSerial(PrusaSerialPrinter),
     Klipper(KlipperPrinter),
@@ -114,8 +134,7 @@ pub enum PrinterHandle {
 impl GenericPrinter for PrinterHandle {
     fn base(&self) -> &PrinterBase {
         match self {
-            PrinterHandle::BambuV1(p) => p.base(),
-            PrinterHandle::BambuV2(p) => p.base(),
+            PrinterHandle::Bambu(p) => p.base(),
             PrinterHandle::PrusaLink(p) => p.base(),
             PrinterHandle::PrusaSerial(p) => p.base(),
             PrinterHandle::Klipper(p) => p.base(),
@@ -124,21 +143,73 @@ impl GenericPrinter for PrinterHandle {
 
     async fn connect(&mut self) -> Result<(), PrinterError> {
         match self {
-            PrinterHandle::BambuV1(p) => p.connect().await,
-            PrinterHandle::BambuV2(p) => p.connect().await,
+            PrinterHandle::Bambu(p) => p.connect().await,
             PrinterHandle::PrusaLink(p) => p.connect().await,
             PrinterHandle::PrusaSerial(p) => p.connect().await,
             PrinterHandle::Klipper(p) => p.connect().await,
         }
     }
 
+    async fn publish(&mut self, payload: String) -> Result<(), PrinterError> {
+        match self {
+            PrinterHandle::Bambu(p) => p.publish(payload).await,
+            PrinterHandle::PrusaLink(p) => p.publish(payload).await,
+            PrinterHandle::PrusaSerial(p) => p.publish(payload).await,
+            PrinterHandle::Klipper(p) => p.publish(payload).await,
+        }
+    }
+
     async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
         match self {
-            PrinterHandle::BambuV1(p) => p.set_fan_speed(percent).await,
-            PrinterHandle::BambuV2(p) => p.set_fan_speed(percent).await,
+            PrinterHandle::Bambu(p) => p.set_fan_speed(percent).await,
             PrinterHandle::PrusaLink(p) => p.set_fan_speed(percent).await,
             PrinterHandle::PrusaSerial(p) => p.set_fan_speed(percent).await,
             PrinterHandle::Klipper(p) => p.set_fan_speed(percent).await,
         }
     }
+
+    async fn upload_project_file(&mut self, file_name: String, data: Vec<u8>) -> Result<(), PrinterError> {
+        match self {
+            PrinterHandle::Bambu(p) => p.upload_project_file(file_name, data).await,
+            PrinterHandle::PrusaLink(p) => p.upload_project_file(file_name, data).await,
+            PrinterHandle::PrusaSerial(p) => p.upload_project_file(file_name, data).await,
+            PrinterHandle::Klipper(p) => p.upload_project_file(file_name, data).await,
+        }
+    }
+
+    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
+        match self {
+            PrinterHandle::Bambu(p) => p.send_gcode_line(gcode_line).await,
+            PrinterHandle::PrusaLink(p) => p.send_gcode_line(gcode_line).await,
+            PrinterHandle::PrusaSerial(p) => p.send_gcode_line(gcode_line).await,
+            PrinterHandle::Klipper(p) => p.send_gcode_line(gcode_line).await,
+        }
+    }
+}
+
+/// Splits `gcode` into lines and sends each one through `printer`'s own
+/// `send_gcode_line` - works for *any* vendor, not just serial ones, since
+/// Bambu/PrusaLink/Klipper each have their own single-line injection
+/// mechanism too (see each `send_gcode_line` impl). They just don't use
+/// this for `upload_project_file` - they have real file storage to bulk-
+/// upload to instead, which is a different, faster path than sending a
+/// whole file one line at a time.
+///
+/// `P: GenericPrinter` makes this a genuine Rust generic: one function
+/// body, compiled separately for whichever concrete printer type you call
+/// it with - like a TS function typed `<P extends GenericPrinter>`, except
+/// Rust generates real, separate machine code per type instead of erasing
+/// the type at compile time. Call it with any of the five printer structs,
+/// or with `&mut PrinterHandle` itself (it implements `GenericPrinter` too,
+/// via the delegation above).
+pub async fn upload_gcode<P: GenericPrinter>(printer: &mut P, gcode: &str) -> Result<(), PrinterError> {
+    for line in gcode.lines() {
+        let line = line.trim();
+        // Blank lines and full-line comments have nothing to send.
+        if line.is_empty() || line.starts_with(';') {
+            continue;
+        }
+        printer.send_gcode_line(line.to_string()).await?;
+    }
+    Ok(())
 }

src/printer/prusa.rs

@@ -29,6 +29,11 @@ impl GenericPrinter for PrusaLinkPrinter {
         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.
@@ -37,6 +42,22 @@ impl GenericPrinter for PrusaLinkPrinter {
         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(())
+    }
 }
 
 
@@ -44,14 +65,22 @@ impl GenericPrinter for PrusaLinkPrinter {
 
 pub struct PrusaSerialPrinter {
     base: PrinterBase,
-    pub com_port: i16,
+    pub port: String,
+    pub baud_rate: u32,
+    /// `None` until `connect()` succeeds - a real serial connection, not a
+    /// stub, so there's something to hold onto: unlike Bambu/PrusaLink/
+    /// Klipper's stubbed sessions, this one has to survive across many
+    /// `send_gcode_line` calls (see `marlin_serial.rs` for why).
+    connection: Option<super::marlin_serial::MarlinSerialConnection>,
 }
 
 impl PrusaSerialPrinter {
-    pub fn new(id: &str, name: &str, com_port: i16) -> Self {
+    pub fn new(id: &str, name: &str, port: &str, baud_rate: u32) -> Self {
         Self {
             base: PrinterBase { id: id.into(), name: name.into(), host: "127.0.0.1".to_string() },
-            com_port,
+            port: port.into(),
+            baud_rate,
+            connection: None,
         }
     }
 }
@@ -61,21 +90,51 @@ impl GenericPrinter for PrusaSerialPrinter {
         &self.base
     }
 
-    /// OVERRIDE: a fourth, again-different body - no network host at all,
-    /// just a local serial port (real version: open the port, e.g. via the
-    /// `serialport`/`tokio-serial` crate, and speak Prusa's G-code-over-
-    /// serial protocol).
+    /// OVERRIDE, and a real one (not a stub, unlike the other four vendors'
+    /// `connect`): opens the port and keeps it, since serial has no
+    /// separate "session" concept to stub out - the open connection *is*
+    /// the whole thing.
     async fn connect(&mut self) -> Result<(), PrinterError> {
-        println!("[{}] connecting over serial (COM{})", self.base.name, self.com_port);
+        println!("[{}] connecting over serial ({}, {} baud)", self.base.name, self.port, self.baud_rate);
+        self.connection = Some(super::marlin_serial::MarlinSerialConnection::open(&self.port, self.baud_rate).await?);
+        Ok(())
+    }
+
+    async fn publish(&mut self, payload: String) -> Result<(), PrinterError> {
+        // TODO: add publishing via serial
         Ok(())
     }
 
     /// OVERRIDE: same gcode as PrusaLink (still M106, still the shared
-    /// conversion), but written straight to the serial port instead of
-    /// sent over HTTP - a third transport for the same underlying command.
+    /// conversion), but sent through `send_gcode_line` below instead of a
+    /// separate transport - serial only has the one way to talk to the
+    /// printer at all.
     async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
         let pwm = super::percent_to_pwm(percent);
-        println!("[{}] writing to COM{}: M106 S{pwm}", self.base.name, self.com_port);
-        Ok(())
+        self.send_gcode_line(format!("M106 S{pwm}")).await
+    }
+
+    /// OVERRIDE, but not just "a fourth transport" like the others above -
+    /// there's no printer-side storage to upload `data` to over a raw
+    /// serial connection at all, so this decodes it as gcode text and
+    /// streams it through `send_gcode_line` one line at a time via the
+    /// generic `upload_gcode` driver, instead of one bulk write like
+    /// Bambu/PrusaLink/Klipper.
+    async fn upload_project_file(&mut self, file_name: String, data: Vec<u8>) -> Result<(), PrinterError> {
+        let gcode = String::from_utf8(data)
+            .map_err(|_| PrinterError(format!("{file_name} is not valid UTF-8 gcode")))?;
+        println!("[{}] streaming {file_name} over {}", self.base.name, self.port);
+        super::upload_gcode(self, &gcode).await
+    }
+
+    /// OVERRIDE, and the real implementation, not a stub: writes `gcode_line`
+    /// to the already-open port and waits for Marlin's "ok" - see
+    /// `marlin_serial.rs` for why that wait matters.
+    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
+        let connection = self
+            .connection
+            .as_mut()
+            .ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
+        connection.send_line(&gcode_line).await
     }
 }