feat(vp): mirror live target printer state to slicer in non-proxy modes

In non-proxy VP modes (Immediate / Review / Print Queue), the slicer now
sees real AMS / FTS / nozzle / k-profile state from the target printer
and streams the live camera — full slicer-as-remote functionality without
giving up Bambuddy's queue / archive / dispatch features.

Architecture (cached-as-base, single source of truth). The bridge caches
the latest real push_status and info.get_version response from Bambuddy's
existing per-printer MQTT subscription — no second session on the printer,
firmware in-flight budget unaffected (#1164). _send_status_report serves
a near-byte-identical copy of the cached push with only the upload-state-
machine fields overridden. Command responses (extrusion_cali_get, AMS
write acks, xcam) fan out raw — they carry sequence_ids the slicer is
waiting on. Slicer-issued commands forward to the printer except
project_file / gcode_file, which still terminate locally because the file
lives on Bambuddy. Camera is a raw TCPProxy on bind_ip:322 → printer:322,
same approach proxy mode uses.

Field-shape gotchas pinned in the bridge module's docstring and the
new test file:
  - Real Bambu pushes use json.dumps(indent=4) wire format. Compact JSON
    fails BambuStudio's Send pre-flight silently.
  - net.info[*].ip is the FTP destination IP (little-endian uint32).
    Without rewriting to the VP bind IP, the slicer FTPs straight to
    the real printer.
  - upgrade_state.sn rewritten to VP serial; AMS-hardware sn fields
    (n3f/0.sn etc.) left alone.
  - ipcam.rtsp_url passes through unchanged; BambuStudio overrides the
    URL host with the device IP it bound on, so :322 lands on the VP's
    TCPProxy.
  - extrusion_cali_get must forward; answering it locally hides the
    user's stored per-filament k-profiles.

Setup nuance for camera: the VP's access code must match the target
printer's because the slicer authenticates RTSPS with whatever access
code is in its profile. MQTT and FTP work either way.

Tested e2e with BambuStudio and OrcaSlicer against H2D (dual-nozzle,
AMS 2 Pro + AMS HT) and X1C across all three non-proxy modes — sync,
send, k-profile lookup, AMS configuration from slicer, and live camera
all work. Proxy mode is untouched: SlicerProxyManager owns its own
proxies and never instantiates SimpleMQTTServer or MQTTBridge.

25 new tests in backend/tests/unit/test_vp_mqtt_bridge.py cover lifecycle,
caching, identity / IP rewriting, wire format, slicer→printer routing,
and the LE-uint32 IP encoder against the real H2D capture value.

README.md

@@ -236,7 +236,8 @@ Perfect for remote print farms, traveling makers, or accessing your home printer
 - Interactive API browser with live testing
 
 ### 🖨️ Virtual Printer & Remote Printing
-- **🌐 Proxy Mode (NEW!)** — Print remotely from anywhere via secure TLS relay
+- **🌐 Proxy Mode** — Print remotely from anywhere via secure TLS relay
+- **🪞 Live target-printer mirror in non-proxy modes (NEW!)** — Immediate / Review / Queue VPs now mirror their target printer's live state to the slicer: AMS slot contents, FTS / dual-extruder routing, k-profiles, AMS load / dry / calibration commands, and the camera stream all flow through the VP. Use the slicer as a full remote for the printer behind the VP without giving up Bambuddy's queue / archive / dispatch features.
 - Emulates a Bambu Lab printer on your network
 - Send prints directly from Bambu Studio/Orca Slicer
 - Configurable printer model (X1C, P1S, A1, H2D, etc.)

backend/app/main.py

@@ -4433,6 +4433,7 @@ async def lifespan(app: FastAPI):
     from backend.app.services.virtual_printer import virtual_printer_manager
 
     virtual_printer_manager.set_session_factory(async_session)
+    virtual_printer_manager.set_printer_manager(printer_manager)
     try:
         await virtual_printer_manager.sync_from_db()
         logging.info("Virtual printer manager synced from database")

backend/app/services/bambu_mqtt.py

@@ -358,6 +358,10 @@ class BambuMQTTClient:
         self._message_log: deque[MQTTLogEntry] = deque(maxlen=100)
         self._logging_enabled: bool = False
         self._last_message_time: float = 0.0  # Track when we last received a message
+        # Raw-message fan-out for VP MQTT bridge (non-proxy modes republish the
+        # printer's pushes verbatim to slicers connected to a virtual printer).
+        # Handlers receive (topic, payload_bytes) before JSON parsing.
+        self._raw_message_handlers: list[Callable[[str, bytes], None]] = []
         self._disconnection_event: threading.Event | None = None
         self._previous_ams_hash: str | None = None  # Track AMS changes
 
@@ -685,6 +689,15 @@ class BambuMQTTClient:
             self.on_state_change(self.state)
 
     def _on_message(self, client, userdata, msg):
+        for handler in self._raw_message_handlers:
+            try:
+                handler(msg.topic, msg.payload)
+            except Exception:
+                logger.exception(
+                    "[%s] raw-message handler crashed for topic=%s",
+                    self.serial_number,
+                    msg.topic,
+                )
         try:
             try:
                 raw = msg.payload.decode()
@@ -3536,6 +3549,39 @@ class BambuMQTTClient:
         """Check if logging is enabled."""
         return self._logging_enabled
 
+    def register_raw_message_handler(self, handler: Callable[[str, bytes], None]) -> None:
+        """Register a handler invoked for every incoming MQTT message.
+
+        Used by the VP MQTT bridge to republish the printer's report pushes to
+        slicers connected to a virtual printer in non-proxy mode. Handlers run
+        on paho's network thread and must not block; exceptions are caught.
+        """
+        if handler not in self._raw_message_handlers:
+            self._raw_message_handlers.append(handler)
+
+    def unregister_raw_message_handler(self, handler: Callable[[str, bytes], None]) -> None:
+        """Unregister a previously-registered raw-message handler."""
+        try:
+            self._raw_message_handlers.remove(handler)
+        except ValueError:
+            pass
+
+    def publish_raw(self, topic: str, payload: bytes | str, qos: int = 1) -> bool:
+        """Publish a pre-formed payload directly to the printer's MQTT broker.
+
+        Used by the VP MQTT bridge to forward slicer-originated commands without
+        going through send_command's sequence-id mangling. Returns False if the
+        underlying paho client isn't ready.
+        """
+        if self._client is None:
+            return False
+        try:
+            info = self._client.publish(topic, payload, qos=qos)
+            return info.rc == mqtt.MQTT_ERR_SUCCESS
+        except Exception:
+            logger.exception("[%s] publish_raw failed for topic=%s", self.serial_number, topic)
+            return False
+
     def send_drying_command(
         self, ams_id: int, temp: int, duration: int, mode: int = 1, filament: str = "", rotate_tray: bool = False
     ):

backend/app/services/virtual_printer/manager.py

@@ -9,15 +9,20 @@ import logging
 from collections.abc import Callable
 from datetime import datetime, timezone
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 from backend.app.core.config import settings as app_settings
 from backend.app.services.virtual_printer.bind_server import BindServer
 from backend.app.services.virtual_printer.certificate import CertificateService
 from backend.app.services.virtual_printer.ftp_server import VirtualPrinterFTPServer
+from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge
 from backend.app.services.virtual_printer.mqtt_server import SimpleMQTTServer
 from backend.app.services.virtual_printer.ssdp_server import SSDPProxy, VirtualPrinterSSDPServer
 from backend.app.services.virtual_printer.tailscale import tailscale_service
-from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager
+from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager, TCPProxy
+
+if TYPE_CHECKING:
+    from backend.app.services.printer_manager import PrinterManager
 
 logger = logging.getLogger(__name__)
 
@@ -117,6 +122,7 @@ class VirtualPrinterInstance:
         tailscale_disabled: bool = True,
         base_dir: Path,
         session_factory: Callable | None = None,
+        printer_manager: "PrinterManager | None" = None,
     ):
         self.id = vp_id
         self.name = name
@@ -133,6 +139,7 @@ class VirtualPrinterInstance:
         self.remote_interface_ip = remote_interface_ip
         self.tailscale_disabled = tailscale_disabled
         self._session_factory = session_factory
+        self._printer_manager = printer_manager
 
         # Directories
         self.upload_dir = base_dir / "uploads" / str(vp_id)
@@ -161,6 +168,8 @@ class VirtualPrinterInstance:
         self._proxy: SlicerProxyManager | None = None
         self._ftp: VirtualPrinterFTPServer | None = None
         self._mqtt: SimpleMQTTServer | None = None
+        self._mqtt_bridge: MQTTBridge | None = None
+        self._rtsp_proxy: TCPProxy | None = None
         self._bind: BindServer | None = None
         self._ssdp: VirtualPrinterSSDPServer | None = None
         self._ssdp_proxy: SSDPProxy | None = None
@@ -621,6 +630,44 @@ class VirtualPrinterInstance:
             )
         )
 
+        # MQTT bridge — fans out the target printer's pushes to slicers connected
+        # to this VP and forwards their commands back to the printer. Only meaningful
+        # when a target printer is configured AND printer_manager was injected (it
+        # always is at runtime; tests may omit it).
+        if self.target_printer_id is not None and self._printer_manager is not None:
+            self._mqtt_bridge = MQTTBridge(
+                vp_id=self.id,
+                vp_name=self.name,
+                vp_serial=self.serial,
+                target_printer_id=self.target_printer_id,
+                mqtt_server=self._mqtt,
+                printer_manager=self._printer_manager,
+            )
+            self._mqtt.set_bridge(self._mqtt_bridge)
+            await self._mqtt_bridge.start()
+
+            # RTSPS camera passthrough on port 322. BambuStudio's camera button
+            # connects to the device IP it bound on (the VP), not the IP in
+            # `ipcam.rtsp_url`. Without a listener on <bind_ip>:322 the slicer
+            # gets connection refused → "LAN connection failed". Same raw TCP
+            # pass-through used by SlicerProxyManager in proxy mode.
+            target_client = self._printer_manager.get_client(self.target_printer_id)
+            target_ip = getattr(target_client, "ip_address", None) if target_client else None
+            if target_ip:
+                self._rtsp_proxy = TCPProxy(
+                    name="RTSP",
+                    listen_port=322,
+                    target_host=target_ip,
+                    target_port=322,
+                    bind_address=bind_addr,
+                )
+                self._tasks.append(
+                    asyncio.create_task(
+                        run_with_logging(self._rtsp_proxy.start(), "RTSP"),
+                        name=f"vp_{self.id}_rtsp",
+                    )
+                )
+
         # Bind server
         self._bind = BindServer(
             serial=self.serial,
@@ -663,6 +710,20 @@ class VirtualPrinterInstance:
         """Stop server-mode services."""
         await self._cancel_renewal_task()
         await self._cancel_restart_task()
+        if self._mqtt_bridge:
+            try:
+                await self._mqtt_bridge.stop()
+            except Exception:
+                logger.exception("[VP %s] MQTT bridge stop failed", self.name)
+            if self._mqtt:
+                self._mqtt.set_bridge(None)
+            self._mqtt_bridge = None
+        if self._rtsp_proxy:
+            try:
+                await self._rtsp_proxy.stop()
+            except Exception:
+                logger.exception("[VP %s] RTSP proxy stop failed", self.name)
+            self._rtsp_proxy = None
         if self._ftp:
             await self._ftp.stop()
             self._ftp = None
@@ -803,6 +864,7 @@ class VirtualPrinterManager:
 
     def __init__(self):
         self._session_factory: Callable | None = None
+        self._printer_manager: PrinterManager | None = None
         self._instances: dict[int, VirtualPrinterInstance] = {}
 
         # Directories
@@ -827,6 +889,10 @@ class VirtualPrinterManager:
         """Set the database session factory."""
         self._session_factory = session_factory
 
+    def set_printer_manager(self, printer_manager: "PrinterManager") -> None:
+        """Inject the global printer_manager so non-proxy VPs can mirror their target's MQTT stream."""
+        self._printer_manager = printer_manager
+
     @property
     def is_enabled(self) -> bool:
         """Check if any virtual printer is running."""
@@ -939,6 +1005,7 @@ class VirtualPrinterManager:
                     tailscale_disabled=vp.tailscale_disabled,
                     base_dir=self._base_dir,
                     session_factory=self._session_factory,
+                    printer_manager=self._printer_manager,
                 )
                 self._instances[vp.id] = instance
                 await instance.start_server()

backend/app/services/virtual_printer/mqtt_bridge.py

@@ -0,0 +1,334 @@
+"""MQTT bridge for non-proxy virtual printers.
+
+Mirrors the target printer's state to slicers connected to a virtual printer
+without opening a second MQTT session on the printer (reuses Bambuddy's
+existing subscription — firmware inflight budget unaffected, see PR #1164).
+
+Architecture (cached-as-base, not a separate fan-out stream):
+
+  - **push_status** snapshots from the printer are CACHED here. The VP's
+    `SimpleMQTTServer._send_status_report` consults that cache and sends
+    a near-byte-identical copy of the real push to the slicer (with
+    sequence_id / gcode_state / etc. overridden). Single source of truth
+    keeps BambuStudio's Send pre-flight happy.
+  - **info.get_version** responses are also cached so the synthetic version
+    response can include the real AMS module list (n3f/n3s/ams entries).
+    Without this BambuStudio's Prepare tab labels every AMS as "unknown".
+  - **Other command responses** (extrusion_cali_get, AMS write acks,
+    xcam responses, …) are fanned out raw to the slicer — they carry
+    sequence_ids the slicer is waiting on; the slicer matches and ignores
+    unrelated ones.
+
+Identity rewriting at cache time:
+
+  - `upgrade_state.sn` (and any other nested dict's `sn` matching the real
+    serial) → VP serial
+  - `net.info[*].ip` little-endian uint32 → VP bind IP. BambuStudio reads
+    this as the FTP destination IP. Without this the slicer FTPs straight
+    to the real printer and bypasses Bambuddy.
+  - `ipcam.rtsp_url` is left unchanged: BambuStudio overrides the URL host
+    with the device IP it bound to (the VP), so the slicer hits the VP's
+    own RTSPS proxy on port 322.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import json
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from backend.app.services.bambu_mqtt import BambuMQTTClient
+    from backend.app.services.printer_manager import PrinterManager
+    from backend.app.services.virtual_printer.mqtt_server import SimpleMQTTServer
+
+logger = logging.getLogger(__name__)
+
+REFRESH_INTERVAL_SECONDS = 30.0
+
+
+def _ip_to_uint32_le(ip_str: str) -> int:
+    """Encode dotted-quad IPv4 as little-endian uint32 (Bambu MQTT's `net.info[].ip` shape)."""
+    parts = [int(x) for x in ip_str.split(".")]
+    if len(parts) != 4 or any(p < 0 or p > 255 for p in parts):
+        raise ValueError(f"invalid IPv4: {ip_str!r}")
+    return parts[0] | (parts[1] << 8) | (parts[2] << 16) | (parts[3] << 24)
+
+
+class MQTTBridge:
+    """Per-VP MQTT fan-out between a real printer and slicers connected to a VP."""
+
+    def __init__(
+        self,
+        *,
+        vp_id: int,
+        vp_name: str,
+        vp_serial: str,
+        target_printer_id: int,
+        mqtt_server: SimpleMQTTServer,
+        printer_manager: PrinterManager,
+    ):
+        self.vp_id = vp_id
+        self.vp_name = vp_name
+        self.vp_serial = vp_serial
+        self.target_printer_id = target_printer_id
+        self._mqtt_server = mqtt_server
+        self._printer_manager = printer_manager
+        self._target_client: BambuMQTTClient | None = None
+        self._target_serial: str | None = None
+        self._target_ip_uint32_le: int | None = None
+        self._vp_ip_uint32_le: int | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._refresh_task: asyncio.Task | None = None
+        self._stopping = False
+        self._latest_print_state: dict | None = None
+        self._latest_version_modules: list | None = None
+
+    @property
+    def is_active(self) -> bool:
+        """True iff a target client is bound and currently connected."""
+        client = self._target_client
+        return bool(client is not None and getattr(client, "state", None) and client.state.connected)
+
+    async def start(self) -> None:
+        """Bind to the target printer (if connected) and start the refresh loop."""
+        self._loop = asyncio.get_running_loop()
+        self._stopping = False
+        self._resolve_client()
+        self._refresh_task = asyncio.create_task(self._refresh_loop())
+
+    async def stop(self) -> None:
+        """Detach from the target printer and stop the refresh loop."""
+        self._stopping = True
+        if self._refresh_task is not None:
+            self._refresh_task.cancel()
+            try:
+                await self._refresh_task
+            except asyncio.CancelledError:
+                pass
+            self._refresh_task = None
+        self._unbind_client()
+        self._loop = None
+
+    async def _refresh_loop(self) -> None:
+        """Re-resolve the target client periodically — paho clients can be replaced.
+
+        BambuMQTTClient is destroyed and recreated on PrinterManager.connect_printer
+        (e.g. printer config update). Without periodic refresh the bridge would lose
+        fan-out after such a churn until the VP itself restarts.
+        """
+        try:
+            while not self._stopping:
+                await asyncio.sleep(REFRESH_INTERVAL_SECONDS)
+                self._resolve_client()
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("[%s] MQTT bridge refresh loop crashed", self.vp_name)
+
+    def _resolve_client(self) -> None:
+        """Look up the current client for target_printer_id and rebind if it changed."""
+        try:
+            current = self._printer_manager.get_client(self.target_printer_id)
+        except Exception:
+            logger.exception("[%s] MQTT bridge: get_client failed", self.vp_name)
+            return
+
+        if current is self._target_client:
+            return
+
+        # Client identity changed — unregister from the old, register on the new.
+        self._unbind_client()
+        if current is None:
+            return
+
+        try:
+            current.register_raw_message_handler(self._on_printer_raw)
+        except Exception:
+            logger.exception("[%s] MQTT bridge: register_raw_message_handler failed", self.vp_name)
+            return
+
+        self._target_client = current
+        self._target_serial = getattr(current, "serial_number", None)
+
+        # Cache printer IP and VP bind IP encoded as little-endian uint32, so we
+        # can rewrite `net.info[*].ip` in cached push_status. BambuStudio reads
+        # that field for the FTP destination IP — without rewriting, the slicer
+        # bypasses the VP and FTPs straight to the real printer.
+        target_ip = getattr(current, "ip_address", None)
+        vp_ip = getattr(self._mqtt_server, "bind_address", None)
+        if target_ip and vp_ip and vp_ip not in ("0.0.0.0", "", None):  # nosec B104
+            try:
+                self._target_ip_uint32_le = _ip_to_uint32_le(target_ip)
+                self._vp_ip_uint32_le = _ip_to_uint32_le(vp_ip)
+            except ValueError:
+                self._target_ip_uint32_le = None
+                self._vp_ip_uint32_le = None
+
+        logger.info(
+            "[%s] MQTT bridge bound to printer %s (serial=%s)",
+            self.vp_name,
+            self.target_printer_id,
+            self._target_serial,
+        )
+
+        # Trigger a fresh get_version + pushall against the printer so the bridge
+        # cache populates immediately. Bambuddy itself queries these on connect,
+        # but that fires before the bridge attaches as a raw-message consumer,
+        # so without this nudge the cache stays empty until the next periodic
+        # query (which can be minutes away).
+        request_fn = getattr(current, "_request_version", None)
+        if callable(request_fn):
+            try:
+                request_fn()
+            except Exception:
+                logger.exception("[%s] MQTT bridge: _request_version failed", self.vp_name)
+        request_status_fn = getattr(current, "request_status_update", None)
+        if callable(request_status_fn):
+            try:
+                request_status_fn()
+            except Exception:
+                logger.exception("[%s] MQTT bridge: request_status_update failed", self.vp_name)
+
+    def _unbind_client(self) -> None:
+        if self._target_client is None:
+            return
+        try:
+            self._target_client.unregister_raw_message_handler(self._on_printer_raw)
+        except Exception:
+            logger.exception("[%s] MQTT bridge: unregister_raw_message_handler failed", self.vp_name)
+        logger.info("[%s] MQTT bridge unbound from printer %s", self.vp_name, self.target_printer_id)
+        self._target_client = None
+        self._target_serial = None
+
+    def _on_printer_raw(self, topic: str, payload: bytes) -> None:
+        """Paho-thread callback — cache the latest push_status for synthetic replay.
+
+        Instead of fanning out a second stream of MQTT messages to the slicer
+        (which trips BambuStudio's Send pre-flight consistency checks), we cache
+        the latest real printer push_status here. The VP's existing 1 Hz
+        synthetic push (which is what Send is built around) consults this cache
+        and replaces its stub fields with real values when available.
+        """
+        if self._stopping:
+            return
+        target_serial = self._target_serial
+        if not target_serial:
+            return
+        prefix = f"device/{target_serial}/"
+        if not topic.startswith(prefix):
+            return
+        suffix = topic[len(prefix) :]
+        if not suffix.startswith("report"):
+            return
+        try:
+            data = json.loads(payload)
+        except json.JSONDecodeError:
+            return
+
+        # Race-free by construction: `json.loads` returns a fresh dict tree per
+        # call so paho-thread mutations below cannot collide with prior cached
+        # state held by the asyncio thread. `_send_status_report`'s shallow
+        # `dict(cached)` is also safe because nothing else writes to the cached
+        # tree after assignment. The defensive deep-copy on store below removes
+        # any future risk if a maintainer later re-enters the cached dict to
+        # mutate it.
+
+        # push_status snapshots → cache the print dict for the periodic 1 Hz
+        # cached-as-base delivery. We do NOT fan these out separately (the
+        # 1 Hz cached-as-base IS the slicer-facing push_status stream).
+        print_data = data.get("print")
+        if isinstance(print_data, dict) and print_data.get("command") == "push_status":
+            for value in print_data.values():
+                if isinstance(value, dict) and value.get("sn") == target_serial:
+                    value["sn"] = self.vp_serial
+            # Note: `ipcam.rtsp_url` carries the real printer's IP. We pass it
+            # through unchanged — the slicer uses it to fetch the live camera
+            # stream directly from the printer. On the same LAN this works as
+            # long as the slicer's stored access code matches the printer's
+            # (i.e. configure the VP with the same access code as its target).
+            # Rewrite real printer IP → VP bind IP in `net.info[*].ip` so the
+            # slicer's FTP destination resolves to the VP, not the real printer.
+            if self._target_ip_uint32_le is not None and self._vp_ip_uint32_le is not None:
+                net = print_data.get("net")
+                if isinstance(net, dict):
+                    info = net.get("info")
+                    if isinstance(info, list):
+                        for entry in info:
+                            if isinstance(entry, dict) and entry.get("ip") == self._target_ip_uint32_le:
+                                entry["ip"] = self._vp_ip_uint32_le
+            # Defensive deep copy on store so the cache is fully decoupled from
+            # the freshly-parsed tree and from any reader's reference.
+            self._latest_print_state = copy.deepcopy(print_data)
+            return
+
+        # info.get_version responses → cache the module list so the synthetic
+        # version response can include the real AMS modules.
+        info_data = data.get("info")
+        if isinstance(info_data, dict) and info_data.get("command") == "get_version":
+            modules = info_data.get("module")
+            if isinstance(modules, list):
+                rewritten: list = []
+                for module in modules:
+                    if isinstance(module, dict):
+                        module = dict(module)
+                        if module.get("sn") == target_serial:
+                            module["sn"] = self.vp_serial
+                    rewritten.append(module)
+                self._latest_version_modules = rewritten
+            # Don't fan out get_version — the slicer's request (when it issues
+            # one) is intercepted locally and answered from the cached modules.
+            return
+
+        # Everything else (extrusion_cali_get response, AMS write acks, xcam
+        # responses, …): fan out to the slicer. These are responses to commands
+        # the slicer (or Bambuddy) issued; the slicer matches by sequence_id and
+        # ignores responses to commands it didn't send. Without this, slicer-
+        # initiated queries like extrusion_cali_get hang forever and BambuStudio
+        # blocks Send waiting for the response.
+        loop = self._loop
+        if loop is None:
+            return
+        target_bytes = target_serial.encode("ascii")
+        if target_bytes in payload:
+            payload = payload.replace(target_bytes, self.vp_serial.encode("ascii"))
+        vp_topic = f"device/{self.vp_serial}/{suffix}"
+        try:
+            asyncio.run_coroutine_threadsafe(
+                self._mqtt_server.push_raw_to_clients(vp_topic, payload),
+                loop,
+            )
+        except RuntimeError:
+            pass
+
+    def get_latest_print_state(self) -> dict | None:
+        """Return the most recent real printer push_status `print` dict, or None."""
+        return self._latest_print_state
+
+    def get_latest_version_modules(self) -> list | None:
+        """Return the most recent real printer get_version `module` list, or None."""
+        return self._latest_version_modules
+
+    def forward_to_printer(self, payload: dict) -> bool:
+        """Publish a slicer-originated command to the real printer's request topic.
+
+        Returns False if no printer client is currently bound.
+        """
+        client = self._target_client
+        target_serial = self._target_serial
+        if client is None or target_serial is None:
+            logger.debug(
+                "[%s] forward_to_printer dropped (printer %s not bound): %s",
+                self.vp_name,
+                self.target_printer_id,
+                list(payload.keys()),
+            )
+            return False
+        topic = f"device/{target_serial}/request"
+        try:
+            return client.publish_raw(topic, json.dumps(payload), qos=1)
+        except Exception:
+            logger.exception("[%s] forward_to_printer publish failed", self.vp_name)
+            return False

backend/app/services/virtual_printer/mqtt_server.py

@@ -10,6 +10,10 @@ import logging
 import ssl
 from collections.abc import Callable
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge
 
 logger = logging.getLogger(__name__)
 
@@ -218,6 +222,12 @@ class SimpleMQTTServer:
         self._current_file = ""
         self._prepare_percent = "0"
 
+        # MQTT bridge for non-proxy modes — set by VirtualPrinterInstance after start().
+        # When the bridge is_active, real printer pushes are fanned out to slicers and
+        # the synthetic 1s push is suspended. When the target printer goes offline the
+        # synthetic fallback resumes automatically.
+        self._bridge: MQTTBridge | None = None
+
     async def start(self) -> None:
         """Start the MQTT server."""
         if self._running:
@@ -346,14 +356,17 @@ class SimpleMQTTServer:
             return None
         return rest[:slash]
 
+    def set_bridge(self, bridge: "MQTTBridge | None") -> None:
+        """Attach (or detach) the MQTT bridge that mirrors the target printer."""
+        self._bridge = bridge
+
     async def _periodic_status_push(self) -> None:
-        """Send periodic status updates to all connected clients."""
+        """Send periodic status updates to all connected clients (1 Hz, exact pre-bridge behaviour)."""
         logger.info("Starting periodic status push task")
         while self._running:
             try:
                 await asyncio.sleep(1)  # Push every 1 second like real printers
 
-                # Send status to all connected clients
                 disconnected = []
                 for client_id, writer in list(self._clients.items()):
                     try:
@@ -378,6 +391,48 @@ class SimpleMQTTServer:
 
         logger.info("Periodic status push task stopped")
 
+    async def push_raw_to_clients(self, topic: str, payload: bytes) -> None:
+        """Publish a pre-serialized MQTT payload on `topic` to every connected slicer.
+
+        Called by MQTTBridge from the asyncio loop (scheduled via
+        run_coroutine_threadsafe from paho's network thread).
+        """
+        topic_bytes = topic.encode("utf-8")
+        # MQTT remaining-length: 2-byte topic length prefix + topic + message body.
+        remaining = 2 + len(topic_bytes) + len(payload)
+        packet = bytearray([0x30])  # PUBLISH, QoS 0
+        while True:
+            byte = remaining % 128
+            remaining //= 128
+            if remaining > 0:
+                byte |= 0x80
+            packet.append(byte)
+            if remaining == 0:
+                break
+        packet.extend([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF])
+        packet.extend(topic_bytes)
+        packet.extend(payload)
+        frame = bytes(packet)
+
+        disconnected = []
+        for client_id, writer in list(self._clients.items()):
+            try:
+                if writer.is_closing():
+                    disconnected.append(client_id)
+                    continue
+                writer.write(frame)
+                try:
+                    await asyncio.wait_for(writer.drain(), timeout=5)
+                except TimeoutError:
+                    logger.debug("MQTT drain timeout pushing bridge frame to %s", client_id)
+            except OSError as e:
+                logger.debug("Failed to push bridge frame to %s: %s", client_id, e)
+                disconnected.append(client_id)
+
+        for client_id in disconnected:
+            self._clients.pop(client_id, None)
+            self._client_serials.pop(client_id, None)
+
     async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
         """Handle an MQTT client connection."""
         addr = writer.get_extra_info("peername")
@@ -575,10 +630,40 @@ class SimpleMQTTServer:
             logger.debug("MQTT SUBSCRIBE error: %s", e)
 
     async def _send_status_report(self, writer: asyncio.StreamWriter, serial: str | None = None) -> None:
-        """Send a status report to the slicer after connection."""
+        """Send a status report to the slicer after connection.
+
+        When a bridge is active and has cached the real printer's latest
+        push_status, send a copy of the real push with only the upload-state-
+        machine fields we own (gcode_state, gcode_file, prepare_percent,
+        subtask_name) overridden. BambuStudio's Send pre-flight checks the
+        push_status shape against what it expects from the printer model, and
+        the synthetic stub introduced fields the real H2D doesn't have (storage,
+        the wrong chamber_temper shape, etc.) which trip the check.
+        """
         try:
-            # Build status message matching Bambu printer format
             self._sequence_id += 1
+
+            cached = self._bridge.get_latest_print_state() if self._bridge is not None else None
+            if isinstance(cached, dict):
+                # Real-printer-shaped response. Copy the cache, then replace the
+                # protocol / upload-state fields with values under our control.
+                print_block = dict(cached)
+                print_block["sequence_id"] = str(self._sequence_id)
+                print_block["command"] = "push_status"
+                print_block["msg"] = 0
+                print_block["gcode_state"] = self._gcode_state
+                print_block["gcode_file"] = self._current_file
+                print_block["gcode_file_prepare_percent"] = self._prepare_percent
+                if self._current_file:
+                    print_block["subtask_name"] = self._current_file.replace(".3mf", "")
+                else:
+                    # Don't override real subtask_name with empty if no upload pending.
+                    print_block.setdefault("subtask_name", "")
+                status = {"print": print_block}
+                await self._publish_to_report(writer, status, serial or self.serial)
+                return
+
+            # No bridge / no cache yet — fall back to the synthetic stub.
             status = {
                 "print": {
                     "sequence_id": str(self._sequence_id),
@@ -724,6 +809,15 @@ class SimpleMQTTServer:
                 }
             }
 
+            # Overlay real version modules from the bridge cache when available
+            # (specifically the AMS modules ams/0, n3f/0, n3s/128 etc. that
+            # BambuStudio's Prepare tab uses to identify AMS hardware — without
+            # them every AMS unit shows as "unknown" in the Prepare panel).
+            if self._bridge is not None:
+                cached_modules = self._bridge.get_latest_version_modules()
+                if isinstance(cached_modules, list) and cached_modules:
+                    version_info["info"]["module"] = cached_modules
+
             await self._publish_to_report(writer, version_info, serial)
             logger.info("Sent version response (product_name=%s)", product_name)
 
@@ -740,9 +834,15 @@ class SimpleMQTTServer:
         self._prepare_percent = prepare_percent
 
     async def _publish_to_report(self, writer: asyncio.StreamWriter, payload: dict, serial: str = "") -> None:
-        """Publish a message on the device report topic."""
+        """Publish a message on the device report topic.
+
+        Real Bambu printers wire-format push_status JSON with 4-space indentation
+        (32254 bytes for an idle H2D push vs 14268 bytes compact). BambuStudio's
+        Send pre-flight rejects compact JSON — without matching the on-wire
+        format the slicer never proceeds to FTP upload.
+        """
         topic = f"device/{serial or self.serial}/report"
-        message = json.dumps(payload)
+        message = json.dumps(payload, indent=4)
 
         topic_bytes = topic.encode("utf-8")
         message_bytes = message.encode("utf-8")
@@ -857,6 +957,15 @@ class SimpleMQTTServer:
                 )
                 return
 
+            # The synthetic flow below is the original (pre-bridge) behaviour and is
+            # what the proven-working FTP "Send" depends on. Do NOT replace any
+            # synthetic response with a forward — only ADD forwarding alongside,
+            # at the bottom, for commands the synthetic flow doesn't handle
+            # (AMS write / xcam / system / etc., which need to actually reach
+            # the real printer).
+
+            handled_locally = False
+
             # Handle pushing command (status request)
             if "pushing" in data:
                 pushing_data = data["pushing"]
@@ -864,13 +973,13 @@ class SimpleMQTTServer:
                 logger.info("MQTT pushing command: %s", command)
 
                 if command == "pushall":
-                    # Slicer is requesting full status - send response
                     logger.info("Sending status report in response to pushall")
                     await self._send_status_report(writer, serial=client_serial)
+                    handled_locally = True
                 elif command == "start":
-                    # Slicer wants periodic status updates - send one now
                     logger.info("Starting status push stream")
                     await self._send_status_report(writer, serial=client_serial)
+                    handled_locally = True
 
             # Handle info commands (get_version, etc.)
             if "info" in data:
@@ -881,6 +990,7 @@ class SimpleMQTTServer:
 
                 if command == "get_version":
                     await self._send_version_response(writer, sequence_id, serial=client_serial)
+                    handled_locally = True
 
             # Handle print commands
             if "print" in data:
@@ -891,13 +1001,18 @@ class SimpleMQTTServer:
 
                 logger.info("MQTT print command: %s for %s", command, filename)
 
-                if command == "project_file":
-                    # Respond with PREPARE status so slicer proceeds with FTP upload
+                if command in ("project_file", "gcode_file"):
+                    # File lives on Bambuddy, not the printer — synthetic only.
                     file_3mf = print_data.get("file", filename)
                     await self._send_print_response(writer, sequence_id, file_3mf, serial=client_serial)
-
                     if self.on_print_command:
                         await self._notify_print_command(filename, print_data)
+                    handled_locally = True
+
+            # Forward anything the synthetic flow didn't handle to the real
+            # printer. AMS load / dry / xcam / system / extrusion_cali_get etc.
+            if not handled_locally and self._bridge is not None and self._bridge.is_active:
+                self._bridge.forward_to_printer(data)
 
         except (IndexError, ValueError, OSError) as e:
             logger.debug("MQTT PUBLISH error: %s", e)

backend/tests/unit/test_vp_mqtt_bridge.py

@@ -0,0 +1,592 @@
+"""Tests for the VP MQTT bridge — non-proxy mirror of target printer state to slicer."""
+
+import asyncio
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.virtual_printer.mqtt_bridge import MQTTBridge, _ip_to_uint32_le
+from backend.app.services.virtual_printer.mqtt_server import SimpleMQTTServer
+
+H2D_SERIAL = "0948BB540200427"
+VP_SERIAL = "09400A391800003"
+H2D_IP = "192.168.255.133"
+VP_IP = "192.168.255.16"
+
+
+def _make_server(serial: str = VP_SERIAL, bind_address: str = VP_IP) -> SimpleMQTTServer:
+    return SimpleMQTTServer(
+        serial=serial,
+        access_code="deadbeef",
+        cert_path=Path("/tmp/unused.crt"),  # nosec B108
+        key_path=Path("/tmp/unused.key"),  # nosec B108
+        model="O1D",
+        bind_address=bind_address,
+    )
+
+
+def _make_paho_client(
+    serial: str = H2D_SERIAL,
+    ip: str = H2D_IP,
+    *,
+    connected: bool = True,
+) -> MagicMock:
+    """Build a mock BambuMQTTClient that satisfies MQTTBridge's interface."""
+    client = MagicMock()
+    client.serial_number = serial
+    client.ip_address = ip
+    client.state = MagicMock()
+    client.state.connected = connected
+    client.publish_raw = MagicMock(return_value=True)
+    client._raw_handlers: list = []
+
+    def _register(handler):
+        client._raw_handlers.append(handler)
+
+    def _unregister(handler):
+        if handler in client._raw_handlers:
+            client._raw_handlers.remove(handler)
+
+    client.register_raw_message_handler.side_effect = _register
+    client.unregister_raw_message_handler.side_effect = _unregister
+    # No-op for _request_version / request_status_update so the post-bind nudge doesn't crash.
+    client._request_version = MagicMock()
+    client.request_status_update = MagicMock()
+    return client
+
+
+def _make_printer_manager(client) -> MagicMock:
+    pm = MagicMock()
+    pm.get_client = MagicMock(return_value=client)
+    return pm
+
+
+def _make_bridge(server: SimpleMQTTServer, target: MagicMock | None = None) -> MQTTBridge:
+    target = target if target is not None else _make_paho_client()
+    pm = _make_printer_manager(target)
+    return MQTTBridge(
+        vp_id=1,
+        vp_name="vp1",
+        vp_serial=VP_SERIAL,
+        target_printer_id=42,
+        mqtt_server=server,
+        printer_manager=pm,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Lifecycle
+# ---------------------------------------------------------------------------
+
+
+class TestBridgeLifecycle:
+    @pytest.mark.asyncio
+    async def test_start_registers_handler_on_target_client(self):
+        target = _make_paho_client()
+        bridge = _make_bridge(_make_server(), target)
+        await bridge.start()
+        assert len(target._raw_handlers) == 1
+        assert bridge.is_active is True
+        await bridge.stop()
+        assert len(target._raw_handlers) == 0
+
+    @pytest.mark.asyncio
+    async def test_start_with_no_target_client_does_not_crash(self):
+        pm = MagicMock()
+        pm.get_client = MagicMock(return_value=None)
+        bridge = MQTTBridge(
+            vp_id=1,
+            vp_name="vp1",
+            vp_serial=VP_SERIAL,
+            target_printer_id=42,
+            mqtt_server=_make_server(),
+            printer_manager=pm,
+        )
+        await bridge.start()
+        assert bridge.is_active is False
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_resolve_rebinds_when_paho_client_replaced(self):
+        """BambuMQTTClient is destroyed and recreated on connect_printer; bridge must rebind."""
+        old_client = _make_paho_client(serial="REAL_OLD")
+        new_client = _make_paho_client(serial="REAL_NEW")
+        pm = _make_printer_manager(old_client)
+        bridge = MQTTBridge(
+            vp_id=1,
+            vp_name="vp1",
+            vp_serial=VP_SERIAL,
+            target_printer_id=42,
+            mqtt_server=_make_server(),
+            printer_manager=pm,
+        )
+        await bridge.start()
+        assert len(old_client._raw_handlers) == 1
+        assert bridge._target_serial == "REAL_OLD"
+
+        pm.get_client.return_value = new_client
+        bridge._resolve_client()
+        assert len(old_client._raw_handlers) == 0
+        assert len(new_client._raw_handlers) == 1
+        assert bridge._target_serial == "REAL_NEW"
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_post_bind_nudge_requests_version_and_status(self):
+        target = _make_paho_client()
+        bridge = _make_bridge(_make_server(), target)
+        await bridge.start()
+        target._request_version.assert_called_once()
+        target.request_status_update.assert_called_once()
+        await bridge.stop()
+
+
+# ---------------------------------------------------------------------------
+# Caching: push_status
+# ---------------------------------------------------------------------------
+
+
+class TestPushStatusCache:
+    """push_status snapshots feed `_send_status_report` via the cache, not a fan-out."""
+
+    @pytest.mark.asyncio
+    async def test_push_status_is_cached_not_fanned_out(self):
+        server = _make_server()
+        server.push_raw_to_clients = AsyncMock()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        payload = json.dumps({"print": {"command": "push_status", "ams": {"ams": []}, "gcode_state": "IDLE"}}).encode()
+        bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
+        await asyncio.sleep(0.01)
+
+        server.push_raw_to_clients.assert_not_awaited()
+        cached = bridge.get_latest_print_state()
+        assert cached is not None
+        assert cached["command"] == "push_status"
+        assert cached["gcode_state"] == "IDLE"
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_serial_rewritten_in_cached_push(self):
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        payload = json.dumps(
+            {
+                "print": {
+                    "command": "push_status",
+                    "upgrade_state": {"sn": H2D_SERIAL, "status": "IDLE"},
+                }
+            }
+        ).encode()
+        bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
+        await asyncio.sleep(0.01)
+
+        cached = bridge.get_latest_print_state()
+        assert cached["upgrade_state"]["sn"] == VP_SERIAL
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_net_info_ip_rewritten_to_vp_ip(self):
+        """BambuStudio reads `net.info[].ip` (LE uint32) for the FTP destination —
+        must be rewritten to the VP's bind IP or the slicer bypasses the VP."""
+        server = _make_server(bind_address=VP_IP)
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        h2d_le = _ip_to_uint32_le(H2D_IP)
+        vp_le = _ip_to_uint32_le(VP_IP)
+        payload = json.dumps(
+            {
+                "print": {
+                    "command": "push_status",
+                    "net": {"info": [{"ip": h2d_le, "mask": 0xFFFFFF}, {"ip": 0, "mask": 0}]},
+                }
+            }
+        ).encode()
+        bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
+        await asyncio.sleep(0.01)
+
+        cached = bridge.get_latest_print_state()
+        assert cached["net"]["info"][0]["ip"] == vp_le
+        assert cached["net"]["info"][1]["ip"] == 0  # untouched
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_request_topic_message_is_ignored(self):
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        payload = json.dumps({"print": {"command": "push_status"}}).encode()
+        bridge._on_printer_raw(f"device/{H2D_SERIAL}/request", payload)
+        await asyncio.sleep(0.01)
+
+        assert bridge.get_latest_print_state() is None
+        await bridge.stop()
+
+
+# ---------------------------------------------------------------------------
+# Caching: get_version response
+# ---------------------------------------------------------------------------
+
+
+class TestVersionCache:
+    @pytest.mark.asyncio
+    async def test_get_version_response_caches_modules(self):
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        payload = json.dumps(
+            {
+                "info": {
+                    "command": "get_version",
+                    "module": [
+                        {"name": "ota", "sn": H2D_SERIAL, "sw_ver": "01.03.00.00"},
+                        {"name": "n3f/0", "sn": "AMS_HW_1", "sw_ver": "04.00.21.87"},
+                    ],
+                }
+            }
+        ).encode()
+        bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", payload)
+        await asyncio.sleep(0.01)
+
+        modules = bridge.get_latest_version_modules()
+        assert modules is not None
+        assert len(modules) == 2
+        # Device-level sn rewritten; AMS-hardware sn left alone.
+        assert modules[0]["sn"] == VP_SERIAL
+        assert modules[1]["sn"] == "AMS_HW_1"
+
+        await bridge.stop()
+
+
+# ---------------------------------------------------------------------------
+# Selective fan-out (everything that's not push_status / get_version)
+# ---------------------------------------------------------------------------
+
+
+class TestCommandResponseFanout:
+    @pytest.mark.asyncio
+    async def test_extrusion_cali_get_response_is_fanned_out(self):
+        """Slicer's extrusion_cali_get goes to the printer; the printer's response
+        must reach the slicer or BambuStudio's pre-flight blocks Send."""
+        server = _make_server()
+        server.push_raw_to_clients = AsyncMock()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        body = json.dumps({"print": {"command": "extrusion_cali_get", "filaments": []}}).encode()
+        bridge._on_printer_raw(f"device/{H2D_SERIAL}/report", body)
+        await asyncio.sleep(0.01)
+
+        server.push_raw_to_clients.assert_awaited_once()
+        topic, _payload = server.push_raw_to_clients.await_args.args
+        assert topic == f"device/{VP_SERIAL}/report"
+
+        await bridge.stop()
+
+
+# ---------------------------------------------------------------------------
+# Forwarding: slicer → printer
+# ---------------------------------------------------------------------------
+
+
+class TestForwardToPrinter:
+    @pytest.mark.asyncio
+    async def test_forward_publishes_to_real_serial_request_topic(self):
+        target = _make_paho_client()
+        bridge = _make_bridge(_make_server(), target)
+        await bridge.start()
+
+        ok = bridge.forward_to_printer({"print": {"command": "stop"}})
+        assert ok is True
+        target.publish_raw.assert_called_once()
+        topic, payload = target.publish_raw.call_args.args
+        assert topic == f"device/{H2D_SERIAL}/request"
+        assert json.loads(payload) == {"print": {"command": "stop"}}
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_forward_returns_false_when_not_bound(self):
+        pm = MagicMock()
+        pm.get_client = MagicMock(return_value=None)
+        bridge = MQTTBridge(
+            vp_id=1,
+            vp_name="vp1",
+            vp_serial=VP_SERIAL,
+            target_printer_id=42,
+            mqtt_server=_make_server(),
+            printer_manager=pm,
+        )
+        await bridge.start()
+        assert bridge.forward_to_printer({"print": {"command": "stop"}}) is False
+        await bridge.stop()
+
+
+# ---------------------------------------------------------------------------
+# SimpleMQTTServer status response: cached-as-base
+# ---------------------------------------------------------------------------
+
+
+class TestStatusReportCachedAsBase:
+    """`_send_status_report` sends near-byte-identical real data when bridge cache exists."""
+
+    def _capture_published(self, server: SimpleMQTTServer):
+        """Wrap _publish_to_report to capture (topic, payload_dict)."""
+        published: list = []
+
+        async def _capture(writer, payload, serial=""):
+            published.append((serial or server.serial, payload))
+
+        server._publish_to_report = _capture  # type: ignore[assignment]
+        return published
+
+    @pytest.mark.asyncio
+    async def test_uses_real_cache_when_bridge_active(self):
+        server = _make_server()
+        bridge = MagicMock()
+        bridge.get_latest_print_state.return_value = {
+            "command": "push_status",
+            "msg": 0,
+            "ams": {"ams": [{"id": "0"}]},
+            "device": {"extruder": {"info": [{"id": 0}, {"id": 1}]}},
+            "nozzle_diameter": "0.4",
+            "nozzle_type": "HH01",  # real H2D value, not synthetic 'hardened_steel'
+        }
+        server.set_bridge(bridge)
+        published = self._capture_published(server)
+
+        await server._send_status_report(MagicMock())
+        assert len(published) == 1
+        _serial, payload = published[0]
+        # AMS / device / nozzle_type all from cache
+        assert payload["print"]["nozzle_type"] == "HH01"
+        assert payload["print"]["device"]["extruder"]["info"][1]["id"] == 1
+        # Protocol fields under our control
+        assert payload["print"]["command"] == "push_status"
+        assert payload["print"]["gcode_state"] == "IDLE"
+
+    @pytest.mark.asyncio
+    async def test_falls_back_to_synthetic_when_no_cache(self):
+        server = _make_server()
+        bridge = MagicMock()
+        bridge.get_latest_print_state.return_value = None
+        server.set_bridge(bridge)
+        published = self._capture_published(server)
+
+        await server._send_status_report(MagicMock())
+        assert len(published) == 1
+        _serial, payload = published[0]
+        # Synthetic baseline has stub fields like nozzle_type='hardened_steel'
+        # and a `storage` field that the real H2D doesn't push.
+        assert payload["print"]["nozzle_type"] == "hardened_steel"
+        assert "storage" in payload["print"]
+
+    @pytest.mark.asyncio
+    async def test_overrides_protocol_fields_even_when_cache_present(self):
+        """Cached value's gcode_state must NOT win over our local upload-state-machine value."""
+        server = _make_server()
+        server._gcode_state = "PREPARE"
+        server._current_file = "foo.3mf"
+        bridge = MagicMock()
+        bridge.get_latest_print_state.return_value = {
+            "command": "push_status",
+            "gcode_state": "IDLE",  # printer is idle; we are mid-FTP-upload
+            "gcode_file": "",
+            "gcode_file_prepare_percent": "0",
+        }
+        server.set_bridge(bridge)
+        published = self._capture_published(server)
+
+        await server._send_status_report(MagicMock())
+        _serial, payload = published[0]
+        assert payload["print"]["gcode_state"] == "PREPARE"
+        assert payload["print"]["gcode_file"] == "foo.3mf"
+
+
+# ---------------------------------------------------------------------------
+# Wire format
+# ---------------------------------------------------------------------------
+
+
+class TestWireFormat:
+    """BambuStudio's Send pre-flight rejects compact JSON — must match real printer's
+    indented format (32K bytes for an idle H2D vs 14K compact)."""
+
+    @pytest.mark.asyncio
+    async def test_publish_uses_indent_4_json_format(self):
+        server = _make_server()
+        captured: list = []
+
+        async def _capture_drain():
+            pass
+
+        writer = MagicMock()
+        writer.write = lambda data: captured.append(data)
+        writer.drain = AsyncMock()
+
+        await server._publish_to_report(writer, {"print": {"command": "push_status", "ams": {}}})
+
+        body = b"".join(captured)
+        assert b'\n    "print"' in body, "publish_to_report must use indent=4 JSON"
+
+
+# ---------------------------------------------------------------------------
+# Routing: _handle_publish
+# ---------------------------------------------------------------------------
+
+
+class TestPublishRouting:
+    """Slicer-issued commands: project_file/gcode_file handled locally, everything
+    else forwarded to the real printer."""
+
+    def _build_publish_payload(self, topic: str, body: bytes) -> bytes:
+        topic_bytes = topic.encode("utf-8")
+        return bytes([len(topic_bytes) >> 8, len(topic_bytes) & 0xFF]) + topic_bytes + body
+
+    def _attach_active_bridge(self, server: SimpleMQTTServer) -> MagicMock:
+        bridge = MagicMock()
+        bridge.is_active = True
+        bridge.forward_to_printer = MagicMock(return_value=True)
+        server.set_bridge(bridge)
+        return bridge
+
+    @pytest.mark.asyncio
+    async def test_project_file_handled_locally_not_forwarded(self):
+        server = _make_server()
+        bridge = self._attach_active_bridge(server)
+        writer = MagicMock()
+        writer.write = MagicMock()
+        writer.drain = AsyncMock()
+
+        body = json.dumps({"print": {"command": "project_file", "subtask_name": "f", "sequence_id": "1"}}).encode()
+        payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
+
+        with patch.object(server, "_send_print_response", new=AsyncMock()) as mock_resp:
+            await server._handle_publish(0x30, payload, writer, "client1")
+
+        bridge.forward_to_printer.assert_not_called()
+        mock_resp.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_gcode_file_handled_locally_not_forwarded(self):
+        server = _make_server()
+        bridge = self._attach_active_bridge(server)
+        writer = MagicMock()
+        writer.write = MagicMock()
+        writer.drain = AsyncMock()
+
+        body = json.dumps({"print": {"command": "gcode_file", "subtask_name": "f.gcode", "sequence_id": "1"}}).encode()
+        payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
+
+        with patch.object(server, "_send_print_response", new=AsyncMock()):
+            await server._handle_publish(0x30, payload, writer, "client1")
+
+        bridge.forward_to_printer.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_pushall_handled_locally_not_forwarded(self):
+        server = _make_server()
+        bridge = self._attach_active_bridge(server)
+        writer = MagicMock()
+        writer.write = MagicMock()
+        writer.drain = AsyncMock()
+
+        body = json.dumps({"pushing": {"command": "pushall", "sequence_id": "0"}}).encode()
+        payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
+
+        with patch.object(server, "_send_status_report", new=AsyncMock()) as mock_status:
+            await server._handle_publish(0x30, payload, writer, "client1")
+
+        # Synthetic answer fires (fast, low latency); no forwarding (the
+        # cache already mirrors what the printer would respond with).
+        bridge.forward_to_printer.assert_not_called()
+        mock_status.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_get_version_handled_locally_not_forwarded(self):
+        server = _make_server()
+        bridge = self._attach_active_bridge(server)
+        writer = MagicMock()
+        writer.write = MagicMock()
+        writer.drain = AsyncMock()
+
+        body = json.dumps({"info": {"command": "get_version", "sequence_id": "1"}}).encode()
+        payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
+
+        with patch.object(server, "_send_version_response", new=AsyncMock()) as mock_ver:
+            await server._handle_publish(0x30, payload, writer, "client1")
+
+        bridge.forward_to_printer.assert_not_called()
+        mock_ver.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_extrusion_cali_get_is_forwarded(self):
+        """extrusion_cali_get fetches per-filament k-profiles — must reach the printer."""
+        server = _make_server()
+        bridge = self._attach_active_bridge(server)
+        writer = MagicMock()
+        writer.write = MagicMock()
+        writer.drain = AsyncMock()
+
+        body = json.dumps(
+            {
+                "print": {
+                    "command": "extrusion_cali_get",
+                    "filament_id": "",
+                    "nozzle_diameter": "0.4",
+                    "sequence_id": "5",
+                }
+            }
+        ).encode()
+        payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
+
+        await server._handle_publish(0x30, payload, writer, "client1")
+
+        bridge.forward_to_printer.assert_called_once()
+        forwarded = bridge.forward_to_printer.call_args.args[0]
+        assert forwarded["print"]["command"] == "extrusion_cali_get"
+
+    @pytest.mark.asyncio
+    async def test_print_stop_is_forwarded(self):
+        server = _make_server()
+        bridge = self._attach_active_bridge(server)
+        writer = MagicMock()
+        writer.write = MagicMock()
+        writer.drain = AsyncMock()
+
+        body = json.dumps({"print": {"command": "stop", "sequence_id": "5"}}).encode()
+        payload = self._build_publish_payload(f"device/{VP_SERIAL}/request", body)
+
+        await server._handle_publish(0x30, payload, writer, "client1")
+
+        bridge.forward_to_printer.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# IP encoding helper
+# ---------------------------------------------------------------------------
+
+
+class TestIpEncoding:
+    def test_le_uint32_matches_real_h2d_capture(self):
+        # 192.168.255.133 captured from real H2D's net.info[0].ip = 2248124608
+        assert _ip_to_uint32_le("192.168.255.133") == 2248124608
+
+    def test_vp_ip_round_trip(self):
+        assert _ip_to_uint32_le("192.168.255.16") == 285190336
+
+    def test_invalid_ip_raises(self):
+        with pytest.raises(ValueError):
+            _ip_to_uint32_le("not.an.ip.actually")

docker-compose.yml

@@ -10,8 +10,10 @@ services:
     # Override with: PUID=$(id -u) PGID=$(id -g) docker compose up -d
     user: "${PUID:-1000}:${PGID:-1000}"
     #
-    # Proxy mode: allow binding to privileged ports (322, 990) as non-root user.
-    # Without this, the FTP and RTSP proxies silently fail.
+    # Allow binding to privileged ports (322, 990) as non-root user — required
+    # for FTPS in every VP mode and for the RTSPS camera proxy in proxy mode +
+    # non-proxy modes that have a target printer configured. Without this, the
+    # FTP and RTSP listeners silently fail.
     cap_add:
       - NET_BIND_SERVICE
     #
@@ -28,7 +30,7 @@ services:
     #  - "8883:8883"                  # Virtual printer MQTT
     #  - "990:990"                    # Virtual printer FTP control
     #  - "6000:6000"                  # Virtual printer file transfer tunnel
-    #  - "322:322"                    # Virtual printer RTSP camera (X1/H2/P2)
+    #  - "322:322"                    # Virtual printer RTSP camera (X1/H2/P2; proxy mode + non-proxy modes with a target printer)
     #  - "2024-2026:2024-2026"        # Virtual printer proprietary ports (A1/P1S)
     #  - "50000-50100:50000-50100"    # Virtual printer FTP passive data
     volumes: