fix: updated stale docs

This reverts commit 910dbcd2e6.

src/slic3r/CMakeLists.txt

@@ -734,8 +734,12 @@ set(SLIC3R_GUI_SOURCES
     Utils/ICameraSignalingChannel.hpp
     Utils/OrcaCloudServiceAgent.cpp
     Utils/OrcaCloudServiceAgent.hpp
+    Utils/OrcaMqttConnection.cpp
+    Utils/OrcaMqttConnection.hpp
     Utils/OrcaPrinterAgent.cpp
     Utils/OrcaPrinterAgent.hpp
+    Utils/OrcaCloudSignalingChannel.cpp
+    Utils/OrcaCloudSignalingChannel.hpp
     Utils/QidiPrinterAgent.cpp
     Utils/QidiPrinterAgent.hpp
     Utils/SnapmakerPrinterAgent.cpp

src/slic3r/Utils/BBLPrinterAgent.cpp

@@ -1,8 +1,11 @@
 #include "BBLPrinterAgent.hpp"
 #include "BBLNetworkPlugin.hpp"
+#include "IPrinterAgent.hpp"
 #include "NetworkAgentFactory.hpp"
 #include "libslic3r/Utils.hpp"
+#include "NetworkAgent.hpp"
 
+#include <boost/format.hpp>
 #include <boost/log/trivial.hpp>
 #include <boost/nowide/fstream.hpp>
 #include <nlohmann/json.hpp>
@@ -10,6 +13,10 @@ using json = nlohmann::json;
 
 #include <type_traits>
 #include <unordered_map>
+#include <memory>
+#include <nlohmann/json.hpp>
+#include <cmath>
+#include <slic3r/GUI/DeviceManager.hpp>
 
 namespace Slic3r {
 
@@ -133,7 +140,7 @@ BBLPrinterAgent::~BBLPrinterAgent() = default;
 
 void BBLPrinterAgent::set_cloud_agent(std::shared_ptr<ICloudServiceAgent> cloud)
 {
-    m_cloud_agent = cloud;
+    (void) cloud;
     // BBL DLL manages tokens internally, so this is just for interface compliance
 }
 
@@ -141,6 +148,163 @@ void BBLPrinterAgent::set_cloud_agent(std::shared_ptr<ICloudServiceAgent> cloud)
 // Communication
 // ============================================================================
 
+std::string BBLPrinterAgent::ams_refresh_rfid_gcode(const std::string& tray_id)
+{
+    return (boost::format("M620 R%1% \n") % tray_id).str();
+}
+
+std::string BBLPrinterAgent::ams_calibrate_gcode(int ams_id)
+{
+    return (boost::format("M620 C%1% \n") % ams_id).str();
+}
+
+std::string BBLPrinterAgent::ams_select_tray_gcode(const std::string& tray_id)
+{
+    return (boost::format("M620 P%1% \n") % tray_id).str();
+}
+
+int BBLPrinterAgent::command_ams_refresh_rfid(std::string dev_id, std::string tray_id, int sequence_id, bool lan_mode)
+{
+    const std::string gcode = ams_refresh_rfid_gcode(tray_id);
+    BOOST_LOG_TRIVIAL(trace) << "ams_debug: gcode_cmd" << gcode;
+    nlohmann::json j;
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = gcode;
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_ams_calibrate(std::string dev_id, int ams_id, int sequence_id, bool lan_mode)
+{
+    const std::string gcode = ams_calibrate_gcode(ams_id);
+    BOOST_LOG_TRIVIAL(trace) << "ams_debug: gcode_cmd" << gcode;
+    nlohmann::json j;
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = gcode;
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_ams_select_tray(std::string dev_id, std::string tray_id, int sequence_id, bool lan_mode)
+{
+    const std::string gcode = ams_select_tray_gcode(tray_id);
+    BOOST_LOG_TRIVIAL(trace) << "ams_debug: gcode_cmd" << gcode;
+    nlohmann::json j;
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = gcode;
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_xyz_abs(std::string dev_id, int sequence_id, bool lan_mode)
+{
+    nlohmann::json j;
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = "G90 \n";
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_auto_leveling(std::string dev_id, int sequence_id, bool lan_mode)
+{
+    nlohmann::json j;
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = "G29 \n";
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_go_home(std::string dev_id, bool is_printing, bool supports_mqtt_homing, int sequence_id, bool lan_mode)
+{
+    nlohmann::json j;
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    if (supports_mqtt_homing) {
+        j["print"]["command"] = "back_to_center";
+        return publish(dev_id, j, lan_mode);
+    }
+
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = is_printing ? "G28 X\n" : "G28 \n";
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_set_bed(std::string dev_id, int temp, bool supports_mqtt_bed_ctrl, int sequence_id, bool lan_mode)
+{
+    nlohmann::json j;
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    if (supports_mqtt_bed_ctrl) {
+        j["print"]["command"] = "set_bed_temp";
+        j["print"]["temp"] = temp;
+        return publish(dev_id, j, lan_mode);
+    }
+
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = (boost::format("M140 S%1%\n") % temp).str();
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_set_nozzle(std::string dev_id, int temp, int sequence_id, bool lan_mode)
+{
+    nlohmann::json j;
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = (boost::format("M104 S%1%\n") % temp).str();
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::command_axis_control(std::string dev_id, std::string axis, double unit, double input_val, int speed,
+                                           bool is_core_xy, bool supports_mqtt_axis_control, int sequence_id, bool lan_mode)
+{
+    nlohmann::json j;
+    j["print"]["sequence_id"] = std::to_string(sequence_id);
+
+    if (supports_mqtt_axis_control) {
+        int dir = input_val > 0 ? 1 : -1;
+        // i3-arch printers move the bed for Y/Z, so the on-screen direction is
+        // reversed -- same negation the g-code fallback below applies.
+        if (!is_core_xy && (axis == "Y" || axis == "Z")) {
+            dir = -dir;
+        }
+
+        j["print"]["command"] = "xyz_ctrl";
+        j["print"]["axis"] = axis;
+        j["print"]["dir"] = dir;
+        j["print"]["mode"] = (std::abs(input_val) >= 10) ? 1 : 0;
+        return publish(dev_id, j, lan_mode);
+    }
+
+    double value = input_val;
+    if (!is_core_xy && (axis == "Y" || axis == "Z")) {
+        value = -1.0 * input_val;
+    }
+
+    std::string value_str = (boost::format("%.1f") % (value * unit)).str();
+    std::string gcode;
+    if (axis == "X" || axis == "Y" || axis == "Z") {
+        gcode = (boost::format("M211 S \nM211 X1 Y1 Z1\nM1002 push_ref_mode\nG91 \nG1 %1%%2% F%3%\nM1002 pop_ref_mode\nM211 R\n")
+                 % axis % value_str % speed).str();
+    } else if (axis == "E") {
+        gcode = (boost::format("M83 \nG0 %1%%2% F%3%\n") % axis % value_str % speed).str();
+    } else {
+        return -1;
+    }
+
+    j["print"]["command"] = "gcode_line";
+    j["print"]["param"] = gcode;
+    return publish(dev_id, j, lan_mode);
+}
+
+int BBLPrinterAgent::publish(const std::string& dev_id, const nlohmann::json& j, bool lan_mode)
+{
+    const int rtn = lan_mode ? send_message_to_printer(dev_id, j.dump(), 0, 0) : send_message(dev_id, j.dump(), 0, 0);
+    if (rtn == 0) {
+        BOOST_LOG_TRIVIAL(info) << "publish_json: " << j.dump() << " code: " << rtn;
+    } else {
+        BOOST_LOG_TRIVIAL(error) << "publish_json: " << j.dump() << " code: " << rtn;
+    }
+    return rtn;
+}
+
 int BBLPrinterAgent::send_message(std::string dev_id, std::string json_str, int qos, int flag)
 {
     json_str = from_orca_payload(std::move(json_str));
@@ -473,8 +637,15 @@ int BBLPrinterAgent::start_local_print_with_record(PrintParams params, OnUpdateS
 
 int BBLPrinterAgent::start_send_gcode_to_sdcard(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn, OnWaitFn wait_fn)
 {
-    return dispatch_start<func_start_send_gcode_to_sdcard_legacy, func_start_send_gcode_to_sdcard_0203>(
+    int result = dispatch_start<func_start_send_gcode_to_sdcard_legacy, func_start_send_gcode_to_sdcard_0203>(
         BBLNetworkPlugin::instance().get_start_send_gcode_to_sdcard(), params, update_fn, cancel_fn, wait_fn);
+    if (result != 0) {
+        BOOST_LOG_TRIVIAL(error) << "start_send_gcode_to_sdcard failed: result=" << result
+            << ", try_emmc_print=" << params.try_emmc_print
+            << ", legacy_mode=" << BBLNetworkPlugin::instance().use_legacy_network()
+            << ", dev_ip=" << params.dev_ip << ", dev_id=" << params.dev_id;
+    }
+    return result;
 }
 
 int BBLPrinterAgent::start_local_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn)

src/slic3r/Utils/BBLPrinterAgent.hpp

@@ -5,6 +5,7 @@
 #include "ICloudServiceAgent.hpp"
 #include <string>
 #include <memory>
+#include <nlohmann/json.hpp>
 
 namespace Slic3r {
 
@@ -28,9 +29,23 @@ public:
 
     // Communication
     int send_message(std::string dev_id, std::string json_str, int qos, int flag) override;
+    static std::string ams_refresh_rfid_gcode(const std::string& tray_id);
+    static std::string ams_calibrate_gcode(int ams_id);
+    static std::string ams_select_tray_gcode(const std::string& tray_id);
+    int command_ams_refresh_rfid(std::string dev_id, std::string tray_id, int sequence_id, bool lan_mode) override;
+    int command_ams_calibrate(std::string dev_id, int ams_id, int sequence_id, bool lan_mode) override;
+    int command_ams_select_tray(std::string dev_id, std::string tray_id, int sequence_id, bool lan_mode) override;
+    int command_xyz_abs(std::string dev_id, int sequence_id, bool lan_mode) override;
+    int command_auto_leveling(std::string dev_id, int sequence_id, bool lan_mode) override;
+    int command_go_home(std::string dev_id, bool is_printing, bool supports_mqtt_homing, int sequence_id, bool lan_mode) override;
+    int command_set_bed(std::string dev_id, int temp, bool supports_mqtt_bed_ctrl, int sequence_id, bool lan_mode) override;
+    int command_set_nozzle(std::string dev_id, int temp, int sequence_id, bool lan_mode) override;
+    int command_axis_control(std::string dev_id, std::string axis, double unit, double input_val, int speed,
+                              bool is_core_xy, bool supports_mqtt_axis_control, int sequence_id, bool lan_mode) override;
     int connect_printer(std::string dev_id, std::string dev_ip, std::string username, std::string password, bool use_ssl) override;
     int disconnect_printer() override;
     int send_message_to_printer(std::string dev_id, std::string json_str, int qos, int flag) override;
+    std::string default_lan_username() const override { return "bblp"; }
 
     // Certificates
     int check_cert() override;
@@ -99,7 +114,8 @@ public:
     static std::string from_orca_payload(std::string json_text);
 
 private:
-    std::shared_ptr<ICloudServiceAgent> m_cloud_agent;
+    // why: the lan/cloud DECISION stays machine-side; keep this mechanical branch in sync with publish_json.
+    int publish(const std::string& dev_id, const nlohmann::json& j, bool lan_mode);
 };
 
 } // namespace Slic3r

src/slic3r/Utils/CrealityPrintAgent.cpp

@@ -282,8 +282,11 @@ bool CrealityPrintAgent::parse_cfs_response(const std::string&    response,
     return true;
 }
 
-bool CrealityPrintAgent::fetch_filament_info(std::string dev_id, FilamentSyncMode /*sync_mode*/)
+bool CrealityPrintAgent::fetch_filament_info(std::string dev_id, FilamentSyncMode sync_mode)
 {
+    if (sync_mode != get_filament_sync_mode())
+        return false;
+
     if (device_info.dev_ip.empty()) {
         BOOST_LOG_TRIVIAL(warning)
             << "CrealityPrintAgent::fetch_filament_info: no device IP, falling back to base agent";

src/slic3r/Utils/CrealityPrintAgent.hpp

@@ -1,6 +1,7 @@
 #ifndef __CREALITY_PRINT_AGENT_HPP__
 #define __CREALITY_PRINT_AGENT_HPP__
 
+#include "IPrinterAgent.hpp"
 #include "MoonrakerPrinterAgent.hpp"
 
 #include <string>

src/slic3r/Utils/MoonrakerPrinterAgent.hpp

@@ -9,11 +9,16 @@
 #include <set>
 #include <string>
 #include <thread>
+#include <condition_variable>
+#include <deque>
+#include <functional>
 
 #include <nlohmann/json.hpp>
 
 namespace Slic3r {
 
+bool moonraker_is_light_name(const std::string& name);
+
 class MoonrakerPrinterAgent : public IPrinterAgent
 {
 public:
@@ -68,10 +73,9 @@ public:
     int set_on_local_connect_fn(OnLocalConnectedFn fn) override;
     int set_on_local_message_fn(OnMessageFn fn) override;
     int set_queue_on_main_fn(QueueOnMainFn fn) override;
-
-    // Pull-mode agent (on-demand filament sync)
-    FilamentSyncMode get_filament_sync_mode() const override { return FilamentSyncMode::pull; }
     bool fetch_filament_info(std::string dev_id, FilamentSyncMode sync_mode = FilamentSyncMode::pull) override;
+    CameraStreamMode get_camera_stream_mode() const override;
+    std::string get_camera_url() const override;
 
 protected:
     struct MoonrakerDeviceInfo
@@ -85,6 +89,7 @@ protected:
         std::string dev_name;
         std::string version;
         std::string klippy_state;
+        float       nozzle_diameter = 0.0f;
         bool        use_ssl = false;
     } device_info;
 
@@ -105,10 +110,17 @@ protected:
     // Methods that derived classes may need to override or access
     virtual bool init_device_info(std::string dev_id, std::string dev_ip, std::string username, std::string password, bool use_ssl);
     virtual bool fetch_device_info(const std::string& base_url, const std::string& api_key, MoonrakerDeviceInfo& info, std::string& error) const;
+    static float parse_nozzle_diameter(const nlohmann::json& response);
 
     // State access for derived classes
     mutable std::recursive_mutex       state_mutex;
 
+    // Counts detached fetch_filament_info() background threads currently touching `this`
+    // (see QidiPrinterAgent::fetch_filament_info). Those threads hold a raw `this` with no
+    // other lifetime protection, so the destructor waits for this to reach 0 before any part
+    // of the object is torn down — see ~MoonrakerPrinterAgent().
+    std::atomic<int> filament_fetch_in_flight{0};
+
     // Helpers
     bool        is_numeric(const std::string& value);
     std::string normalize_base_url(std::string host, const std::string& port);
@@ -121,6 +133,18 @@ protected:
     // Map filament type to OrcaFilamentLibrary preset ID for AMS sync compatibility
     static std::string map_filament_type_to_generic_id(const std::string& filament_type);
 
+    // Send a G-code script via Moonraker (/printer/gcode/script)
+    bool send_gcode(const std::string& dev_id, const std::string& gcode) const;
+    bool send_gcode(const std::string& dev_id, const std::string& gcode,
+                    const std::string& base_url, const std::string& api_key) const;
+    bool post_print_action(const std::string& action) const;
+    bool post_print_action(const std::string& action,
+                           const std::string& base_url, const std::string& api_key) const;
+
+    bool send_ws_rpc(const std::string& method, const nlohmann::json& params);
+
+    virtual void on_status_loop_tick(const std::string& dev_id) {}
+
 private:
     int handle_request(const std::string& dev_id, const std::string& json_str);
     int send_version_info(const std::string& dev_id);
@@ -128,7 +152,6 @@ private:
 
     bool fetch_object_list(const std::string& base_url, const std::string& api_key, std::set<std::string>& objects, std::string& error) const;
     bool query_printer_status(const std::string& base_url, const std::string& api_key, nlohmann::json& status, std::string& error) const;
-    bool send_gcode(const std::string& dev_id, const std::string& gcode) const;
 
     void announce_printhost_device();
     void dispatch_local_connect(int state, const std::string& dev_id, const std::string& msg);
@@ -137,7 +160,8 @@ private:
     void start_status_stream(const std::string& dev_id, const std::string& base_url, const std::string& api_key);
     void stop_status_stream();
     void run_status_stream(std::string dev_id, std::string base_url, std::string api_key);
-    void handle_ws_message(const std::string& dev_id, const std::string& payload);
+    void handle_ws_message(std::string dev_id, std::string payload, std::string base_url, std::string api_key);
+    void refresh_thumbnail_url(std::string base_url, std::string api_key);
     void update_status_cache(const nlohmann::json& updates);
     nlohmann::json build_print_payload_locked() const;
 
@@ -151,9 +175,10 @@ private:
                       const std::string& base_url, const std::string& api_key,
                       OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn);
 
-    // JSON-RPC helper
+    // Start a print of a previously uploaded G-code file (path relative to the
-    bool send_jsonrpc_command(const std::string& base_url, const std::string& api_key,
+    // Moonraker gcodes root).
-                              const nlohmann::json& request, std::string& response) const;
+    bool start_print_file(const std::string& base_url, const std::string& api_key,
+                          const std::string& filename, std::string& error_msg) const;
 
     // Connection thread management
     void perform_connection_async(const std::string& dev_id,
@@ -161,6 +186,12 @@ private:
                                    const std::string& api_key,
                                    uint64_t generation);
 
+    // why: a printer with no /server/webcams/list entry can still name its stream directly;
+    // subclasses (e.g. printers with a fixed webcam path) can override this instead.
+    virtual std::string webcam_stream_override(const std::string& base_url) const { return {}; }
+    void refresh_webcam_info() const;
+    bool fetch_webcam_info(const std::string& base_url, const std::string& api_key, uint64_t generation) const;
+
     // System-specific filament fetch methods
     bool fetch_hh_filament_info(std::vector<AmsTrayData>& trays, int& max_lane_index);
     bool fetch_moonraker_filament_data(std::vector<AmsTrayData>& trays, int& max_lane_index);
@@ -189,15 +220,38 @@ private:
 
     mutable std::recursive_mutex payload_mutex;
     nlohmann::json     status_cache;
+    // note: guarded by payload_mutex; filled by refresh_thumbnail_url(), empty url = looked up, none found
+    std::string        thumbnail_filename;
+    std::string        thumbnail_url;
+    mutable std::string        webcam_stream_url;
+    mutable CameraStreamMode   webcam_stream_mode = CameraStreamMode::none;
+    mutable uint64_t            webcam_info_last_lookup_ms = 0;
+    mutable uint64_t            webcam_info_generation = 0;
+    unsigned            thumbnail_lookup_attempts = 0;
+
+    static constexpr uint64_t WEBCAM_INFO_REFRESH_INTERVAL_MS = 1000;
 
     std::atomic<int>       next_jsonrpc_id{1};
     std::set<std::string>  available_objects;  // Track for feature detection
+    bool                   assumed_light_on = false;
 
     std::atomic<bool>   ws_stop{false};
     std::atomic<bool>   ws_reconnect_requested{false};  // Flag to trigger reconnection
     std::atomic<uint64_t> ws_last_emit_ms{0};
     std::thread         ws_thread;
 
+    // stop_status_stream() invokes ws_abort_io to wake a blocked synchronous
+    // ws.read()/ws.write()/handshake in run_status_stream(): ws_stop is only
+    // observed between reads, and Beast's expires_after() does not bound
+    // synchronous operations.
+    std::mutex             ws_abort_mutex;
+    std::function<void()>  ws_abort_io;  // guarded by ws_abort_mutex
+
+    // AMS/filament refresh cadence, independent of telemetry dispatch so a steady
+    // stream of status updates can't starve it (ws_last_emit_ms is reset by those).
+    static constexpr uint64_t AMS_REFRESH_INTERVAL_MS = 10000;
+    std::atomic<uint64_t> ams_last_fetch_ms{0};
+
     // Throttling configuration for WebSocket updates
     // Critical changes (state transitions) dispatch immediately; telemetry is throttled
     static constexpr uint64_t STATUS_UPDATE_INTERVAL_MS = 1000;  // 1 update/sec for telemetry
@@ -207,7 +261,15 @@ private:
     // Connection thread management
     std::atomic<uint64_t>  connect_generation{0};
     std::thread            connect_thread;
-    std::recursive_mutex   connect_mutex;
+    mutable std::recursive_mutex connect_mutex;
+
+    void enqueue_command(std::function<void()> fn);
+    void run_command_worker();
+    std::thread cmd_thread;
+    std::deque<std::function<void()>> cmd_queue;
+    std::mutex cmd_mutex;
+    std::condition_variable cmd_cv;
+    bool cmd_stop = false;
 };
 
 } // namespace Slic3r

src/slic3r/Utils/OrcaCloudServiceAgent.cpp

@@ -6,6 +6,8 @@
 
 #include <boost/asio.hpp>
 #include <boost/beast/core/detail/base64.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/websocket.hpp>
 #include <boost/filesystem.hpp>
 #include <boost/log/trivial.hpp>
 #include <boost/uuid/uuid.hpp>
@@ -21,14 +23,18 @@
 #include <openssl/hmac.h>
 #include <openssl/rand.h>
 #include <openssl/sha.h>
+#include <openssl/ssl.h>
 
 #include <algorithm>
 #include <cctype>
+#include <condition_variable>
+#include <cstdint>
 #include <cstdlib>
 #include <fstream>
 #include <iomanip>
 #include <optional>
 #include <random>
+#include <set>
 #include <sstream>
 
 #include <string>
@@ -492,6 +498,7 @@ OrcaCloudServiceAgent::OrcaCloudServiceAgent(std::string log_dir)
     , api_base_url(ORCA_DEFAULT_API_URL)
     , auth_base_url(ORCA_DEFAULT_AUTH_URL)
     , cloud_base_url(ORCA_DEFAULT_CLOUD_URL)
+    , mqtt_connection(std::make_unique<OrcaMqttConnection>())
 {
     auth_headers["apikey"]    = ORCA_DEFAULT_PUB_KEY;
     pkce_bundle.loopback_port = choose_loopback_port();
@@ -502,6 +509,8 @@ OrcaCloudServiceAgent::OrcaCloudServiceAgent(std::string log_dir)
 
 OrcaCloudServiceAgent::~OrcaCloudServiceAgent()
 {
+    if (mqtt_connection)
+        mqtt_connection->stop();
     if (refresh_thread.joinable()) {
         refresh_thread.join();
     }
@@ -942,22 +951,45 @@ bool OrcaCloudServiceAgent::ensure_token_fresh(const std::string& reason) { retu
 
 int OrcaCloudServiceAgent::connect_server()
 {
+    const bool logged_in = is_user_login();
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: OrcaCloudServiceAgent::connect_server logged_in=" << logged_in
+                            << " api_base_url=" << api_base_url;
+    if (!logged_in) {
+        if (mqtt_connection)
+            mqtt_connection->stop();
+        {
+            std::lock_guard<std::recursive_mutex> lock(state_mutex);
+            is_connected = false;
+        }
+        BOOST_LOG_TRIVIAL(warning) << "OrcaCloudServiceAgent: connect_server requires a logged-in user";
+        invoke_server_connected_callback(-1, 401);
+        return BAMBU_NETWORK_ERR_INVALID_HANDLE;
+    }
+
     std::string response;
     unsigned int http_code = 0;
     int result             = http_get(ORCA_HEALTH_PATH, &response, &http_code);
 
     bool connected = (result == BAMBU_NETWORK_SUCCESS && http_code >= 200 && http_code < 300);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: cloud health result=" << result << " http_code=" << http_code
+                            << " connected=" << connected << " response_bytes=" << response.size();
+
+    // connect_server() remains a REST health probe. The long-lived fleet MQTT socket
+    // is started lazily by set_user_selected_machine -> configure_selected_printer_mqtt;
+    // subscriptions queued before that point are replayed when it starts.
     {
         std::lock_guard<std::recursive_mutex> lock(state_mutex);
         is_connected = connected;
     }
-
     invoke_server_connected_callback(connected ? 0 : -1, http_code);
     return connected ? BAMBU_NETWORK_SUCCESS : BAMBU_NETWORK_ERR_CONNECTION_TO_SERVER_FAILED;
 }
 
 bool OrcaCloudServiceAgent::is_server_connected()
 {
+    // The REST health probe is the signal; the per-printer MQTT socket does not gate
+    // whole-cloud connectivity (one printer reconnecting must not report the whole
+    // cloud as lost).
     std::lock_guard<std::recursive_mutex> lock(state_mutex);
     return is_connected;
 }
@@ -978,13 +1010,232 @@ int OrcaCloudServiceAgent::stop_subscribe(std::string module)
 
 int OrcaCloudServiceAgent::add_subscribe(std::vector<std::string> dev_list)
 {
-    (void) dev_list;
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: OrcaCloudServiceAgent::add_subscribe count=" << dev_list.size()
-    return BAMBU_NETWORK_SUCCESS;
+                            << " logged_in=" << is_user_login() << " mqtt_connection=" << (mqtt_connection ? "set" : "null");
+    if (!is_user_login() || !mqtt_connection) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: add_subscribe rejected because cloud is not ready";
+        return BAMBU_NETWORK_ERR_INVALID_HANDLE;
+    }
+    bool queued = true;
+    for (const std::string& dev_id : dev_list)
+        queued = mqtt_connection->subscribe(dev_id) && queued;
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: add_subscribe queued=" << queued;
+    return queued ? BAMBU_NETWORK_SUCCESS : BAMBU_NETWORK_ERR_CONNECT_FAILED;
 }
 
 int OrcaCloudServiceAgent::del_subscribe(std::vector<std::string> dev_list)
 {
-    (void) dev_list;
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: OrcaCloudServiceAgent::del_subscribe count=" << dev_list.size()
+                            << " logged_in=" << is_user_login() << " mqtt_connection=" << (mqtt_connection ? "set" : "null");
+    if (!is_user_login() || !mqtt_connection) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: del_subscribe rejected because cloud is not ready";
+        return BAMBU_NETWORK_ERR_INVALID_HANDLE;
+    }
+    bool queued = true;
+    for (const std::string& dev_id : dev_list)
+        queued = mqtt_connection->unsubscribe(dev_id) && queued;
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: del_subscribe queued=" << queued;
+    return queued ? BAMBU_NETWORK_SUCCESS : BAMBU_NETWORK_ERR_CONNECT_FAILED;
+}
+
+int OrcaCloudServiceAgent::configure_selected_printer_mqtt(const std::string& dev_id,
+                                                           OrcaMqttConnection::StateHandler state_handler)
+{
+    (void) dev_id;
+    OrcaMqttConnection::Config cfg;
+    cfg.url               = "wss://" + api_base_url + "/api/v1/printers/mqtt";
+    cfg.use_tls           = true;
+    cfg.bearer_provider   = [this] { return get_access_token(); };
+    cfg.client_id         = "OrcaSlicer";
+    cfg.keepalive_seconds = 300;
+
+    {
+        std::lock_guard<std::mutex> lock(m_selected_url_mutex);
+        m_selected_printer_mqtt_url = cfg.url;
+    }
+
+    if (mqtt_connection->is_running()) {
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: fleet MQTT connection already running";
+        return BAMBU_NETWORK_SUCCESS;
+    }
+
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: configuring fleet MQTT endpoint=" << cfg.url;
+
+    // NOTE: no lock is held across start() — it blocks for the whole initial connect
+    // attempt (up to ~10s), and the message handler below re-enters callback_mutex on
+    // the MQTT worker thread.
+    const bool ok = mqtt_connection->start(
+        cfg,
+        [this](const std::string& id, const std::string& payload) { deliver_cloud_message(id, payload); },
+        std::move(state_handler));
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: fleet MQTT start returned=" << ok;
+    return ok ? BAMBU_NETWORK_SUCCESS : BAMBU_NETWORK_ERR_CONNECTION_TO_SERVER_FAILED;
+}
+
+void OrcaCloudServiceAgent::teardown_selected_printer_mqtt()
+{
+    if (mqtt_connection) {
+        mqtt_connection->stop();
+        // The connection object is reused for the next printer; drop this printer's
+        // report topic so its 1:1 socket does not re-subscribe the previous device.
+        mqtt_connection->clear_subscriptions();
+    }
+    std::lock_guard<std::mutex> lock(m_selected_url_mutex);
+    m_selected_printer_mqtt_url.clear();
+}
+
+std::string OrcaCloudServiceAgent::selected_printer_mqtt_url() const
+{
+    std::lock_guard<std::mutex> lock(m_selected_url_mutex);
+    return m_selected_printer_mqtt_url;
+}
+
+void OrcaCloudServiceAgent::deliver_cloud_message(const std::string& dev_id, const std::string& payload)
+{
+    OnMessageFn callback;
+    {
+        std::lock_guard<std::mutex> lock(callback_mutex);
+        callback = printer_status_callback;
+    }
+    if (callback)
+        callback(dev_id, payload);
+}
+
+int OrcaCloudServiceAgent::set_printer_status_callback(OnMessageFn fn)
+{
+    std::lock_guard<std::mutex> lock(callback_mutex);
+    printer_status_callback = std::move(fn);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: printer status callback=" << (printer_status_callback ? "set" : "clear");
+    return BAMBU_NETWORK_SUCCESS;
+}
+
+int OrcaCloudServiceAgent::send_printer_command(const std::string& dev_id, const std::string& body)
+{
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: send_printer_command dev_id=" << dev_id
+                            << " body_bytes=" << body.size() << " logged_in=" << is_user_login();
+    if (dev_id.empty() || !is_user_login()) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: send_printer_command rejected";
+        return BAMBU_NETWORK_ERR_INVALID_HANDLE;
+    }
+
+    const std::string path = std::string(ORCA_CLOUD_PRINTER) + "/" + dev_id + "/commands";
+    std::string  response;
+    unsigned int http_code = 0;
+    int          result    = http_post(path, body, &response, &http_code);
+    BOOST_LOG_TRIVIAL(info) << "OrcaCloudServiceAgent: command dev=" << dev_id
+                            << " http=" << http_code << " result=" << result
+                            << " response_bytes=" << response.size();
+    return (result == BAMBU_NETWORK_SUCCESS && http_code >= 200 && http_code < 300)
+               ? BAMBU_NETWORK_SUCCESS
+               : BAMBU_NETWORK_ERR_CONNECT_FAILED;
+}
+
+int OrcaCloudServiceAgent::upload_gcode_via_cloud(const std::string& dev_id,
+                                                  const std::string& local_gcode_path,
+                                                  std::string* job_id,
+                                                  OnUpdateStatusFn update_fn,
+                                                  WasCancelledFn cancel_fn)
+{
+    if (dev_id.empty() || local_gcode_path.empty() || !is_user_login())
+        return BAMBU_NETWORK_ERR_INVALID_HANDLE;
+    if (cancel_fn && cancel_fn())
+        return BAMBU_NETWORK_ERR_CANCELED;
+
+    // Step 1: POST print-jobs/uploads -> a short-lived presigned R2 PUT URL. No
+    // metadata rides this request; filename/start are only relevant to the HTTP
+    // .../start finalize route, which this MQTT-driven flow does not call.
+    const std::string uploads_path = std::string(ORCA_CLOUD_PRINTER) + "/" + Http::url_encode(dev_id) + "/print-jobs/uploads";
+    std::string response;
+    unsigned int http_code = 0;
+    int result = http_post(uploads_path, "{}", &response, &http_code);
+    if (result != BAMBU_NETWORK_SUCCESS || http_code < 200 || http_code >= 300) {
+        BOOST_LOG_TRIVIAL(warning) << "OrcaCloudServiceAgent: print-jobs/uploads failed http_code=" << http_code;
+        return BAMBU_NETWORK_ERR_CONNECT_FAILED;
+    }
+
+    std::string upload_job_id;
+    std::string upload_url;
+    try {
+        const nlohmann::json j = nlohmann::json::parse(response);
+        upload_job_id = j.value("job_id", "");
+        upload_url    = j.value("upload_url", "");
+    } catch (const std::exception& e) {
+        BOOST_LOG_TRIVIAL(error) << "OrcaCloudServiceAgent: failed to parse print-jobs/uploads response: " << e.what();
+        return BAMBU_NETWORK_ERR_CONNECT_FAILED;
+    }
+    if (upload_job_id.empty() || upload_url.empty()) {
+        BOOST_LOG_TRIVIAL(error) << "OrcaCloudServiceAgent: print-jobs/uploads response missing job_id/upload_url";
+        return BAMBU_NETWORK_ERR_CONNECT_FAILED;
+    }
+
+    if (cancel_fn && cancel_fn())
+        return BAMBU_NETWORK_ERR_CANCELED;
+
+    // Step 2: PUT the G-code straight to R2 with the one-time URL from step 1. This
+    // is a scoped, PUT-only, short-TTL capability with no bearer token of its own,
+    // so it bypasses http_put (which always prefixes api_base_url and attaches the
+    // cloud session's Authorization header - neither belongs on an R2 PUT).
+    bool canceled = false;
+    unsigned put_status = 0;
+    std::string put_error;
+    Http::put(upload_url)
+        .tls_verify(true)
+        .header("Content-Type", "text/x.gcode")
+        .set_put_body(boost::filesystem::path(local_gcode_path))
+        .timeout_connect(5)
+        .timeout_max(300) // large G-code over a slow link
+        .on_progress([&](Http::Progress progress, bool& cancel) {
+            if (cancel_fn && cancel_fn()) {
+                cancel   = true;
+                canceled = true;
+                return;
+            }
+            if (update_fn && progress.ultotal > 0) {
+                const int percent = static_cast<int>((progress.ulnow * 100) / progress.ultotal);
+                update_fn(PrintingStageUpload, percent, "Uploading...");
+            }
+        })
+        .on_complete([&](std::string, unsigned status) { put_status = status; })
+        .on_error([&](std::string, std::string err, unsigned status) {
+            put_status = status;
+            put_error  = std::move(err);
+        })
+        .perform_sync();
+
+    if (canceled)
+        return BAMBU_NETWORK_ERR_CANCELED;
+    if (put_status < 200 || put_status >= 300) {
+        BOOST_LOG_TRIVIAL(warning) << "OrcaCloudServiceAgent: R2 upload failed status=" << put_status << " error=" << put_error;
+        return BAMBU_NETWORK_ERR_PRINT_SG_UPLOAD_FTP_FAILED;
+    }
+
+    if (job_id)
+        *job_id = std::move(upload_job_id);
+    return BAMBU_NETWORK_SUCCESS;
+}
+
+int OrcaCloudServiceAgent::start_cloud_print_job(const std::string& dev_id,
+                                                 const std::string& job_id,
+                                                 const std::string& filename,
+                                                 bool start)
+{
+    if (dev_id.empty() || job_id.empty() || !is_user_login())
+        return BAMBU_NETWORK_ERR_INVALID_HANDLE;
+
+    nlohmann::json body;
+    if (!filename.empty())
+        body["filename"] = filename;
+    body["start"] = start;
+
+    const std::string path = std::string(ORCA_CLOUD_PRINTER) + "/" + Http::url_encode(dev_id) + "/print-jobs/" +
+                             Http::url_encode(job_id) + "/start";
+    std::string response;
+    unsigned int http_code = 0;
+    const int result = http_post(path, body.dump(), &response, &http_code);
+    if (result != BAMBU_NETWORK_SUCCESS || http_code < 200 || http_code >= 300) {
+        BOOST_LOG_TRIVIAL(warning) << "OrcaCloudServiceAgent: print-jobs/" << job_id << "/start failed http_code=" << http_code;
+        return BAMBU_NETWORK_ERR_CONNECT_FAILED;
+    }
     return BAMBU_NETWORK_SUCCESS;
 }
 
@@ -2025,6 +2276,10 @@ bool OrcaCloudServiceAgent::set_user_session(const json& session_json, bool noti
 
 void OrcaCloudServiceAgent::clear_session()
 {
+    if (mqtt_connection) {
+        mqtt_connection->stop();
+        mqtt_connection->clear_subscriptions();
+    }
     {
         std::lock_guard<std::mutex> lock(session_mutex);
         session = SessionInfo{};
@@ -2140,7 +2395,11 @@ int OrcaCloudServiceAgent::http_get(const std::string& path, std::string* respon
     return (res.success && !suppress) ? BAMBU_NETWORK_SUCCESS : BAMBU_NETWORK_ERR_CONNECT_FAILED;
 }
 
-int OrcaCloudServiceAgent::http_post(const std::string& path, const std::string& body, std::string* response_body, unsigned int* http_code)
+int OrcaCloudServiceAgent::http_post(const std::string& path,
+                                     const std::string& body,
+                                     std::string* response_body,
+                                     unsigned int* http_code,
+                                     const std::string& content_type)
 {
     std::string url = api_base_url + path;
     BOOST_LOG_TRIVIAL(trace) << "OrcaCloudServiceAgent: POST " << url;
@@ -2164,7 +2423,7 @@ int OrcaCloudServiceAgent::http_post(const std::string& path, const std::string&
                 http.header("Authorization", "Bearer " + token);
             }
 
-            http.header("Content-Type", "application/json");
+            http.header("Content-Type", content_type);
             http.set_post_body(body);
 
             http.on_complete([&](std::string resp_body, unsigned resp_status) {
@@ -2631,19 +2890,28 @@ int OrcaCloudServiceAgent::get_user_print_info(unsigned int* http_code, std::str
     if (http_code)
         *http_code = code;
 
-    if (result != 0 || code != 200)
+    if (result != 0 || code != 200) {
+        BOOST_LOG_TRIVIAL(error) << "OrcaCloudServiceAgent: get_user_print_info failed - http_code=" << code << ", response=" << response;
         return result != 0 ? result : BAMBU_NETWORK_ERR_GET_SETTING_LIST_FAILED;
+    }
+
+    BOOST_LOG_TRIVIAL(trace) << "OrcaCloudServiceAgent: get_user_print_info fetched - http_code=" << code << ", response=" << response;
 
     try {
         auto resp_json = nlohmann::json::parse(response);
         nlohmann::json devices = nlohmann::json::array();
 
         for (const auto& printer : resp_json.value("data", nlohmann::json::array())) {
-            const std::string role = printer.value("access_role", "");
+            nlohmann::json device;
+            std::string role = printer.value("access_role", "");
+
+            // A printer with the role "view" only has monitoring access for orca cloud.
+            // The printer is owned by a different person and was shared to the current user without
+            // any permission to control the printer so we discard this printer. Comment this out if 
+            // OrcaSlicer wants to support view only printers.
             if (role.empty() || role == "viewer")
                 continue;
 
-            nlohmann::json device;
             device["dev_id"]   = printer.value("id", "");
             device["dev_name"] = printer.value("name", "");
             if (printer.contains("model") && printer["model"].is_string())
@@ -2657,15 +2925,20 @@ int OrcaCloudServiceAgent::get_user_print_info(unsigned int* http_code, std::str
                     device["task_status"] = status["job"].value("state", "");
             }
             device["dev_online"] = online;
-            devices.push_back(std::move(device));
+
+            devices.push_back(device);
         }
 
         if (http_body) {
             nlohmann::json out;
-            out["devices"] = std::move(devices);
+            out["devices"] = devices;
             *http_body = out.dump();
         }
-    } catch (const std::exception&) {
+
+        BOOST_LOG_TRIVIAL(debug) << "OrcaCloudServiceAgent: get_user_print_info parsed - device_count=" << devices.size()
+                                  << ", devices=" << devices.dump();
+    } catch (const std::exception& e) {
+        BOOST_LOG_TRIVIAL(error) << "OrcaCloudServiceAgent: get_user_print_info parse exception - " << e.what();
         return BAMBU_NETWORK_ERR_GET_SETTING_LIST_FAILED;
     }
 

src/slic3r/Utils/OrcaCloudServiceAgent.hpp

@@ -2,6 +2,12 @@
 #define __ORCA_CLOUD_SERVICE_AGENT_HPP__
 
 #include "ICloudServiceAgent.hpp"
+
+#include <boost/asio.hpp>
+#include <boost/asio/ssl.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/ssl.hpp>
+#include <boost/beast/websocket.hpp>
 #include <cstdlib>
 #include <string>
 #include <map>
@@ -9,12 +15,17 @@
 #include <atomic>
 #include <chrono>
 #include <functional>
+#include <condition_variable>
+#include <cstdint>
+#include <set>
 #include <memory>
 #include <thread>
 #include <unordered_map>
 #include <vector>
 #include <nlohmann/json.hpp>
 
+#include "OrcaMqttConnection.hpp"
+
 class wxSecretStore;
 
 namespace Slic3r {
@@ -206,6 +217,19 @@ public:
     int add_subscribe(std::vector<std::string> dev_list) override;
     int del_subscribe(std::vector<std::string> dev_list) override;
     void enable_multi_machine(bool enable) override;
+    int set_printer_status_callback(OnMessageFn fn);
+    int send_printer_command(const std::string& dev_id, const std::string& body);
+
+    int upload_gcode_via_cloud(const std::string& dev_id,
+                               const std::string& local_gcode_path,
+                               std::string* job_id,
+                               OnUpdateStatusFn update_fn,
+                               WasCancelledFn cancel_fn);
+
+    int start_cloud_print_job(const std::string& dev_id,
+                              const std::string& job_id,
+                              const std::string& filename,
+                              bool start = true);
 
     // ========================================================================
     // ICloudServiceAgent Interface Implementation - Settings Synchronization
@@ -346,7 +370,29 @@ public:
 
     static std::string generate_uuid_for_setting_id(const std::string& name, const std::string& user_id = "");
 
+    OrcaMqttConnection* get_mqtt_connection() noexcept {
+        return mqtt_connection.get();
+    }
+
+    const OrcaMqttConnection* get_mqtt_connection() const noexcept {
+        return mqtt_connection.get();
+    }
+
+    // Account-scoped cloud socket: wss://<api_base_url>/api/v1/printers/mqtt.
+    // configure_ blocks for the duration of the initial connect attempt, so callers
+    // drive it off the UI thread; teardown_ is synchronous. The dev_id argument is
+    // retained for source compatibility with the printer-agent lifecycle; it does
+    // not participate in endpoint construction.
+    int         configure_selected_printer_mqtt(const std::string& dev_id,
+                                                OrcaMqttConnection::StateHandler state_handler = {});
+    void        teardown_selected_printer_mqtt();
+    // Test hook: the wss:// URL of the current fleet socket ("" when none).
+    std::string selected_printer_mqtt_url() const;
+
 private:
+    // Fans one inbound fleet MQTT message out to printer_status_callback.
+    void deliver_cloud_message(const std::string& dev_id, const std::string& payload);
+
     // Sync protocol helpers
     int sync_pull(
         std::function<void(const SyncPullResponse&)> on_success,
@@ -370,7 +416,11 @@ private:
 
     // HTTP request helpers
     int http_get(const std::string& path, std::string* response_body, unsigned int* http_code);
-    int http_post(const std::string& path, const std::string& body, std::string* response_body, unsigned int* http_code);
+    int http_post(const std::string& path,
+                  const std::string& body,
+                  std::string* response_body,
+                  unsigned int* http_code,
+                  const std::string& content_type = "application/json");
     int http_put(const std::string& path, const std::string& body, std::string* response_body, unsigned int* http_code);
     int http_delete(const std::string& path, std::string* response_body, unsigned int* http_code);
     std::map<std::string, std::string> data_headers();
@@ -423,6 +473,9 @@ private:
                                std::chrono::system_clock::now().time_since_epoch()).count()};
 
     // Member variables - connection state
+    std::unique_ptr<OrcaMqttConnection> mqtt_connection;
+    std::string        m_selected_printer_mqtt_url;   // guarded by m_selected_url_mutex
+    mutable std::mutex m_selected_url_mutex;
     bool is_connected{false};
     bool enable_track{false};
     bool multi_machine_enabled{false};
@@ -436,6 +489,7 @@ private:
     AppOnHttpErrorFn on_http_error_fn;
     GetCountryCodeFn get_country_code_fn;
     QueueOnMainFn queue_on_main_fn;
+    OnMessageFn printer_status_callback;
     mutable std::mutex callback_mutex;
 
     // Thread safety

src/slic3r/Utils/OrcaCloudSignalingChannel.cpp

@@ -0,0 +1,320 @@
+#include "OrcaCloudSignalingChannel.hpp"
+
+#include "Http.hpp"
+
+#include <boost/asio/connect.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/log/trivial.hpp>
+#include <nlohmann/json.hpp>
+
+#include <openssl/ssl.h>
+
+#include <cctype>
+#include <iomanip>
+#include <sstream>
+#include <stdexcept>
+
+namespace Slic3r {
+
+OrcaCloudSignalingChannel::OrcaCloudSignalingChannel(std::shared_ptr<ICloudServiceAgent> cloud, std::string dev_id)
+    : m_cloud(std::move(cloud))
+    , m_dev_id(std::move(dev_id))
+{
+}
+
+OrcaCloudSignalingChannel::~OrcaCloudSignalingChannel()
+{
+    close();
+}
+
+void OrcaCloudSignalingChannel::open()
+{
+    bool expected = false;
+    if (!m_open.compare_exchange_strong(expected, true))
+        return;
+    m_stop.store(false);
+    m_thread = std::thread([this] { run(); });
+}
+
+void OrcaCloudSignalingChannel::close()
+{
+    m_stop.store(true);
+    std::shared_ptr<Connection> conn;
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        conn = m_conn;
+    }
+    if (conn) {
+        // Established session: close the socket on the io_context's own thread so
+        // the pending async_read completes and io_context.run() unwinds.
+        boost::asio::post(conn->io_context, [conn] {
+            boost::system::error_code ec;
+            boost::beast::get_lowest_layer(conn->websocket).cancel(ec);
+            boost::beast::get_lowest_layer(conn->websocket).close(ec);
+        });
+        // Pre-run() phase (still in the synchronous connect/handshake): best-effort
+        // direct interruption.
+        boost::system::error_code ec;
+        boost::beast::get_lowest_layer(conn->websocket).cancel(ec);
+        boost::beast::get_lowest_layer(conn->websocket).close(ec);
+    }
+    if (m_thread.joinable())
+        m_thread.join();
+    m_open.store(false);
+}
+
+void OrcaCloudSignalingChannel::send_offer(std::string sdp)
+{
+    send_json(nlohmann::json{{"type", "webrtc.offer"}, {"sdp", std::move(sdp)}}.dump());
+}
+
+void OrcaCloudSignalingChannel::send_ice(std::string candidate, std::string mid)
+{
+    send_json(nlohmann::json{{"type", "webrtc.ice"},
+                             {"candidate", std::move(candidate)},
+                             {"sdpMid", std::move(mid)}}
+                  .dump());
+}
+
+std::string OrcaCloudSignalingChannel::encode_path_component(const std::string& value)
+{
+    std::ostringstream encoded;
+    encoded << std::uppercase << std::hex;
+    for (unsigned char c : value) {
+        if (std::isalnum(c) || c == '-' || c == '_' || c == '.' || c == '~')
+            encoded << c;
+        else
+            encoded << '%' << std::setw(2) << std::setfill('0') << static_cast<unsigned int>(c);
+    }
+    return encoded.str();
+}
+
+std::string OrcaCloudSignalingChannel::host_without_scheme(std::string value)
+{
+    const auto scheme = value.find("://");
+    if (scheme != std::string::npos)
+        value.erase(0, scheme + 3);
+    const auto slash = value.find('/');
+    if (slash != std::string::npos)
+        value.erase(slash);
+    return value;
+}
+
+void OrcaCloudSignalingChannel::unavailable(CameraUnavailableReason reason, std::string detail)
+{
+    if (on_unavailable)
+        on_unavailable(reason, std::move(detail));
+}
+
+void OrcaCloudSignalingChannel::run()
+{
+    try {
+        if (!m_cloud || !m_cloud->ensure_token_fresh("camera")) {
+            unavailable(CameraUnavailableReason::Error, "Unable to refresh OrcaCloud credentials");
+            m_open.store(false);
+            return;
+        }
+        const std::string token = m_cloud->get_access_token();
+        const std::string host = host_without_scheme(m_cloud->get_cloud_service_host());
+        if (token.empty() || host.empty()) {
+            unavailable(CameraUnavailableReason::Error, "OrcaCloud session is unavailable");
+            m_open.store(false);
+            return;
+        }
+
+        const std::string live_token_url =
+            "https://" + host + "/api/v1/printers/" + encode_path_component(m_dev_id) + "/live-token";
+        BOOST_LOG_TRIVIAL(info) << "signaling: POST " << live_token_url << " (dev_id=" << m_dev_id << ")";
+
+        nlohmann::json token_response;
+        std::string token_body;
+        std::string token_error;
+        unsigned int http_code = 0;
+        auto request = Http::post(live_token_url);
+        request.set_post_body(std::string("{}"))
+            .header("Authorization", "Bearer " + token)
+            .header("Content-Type", "application/json")
+            .tls_verify(true)
+            .timeout_max(30)
+            .on_complete([&token_body, &http_code](std::string body, unsigned status) {
+                http_code = status;
+                token_body = std::move(body);
+            })
+            .on_error([&token_body, &token_error, &http_code](std::string body, std::string error, unsigned status) {
+                http_code = status;
+                token_body = std::move(body);
+                token_error = std::move(error);
+            })
+            .perform_sync();
+        BOOST_LOG_TRIVIAL(info) << "signaling: live-token HTTP " << http_code
+                                << (token_error.empty() ? "" : " error=" + token_error)
+                                << " body=" << token_body.substr(0, 512);
+        try {
+            token_response = nlohmann::json::parse(token_body);
+        } catch (const std::exception& e) {
+            BOOST_LOG_TRIVIAL(warning) << "signaling: live-token body is not JSON: " << e.what();
+        }
+        if (http_code < 200 || http_code >= 300 || !token_response.contains("token")) {
+            unavailable(CameraUnavailableReason::Error, "Unable to mint camera live token");
+            m_open.store(false);
+            return;
+        }
+
+        std::vector<CameraIceServer> ice_servers;
+        if (token_response.contains("ice_servers") && token_response["ice_servers"].is_array()) {
+            for (const auto& entry : token_response["ice_servers"]) {
+                if (entry.is_string()) {
+                    ice_servers.push_back({entry.get<std::string>(), {}, {}});
+                } else if (entry.is_object()) {
+                    // RTCIceServer.urls is "string | string[]" (Cloudflare
+                    // Realtime returns an array). Emit one CameraIceServer per
+                    // URL, sharing the credentials.
+                    const std::string username = entry.value("username", std::string{});
+                    const std::string credential = entry.value("credential", std::string{});
+                    const auto add_url = [&](const nlohmann::json& url) {
+                        if (url.is_string() && !url.get<std::string>().empty())
+                            ice_servers.push_back({url.get<std::string>(), username, credential});
+                    };
+                    const auto urls = entry.find("urls");
+                    if (urls != entry.end()) {
+                        if (urls->is_array()) {
+                            for (const auto& url : *urls)
+                                add_url(url);
+                        } else {
+                            add_url(*urls);
+                        }
+                    }
+                }
+            }
+        }
+
+        auto conn = std::make_shared<Connection>();
+        conn->ssl_context.set_default_verify_paths();
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_conn = conn;
+        }
+        auto& websocket = conn->websocket;
+        boost::asio::ip::tcp::resolver resolver(conn->io_context);
+        const auto endpoints = resolver.resolve(host, "443");
+        boost::asio::connect(boost::beast::get_lowest_layer(websocket), endpoints);
+        if (!SSL_set_tlsext_host_name(websocket.next_layer().native_handle(), host.c_str()))
+            throw std::runtime_error("Unable to configure TLS server name");
+        websocket.next_layer().set_verify_mode(boost::asio::ssl::verify_peer);
+        websocket.next_layer().handshake(boost::asio::ssl::stream_base::client);
+        const std::string ws_target = "/api/v1/printers/" + encode_path_component(m_dev_id) +
+                                      "/camera/live?token=" +
+                                      encode_path_component(token_response["token"].get<std::string>());
+        websocket.handshake(host, ws_target);
+        BOOST_LOG_TRIVIAL(info) << "signaling: websocket handshake ok (" << ice_servers.size()
+                                << " ice servers)";
+
+        if (on_ready)
+            on_ready(std::move(ice_servers));
+        send_json(nlohmann::json{{"type", "camera.mode"}, {"mode", "webrtc"}}.dump());
+        // Async read loop, driven by the connection's own io_context. run()
+        // returns once close() has shut the socket down, giving a bounded,
+        // deadlock-free teardown from any thread.
+        do_read(conn);
+        conn->io_context.run();
+    } catch (const std::exception& e) {
+        BOOST_LOG_TRIVIAL(warning) << "signaling: run() exception: " << e.what();
+        if (!m_stop.load())
+            unavailable(CameraUnavailableReason::Closed, e.what());
+    }
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_conn.reset();
+    }
+    m_open.store(false);
+}
+
+void OrcaCloudSignalingChannel::do_read(std::shared_ptr<Connection> conn)
+{
+    auto buffer = std::make_shared<boost::beast::flat_buffer>();
+    conn->websocket.async_read(
+        *buffer, [this, conn, buffer](boost::system::error_code ec, std::size_t) {
+            if (ec) {
+                if (!m_stop.load())
+                    unavailable(CameraUnavailableReason::Closed, ec.message());
+                return; // do not re-arm; io_context.run() unwinds
+            }
+            const std::string raw = boost::beast::buffers_to_string(buffer->data());
+            // A malformed or unexpectedly-shaped message must not tear down the
+            // session: parse/dispatch is guarded.
+            try {
+                dispatch_message(nlohmann::json::parse(raw), raw);
+            } catch (const std::exception& e) {
+                BOOST_LOG_TRIVIAL(warning) << "signaling: ignoring malformed message: " << e.what()
+                                           << " raw=" << raw.substr(0, 256);
+            }
+            if (!m_stop.load())
+                do_read(conn);
+        });
+}
+
+// Returns the string at key, or "" if absent or not a string (JSON null included).
+static std::string json_string(const nlohmann::json& object, const char* key)
+{
+    const auto it = object.find(key);
+    return (it != object.end() && it->is_string()) ? it->get<std::string>() : std::string{};
+}
+
+void OrcaCloudSignalingChannel::dispatch_message(const nlohmann::json& message, const std::string& raw)
+{
+    const std::string type = json_string(message, "type");
+    BOOST_LOG_TRIVIAL(info) << "signaling: recv type=" << type << " raw=" << raw.substr(0, 256);
+    if (type == "webrtc.answer") {
+        const std::string sdp = json_string(message, "sdp");
+        if (on_answer && !sdp.empty())
+            on_answer(sdp);
+    } else if (type == "webrtc.ice" && message.contains("candidate")) {
+        // The peer may send "candidate" as a flat string or as a nested
+        // RTCIceCandidateInit object { candidate, sdpMid, sdpMLineIndex }.
+        const nlohmann::json& candidate = message["candidate"];
+        std::string sdp_candidate;
+        std::string mid = json_string(message, "sdpMid");
+        if (candidate.is_string()) {
+            sdp_candidate = candidate.get<std::string>();
+        } else if (candidate.is_object()) {
+            sdp_candidate = json_string(candidate, "candidate");
+            std::string nested_mid = json_string(candidate, "sdpMid");
+            if (!nested_mid.empty())
+                mid = std::move(nested_mid);
+        }
+        if (on_ice && !sdp_candidate.empty())
+            on_ice(sdp_candidate, mid);
+    } else if (type == "webrtc.unavailable") {
+        const std::string reason = json_string(message, "reason");
+        unavailable(reason == "busy" ? CameraUnavailableReason::Busy
+                                     : reason == "disabled" ? CameraUnavailableReason::Disabled
+                                                            : CameraUnavailableReason::Error,
+                    reason.empty() ? "error" : reason);
+    }
+}
+
+void OrcaCloudSignalingChannel::send_json(const std::string& message)
+{
+    std::shared_ptr<Connection> conn;
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        conn = m_conn;
+    }
+    if (!conn || m_stop.load())
+        return;
+    // Serialize the write onto the io_context thread (same thread that runs
+    // async_read), so reads and writes never touch the stream concurrently.
+    auto payload = std::make_shared<std::string>(message);
+    boost::asio::post(conn->io_context, [this, conn, payload] {
+        if (m_stop.load())
+            return;
+        boost::system::error_code ec;
+        conn->websocket.write(boost::asio::buffer(*payload), ec);
+        if (ec && !m_stop.load())
+            unavailable(CameraUnavailableReason::Closed, ec.message());
+    });
+}
+
+} // namespace Slic3r

src/slic3r/Utils/OrcaCloudSignalingChannel.hpp

@@ -0,0 +1,63 @@
+#pragma once
+
+#include "ICameraSignalingChannel.hpp"
+#include "ICloudServiceAgent.hpp"
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/ssl.hpp>
+#include <boost/beast/ssl.hpp>
+#include <boost/beast/websocket.hpp>
+
+#include <nlohmann/json_fwd.hpp>
+
+#include <atomic>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+
+namespace Slic3r {
+
+class OrcaCloudSignalingChannel : public ICameraSignalingChannel {
+public:
+    OrcaCloudSignalingChannel(std::shared_ptr<ICloudServiceAgent> cloud, std::string dev_id);
+    ~OrcaCloudSignalingChannel() override;
+
+    void open() override;
+    void close() override;
+    void send_offer(std::string sdp) override;
+    void send_ice(std::string candidate, std::string mid) override;
+
+private:
+    using WebSocket = boost::beast::websocket::stream<
+        boost::beast::ssl_stream<boost::asio::ip::tcp::socket>>;
+
+    // The io_context and ssl_context must outlive the websocket stream that
+    // references them. Bundling them here with the stream declared last makes
+    // the destruction order correct (stream first, then contexts), and lets a
+    // single shared_ptr own the whole set.
+    struct Connection {
+        boost::asio::io_context io_context;
+        boost::asio::ssl::context ssl_context{boost::asio::ssl::context::tls_client};
+        WebSocket websocket{io_context, ssl_context};
+    };
+
+    void run();
+    void do_read(std::shared_ptr<Connection> conn);
+    void dispatch_message(const nlohmann::json& message, const std::string& raw);
+    void send_json(const std::string& message);
+    void unavailable(CameraUnavailableReason reason, std::string detail);
+    static std::string encode_path_component(const std::string& value);
+    static std::string host_without_scheme(std::string value);
+
+    std::shared_ptr<ICloudServiceAgent> m_cloud;
+    std::string m_dev_id;
+    std::atomic<bool> m_stop{false};
+    std::atomic<bool> m_open{false};
+    std::thread m_thread;
+    mutable std::mutex m_mutex;
+    std::shared_ptr<Connection> m_conn;
+};
+
+} // namespace Slic3r

src/slic3r/Utils/OrcaMqttConnection.cpp

@@ -0,0 +1,839 @@
+#include "OrcaMqttConnection.hpp"
+
+#include <boost/asio.hpp>
+#include <boost/asio/ssl.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/ssl.hpp>
+#include <boost/beast/websocket.hpp>
+#include <boost/log/trivial.hpp>
+
+#include <openssl/ssl.h>
+
+#include <algorithm>
+#include <chrono>
+#include <memory>
+#include <optional>
+#include <sstream>
+#include <stdexcept>
+#include <utility>
+
+namespace Slic3r {
+
+struct OrcaMqttConnection::Connection {
+    boost::asio::io_context io_context;
+    boost::asio::ssl::context ssl_context;
+    boost::asio::ip::tcp::resolver resolver;
+    // Exactly one of these is engaged once ws_handshake() has run: wss for
+    // wss:// endpoints, ws for plaintext ws://.
+    std::optional<TlsWebSocket>   wss;
+    std::optional<PlainWebSocket> ws;
+
+    Connection()
+        : ssl_context(boost::asio::ssl::context::tls_client)
+        , resolver(io_context)
+    {}
+};
+
+namespace {
+// Apply / clear a tcp_stream timeout on whichever websocket is engaged.
+// Templated on the connection type only because Connection is a private nested
+// type: a deduced parameter needs no (inaccessible) name for it.
+template<class Conn> void expires_after(Conn& conn, std::chrono::seconds timeout) {
+    if (conn.wss)     boost::beast::get_lowest_layer(*conn.wss).expires_after(timeout);
+    else if (conn.ws) boost::beast::get_lowest_layer(*conn.ws).expires_after(timeout);
+}
+template<class Conn> void expires_never(Conn& conn) {
+    if (conn.wss)     boost::beast::get_lowest_layer(*conn.wss).expires_never();
+    else if (conn.ws) boost::beast::get_lowest_layer(*conn.ws).expires_never();
+}
+} // namespace
+
+OrcaMqttConnection::~OrcaMqttConnection() { stop(); }
+
+bool OrcaMqttConnection::start(const Config& config, MessageHandler on_message, StateHandler on_state) {
+    std::lock_guard<std::recursive_mutex> lifecycle_lock(lifecycle_mutex);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT start url=" << config.url
+                            << " use_tls=" << config.use_tls
+                            << " bearer_provider=" << (config.bearer_provider ? "set" : "null")
+                            << " username_present=" << (!config.username.empty())
+                            << " password_present=" << (!config.password.empty())
+                            << " client_id=" << config.client_id
+                            << " keepalive_seconds=" << config.keepalive_seconds
+                            << " message_callback=" << (on_message ? "set" : "null")
+                            << " state_callback=" << (on_state ? "set" : "null");
+    stop();
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        current_config    = config;
+        this->on_message  = std::move(on_message);
+        this->on_state    = std::move(on_state);
+        initial_result    = false;
+        initial_completed = false;
+        connected         = false;
+        m_last_connack_rc.store(-1);
+    }
+    stopping.store(false);
+    worker = std::thread(&OrcaMqttConnection::run, this);
+
+    std::unique_lock<std::mutex> lock(mutex);
+    if (!initial_cv.wait_for(lock, std::chrono::seconds(10), [this] { return initial_completed; })) {
+        initial_completed = true;
+        initial_result    = false;
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT initial connection timed out after 10 seconds"
+                                   << " url=" << current_config.url
+                                   << " last_connack_rc=" << m_last_connack_rc.load()
+                                   << " connected=" << connected.load()
+                                   << "; worker will retry";
+    }
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT start initial_result=" << initial_result
+                            << " initial_completed=" << initial_completed
+                            << " last_connack_rc=" << m_last_connack_rc.load()
+                            << " worker_running=" << (worker.joinable() && !stopping.load());
+    return initial_result;
+}
+
+void OrcaMqttConnection::stop() {
+    std::lock_guard<std::recursive_mutex> lifecycle_lock(lifecycle_mutex);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT stop requested"
+                            << " url=" << current_config.url
+                            << " connected=" << connected.load()
+                            << " worker_joinable=" << worker.joinable()
+                            << " last_connack_rc=" << m_last_connack_rc.load();
+    stopping.store(true);
+    state_cv.notify_all();
+    {
+        std::lock_guard<std::mutex> lock(connection_mutex);
+        if (active_connection) {
+            // Generic so it accepts either the TLS or the plaintext websocket.
+            auto shutdown_socket = [](auto& websocket) {
+                auto& socket = boost::beast::get_lowest_layer(websocket).socket();
+                boost::system::error_code socket_error;
+                socket.cancel(socket_error);
+                socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, socket_error);
+                socket.close(socket_error);
+            };
+            if (active_connection->wss)
+                shutdown_socket(*active_connection->wss);
+            else if (active_connection->ws)
+                shutdown_socket(*active_connection->ws);
+            active_connection->resolver.cancel();
+        }
+    }
+    if (worker.joinable())
+        worker.join();
+
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        connected = false;
+        acknowledged_subscriptions.clear();
+        pending_subscribe_packets.clear();
+        pending_requests.clear();
+        if (!initial_completed) {
+            initial_completed = true;
+            initial_result    = false;
+        }
+    }
+    initial_cv.notify_all();
+}
+
+bool OrcaMqttConnection::is_running() const {
+    return worker.joinable() && !stopping.load();
+}
+
+void OrcaMqttConnection::flush_subscription_change() {
+    std::shared_ptr<Connection> conn;
+    {
+        std::lock_guard<std::mutex> lock(connection_mutex);
+        conn = active_connection;
+    }
+    bool connacked;
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        connacked = connected;
+    }
+    if (!conn || !connacked)
+        return; // no live MQTT session yet — the worker sends the set on CONNACK
+
+    // beast permits a concurrent writer while the worker is blocked in
+    // websocket.read(); every write is serialised by write_mutex inside ws_write().
+    try {
+        send_pending_subscriptions(*conn);
+    } catch (const std::exception& e) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: direct subscription write failed (" << e.what()
+                                   << "); worker will resend the full set on reconnect";
+    }
+}
+
+bool OrcaMqttConnection::subscribe(const std::string& dev_id) {
+    if (dev_id.empty()) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT subscribe rejected empty dev_id";
+        return false;
+    }
+    const std::string topic = report_topic(dev_id);
+    if (topic.size() > 96) { // MQTT topic filter cap enforced by the service
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT subscribe rejected oversized topic=" << topic;
+        return false;
+    }
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        if (subscriptions.count(topic) != 0 && pending_unsubscriptions.count(topic) == 0) {
+            BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT subscribe already queued or active topic=" << topic
+                                    << " acknowledged=" << (acknowledged_subscriptions.count(topic) != 0);
+            return true;
+        }
+        subscriptions.insert(topic);
+        pending_unsubscriptions.erase(topic);
+        pending_subscriptions.insert(topic);
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT subscribe queued topic=" << topic
+                                << " total_subscriptions=" << subscriptions.size()
+                                << " connected=" << connected.load();
+    }
+    state_cv.notify_all();
+    flush_subscription_change(); // emit SUBSCRIBE now on the live socket (no reconnect)
+    return true;
+}
+
+bool OrcaMqttConnection::unsubscribe(const std::string& dev_id) {
+    const std::string topic = report_topic(dev_id);
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        subscriptions.erase(topic);
+        acknowledged_subscriptions.erase(topic);
+        pending_subscriptions.erase(topic);
+        pending_unsubscriptions.insert(topic);
+        for (auto it = pending_subscribe_packets.begin(); it != pending_subscribe_packets.end();) {
+            if (it->second == topic)
+                it = pending_subscribe_packets.erase(it);
+            else
+                ++it;
+        }
+        for (auto it = pending_requests.begin(); it != pending_requests.end();) {
+            if (it->first == dev_id)
+                it = pending_requests.erase(it);
+            else
+                ++it;
+        }
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT unsubscribe queued topic=" << topic
+                                << " total_subscriptions=" << subscriptions.size()
+                                << " connected=" << connected.load();
+    }
+    state_cv.notify_all();
+    flush_subscription_change(); // emit UNSUBSCRIBE now on the live socket (no reconnect)
+    return true;
+}
+
+void OrcaMqttConnection::clear_subscriptions() {
+    std::lock_guard<std::mutex> lock(mutex);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT clear subscriptions count=" << subscriptions.size();
+    subscriptions.clear();
+    pending_subscriptions.clear();
+    pending_unsubscriptions.clear();
+    acknowledged_subscriptions.clear();
+    pending_subscribe_packets.clear();
+    pending_requests.clear();
+}
+
+bool OrcaMqttConnection::parse_endpoint(const std::string& url, Endpoint& endpoint) {
+    std::string rest;
+    std::string default_port;
+    if      (url.rfind("wss://", 0) == 0) { rest = url.substr(6); default_port = "443"; }
+    else if (url.rfind("ws://",  0) == 0) { rest = url.substr(5); default_port = "80";  }
+    else return false;
+
+    const auto slash = rest.find('/');
+    const std::string authority = rest.substr(0, slash);
+    endpoint.target = (slash == std::string::npos) ? "/" : rest.substr(slash);
+
+    // host[:port] — leave an unbracketed IPv6 literal alone
+    const auto colon = authority.rfind(':');
+    if (colon != std::string::npos && authority.find(']') == std::string::npos) {
+        endpoint.host = authority.substr(0, colon);
+        endpoint.port = authority.substr(colon + 1);
+    } else {
+        endpoint.host = authority;
+        endpoint.port = default_port;
+    }
+    return !endpoint.host.empty() && !endpoint.port.empty() && !endpoint.target.empty();
+}
+
+void OrcaMqttConnection::append_string(std::vector<uint8_t>& packet, const std::string& value) {
+    if (value.size() > 0xffff)
+        throw std::runtime_error("MQTT string is too long");
+    packet.push_back(static_cast<uint8_t>(value.size() >> 8));
+    packet.push_back(static_cast<uint8_t>(value.size() & 0xff));
+    packet.insert(packet.end(), value.begin(), value.end());
+}
+
+void OrcaMqttConnection::prepend_remaining_length(std::vector<uint8_t>& packet, size_t length) {
+    std::vector<uint8_t> encoded;
+    do {
+        uint8_t byte = static_cast<uint8_t>(length % 128);
+        length /= 128;
+        if (length != 0)
+            byte |= 0x80;
+        encoded.push_back(byte);
+    } while (length != 0);
+    packet.insert(packet.begin() + 1, encoded.begin(), encoded.end());
+}
+
+std::vector<uint8_t> OrcaMqttConnection::make_connect_packet(
+    const std::string& client_id, const std::string& username,
+    const std::string& password, int keepalive_seconds) {
+    std::vector<uint8_t> packet{0x10};
+    append_string(packet, "MQTT");
+    packet.push_back(4); // protocol level 3.1.1
+
+    uint8_t flags = 0x02; // clean session
+    if (!username.empty()) { flags |= 0x80; if (!password.empty()) flags |= 0x40; }
+    packet.push_back(flags);
+
+    packet.push_back(static_cast<uint8_t>(keepalive_seconds >> 8));
+    packet.push_back(static_cast<uint8_t>(keepalive_seconds & 0xff));
+
+    append_string(packet, client_id.empty() ? "OrcaSlicer" : client_id);
+    if (!username.empty()) {
+        append_string(packet, username);
+        if (!password.empty()) append_string(packet, password);
+    }
+    prepend_remaining_length(packet, packet.size() - 1);
+    return packet;
+}
+
+std::string OrcaMqttConnection::report_topic(const std::string& device_id) { return "device/" + device_id + "/report"; }
+
+std::string OrcaMqttConnection::request_topic(const std::string& id) { return "device/" + id + "/request"; }
+
+std::vector<uint8_t> OrcaMqttConnection::make_publish_packet(const std::string& topic, const std::string& payload) {
+    std::vector<uint8_t> packet{0x30}; // PUBLISH, QoS 0, no retain
+    append_string(packet, topic);      // no packet id at QoS 0
+    packet.insert(packet.end(), payload.begin(), payload.end());
+    prepend_remaining_length(packet, packet.size() - 1);
+    return packet;
+}
+
+std::vector<uint8_t> OrcaMqttConnection::make_subscribe_packet(uint16_t id, const std::string& topic, uint8_t qos) {
+    std::vector<uint8_t> packet{0x82};
+    packet.push_back(id >> 8); packet.push_back(id & 0xff);
+    append_string(packet, topic);
+    packet.push_back(qos);
+    prepend_remaining_length(packet, packet.size() - 1);
+    return packet;
+}
+
+std::vector<uint8_t> OrcaMqttConnection::make_unsubscribe_packet(uint16_t id, const std::string& topic) {
+    std::vector<uint8_t> packet{0xA2};
+    packet.push_back(id >> 8); packet.push_back(id & 0xff);
+    append_string(packet, topic);
+    prepend_remaining_length(packet, packet.size() - 1);
+    return packet;
+}
+
+std::vector<uint8_t> OrcaMqttConnection::make_ping_packet() { return {0xc0, 0}; }
+
+void OrcaMqttConnection::ws_write(Connection& conn, const std::vector<uint8_t>& packet) {
+    if (packet.empty()) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: attempted to send empty MQTT packet";
+        return;
+    }
+    // Writes come from the worker thread AND, for dynamic (un)subscribes, the
+    // caller thread. Serialise them; the worker's concurrent read is fine (beast
+    // allows one reader + one writer).
+    std::lock_guard<std::mutex> lock(write_mutex);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending MQTT packet type=0x" << std::hex
+                            << static_cast<unsigned int>(packet[0] >> 4) << std::dec
+                            << " bytes=" << packet.size();
+    if (conn.wss) {
+        conn.wss->binary(true);
+        conn.wss->write(boost::asio::buffer(packet));
+    } else if (conn.ws) {
+        conn.ws->binary(true);
+        conn.ws->write(boost::asio::buffer(packet));
+    }
+}
+
+std::size_t OrcaMqttConnection::ws_read(Connection& conn, boost::beast::flat_buffer& buffer,
+                                        boost::system::error_code& ec) {
+    if (conn.wss)
+        return conn.wss->read(buffer, ec);
+    if (conn.ws)
+        return conn.ws->read(buffer, ec);
+    ec = boost::asio::error::not_connected;
+    return 0;
+}
+
+void OrcaMqttConnection::ws_close(Connection& conn) {
+    boost::system::error_code close_error;
+    if (conn.wss)
+        conn.wss->close(boost::beast::websocket::close_code::normal, close_error);
+    else if (conn.ws)
+        conn.ws->close(boost::beast::websocket::close_code::normal, close_error);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT connection closed code=" << close_error.value()
+                            << " message=" << close_error.message();
+}
+
+void OrcaMqttConnection::ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint) {
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: WebSocket resolve starting host=" << endpoint.host
+                            << " port=" << endpoint.port << " target=" << endpoint.target
+                            << " tls=" << config.use_tls;
+    const auto results = conn.resolver.resolve(endpoint.host, endpoint.port);
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT DNS resolution succeeded host=" << endpoint.host;
+
+    std::string token;
+    if (config.bearer_provider) {
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: requesting bearer token for WebSocket upgrade";
+        token = config.bearer_provider();
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: bearer token callback completed token_present=" << !token.empty();
+    }
+    auto decorator = [token](boost::beast::websocket::request_type& request) {
+        request.set(boost::beast::http::field::user_agent, "OrcaSlicer");
+        if (!token.empty())
+            request.set(boost::beast::http::field::authorization, "Bearer " + token);
+        request.set("Sec-WebSocket-Protocol", "mqtt");
+    };
+    boost::beast::http::response<boost::beast::http::string_body> response;
+    boost::system::error_code handshake_error;
+
+    if (config.use_tls) {
+        // stop() inspects the engaged optional under connection_mutex; publish it
+        // under the same lock, then release before the blocking connect.
+        {
+            std::lock_guard<std::mutex> lock(connection_mutex);
+            conn.wss.emplace(conn.io_context, conn.ssl_context);
+        }
+        auto& websocket = *conn.wss;
+        auto& stream    = boost::beast::get_lowest_layer(websocket);
+        stream.expires_after(std::chrono::seconds(10));
+        stream.connect(results);
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT TCP connection established host=" << endpoint.host
+                                << " port=" << endpoint.port;
+        // Set SNI before the TLS handshake so the cloud edge selects the correct
+        // certificate.
+        auto& tls_stream = websocket.next_layer();
+        if (!SSL_set_tlsext_host_name(tls_stream.native_handle(), endpoint.host.c_str()))
+            throw std::runtime_error("failed to set Orca Cloud TLS server name");
+        conn.ssl_context.set_default_verify_paths();
+        tls_stream.set_verify_mode(boost::asio::ssl::verify_peer);
+        tls_stream.set_verify_callback(boost::asio::ssl::host_name_verification(endpoint.host));
+        tls_stream.handshake(boost::asio::ssl::stream_base::client);
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT TLS handshake completed host=" << endpoint.host;
+        websocket.set_option(boost::beast::websocket::stream_base::decorator(decorator));
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending TLS WebSocket upgrade target=" << endpoint.target
+                                << " bearer_header=" << (!token.empty());
+        websocket.handshake(response, endpoint.host, endpoint.target, handshake_error);
+    } else {
+        {
+            std::lock_guard<std::mutex> lock(connection_mutex);
+            conn.ws.emplace(conn.io_context);
+        }
+        auto& websocket = *conn.ws;
+        auto& stream    = boost::beast::get_lowest_layer(websocket);
+        stream.expires_after(std::chrono::seconds(10));
+        stream.connect(results);
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT TCP connection established host=" << endpoint.host
+                                << " port=" << endpoint.port << " (plaintext)";
+        websocket.set_option(boost::beast::websocket::stream_base::decorator(decorator));
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending plaintext WebSocket upgrade target=" << endpoint.target
+                                << " bearer_header=" << (!token.empty());
+        websocket.handshake(response, endpoint.host, endpoint.target, handshake_error);
+    }
+
+    if (handshake_error) {
+        // Surface the server's HTTP status so a persistent rejection (stale token,
+        // missing api key, wrong route) is diagnosable from the log rather than an
+        // opaque "handshake declined".
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: WS handshake rejected http="
+                                   << response.result_int() << " (" << response.reason() << "), "
+                                   << handshake_error.message();
+        throw boost::system::system_error(handshake_error, "Orca WebSocket handshake");
+    }
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: WebSocket handshake completed http=" << response.result_int()
+                            << " negotiated_protocol=" << response["Sec-WebSocket-Protocol"];
+    if (response["Sec-WebSocket-Protocol"] != "mqtt") {
+        BOOST_LOG_TRIVIAL(error) << "Orca diagnostic: WebSocket handshake did not negotiate MQTT";
+        throw std::runtime_error("Orca WebSocket did not negotiate MQTT");
+    }
+}
+
+bool OrcaMqttConnection::send_request(const std::string& dev_id, const std::string& payload) {
+    if (dev_id.empty()) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT send_request rejected empty dev_id";
+        return false;
+    }
+    if (!connected.load()) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT send_request rejected because connection is not ready"
+                                   << " dev_id=" << dev_id << " last_connack_rc=" << m_last_connack_rc.load();
+        return false;
+    }
+    std::shared_ptr<Connection> conn;
+    {
+        std::lock_guard<std::mutex> lock(connection_mutex);
+        conn = active_connection;
+    }
+    if (!conn) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT send_request rejected because active connection is null"
+                                   << " dev_id=" << dev_id;
+        return false;
+    }
+    const std::string report = report_topic(dev_id);
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        if (subscriptions.count(report) != 0 && acknowledged_subscriptions.count(report) == 0) {
+            pending_requests.emplace_back(dev_id, payload);
+            BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT request queued until SUBACK"
+                                    << " dev_id=" << dev_id << " payload_bytes=" << payload.size()
+                                    << " pending_requests=" << pending_requests.size();
+            return true;
+        }
+    }
+    try {
+        // ws_write() serialises the write via write_mutex; do not lock it here.
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT PUBLISH request dev_id=" << dev_id
+                                << " topic=" << request_topic(dev_id)
+                                << " payload_bytes=" << payload.size();
+        ws_write(*conn, make_publish_packet(request_topic(dev_id), payload));
+    } catch (const std::exception& e) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: send_request failed dev_id=" << dev_id
+                                   << " (" << e.what() << ")";
+        return false;
+    }
+    return true;
+}
+
+void OrcaMqttConnection::connect_and_read() {
+    m_connection_stage = "creating connection";
+    auto connection = std::make_shared<Connection>();
+    {
+        std::lock_guard<std::mutex> lock(connection_mutex);
+        active_connection = connection;
+        if (stopping.load())
+            return;
+    }
+
+    m_connection_stage = "parsing endpoint";
+    Endpoint endpoint;
+    if (!parse_endpoint(current_config.url, endpoint)) {
+        BOOST_LOG_TRIVIAL(error) << "Orca diagnostic: invalid MQTT endpoint=" << current_config.url;
+        throw std::runtime_error("invalid Orca Cloud WebSocket endpoint");
+    }
+
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT connecting host=" << endpoint.host
+                            << " port=" << endpoint.port << " target=" << endpoint.target;
+
+    m_connection_stage = "WebSocket handshake";
+    ws_handshake(*connection, current_config, endpoint);
+
+    m_connection_stage = "sending MQTT CONNECT";
+    expires_never(*connection);
+    // Auth precedence: a bearer_provider authenticates the WebSocket upgrade, so the
+    // CONNECT username/password fields are omitted entirely (the cloud form).
+    const bool use_bearer = static_cast<bool>(current_config.bearer_provider);
+    ws_write(*connection, make_connect_packet(current_config.client_id,
+                                              use_bearer ? std::string() : current_config.username,
+                                              use_bearer ? std::string() : current_config.password,
+                                              current_config.keepalive_seconds));
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT CONNECT packet sent";
+
+    m_connection_stage = "waiting for MQTT CONNACK";
+    boost::beast::flat_buffer buffer;
+    expires_after(*connection, std::chrono::seconds(10));
+    boost::system::error_code connack_error;
+    ws_read(*connection, buffer, connack_error);
+    if (connack_error)
+        throw boost::system::system_error(connack_error, "read Orca MQTT CONNACK");
+    const std::string connack = boost::beast::buffers_to_string(buffer.data());
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT CONNACK received bytes=" << connack.size()
+                            << " header=" << (connack.empty() ? -1 : static_cast<int>(static_cast<uint8_t>(connack[0])))
+                            << " return_code=" << (connack.size() > 3 ? static_cast<int>(static_cast<uint8_t>(connack[3])) : -1);
+    // rc: 0 accepted, 1..5 refusal, -1 malformed/not a CONNACK.
+    const int rc = (connack.size() == 4 && static_cast<uint8_t>(connack[0]) == 0x20)
+                       ? static_cast<int>(static_cast<uint8_t>(connack[3]))
+                       : -1;
+    m_last_connack_rc.store(rc);
+    if (rc != 0) {
+        BOOST_LOG_TRIVIAL(error) << "Orca diagnostic: MQTT CONNECT refused rc=" << rc;
+        if (rc == 4 || rc == 5) {
+            // Bad credentials / not authorized — retrying cannot help. Make run()'s
+            // loop exit and unblock any waiting start().
+            stopping.store(true);
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                initial_completed = true;
+                initial_result    = false;
+            }
+            initial_cv.notify_all();
+        }
+        throw std::runtime_error("Orca MQTT CONNECT refused rc=" + std::to_string(rc));
+    }
+
+    m_connection_stage = "reading MQTT messages";
+    // The subscription acknowledgement belongs to this MQTT session. Clear
+    // the previous session's state before notifying the owner, because the
+    // reconnect callback immediately queues the printer's initial requests.
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        acknowledged_subscriptions.clear();
+        pending_subscribe_packets.clear();
+    }
+    notify_state(true);
+    reconnect_delay_seconds.store(1); // a fresh CONNACK resets the backoff
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT connection is ready; sending current subscriptions";
+    send_current_subscriptions(*connection);
+    std::chrono::steady_clock::time_point next_ping = std::chrono::steady_clock::now() + std::chrono::seconds(30);
+
+    while (!stopping.load()) {
+        send_pending_subscriptions(*connection);
+        // Keepalive is driven every iteration, not only from the read-timeout branch:
+        // a printer pushing faster than the 1s read deadline would otherwise keep the
+        // read hot and the broker would drop us at 1.5 x keepalive.
+        if (std::chrono::steady_clock::now() >= next_ping) {
+            ws_write(*connection, make_ping_packet());
+            BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT PINGREQ sent";
+            next_ping = std::chrono::steady_clock::now() + std::chrono::seconds(30);
+        }
+        buffer.consume(buffer.size());
+        m_connection_stage = "reading MQTT frame";
+        expires_after(*connection, std::chrono::seconds(1));
+        boost::system::error_code error;
+        ws_read(*connection, buffer, error);
+        if (error == boost::beast::error::timeout)
+            continue;
+        if (error) {
+            BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT WebSocket read failed code=" << error.value()
+                                       << " message=" << error.message();
+            throw boost::system::system_error(error, "read Orca MQTT message");
+        }
+        handle_packet(boost::beast::buffers_to_string(buffer.data()));
+    }
+
+    ws_close(*connection);
+    if (!stopping.load())
+        notify_state(false);
+}
+
+void OrcaMqttConnection::send_current_subscriptions(Connection& conn) {
+    std::vector<std::string> topics;
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        topics.assign(subscriptions.begin(), subscriptions.end());
+        acknowledged_subscriptions.clear();
+        pending_subscribe_packets.clear();
+        for (const std::string& topic : topics)
+            pending_subscriptions.erase(topic);
+    }
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending current MQTT subscriptions count=" << topics.size();
+    for (const std::string& topic : topics) {
+        const uint16_t packet_id = next_packet_id++;
+        {
+            std::lock_guard<std::mutex> lock(mutex);
+            pending_subscribe_packets[packet_id] = topic;
+        }
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending SUBSCRIBE topic=" << topic << " packet_id=" << packet_id;
+        ws_write(conn, make_subscribe_packet(packet_id, topic, 1));
+    }
+}
+
+void OrcaMqttConnection::send_pending_subscriptions(Connection& conn) {
+    std::vector<std::string> subscribe_topics;
+    std::vector<std::string> unsubscribe_topics;
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        subscribe_topics.assign(pending_subscriptions.begin(), pending_subscriptions.end());
+        unsubscribe_topics.assign(pending_unsubscriptions.begin(), pending_unsubscriptions.end());
+        pending_subscriptions.clear();
+        pending_unsubscriptions.clear();
+    }
+    for (const std::string& topic : subscribe_topics) {
+        const uint16_t packet_id = next_packet_id++;
+        {
+            std::lock_guard<std::mutex> lock(mutex);
+            pending_subscribe_packets[packet_id] = topic;
+        }
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending pending SUBSCRIBE topic=" << topic
+                                << " packet_id=" << packet_id;
+        ws_write(conn, make_subscribe_packet(packet_id, topic, 1));
+    }
+    for (const std::string& topic : unsubscribe_topics) {
+        const uint16_t packet_id = next_packet_id++;
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: sending pending UNSUBSCRIBE topic=" << topic
+                                << " packet_id=" << packet_id;
+        ws_write(conn, make_unsubscribe_packet(packet_id, topic));
+    }
+}
+
+void OrcaMqttConnection::handle_packet(const std::string& packet) {
+    if (packet.size() < 2) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: received undersized MQTT packet bytes=" << packet.size();
+        return;
+    }
+    const uint8_t header = static_cast<uint8_t>(packet[0]);
+    const uint8_t packet_type = header >> 4;
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: received MQTT packet type=" << static_cast<unsigned int>(packet_type)
+                            << " header=0x" << std::hex << static_cast<unsigned int>(header) << std::dec
+                            << " bytes=" << packet.size();
+    if (packet_type != 3) { // Only QoS 0 PUBLISH carries printer status.
+        if (packet_type == 9 && packet.size() >= 5) {
+            const uint16_t packet_id = (static_cast<unsigned int>(static_cast<uint8_t>(packet[2])) << 8) |
+                                       static_cast<unsigned int>(static_cast<uint8_t>(packet[3]));
+            std::ostringstream result_codes;
+            for (size_t index = 4; index < packet.size(); ++index) {
+                if (index != 4)
+                    result_codes << ',';
+                result_codes << "0x" << std::hex << static_cast<unsigned int>(static_cast<uint8_t>(packet[index]));
+            }
+            BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: received SUBACK packet_id="
+                                    << packet_id
+                                    << " result_codes=" << result_codes.str();
+
+            // Each production SUBSCRIBE packet currently contains one topic.
+            // MQTT grants QoS 0 or 1 for a requested QoS 1 subscription; 0x80
+            // means the subscription was rejected.
+            const uint8_t result = static_cast<uint8_t>(packet[4]);
+            std::string topic;
+            std::deque<std::pair<std::string, std::string>> requests;
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                auto pending = pending_subscribe_packets.find(packet_id);
+                if (pending != pending_subscribe_packets.end()) {
+                    topic = pending->second;
+                    pending_subscribe_packets.erase(pending);
+                    for (auto it = pending_requests.begin(); it != pending_requests.end();) {
+                        if (report_topic(it->first) == topic) {
+                            requests.push_back(std::move(*it));
+                            it = pending_requests.erase(it);
+                        } else {
+                            ++it;
+                        }
+                    }
+                    if (result == 0 || result == 1) {
+                        acknowledged_subscriptions.insert(topic);
+                    }
+                }
+            }
+            if (topic.empty()) {
+                BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: SUBACK has no pending topic packet_id=" << packet_id;
+            } else if (result == 0 || result == 1) {
+                BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: report subscription active topic=" << topic
+                                        << " granted_qos=" << static_cast<unsigned int>(result)
+                                        << " releasing_requests=" << requests.size();
+                for (const auto& request : requests) {
+                    if (!send_request(request.first, request.second)) {
+                        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: queued MQTT request could not be sent"
+                                                   << " after SUBACK dev_id=" << request.first;
+                    }
+                }
+            } else {
+                BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: report subscription rejected topic=" << topic
+                                           << " result_code=0x" << std::hex << static_cast<unsigned int>(result) << std::dec
+                                           << " dropped_requests=" << requests.size();
+            }
+        }
+        return;
+    }
+    size_t index = 1;
+    size_t multiplier = 1;
+    size_t remaining = 0;
+    uint8_t encoded = 0;
+    do {
+        if (index >= packet.size() || multiplier > 128 * 128 * 128) {
+            BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: malformed MQTT PUBLISH remaining length";
+            return;
+        }
+        encoded = static_cast<uint8_t>(packet[index++]);
+        remaining += (encoded & 0x7f) * multiplier;
+        multiplier *= 128;
+    } while ((encoded & 0x80) != 0);
+    const size_t remaining_end = index + remaining;
+    if (remaining_end > packet.size() || remaining < 2 || index + 2 > remaining_end) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: malformed MQTT PUBLISH body remaining=" << remaining
+                                   << " packet_bytes=" << packet.size();
+        return;
+    }
+    const uint16_t topic_length = (static_cast<uint8_t>(packet[index]) << 8) |
+                                  static_cast<uint8_t>(packet[index + 1]);
+    index += 2;
+    if (topic_length > packet.size() - index) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: malformed MQTT PUBLISH topic length=" << topic_length;
+        return;
+    }
+    const std::string topic(packet.data() + index, topic_length);
+    index += topic_length;
+    if (((header >> 1) & 0x03) != 0) {
+        if (index + 2 > remaining_end) {
+            BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: malformed MQTT PUBLISH packet identifier";
+            return;
+        }
+        index += 2; // QoS 1/2 packet identifier; the service currently sends QoS 0.
+    }
+    const size_t payload_size = remaining_end - index;
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: received PUBLISH topic=" << topic
+                            << " payload_bytes=" << payload_size
+                            << " message_callback=" << (on_message ? "set" : "null");
+    // topic is "device/<id>/report" (or "/request"); hand the id up, drop anything else.
+    std::string dev_id;
+    if (topic.rfind("device/", 0) == 0) {
+        const size_t id_start = 7;
+        const size_t id_end   = topic.rfind('/');
+        if (id_end != std::string::npos && id_end > id_start)
+            dev_id = topic.substr(id_start, id_end - id_start);
+    }
+    if (dev_id.empty()) {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: dropping PUBLISH on unrecognized topic=" << topic;
+    } else if (on_message) {
+        on_message(dev_id, packet.substr(index, remaining_end - index));
+    } else {
+        BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: dropping PUBLISH because message callback is not set";
+    }
+}
+
+void OrcaMqttConnection::notify_state(bool is_now_connected) {
+    StateHandler callback;
+    bool initial = false;
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        connected = is_now_connected;
+        initial   = !initial_completed;
+        if (initial) {
+            initial_result    = is_now_connected;
+            initial_completed = true;
+        }
+        callback = on_state;
+    }
+    BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT state changed connected=" << is_now_connected
+                            << " initial=" << initial << " state_callback=" << (callback ? "set" : "null");
+    if (initial)
+        initial_cv.notify_all();
+    else if (callback)
+        callback(is_now_connected, false);
+}
+
+void OrcaMqttConnection::run() {
+    while (!stopping.load()) {
+        const int retry_seconds = reconnect_delay_seconds.load();
+        const uint64_t attempt = ++m_attempt_number;
+        m_connection_stage = "starting attempt";
+        try {
+            BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT connection attempt=" << attempt
+                                    << " retry_delay=" << retry_seconds
+                                    << " url=" << current_config.url;
+            connect_and_read();
+        } catch (const std::exception& error) {
+            BOOST_LOG_TRIVIAL(warning) << "Orca diagnostic: MQTT connection attempt=" << attempt
+                                       << " failed stage=" << m_connection_stage
+                                       << " error=" << error.what()
+                                       << " last_connack_rc=" << m_last_connack_rc.load()
+                                       << " stopping=" << stopping.load();
+            if (!stopping.load())
+                notify_state(false);
+        }
+        if (stopping.load())
+            break;
+        // Grow the backoff only across attempts that never reached CONNACK; a
+        // successful connection resets reconnect_delay_seconds to 1 (connect_and_read).
+        reconnect_delay_seconds.store(std::min(retry_seconds * 2, 30));
+        std::unique_lock<std::mutex> lock(mutex);
+        BOOST_LOG_TRIVIAL(info) << "Orca diagnostic: MQTT waiting before reconnect seconds=" << retry_seconds;
+        state_cv.wait_for(lock, std::chrono::seconds(retry_seconds), [this] { return stopping.load(); });
+    }
+}
+
+} // namespace Slic3r

src/slic3r/Utils/OrcaMqttConnection.hpp

@@ -0,0 +1,152 @@
+#ifndef slic3r_OrcaMqttConnection_hpp_
+#define slic3r_OrcaMqttConnection_hpp_
+
+#include <boost/asio/ssl.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/ssl.hpp>
+#include <boost/beast/websocket.hpp>
+#include <atomic>
+#include <condition_variable>
+#include <deque>
+#include <functional>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <set>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+#include <cstddef>
+#include <cstdint>
+
+namespace Slic3r {
+
+// Minimal MQTT 3.1.1 codec + WebSocket transport (ws:// and wss://), shared by the
+// LAN (OrcaSonar) and cloud (fleet) printer connections. Both PUBLISH
+// commands to device/<id>/request and SUBSCRIBE device/<id>/report; Config is the
+// only per-transport difference.
+class OrcaMqttConnection
+{
+public:
+    using TokenProvider  = std::function<std::string()>;
+    using MessageHandler = std::function<void(const std::string&, const std::string&)>;
+    using StateHandler   = std::function<void(bool connected, bool initial)>;
+
+    struct Endpoint { std::string host; std::string port; std::string target; };
+
+    struct Config {
+        std::string   url;
+        bool          use_tls = false;
+        TokenProvider bearer_provider;       // set => bearer on WS upgrade, CONNECT creds omitted
+        std::string   username;
+        std::string   password;
+        std::string   client_id  = "OrcaSlicer";
+        int           keepalive_seconds = 60;
+    };
+
+    static bool parse_endpoint(const std::string& url, Endpoint& endpoint);
+
+    // Build an MQTT 3.1.1 CONNECT packet. Clean-session is always set; the
+    // username/password connect flags and payload fields are added only when
+    // username is non-empty (the cloud form authenticates via a bearer on the
+    // WebSocket upgrade and omits CONNECT credentials). Public for unit tests.
+    static std::vector<uint8_t> make_connect_packet(const std::string& client_id,
+                                                    const std::string& username,
+                                                    const std::string& password,
+                                                    int keepalive_seconds);
+
+    // Topic-string helpers for the per-device request/report channels and the
+    // MQTT 3.1.1 PUBLISH / SUBSCRIBE / UNSUBSCRIBE packet builders. All public
+    // for unit tests. make_publish_packet emits QoS 0 (no packet identifier).
+    static std::string request_topic(const std::string& dev_id);            // "device/<id>/request"
+    static std::string report_topic(const std::string& dev_id);             // "device/<id>/report"
+    static std::vector<uint8_t> make_publish_packet(const std::string& topic, const std::string& payload);
+    static std::vector<uint8_t> make_subscribe_packet(uint16_t packet_id, const std::string& topic, uint8_t qos);
+    static std::vector<uint8_t> make_unsubscribe_packet(uint16_t packet_id, const std::string& topic);
+
+    ~OrcaMqttConnection();
+
+    bool start(const Config& config, MessageHandler on_message, StateHandler on_state);
+    void stop();
+    // True while the worker thread is alive (connected OR retrying). Lets callers
+    // avoid restarting a healthy connection.
+    bool is_running() const;
+    // True once CONNACK has been received and the socket has not since dropped.
+    bool is_connected() const { return connected.load(); }
+    bool subscribe(const std::string& dev_id);
+    bool unsubscribe(const std::string& dev_id);
+    // Last MQTT CONNACK return code: 0 ok, 1..5 refusal, -1 none seen this attempt.
+    int  last_connack_rc() const { return m_last_connack_rc.load(); }
+    void clear_subscriptions();
+    bool send_request(const std::string& dev_id, const std::string& payload);
+
+private:
+    // The endpoint may be either a TLS (wss://) or a plaintext (ws://) WebSocket;
+    // Connection holds whichever one is engaged and the ws_* helpers below
+    // dispatch on it.
+    using TlsWebSocket   = boost::beast::websocket::stream<
+        boost::asio::ssl::stream<boost::beast::tcp_stream>>;
+    using PlainWebSocket = boost::beast::websocket::stream<boost::beast::tcp_stream>;
+    struct Connection;
+
+    static void append_string(std::vector<uint8_t>& packet, const std::string& value);
+    static void prepend_remaining_length(std::vector<uint8_t>& packet, size_t length);
+    static std::vector<uint8_t> make_ping_packet();
+
+    // Transport dispatch: each forwards to conn.wss (TLS) or conn.ws (plaintext).
+    void        ws_write(Connection& conn, const std::vector<uint8_t>& packet); // locks write_mutex
+    std::size_t ws_read(Connection& conn, boost::beast::flat_buffer& buffer, boost::system::error_code& ec);
+    void        ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint);
+    void        ws_close(Connection& conn);
+    // Emit a queued SUBSCRIBE/UNSUBSCRIBE on the live socket right now (from the
+    // caller thread), so a selection change is applied without waiting for the
+    // blocking read loop to next return. No-op if no CONNACKed socket exists yet
+    // (the worker sends the set on connect). The WebSocket is never dropped for a
+    // subscription change.
+    void flush_subscription_change();
+    void connect_and_read();
+    void send_current_subscriptions(Connection& conn);
+    void send_pending_subscriptions(Connection& conn);
+    void handle_packet(const std::string& packet);
+    void notify_state(bool is_now_connected);
+    void run();
+
+    std::atomic_bool stopping{true};
+    std::atomic_int reconnect_delay_seconds{1};
+    // Serialises the whole of start() and stop() against each other, so the UI
+    // thread's stop() (disconnect / dtor) cannot race the connect thread's start()
+    // into a concurrent worker.join(). Recursive because start() calls stop().
+    std::recursive_mutex lifecycle_mutex;
+    std::thread worker;
+    std::mutex mutex;
+    std::mutex connection_mutex;
+    std::mutex write_mutex; // serialises every websocket write (worker + caller threads)
+    std::shared_ptr<Connection> active_connection;
+    std::condition_variable initial_cv;
+    std::condition_variable state_cv;
+    Config current_config;
+    MessageHandler on_message;
+    StateHandler on_state;
+    // Full report-topic strings ("device/<id>/report"), not bare device ids.
+    std::set<std::string> subscriptions;
+    std::set<std::string> pending_subscriptions;
+    std::set<std::string> pending_unsubscriptions;
+    // Requests for a subscribed device wait until the corresponding SUBACK is
+    // received. Otherwise an immediate pushall response can be published by
+    // the broker before this client is actually subscribed to the report topic.
+    std::set<std::string> acknowledged_subscriptions;
+    std::map<uint16_t, std::string> pending_subscribe_packets;
+    std::deque<std::pair<std::string, std::string>> pending_requests;
+    std::atomic<uint16_t> next_packet_id{1};
+    std::atomic<int> m_last_connack_rc{-1};
+    uint64_t m_attempt_number{0};       // worker-thread diagnostic sequence
+    std::string m_connection_stage;     // worker-thread diagnostic stage
+    bool initial_result{false};
+    bool initial_completed{false};
+    std::atomic_bool connected{false};
+};
+
+} // namespace Slic3r
+
+#endif // slic3r_OrcaMqttConnection_hpp_

src/slic3r/Utils/OrcaPrinterAgent.hpp

@@ -3,19 +3,28 @@
 
 #include "IPrinterAgent.hpp"
 #include "ICloudServiceAgent.hpp"
+#include "OrcaCloudServiceAgent.hpp"
+#include "OrcaMqttConnection.hpp"
+#include <atomic>
+#include <cstdint>
+#include <functional>
 #include <string>
 #include <mutex>
 #include <memory>
+#include <thread>
 
 namespace Slic3r {
 
+class OrcaCloudServiceAgent;
+
 /**
- * OrcaPrinterAgent - Stub implementation for printer operations.
+ * OrcaPrinterAgent - OrcaSonar MQTT printer agent.
  *
- * All printer-related operations are currently stubs that return success.
+ * LAN and cloud commands use the same OrcaSonar protocol payloads; only the
- * Actual printer connectivity requires the BBL SDK or future Orca implementation.
+ * MQTT connection selected by route_send() differs.
  */
-class OrcaPrinterAgent : public IPrinterAgent {
+class OrcaPrinterAgent : public IPrinterAgent
+{
 public:
     explicit OrcaPrinterAgent(std::string log_dir);
     ~OrcaPrinterAgent() override;
@@ -25,6 +34,10 @@ public:
     // ========================================================================
 
     void set_cloud_agent(std::shared_ptr<ICloudServiceAgent> cloud) override;
+    CameraStreamMode get_camera_stream_mode() const override;
+    std::string get_camera_url() const override;
+    std::unique_ptr<ICameraSignalingChannel>
+    create_camera_signaling_channel(const std::string& dev_id) override;
 
     // Communication
     int send_message(std::string dev_id, std::string json_str, int qos, int flag) override;
@@ -42,7 +55,13 @@ public:
     // Binding
     int ping_bind(std::string ping_code) override;
     int bind_detect(std::string dev_ip, std::string sec_link, detectResult& detect) override;
-    int bind(std::string dev_ip, std::string dev_id, std::string dev_model, std::string sec_link, std::string timezone, bool improved, OnUpdateStatusFn update_fn) override;
+    int bind(std::string dev_ip,
+             std::string dev_id,
+             std::string dev_model,
+             std::string sec_link,
+             std::string timezone,
+             bool improved,
+             OnUpdateStatusFn update_fn) override;
     int unbind(std::string dev_id) override;
     int request_bind_ticket(std::string* ticket) override;
     int get_hms_snapshot(std::string dev_id, std::string file_name, std::function<void(std::string, int)> callback) override;
@@ -77,10 +96,127 @@ public:
     int set_on_local_message_fn(OnMessageFn fn) override;
     int set_queue_on_main_fn(QueueOnMainFn fn) override;
 
+    int command_ams_refresh_rfid(std::string dev_id, std::string tray_id, int sequence_id, bool lan_mode) override;
+    int command_ams_calibrate(std::string dev_id, int ams_id, int sequence_id, bool lan_mode) override;
+    int command_ams_select_tray(std::string dev_id, std::string tray_id, int sequence_id, bool lan_mode) override;
+    int command_start_camera(std::string dev_id) override;
+    int command_xyz_abs(std::string dev_id, int sequence_id, bool lan_mode) override;
+    int command_auto_leveling(std::string dev_id, int sequence_id, bool lan_mode) override;
+    int command_go_home(std::string dev_id, bool is_printing, bool supports_mqtt_homing, int sequence_id, bool lan_mode) override;
+    int command_set_bed(std::string dev_id, int temp, bool supports_mqtt_bed_ctrl, int sequence_id, bool lan_mode) override;
+    int command_set_nozzle(std::string dev_id, int temp, int sequence_id, bool lan_mode) override;
+    int command_axis_control(std::string dev_id,
+                             std::string axis,
+                             double unit,
+                             double input_val,
+                             int speed,
+                             bool is_core_xy,
+                             bool supports_mqtt_axis_control,
+                             int sequence_id,
+                             bool lan_mode) override;
+
+    // Test-only: drive emit_connect_sequence directly (no socket).
+    void run_connect_sequence_for_test(const std::string& dev_id)
+    {
+        emit_connect_sequence(dev_id, [](const std::string&) {}, [](const std::string&) {});
+    }
+
+    // Test-only: advance the LAN connection epoch without a connect/disconnect cycle.
+    void bump_lan_generation_for_test() { ++m_lan_generation; }
+    // Test-only: the same for the (independent) cloud selection epoch.
+    void bump_cloud_generation_for_test() { ++m_cloud_generation; }
+
+protected:
+    // Forward one inbound printer message to on_message_fn or on_local_message_fn (marshalled onto the UI
+    // thread via queue_on_main_fn when set). Body of every connection's MessageHandler.
+    void deliver_to_sink(const std::string& dev_id, const std::string& payload, bool local);
+
+    // Extract OrcaSonar's print.ipcam.stream_mode from LAN reports before they
+    // are forwarded to the GUI. The getters below then read this agent-owned state.
+    void parse_ipcam_info(const std::string& dev_id, const std::string& payload);
+
+    // Orca-dialect -> Bambu-dialect compatibility shim for inbound reports: the single
+    // place Orca Protocol JSON is rewritten into the shapes MachineObject::parse_json
+    // already handles, so parse_json needs no Orca-specific changes. Self-contained
+    // (its cache is a function-local static) and deletable together with its call site
+    // once parse_json reads the Orca dialect natively. See the definition for the
+    // per-rule detail. Returns the payload unchanged when no rule applies.
+    std::string merge_capabilities(const std::string& dev_id, const std::string& payload);
+
+    // Report the asynchronous LAN connection state using the same callback contract as
+    // the other printer agents. The transport result cannot be returned by
+    // connect_printer(), which only starts the worker.
+    void dispatch_local_connect(int state, const std::string& dev_id, const std::string& message);
+
+    // The LAN inbound-message handler for one connection generation: forwards to
+    // deliver_to_sink only while `generation` is still the live epoch.
+    std::function<void(const std::string&, const std::string&)> make_lan_message_handler(uint64_t generation);
+
+    // Pure LAN-address parsing + client-id. protected static so the test Probe reaches them.
+    static bool parse_lan_endpoint(const std::string& dev_ip, std::string& host, std::string& port);
+    static std::string make_lan_client_id(const std::string& dev_id);
+    // Test hook: the ws:// URL connect_printer built for the current LAN session ("" if none).
+    std::string lan_connection_target() const;
+    // Shared post-connect sequence: SUBSCRIBE, then pushing.start, pushall,
+    // info.get_version, info.get_capabilities. Runs identically on LAN and cloud.
+    void on_connected(const std::string& dev_id, OrcaMqttConnection* conn, uint64_t generation);
+
+    // The post-connect command sequence, factored behind a seam so a test can
+    // observe the SUBSCRIBE + 4 request payloads without a live OrcaMqttConnection.
+    virtual void emit_connect_sequence(const std::string& dev_id,
+                                       std::function<void(const std::string&)> subscribe,
+                                       std::function<void(const std::string&)> request);
+    static std::string seq(int n); // decimal string in the OrcaSlicer 20000..29999 band
+    static std::string build_pushing_start(const std::string& sequence_id);
+    static std::string build_pushing_stop(const std::string& sequence_id);
+    static std::string build_pushall(const std::string& sequence_id);
+    static std::string build_get_version(const std::string& sequence_id);
+    static std::string build_get_capabilities(const std::string& sequence_id);
+
 private:
+    class OrcaSonarDiscovery;
+
     std::string log_dir;
     std::string selected_machine;
+
+    enum CurrentConn { NONE, CLOUD, LAN };
+    static const char* connection_type_name(CurrentConn connection);
+
+    // The transport for the printer currently selected by the UI.  LAN and
+    // cloud sessions have separate connection objects, so this is selection
+    // state rather than an inference from whichever socket happens to exist.
+    CurrentConn m_current_connection = NONE;
+
     std::shared_ptr<ICloudServiceAgent> m_cloud_agent;
+    std::unique_ptr<OrcaMqttConnection> lan_mqtt_connection;
+
+    // Two independent epochs: a cloud (de)selection must not fence the live LAN
+    // feed, and vice versa. Each transport's connect thread and inbound handler
+    // compare against their own counter only.
+    std::atomic<uint64_t> m_lan_generation{0};
+    std::atomic<uint64_t> m_cloud_generation{0};
+
+    // The short-lived threads that run the blocking initial connect for the current
+    // LAN / cloud session. Joined members (never detached) so they cannot outlive
+    // *this or the connection they hold a raw pointer to.
+    std::thread m_lan_connect_thread;
+    std::thread m_cloud_connect_thread;
+
+    std::unique_ptr<OrcaSonarDiscovery> m_discovery;
+
+    std::string m_lan_dev_id; // guarded by state_mutex
+    std::string m_lan_url;    // guarded by state_mutex — the Config.url of the live LAN session
+    CameraStreamMode m_camera_stream_mode = CameraStreamMode::none; // guarded by state_mutex
+    std::string m_camera_url; // guarded by state_mutex
+
+    OrcaCloudServiceAgent* get_orca_cloud_agent();
+
+    OrcaMqttConnection* get_appropriate_mqtt_connection(bool is_lan = true);
+    static bool parse_nonnegative_command_id(const std::string& value, int& result);
+
+    // Route one command payload to device/<dev_id>/request on the LAN or the shared
+    // cloud connection. The uniform send path for both send_message* overrides.
+    int route_send(bool is_lan, const std::string& dev_id, const std::string& json_str);
 
     // Callbacks
     OnMsgArrivedFn on_ssdp_msg_fn;

src/slic3r/Utils/QidiPrinterAgent.cpp

@@ -1,5 +1,6 @@
 #include "QidiPrinterAgent.hpp"
 #include "Http.hpp"
+#include "IPrinterAgent.hpp"
 #include "libslic3r/PresetBundle.hpp"
 #include "slic3r/GUI/GUI_App.hpp"
 
@@ -8,6 +9,7 @@
 #include <boost/log/trivial.hpp>
 #include <cctype>
 #include <sstream>
+#include <thread>
 
 using json = nlohmann::json;
 
@@ -27,6 +29,15 @@ bool has_visible_base_preset(const PresetCollection& filaments, const std::strin
     return false;
 }
 
+// RAII decrement for MoonrakerPrinterAgent::filament_fetch_in_flight — guarantees the
+// counter drops back down on every exit path (early return or fall-through) inside the
+// detached fetch thread below, so ~MoonrakerPrinterAgent()'s wait loop can't stall forever.
+struct InFlightGuard
+{
+    std::atomic<int>& counter;
+    ~InFlightGuard() { counter.fetch_sub(1, std::memory_order_relaxed); }
+};
+
 } // anonymous namespace
 
 const std::string QidiPrinterAgent_VERSION = "0.0.1";
@@ -40,42 +51,141 @@ AgentInfo QidiPrinterAgent::get_agent_info_static()
     return AgentInfo{"qidi", "Qidi", QidiPrinterAgent_VERSION, "Qidi printer agent"};
 }
 
-bool QidiPrinterAgent::fetch_filament_info(std::string dev_id, FilamentSyncMode /*sync_mode*/)
+FilamentSyncMode QidiPrinterAgent::get_filament_sync_mode() const
 {
-    std::string error;
+    if (GUI::wxGetApp().app_config->get_bool("use_printer_agents"))
+        return FilamentSyncMode::subscription;
+    return FilamentSyncMode::pull;
+}
 
-    // 1. Fetch device info and infer series_id
+bool QidiPrinterAgent::fetch_filament_info(std::string dev_id, FilamentSyncMode sync_mode)
-    std::string series_id;
+{
-    {
+    if (sync_mode != get_filament_sync_mode())
-        MoonrakerDeviceInfo info;
+        return false;
-        if (fetch_device_info(device_info.base_url, device_info.api_key, info, error)) {
+
-            series_id = infer_series_id(info.model_id, info.dev_name);
+    // Snapshot only what the fetch needs, rather than reading device_info live from the
+    // background thread below — device_info can be concurrently rewritten by a reconnect
+    // on another thread while this fetch is still in flight.
+    std::string base_url   = device_info.base_url;
+    std::string api_key    = device_info.api_key;
+    std::string model_id   = device_info.model_id;
+    std::string model_name = device_info.model_name;
+
+    filament_fetch_in_flight.fetch_add(1, std::memory_order_relaxed);
+
+    std::thread([this, base_url, api_key, model_id, model_name]() {
+        InFlightGuard guard{filament_fetch_in_flight};
+
+        std::string error;
+
+        // 1. Fetch device info and infer series_id
+        std::string series_id;
+        {
+            MoonrakerDeviceInfo info;
+            if (fetch_device_info(base_url, api_key, info, error)) {
+                series_id = infer_series_id(info.model_id, info.dev_name);
+            }
+        }
+        if (series_id.empty()) {
+            // Fall back to the configured Orca model if Moonraker doesn't expose a usable identifier.
+            series_id = infer_series_id(model_id, model_name);
         }
-    }
-    if (series_id.empty()) {
-        // Fall back to the configured Orca model if Moonraker doesn't expose a usable identifier.
-        series_id = infer_series_id(device_info.model_id, device_info.model_name);
-    }
 
-    // 2. Fetch filament dictionary
+        // 2. Fetch filament dictionary
-    QidiFilamentDict dict;
+        QidiFilamentDict dict;
-    if (!fetch_filament_dict(device_info.base_url, device_info.api_key, dict, error)) {
+        if (!fetch_filament_dict(base_url, api_key, dict, error)) {
-        BOOST_LOG_TRIVIAL(warning) << "QidiPrinterAgent::fetch_filament_info: Failed to fetch filament dict: " << error;
+            BOOST_LOG_TRIVIAL(warning) << "QidiPrinterAgent::fetch_filament_info: Failed to fetch filament dict: " << error;
-    }
+        }
 
-    // 3. Fetch slot info and build AmsTrayData directly
+        // 3. Fetch slot info and build AmsTrayData directly
-    std::vector<AmsTrayData> trays;
+        std::vector<AmsTrayData> trays;
-    int                      box_count = 0;
+        int box_count = 0;
-    if (!fetch_slot_info(device_info.base_url, device_info.api_key, dict, series_id, trays, box_count, error)) {
+        if (!fetch_slot_info(base_url, api_key, dict, series_id, trays, box_count, error)) {
-        BOOST_LOG_TRIVIAL(warning) << "QidiPrinterAgent::fetch_filament_info: Failed to fetch slot info: " << error;
+            BOOST_LOG_TRIVIAL(warning) << "QidiPrinterAgent::fetch_filament_info: Failed to fetch slot info: " << error;
+            return;
+        }
+
+        // 4. Build the AMS payload
+        build_ams_payload(box_count, box_count * 4 - 1, trays);
+    }).detach();
+    return true;
+}
+
+bool QidiPrinterAgent::apply_box_mapping(const PrintParams& params) const
+{
+    // enable_box mirrors task_use_ams: engage the multi-color box only when this
+    // job actually routes filament through it. (See qidi-ams-findings.md §2/§8.3 —
+    // if firmware treats enable_box as "a box exists" rather than "use it this job",
+    // switch this gate to HasAms()/box_count instead.)
+    const int enable = params.task_use_ams ? 1 : 0;
+    if (!send_gcode(device_info.dev_id, "SAVE_VARIABLE VARIABLE=enable_box VALUE=" + std::to_string(enable))) {
+        BOOST_LOG_TRIVIAL(error) << "QidiPrinterAgent::apply_box_mapping: failed to set enable_box";
         return false;
     }
 
-    // 4. Build the AMS payload
+    // When the box isn't used this job, leave the existing value_t<tool> slot
-    build_ams_payload(box_count, box_count * 4 - 1, trays);
+    // assignments untouched (enable_box=0 is enough to disengage it).
+    if (!enable)
+        return true;
+
+    if (params.ams_mapping.empty()) {
+        BOOST_LOG_TRIVIAL(warning) << "QidiPrinterAgent::apply_box_mapping: enable_box set but ams_mapping is empty";
+        return true;
+    }
+
+    // ams_mapping (v0) is a JSON array indexed by filament/tool; each value is the
+    // physical box slot (-1 = unmapped). Mirror it onto the printer's value_t<tool>
+    // variables: SAVE_VARIABLE VARIABLE=value_t<tool> VALUE='slot<n>'.
+    auto mapping = nlohmann::json::parse(params.ams_mapping, nullptr, /*allow_exceptions*/ false);
+    if (mapping.is_discarded() || !mapping.is_array()) {
+        BOOST_LOG_TRIVIAL(error) << "QidiPrinterAgent::apply_box_mapping: invalid ams_mapping: " << params.ams_mapping;
+        return false;
+    }
+
+    for (size_t tool = 0; tool < mapping.size(); ++tool) {
+        if (!mapping[tool].is_number_integer())
+            continue;
+        const int slot = mapping[tool].get<int>();
+        if (slot < 0)
+            continue; // unmapped filament — skip
+        const std::string gcode = "SAVE_VARIABLE VARIABLE=value_t" + std::to_string(tool) +
+                                  " VALUE=\"'slot" + std::to_string(slot) + "'\"";
+        if (!send_gcode(device_info.dev_id, gcode)) {
+            BOOST_LOG_TRIVIAL(error) << "QidiPrinterAgent::apply_box_mapping: failed to set value_t" << tool;
+            return false;
+        }
+    }
     return true;
 }
 
+int QidiPrinterAgent::start_local_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn)
+{
+    if (!apply_box_mapping(params))
+        return BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED;
+    return MoonrakerPrinterAgent::start_local_print(std::move(params), update_fn, cancel_fn);
+}
+
+int QidiPrinterAgent::start_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn, OnWaitFn wait_fn)
+{
+    if (!apply_box_mapping(params))
+        return BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED;
+    return MoonrakerPrinterAgent::start_print(std::move(params), update_fn, cancel_fn, wait_fn);
+}
+
+int QidiPrinterAgent::start_local_print_with_record(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn, OnWaitFn wait_fn)
+{
+    if (!apply_box_mapping(params))
+        return BAMBU_NETWORK_ERR_PRINT_WR_UPLOAD_FTP_FAILED;
+    return MoonrakerPrinterAgent::start_local_print_with_record(std::move(params), update_fn, cancel_fn, wait_fn);
+}
+
+int QidiPrinterAgent::start_sdcard_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn)
+{
+    if (!apply_box_mapping(params))
+        return BAMBU_NETWORK_ERR_PRINT_LP_PUBLISH_MSG_FAILED;
+    return MoonrakerPrinterAgent::start_sdcard_print(std::move(params), update_fn, cancel_fn);
+}
+
 bool QidiPrinterAgent::fetch_slot_info(const std::string&        base_url,
                                        const std::string&        api_key,
                                        const QidiFilamentDict&   dict,
@@ -120,20 +230,10 @@ bool QidiPrinterAgent::fetch_slot_info(const std::string&        base_url,
         return false;
     }
 
-    auto json = nlohmann::json::parse(response_body, nullptr, false, true);
+    nlohmann::json status;
-    if (json.is_discarded()) {
+    nlohmann::json variables;
-        error = "Invalid JSON response";
+    if (!parse_slot_response(response_body, status, variables, error))
         return false;
-    }
-
-    if (!json.contains("result") || !json["result"].contains("status") || !json["result"]["status"].contains("save_variables") ||
-        !json["result"]["status"]["save_variables"].contains("variables")) {
-        error = "Unexpected JSON structure";
-        return false;
-    }
-
-    auto& variables = json["result"]["status"]["save_variables"]["variables"];
-    auto& status    = json["result"]["status"];
 
     box_count = variables.value("box_count", 1);
     if (box_count < 0) {
@@ -208,6 +308,31 @@ bool QidiPrinterAgent::fetch_slot_info(const std::string&        base_url,
     return true;
 }
 
+bool QidiPrinterAgent::parse_slot_response(const std::string& response_body,
+                                           nlohmann::json&    status,
+                                           nlohmann::json&    variables,
+                                           std::string&       error)
+{
+    auto json = nlohmann::json::parse(response_body, nullptr, false, true);
+    if (json.is_discarded()) {
+        error = "Invalid JSON response";
+        return false;
+    }
+
+    if (!json.is_object() || !json.contains("result") || !json["result"].is_object() || !json["result"].contains("status") ||
+        !json["result"]["status"].is_object() || !json["result"]["status"].contains("save_variables") ||
+        !json["result"]["status"]["save_variables"].is_object() || !json["result"]["status"]["save_variables"].contains("variables") ||
+        !json["result"]["status"]["save_variables"]["variables"].is_object()) {
+        // why: Qidi firmware may send null here, but json::value() throws for it.
+        error = "Unexpected JSON structure: save_variables.variables must be an object";
+        return false;
+    }
+
+    status    = json["result"]["status"];
+    variables = status["save_variables"]["variables"];
+    return true;
+}
+
 bool QidiPrinterAgent::fetch_filament_dict(const std::string& base_url,
                                            const std::string& api_key,
                                            QidiFilamentDict& dict,

src/slic3r/Utils/QidiPrinterAgent.hpp

@@ -1,7 +1,9 @@
 #ifndef __QIDI_PRINTER_AGENT_HPP__
 #define __QIDI_PRINTER_AGENT_HPP__
 
+#include "IPrinterAgent.hpp"
 #include "MoonrakerPrinterAgent.hpp"
+#include "nlohmann/json_fwd.hpp"
 
 #include <map>
 #include <string>
@@ -21,7 +23,23 @@ public:
     // Override filament sync (Qidi-specific implementation)
     bool fetch_filament_info(std::string dev_id, FilamentSyncMode sync_mode = FilamentSyncMode::pull) override;
 
+    static bool parse_slot_response(const std::string& response_body,
+                                    nlohmann::json&    status,
+                                    nlohmann::json&    variables,
+                                    std::string&       error);
+
+    // Print operations — emit QiDi multi-color box config, then delegate to base.
+    int start_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn, OnWaitFn wait_fn) override;
+    int start_local_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn) override;
+    int start_local_print_with_record(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn, OnWaitFn wait_fn) override;
+    int start_sdcard_print(PrintParams params, OnUpdateStatusFn update_fn, WasCancelledFn cancel_fn) override;
+
+    FilamentSyncMode get_filament_sync_mode() const override;
+
 private:
+    // Push enable_box + value_t<tool> SAVE_VARIABLEs before a print starts.
+    // Returns false if any command fails (caller should abort the print).
+    bool apply_box_mapping(const PrintParams& params) const;
     struct QidiFilamentDict
     {
         std::map<int, std::string> colors;

src/slic3r/Utils/SnapmakerPrinterAgent.cpp

@@ -1,10 +1,14 @@
 #include "SnapmakerPrinterAgent.hpp"
 #include "Http.hpp"
+#include "IPrinterAgent.hpp"
 #include "libslic3r/PresetBundle.hpp"
 #include "slic3r/GUI/GUI_App.hpp"
 
 #include "nlohmann/json.hpp"
 #include <boost/log/trivial.hpp>
+#include <chrono>
+#include <sstream>
+#include <thread>
 
 using json = nlohmann::json;
 
@@ -13,6 +17,13 @@ namespace Slic3r {
 namespace {
 
 constexpr const char* SNAPMAKER_AGENT_VERSION = "0.0.1";
+constexpr int64_t     CAMERA_REFRESH_INTERVAL_MS = 300'000;
+
+int64_t now_ms()
+{
+    return std::chrono::duration_cast<std::chrono::milliseconds>(
+        std::chrono::steady_clock::now().time_since_epoch()).count();
+}
 
 // Safely access a parallel array by index, returning a fallback if out of bounds.
 template<typename T>
@@ -69,6 +80,31 @@ std::string find_closest_color_preset_by_vendor_and_type(const PresetCollection&
 
 SnapmakerPrinterAgent::SnapmakerPrinterAgent(std::string log_dir) : MoonrakerPrinterAgent(std::move(log_dir)) {}
 
+void SnapmakerPrinterAgent::start_camera_monitor()
+{
+    std::thread([this] {
+        send_ws_rpc("camera.start_monitor",
+                    {{"domain", "lan"}, {"interval", 0}, {"expect_pw", false}});
+    }).detach();
+    m_camera_last_fire_ms.store(now_ms());
+}
+
+void SnapmakerPrinterAgent::on_status_loop_tick(const std::string& dev_id)
+{
+    (void) dev_id;
+    const int64_t last = m_camera_last_fire_ms.load();
+    if (last == 0 || now_ms() - last >= CAMERA_REFRESH_INTERVAL_MS) {
+        start_camera_monitor();
+    }
+}
+
+int SnapmakerPrinterAgent::command_start_camera(std::string dev_id)
+{
+    (void) dev_id;
+    start_camera_monitor();
+    return BAMBU_NETWORK_SUCCESS;
+}
+
 AgentInfo SnapmakerPrinterAgent::get_agent_info_static()
 {
     return AgentInfo{"snapmaker", "Snapmaker", SNAPMAKER_AGENT_VERSION, "Snapmaker printer agent"};
@@ -104,127 +140,149 @@ std::string SnapmakerPrinterAgent::combine_filament_type(const std::string& type
     return base;
 }
 
-bool SnapmakerPrinterAgent::fetch_filament_info(std::string dev_id, FilamentSyncMode /*sync_mode*/)
+bool SnapmakerPrinterAgent::fetch_filament_info(std::string dev_id, FilamentSyncMode sync_mode)
 {
-    std::string url = join_url(device_info.base_url, "/printer/objects/query?print_task_config&filament_detect");
+    (void) dev_id;
-
+    if (sync_mode != get_filament_sync_mode())
-    std::string response_body;
-    bool        success = false;
-    std::string http_error;
-
-    auto http = Http::get(url);
-    if (!device_info.api_key.empty()) {
-        http.header("X-Api-Key", device_info.api_key);
-    }
-    http.timeout_connect(5)
-        .timeout_max(10)
-        .on_complete([&](std::string body, unsigned status) {
-            if (status == 200) {
-                response_body = body;
-                success       = true;
-            } else {
-                http_error = "HTTP error: " + std::to_string(status);
-            }
-        })
-        .on_error([&](std::string body, std::string err, unsigned status) {
-            http_error = err;
-            if (status > 0) {
-                http_error += " (HTTP " + std::to_string(status) + ")";
-            }
-        })
-        .perform_sync();
-
-    if (!success) {
-        BOOST_LOG_TRIVIAL(warning) << "SnapmakerPrinterAgent::fetch_filament_info: HTTP request failed: " << http_error;
         return false;
-    }
 
-    auto json = nlohmann::json::parse(response_body, nullptr, false, true);
+    const std::string base_url = device_info.base_url;
-    if (json.is_discarded()) {
+    const std::string api_key  = device_info.api_key;
-        BOOST_LOG_TRIVIAL(warning) << "SnapmakerPrinterAgent::fetch_filament_info: Invalid JSON response";
-        return false;
-    }
 
-    // Navigate to result.status.print_task_config
+    filament_fetch_in_flight.fetch_add(1, std::memory_order_relaxed);
-    if (!json.contains("result") || !json["result"].contains("status") ||
-        !json["result"]["status"].contains("print_task_config")) {
-        BOOST_LOG_TRIVIAL(warning) << "SnapmakerPrinterAgent::fetch_filament_info: Missing print_task_config in response";
-        return false;
-    }
 
-    auto& ptc = json["result"]["status"]["print_task_config"];
+    std::thread([this, base_url, api_key]() {
+        struct InFlightGuard
+        {
+            std::atomic<int>& counter;
+            ~InFlightGuard() { counter.fetch_sub(1, std::memory_order_relaxed); }
+        } guard{filament_fetch_in_flight};
 
-    // Read parallel arrays from print_task_config
+        const std::string url = join_url(base_url, "/printer/objects/query?print_task_config&filament_detect");
-    auto filament_exist    = ptc.value("filament_exist", std::vector<bool>{});
-    auto filament_type     = ptc.value("filament_type", std::vector<std::string>{});
-    auto filament_sub_type = ptc.value("filament_sub_type", std::vector<std::string>{});
-    auto filament_color    = ptc.value("filament_color_rgba", std::vector<std::string>{});
-    auto filament_vendor   = ptc.value("filament_vendor", std::vector<std::string>{});
 
-    const int slot_count = static_cast<int>(filament_exist.size());
+        std::string response_body;
-    if (slot_count == 0) {
+        bool success = false;
-        BOOST_LOG_TRIVIAL(info) << "SnapmakerPrinterAgent::fetch_filament_info: No filament slots reported";
+        std::string http_error;
-        return false;
-    }
 
-    // Read NFC filament_detect data for temperature info (optional)
+        auto http = Http::get(url);
-    nlohmann::json nfc_info;
+        if (!api_key.empty()) {
-    if (json["result"]["status"].contains("filament_detect") &&
+            http.header("X-Api-Key", api_key);
-        json["result"]["status"]["filament_detect"].contains("info")) {
+        }
-        nfc_info = json["result"]["status"]["filament_detect"]["info"];
+        http.timeout_connect(5)
-    }
+            .timeout_max(10)
-
+            .on_complete([&](std::string body, unsigned status) {
-    static const std::string empty_str;
+                if (status == 200) {
-    static const std::string default_color = "FFFFFFFF";
+                    response_body = body;
-
+                    success       = true;
-    std::vector<AmsTrayData> trays;
-    trays.reserve(slot_count);
-
-    for (int i = 0; i < slot_count; ++i) {
-        AmsTrayData tray;
-        tray.slot_index   = i;
-        tray.has_filament = filament_exist[i];
-
-        if (tray.has_filament) {
-            tray.tray_type     = combine_filament_type(safe_at(filament_type, i, empty_str),
-                                                       safe_at(filament_sub_type, i, empty_str));
-            tray.tray_color    = safe_at(filament_color, i, default_color);
-
-            auto* bundle = GUI::wxGetApp().preset_bundle;
-            // Try to find a matching preset for this filament based on vendor, type and color.
-            // If not found, default to traditional search by type only or generic type mapping.
-            if (bundle) {
-                std::string vendor      = safe_at(filament_vendor, i, empty_str);
-                std::string filament_id = find_closest_color_preset_by_vendor_and_type(bundle->filaments, vendor, tray.tray_type,
-                                                                                       tray.tray_color);
-
-                if (!filament_id.empty()) {
-                    tray.tray_info_idx = filament_id;
-                    BOOST_LOG_TRIVIAL(warning) << "Filament sync: Found manufacturer-specific profile for slot " << i << ": "
-                                               << filament_id;
                 } else {
-                    tray.tray_info_idx = bundle->filaments.filament_id_by_type(tray.tray_type);
+                    http_error = "HTTP error: " + std::to_string(status);
                 }
-            } else {
+            })
-                tray.tray_info_idx = map_filament_type_to_generic_id(tray.tray_type);
+            .on_error([&](std::string body, std::string err, unsigned status) {
-            }
+                http_error = err;
+                if (status > 0) {
+                    http_error += " (HTTP " + std::to_string(status) + ")";
+                }
+            })
+            .perform_sync();
 
-            // Extract NFC temperature data if available
+        if (!success) {
-            if (nfc_info.is_array() && i < static_cast<int>(nfc_info.size()) && nfc_info[i].is_object()) {
+            BOOST_LOG_TRIVIAL(warning) << "SnapmakerPrinterAgent::fetch_filament_info: HTTP request failed: " << http_error;
-                auto& nfc_slot = nfc_info[i];
+            return;
-                std::string vendor = nfc_slot.value("VENDOR", "NONE");
-                if (vendor != "NONE" && !vendor.empty()) {
-                    tray.bed_temp    = nfc_slot.value("BED_TEMP", 0);
-                    tray.nozzle_temp = nfc_slot.value("FIRST_LAYER_TEMP", 0);
-                }
-            }
         }
 
-        trays.emplace_back(std::move(tray));
+        auto json = nlohmann::json::parse(response_body, nullptr, false, true);
-    }
+        if (json.is_discarded()) {
+            BOOST_LOG_TRIVIAL(warning) << "SnapmakerPrinterAgent::fetch_filament_info: Invalid JSON response";
+            return;
+        }
+
+        // Navigate to result.status.print_task_config
+        if (!json.contains("result") || !json["result"].contains("status") || !json["result"]["status"].contains("print_task_config")) {
+            BOOST_LOG_TRIVIAL(warning) << "SnapmakerPrinterAgent::fetch_filament_info: Missing print_task_config in response";
+            return;
+        }
+
+        auto& ptc = json["result"]["status"]["print_task_config"];
+
+        // Read parallel arrays from print_task_config
+        auto filament_exist    = ptc.value("filament_exist", std::vector<bool>{});
+        auto filament_type     = ptc.value("filament_type", std::vector<std::string>{});
+        auto filament_sub_type = ptc.value("filament_sub_type", std::vector<std::string>{});
+        auto filament_color    = ptc.value("filament_color_rgba", std::vector<std::string>{});
+        auto filament_vendor   = ptc.value("filament_vendor", std::vector<std::string>{});
+
+        const int slot_count = static_cast<int>(filament_exist.size());
+        if (slot_count == 0) {
+            BOOST_LOG_TRIVIAL(info) << "SnapmakerPrinterAgent::fetch_filament_info: No filament slots reported";
+            return;
+        }
+
+        // Read NFC filament_detect data for temperature info (optional)
+        nlohmann::json nfc_info;
+        if (json["result"]["status"].contains("filament_detect") && json["result"]["status"]["filament_detect"].contains("info")) {
+            nfc_info = json["result"]["status"]["filament_detect"]["info"];
+        }
+
+        static const std::string empty_str;
+        static const std::string default_color = "FFFFFFFF";
+
+        std::vector<AmsTrayData> trays;
+        trays.reserve(slot_count);
+
+        for (int i = 0; i < slot_count; ++i) {
+            AmsTrayData tray;
+            tray.slot_index   = i;
+            tray.has_filament = filament_exist[i];
+
+            if (tray.has_filament) {
+                tray.tray_type  = combine_filament_type(safe_at(filament_type, i, empty_str), safe_at(filament_sub_type, i, empty_str));
+                tray.tray_color = safe_at(filament_color, i, default_color);
+
+                auto* bundle = GUI::wxGetApp().preset_bundle;
+                // Try to find a matching preset for this filament based on vendor, type and color.
+                // If not found, default to traditional search by type only or generic type mapping.
+                if (bundle) {
+                    std::string vendor      = safe_at(filament_vendor, i, empty_str);
+                    std::string filament_id = find_closest_color_preset_by_vendor_and_type(bundle->filaments, vendor, tray.tray_type,
+                                                                                           tray.tray_color);
+
+                    if (!filament_id.empty()) {
+                        tray.tray_info_idx = filament_id;
+                        BOOST_LOG_TRIVIAL(warning)
+                            << "Filament sync: Found manufacturer-specific profile for slot " << i << ": " << filament_id;
+                    } else {
+                        tray.tray_info_idx = bundle->filaments.filament_id_by_type(tray.tray_type);
+                    }
+                } else {
+                    tray.tray_info_idx = map_filament_type_to_generic_id(tray.tray_type);
+                }
+
+                // Extract NFC temperature data if available
+                if (nfc_info.is_array() && i < static_cast<int>(nfc_info.size()) && nfc_info[i].is_object()) {
+                    auto& nfc_slot     = nfc_info[i];
+                    std::string vendor = nfc_slot.value("VENDOR", "NONE");
+                    if (vendor != "NONE" && !vendor.empty()) {
+                        tray.bed_temp    = nfc_slot.value("BED_TEMP", 0);
+                        tray.nozzle_temp = nfc_slot.value("FIRST_LAYER_TEMP", 0);
+                    }
+                }
+            }
+
+            trays.emplace_back(std::move(tray));
+        }
+
+        build_ams_payload(1, slot_count - 1, trays);
+    }).detach();
 
-    build_ams_payload(1, slot_count - 1, trays);
     return true;
 }
 
+FilamentSyncMode SnapmakerPrinterAgent::get_filament_sync_mode() const
+{
+    if (GUI::wxGetApp().app_config->get_bool("use_printer_agents"))
+        return FilamentSyncMode::subscription;
+    return FilamentSyncMode::pull;
+}
+
 } // namespace Slic3r

src/slic3r/Utils/SnapmakerPrinterAgent.hpp

@@ -1,7 +1,10 @@
 #pragma once
 
+#include "IPrinterAgent.hpp"
 #include "MoonrakerPrinterAgent.hpp"
 
+#include <atomic>
+#include <cstdint>
 #include <string>
 
 namespace Slic3r {
@@ -16,10 +19,19 @@ public:
     AgentInfo        get_agent_info() override { return get_agent_info_static(); }
 
     bool fetch_filament_info(std::string dev_id, FilamentSyncMode sync_mode = FilamentSyncMode::pull) override;
+    FilamentSyncMode get_filament_sync_mode() const override;
+    int command_start_camera(std::string dev_id) override;
+    CameraStreamMode get_camera_stream_mode() const override { return CameraStreamMode::http_snapshot; }
+    std::string get_camera_url() const override { return device_info.base_url + "/server/files/camera/monitor.jpg"; }
 
 private:
     // Combine filament_type + filament_sub_type into a unified type string
     static std::string combine_filament_type(const std::string& type, const std::string& sub_type);
+
+    void start_camera_monitor();
+    void on_status_loop_tick(const std::string& dev_id) override;
+
+    std::atomic<int64_t> m_camera_last_fire_ms{0};
 };
 
 } // namespace Slic3r

tests/slic3rutils/CMakeLists.txt

@@ -14,6 +14,9 @@ add_executable(${_TEST_NAME}_tests
     test_plugin_capabilities_in_use.cpp
     test_plugin_status.cpp
     test_printer_agent.cpp
+    test_qidi_printer_agent.cpp
+    test_orca_mqtt_connection.cpp
+    test_orca_printer_agent.cpp
     test_plugin_install.cpp
     test_plugin_lifecycle.cpp
     test_slicing_pipeline_bindings.cpp

tests/slic3rutils/orca_mqtt_mock_broker.hpp

@@ -0,0 +1,317 @@
+#pragma once
+
+// In-process plaintext MQTT-over-WebSocket broker for the OrcaMqtt tests.
+//
+// It speaks just enough of MQTT 3.1.1 to drive OrcaMqttConnection /
+// OrcaPrinterAgent end to end without a real network: CONNECT/CONNACK,
+// SUBSCRIBE/SUBACK, UNSUBSCRIBE/UNSUBACK, client PUBLISH (QoS 0), PINGREQ and
+// DISCONNECT. The outbound PUBLISH frame is built with the production
+// OrcaMqttConnection::make_publish_packet() so the tests never depend on a
+// second, hand-rolled MQTT encoder.
+
+#include <slic3r/Utils/OrcaMqttConnection.hpp>
+
+#include <boost/asio.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/beast/websocket.hpp>
+
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+namespace orca_mqtt_test {
+
+namespace net   = boost::asio;
+namespace beast = boost::beast;
+namespace ws    = boost::beast::websocket;
+using tcp       = boost::asio::ip::tcp;
+
+// Decode an MQTT remaining-length varint starting at packet[offset].
+// Returns {value, bytes_consumed}; bytes_consumed == 0 means malformed.
+inline std::pair<std::size_t, std::size_t> mqtt_decode_remaining_length(const std::string& packet, std::size_t offset)
+{
+    std::size_t value      = 0;
+    std::size_t multiplier = 1;
+    std::size_t used       = 0;
+    while (offset + used < packet.size() && used < 4) {
+        const std::uint8_t byte = static_cast<std::uint8_t>(packet[offset + used]);
+        value += static_cast<std::size_t>(byte & 0x7f) * multiplier;
+        multiplier *= 128;
+        ++used;
+        if ((byte & 0x80) == 0)
+            return {value, used};
+    }
+    return {0, 0};
+}
+
+inline bool mqtt_topic_is_request(const std::string& topic)
+{
+    static const std::string suffix = "/request";
+    return topic.size() >= suffix.size() &&
+           topic.compare(topic.size() - suffix.size(), suffix.size(), suffix) == 0;
+}
+
+class MockBroker
+{
+public:
+    // refuse_auth: answer every CONNECT with CONNACK rc 5 (not authorized) and
+    // close, so the reconnect/refusal paths can be exercised.
+    explicit MockBroker(bool refuse_auth = false) : m_refuse_auth(refuse_auth), m_acceptor(m_io)
+    {
+        const tcp::endpoint endpoint(net::ip::make_address("127.0.0.1"), 0);
+        m_acceptor.open(endpoint.protocol());
+        m_acceptor.set_option(net::socket_base::reuse_address(true));
+        m_acceptor.bind(endpoint);
+        m_acceptor.listen(net::socket_base::max_listen_connections);
+        m_port = std::to_string(m_acceptor.local_endpoint().port());
+        // why: a non-blocking acceptor lets the accept loop poll a stop flag, so
+        // the destructor never has to interrupt a blocking accept().
+        m_acceptor.non_blocking(true);
+        m_thread = std::thread([this] { run(); });
+    }
+
+    ~MockBroker()
+    {
+        m_stopping.store(true);
+        drop_client();                     // unblocks the worker's blocking read
+        if (m_thread.joinable())
+            m_thread.join();
+        boost::system::error_code ec;
+        m_acceptor.close(ec);              // after join: the acceptor is worker-owned
+        m_io.stop();
+    }
+
+    MockBroker(const MockBroker&)            = delete;
+    MockBroker& operator=(const MockBroker&) = delete;
+
+    std::string ws_url() const { return "ws://127.0.0.1:" + m_port + "/mqtt"; }
+
+    std::pair<std::string, std::string> host_port() const { return {std::string("127.0.0.1"), m_port}; }
+
+    // Server -> client PUBLISH on device/<dev_id>/report.
+    void push_report(const std::string& dev_id, const std::string& payload)
+    {
+        const std::vector<std::uint8_t> packet =
+            Slic3r::OrcaMqttConnection::make_publish_packet("device/" + dev_id + "/report", payload);
+        std::lock_guard<std::mutex> lock(m_mutex);
+        if (!m_stream || !m_stream_ready)
+            return;
+        boost::system::error_code ec;
+        m_stream->binary(true);
+        m_stream->write(net::buffer(packet), ec); // a vanished client is not a test failure
+    }
+
+    // Force-close the live client socket; the worker's read returns an error and
+    // the accept loop picks up the client's reconnect.
+    void drop_client()
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        close_client_locked();
+    }
+
+    // Payloads the client PUBLISHed to any device/<id>/request topic.
+    std::vector<std::string> received_requests() const
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        return m_received_requests;
+    }
+
+    // MQTT CONNECTs seen; increments again after a reconnect.
+    int connect_count() const { return m_connect_count.load(); }
+
+private:
+    void run()
+    {
+        try {
+            while (!m_stopping.load()) {
+                tcp::socket               socket(m_io);
+                boost::system::error_code ec;
+                m_acceptor.accept(socket, ec);
+                if (ec == net::error::would_block || ec == net::error::try_again) {
+                    std::this_thread::sleep_for(std::chrono::milliseconds(5));
+                    continue;
+                }
+                if (ec)
+                    return;
+                try {
+                    serve(std::move(socket));
+                } catch (...) {
+                    // a client dying mid-session must not take the broker down
+                }
+                std::lock_guard<std::mutex> lock(m_mutex);
+                close_client_locked();
+                m_stream.reset();
+            }
+        } catch (...) {
+            // never let an exception escape the broker thread
+        }
+    }
+
+    void serve(tcp::socket socket)
+    {
+        ws::stream<beast::tcp_stream>* stream = nullptr;
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_stream.emplace(std::move(socket));
+            m_stream_ready = false;
+            stream         = &*m_stream;
+        }
+        // why: no io_context is ever run here, so a tcp_stream timer would never
+        // fire; the sync operations below carry no timeout of their own.
+        beast::get_lowest_layer(*stream).expires_never();
+        stream->set_option(ws::stream_base::decorator(
+            [](ws::response_type& res) { res.set("Sec-WebSocket-Protocol", "mqtt"); }));
+
+        boost::system::error_code ec;
+        stream->accept(ec);
+        if (ec)
+            return;
+        stream->binary(true);
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_stream_ready = true;
+        }
+        read_loop(*stream);
+    }
+
+    // The client sends every MQTT packet as one binary WebSocket message, so one
+    // read yields exactly one packet.
+    void read_loop(ws::stream<beast::tcp_stream>& stream)
+    {
+        beast::flat_buffer buffer;
+        while (!m_stopping.load()) {
+            boost::system::error_code ec;
+            buffer.clear();
+            stream.read(buffer, ec);
+            if (ec)
+                return;
+            const std::string packet = beast::buffers_to_string(buffer.data());
+            if (packet.empty())
+                continue;
+            if (!handle_packet(stream, packet))
+                return;
+        }
+    }
+
+    // Returns false when the session must be closed.
+    bool handle_packet(ws::stream<beast::tcp_stream>& stream, const std::string& packet)
+    {
+        switch (static_cast<std::uint8_t>(packet[0]) & 0xf0) {
+        case 0x10: { // CONNECT
+            ++m_connect_count;
+            if (m_refuse_auth) {
+                write_packet(stream, {0x20, 0x02, 0x00, 0x05}); // CONNACK not authorized
+                return false;
+            }
+            write_packet(stream, {0x20, 0x02, 0x00, 0x00}); // CONNACK accepted
+            return true;
+        }
+        case 0x80: { // SUBSCRIBE (0x82) - packet id follows the remaining-length varint
+            const auto id = packet_id(packet);
+            if (id)
+                write_packet(stream, {0x90, 0x03, id->first, id->second, 0x00}); // SUBACK, QoS 0
+            return true;
+        }
+        case 0xa0: { // UNSUBSCRIBE (0xa2)
+            const auto id = packet_id(packet);
+            if (id)
+                write_packet(stream, {0xb0, 0x02, id->first, id->second}); // UNSUBACK
+            return true;
+        }
+        case 0x30: { // PUBLISH, QoS 0 (no packet identifier)
+            record_publish(packet);
+            return true;
+        }
+        case 0xc0: // PINGREQ
+            write_packet(stream, {0xd0, 0x00});
+            return true;
+        case 0xe0: // DISCONNECT
+            return false;
+        default:
+            return true;
+        }
+    }
+
+    // The two packet-identifier bytes sitting right after the remaining-length varint.
+    static std::optional<std::pair<std::uint8_t, std::uint8_t>> packet_id(const std::string& packet)
+    {
+        const auto varint = mqtt_decode_remaining_length(packet, 1);
+        if (varint.second == 0)
+            return std::nullopt;
+        const std::size_t pos = 1 + varint.second;
+        if (pos + 2 > packet.size())
+            return std::nullopt;
+        return std::make_pair(static_cast<std::uint8_t>(packet[pos]), static_cast<std::uint8_t>(packet[pos + 1]));
+    }
+
+    void record_publish(const std::string& packet)
+    {
+        const auto varint = mqtt_decode_remaining_length(packet, 1);
+        if (varint.second == 0)
+            return;
+        std::size_t pos = 1 + varint.second;
+        if (pos + 2 > packet.size())
+            return;
+        const std::size_t topic_len = (static_cast<std::size_t>(static_cast<std::uint8_t>(packet[pos])) << 8) |
+                                      static_cast<std::uint8_t>(packet[pos + 1]);
+        pos += 2;
+        if (pos + topic_len > packet.size())
+            return;
+        const std::string topic = packet.substr(pos, topic_len);
+        pos += topic_len;
+        const std::size_t end = std::min(packet.size(), 1 + varint.second + varint.first);
+        if (end < pos)
+            return;
+        if (!mqtt_topic_is_request(topic))
+            return;
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_received_requests.push_back(packet.substr(pos, end - pos));
+    }
+
+    // Every write - the worker's own replies and push_report() from the test
+    // thread - is serialised by m_mutex. beast permits a writer while the worker
+    // is blocked in read(), which is the same arrangement OrcaMqttConnection uses.
+    void write_packet(ws::stream<beast::tcp_stream>& stream, const std::vector<std::uint8_t>& packet)
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        boost::system::error_code   ec;
+        stream.binary(true);
+        stream.write(net::buffer(packet), ec);
+    }
+
+    void close_client_locked()
+    {
+        if (!m_stream)
+            return;
+        m_stream_ready = false;
+        boost::system::error_code ec;
+        auto&                     socket = beast::get_lowest_layer(*m_stream).socket();
+        socket.cancel(ec);
+        // shutdown() before close() is what actually wakes a blocking read on the
+        // worker thread; close() alone does not on POSIX.
+        socket.shutdown(tcp::socket::shutdown_both, ec);
+        socket.close(ec);
+    }
+
+    const bool                                   m_refuse_auth;
+    net::io_context                              m_io;
+    tcp::acceptor                                m_acceptor;
+    std::string                                  m_port;
+    std::thread                                  m_thread;
+    std::atomic_bool                             m_stopping{false};
+    std::atomic<int>                             m_connect_count{0};
+    mutable std::mutex                           m_mutex;
+    std::optional<ws::stream<beast::tcp_stream>> m_stream;      // guarded by m_mutex
+    bool                                         m_stream_ready = false; // guarded by m_mutex
+    std::vector<std::string>                     m_received_requests;    // guarded by m_mutex
+};
+
+} // namespace orca_mqtt_test

tests/slic3rutils/test_orca_mqtt_connection.cpp

@@ -0,0 +1,229 @@
+#include <catch2/catch_test_macros.hpp>
+#include <slic3r/Utils/OrcaMqttConnection.hpp>
+
+#include "orca_mqtt_mock_broker.hpp"
+
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <cstdint>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+using Slic3r::OrcaMqttConnection;
+
+// Offset of the CONNECT variable header: 1 (fixed header) + N remaining-length varint bytes.
+static size_t mqtt_varheader_offset(const std::vector<uint8_t>& p) {
+    size_t i = 1;
+    while (i < p.size() && (p[i] & 0x80)) ++i;   // skip varint continuation bytes
+    return i + 1;                                // + the final varint byte
+}
+
+TEST_CASE("OrcaMqtt parse_endpoint handles ws and wss", "[OrcaMqtt]") {
+    OrcaMqttConnection::Endpoint ep;
+
+    REQUIRE(OrcaMqttConnection::parse_endpoint("ws://printer.local:8280/mqtt", ep));
+    CHECK(ep.host   == "printer.local");
+    CHECK(ep.port   == "8280");
+    CHECK(ep.target == "/mqtt");
+
+    REQUIRE(OrcaMqttConnection::parse_endpoint("ws://10.0.0.5/mqtt", ep));
+    CHECK(ep.port == "80");
+
+    REQUIRE(OrcaMqttConnection::parse_endpoint("wss://api.example.com/api/v1/printers/abc/mqtt", ep));
+    CHECK(ep.host   == "api.example.com");
+    CHECK(ep.port   == "443");
+    CHECK(ep.target == "/api/v1/printers/abc/mqtt");
+
+    CHECK_FALSE(OrcaMqttConnection::parse_endpoint("http://x/y", ep));
+}
+
+TEST_CASE("OrcaMqtt CONNECT packet - no auth (cloud form)", "[OrcaMqtt]") {
+    auto p = OrcaMqttConnection::make_connect_packet("OrcaSlicer", "", "", 300);
+    REQUIRE(p.size() >= 12);
+    CHECK(p[0] == 0x10);                              // CONNECT fixed header
+    const size_t v = mqtt_varheader_offset(p);
+    CHECK(p[v + 0] == 0x00); CHECK(p[v + 1] == 0x04); // protocol name length
+    CHECK(p[v + 2] == 'M'); CHECK(p[v + 3] == 'Q');
+    CHECK(p[v + 4] == 'T'); CHECK(p[v + 5] == 'T');
+    CHECK(p[v + 6] == 0x04);                          // protocol level 3.1.1
+    CHECK(p[v + 7] == 0x02);                          // connect flags: clean session only
+    CHECK(((p[v + 8] << 8) | p[v + 9]) == 300);       // keepalive
+}
+
+TEST_CASE("OrcaMqtt CONNECT packet - username/password (LAN form)", "[OrcaMqtt]") {
+    auto p = OrcaMqttConnection::make_connect_packet("orcaslicer-lan-x", "orcasonar", "code123", 60);
+    CHECK(p[0] == 0x10);
+    const size_t v = mqtt_varheader_offset(p);
+    CHECK(p[v + 7] == (0x02 | 0x80 | 0x40));          // clean session + username + password flags
+    const std::string blob(p.begin(), p.end());
+    CHECK(blob.find("orcaslicer-lan-x") != std::string::npos);
+    CHECK(blob.find("orcasonar")        != std::string::npos);
+    CHECK(blob.find("code123")          != std::string::npos);
+}
+
+// Auth precedence (spec O3): when a bearer_provider is configured, connect_and_read
+// passes empty CONNECT credentials, so the packet must carry clean-session only and
+// no username/password flags or payload fields. (The precedence branch itself lives
+// in connect_and_read; the [.integration] cloud-style round trip exercises it live.)
+TEST_CASE("OrcaMqtt CONNECT omits creds when a bearer is configured", "[OrcaMqtt]") {
+    auto p = OrcaMqttConnection::make_connect_packet("cid", "", "", 60);
+    const size_t v = mqtt_varheader_offset(p);
+    CHECK(p[v + 7] == 0x02);                       // clean session only: no 0x80 / 0x40
+    const std::string blob(p.begin(), p.end());
+    CHECK(blob.find("orcasonar") == std::string::npos);
+}
+
+TEST_CASE("OrcaMqtt topic helpers", "[OrcaMqtt]") {
+    CHECK(OrcaMqttConnection::request_topic("abc") == "device/abc/request");
+    CHECK(OrcaMqttConnection::report_topic("abc")  == "device/abc/report");
+}
+
+TEST_CASE("OrcaMqtt PUBLISH packet QoS0", "[OrcaMqtt]") {
+    auto p = OrcaMqttConnection::make_publish_packet("device/abc/request", "{\"ok\":1}");
+    CHECK((p[0] & 0xf0) == 0x30);   // PUBLISH
+    CHECK((p[0] & 0x06) == 0x00);   // QoS 0
+    const std::string blob(p.begin(), p.end());
+    CHECK(blob.find("device/abc/request") != std::string::npos);
+    CHECK(blob.find("{\"ok\":1}")          != std::string::npos);
+}
+
+TEST_CASE("OrcaMqtt SUBSCRIBE packet", "[OrcaMqtt]") {
+    auto p = OrcaMqttConnection::make_subscribe_packet(7, "device/abc/report", 1);
+    CHECK(p[0] == 0x82);                          // SUBSCRIBE + reserved bit
+    const size_t v = mqtt_varheader_offset(p);
+    CHECK(((p[v] << 8) | p[v + 1]) == 7);         // packet id
+    CHECK(p.back() == 1);                         // requested QoS
+}
+
+TEST_CASE("OrcaMqtt send_request refuses when not connected", "[OrcaMqtt]") {
+    OrcaMqttConnection conn;
+    CHECK_FALSE(conn.send_request("abc", "{\"pushing\":{\"command\":\"pushall\",\"sequence_id\":\"20001\"}}"));
+}
+
+TEST_CASE("OrcaMqtt start takes a Config", "[OrcaMqtt]") {
+    OrcaMqttConnection conn;
+    OrcaMqttConnection::Config cfg;
+    cfg.url = "ws://127.0.0.1:1/mqtt";           // nothing listening
+    cfg.keepalive_seconds = 42;
+    // start() returns false (no server) but must compile with the Config overload
+    const bool ok = conn.start(cfg, [](auto, auto){}, [](bool, bool){});
+    CHECK_FALSE(ok);
+    CHECK(conn.last_connack_rc() == -1);
+    conn.stop();
+}
+
+TEST_CASE("MockBroker starts and reports a url", "[OrcaMqtt][.integration]") {
+    orca_mqtt_test::MockBroker b;
+    CHECK(b.ws_url().rfind("ws://127.0.0.1:", 0) == 0);
+    CHECK(b.connect_count() == 0);
+}
+
+// --- End-to-end integration: OrcaMqttConnection against the in-process MockBroker.
+// All hidden behind [.integration] (run explicitly). These prove a LAN-style config
+// (CONNECT username/password) and a cloud-style config (bearer on the WS upgrade,
+// no CONNECT creds) drive the *same* OrcaMqttConnection code path with identical
+// assertions.
+
+static void run_round_trip(bool use_tls_flag_only) {
+    orca_mqtt_test::MockBroker broker;
+    OrcaMqttConnection conn;
+    OrcaMqttConnection::Config cfg;
+    cfg.url = broker.ws_url();                     // plaintext regardless
+    cfg.use_tls = false;                           // the mock is plaintext; the flag path is unit-tested elsewhere
+    if (use_tls_flag_only) cfg.bearer_provider = []{ return std::string("tok"); };
+    else { cfg.username = "orcasonar"; cfg.password = "code"; }
+
+    // A mutex + condition_variable rather than a promise: the handler runs on the MQTT
+    // worker thread and a second inbound message would throw std::future_error there.
+    std::mutex              got_mutex;
+    std::condition_variable got_cv;
+    bool                    got_any = false;
+    std::string             got_id, got_payload;
+
+    REQUIRE(conn.start(cfg,
+        [&](const std::string& id, const std::string& payload){
+            {
+                std::lock_guard<std::mutex> l(got_mutex);
+                if (got_any) return;              // keep the first message only
+                got_any = true; got_id = id; got_payload = payload;
+            }
+            got_cv.notify_all();
+        },
+        [](bool,bool){}));
+    REQUIRE(conn.subscribe("dev-1"));
+    REQUIRE(conn.send_request("dev-1", R"({"pushing":{"command":"pushall","sequence_id":"20001"}})"));
+
+    broker.push_report("dev-1", R"({"print":{"command":"push_status","sequence_id":"20001","result":"success"}})");
+    std::string id, payload;
+    {
+        std::unique_lock<std::mutex> l(got_mutex);
+        REQUIRE(got_cv.wait_for(l, std::chrono::seconds(3), [&]{ return got_any; }));
+        id = got_id; payload = got_payload;
+    }
+    CHECK(id == "dev-1");
+    CHECK(payload.find("push_status") != std::string::npos);
+
+    // the client's command reached the broker on the request topic. The mock records
+    // the PUBLISH on its own read-loop thread, so poll rather than check immediately.
+    std::vector<std::string> reqs;
+    for (int i = 0; i < 200; ++i) {
+        reqs = broker.received_requests();
+        if (!reqs.empty()) break;
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    REQUIRE(reqs.size() >= 1);
+    CHECK(reqs.front().find("pushall") != std::string::npos);
+    conn.stop();
+}
+
+TEST_CASE("OrcaMqtt round-trip — LAN-style config",   "[OrcaMqtt][.integration]") { run_round_trip(false); }
+TEST_CASE("OrcaMqtt round-trip — cloud-style config", "[OrcaMqtt][.integration]") { run_round_trip(true);  }
+
+TEST_CASE("OrcaMqtt reconnects and re-subscribes after a socket drop", "[OrcaMqtt][.integration]") {
+    orca_mqtt_test::MockBroker broker;
+    OrcaMqttConnection conn;
+    OrcaMqttConnection::Config cfg; cfg.url = broker.ws_url(); cfg.use_tls = false; cfg.username = "u"; cfg.password = "p";
+
+    std::mutex m; std::vector<std::string> got;
+    REQUIRE(conn.start(cfg,
+        [&](const std::string&, const std::string& p){ std::lock_guard<std::mutex> l(m); got.push_back(p); },
+        [](bool,bool){}));
+    REQUIRE(conn.subscribe("dev-1"));
+
+    broker.drop_client();
+    // the worker reconnects with ~1s backoff
+    for (int i = 0; i < 300 && broker.connect_count() < 2; ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    CHECK(broker.connect_count() >= 2);
+
+    // a report after the reconnect must still be delivered -> the SUBSCRIBE was re-sent
+    broker.push_report("dev-1", R"({"print":{"command":"push_status","sequence_id":"20002"}})");
+    bool delivered = false;
+    for (int i = 0; i < 200 && !delivered; ++i) {
+        { std::lock_guard<std::mutex> l(m); delivered = !got.empty(); }
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    CHECK(delivered);
+    conn.stop();
+}
+
+TEST_CASE("OrcaMqtt auth rejection is terminal (no retry storm)", "[OrcaMqtt][.integration]") {
+    orca_mqtt_test::MockBroker broker(/*refuse_auth=*/true);
+    OrcaMqttConnection conn;
+    OrcaMqttConnection::Config cfg; cfg.url = broker.ws_url(); cfg.use_tls = false; cfg.username = "u"; cfg.password = "bad";
+
+    const bool ok = conn.start(cfg, [](const std::string&, const std::string&){}, [](bool,bool){});
+    CHECK_FALSE(ok);
+    CHECK(conn.last_connack_rc() == 5);
+    // worker must have stopped itself (rc 5 is terminal) — give it a moment
+    for (int i = 0; i < 100 && conn.is_running(); ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    CHECK_FALSE(conn.is_running());
+    // and it must NOT have hammered the broker with retries
+    std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    CHECK(broker.connect_count() <= 2);
+    conn.stop();
+}

tests/slic3rutils/test_orca_printer_agent.cpp

@@ -0,0 +1,204 @@
+#include <catch2/catch_test_macros.hpp>
+#include <slic3r/Utils/OrcaCloudServiceAgent.hpp>
+#include <slic3r/Utils/OrcaPrinterAgent.hpp>
+
+#include "orca_mqtt_mock_broker.hpp"
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+using Slic3r::OrcaPrinterAgent;
+
+namespace {
+// Probe exposes the protected internals the tests drive.
+struct Probe : OrcaPrinterAgent {
+    using OrcaPrinterAgent::OrcaPrinterAgent;
+    using OrcaPrinterAgent::deliver_to_sink;
+    using OrcaPrinterAgent::parse_lan_endpoint;
+    using OrcaPrinterAgent::make_lan_client_id;
+    using OrcaPrinterAgent::lan_connection_target;
+};
+}
+
+TEST_CASE("OrcaPrinterAgent forwards a status payload to on_message_fn", "[OrcaPrinterAgent]") {
+    Probe agent("/tmp");
+    std::string got_id, got_payload;
+    agent.set_on_message_fn([&](std::string id, std::string p){ got_id = std::move(id); got_payload = std::move(p); });
+    agent.deliver_to_sink("dev-1", R"({"print":{"command":"push_status"}})", /*local=*/false);
+    CHECK(got_id == "dev-1");
+    CHECK(got_payload.find("push_status") != std::string::npos);
+}
+
+TEST_CASE("OrcaPrinterAgent stamps the get_capabilities nozzle diameter onto push_status frames", "[OrcaPrinterAgent]") {
+    Probe agent("/tmp");
+    std::string last_payload;
+    agent.set_on_message_fn([&](std::string, std::string p){ last_payload = std::move(p); });
+
+    // Before any capabilities reply, a push_status frame is forwarded untouched.
+    agent.deliver_to_sink("dev-1", R"({"print":{"command":"push_status","mc_percent":10}})", /*local=*/false);
+    CHECK(last_payload.find("nozzle_diameter") == std::string::npos);
+
+    // The get_capabilities reply is forwarded verbatim; its topology nozzle diameter
+    // is cached for the device.
+    agent.deliver_to_sink(
+        "dev-1",
+        R"({"info":{"command":"get_capabilities","capabilities":{"topology":{"tools":[{"id":"T0","nozzle":{"diameter_mm":0.4}}]}}}})",
+        /*local=*/false);
+    CHECK(last_payload.find("\"command\":\"get_capabilities\"") != std::string::npos);
+    CHECK(last_payload.find("\"print\"") == std::string::npos);
+
+    // Later push_status frames for that device get the cached diameter plus a neutral
+    // nozzle_type, so MachineObject::parse_json's legacy nozzle parser can run.
+    agent.deliver_to_sink("dev-1", R"({"print":{"command":"push_status","mc_percent":20}})", /*local=*/false);
+    CHECK(last_payload.find("\"nozzle_diameter\":0.4") != std::string::npos);
+    CHECK(last_payload.find("\"nozzle_type\":\"N/A\"") != std::string::npos);
+
+    // A different device is unaffected.
+    agent.deliver_to_sink("dev-2", R"({"print":{"command":"push_status"}})", /*local=*/false);
+    CHECK(last_payload.find("nozzle_diameter") == std::string::npos);
+
+    // A frame that already carries real nozzle data is not overridden.
+    agent.deliver_to_sink("dev-1", R"({"print":{"command":"push_status","nozzle_diameter":0.6}})", /*local=*/false);
+    CHECK(last_payload.find("\"nozzle_diameter\":0.6") != std::string::npos);
+    CHECK(last_payload.find("N/A") == std::string::npos);
+}
+
+TEST_CASE("OrcaPrinterAgent::parse_lan_endpoint", "[OrcaPrinterAgent]") {
+    std::string h, p;
+    REQUIRE(Probe::parse_lan_endpoint("192.168.1.9", h, p));
+    CHECK(h == "192.168.1.9"); CHECK(p == "8280");
+    REQUIRE(Probe::parse_lan_endpoint("http://host.local:9000/x", h, p));
+    CHECK(h == "host.local"); CHECK(p == "9000");
+    CHECK_FALSE(Probe::parse_lan_endpoint("", h, p));
+}
+
+TEST_CASE("OrcaPrinterAgent::make_lan_client_id is stable and prefixed", "[OrcaPrinterAgent]") {
+    const auto a = Probe::make_lan_client_id("dev-1");
+    const auto b = Probe::make_lan_client_id("dev-1");
+    CHECK(a == b);                                   // drawn once per process
+    CHECK(a.rfind("orcaslicer-lan-dev-1-", 0) == 0);
+}
+
+TEST_CASE("connect_printer wires up a LAN Config", "[OrcaPrinterAgent][.integration]") {
+    Probe agent("/tmp");
+    const int rc = agent.connect_printer("dev-1", "10.255.255.1", "orcasonar", "code", false);
+    CHECK(rc == BAMBU_NETWORK_SUCCESS);
+    CHECK(agent.lan_connection_target() == "ws://10.255.255.1:8280/mqtt");
+    CHECK(agent.get_user_selected_machine().empty());   // LAN path must not touch the cloud selection
+    agent.disconnect_printer();
+}
+
+TEST_CASE("post-connect sequence is subscribe then 4 requests in order", "[OrcaPrinterAgent]") {
+    struct SeqProbe : OrcaPrinterAgent {
+        using OrcaPrinterAgent::OrcaPrinterAgent;
+        std::vector<std::string> calls;
+        void emit_connect_sequence(const std::string& dev_id,
+            std::function<void(const std::string&)> /*sub*/,
+            std::function<void(const std::string&)> /*req*/) override {
+            OrcaPrinterAgent::emit_connect_sequence(dev_id,
+                [&](const std::string& id){ calls.push_back("sub:" + id); },
+                [&](const std::string& body){ calls.push_back(body); });
+        }
+    } probe("/tmp");
+    probe.run_connect_sequence_for_test("dev-1");
+    REQUIRE(probe.calls.size() == 5);
+    CHECK(probe.calls[0] == "sub:dev-1");
+    CHECK(probe.calls[1].find("\"pushing\"") != std::string::npos);
+    CHECK(probe.calls[1].find("\"start\"")   != std::string::npos);
+    CHECK(probe.calls[2].find("pushall")     != std::string::npos);
+    CHECK(probe.calls[3].find("get_version") != std::string::npos);
+    CHECK(probe.calls[4].find("get_capabilities") != std::string::npos);
+    for (auto& c : probe.calls)
+        if (auto pos = c.find("sequence_id"); pos != std::string::npos)
+            CHECK(c.substr(pos).find("\"2") != std::string::npos);
+}
+
+// Hidden: spawns the connect worker and attempts a real (failing) connect.
+TEST_CASE("selecting a cloud printer configures the fleet socket", "[OrcaPrinterAgent][.integration]") {
+    auto cloud = std::make_shared<Slic3r::OrcaCloudServiceAgent>("/tmp");
+    cloud->set_api_base_url("api.example.com");
+    OrcaPrinterAgent agent("/tmp");
+    agent.set_cloud_agent(cloud);
+
+    agent.set_user_selected_machine("printer-uuid-1");
+    // The configure runs on the connect worker; poll rather than racing it.
+    std::string url;
+    for (int i = 0; i < 300; ++i) {
+        url = cloud->selected_printer_mqtt_url();
+        if (!url.empty()) break;
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    CHECK(url == "wss://api.example.com/api/v1/printers/mqtt");
+
+    agent.set_user_selected_machine("");     // selection changes do not tear down the fleet socket
+    CHECK(cloud->selected_printer_mqtt_url() == "wss://api.example.com/api/v1/printers/mqtt");
+}
+
+TEST_CASE("a stale-generation inbound message is dropped", "[OrcaPrinterAgent]") {
+    struct GenProbe : OrcaPrinterAgent {
+        using OrcaPrinterAgent::OrcaPrinterAgent;
+        using OrcaPrinterAgent::make_lan_message_handler;   // expose for the test
+    };
+    GenProbe agent("/tmp");
+    int hits = 0;
+    agent.set_on_message_fn([&](std::string, std::string){ ++hits; });
+    auto handler_gen1 = agent.make_lan_message_handler(/*generation=*/1);
+    // m_lan_generation starts at 0; two bumps -> 2, so the epoch-1 handler is stale.
+    agent.bump_lan_generation_for_test();
+    agent.bump_lan_generation_for_test();
+    handler_gen1("dev-1", "{}");                            // late callback from gen 1
+    CHECK(hits == 0);
+}
+
+TEST_CASE("connect_server does not start an MQTT socket", "[OrcaCloud]") {
+    auto cloud = std::make_shared<Slic3r::OrcaCloudServiceAgent>("/tmp");
+    cloud->set_api_base_url("127.0.0.1:1");     // no session -> connect_server short-circuits before any probe
+    cloud->connect_server();
+    REQUIRE(cloud->get_mqtt_connection() != nullptr);       // created in the ctor
+    CHECK_FALSE(cloud->get_mqtt_connection()->is_running()); // never started
+    CHECK(cloud->selected_printer_mqtt_url().empty());
+}
+
+TEST_CASE("send_message* reject when there is no connection", "[OrcaPrinterAgent]") {
+    OrcaPrinterAgent agent("/tmp");   // no cloud agent, no LAN connection
+    CHECK(agent.send_message("d", "{}", 0, 0)            == BAMBU_NETWORK_ERR_INVALID_HANDLE);
+    CHECK(agent.send_message_to_printer("d", "{}", 0, 0) == BAMBU_NETWORK_ERR_INVALID_HANDLE);
+    CHECK(agent.send_message("", "{}", 0, 0)             == BAMBU_NETWORK_ERR_INVALID_HANDLE);   // empty dev_id
+}
+
+TEST_CASE("send_message_to_printer publishes on the LAN connection", "[OrcaPrinterAgent][.integration]") {
+    orca_mqtt_test::MockBroker broker;
+    OrcaPrinterAgent agent("/tmp");
+    const auto ep = broker.host_port();
+    agent.connect_printer("dev-1", ep.first + ":" + ep.second, "orcasonar", "code", false);
+
+    for (int i = 0; i < 150 && broker.connect_count() == 0; ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    REQUIRE(broker.connect_count() >= 1);
+
+    CHECK(agent.send_message_to_printer("dev-1", R"({"print":{"command":"pause","sequence_id":"20007"}})", 0, 0)
+          == BAMBU_NETWORK_SUCCESS);
+
+    // on_connected also publishes 4 requests; poll until "pause" specifically shows up.
+    bool saw_pause = false;
+    for (int i = 0; i < 150 && !saw_pause; ++i) {
+        for (const auto& r : broker.received_requests())
+            if (r.find("pause") != std::string::npos) { saw_pause = true; break; }
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    }
+    CHECK(saw_pause);
+    agent.disconnect_printer();
+}
+
+TEST_CASE("destroying an agent mid-connect does not hang or crash", "[OrcaPrinterAgent]") {
+    for (int i = 0; i < 20; ++i) {
+        auto agent = std::make_unique<OrcaPrinterAgent>("/tmp");
+        agent->connect_printer("dev-1", "127.0.0.1:1", "orcasonar", "code", false);  // nothing listening: instant ECONNREFUSED
+        agent.reset();   // ~OrcaPrinterAgent must stop the conn, join the thread, and not hang/crash
+    }
+    SUCCEED();
+}

tests/slic3rutils/test_printer_agent.cpp

@@ -1,5 +1,7 @@
 #include <catch2/catch_all.hpp>
 
+#include <slic3r/Utils/BBLPrinterAgent.hpp>
+#include <slic3r/Utils/MoonrakerPrinterAgent.hpp>
 #include <slic3r/Utils/NetworkAgentFactory.hpp>
 #include <slic3r/plugin/PythonPluginBridge.hpp>
 
@@ -8,11 +10,164 @@
 #include <pybind11/embed.h>
 #include <pybind11/pybind11.h>
 
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
 #include <string>
+#include <thread>
 
 using namespace Slic3r;
 namespace py = pybind11;
 
+class MoonrakerParserProbe : public MoonrakerPrinterAgent
+{
+public:
+    using MoonrakerPrinterAgent::parse_nozzle_diameter;
+
+    explicit MoonrakerParserProbe(std::string log_dir) : MoonrakerPrinterAgent(std::move(log_dir)) {}
+};
+
+TEST_CASE("Moonraker parses nozzle diameter from configfile settings", "[unit][moonraker]")
+{
+    const auto response = nlohmann::json::parse(R"({
+        "result": {
+            "status": {
+                "configfile": {
+                    "settings": {
+                        "extruder": {
+                            "nozzle_diameter": 0.6
+                        }
+                    }
+                }
+            }
+        }
+    })");
+
+    CHECK(MoonrakerParserProbe::parse_nozzle_diameter(response) == Catch::Approx(0.6f));
+}
+
+TEST_CASE("Moonraker parses nozzle diameter from raw config and tolerates missing data", "[unit][moonraker]")
+{
+    const auto raw_config_response = nlohmann::json::parse(R"({
+        "result": {
+            "status": {
+                "configfile": {
+                    "config": {
+                        "extruder": {
+                            "nozzle_diameter": "0.8"
+                        }
+                    }
+                }
+            }
+        }
+    })");
+    const auto missing_response = nlohmann::json::object();
+
+    CHECK(MoonrakerParserProbe::parse_nozzle_diameter(raw_config_response) == Catch::Approx(0.8f));
+    CHECK(MoonrakerParserProbe::parse_nozzle_diameter(missing_response) == 0.0f);
+}
+
+// why: these builders preserve the Bambu firmware dialect byte-for-byte, including its trailing space.
+TEST_CASE("unit: BBL AMS gcode builders preserve command bytes", "[unit][bbl]")
+{
+    CHECK(BBLPrinterAgent::ams_refresh_rfid_gcode("123") == "M620 R123 \n");
+    CHECK(BBLPrinterAgent::ams_calibrate_gcode(123) == "M620 C123 \n");
+    CHECK(BBLPrinterAgent::ams_select_tray_gcode("123") == "M620 P123 \n");
+}
+
+// why: an agent without a Bambu-dialect translation must refuse these commands before any network or wx path.
+TEST_CASE("unit: default AMS commands report not supported", "[unit][moonraker]")
+{
+    MoonrakerPrinterAgent agent("");
+
+    CHECK(agent.command_ams_refresh_rfid("dev", "123", 1, false) == ORCA_NETWORK_ERR_CMD_NOT_SUPPORTED);
+    CHECK(agent.command_ams_calibrate("dev", 1, 2, false) == ORCA_NETWORK_ERR_CMD_NOT_SUPPORTED);
+    CHECK(agent.command_ams_select_tray("dev", "123", 3, false) == ORCA_NETWORK_ERR_CMD_NOT_SUPPORTED);
+}
+
+TEST_CASE("unit: Moonraker light name matching", "[unit][moonraker]")
+{
+    CHECK(moonraker_is_light_name("caselight"));
+    CHECK(moonraker_is_light_name("LED_STRIP"));
+    CHECK_FALSE(moonraker_is_light_name("beeper"));
+    CHECK(moonraker_is_light_name("FLASHLIGHT_SWITCH"));
+    CHECK(moonraker_is_light_name("MODLELIGHT_SWITCH"));
+}
+
+// ===========================================================================
+// UNIT - handle_request's not-supported default.
+// The agent is the only thing that knows what it can translate, so an untranslated
+// command has to say so instead of returning success and letting the UI believe the
+// control worked. Guards the inverse too: the pushing namespace is genuinely
+// satisfied by the websocket status stream, and it re-fires from the keepalive timer
+// roughly once a second, so it must stay a success or it would raise a dialog on a
+// timer. Only branches that touch neither the network nor wx are exercised.
+// ===========================================================================
+TEST_CASE("unit: Moonraker reports untranslated commands as not supported", "[unit][moonraker]")
+{
+    MoonrakerPrinterAgent agent("");
+
+    CHECK(agent.send_message("dev", R"({"print":{"command":"ams_change_filament"}})", 0, 0) ==
+          ORCA_NETWORK_ERR_CMD_NOT_SUPPORTED);
+    CHECK(agent.send_message("dev", R"({"system":{"command":"set_door_stat"}})", 0, 0) ==
+          ORCA_NETWORK_ERR_CMD_NOT_SUPPORTED);
+    CHECK(agent.send_message("dev", R"({"xcam":{"command":"xcam_control_set"}})", 0, 0) ==
+          ORCA_NETWORK_ERR_CMD_NOT_SUPPORTED);
+
+    CHECK(agent.send_message("dev", R"({"pushing":{"command":"pushall"}})", 0, 0) == BAMBU_NETWORK_SUCCESS);
+    CHECK(agent.send_message("dev", R"({"pushing":{"command":"start"}})", 0, 0) == BAMBU_NETWORK_SUCCESS);
+
+    // why: malformed input is a different failure than an untranslated command, and the
+    // default must not swallow it into a misleading not-supported verdict.
+    CHECK(agent.send_message("dev", "{not json", 0, 0) == BAMBU_NETWORK_ERR_INVALID_RESULT);
+}
+
+// why: IPrinterAgent::fetch_filament_info is the single virtual hook derived agents override
+// (MoonrakerPrinterAgent's own override is synchronous, but QidiPrinterAgent's override is
+// fire-and-forget: it spawns a detached thread and returns immediately). QidiPrinterAgent is
+// `final`, so this probes the same contract with a controllable double instead.
+TEST_CASE("unit: a fire-and-forget override of fetch_filament_info is not waited on by the caller",
+          "[unit][moonraker]")
+{
+    class RecordingAgent : public Slic3r::MoonrakerPrinterAgent
+    {
+    public:
+        explicit RecordingAgent(std::string log_dir) : MoonrakerPrinterAgent(std::move(log_dir)) {}
+
+        std::atomic<bool>  invoked{false};
+        std::promise<void> release_gate;
+        std::promise<void> done_promise;
+
+        bool fetch_filament_info(std::string /*dev_id*/, FilamentSyncMode /*sync_mode*/ = FilamentSyncMode::pull) override
+        {
+            std::thread([this]() {
+                invoked.store(true);
+                // Block here until the test explicitly releases us, proving the caller
+                // (fetch_filament_info) does not wait for this to run.
+                release_gate.get_future().wait();
+                done_promise.set_value();
+            }).detach();
+            return true;
+        }
+    };
+
+    auto agent = std::make_shared<RecordingAgent>(std::string{});
+    auto done_future = agent->done_promise.get_future();
+
+    bool immediate_result = agent->fetch_filament_info("test-dev");
+
+    // fetch_filament_info must return before its background work completes — prove
+    // it by confirming the background call is still blocked on the gate right now.
+    REQUIRE(immediate_result == true);
+    REQUIRE(done_future.wait_for(std::chrono::milliseconds(100)) == std::future_status::timeout);
+
+    // Now let the background call finish and confirm it actually ran (polymorphic dispatch).
+    agent->release_gate.set_value();
+    REQUIRE(done_future.wait_for(std::chrono::seconds(2)) == std::future_status::ready);
+    REQUIRE(agent->invoked.load() == true);
+}
+
 // ===========================================================================
 // UNIT - printer-agent registry duplicate handling.
 // Confirms a duplicate agent id is rejected so a plugin cannot shadow a built-in

tests/slic3rutils/test_qidi_printer_agent.cpp

@@ -0,0 +1,131 @@
+#include <catch2/catch_all.hpp>
+
+#include <nlohmann/json.hpp>
+
+#include <slic3r/Utils/QidiPrinterAgent.hpp>
+
+#include <string>
+
+using namespace Slic3r;
+
+TEST_CASE("Qidi slot response rejects null variables without throwing", "[QidiPrinterAgent]")
+{
+    const std::string response = R"({
+        "result": {
+            "status": {
+                "save_variables": {
+                    "variables": null
+                }
+            }
+        }
+    })";
+    nlohmann::json status;
+    nlohmann::json variables;
+    std::string    error;
+    bool           parsed = true;
+
+    REQUIRE_NOTHROW(parsed = QidiPrinterAgent::parse_slot_response(response, status, variables, error));
+    CHECK_FALSE(parsed);
+    CHECK_THAT(error, Catch::Matchers::ContainsSubstring("variables"));
+    CHECK_THAT(error, Catch::Matchers::ContainsSubstring("object"));
+}
+
+TEST_CASE("Qidi slot response rejects missing and non-object fields without throwing", "[QidiPrinterAgent]")
+{
+    std::string response;
+
+    SECTION("missing result")
+    {
+        response = R"({})";
+    }
+
+    SECTION("non-object result")
+    {
+        response = R"({"result":null})";
+    }
+
+    SECTION("missing status")
+    {
+        response = R"({"result":{}})";
+    }
+
+    SECTION("non-object status")
+    {
+        response = R"({"result":{"status":null}})";
+    }
+
+    SECTION("missing save_variables")
+    {
+        response = R"({"result":{"status":{}}})";
+    }
+
+    SECTION("non-object save_variables")
+    {
+        response = R"({"result":{"status":{"save_variables":null}}})";
+    }
+
+    SECTION("missing variables")
+    {
+        response = R"({"result":{"status":{"save_variables":{}}}})";
+    }
+
+    SECTION("scalar")
+    {
+        response = R"({"result":{"status":{"save_variables":{"variables":42}}}})";
+    }
+
+    SECTION("array")
+    {
+        response = R"({"result":{"status":{"save_variables":{"variables":[]}}}})";
+    }
+
+    nlohmann::json status;
+    nlohmann::json variables;
+    std::string    error;
+    bool           parsed = true;
+
+    REQUIRE_NOTHROW(parsed = QidiPrinterAgent::parse_slot_response(response, status, variables, error));
+    CHECK_FALSE(parsed);
+}
+
+TEST_CASE("Qidi slot response exposes valid status and variables", "[QidiPrinterAgent]")
+{
+    const std::string response = R"({
+        "result": {
+            "status": {
+                "save_variables": {
+                    "variables": {
+                        "box_count": 2,
+                        "color_slot0": 3
+                    }
+                },
+                "box_stepper slot0": {
+                    "runout_button": 0
+                }
+            }
+        }
+    })";
+    nlohmann::json status;
+    nlohmann::json variables;
+    std::string    error;
+    bool           parsed = false;
+
+    REQUIRE_NOTHROW(parsed = QidiPrinterAgent::parse_slot_response(response, status, variables, error));
+    REQUIRE(parsed);
+    CHECK(status.is_object());
+    CHECK(variables.is_object());
+    CHECK(variables.at("box_count") == 2);
+    CHECK(status.contains("box_stepper slot0"));
+}
+
+TEST_CASE("Qidi slot response rejects invalid JSON", "[QidiPrinterAgent]")
+{
+    nlohmann::json status;
+    nlohmann::json variables;
+    std::string    error;
+    bool           parsed = true;
+
+    REQUIRE_NOTHROW(parsed = QidiPrinterAgent::parse_slot_response("{not json", status, variables, error));
+    CHECK_FALSE(parsed);
+    CHECK(error == "Invalid JSON response");
+}