fix(vp): route non-proxy camera passthrough by target model — 6000 for A1/P1 (#1868)

Non-proxy VP mode hardcoded the camera-passthrough TCPProxy to
listen_port=322 / target_port=322 regardless of the target printer's
model. That port is correct for RTSPS models (X1/X2/H2/P2S), but A1 /
A1 Mini / P1P / P1S use Bambu's proprietary chamber-image protocol on
port 6000. Result: A1/P1 targets got a 322 listener with no upstream,
OrcaSlicer Liveview failed with [2:-10061], BambuStudio's camera button
timed out.

Reporter confirmed a raw socat forwarder `<VP-IP>:6000 → <P1S-IP>:6000`
restored the stream — the target camera works, the VP just wasn't
publishing it.

Proxy mode was unaffected because SlicerProxyManager already opens 6000
(nominally file-transfer; Bambu reuses the port for chamber-image), so
the passthrough coincidentally works there.

Fix: read the target's model from
`printer_manager.get_client(target_id).model` at the same point we read
target_ip, then use `get_camera_port(target_model)` — the same source of
truth as routes/camera.py — to pick 322 or 6000. Model comes from the
physical printer, NOT self.model (the VP's spoofed identity has no
bearing on how the real device serves its camera).

Renamed the log tag from "RTSP" to f"Camera-{camera_port}" so support
bundles show which protocol the VP is fronting at a glance. Kept the
_rtsp_proxy attribute name to keep the diff tight; the block comment
spells out that it doubles as chamber-image passthrough on A1/P1.

backend/app/services/virtual_printer/manager.py

@@ -1095,25 +1095,36 @@ class VirtualPrinterInstance:
             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.
+            # Camera passthrough. BambuStudio / OrcaSlicer connect the "camera"
+            # button to the device IP they bound on (the VP), not the IP in the
+            # printer's `ipcam.rtsp_url`. Without a listener the slicer gets
+            # connection refused → "LAN connection failed" (RTSP models) or
+            # OrcaSlicer error `[2:-10061]` (chamber-image models, #1868).
+            #
+            # The port depends on the TARGET printer's model:
+            #   RTSPS (X1/X2/H2/P2S)        → 322
+            #   chamber-image (A1/P1P/P1S)  → 6000
+            #
+            # `get_camera_port()` is the same source of truth used by
+            # `routes/camera.py`, so slicer and Bambuddy UI agree.
             target_client = self._printer_manager.get_client(self.target_printer_id)
             target_ip = getattr(target_client, "ip_address", None) if target_client else None
+            target_model = getattr(target_client, "model", None) if target_client else None
             if target_ip:
+                from backend.app.services.camera import get_camera_port
+
+                camera_port = get_camera_port(target_model)
                 self._rtsp_proxy = TCPProxy(
-                    name="RTSP",
-                    listen_port=322,
+                    name=f"Camera-{camera_port}",
+                    listen_port=camera_port,
                     target_host=target_ip,
-                    target_port=322,
+                    target_port=camera_port,
                     bind_address=bind_addr,
                 )
                 self._tasks.append(
                     asyncio.create_task(
-                        run_with_logging(self._rtsp_proxy.start(), "RTSP"),
-                        name=f"vp_{self.id}_rtsp",
+                        run_with_logging(self._rtsp_proxy.start(), f"Camera-{camera_port}"),
+                        name=f"vp_{self.id}_camera",
                     )
                 )
 
@@ -1197,7 +1208,7 @@ class VirtualPrinterInstance:
             try:
                 await self._rtsp_proxy.stop()
             except Exception:
-                logger.exception("[VP %s] RTSP proxy stop failed", self.name)
+                logger.exception("[VP %s] Camera proxy stop failed", self.name)
             self._rtsp_proxy = None
         if self._ftp:
             await self._ftp.stop()

backend/tests/unit/services/test_virtual_printer.py

@@ -3175,6 +3175,146 @@ class TestVirtualPrinterManagerDirectories:
         assert (tmp_path / "certs" / "42").exists()
 
 
+class TestVirtualPrinterCameraPassthrough:
+    """Tests for the non-proxy VP camera pass-through port selection (#1868).
+
+    The camera port must follow the TARGET printer's model, not the VP's
+    spoofed model: RTSP models (X1/X2/H2/P2S) use 322, chamber-image models
+    (A1/P1) use 6000. Before the fix, the manager hardcoded 322, so P1S /
+    A1 targets got a 322 listener with no upstream and BambuStudio /
+    OrcaSlicer Liveview failed with error `[2:-10061]` (the reporter's
+    symptom in #1868).
+    """
+
+    @staticmethod
+    def _patch_start_server(monkeypatch):
+        """Patch every non-camera constructor start_server touches so the
+        test only exercises the camera-port branch. Returns the list that
+        ``TCPProxy`` calls get captured into."""
+        from backend.app.services.virtual_printer import manager as vp_manager
+
+        tcp_calls: list[dict] = []
+
+        class FakeTCPProxy:
+            def __init__(self, **kwargs):
+                tcp_calls.append(kwargs)
+                self.kwargs = kwargs
+
+            async def start(self):
+                return None
+
+            async def stop(self):
+                return None
+
+        # Camera pass-through is the only TCPProxy start_server constructs.
+        monkeypatch.setattr(vp_manager, "TCPProxy", FakeTCPProxy)
+
+        # No-op every other service constructor + resolve_cert_and_advertise.
+        # start_server() awaits `.ready.wait()` on FTP/MQTT/Bind/SSDP after
+        # spawning them, so each fake instance must carry an already-set
+        # asyncio.Event as `.ready` — a plain MagicMock returns another
+        # MagicMock for `.wait()`, which `asyncio.gather` then rejects with
+        # `TypeError: An asyncio.Future, a coroutine or an awaitable is
+        # required`.
+        def _service_factory():
+            def make(*args, **kwargs):
+                inst = MagicMock()
+                inst.start = AsyncMock(return_value=None)
+                inst.stop = AsyncMock(return_value=None)
+                ready = asyncio.Event()
+                ready.set()
+                inst.ready = ready
+                return inst
+
+            return make
+
+        for name in (
+            "VirtualPrinterFTPServer",
+            "SimpleMQTTServer",
+            "MQTTBridge",
+            "BindServer",
+            "VirtualPrinterSSDPServer",
+            "SSDPProxy",
+        ):
+            monkeypatch.setattr(vp_manager, name, _service_factory())
+
+        return tcp_calls
+
+    async def _run_start_server(
+        self,
+        tmp_path,
+        monkeypatch,
+        *,
+        target_model: str,
+        target_ip: str = "192.168.1.100",
+    ) -> list[dict]:
+        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance
+
+        tcp_calls = self._patch_start_server(monkeypatch)
+
+        instance = VirtualPrinterInstance(
+            vp_id=99,
+            name="CamTest",
+            mode="archive",
+            model="BL-P001",  # VP's spoofed identity — irrelevant here
+            access_code="12345678",
+            serial_suffix="391800099",
+            target_printer_id=7,
+            base_dir=tmp_path,
+        )
+
+        # printer_manager stub returns a client with the model + ip we want.
+        client = MagicMock()
+        client.ip_address = target_ip
+        client.model = target_model
+        printer_manager = MagicMock()
+        printer_manager.get_client.return_value = client
+        instance._printer_manager = printer_manager
+
+        # Cert / advertise resolution — start_server calls this early. Patch
+        # to a fixed tuple so no filesystem I/O is required.
+        monkeypatch.setattr(
+            instance,
+            "_resolve_cert_and_advertise",
+            lambda: (Path("/tmp/cert.pem"), Path("/tmp/key.pem"), "192.168.1.1"),  # nosec B108
+        )
+
+        try:
+            await instance.start_server()
+        finally:
+            for task in instance._tasks:
+                task.cancel()
+            await asyncio.gather(*instance._tasks, return_exceptions=True)
+
+        return tcp_calls
+
+    @pytest.mark.asyncio
+    async def test_rtsp_model_p2s_opens_port_322(self, tmp_path, monkeypatch):
+        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="P2S")
+        assert any(c["listen_port"] == 322 and c["target_port"] == 322 for c in calls), (
+            f"Expected 322 pass-through for RTSP model P2S, got {calls}"
+        )
+
+    @pytest.mark.asyncio
+    async def test_chamber_image_model_p1s_opens_port_6000(self, tmp_path, monkeypatch):
+        """#1868 regression guard: P1S target must expose 6000, not 322."""
+        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="P1S")
+        assert any(c["listen_port"] == 6000 and c["target_port"] == 6000 for c in calls), (
+            f"Expected 6000 pass-through for chamber-image model P1S (#1868), got {calls}"
+        )
+        assert not any(c["listen_port"] == 322 for c in calls), f"P1S should NOT get a 322 listener, got {calls}"
+
+    @pytest.mark.asyncio
+    async def test_chamber_image_model_a1_opens_port_6000(self, tmp_path, monkeypatch):
+        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="A1")
+        assert any(c["listen_port"] == 6000 and c["target_port"] == 6000 for c in calls)
+
+    @pytest.mark.asyncio
+    async def test_rtsp_model_x1c_opens_port_322(self, tmp_path, monkeypatch):
+        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="X1C")
+        assert any(c["listen_port"] == 322 and c["target_port"] == 322 for c in calls)
+
+
 class TestVirtualPrinterInstanceProxyMode:
     """Tests for VirtualPrinterInstance proxy mode."""