diff --git a/tests/slic3rutils/CMakeLists.txt b/tests/slic3rutils/CMakeLists.txt
index 566835f495..1ade7b3faf 100644
--- a/tests/slic3rutils/CMakeLists.txt
+++ b/tests/slic3rutils/CMakeLists.txt
@@ -24,6 +24,8 @@ add_executable(${_TEST_NAME}_tests
     test_plugin_capabilities_in_use.cpp
     test_plugin_status.cpp
     test_printer_agent.cpp
+    test_orca_mqtt_connection.cpp
+    test_orca_printer_agent.cpp
     test_plugin_install.cpp
     test_plugin_lifecycle.cpp
     test_plugin_printer_agent.cpp
diff --git a/tests/slic3rutils/orca_mqtt_mock_broker.hpp b/tests/slic3rutils/orca_mqtt_mock_broker.hpp
new file mode 100644
index 0000000000..555a4a9a05
--- /dev/null
+++ b/tests/slic3rutils/orca_mqtt_mock_broker.hpp
@@ -0,0 +1,325 @@
+#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(); }
+
+    // MQTT PINGREQs seen while the client has no other traffic.
+    int ping_count() const { return m_ping_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
+            ++m_ping_count;
+            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. The production client keeps all of its
+    // WebSocket operations on its MQTT worker instead.
+    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();
+        if (socket.cancel(ec))
+            return;
+        // shutdown() before close() is what actually wakes a blocking read on the
+        // worker thread; close() alone does not on POSIX.
+        if (socket.shutdown(tcp::socket::shutdown_both, ec))
+            return;
+        if (socket.close(ec))
+            return;
+    }
+
+    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};
+    std::atomic<int>                             m_ping_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
diff --git a/tests/slic3rutils/test_orca_mqtt_connection.cpp b/tests/slic3rutils/test_orca_mqtt_connection.cpp
new file mode 100644
index 0000000000..1bb4983901
--- /dev/null
+++ b/tests/slic3rutils/test_orca_mqtt_connection.cpp
@@ -0,0 +1,243 @@
+#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 keepalive runs while the connection is idle", "[OrcaMqtt][.integration]") {
+    orca_mqtt_test::MockBroker broker;
+    OrcaMqttConnection conn;
+    OrcaMqttConnection::Config cfg;
+    cfg.url = broker.ws_url();
+    cfg.keepalive_seconds = 2;
+
+    REQUIRE(conn.start(cfg, [](const std::string&, const std::string&) {}, [](bool, bool) {}));
+    for (int i = 0; i < 200 && broker.ping_count() == 0; ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    CHECK(broker.ping_count() > 0);
+    conn.stop();
+}
+
+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();
+}
diff --git a/tests/slic3rutils/test_orca_printer_agent.cpp b/tests/slic3rutils/test_orca_printer_agent.cpp
new file mode 100644
index 0000000000..1ff111a879
--- /dev/null
+++ b/tests/slic3rutils/test_orca_printer_agent.cpp
@@ -0,0 +1,208 @@
+#include <catch2/catch_test_macros.hpp>
+#include <slic3r/Utils/IPrinterAgent.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");
+    Slic3r::PrinterConnectionParams params{
+        "dev-1", "10.255.255.1", "orcasonar", "code", "", false, ""
+    };
+    const int rc = agent.connect_printer(params);
+    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(Slic3r::PrinterConnectionParams{"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(Slic3r::PrinterConnectionParams{"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();
+}