Merge branch 'dev' into fix/camera-rotation-finish-photo-timelapse

backend/app/api/routes/_oidc_helpers.py

@@ -13,7 +13,12 @@ from __future__ import annotations
 import ipaddress
 from urllib.parse import urlparse
 
-from backend.app.api.routes._url_safety import CLOUD_METADATA_IPS, NUMERIC_IP_RE, unwrap_ipv4_mapped
+from backend.app.api.routes._url_safety import (
+    CLOUD_METADATA_HOSTNAMES,
+    CLOUD_METADATA_IPS,
+    NUMERIC_IP_RE,
+    unwrap_ipv4_mapped,
+)
 
 
 def assert_safe_public_https_url(url: str) -> None:
@@ -38,10 +43,12 @@ def assert_safe_public_https_url(url: str) -> None:
     - IPv4-mapped IPv6 (``::ffff:127.0.0.1``) — unwrapped before the IP-class
       check so an attacker can't bypass via IPv6 encoding.
 
-    Hostname-based addresses are accepted without DNS resolution — the
-    operator is trusted to configure a sensible IdP host, and resolving here
-    would both add a TOCTOU gap (DNS can change between validation and
-    request) and make the validator issue network requests of its own.
+    Hostname-based addresses are otherwise accepted without DNS resolution —
+    the operator is trusted to configure a sensible IdP host, and resolving
+    here would both add a TOCTOU gap (DNS can change between validation and
+    request) and make the validator issue network requests of its own. The
+    fixed cloud-metadata hostnames are the exception: matching them is a
+    literal string comparison, not a resolution.
     """
     parsed = urlparse(url)
     if parsed.scheme.lower() != "https":
@@ -49,6 +56,14 @@ def assert_safe_public_https_url(url: str) -> None:
 
     hostname = (parsed.hostname or "").lower()
 
+    # "https:///path" parses to an empty hostname; without this it reaches the
+    # ip_address() ValueError branch and is accepted as a symbolic hostname.
+    if not hostname:
+        raise ValueError("icon URL must include a hostname")
+
+    if hostname in CLOUD_METADATA_HOSTNAMES:
+        raise ValueError("icon URL must not point to a cloud metadata endpoint")
+
     if NUMERIC_IP_RE.match(hostname):
         raise ValueError("icon URL must not use numeric-encoded IP addresses")
 

backend/app/api/routes/_url_safety.py

@@ -40,6 +40,18 @@ CLOUD_METADATA_IPS = frozenset(
     }
 )
 
+# The DNS-name form of the same targets. Neither guard resolves hostnames (see
+# the TOCTOU note on each), so an IP blocklist alone cannot catch these — but a
+# literal-string match needs no resolution and costs nothing. These names only
+# resolve inside the respective cloud, so there is no legitimate reason for any
+# Bambuddy integration to point at one.
+CLOUD_METADATA_HOSTNAMES = frozenset(
+    {
+        "metadata.google.internal",  # GCP
+        "metadata.goog",  # GCP short form
+    }
+)
+
 
 # libc and browsers parse numeric-encoded IP forms (decimal ``2130706433``
 # for 127.0.0.1, hex ``0x7f000001``) but Python's ``ipaddress.ip_address``
@@ -90,10 +102,11 @@ def assert_safe_lan_service_url(url: str, *, label: str) -> None:
       indicative of misuse.
     - IPv4-mapped IPv6 encodings of any of the above.
 
-    Symbolic hostnames are accepted without DNS resolution, matching the
-    public-internet guard: resolution here would be both a TOCTOU (DNS can
+    Symbolic hostnames are otherwise accepted without DNS resolution, matching
+    the public-internet guard: resolution here would be both a TOCTOU (DNS can
     change between validation and request) and a request the validator
-    shouldn't be making.
+    shouldn't be making. The one exception is the fixed set of cloud-metadata
+    hostnames, which is a literal-string match and needs no resolution.
     """
     parsed = urlparse(url)
     if parsed.scheme.lower() not in ("http", "https"):
@@ -101,6 +114,15 @@ def assert_safe_lan_service_url(url: str, *, label: str) -> None:
 
     hostname = (parsed.hostname or "").lower()
 
+    # "http:///path" parses to an empty hostname. Never a valid destination,
+    # and without this it falls through the ip_address() ValueError branch
+    # below and is accepted as if it were a symbolic hostname.
+    if not hostname:
+        raise ValueError(f"{label} must include a hostname")
+
+    if hostname in CLOUD_METADATA_HOSTNAMES:
+        raise ValueError(f"{label} must not point to a cloud metadata endpoint")
+
     if NUMERIC_IP_RE.match(hostname):
         raise ValueError(f"{label} must not use numeric-encoded IP addresses; use standard dotted-decimal notation")
 

backend/app/api/routes/github_backup.py

@@ -49,7 +49,21 @@ async def _enforce_private_repo(repo_url: str, token: str, provider: str) -> Non
 
     Used by POST and PATCH /config so a backup configuration can never be
     saved against a public repository.
+
+    The URL is policy-checked first: the Gitea and Forgejo backends derive
+    their API base from this value (``get_api_base``) and then request it with
+    the supplied token, so an unchecked repository_url is an outbound fetch to
+    an operator-supplied host. A self-hosted Gitea on the LAN is the normal
+    case, so the LAN-service tier applies — this only rules out the targets
+    that are wrong under any topology.
     """
+    from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
+    try:
+        assert_safe_lan_service_url(repo_url, label="Repository URL")
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
     result = await github_backup_service.test_connection(repo_url, token, provider=provider)
     if not result.get("success"):
         message = result.get("message") or "Connection test failed"

backend/app/api/routes/printers.py

@@ -64,6 +64,7 @@ from backend.app.services.printer_manager import (
 )
 from backend.app.utils.filament_ids import filament_id_to_setting_id
 from backend.app.utils.http import build_content_disposition
+from backend.app.utils.printer_models import uses_exhaust_fan_label
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/printers", tags=["printers"])
@@ -798,6 +799,8 @@ async def get_printer_status(
         big_fan1_speed=state.big_fan1_speed,
         big_fan2_speed=state.big_fan2_speed,
         heatbreak_fan_speed=state.heatbreak_fan_speed,
+        left_aux_fan_speed=state.left_aux_fan_speed,
+        exhaust_fan_present=state.exhaust_fan_present,
         firmware_version=state.firmware_version,
         developer_mode=state.developer_mode if state else None,
         ams_filament_backup=state.ams_filament_backup if state else None,
@@ -3193,16 +3196,28 @@ async def set_chamber_temperature(
 @router.post("/{printer_id}/fan-speed")
 async def set_fan_speed(
     printer_id: int,
-    fan: str = Query(..., description="Fan to control: part, aux, or chamber"),
+    fan: str = Query(..., description="Fan to control: part, aux, aux2 (left aux), or chamber"),
     speed: int = Query(..., ge=0, le=100, description="Fan speed percentage"),
     _=RequirePermissionIfAuthEnabled(Permission.PRINTERS_CONTROL),
     db: AsyncSession = Depends(get_db),
 ):
-    """Set a fan speed by percentage."""
-    fan_ids = {"part": 1, "aux": 2, "chamber": 3}
+    """Set a fan speed by percentage.
+
+    Fan index 10 ("aux2") is the optional left auxiliary part cooling fan on
+    P2S/X2D — driven with "M106 P10" exactly like Bambu's official machine
+    profile gcode does. It only exists when the printer reports airduct part 10,
+    so the request is rejected rather than sending M106 P10 into the void on a
+    machine that has no such fan.
+
+    That gate also rejects for the short window between connecting and the
+    first airduct push, when nothing is known about the fan yet. The card hides
+    the badge over the same window, so there is no control to click; a direct
+    API caller gets a 400 and should retry once the status reports the fan.
+    """
+    fan_ids = {"part": 1, "aux": 2, "chamber": 3, "aux2": 10}
     fan_id = fan_ids.get(fan)
     if fan_id is None:
-        raise HTTPException(400, "fan must be 'part', 'aux', or 'chamber'")
+        raise HTTPException(400, "fan must be 'part', 'aux', 'aux2', or 'chamber'")
 
     result = await db.execute(select(Printer).where(Printer.id == printer_id))
     printer = result.scalar_one_or_none()
@@ -3213,12 +3228,31 @@ async def set_fan_speed(
     if not client:
         raise HTTPException(400, "Printer not connected")
 
+    # Presence gate for the accessory fan. Without this, aux2 is accepted for
+    # every model and an A1 would be sent M106 P10 for a fan it does not have.
+    # The UI already hides the badge; this closes the same hole on the API.
+    if fan == "aux2" and getattr(client.state, "left_aux_fan_speed", None) is None:
+        raise HTTPException(
+            400,
+            "This printer does not report a left auxiliary fan "
+            "(no airduct part 10). The fan is an accessory kit on the P2S "
+            "and factory-fitted on the X2D.",
+        )
+
     pwm_speed = round(speed * 255 / 100)
     success = client.set_fan_speed(fan_id, pwm_speed)
     if not success:
         raise HTTPException(500, "Failed to set fan speed")
 
-    fan_names = {"part": "Part cooling fan", "aux": "Auxiliary fan", "chamber": "Chamber fan"}
+    # The enclosure fan is called "Exhaust" on P2S/X2D and "Chamber" elsewhere;
+    # match whatever the printer card badge shows so the toast agrees with the
+    # control the user just clicked.
+    fan_names = {
+        "part": "Part cooling fan",
+        "aux": "Auxiliary fan",
+        "aux2": "Left auxiliary fan",
+        "chamber": "Exhaust fan" if uses_exhaust_fan_label(printer.model) else "Chamber fan",
+    }
     return {"success": True, "message": f"{fan_names[fan]} set to {speed}%"}
 
 

backend/app/main.py

@@ -6858,6 +6858,18 @@ async def lifespan(app: FastAPI):
     # Start camera stream orphan cleanup
     start_camera_cleanup()
 
+    # One-shot sweep for timelapse session directories orphaned by a crash
+    # or restart that happened mid-print (in-memory session tracking can't
+    # survive that, and nothing else reaps the leftover frames/output file)
+    try:
+        from backend.app.services.layer_timelapse import cleanup_orphaned_timelapse_sessions
+
+        removed = cleanup_orphaned_timelapse_sessions()
+        if removed:
+            logging.getLogger(__name__).info("Removed %d orphaned timelapse session artifact(s)", removed)
+    except Exception as e:
+        logging.getLogger(__name__).warning("Orphaned timelapse session cleanup failed: %s", e)
+
     # Start expected-print TTL eviction (prevents memory leak when prints are
     # registered but on_print_start never fires)
     start_expected_prints_cleanup()

backend/app/schemas/printer.py

@@ -360,6 +360,11 @@ class PrinterStatus(BaseModel):
     big_fan1_speed: int | None = None  # Auxiliary fan
     big_fan2_speed: int | None = None  # Chamber/exhaust fan
     heatbreak_fan_speed: int | None = None  # Hotend heatbreak fan
+    # Left auxiliary part cooling fan (optional P2S/X2D accessory, airduct part id 10).
+    # None = not installed / not reported by this model.
+    left_aux_fan_speed: int | None = None
+    # Chamber exhaust fan present (P2S/X2D External Exhaust Fan kit; airduct part id 3).
+    exhaust_fan_present: bool = False
     # Firmware version (from info.module[name="ota"].sw_ver)
     firmware_version: str | None = None
     # Developer LAN mode: True = enabled, False = disabled (MQTT encryption), None = unknown

backend/app/services/bambu_mqtt.py

@@ -472,6 +472,21 @@ class PrinterState:
     big_fan1_speed: int | None = None  # Auxiliary fan
     big_fan2_speed: int | None = None  # Chamber/exhaust fan
     heatbreak_fan_speed: int | None = None  # Hotend heatbreak fan
+    # Left auxiliary part cooling fan (optional accessory on P2S/X2D). Reported ONLY
+    # via device.airduct.parts (decoded part id 10 = FAN_REMOTE_COOLING_1 in Bambu
+    # Studio's AIR_FUN enum) — the firmware does NOT mirror it into any flat
+    # big_fanX_speed field, which is why it was previously dropped. 0-100 percent.
+    left_aux_fan_speed: int | None = None
+    # Chamber exhaust fan, derived from the airduct parts list containing decoded
+    # id 3. On the P2S this is the External Exhaust Fan kit and a base machine
+    # omits it, which is the case this flag exists to detect.
+    #
+    # NOTE: the flag is not P2S/X2D-specific despite the name. The H2 series
+    # (H2C/H2D/H2S) also reports part 3, so this goes True there too. That is
+    # harmless because only the P2S/X2D badge consults it — those models keep
+    # their unconditional "Chamber Fan" badge — but do not read this as
+    # "an exhaust kit is fitted" without also checking the model.
+    exhaust_fan_present: bool = False
     # Tray change history during current print: [(global_tray_id, layer_num), ...]
     # Used by usage tracker to split filament weight on mid-print tray switch
     tray_change_log: list = field(default_factory=list)
@@ -3522,6 +3537,83 @@ class BambuMQTTClient:
                             f"[{self.serial_number}] airduct_mode changed: {self.state.airduct_mode} -> {new_mode}"
                         )
                     self.state.airduct_mode = new_mode
+                # Parse individual airduct fan parts (new-protocol models: P2S/X2D/H2*).
+                # Raw part ids are bit-packed — decoded id = raw_id >> 4 (bits 4-11),
+                # mirroring Bambu Studio DevFan::ParseV3_0. Decoded ids follow the
+                # AIR_FUN enum: 1=part cooling, 2=right aux, 3=chamber/exhaust,
+                # 10=left aux (FAN_REMOTE_COOLING_1). The airduct `parts` list only
+                # contains the fans that physically exist, so it doubles as a
+                # presence signal for the two P2S/X2D add-on kits:
+                #   - id 10 (left auxiliary part cooling fan) — reported ONLY here,
+                #     never mirrored into a flat big_fanX_speed field.
+                #   - id 3 (chamber exhaust fan) — its speed is mirrored into
+                #     big_fan2_speed, but the part is only listed when the External
+                #     Exhaust Fan kit (get_version module "eef") is installed.
+                # `state` is already a 0-100 percentage.
+                parts = airduct_data.get("parts")
+                if isinstance(parts, list):
+                    speeds: dict[int, int] = {}
+                    for part in parts:
+                        if not isinstance(part, dict):
+                            continue
+                        try:
+                            # Studio reads the id with get_flag_bits(id, 4, 8),
+                            # so mask after shifting for the same reason `state`
+                            # is masked below. Every id seen in the wild
+                            # (16/32/48/160) decodes identically either way —
+                            # this is consistency, not a live bug.
+                            part_id = (int(part["id"]) >> 4) & 0xFF
+                            # `state` is bit-packed like its sibling `range`
+                            # (end << 16 | start), so take only the low 8 bits —
+                            # the same decode Bambu Studio does with
+                            # get_flag_bits(state, 0, 8). Without the mask a
+                            # packed value would clamp to 100 instead of
+                            # decoding to the real percentage.
+                            part_state = int(part["state"]) & 0xFF
+                        except (KeyError, ValueError, TypeError):
+                            continue
+                        # Ids seen across the support-package archive:
+                        #   1 part cooling, 2 aux, 3 chamber/exhaust,
+                        #   6 (H2 series, unmapped), 10 left aux.
+                        speeds[part_id] = max(0, min(100, part_state))
+
+                    # Absence in this list is what tells us a kit is NOT fitted,
+                    # so it may only be trusted when the list is a full
+                    # inventory rather than a diff frame. `device.airduct` is
+                    # pushed field by field — the `modeCur` handler above exists
+                    # for exactly that reason — and a truncated `parts` read as
+                    # gospel would retract both accessory badges mid-print and
+                    # start rejecting `aux2` on a printer that has the fan.
+                    #
+                    # Every airduct layout in the support-package archive
+                    # (P2S base 1,2 / P2S+kit 1,2,3 / X2D 1,2,3,10 /
+                    # H2C,H2D,H2S 1,2,3,6 — 37 of 37 bundles) contains both the
+                    # part cooling fan and the aux fan, neither of which is
+                    # optional on any machine that reports an airduct at all.
+                    # A list carrying both is therefore a complete inventory; a
+                    # list missing either is a partial frame, and we take its
+                    # speeds without touching presence.
+                    is_full_inventory = 1 in speeds and 2 in speeds
+
+                    left_aux_speed = speeds.get(10)
+                    if left_aux_speed is None and not is_full_inventory:
+                        # Partial frame that didn't mention the left aux fan —
+                        # keep whatever we already knew about it.
+                        left_aux_speed = self.state.left_aux_fan_speed
+                    if left_aux_speed != self.state.left_aux_fan_speed:
+                        logger.debug(
+                            f"[{self.serial_number}] left_aux_fan_speed changed: "
+                            f"{self.state.left_aux_fan_speed} -> {left_aux_speed}"
+                        )
+                    # A FULL parts list without id 10 means the left aux fan is
+                    # not installed — report None so the UI can hide the widget.
+                    self.state.left_aux_fan_speed = left_aux_speed
+                    # id 3 present == chamber exhaust fan installed (base P2S
+                    # omits it). Only ever retracted on a full inventory.
+                    if 3 in speeds:
+                        self.state.exhaust_fan_present = True
+                    elif is_full_inventory:
+                        self.state.exhaust_fan_present = False
                 # Parse chamber temp - may be encoded as (target*65536+current) when > 500
                 # Check if we recently set the target locally (within 5 seconds)
                 local_set_time = self.state.temperatures.get("_chamber_target_set_time", 0)
@@ -5793,13 +5885,16 @@ class BambuMQTTClient:
         """Set fan speed.
 
         Args:
-            fan: Fan index (1=part cooling, 2=auxiliary, 3=chamber)
+            fan: Fan index (1=part cooling, 2=auxiliary, 3=chamber, 10=left auxiliary).
+                Index 10 is the optional left auxiliary part cooling fan on P2S/X2D
+                (airduct part id 10); Bambu's official machine profiles drive it with
+                "M106 P10" in start/layer-change gcode.
             speed: Speed 0-255 (0=off, 255=full)
 
         Returns:
             True if command was sent, False otherwise
         """
-        if fan not in (1, 2, 3):
+        if fan not in (1, 2, 3, 10):
             logger.warning("[%s] Invalid fan index: %s", self.serial_number, fan)
             return False
 
@@ -5818,6 +5913,10 @@ class BambuMQTTClient:
         """Set chamber fan speed (0-255)."""
         return self.set_fan_speed(3, speed)
 
+    def set_left_aux_fan(self, speed: int) -> bool:
+        """Set left auxiliary part cooling fan speed (0-255). P2S/X2D accessory."""
+        return self.set_fan_speed(10, speed)
+
     def set_airduct_mode(self, mode: str) -> bool:
         """Set air conditioning mode (cooling or heating).
 

backend/app/services/external_camera.py

@@ -8,6 +8,7 @@ to ensure they are well-formed before use.
 """
 
 import asyncio
+import functools
 import logging
 import re
 import shutil
@@ -175,6 +176,70 @@ def get_ffmpeg_path() -> str | None:
     return None
 
 
+# In-flight one-shot captures, keyed by (url, camera_type, snapshot_url) —
+# the tuple that actually identifies the physical resource being contended
+# (#2707 comment thread, following #2705's shape for the built-in path).
+#
+# V4L2 USB devices allow exactly one open handle, and is_stream_active() /
+# try_get_active_buffered_frame() (#2707) only stop a one-shot capturer from
+# competing with the fan-out live view. They do nothing for capturer-vs-
+# capturer with no viewer attached, where every consumer correctly concludes
+# it isn't competing with a viewer and then collides with the others -
+# exactly the #2705 report, just for this module's callers instead of
+# capture_camera_frame_bytes()'s (Obico polling, the in-print frame bank,
+# the finish-photo moment, plate detection, and the notification snapshot
+# all reach capture_frame() independently).
+#
+# snapshot_url is part of the key (not just url/camera_type) because it
+# routes to a completely different endpoint (#1177) - two printers that
+# share a camera_url but differ only in snapshot_url must not coalesce.
+_inflight_captures: dict[tuple[str, str, str | None], asyncio.Task[bytes | None]] = {}
+
+
+def capture_in_flight(url: str, camera_type: str, snapshot_url: str | None = None) -> bool:
+    """Return True iff a one-shot capture for this key is running right now.
+
+    Mirrors camera.py's capture_in_flight() for the built-in path - for a
+    caller that needs to know it will JOIN someone else's capture rather
+    than open its own connection. Ordinary consumers should ignore this:
+    they want "a recent frame", and capture_frame() already does the right
+    thing for them.
+    """
+    task = _inflight_captures.get((url, camera_type, snapshot_url))
+    return task is not None and not task.done()
+
+
+def _discard_inflight_capture(key: tuple[str, str, str | None], task: asyncio.Task) -> None:
+    """Done-callback: drop the finished task from the in-flight registry.
+
+    Guarded on identity so a slow task that finishes after a newer capture
+    has registered for the same key can't evict its successor.
+
+    Also retrieves the exception, if any: the leader normally awaits the
+    task and would surface it, but a leader whose own caller was cancelled
+    leaves nobody to collect it, and an unretrieved task exception is
+    logged by asyncio as a warning with a traceback at an arbitrary later
+    point otherwise.
+    """
+    if _inflight_captures.get(key) is task:
+        del _inflight_captures[key]
+    if not task.cancelled() and task.exception() is not None:
+        logger.debug("In-flight external-camera capture for %s ended in an exception", _log_key(key))
+
+
+def _log_key(key: tuple[str, str, str | None]) -> str:
+    """Render an in-flight key for a log line, with credentials redacted.
+
+    Unlike camera.py's coalescing — which is keyed by IP address and so has
+    nothing to hide — these keys carry the camera URL, and an RTSP camera URL
+    routinely embeds ``user:pass@``. Redact before truncating: slicing first
+    can cut the URL short of the ``@`` the pattern anchors on and leave the
+    password in the log, which is why every other URL log in this module does
+    it in this order.
+    """
+    return redact_url_credentials(key[0])[:50] if key[0] else "None"
+
+
 async def capture_frame(
     url: str,
     camera_type: str,
@@ -186,7 +251,10 @@ async def capture_frame(
     Args:
         url: Live-stream URL (MJPEG stream, RTSP URL, HTTP snapshot URL, or USB device path).
         camera_type: "mjpeg", "rtsp", "snapshot", or "usb".
-        timeout: Connection timeout in seconds.
+        timeout: Connection timeout in seconds. Applies to this caller's own
+            wait, including when it joins another caller's capture - call
+            sites disagree about the value, and a follower must not silently
+            inherit the leader's deadline in either direction.
         snapshot_url: Optional override for single-frame capture. When set, fetched
             via plain HTTP GET regardless of `camera_type`. Bypasses MJPEG warm-up
             handling on sources that expose a dedicated frame endpoint (e.g. go2rtc's
@@ -195,27 +263,120 @@ async def capture_frame(
 
     Returns:
         JPEG bytes or None on failure
+
+    Concurrent callers for the same (url, camera_type, snapshot_url) share
+    one capture (#2705-shape fix, filed for the external-camera path as a
+    follow-up on #2707): the first opens the connection, everyone arriving
+    while it's in flight awaits the same result. This coalesces; it does
+    not cache - a call that arrives after the previous capture finished
+    always captures fresh, since plate detection and the finish-photo path
+    judge a running print from these frames and a stale one there is worse
+    than a slow one (#1397).
     """
-    if snapshot_url:
-        # Redact before truncating — slicing first can cut the URL short of the
-        # ``@`` the pattern anchors on and leave the password in the log.
-        logger.debug("capture_frame using snapshot override url=%s...", redact_url_credentials(snapshot_url)[:50])
-        return await _capture_snapshot(snapshot_url, timeout)
-    logger.debug(
-        "capture_frame called: type=%s, url=%s...",
-        camera_type,
-        redact_url_credentials(url)[:50] if url else "None",
-    )
-    if camera_type == "mjpeg":
-        return await _capture_mjpeg_frame(url, timeout)
-    elif camera_type == "rtsp":
-        return await _capture_rtsp_frame(url, timeout)
-    elif camera_type == "snapshot":
-        return await _capture_snapshot(url, timeout)
-    elif camera_type == "usb":
-        return await _capture_usb_frame(url, timeout)
+    key = (url, camera_type, snapshot_url)
+
+    # A follower whose leader fails takes a turn of its own rather than
+    # inheriting a failure it never had a chance to avoid - by then the
+    # leader has finished, so there's no connection left to compete with.
+    # Bounded at two rounds: if the capture we joined AND its replacement
+    # both failed, a third attempt won't help, and this caller has already
+    # spent its patience.
+    for _ in range(2):
+        leader = _inflight_captures.get(key)
+        if leader is None or leader.done():
+            break
+        try:
+            frame = await asyncio.wait_for(asyncio.shield(leader), timeout=timeout)
+        except TimeoutError:
+            # shield() keeps the capture running for whoever else is still
+            # waiting on it - giving up is this caller's decision alone.
+            logger.warning(
+                "Gave up waiting %ss on the in-flight external-camera capture for %s", timeout, _log_key(key)
+            )
+            return None
+        except asyncio.CancelledError:
+            # Distinguish "the capture I joined was cancelled" from "I was
+            # cancelled". Only the former is ours to recover from.
+            if not leader.cancelled():
+                raise
+            logger.info("In-flight external-camera capture for %s was cancelled; capturing our own", _log_key(key))
+            continue
+        if frame is not None:
+            logger.debug(
+                "Reusing in-flight external-camera capture for %s: %d bytes (no second connection opened)",
+                _log_key(key),
+                len(frame),
+            )
+            return frame
+        logger.debug("In-flight external-camera capture for %s failed; capturing our own", _log_key(key))
     else:
-        logger.warning("Unknown camera type: %s", camera_type)
+        return None
+
+    task = asyncio.create_task(_capture_frame_uncoalesced(url, camera_type, timeout, snapshot_url))
+    _inflight_captures[key] = task
+    task.add_done_callback(functools.partial(_discard_inflight_capture, key))
+    # No wait_for here: this caller IS the capture, and each dispatched
+    # _capture_* function already enforces `timeout` internally, where it
+    # can also kill the ffmpeg process - a second deadline on top would
+    # abandon the subprocess instead of killing it. shield() so a cancelled
+    # leader (a client navigating away mid-request is routine) doesn't take
+    # the capture down with it - followers already waiting on it still get
+    # their frame.
+    return await asyncio.shield(task)
+
+
+async def _capture_frame_uncoalesced(
+    url: str,
+    camera_type: str,
+    timeout: int,
+    snapshot_url: str | None,
+) -> bytes | None:
+    """Open a connection and capture one frame. See capture_frame().
+
+    Callers want that wrapper, not this: it opens a connection
+    unconditionally, which is the collision #2705/#2707 are about.
+
+    Failure is reported as ``None``, never as an exception. That is load-
+    bearing now that captures are shared: the coalescing wrapper hands one
+    task's outcome to every caller waiting on it, and it can only give a
+    follower its own turn for an outcome it can recognise. An exception
+    escaping here would instead propagate to every follower at once —
+    turning one caller's failure into N — and none of them would retry.
+    The per-type helpers below each catch what they expect and return None,
+    but they catch narrowly (``aiohttp.ClientError``/``OSError``/timeouts),
+    so this is the structural guarantee rather than one contingent on their
+    coverage. Mirrors ``_capture_camera_frame_bytes_uncoalesced`` in
+    camera.py, which ends in the same blanket catch for the same reason.
+    """
+    try:
+        if snapshot_url:
+            # Redact before truncating — slicing first can cut the URL short of the
+            # ``@`` the pattern anchors on and leave the password in the log.
+            logger.debug("capture_frame using snapshot override url=%s...", redact_url_credentials(snapshot_url)[:50])
+            return await _capture_snapshot(snapshot_url, timeout)
+        logger.debug(
+            "capture_frame called: type=%s, url=%s...",
+            camera_type,
+            redact_url_credentials(url)[:50] if url else "None",
+        )
+        if camera_type == "mjpeg":
+            return await _capture_mjpeg_frame(url, timeout)
+        elif camera_type == "rtsp":
+            return await _capture_rtsp_frame(url, timeout)
+        elif camera_type == "snapshot":
+            return await _capture_snapshot(url, timeout)
+        elif camera_type == "usb":
+            return await _capture_usb_frame(url, timeout)
+        else:
+            logger.warning("Unknown camera type: %s", camera_type)
+            return None
+    except asyncio.CancelledError:
+        # Cancellation is not a capture failure and must stay distinguishable:
+        # the wrapper checks ``leader.cancelled()`` to decide whether a
+        # follower may take its own turn.
+        raise
+    except Exception:
+        logger.exception("External camera capture failed for %s", redact_url_credentials(url)[:50] if url else "None")
         return None
 
 
@@ -566,12 +727,26 @@ async def test_connection(url: str, camera_type: str) -> dict:
     """Test camera connection.
 
     Returns:
-        Dict with {success: bool, error?: str, resolution?: str}
+        Dict with {success: bool, error?: str, resolution?: str, coalesced: bool}
+
+    ``coalesced`` is True when the frame came from a capture that was already
+    running rather than from a connection this test opened. Captures are shared
+    (see ``capture_frame``), so a test that lands while Obico is polling — or
+    while any other one-shot consumer is mid-capture — gets that frame back and
+    would otherwise report a healthy connection it never made, which is the one
+    answer a *connection test* must not give silently. Forcing an uncoalesced
+    capture here would be worse: it would open the second handle to a
+    single-reader device that this whole mechanism exists to prevent. So the
+    test still shares, and says so. Mirrors the ``coalesced_capture`` code the
+    built-in diagnostic reports for the same situation (camera_diagnose.py).
     """
     logger.info("Testing camera connection: type=%s, url=%s...", camera_type, redact_url_credentials(url)[:50])
+    # Sampled before the call, while it can still distinguish "someone else is
+    # mid-capture" from "I am the one capturing".
+    coalesced = capture_in_flight(url, camera_type)
     try:
         frame = await capture_frame(url, camera_type, timeout=10)
-        logger.info("Capture result: %s bytes", len(frame) if frame else 0)
+        logger.info("Capture result: %s bytes%s", len(frame) if frame else 0, " (coalesced)" if coalesced else "")
 
         if frame:
             # Try to get resolution from JPEG header
@@ -590,15 +765,15 @@ async def test_connection(url: str, camera_type: str) -> dict:
             except (IndexError, ValueError):
                 pass  # Resolution detection is optional; fall back to default
 
-            return {"success": True, "resolution": resolution}
+            return {"success": True, "resolution": resolution, "coalesced": coalesced}
         else:
-            return {"success": False, "error": "Failed to capture frame from camera"}
+            return {"success": False, "error": "Failed to capture frame from camera", "coalesced": coalesced}
 
     except Exception as e:
         # Sanitize error message - don't expose internal details
         error_type = type(e).__name__
         logger.error("Camera connection test failed: %s", e)
-        return {"success": False, "error": f"Connection failed: {error_type}"}
+        return {"success": False, "error": f"Connection failed: {error_type}", "coalesced": coalesced}
 
 
 async def generate_mjpeg_stream(

backend/app/services/homeassistant.py

@@ -187,17 +187,40 @@ class HomeAssistantService:
 
     @staticmethod
     def _validate_url(url: str) -> str | None:
-        """Validate HA URL scheme and block dangerous destinations."""
+        """Normalise a caller-supplied HA URL, or return None if it is unsafe.
+
+        The stored ``ha_url`` setting is already validated at the schema layer
+        (``LAN_SERVICE_URL_SETTINGS`` in schemas/settings.py), but
+        ``test_connection`` takes its URL straight from the request body, so
+        the same policy has to be applied here.
+
+        Delegates to ``_url_safety.assert_safe_lan_service_url`` rather than
+        the string blocklist this replaces. That blocklist only knew three
+        literal hostnames plus a ``169.254.`` prefix and never parsed the
+        hostname as an IP, so it let through the Alibaba (100.100.100.200)
+        and AWS-IPv6 (fd00:ec2::254) metadata endpoints, numeric-encoded
+        loopback, multicast, and IPv4-mapped IPv6 encodings of the IMDS
+        address it did know about.
+
+        Loopback and RFC-1918 remain permitted — Home Assistant is a
+        LAN-resident service by design, and the shared guard is documented
+        that way.
+        """
+        from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
         try:
-            parsed = urlparse(url)
+            assert_safe_lan_service_url(url, label="Home Assistant URL")
         except ValueError:
             return None
-        if parsed.scheme not in ("http", "https") or not parsed.hostname:
-            return None
-        blocked = ("169.254.169.254", "metadata.google.internal", "0.0.0.0")  # nosec B104
-        if parsed.hostname.lower() in blocked or (parsed.hostname or "").startswith("169.254."):
+        # Guard passed, so the scheme is http/https and a hostname is present;
+        # re-parse only to drop query/fragment and normalise the authority.
+        parsed = urlparse(url)
+        if not parsed.hostname:
             return None
-        return f"{parsed.scheme}://{parsed.hostname}" + (f":{parsed.port}" if parsed.port else "") + (parsed.path or "")
+        # urlparse strips the brackets off an IPv6 literal, so they have to go
+        # back on or the rebuilt URL is unparseable ("http://fd00::1:8123").
+        host = f"[{parsed.hostname}]" if ":" in parsed.hostname else parsed.hostname
+        return f"{parsed.scheme.lower()}://{host}" + (f":{parsed.port}" if parsed.port else "") + (parsed.path or "")
 
     async def test_connection(self, url: str, token: str) -> dict:
         """Test connection to Home Assistant.

backend/app/services/layer_timelapse.py

@@ -6,6 +6,7 @@ Captures a frame on each layer change and stitches them into a video on print co
 import asyncio
 import logging
 import shutil
+import time
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -19,6 +20,15 @@ logger = logging.getLogger(__name__)
 # Active timelapse sessions: {printer_id: TimelapseSession}
 _active_sessions: dict[int, "TimelapseSession"] = {}
 
+# Sessions whose frames are being stitched right now: {printer_id: session_id}.
+# on_print_complete removes the session from _active_sessions *before* handing
+# frames_dir to ffmpeg, so for the length of a stitch (up to 300s) nothing in
+# _active_sessions marks that directory as in use. Without this second registry
+# the only thing standing between an in-progress stitch and
+# cleanup_orphaned_timelapse_sessions() is the age margin — whose default is
+# exactly the stitch timeout, so there is no headroom at all.
+_finalizing_sessions: dict[int, str] = {}
+
 
 def get_ffmpeg_path() -> str | None:
     """Get the path to ffmpeg executable."""
@@ -281,6 +291,12 @@ async def on_print_complete(printer_id: int) -> Path | None:
     # Create output path in parent of frames dir
     output_path = session.frames_dir.parent / f"timelapse_{session.session_id}.mp4"
 
+    # The session is already out of _active_sessions, so mark it finalizing for
+    # the length of the stitch — otherwise a sweep running now sees a frames
+    # directory that matches no session and whose mtime is the last layer's
+    # write, which on a tall print's final layer is easily older than the age
+    # margin, and deletes ffmpeg's input from under it.
+    _finalizing_sessions[printer_id] = session.session_id
     try:
         success = await session.stitch(output_path)
         if success:
@@ -294,6 +310,8 @@ async def on_print_complete(printer_id: int) -> Path | None:
         logger.error("Timelapse completion failed: %s", e)
         session.cleanup()
         return None
+    finally:
+        _finalizing_sessions.pop(printer_id, None)
 
 
 def cancel_session(printer_id: int):
@@ -311,3 +329,92 @@ def cancel_session(printer_id: int):
 def get_active_sessions() -> dict[int, TimelapseSession]:
     """Get all active timelapse sessions."""
     return _active_sessions.copy()
+
+
+def cleanup_orphaned_timelapse_sessions(min_age_seconds: float = 300) -> int:
+    """Remove timelapse_frames/<printer_id>/* left behind by a crash or
+    restart that happened while a session was active.
+
+    _active_sessions is in-memory only, so a process restart loses track of
+    any in-flight session without ever calling cancel_session()/cleanup() -
+    the frames directory (and, if stitching had already produced output
+    before the restart, a stray `timelapse_<session_id>.mp4`) are then
+    orphaned on disk with nothing else to reap them (unlike the ffmpeg
+    orphan janitor in routes/camera.py, there was no equivalent here).
+
+    Safe to call once at startup: normal operation always cleans up via
+    on_print_complete/cancel_session, so anything found here predates this
+    process - and a restart-recovered print doesn't get a new timelapse
+    session either (`_maybe_start_layer_timelapse` is only wired into fresh
+    PRINT_START events, see #1353), so an orphaned directory can never be
+    resumed.
+
+    Also safe to call mid-run, which needs all three guards rather than the
+    age margin alone:
+
+    * `_active_sessions` covers a session that is still capturing.
+    * `_finalizing_sessions` covers the stitch window. on_print_complete drops
+      the session from `_active_sessions` before handing frames_dir to ffmpeg,
+      so without this the directory matches no session for up to 300s while
+      being actively read.
+    * `min_age_seconds` covers the remaining gap - a session in the middle of
+      being created, and the stitched `.mp4` between ffmpeg finishing it and
+      the caller attaching and unlinking it. Both are freshly written, so the
+      margin has real headroom there; it did NOT have any for the stitch
+      window, whose length is bounded by the same 300s.
+
+    Returns the number of orphaned directories/files removed.
+    """
+    base_dir = settings.base_dir / "timelapse_frames"
+    if not base_dir.exists():
+        return 0
+
+    now = time.time()
+    removed = 0
+    for printer_dir in base_dir.iterdir():
+        if not printer_dir.is_dir():
+            continue
+        try:
+            printer_id = int(printer_dir.name)
+        except ValueError:
+            continue
+
+        active_session = _active_sessions.get(printer_id)
+        in_use_session_ids = {
+            active_session.session_id if active_session else None,
+            _finalizing_sessions.get(printer_id),
+        } - {None}
+
+        for entry in printer_dir.iterdir():
+            # Frame dirs are named "<session_id>/"; stitched-but-not-yet-
+            # attached output files are "timelapse_<session_id>.mp4" (see
+            # on_print_complete's output_path). Anything else under here was
+            # not written by this module, so leave it alone rather than
+            # deleting a file on the strength of its age.
+            if entry.is_dir():
+                entry_session_id = entry.name
+            elif entry.name.startswith("timelapse_") and entry.name.endswith(".mp4"):
+                entry_session_id = entry.name[len("timelapse_") : -len(".mp4")]
+            else:
+                continue
+            if entry_session_id in in_use_session_ids:
+                continue
+            try:
+                if now - entry.stat().st_mtime < min_age_seconds:
+                    continue
+            except OSError:
+                continue
+            try:
+                # No ignore_errors: it would swallow a failed removal while the
+                # count and the log line below still claimed success, and that
+                # log is the only evidence an operator has of what was deleted.
+                if entry.is_dir():
+                    shutil.rmtree(entry)
+                else:
+                    entry.unlink(missing_ok=True)
+                removed += 1
+                logger.info("Removed orphaned timelapse artifact: %s", entry)
+            except OSError as e:
+                logger.warning("Failed to remove orphaned timelapse artifact %s: %s", entry, e)
+
+    return removed

backend/app/services/mqtt_relay.py

@@ -282,6 +282,8 @@ class MQTTRelayService:
             "big_fan1_speed": state.big_fan1_speed,
             "big_fan2_speed": state.big_fan2_speed,
             "heatbreak_fan_speed": state.heatbreak_fan_speed,
+            "left_aux_fan_speed": state.left_aux_fan_speed,
+            "exhaust_fan_present": state.exhaust_fan_present,
             # Bambuddy-side gate, not printer telemetry (#2525). Mirrors what the
             # Web UI already receives via printer_state_to_dict, so an external
             # automation can tell "finished" from "finished and still waiting for

backend/app/services/obico_detection.py

@@ -365,7 +365,23 @@ class ObicoDetectionService:
         }
 
     async def test_connection(self, url: str) -> dict:
-        """Ping the ML API health endpoint. Returns {ok, status_code, body, error}."""
+        """Ping the ML API health endpoint. Returns {ok, status_code, body, error}.
+
+        The stored ``obico_ml_url`` setting is validated at the schema layer,
+        but this route takes its URL from the request body, so the same
+        LAN-service policy has to be applied here or the guard is trivially
+        sidestepped by testing a URL instead of saving it. The response body
+        is returned to the caller (it is the health signal — the endpoint
+        answers "ok"), which is exactly why the destination must be inside
+        policy before the request is made.
+        """
+        from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
+        try:
+            assert_safe_lan_service_url(url, label="Obico ML URL")
+        except ValueError as exc:
+            return {"ok": False, "status_code": None, "body": None, "error": str(exc)}
+
         target = f"{url.rstrip('/')}/hc/"
         try:
             async with httpx.AsyncClient(timeout=HEALTH_TIMEOUT) as client:

backend/app/services/printer_manager.py

@@ -1418,6 +1418,8 @@ def printer_state_to_dict(
         "big_fan1_speed": state.big_fan1_speed,
         "big_fan2_speed": state.big_fan2_speed,
         "heatbreak_fan_speed": state.heatbreak_fan_speed,
+        "left_aux_fan_speed": state.left_aux_fan_speed,
+        "exhaust_fan_present": state.exhaust_fan_present,
         # Chamber light state
         "chamber_light": state.chamber_light,
         # Active extruder for dual-nozzle printers (0=right, 1=left)

backend/app/services/rest_smart_plug.py

@@ -1,10 +1,8 @@
 """Service for controlling smart plugs via generic REST/HTTP API."""
 
-import ipaddress
 import json
 import logging
 from typing import TYPE_CHECKING, Any
-from urllib.parse import urlparse
 
 import httpx
 
@@ -24,18 +22,39 @@ class RESTSmartPlugService:
         self.timeout = timeout
 
     @staticmethod
-    def _validate_url(url: str) -> bool:
-        """Block cloud metadata and link-local IPs."""
+    def _url_error(url: str) -> str | None:
+        """Return why *url* is rejected by the LAN-service policy, else None.
+
+        Split out from ``_validate_url`` so ``test_connection`` can tell the
+        user which rule the URL broke instead of a single fixed sentence.
+        """
+        from backend.app.api.routes._url_safety import assert_safe_lan_service_url
+
         try:
-            parsed = urlparse(url)
-            hostname = parsed.hostname
-            if not hostname:
-                return False
-            addr = ipaddress.ip_address(hostname)
-            return not addr.is_loopback and not addr.is_link_local
-        except ValueError:
-            # Hostname is not an IP (e.g., "openhab.local") — allow it
-            return True
+            assert_safe_lan_service_url(url, label="REST plug URL")
+        except ValueError as exc:
+            return str(exc)
+        return None
+
+    @staticmethod
+    def _validate_url(url: str) -> bool:
+        """Apply the shared LAN-service SSRF policy to a REST plug URL.
+
+        Delegates to ``_url_safety.assert_safe_lan_service_url`` — the same
+        guard Spoolman, the notification providers and the LAN-service
+        settings use — rather than reimplementing a narrower check. The
+        hand-rolled version this replaces got the policy wrong in both
+        directions: it rejected a literal ``127.0.0.1`` (so an openHAB or
+        Node-RED instance on the same host could only be reached by spelling
+        it ``localhost``), while allowing every target the shared policy
+        rejects unconditionally — Alibaba/AWS-IPv6 metadata endpoints,
+        numeric-encoded IPs, multicast and the unspecified address — because
+        anything that wasn't a bare IP literal fell through to ``True``.
+
+        Loopback and RFC-1918 stay permitted on purpose: a REST-controlled
+        plug bridge running next to Bambuddy is the normal topology.
+        """
+        return RESTSmartPlugService._url_error(url) is None
 
     def _parse_headers(self, headers_json: str | None) -> dict[str, str]:
         """Parse JSON string to dict of headers."""
@@ -273,8 +292,9 @@ class RESTSmartPlugService:
             - success: bool
             - error: error message if failed
         """
-        if not self._validate_url(url):
-            return {"success": False, "error": "Invalid URL (loopback/link-local addresses are blocked)"}
+        url_error = self._url_error(url)
+        if url_error:
+            return {"success": False, "error": url_error}
 
         parsed_headers = self._parse_headers(headers)
 

backend/app/services/tasmota.py

@@ -26,12 +26,32 @@ class TasmotaService:
 
     @staticmethod
     def _validate_ip(ip: str) -> bool:
-        """Block cloud metadata and link-local IPs."""
+        """Block cloud metadata, loopback and link-local destinations.
+
+        Deliberately stricter than the shared LAN-service guard, and kept that
+        way: a Tasmota plug is always a separate device on the LAN, so a bare
+        IP literal is the only sensible value. Anything that is not one —
+        including a symbolic hostname — still fails closed here, which is why
+        this does not simply delegate to ``assert_safe_lan_service_url``.
+
+        What it borrows from the shared guard is the destination set that is
+        dangerous under any topology: cloud-metadata endpoints beyond the AWS
+        IPv4 address (Alibaba's 100.100.100.200, AWS's fd00:ec2::254),
+        multicast and unspecified addresses, and IPv4-mapped IPv6 encodings
+        used to smuggle any of the above past the per-class checks.
+        """
+        from backend.app.api.routes._url_safety import CLOUD_METADATA_IPS, unwrap_ipv4_mapped
+
         try:
             addr = ipaddress.ip_address(ip)
         except ValueError:
             return False  # Not a valid IP
-        return not addr.is_loopback and not addr.is_link_local
+        effective = unwrap_ipv4_mapped(addr)
+        if effective in CLOUD_METADATA_IPS:
+            return False
+        if effective.is_multicast or effective.is_unspecified:
+            return False
+        return not effective.is_loopback and not effective.is_link_local
 
     async def _send_command(
         self,

backend/app/utils/printer_models.py

@@ -212,6 +212,36 @@ DUAL_NOZZLE_MODELS = frozenset(
 )
 
 
+# Models where Bambu's own firmware/UI names the enclosure fan (big_fan2 /
+# airduct part id 3) "Exhaust" rather than "Chamber". On these the printer's
+# touchscreen and Bambu Studio both call it the exhaust fan, and on the P2S it
+# is an add-on kit rather than built-in hardware. Other enclosed models
+# (X1 / P1S / H2 series) keep the "Chamber" naming.
+EXHAUST_FAN_LABEL_MODELS = frozenset(
+    [
+        # Display names (uppercase, no spaces)
+        "P2S",
+        "X2D",
+        # Internal codes
+        "N7",  # P2S
+        "N6",  # X2D
+    ]
+)
+
+
+def uses_exhaust_fan_label(model: str | None) -> bool:
+    """Return True if this model calls the big_fan2 enclosure fan "Exhaust".
+
+    P2S/X2D name that fan "Exhaust" in Bambu's firmware/UI; everything else
+    enclosed calls it the chamber fan. Used so the UI badge and the API
+    response message agree on what the user sees.
+    """
+    if not model:
+        return False
+    normalized = model.strip().upper().replace(" ", "").replace("-", "")
+    return normalized in EXHAUST_FAN_LABEL_MODELS
+
+
 def has_ethernet(model: str | None) -> bool:
     """Return True if the printer model has an ethernet port."""
     if not model:

backend/tests/integration/test_printers_api.py

@@ -3855,8 +3855,10 @@ class TestSetChamberTemperatureAPI:
 class TestSetFanSpeedAPI:
     """Integration tests for POST /printers/{id}/fan-speed (#1661).
 
-    The fan-id mapping (part->1, aux->2, chamber->3) is the critical
-    correctness gate — wrong mapping would target the wrong physical fan.
+    The fan-id mapping (part->1, aux->2, chamber->3, aux2->10) is the
+    critical correctness gate — wrong mapping would target the wrong
+    physical fan. "aux2" (M106 P10) is the optional left auxiliary part
+    cooling fan on P2S/X2D.
     """
 
     @pytest.mark.asyncio
@@ -3879,13 +3881,17 @@ class TestSetFanSpeedAPI:
     @pytest.mark.integration
     @pytest.mark.parametrize(
         "fan_name,expected_fan_id",
-        [("part", 1), ("aux", 2), ("chamber", 3)],
+        [("part", 1), ("aux", 2), ("chamber", 3), ("aux2", 10)],
     )
     async def test_fan_id_mapping(self, async_client: AsyncClient, printer_factory, fan_name, expected_fan_id):
         """Verify each fan name maps to the correct hardware fan-id."""
         printer = await printer_factory(name="P", model="X1C")
         mock_client = MagicMock()
         mock_client.set_fan_speed.return_value = True
+        # aux2 is presence-gated on the printer reporting airduct part 10, so
+        # give the mock a reported speed. Set explicitly rather than leaning on
+        # MagicMock's auto-attribute, which would satisfy the gate by accident.
+        mock_client.state.left_aux_fan_speed = 0
         with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
             mock_pm.get_client.return_value = mock_client
             response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan={fan_name}&speed=100")
@@ -3893,6 +3899,39 @@ class TestSetFanSpeedAPI:
         called_fan_id, called_pwm = mock_client.set_fan_speed.call_args.args
         assert called_fan_id == expected_fan_id
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_aux2_rejected_when_printer_has_no_left_aux_fan(self, async_client: AsyncClient, printer_factory):
+        """A printer that never reports airduct part 10 must not be sent M106 P10.
+
+        Without the gate the endpoint accepted aux2 for every model, so a POST
+        against an A1 would fire a command for hardware that does not exist.
+        """
+        printer = await printer_factory(name="P", model="A1")
+        mock_client = MagicMock()
+        mock_client.set_fan_speed.return_value = True
+        mock_client.state.left_aux_fan_speed = None
+        with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=aux2&speed=50")
+        assert response.status_code == 400
+        assert "left auxiliary fan" in response.json()["detail"]
+        mock_client.set_fan_speed.assert_not_called()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_other_fans_unaffected_by_the_aux2_gate(self, async_client: AsyncClient, printer_factory):
+        """The gate is aux2-only — a base P2S can still drive its built-in fans."""
+        printer = await printer_factory(name="P", model="P2S")
+        mock_client = MagicMock()
+        mock_client.set_fan_speed.return_value = True
+        mock_client.state.left_aux_fan_speed = None
+        with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            for fan_name in ("part", "aux", "chamber"):
+                response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan={fan_name}&speed=50")
+                assert response.status_code == 200, fan_name
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     @pytest.mark.parametrize(
@@ -3911,6 +3950,36 @@ class TestSetFanSpeedAPI:
         _called_fan_id, called_pwm = mock_client.set_fan_speed.call_args.args
         assert called_pwm == expected_pwm
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    @pytest.mark.parametrize(
+        "model,expected_label",
+        [
+            ("P2S", "Exhaust fan"),
+            ("X2D", "Exhaust fan"),
+            ("X1C", "Chamber fan"),
+            ("P1S", "Chamber fan"),
+            ("H2D", "Chamber fan"),
+        ],
+    )
+    async def test_chamber_fan_message_matches_model_label(
+        self, async_client: AsyncClient, printer_factory, model, expected_label
+    ):
+        """The success toast must use the same name as the printer card badge.
+
+        On P2S/X2D the big_fan2 fan is labelled "Exhaust"; everywhere else it
+        stays "Chamber". A mismatch means the user clicks "Exhaust" and gets
+        told "Chamber fan set to N%".
+        """
+        printer = await printer_factory(name="P", model=model)
+        mock_client = MagicMock()
+        mock_client.set_fan_speed.return_value = True
+        with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            response = await async_client.post(f"/api/v1/printers/{printer.id}/fan-speed?fan=chamber&speed=50")
+        assert response.status_code == 200
+        assert response.json()["message"] == f"{expected_label} set to 50%"
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_speed_out_of_range_rejected(self, async_client: AsyncClient, printer_factory):

backend/tests/unit/services/test_external_camera_capture_coalescing.py

@@ -0,0 +1,493 @@
+"""Single-flight coalescing of one-shot external-camera captures (#2705-shape
+fix, filed against the external-camera path as a follow-up on #2707).
+
+V4L2 USB devices allow exactly one open handle - the same one-connection
+limit #2705 covers for Bambu firmware. The #2707 guards (``is_stream_active``
+/ ``try_get_active_buffered_frame``) only keep a one-shot capturer from
+competing with the fan-out live view; nothing kept the capturers from
+competing with EACH OTHER when no viewer is attached, so an Obico poll and
+the in-print frame bank (say) could each open their own connection to the
+same USB device and collide.
+
+These tests drive ``capture_frame`` at the public boundary and count how
+many times the underlying capture ran, since "how many connections did we
+open" is the entire point of the fix. Mirrors
+``test_camera_capture_coalescing.py``'s structure for the built-in path.
+"""
+
+import asyncio
+
+import pytest
+
+from backend.app.services import external_camera as ec_module
+from backend.app.services.external_camera import capture_frame, capture_in_flight
+
+FRAME_A = b"\xff\xd8" + b"a" * 200 + b"\xff\xd9"
+FRAME_B = b"\xff\xd8" + b"b" * 200 + b"\xff\xd9"
+
+
+@pytest.fixture(autouse=True)
+def _clear_inflight():
+    """The registry is module-global; don't leak tasks between tests."""
+    ec_module._inflight_captures.clear()
+    yield
+    ec_module._inflight_captures.clear()
+
+
+class RecordingCapture:
+    """Stand-in for the real capture, recording each call.
+
+    ``gate`` (when set) holds every capture open until released, which is how
+    these tests create the overlap window that used to produce two
+    connections.
+    """
+
+    def __init__(self, frames=(FRAME_A, FRAME_B), gate: asyncio.Event | None = None):
+        self.calls: list[tuple[str, str, str | None, int]] = []
+        self._frames = list(frames)
+        self._gate = gate
+        self.started = asyncio.Event()
+
+    async def __call__(self, url, camera_type, timeout, snapshot_url):
+        self.calls.append((url, camera_type, snapshot_url, timeout))
+        self.started.set()
+        if self._gate is not None:
+            await self._gate.wait()
+        return self._frames.pop(0) if self._frames else None
+
+    @property
+    def count(self) -> int:
+        return len(self.calls)
+
+
+@pytest.fixture
+def patch_capture(monkeypatch):
+    def _install(capture):
+        monkeypatch.setattr(ec_module, "_capture_frame_uncoalesced", capture)
+        return capture
+
+    return _install
+
+
+async def _let_leader_start(capture: RecordingCapture) -> None:
+    """Wait until the leader is inside the capture, so the next caller joins it.
+
+    Without this the second caller can reach the registry before the first
+    has even been scheduled, which tests a different (and uninteresting) race.
+    """
+    await asyncio.wait_for(capture.started.wait(), timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_simultaneous_callers_share_one_capture(patch_capture):
+    """The reported collision: two consumers, one connection, two frames."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=20))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=15))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert await follower == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_five_callers_one_capture(patch_capture):
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    first = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    rest = [asyncio.create_task(capture_frame("/dev/video1", "usb")) for _ in range(4)]
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await asyncio.gather(first, *rest) == [FRAME_A] * 5
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_different_cameras_do_not_coalesce(patch_capture):
+    """The one-connection limit is per camera, so the key must be too."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    one = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    two = asyncio.create_task(capture_frame("/dev/video2", "usb"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert {await one, await two} == {FRAME_A, FRAME_B}
+    assert capture.count == 2
+    assert {url for url, *_ in capture.calls} == {"/dev/video1", "/dev/video2"}
+
+
+@pytest.mark.asyncio
+async def test_different_snapshot_url_does_not_coalesce(patch_capture):
+    """#1177's snapshot_url override routes to a different endpoint entirely -
+    two printers sharing a camera_url but differing only in snapshot_url must
+    not share a capture."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    one = asyncio.create_task(capture_frame("http://cam/", "mjpeg", snapshot_url="http://cam/frame1.jpg"))
+    await _let_leader_start(capture)
+    two = asyncio.create_task(capture_frame("http://cam/", "mjpeg", snapshot_url="http://cam/frame2.jpg"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert {await one, await two} == {FRAME_A, FRAME_B}
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_coalescing_is_not_caching(patch_capture):
+    """Sequential callers each capture fresh.
+
+    Deliberate: plate detection and the finish-photo path decide things about
+    a running print from these frames, and #1397 was a finish photo a few
+    seconds stale showing the bed already lowered.
+    """
+    capture = patch_capture(RecordingCapture())
+
+    assert await capture_frame("/dev/video1", "usb") == FRAME_A
+    assert await capture_frame("/dev/video1", "usb") == FRAME_B
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_registry_is_empty_after_a_capture_finishes(patch_capture):
+    """No leak, and nothing left behind for the next caller to join."""
+    patch_capture(RecordingCapture())
+
+    await capture_frame("/dev/video1", "usb")
+    await asyncio.sleep(0)  # let the done-callback run
+
+    assert ec_module._inflight_captures == {}
+    assert capture_in_flight("/dev/video1", "usb") is False
+
+
+@pytest.mark.asyncio
+async def test_failed_leader_does_not_poison_its_followers(patch_capture):
+    """A follower that never got its own attempt gets one when the leader fails.
+
+    Safe by then: the leader has finished, so there is no connection to
+    compete with. This also covers the follower whose timeout is LONGER than
+    the leader's — it isn't cut short by someone else's deadline.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, FRAME_B), gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=10))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=20))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader is None
+    assert await follower == FRAME_B
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_two_consecutive_failures_give_up(patch_capture):
+    """Bounded retry: a follower doesn't chase failing captures forever.
+
+    Two followers behind a failing leader. The first takes its own turn, the
+    second joins THAT capture, and when it fails too the second gives up
+    rather than opening a third connection.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, None), gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    first = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+    second = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader is None
+    assert await first is None
+    assert await second is None
+    # The leader's capture plus one retry — not one per disappointed caller.
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_follower_timeout_does_not_sabotage_the_capture(patch_capture):
+    """A follower giving up leaves the capture running for everyone else.
+
+    Call sites disagree about the timeout, so a follower must be able to
+    abandon a join without cancelling a capture other callers are still
+    waiting on.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=30))
+    await _let_leader_start(capture)
+    impatient = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=0.01))
+    patient = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=30))
+
+    assert await impatient is None  # gave up on its own deadline
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert await patient == FRAME_A  # unaffected by the one that walked away
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_cancelled_leader_still_delivers_to_followers(patch_capture):
+    """Snapshot/capture requests get cancelled routinely (client navigates
+    away mid-request). The follower must not lose the frame because the
+    caller that happened to open the connection went away."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+
+    leader.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await leader
+    gate.set()
+
+    assert await follower == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_cancelling_a_follower_leaves_the_leader_alone(patch_capture):
+    """The mirror case: the follower's cancellation is its own business."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+
+    follower.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await follower
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_capture_in_flight_reports_the_window(patch_capture):
+    """The predicate a diagnose-style caller would use to know it will join,
+    not measure its own connection."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    assert capture_in_flight("/dev/video1", "usb") is False
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+
+    assert capture_in_flight("/dev/video1", "usb") is True
+    assert capture_in_flight("/dev/video2", "usb") is False  # per camera
+
+    gate.set()
+    await leader
+    await asyncio.sleep(0)
+
+    assert capture_in_flight("/dev/video1", "usb") is False
+
+
+# ---------------------------------------------------------------------------
+# Failure must arrive as None, never as an exception
+# ---------------------------------------------------------------------------
+#
+# `test_failed_leader_does_not_poison_its_followers` above covers a leader that
+# RETURNS None. A leader that RAISES is a different path: the wrapper's retry
+# loop only catches TimeoutError and CancelledError, so an escaping exception
+# would reach every follower at once and none of them would take a turn of
+# their own — one caller's failure becoming N. The per-type helpers catch
+# narrowly (aiohttp.ClientError / OSError / timeouts), so the guarantee lives
+# in _capture_frame_uncoalesced's own blanket catch.
+
+
+@pytest.mark.asyncio
+async def test_an_unexpected_error_is_reported_as_a_failed_capture():
+    """Not every failure is an OSError. An IncompleteReadError is an EOFError,
+    which none of the per-type helpers catch."""
+
+    async def raising(url, timeout):
+        raise asyncio.IncompleteReadError(partial=b"", expected=4)
+
+    import backend.app.services.external_camera as ec
+
+    original = ec._capture_snapshot
+    ec._capture_snapshot = raising
+    try:
+        result = await ec._capture_frame_uncoalesced("http://cam/snap", "snapshot", 5, None)
+    finally:
+        ec._capture_snapshot = original
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_a_raising_leader_does_not_take_its_followers_down_with_it(monkeypatch):
+    """The whole point of coalescing is that one caller's connection serves
+    several. It must not also mean one caller's crash fails several.
+
+    Patches the per-type helper rather than ``_capture_frame_uncoalesced``,
+    deliberately: the guarantee lives in that function's blanket catch, so a
+    stand-in installed in its place would test the wrapper against a shape the
+    wrapper can no longer be handed.
+    """
+    gate = asyncio.Event()
+    attempts: list[str] = []
+
+    async def raise_then_succeed(url, timeout):
+        attempts.append(url)
+        if len(attempts) == 1:
+            await gate.wait()
+            raise RuntimeError("ffmpeg died in a way nobody catches")
+        return FRAME_B
+
+    monkeypatch.setattr(ec_module, "_capture_rtsp_frame", raise_then_succeed)
+
+    leader = asyncio.create_task(capture_frame("rtsp://cam/1", "rtsp", timeout=5))
+    await asyncio.sleep(0)
+    follower = asyncio.create_task(capture_frame("rtsp://cam/1", "rtsp", timeout=5))
+    await asyncio.sleep(0)
+    gate.set()
+
+    leader_result, follower_result = await asyncio.gather(leader, follower, return_exceptions=True)
+
+    assert not isinstance(leader_result, BaseException), f"leader raised {leader_result!r}"
+    assert not isinstance(follower_result, BaseException), f"follower raised {follower_result!r}"
+    assert leader_result is None, "the leader's own capture failed, so it gets None"
+    assert follower_result == FRAME_B, "the follower took its own turn and succeeded"
+
+
+# ---------------------------------------------------------------------------
+# The connection test must not claim a connection it never opened
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_connection_reports_when_it_shared_someone_elses_capture(patch_capture):
+    """A test landing while Obico is mid-poll gets that frame back. Reporting a
+    bare success would credit a connection this test never made — and forcing
+    its own would open the second handle the coalescing exists to prevent."""
+    from backend.app.services.external_camera import test_connection
+
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,), gate=gate))
+
+    other = asyncio.create_task(capture_frame("rtsp://cam/1", "rtsp", timeout=5))
+    await _let_leader_start(capture)
+
+    tested = asyncio.create_task(test_connection("rtsp://cam/1", "rtsp"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    result = await tested
+    await other
+
+    assert result["success"] is True
+    assert result["coalesced"] is True
+    assert capture.count == 1, "no second connection was opened"
+
+
+@pytest.mark.asyncio
+async def test_connection_reports_its_own_capture_as_not_coalesced(patch_capture):
+    from backend.app.services.external_camera import test_connection
+
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,)))
+    result = await test_connection("rtsp://cam/1", "rtsp")
+
+    assert result["success"] is True
+    assert result["coalesced"] is False
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_connection_reports_coalesced_on_the_failure_path_too(patch_capture):
+    """The flag describes where the answer came from, not whether it was good."""
+    from backend.app.services.external_camera import test_connection
+
+    capture = patch_capture(RecordingCapture(frames=(None,)))
+    result = await test_connection("rtsp://cam/1", "rtsp")
+
+    assert result["success"] is False
+    assert result["coalesced"] is False
+    assert capture.count == 1
+
+
+# ---------------------------------------------------------------------------
+# Credentials must not reach the log
+# ---------------------------------------------------------------------------
+#
+# camera.py's coalescing is keyed by IP address and has nothing to redact.
+# These keys carry the camera URL, and an RTSP camera URL routinely embeds
+# user:pass@ — which is why every other URL log in the module redacts.
+
+CREDENTIALED_URL = "rtsp://admin:hunter2@192.168.1.50:554/Streaming/Channels/101"
+
+
+@pytest.mark.asyncio
+async def test_the_reuse_log_line_redacts_the_password(patch_capture, caplog):
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,), gate=gate))
+
+    with caplog.at_level("DEBUG", logger=ec_module.__name__):
+        leader = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await _let_leader_start(capture)
+        follower = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await asyncio.sleep(0)
+        gate.set()
+        await asyncio.gather(leader, follower)
+
+    assert not [r.getMessage() for r in caplog.records if "hunter2" in r.getMessage()]
+
+
+@pytest.mark.asyncio
+async def test_the_gave_up_waiting_log_line_redacts_the_password(patch_capture, caplog):
+    """This one is a warning, so it shows at the default level and lands in
+    support bundles."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,), gate=gate))
+
+    with caplog.at_level("DEBUG", logger=ec_module.__name__):
+        leader = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await _let_leader_start(capture)
+        assert await capture_frame(CREDENTIALED_URL, "rtsp", timeout=0) is None
+        gate.set()
+        await leader
+
+    messages = [r.getMessage() for r in caplog.records]
+    assert any("Gave up waiting" in m for m in messages), "the timeout path did not run"
+    assert not [m for m in messages if "hunter2" in m]
+
+
+@pytest.mark.asyncio
+async def test_the_failed_capture_log_line_redacts_the_password(patch_capture, caplog):
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, FRAME_B), gate=gate))
+
+    with caplog.at_level("DEBUG", logger=ec_module.__name__):
+        leader = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await _let_leader_start(capture)
+        follower = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await asyncio.sleep(0)
+        gate.set()
+        await asyncio.gather(leader, follower)
+
+    assert not [r.getMessage() for r in caplog.records if "hunter2" in r.getMessage()]

backend/tests/unit/services/test_layer_timelapse.py

@@ -4,6 +4,7 @@ Tests for the layer timelapse service.
 These tests cover session management and pure logic functions.
 """
 
+import time
 from datetime import datetime
 from pathlib import Path
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
@@ -411,3 +412,235 @@ class TestGetActiveSessions:
                 assert 1 in _active_sessions
 
                 cancel_session(1)
+
+
+class TestCleanupOrphanedTimelapseSessions:
+    """_active_sessions is in-memory only, so a process restart mid-print
+    loses track of an active session without ever cleaning up its frames
+    directory (or a stitched-but-not-attached output .mp4). Confirmed live:
+    38MB of exactly this leftover on Carl's OrangePi after several restarts
+    during testing. cleanup_orphaned_timelapse_sessions() sweeps for it."""
+
+    def _touch_old(self, path, age_seconds=600):
+        import os
+
+        path.touch()
+        old = time.time() - age_seconds
+        os.utime(path, (old, old))
+
+    def _mkdir_old(self, path, age_seconds=600):
+        import os
+
+        path.mkdir(parents=True)
+        old = time.time() - age_seconds
+        os.utime(path, (old, old))
+
+    def test_removes_orphaned_frame_dir_and_stray_output(self, tmp_path):
+        from backend.app.services.layer_timelapse import (
+            _active_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+        printer_dir = tmp_path / "timelapse_frames" / "1"
+        self._mkdir_old(printer_dir / "20260101_000000")
+        self._touch_old(printer_dir / "timelapse_20260101_000000.mp4")
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
+
+        assert removed == 2
+        assert not (printer_dir / "20260101_000000").exists()
+        assert not (printer_dir / "timelapse_20260101_000000.mp4").exists()
+
+    def test_spares_the_currently_active_session(self, tmp_path):
+        from backend.app.services.layer_timelapse import (
+            TimelapseSession,
+            _active_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            session = TimelapseSession(1, 100, "/dev/video1", "usb")
+            _active_sessions[1] = session
+            import os
+
+            old = time.time() - 600
+            os.utime(session.frames_dir, (old, old))
+
+            removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
+
+        assert removed == 0
+        assert session.frames_dir.exists()
+        _active_sessions.clear()
+
+    def test_spares_recently_modified_entries(self, tmp_path):
+        """Defensive margin: something modified within min_age_seconds is
+        left alone even if it doesn't match an active session, in case this
+        is ever invoked while a session is mid-creation."""
+        from backend.app.services.layer_timelapse import (
+            _active_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+        printer_dir = tmp_path / "timelapse_frames" / "1"
+        printer_dir.mkdir(parents=True)
+        (printer_dir / "20260101_000000").mkdir()
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
+
+        assert removed == 0
+        assert (printer_dir / "20260101_000000").exists()
+
+    def test_no_base_dir_is_a_no_op(self, tmp_path):
+        from backend.app.services.layer_timelapse import cleanup_orphaned_timelapse_sessions
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path / "does-not-exist"
+            removed = cleanup_orphaned_timelapse_sessions()
+
+        assert removed == 0
+
+    def test_ignores_non_numeric_printer_dirs(self, tmp_path):
+        """Defensive: unrelated directories under timelapse_frames/ (there
+        shouldn't be any, but printer_id is parsed from the dir name) must
+        not raise."""
+        from backend.app.services.layer_timelapse import (
+            _active_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+        (tmp_path / "timelapse_frames" / "not-a-printer-id").mkdir(parents=True)
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            removed = cleanup_orphaned_timelapse_sessions()
+
+        assert removed == 0
+
+    def test_spares_a_session_that_is_mid_stitch(self, tmp_path):
+        """on_print_complete drops the session from _active_sessions before it
+        hands frames_dir to ffmpeg, so for the length of a stitch (up to 300s)
+        the directory matches no active session. Its mtime is the last layer's
+        frame write, which on a tall print's final layer is easily older than
+        the age margin — and the margin's default IS the stitch timeout, so it
+        offers no headroom here. _finalizing_sessions covers that window."""
+        import os
+
+        from backend.app.services.layer_timelapse import (
+            TimelapseSession,
+            _active_sessions,
+            _finalizing_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+        _finalizing_sessions.clear()
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            session = TimelapseSession(1, 100, "/dev/video1", "usb")
+            (session.frames_dir / "layer_00001.jpg").write_bytes(b"x")
+            old = time.time() - 600
+            os.utime(session.frames_dir, (old, old))
+
+            # Exactly the state on_print_complete is in while ffmpeg runs.
+            _active_sessions.pop(1, None)
+            _finalizing_sessions[1] = session.session_id
+
+            removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
+
+        assert removed == 0
+        assert session.frames_dir.exists(), "ffmpeg's input was deleted mid-stitch"
+        _finalizing_sessions.clear()
+
+    @pytest.mark.asyncio
+    async def test_on_print_complete_clears_the_finalizing_marker(self, tmp_path):
+        """Including when the stitch fails — a leaked marker would make the
+        sweep skip that printer's leftovers forever."""
+        from backend.app.services.layer_timelapse import (
+            TimelapseSession,
+            _active_sessions,
+            _finalizing_sessions,
+            on_print_complete,
+        )
+
+        _active_sessions.clear()
+        _finalizing_sessions.clear()
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            session = TimelapseSession(1, 100, "/dev/video1", "usb")
+            session.frame_count = 3
+            _active_sessions[1] = session
+
+            with patch.object(TimelapseSession, "stitch", AsyncMock(side_effect=RuntimeError("ffmpeg died"))):
+                result = await on_print_complete(1)
+
+        assert result is None
+        assert 1 not in _finalizing_sessions
+
+    def test_leaves_unrelated_files_alone(self, tmp_path):
+        """Only this module's own artifacts are swept. A file that is neither a
+        session directory nor timelapse_<id>.mp4 was put there by something
+        else, and age is not a reason to delete it."""
+        from backend.app.services.layer_timelapse import (
+            _active_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+        printer_dir = tmp_path / "timelapse_frames" / "1"
+        printer_dir.mkdir(parents=True)
+        stranger = printer_dir / "notes.txt"
+        self._touch_old(stranger)
+        self._touch_old(printer_dir / "timelapse_20260101_000000.mp4")
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
+
+        assert removed == 1
+        assert stranger.exists()
+        assert not (printer_dir / "timelapse_20260101_000000.mp4").exists()
+
+    def test_a_removal_that_fails_is_not_counted_as_removed(self, tmp_path):
+        """The count and the log line are the only evidence an operator has of
+        what was deleted, so a failed rmtree must not be reported as a success.
+
+        The stub honours rmtree's real contract — ignore_errors=True swallows
+        the failure and returns normally — because that is the whole point: a
+        caller passing it gets a silent no-op that the surrounding
+        ``except OSError`` can never see, and would still count and log the
+        directory as removed. A stub that raised unconditionally would pass
+        either way and prove nothing.
+        """
+        from backend.app.services.layer_timelapse import (
+            _active_sessions,
+            cleanup_orphaned_timelapse_sessions,
+        )
+
+        _active_sessions.clear()
+        printer_dir = tmp_path / "timelapse_frames" / "1"
+        self._mkdir_old(printer_dir / "20260101_000000")
+
+        def rmtree_on_read_only_fs(path, ignore_errors=False, **kwargs):
+            if ignore_errors:
+                return  # silently does nothing, exactly like the real thing
+            raise OSError("read-only fs")
+
+        with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+            mock_settings.base_dir = tmp_path
+            with patch("backend.app.services.layer_timelapse.shutil.rmtree", rmtree_on_read_only_fs):
+                removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
+
+        assert removed == 0, "a directory that is still on disk was reported as removed"
+        assert (printer_dir / "20260101_000000").exists()

backend/tests/unit/services/test_p2s_accessory_fans.py

@@ -0,0 +1,394 @@
+"""Tests for the P2S/X2D left auxiliary part cooling fan (#2576).
+
+The "Auxiliary Part Cooling Fan - Left" (also fits X2D) is reported ONLY as
+device.airduct part with raw id 160 (decoded id = 160 >> 4 = 10,
+AIR_FUN.FAN_REMOTE_COOLING_1 in Bambu Studio) — the firmware does NOT mirror
+it into any flat big_fanX_speed field, which is why it was previously dropped.
+It is controlled with "M106 P10", exactly like Bambu's official P2S machine-
+profile gcode does.
+
+The airduct payloads below are verbatim captures from a live P2S
+(fw 01.02.00.00) with the accessory installed.
+"""
+
+import pytest
+
+
+@pytest.fixture
+def mqtt_client():
+    from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+    return BambuMQTTClient(
+        ip_address="192.168.1.100",
+        serial_number="TESTP2S",
+        access_code="12345678",
+    )
+
+
+def _airduct_device(parts):
+    """Wrap airduct parts in the device envelope as pushed by a P2S."""
+    return {
+        "device": {
+            "airduct": {
+                "modeCur": 0,
+                "modeFunc": 0,
+                "modeList": [
+                    {"ctrl": [16, 32, 160, 48], "modeId": 0, "off": []},
+                    {"ctrl": [16, 32, 48], "modeId": 1, "off": [160]},
+                ],
+                "modeVisable": 7,
+                "parts": parts,
+                "subFunc": 0,
+                "subMode": 0,
+                "subVisable": 7,
+                "version": 1,
+            },
+            "type": 1,
+        }
+    }
+
+
+# Verbatim parts list from a live P2S: part cooling ramping (state 30,
+# target 90), right aux at 40%, left aux OFF, chamber at 70%.
+P2S_PARTS_LEFT_AUX_OFF = [
+    {"func": 0, "id": 16, "range": 6553600, "state": 30, "tar_state": 90},
+    {"func": 6, "id": 32, "range": 6553600, "state": 40, "tar_state": 40},
+    {"func": 5, "id": 160, "range": 6553600, "state": 0, "tar_state": 0},
+    {"func": 2, "id": 48, "range": 6553600, "state": 70, "tar_state": 70},
+]
+
+# Same printer later in the print: left aux running at 80%.
+P2S_PARTS_LEFT_AUX_80 = [
+    {"func": 0, "id": 16, "range": 6553600, "state": 60, "tar_state": 60},
+    {"func": 6, "id": 32, "range": 6553600, "state": 100, "tar_state": 100},
+    {"func": 5, "id": 160, "range": 6553600, "state": 80, "tar_state": 80},
+    {"func": 2, "id": 48, "range": 6553600, "state": 80, "tar_state": 80},
+]
+
+
+class TestLeftAuxFanParsing:
+    """device.airduct part id 10 (raw 160) -> state.left_aux_fan_speed."""
+
+    def test_defaults_to_none(self, mqtt_client):
+        assert mqtt_client.state.left_aux_fan_speed is None
+
+    def test_parses_left_aux_running(self, mqtt_client):
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        assert mqtt_client.state.left_aux_fan_speed == 80
+
+    def test_parses_left_aux_off(self, mqtt_client):
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_OFF))
+        assert mqtt_client.state.left_aux_fan_speed == 0
+
+    def test_raw_id_is_bit_unpacked(self, mqtt_client):
+        """Raw id 160 must decode to part id 10 (id >> 4), NOT match on 160."""
+        # A hypothetical raw id of 10 would decode to part id 0 — must not match.
+        parts = [{"func": 5, "id": 10, "range": 6553600, "state": 50, "tar_state": 50}]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.left_aux_fan_speed is None
+
+    def test_parts_without_left_aux_reports_none(self, mqtt_client):
+        """A full parts list without id 10 means the fan is not installed."""
+        mqtt_client.state.left_aux_fan_speed = 80  # previously seen
+        parts = [p for p in P2S_PARTS_LEFT_AUX_80 if p["id"] != 160]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.left_aux_fan_speed is None
+
+    def test_diff_push_without_device_preserves_value(self, mqtt_client):
+        """P-series diff pushes omit device.airduct — value must survive."""
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        mqtt_client._update_state({"nozzle_temper": 250.0})
+        assert mqtt_client.state.left_aux_fan_speed == 80
+
+    def test_state_clamped_to_0_100(self, mqtt_client):
+        parts = [{"func": 5, "id": 160, "range": 6553600, "state": 250, "tar_state": 0}]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.left_aux_fan_speed == 100
+
+    def test_packed_state_decodes_from_low_8_bits(self, mqtt_client):
+        """`state` is bit-packed like its sibling `range` (end << 16 | start).
+
+        Bambu Studio decodes it with get_flag_bits(state, 0, 8), so only the low
+        byte carries the percentage. Without the mask a packed value would clamp
+        to 100 instead of decoding to the real speed.
+        """
+        packed = (60 << 16) | 45  # sibling field in the high bits, 45% in the low byte
+        parts = [{"func": 5, "id": 160, "range": 6553600, "state": packed, "tar_state": 0}]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.left_aux_fan_speed == 45
+
+    def test_unpacked_state_is_unaffected_by_the_mask(self, mqtt_client):
+        # Plain 0-100 values (what a P2S actually sends) must round-trip exactly.
+        for speed in (0, 30, 80, 100):
+            parts = [{"func": 5, "id": 160, "range": 6553600, "state": speed, "tar_state": speed}]
+            mqtt_client._update_state(_airduct_device(parts))
+            assert mqtt_client.state.left_aux_fan_speed == speed
+
+    def test_malformed_part_entries_ignored(self, mqtt_client):
+        parts = [
+            "not-a-dict",
+            {"func": 5},  # no id/state
+            {"id": "garbage", "state": 10},
+            {"func": 5, "id": 160, "range": 6553600, "state": 30, "tar_state": 30},
+        ]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.left_aux_fan_speed == 30
+
+    def test_flat_fan_fields_unaffected(self, mqtt_client):
+        """Regression: flat fields keep coming from the flat MQTT keys."""
+        payload = {
+            "cooling_fan_speed": "4",
+            "big_fan1_speed": "6",
+            "big_fan2_speed": "10",
+            "heatbreak_fan_speed": "14",
+            **_airduct_device(P2S_PARTS_LEFT_AUX_OFF),
+        }
+        mqtt_client._update_state(payload)
+        assert mqtt_client.state.cooling_fan_speed == 27  # 4/15
+        assert mqtt_client.state.big_fan1_speed == 40  # 6/15
+        assert mqtt_client.state.big_fan2_speed == 67  # 10/15
+        assert mqtt_client.state.heatbreak_fan_speed == 93  # 14/15
+        assert mqtt_client.state.left_aux_fan_speed == 0
+
+
+class TestExhaustFanPresence:
+    """device.airduct part id 3 (raw 48) presence -> state.exhaust_fan_present.
+
+    The chamber exhaust fan is a P2S/X2D add-on kit (get_version module "eef").
+    Its speed rides on the flat big_fan2_speed field, but the airduct only lists
+    part id 3 when the kit is physically installed — so part-3 presence is the
+    signal the UI uses to show/hide the Exhaust tile.
+    """
+
+    def test_defaults_to_false(self, mqtt_client):
+        assert mqtt_client.state.exhaust_fan_present is False
+
+    def test_present_when_part_3_reported(self, mqtt_client):
+        # Full P2S parts list includes id 48 (>>4 = 3).
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        assert mqtt_client.state.exhaust_fan_present is True
+
+    def test_absent_when_part_3_missing(self, mqtt_client):
+        mqtt_client.state.exhaust_fan_present = True  # previously seen
+        parts = [p for p in P2S_PARTS_LEFT_AUX_80 if p["id"] != 48]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.exhaust_fan_present is False
+
+    def test_base_p2s_only_part_cooling_and_aux(self, mqtt_client):
+        # A base P2S (no exhaust kit, no left aux kit) lists only ids 1 and 2.
+        parts = [
+            {"func": 0, "id": 16, "range": 6553600, "state": 0, "tar_state": 0},
+            {"func": 6, "id": 32, "range": 6553600, "state": 0, "tar_state": 0},
+        ]
+        mqtt_client._update_state(_airduct_device(parts))
+        assert mqtt_client.state.exhaust_fan_present is False
+        assert mqtt_client.state.left_aux_fan_speed is None
+
+    def test_diff_push_without_device_preserves_value(self, mqtt_client):
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        mqtt_client._update_state({"nozzle_temper": 250.0})
+        assert mqtt_client.state.exhaust_fan_present is True
+
+
+class TestPartialPartsFrames:
+    """A `parts` list that is not a full inventory must not retract presence.
+
+    `device.airduct` is pushed field by field — the `modeCur` handler reads it
+    with an `in` check for exactly that reason — so a frame can carry `parts`
+    without carrying every fan. Absence is what tells us a kit is not fitted, so
+    it is only trustworthy on a complete list. Read as gospel, a truncated frame
+    would make both accessory badges vanish mid-print and start rejecting
+    ``fan=aux2`` on a printer that does have the fan.
+
+    Completeness is judged on ids 1 (part cooling) and 2 (aux) being present:
+    neither is optional on any machine that reports an airduct at all, and both
+    appear in every layout in the support-package archive (P2S base 1,2 /
+    P2S+kit 1,2,3 / X2D 1,2,3,10 / H2C,H2D,H2S 1,2,3,6).
+    """
+
+    def test_partial_frame_does_not_retract_the_left_aux_fan(self, mqtt_client):
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        assert mqtt_client.state.left_aux_fan_speed == 80
+
+        # Only the part cooling fan changed — the frame says nothing about the
+        # left aux fan, which is not the same as saying it is gone.
+        mqtt_client._update_state(
+            _airduct_device([{"func": 0, "id": 16, "range": 6553600, "state": 70, "tar_state": 70}])
+        )
+
+        assert mqtt_client.state.left_aux_fan_speed == 80
+
+    def test_partial_frame_does_not_retract_the_exhaust_fan(self, mqtt_client):
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        assert mqtt_client.state.exhaust_fan_present is True
+
+        mqtt_client._update_state(
+            _airduct_device([{"func": 0, "id": 16, "range": 6553600, "state": 70, "tar_state": 70}])
+        )
+
+        assert mqtt_client.state.exhaust_fan_present is True
+
+    def test_a_partial_frame_still_applies_the_speed_it_carries(self, mqtt_client):
+        """Not-authoritative-for-absence is not the same as ignored."""
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        assert mqtt_client.state.left_aux_fan_speed == 80
+
+        mqtt_client._update_state(
+            _airduct_device([{"func": 5, "id": 160, "range": 6553600, "state": 25, "tar_state": 25}])
+        )
+
+        assert mqtt_client.state.left_aux_fan_speed == 25
+
+    def test_a_partial_frame_can_still_reveal_a_fan(self, mqtt_client):
+        """Presence may always be added — only retraction needs a full list."""
+        mqtt_client._update_state(
+            _airduct_device([{"func": 2, "id": 48, "range": 6553600, "state": 70, "tar_state": 70}])
+        )
+
+        assert mqtt_client.state.exhaust_fan_present is True
+
+    def test_a_full_frame_still_retracts_both(self, mqtt_client):
+        """The kits really can be removed, and a complete list must say so —
+        this is the behaviour the presence gate exists for."""
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        assert mqtt_client.state.left_aux_fan_speed == 80
+        assert mqtt_client.state.exhaust_fan_present is True
+
+        # Base P2S layout: part cooling + aux only.
+        mqtt_client._update_state(
+            _airduct_device(
+                [
+                    {"func": 0, "id": 16, "range": 6553600, "state": 0, "tar_state": 0},
+                    {"func": 6, "id": 32, "range": 6553600, "state": 0, "tar_state": 0},
+                ]
+            )
+        )
+
+        assert mqtt_client.state.left_aux_fan_speed is None
+        assert mqtt_client.state.exhaust_fan_present is False
+
+    def test_an_empty_parts_list_changes_nothing(self, mqtt_client):
+        mqtt_client._update_state(_airduct_device(P2S_PARTS_LEFT_AUX_80))
+        mqtt_client._update_state(_airduct_device([]))
+
+        assert mqtt_client.state.left_aux_fan_speed == 80
+        assert mqtt_client.state.exhaust_fan_present is True
+
+
+class TestLeftAuxFanCommand:
+    """set_fan_speed must accept index 10 and emit M106 P10."""
+
+    def test_set_fan_speed_10_sends_m106_p10(self, mqtt_client, monkeypatch):
+        sent = []
+        monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
+        assert mqtt_client.set_fan_speed(10, 204) is True
+        assert sent == ["M106 P10 S204"]
+
+    def test_set_left_aux_fan_helper(self, mqtt_client, monkeypatch):
+        sent = []
+        monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
+        assert mqtt_client.set_left_aux_fan(255) is True
+        assert sent == ["M106 P10 S255"]
+
+    def test_speed_clamped_to_255(self, mqtt_client, monkeypatch):
+        sent = []
+        monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
+        mqtt_client.set_left_aux_fan(999)
+        assert sent == ["M106 P10 S255"]
+
+    def test_invalid_fan_index_rejected(self, mqtt_client, monkeypatch):
+        sent = []
+        monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
+        assert mqtt_client.set_fan_speed(4, 100) is False
+        assert mqtt_client.set_fan_speed(11, 100) is False
+        assert sent == []
+
+    def test_existing_fan_indexes_still_accepted(self, mqtt_client, monkeypatch):
+        sent = []
+        monkeypatch.setattr(mqtt_client, "send_gcode", lambda g: sent.append(g) or True)
+        for idx in (1, 2, 3):
+            assert mqtt_client.set_fan_speed(idx, 128) is True
+        assert sent == ["M106 P1 S128", "M106 P2 S128", "M106 P3 S128"]
+
+
+class TestExhaustFanLabelModels:
+    """P2S/X2D call the big_fan2 enclosure fan "Exhaust"; others say "Chamber"."""
+
+    def test_p2s_and_x2d_use_exhaust_label(self):
+        from backend.app.utils.printer_models import uses_exhaust_fan_label
+
+        for model in ("P2S", "X2D", "p2s", " P2S ", "N7", "N6"):
+            assert uses_exhaust_fan_label(model) is True, model
+
+    def test_other_enclosed_models_keep_chamber_label(self):
+        from backend.app.utils.printer_models import uses_exhaust_fan_label
+
+        for model in ("X1C", "X1", "X1E", "P1S", "H2D", "H2C", "H2S", "A1"):
+            assert uses_exhaust_fan_label(model) is False, model
+
+    def test_unknown_or_missing_model_defaults_to_chamber(self):
+        from backend.app.utils.printer_models import uses_exhaust_fan_label
+
+        assert uses_exhaust_fan_label(None) is False
+        assert uses_exhaust_fan_label("") is False
+        assert uses_exhaust_fan_label("SomeFutureModel") is False
+
+
+class TestExhaustLabelModelListsAgree:
+    """The exhaust-label model list is duplicated across the stack.
+
+    The backend keeps ``EXHAUST_FAN_LABEL_MODELS`` (display names plus the N7/N6
+    internal codes, since the API can be handed either) and the frontend keeps
+    ``MODELS_WITH_EXHAUST_LABEL`` in PrintersPage.tsx (display names only —
+    ``printer.model`` is always a display name by the time it reaches the card).
+    Both are correct as written, but nothing stopped them drifting apart: adding
+    a model to one and forgetting the other silently produces a card labelled
+    "Exhaust" whose control toast says "Chamber fan", or vice versa.
+    """
+
+    def _frontend_models(self) -> set[str]:
+        import re
+        from pathlib import Path
+
+        import pytest
+
+        # Walk up rather than hard-coding a parent depth, so the test survives
+        # the file being moved and works whatever directory pytest runs from.
+        relative = Path("frontend") / "src" / "pages" / "PrintersPage.tsx"
+        source = next(
+            (candidate for parent in Path(__file__).resolve().parents if (candidate := parent / relative).is_file()),
+            None,
+        )
+        if source is None:
+            pytest.skip("frontend sources not present in this checkout")
+        text = source.read_text(encoding="utf-8")
+        match = re.search(
+            r"const MODELS_WITH_EXHAUST_LABEL:[^=]*=\s*new Set\(\[(.*?)\]\)",
+            text,
+            re.DOTALL,
+        )
+        assert match, "MODELS_WITH_EXHAUST_LABEL not found in PrintersPage.tsx"
+        return set(re.findall(r"['\"]([^'\"]+)['\"]", match.group(1)))
+
+    def test_frontend_list_is_the_display_name_subset_of_the_backend_list(self):
+        from backend.app.utils.printer_models import EXHAUST_FAN_LABEL_MODELS
+
+        frontend = self._frontend_models()
+        assert frontend, "frontend list parsed as empty"
+        missing = frontend - set(EXHAUST_FAN_LABEL_MODELS)
+        assert not missing, (
+            f"models {sorted(missing)} label the fan 'Exhaust' in the UI but the backend "
+            f"would report 'Chamber fan' — add them to EXHAUST_FAN_LABEL_MODELS"
+        )
+
+    def test_every_backend_display_name_is_handled_by_the_frontend(self):
+        from backend.app.utils.printer_models import EXHAUST_FAN_LABEL_MODELS
+
+        # N7/N6 are internal codes that never reach the card, so exclude them.
+        internal_codes = {"N7", "N6"}
+        backend_display = set(EXHAUST_FAN_LABEL_MODELS) - internal_codes
+        missing = backend_display - self._frontend_models()
+        assert not missing, (
+            f"models {sorted(missing)} say 'Exhaust fan' in the API response but the card "
+            f"would still show 'Chamber Fan' — add them to MODELS_WITH_EXHAUST_LABEL"
+        )

backend/tests/unit/services/test_rest_smart_plug.py

@@ -49,11 +49,39 @@ class TestURLValidation:
     def test_hostname_url(self, service):
         assert service._validate_url("http://openhab.local:8080/api") is True
 
-    def test_loopback_blocked(self, service):
-        assert service._validate_url("http://127.0.0.1/api") is False
+    def test_loopback_allowed(self, service):
+        """Deliberate change: the LAN-service policy permits loopback, because
+        an openHAB/Node-RED bridge on the same host is the normal topology.
 
-    def test_link_local_blocked(self, service):
-        assert service._validate_url("http://169.254.1.1/api") is False
+        The previous check rejected a literal 127.0.0.1 while accepting the
+        equivalent "localhost", so the same target was configurable one way and
+        not the other. See test_outbound_url_ssrf_guards.py for the policy.
+        """
+        assert service._validate_url("http://127.0.0.1/api") is True
+
+    def test_link_local_allowed(self, service):
+        """Also deliberate: a generic APIPA address is a LAN host like any
+        other. The cloud-metadata address inside that range is blocked by
+        name, not by rejecting the whole /16 — see test_metadata_blocked."""
+        assert service._validate_url("http://169.254.1.1/api") is True
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://169.254.169.254/latest/meta-data/",
+            "http://100.100.100.200/",
+            "http://[fd00:ec2::254]/",
+            "http://metadata.google.internal/",
+            "http://[::ffff:169.254.169.254]/",
+            "http://2130706433/",
+            "http://0.0.0.0/",
+        ],
+    )
+    def test_metadata_and_encoded_targets_blocked(self, service, url):
+        """The gap the previous hand-rolled check left: anything that was not a
+        bare IP literal fell through to True, and the literals it did parse were
+        only tested for loopback/link-local."""
+        assert service._validate_url(url) is False
 
     def test_empty_hostname(self, service):
         assert service._validate_url("http:///api") is False
@@ -405,6 +433,10 @@ class TestTestConnection:
 
     @pytest.mark.asyncio
     async def test_connection_invalid_url(self, service):
-        result = await service.test_connection("http://127.0.0.1/api")
+        """127.0.0.1 is now permitted (see TestURLValidation), so the rejection
+        case here is a target that is out of policy under any topology. The
+        error is the guard's own message rather than a fixed sentence, so the
+        user learns which rule the URL broke."""
+        result = await service.test_connection("http://169.254.169.254/latest/meta-data/")
         assert result["success"] is False
-        assert "blocked" in result["error"].lower()
+        assert "cloud metadata" in result["error"].lower()

backend/tests/unit/test_outbound_url_ssrf_guards.py

@@ -40,6 +40,9 @@ from backend.app.api.routes._url_safety import assert_safe_lan_service_url
 from backend.app.schemas.auth import OIDCProviderCreate, OIDCProviderUpdate
 from backend.app.schemas.settings import LAN_SERVICE_URL_SETTINGS, AppSettingsUpdate
 from backend.app.services import notification_service as ns
+from backend.app.services.homeassistant import HomeAssistantService
+from backend.app.services.rest_smart_plug import RESTSmartPlugService
+from backend.app.services.tasmota import TasmotaService
 
 # Dangerous under any topology — both tiers must reject all of these.
 UNIVERSALLY_BLOCKED = [
@@ -54,6 +57,11 @@ UNIVERSALLY_BLOCKED = [
     "http://[::ffff:169.254.169.254]/",
     "http://0.0.0.0/",
     "http://239.255.255.250/",
+    # The DNS-name form of the same target. Neither tier resolves hostnames,
+    # but these are a fixed literal set, so matching them costs no lookup.
+    "http://metadata.google.internal/",
+    "http://METADATA.GOOGLE.INTERNAL/computeMetadata/v1/",
+    "http://metadata.goog/",
 ]
 
 # The normal self-hosted topology — the LAN tier must permit all of these.
@@ -374,3 +382,269 @@ async def test_test_config_refuses_metadata_targets_without_a_request(provider_t
     assert success is False
     assert called is False
     assert "cloud metadata" in message
+
+
+# ---------------------------------------------------------------------------
+# Smart plugs: the same request-body-URL shape as the notification test endpoint
+# ---------------------------------------------------------------------------
+#
+# POST /smart-plugs/{ha,rest}/test-connection take their URL from the request
+# body and never persist it, so the schema-layer validator on ``ha_url`` does
+# not apply. Both are reachable with only ``SMART_PLUGS_CONTROL``, which the
+# default Operators group carries and which does NOT imply ``SETTINGS_UPDATE``
+# — identical to the notification case above.
+#
+# Both previously used hand-rolled checks that got the policy wrong in both
+# directions: the REST one rejected a literal ``127.0.0.1`` while allowing
+# every non-literal hostname, and the HA one matched three literal strings and
+# never parsed the hostname as an IP at all.
+
+
+@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
+def test_rest_plug_guard_rejects_dangerous_targets(url: str):
+    assert RESTSmartPlugService._validate_url(url) is False
+
+
+@pytest.mark.parametrize("url", LAN_ALLOWED)
+def test_rest_plug_guard_permits_the_normal_self_hosted_topology(url: str):
+    """Includes literal 127.0.0.1, which the previous implementation rejected
+    while accepting the equivalent "localhost" — a plug bridge on the same
+    host could only be configured by spelling it one particular way."""
+    assert RESTSmartPlugService._validate_url(url) is True
+
+
+@pytest.mark.parametrize("url", UNIVERSALLY_BLOCKED)
+def test_ha_guard_rejects_dangerous_targets(url: str):
+    assert HomeAssistantService._validate_url(url) is None
+
+
+@pytest.mark.parametrize("url", LAN_ALLOWED)
+def test_ha_guard_permits_the_normal_self_hosted_topology(url: str):
+    assert HomeAssistantService._validate_url(url) is not None
+
+
+def test_ha_guard_still_normalises_the_url_it_returns():
+    """Delegating the policy must not change what the caller gets back:
+    scheme+host+port+path, with query and fragment dropped."""
+    assert HomeAssistantService._validate_url("http://192.168.1.5:8123/base?x=1#f") == "http://192.168.1.5:8123/base"
+    assert HomeAssistantService._validate_url("http://ha.lan") == "http://ha.lan"
+
+
+def test_ha_guard_keeps_ipv6_literals_bracketed():
+    """urlparse strips the brackets off an IPv6 host; re-emitting it without
+    them yields an unparseable URL that httpx cannot dial."""
+    assert HomeAssistantService._validate_url("http://[fd00::1]:8123/api") == "http://[fd00::1]:8123/api"
+
+
+@pytest.mark.parametrize("ip", ["169.254.169.254", "100.100.100.200", "fd00:ec2::254", "0.0.0.0", "239.255.255.250"])
+def test_tasmota_guard_rejects_metadata_and_misuse_addresses(ip: str):
+    """Tasmota keeps its own stricter rule (bare IP literals only, loopback
+    rejected — a plug is always a separate LAN device), but must not miss the
+    destinations that are dangerous regardless of topology."""
+    assert TasmotaService._validate_ip(ip) is False
+
+
+@pytest.mark.parametrize("ip", ["::ffff:169.254.169.254", "::ffff:100.100.100.200"])
+def test_tasmota_guard_unwraps_ipv4_mapped_ipv6(ip: str):
+    assert TasmotaService._validate_ip(ip) is False
+
+
+@pytest.mark.parametrize("ip", ["192.168.1.50", "10.0.0.7", "172.16.4.9"])
+def test_tasmota_guard_still_permits_a_normal_lan_plug(ip: str):
+    assert TasmotaService._validate_ip(ip) is True
+
+
+@pytest.mark.parametrize("ip", ["127.0.0.1", "tasmota.local", "not-an-ip"])
+def test_tasmota_guard_keeps_failing_closed_on_non_lan_device_values(ip: str):
+    """Deliberately stricter than the shared LAN guard, and unchanged here."""
+    assert TasmotaService._validate_ip(ip) is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "target",
+    ["http://169.254.169.254/", "http://100.100.100.200/", "http://metadata.google.internal/"],
+)
+async def test_rest_test_connection_refuses_metadata_without_a_request(target: str, monkeypatch):
+    """End-to-end shape of the reported attack, mirroring the notification
+    test above: an unsaved URL aimed at IMDS via the test endpoint must be
+    refused before any HTTP call is made."""
+
+    def _fail_if_called(*_a, **_kw):
+        raise AssertionError("outbound request should not have been attempted")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _fail_if_called)
+    result = await RESTSmartPlugService().test_connection(target, "GET", None)
+
+    assert result["success"] is False
+    assert "cloud metadata" in result["error"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "target",
+    ["http://169.254.169.254", "http://100.100.100.200", "http://metadata.google.internal"],
+)
+async def test_ha_test_connection_refuses_metadata_without_a_request(target: str, monkeypatch):
+    def _fail_if_called(*_a, **_kw):
+        raise AssertionError("outbound request should not have been attempted")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _fail_if_called)
+    result = await HomeAssistantService().test_connection(target, "token")
+
+    assert result["success"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("target", ["http://169.254.169.254", "http://metadata.google.internal"])
+async def test_obico_test_connection_refuses_metadata_without_a_request(target: str, monkeypatch):
+    """Same shape again: obico_ml_url is guarded when saved via settings, but
+    this route takes the URL from the request body and echoes the response."""
+    from backend.app.services.obico_detection import ObicoDetectionService
+
+    def _fail_if_called(*_a, **_kw):
+        raise AssertionError("outbound request should not have been attempted")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _fail_if_called)
+    result = await ObicoDetectionService().test_connection(target)
+
+    assert result["ok"] is False
+    assert result["body"] is None
+    assert "cloud metadata" in result["error"]
+
+
+# ---------------------------------------------------------------------------
+# Drift backstop, part 2: URLs that arrive in a request body
+# ---------------------------------------------------------------------------
+#
+# `test_every_url_setting_is_either_guarded_or_explicitly_exempt` above only
+# walks `AppSettingsUpdate`. That is why the notification test endpoint, and
+# then the two smart-plug test endpoints, each had to be found by hand: a URL
+# that arrives in a request body and is never persisted is not a settings
+# field, so nothing enumerated it. This walks the live route table instead.
+
+
+def _request_body_url_fields() -> set[tuple[str, str]]:
+    """Every (model, field) pair on a mutating route whose body carries a URL."""
+    from fastapi.routing import APIRoute
+    from pydantic import BaseModel
+
+    from backend.app.main import app
+
+    found: set[tuple[str, str]] = set()
+    for route in app.routes:
+        if not isinstance(route, APIRoute) or not ({"POST", "PUT", "PATCH"} & set(route.methods or ())):
+            continue
+        for param in route.dependant.body_params:
+            # FastAPI moved the resolved annotation from `type_` onto
+            # `field_info.annotation`; read both so this can't silently
+            # enumerate nothing (which would make the assertions vacuous).
+            annotation = getattr(param, "type_", None)
+            if annotation is None:
+                annotation = getattr(getattr(param, "field_info", None), "annotation", None)
+            if not (isinstance(annotation, type) and issubclass(annotation, BaseModel)):
+                continue
+            for field in annotation.model_fields:
+                if field == "url" or field.endswith("_url"):
+                    found.add((annotation.__name__, field))
+    return found
+
+
+# Guarded: the handler (or the service it calls) puts the value through one of
+# the two tiers before any request is issued.
+GUARDED_BODY_URLS = {
+    ("AppSettingsUpdate", "bambu_studio_api_url"),
+    ("AppSettingsUpdate", "ha_url"),
+    ("AppSettingsUpdate", "obico_ml_url"),
+    ("AppSettingsUpdate", "orcaslicer_api_url"),
+    ("AppSettingsUpdate", "spoolman_url"),  # assert_safe_spoolman_url at each consumer
+    ("HATestConnectionRequest", "url"),  # homeassistant._validate_url
+    ("RESTTestConnectionRequest", "url"),  # rest_smart_plug._validate_url
+    ("TestConnectionRequest", "url"),  # obico_detection.test_connection
+    ("OIDCProviderCreate", "issuer_url"),  # public tier, via schemas.auth
+    ("OIDCProviderCreate", "icon_url"),
+    ("OIDCProviderUpdate", "issuer_url"),
+    ("OIDCProviderUpdate", "icon_url"),
+    # Gitea/Forgejo derive their API base from this and request it with the
+    # stored token, so it is a real fetch target — guarded in
+    # github_backup._enforce_private_repo, which both POST and PATCH funnel through.
+    ("GitHubBackupConfigCreate", "repository_url"),
+    ("GitHubBackupConfigUpdate", "repository_url"),
+    # SmartPlug{Create,Update} persist these; every read goes back out through
+    # RESTSmartPlugService._send_request, which applies the same guard.
+    ("SmartPlugCreate", "rest_on_url"),
+    ("SmartPlugCreate", "rest_off_url"),
+    ("SmartPlugCreate", "rest_status_url"),
+    ("SmartPlugCreate", "rest_power_url"),
+    ("SmartPlugCreate", "rest_energy_url"),
+    ("SmartPlugUpdate", "rest_on_url"),
+    ("SmartPlugUpdate", "rest_off_url"),
+    ("SmartPlugUpdate", "rest_status_url"),
+    ("SmartPlugUpdate", "rest_power_url"),
+    ("SmartPlugUpdate", "rest_energy_url"),
+}
+
+# Not a destination Bambuddy requests — no guard applies.
+NOT_A_FETCH_TARGET = {
+    ("AppSettingsUpdate", "external_url"),  # Bambuddy's own address (see exempt list above)
+    ("AppSettingsUpdate", "ldap_server_url"),  # ldap://, handed to an LDAP client
+    ("ProjectCreate", "url"),  # stored link, rendered in the UI, never fetched
+    ("ProjectUpdate", "url"),
+    ("BOMItemCreate", "sourcing_url"),  # stored supplier link, never fetched
+    ("BOMItemUpdate", "sourcing_url"),
+    ("MakerWorldResolveRequest", "url"),  # parsed for a model id; fetches go to a pinned CDN allowlist
+    ("DeviceRegisterRequest", "backend_url"),  # the device's view of Bambuddy's own address
+    ("HeartbeatRequest", "backend_url"),
+    ("SystemConfigRequest", "backend_url"),
+    ("ExternalLinkCreate", "url"),  # sidebar link, rendered in the UI, never requested
+    ("ExternalLinkUpdate", "url"),
+    ("MaintenanceTypeCreate", "wiki_url"),  # documentation link surfaced in the UI/notifications
+    ("MaintenanceTypeUpdate", "wiki_url"),
+    ("ArchiveUpdate", "external_url"),  # stored source link for the model, never fetched
+}
+
+# Genuinely unguarded, and deliberately recorded rather than quietly exempted.
+# These reach `external_camera.capture_frame`, which dials rtsp:// as well as
+# http(s):// — the LAN-service guard rejects any non-HTTP scheme, so wiring it
+# up as-is would break every RTSP camera. Closing these needs a scheme-aware
+# variant of the guard, not a one-line delegation.
+KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD = {
+    ("PrinterCreate", "external_camera_url"),
+    ("PrinterCreate", "external_camera_snapshot_url"),
+    ("PrinterUpdate", "external_camera_url"),
+    ("PrinterUpdate", "external_camera_snapshot_url"),
+}
+
+
+def test_the_route_walk_actually_finds_something():
+    """Guards the guard. If FastAPI's internals move again and the walk starts
+    returning nothing, both assertions below pass vacuously and the backstop
+    silently stops working — which is the exact failure it exists to prevent."""
+    found = _request_body_url_fields()
+    assert ("RESTTestConnectionRequest", "url") in found
+    assert ("HATestConnectionRequest", "url") in found
+    assert len(found) > 20
+
+
+def test_every_request_body_url_is_classified():
+    """A new URL-bearing request field can't land without a decision.
+
+    Add it to GUARDED_BODY_URLS once the handler runs it through a guard, or
+    to NOT_A_FETCH_TARGET with the reason it is never requested. Do not add
+    anything to KNOWN_UNGUARDED_* without also raising it.
+    """
+    classified = GUARDED_BODY_URLS | NOT_A_FETCH_TARGET | KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD
+    unclassified = _request_body_url_fields() - classified
+    assert not unclassified, (
+        f"Unclassified request-body URL field(s): {sorted(unclassified)}. Route the value "
+        f"through a guard and list it in GUARDED_BODY_URLS, or list it in NOT_A_FETCH_TARGET "
+        f"with the reason it is never fetched."
+    )
+
+
+def test_classification_lists_do_not_drift_from_the_routes():
+    """The reverse direction: a stale entry means a route was renamed or
+    removed and the list was not updated, which would hide the next one."""
+    actual = _request_body_url_fields()
+    stale = (GUARDED_BODY_URLS | NOT_A_FETCH_TARGET | KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD) - actual
+    assert not stale, f"Classification entries no longer match any route: {sorted(stale)}"

backend/tests/unit/test_plate_clear_mqtt_notification.py

@@ -44,6 +44,8 @@ def _state() -> SimpleNamespace:
         big_fan1_speed=0,
         big_fan2_speed=0,
         heatbreak_fan_speed=0,
+        left_aux_fan_speed=None,
+        exhaust_fan_present=False,
     )
 
 

backend/tests/unit/test_printer_manager_status_broadcast.py

@@ -90,6 +90,8 @@ def _fake_state(**overrides):
         "firmware_version": None,
         "gcode_file": None,
         "heatbreak_fan_speed": None,
+        "left_aux_fan_speed": None,
+        "exhaust_fan_present": False,
         "layer_num": None,
         "remaining_time": None,
         "speed_level": None,

frontend/src/__tests__/pages/PrintersPage.test.tsx

@@ -310,6 +310,113 @@ describe('PrintersPage', () => {
         expect(screen.getByTitle('Chamber Fan')).toBeInTheDocument();
       });
     });
+
+    // P2S/X2D left auxiliary part cooling fan (airduct part id 10) — optional
+    // hardware, so the badge must only appear when the firmware reports it.
+    const renderWithStatus = (
+      printer: typeof mockPrinters[number],
+      status: Record<string, unknown>,
+    ) => {
+      server.use(
+        http.get('/api/v1/printers/', () => HttpResponse.json([printer])),
+        http.get('/api/v1/printers/:id/status', () => HttpResponse.json(status)),
+      );
+      render(<PrintersPage />);
+    };
+
+    it('shows the exhaust tile labeled "Exhaust" on P2S when the kit is present', async () => {
+      // Exhaust fan is an add-on kit; the tile appears only when the printer
+      // reports it (airduct part id 3 -> exhaust_fan_present).
+      renderWithStatus(
+        { ...mockPrinters[0], model: 'P2S' },
+        { ...statusWithFans, exhaust_fan_present: true },
+      );
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Part Cooling Fan')).toBeInTheDocument();
+      });
+      expect(screen.getByTitle('Exhaust')).toBeInTheDocument();
+      expect(screen.queryByTitle('Chamber Fan')).not.toBeInTheDocument();
+    });
+
+    it('hides the exhaust tile on a base P2S without the kit', async () => {
+      renderWithStatus(
+        { ...mockPrinters[0], model: 'P2S' },
+        { ...statusWithFans, exhaust_fan_present: false },
+      );
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Part Cooling Fan')).toBeInTheDocument();
+      });
+      expect(screen.queryByTitle('Exhaust')).not.toBeInTheDocument();
+      expect(screen.queryByTitle('Chamber Fan')).not.toBeInTheDocument();
+    });
+
+    it('keeps the always-on "Chamber Fan" tile on X1C regardless of exhaust_fan_present', async () => {
+      renderWithStatus(
+        { ...mockPrinters[0], model: 'X1C' },
+        { ...statusWithFans, exhaust_fan_present: false },
+      );
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Chamber Fan')).toBeInTheDocument();
+      });
+      expect(screen.queryByTitle('Exhaust')).not.toBeInTheDocument();
+    });
+
+    it('hides the left aux badge when the accessory is not reported', async () => {
+      renderWithStatus({ ...mockPrinters[0], model: 'P2S' }, statusWithFans);
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Part Cooling Fan')).toBeInTheDocument();
+      });
+      expect(screen.queryByTitle('Left Auxiliary Fan')).not.toBeInTheDocument();
+    });
+
+    it('orders the fan badges left-to-right: part, left aux, aux, exhaust', async () => {
+      // The two aux badges should read in the same order as the physical
+      // hardware, so the left fan sits before the right one.
+      renderWithStatus(
+        { ...mockPrinters[0], model: 'P2S' },
+        { ...statusWithFans, left_aux_fan_speed: 80, exhaust_fan_present: true },
+      );
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Left Auxiliary Fan')).toBeInTheDocument();
+      });
+
+      const order = ['Part Cooling Fan', 'Left Auxiliary Fan', 'Auxiliary Fan', 'Exhaust'].map(
+        (title) => screen.getByTitle(title),
+      );
+      for (let i = 1; i < order.length; i++) {
+        // Node.compareDocumentPosition returns FOLLOWING (4) when the argument
+        // comes after the reference node in document order.
+        expect(order[i - 1].compareDocumentPosition(order[i])).toBe(Node.DOCUMENT_POSITION_FOLLOWING);
+      }
+    });
+
+    it('shows left aux fan badge when the accessory is installed (P2S)', async () => {
+      renderWithStatus(
+        { ...mockPrinters[0], model: 'P2S' },
+        { ...statusWithFans, left_aux_fan_speed: 80 },
+      );
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Left Auxiliary Fan')).toBeInTheDocument();
+      });
+    });
+
+    it('shows left aux fan badge even at 0% while installed', async () => {
+      renderWithStatus(
+        { ...mockPrinters[0], model: 'P2S' },
+        { ...statusWithFans, left_aux_fan_speed: 0 },
+      );
+
+      await waitFor(() => {
+        expect(screen.getByTitle('Left Auxiliary Fan')).toBeInTheDocument();
+      });
+    });
+
   });
 
   describe('empty state', () => {

frontend/src/api/client.ts

@@ -565,6 +565,11 @@ export interface PrinterStatus {
   big_fan1_speed: number | null;     // Auxiliary fan
   big_fan2_speed: number | null;     // Chamber/exhaust fan
   heatbreak_fan_speed: number | null; // Hotend heatbreak fan
+  // Left auxiliary part cooling fan (optional P2S/X2D accessory, M106 P10).
+  // null = not installed / not reported by this model.
+  left_aux_fan_speed: number | null;
+  // Chamber exhaust fan present (P2S/X2D External Exhaust Fan kit, airduct part 3).
+  exhaust_fan_present: boolean;
   firmware_version: string | null;   // Firmware version from MQTT
   // Developer LAN mode: true = enabled, false = disabled, null = unknown
   developer_mode: boolean | null;
@@ -3885,7 +3890,11 @@ export const api = {
       method: 'POST',
     }),
   testExternalCamera: (printerId: number, url: string, cameraType: string) =>
-    request<{ success: boolean; error?: string; resolution?: string }>(
+    // `coalesced` is true when the frame came from a capture that was already
+    // running (Obico polling, a snapshot) rather than a connection this test
+    // opened — a single-reader camera is shared rather than opened twice, so
+    // the result is real but says nothing about reaching the camera just now.
+    request<{ success: boolean; error?: string; resolution?: string; coalesced?: boolean }>(
       `/printers/${printerId}/camera/external/test?url=${encodeURIComponent(url)}&camera_type=${encodeURIComponent(cameraType)}`,
       { method: 'POST' }
     ),
@@ -3934,7 +3943,7 @@ export const api = {
       method: 'POST',
     }),
 
-  setFanSpeed: (printerId: number, fan: 'part' | 'aux' | 'chamber', speed: number) =>
+  setFanSpeed: (printerId: number, fan: 'part' | 'aux' | 'aux2' | 'chamber', speed: number) =>
     request<{ success: boolean; message: string }>(`/printers/${printerId}/fan-speed?fan=${fan}&speed=${speed}`, {
       method: 'POST',
     }),

frontend/src/i18n/locales/de.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: 'Bauteilkühlung',
       auxiliary: 'Hilfsventilator',
+      leftAuxiliary: 'Linker Hilfsventilator',
+      exhaust: 'Abluft',
       chamber: 'Kammerventilator',
     },
     // HMS errors
@@ -2300,6 +2302,7 @@ export default {
       connectionFailed: 'Verbindung fehlgeschlagen',
       testFailed: 'Test fehlgeschlagen',
       cameraConnected: 'Kamera verbunden{{resolution}}',
+      cameraConnectedCoalesced: 'Kamera verbunden{{resolution}} (geteilt mit einer bereits laufenden Aufnahme)',
     },
     testConnection: 'Verbindung testen',
     catalog: {

frontend/src/i18n/locales/en.ts

@@ -672,6 +672,8 @@ export default {
     fans: {
       partCooling: 'Part Cooling Fan',
       auxiliary: 'Auxiliary Fan',
+      leftAuxiliary: 'Left Auxiliary Fan',
+      exhaust: 'Exhaust',
       chamber: 'Chamber Fan',
     },
     // HMS errors
@@ -2319,6 +2321,7 @@ export default {
       connectionFailed: 'Connection failed',
       testFailed: 'Test failed',
       cameraConnected: 'Camera connected{{resolution}}',
+      cameraConnectedCoalesced: 'Camera connected{{resolution}} (shared with a capture already running)',
     },
     testConnection: 'Test Connection',
     catalog: {

frontend/src/i18n/locales/es.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: 'Ventilador de refrigeración de piezas',
       auxiliary: 'Ventilador auxiliar',
+      leftAuxiliary: 'Ventilador auxiliar izquierdo',
+      exhaust: 'Extracción',
       chamber: 'Ventilador de la cámara',
     },
     // HMS errors
@@ -2303,6 +2305,7 @@ export default {
       connectionFailed: 'Error de conexión',
       testFailed: 'La prueba falló',
       cameraConnected: 'Cámara conectada{{resolution}}',
+      cameraConnectedCoalesced: 'Cámara conectada{{resolution}} (compartida con una captura ya en curso)',
     },
     testConnection: 'Probar conexión',
     catalog: {

frontend/src/i18n/locales/fr.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: 'Ventilateur pièce',
       auxiliary: 'Ventilateur auxiliaire',
+      leftAuxiliary: 'Ventilateur auxiliaire gauche',
+      exhaust: 'Extraction',
       chamber: 'Ventilateur chambre',
     },
     // HMS errors
@@ -2256,6 +2258,7 @@ export default {
       connectionFailed: 'Échec connexion',
       testFailed: 'Échec test',
       cameraConnected: 'Caméra connectée {{resolution}}',
+      cameraConnectedCoalesced: 'Caméra connectée {{resolution}} (partagée avec une capture déjà en cours)',
     },
     testConnection: 'Tester la connexion',
     catalog: {

frontend/src/i18n/locales/it.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: 'Ventola raffreddamento parte',
       auxiliary: 'Ventola ausiliaria',
+      leftAuxiliary: 'Ventola ausiliaria sinistra',
+      exhaust: 'Estrazione',
       chamber: 'Ventola camera',
     },
     // HMS errors
@@ -2256,6 +2258,7 @@ export default {
       connectionFailed: 'Connessione fallita',
       testFailed: 'Test fallito',
       cameraConnected: 'Camera connessa{{resolution}}',
+      cameraConnectedCoalesced: 'Camera connessa{{resolution}} (condivisa con un\'acquisizione già in corso)',
     },
     testConnection: 'Testa connessione',
     catalog: {

frontend/src/i18n/locales/ja.ts

@@ -667,6 +667,8 @@ export default {
     fans: {
       partCooling: 'パーツ冷却ファン',
       auxiliary: '補助ファン',
+      leftAuxiliary: '左補助ファン',
+      exhaust: '排気',
       chamber: 'チャンバーファン',
     },
     // HMS errors
@@ -2299,6 +2301,7 @@ export default {
       connectionFailed: '接続失敗',
       testFailed: 'テスト通知の送信に失敗しました',
       cameraConnected: 'カメラ接続{{resolution}}',
+      cameraConnectedCoalesced: 'カメラ接続{{resolution}}（実行中のキャプチャと共有）',
     },
     testConnection: '接続テスト',
     catalog: {

frontend/src/i18n/locales/ko.ts

@@ -625,6 +625,8 @@ export default {
     fans: {
       partCooling: '파트 냉각 팬',
       auxiliary: '보조 팬',
+      leftAuxiliary: '왼쪽 보조 팬',
+      exhaust: '배기',
       chamber: '챔버 팬'
     },
     clickToViewHmsErrors: 'HMS 오류 보기 클릭',
@@ -2168,6 +2170,7 @@ export default {
       connectionFailed: '연결 실패',
       testFailed: '테스트 실패',
       cameraConnected: '카메라 연결됨{{resolution}}',
+      cameraConnectedCoalesced: '카메라 연결됨{{resolution}} (이미 진행 중인 캡처와 공유됨)',
       passwordNeedsUppercase: '비밀번호에 대문자가 최소 1개 포함되어야 합니다',
       passwordNeedsLowercase: '비밀번호에 소문자가 최소 1개 포함되어야 합니다',
       passwordNeedsDigit: '비밀번호에 숫자가 최소 1개 포함되어야 합니다',

frontend/src/i18n/locales/pt-BR.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: 'Ventilador de resfriamento da peça',
       auxiliary: 'Ventilador auxiliar',
+      leftAuxiliary: 'Ventilador auxiliar esquerdo',
+      exhaust: 'Exaustão',
       chamber: 'Ventilador da câmara',
     },
     // HMS errors
@@ -2256,6 +2258,7 @@ export default {
       connectionFailed: 'Falha na conexão',
       testFailed: 'Falha no teste',
       cameraConnected: 'Câmera conectada{{resolution}}',
+      cameraConnectedCoalesced: 'Câmera conectada{{resolution}} (compartilhada com uma captura já em andamento)',
     },
     testConnection: 'Testar Conexão',
     catalog: {

frontend/src/i18n/locales/ru.ts

@@ -630,6 +630,8 @@ export default {
     fans: {
       partCooling: "Вентилятор обдува модели",
       auxiliary: "Дополнительный вентилятор",
+      leftAuxiliary: "Левый дополнительный вентилятор",
+      exhaust: "Вытяжка",
       chamber: "Вентилятор камеры",
     },
     clickToViewHmsErrors: "Нажмите, чтобы посмотреть ошибки HMS",
@@ -2173,6 +2175,7 @@ export default {
       connectionFailed: "Не удалось подключиться",
       testFailed: "Проверка завершилась ошибкой",
       cameraConnected: "Камера подключена{{resolution}}",
+      cameraConnectedCoalesced: "Камера подключена{{resolution}} (используется уже выполняющийся захват)",
     },
     testConnection: "Проверить подключение",
     catalog: {

frontend/src/i18n/locales/tr.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: 'Parça Soğutma Fanı',
       auxiliary: 'Yardımcı Fan',
+      leftAuxiliary: 'Sol Yardımcı Fan',
+      exhaust: 'Egzoz',
       chamber: 'Hazne Fanı',
     },
     // HMS hataları
@@ -2304,6 +2306,7 @@ export default {
       connectionFailed: 'Bağlantı başarısız',
       testFailed: 'Test başarısız',
       cameraConnected: 'Kamera bağlandı{{resolution}}',
+      cameraConnectedCoalesced: 'Kamera bağlandı{{resolution}} (hâlihazırda süren bir yakalamayla paylaşıldı)',
     },
     testConnection: 'Bağlantıyı Test Et',
     catalog: {

frontend/src/i18n/locales/uk.ts

@@ -672,6 +672,8 @@ export default {
     fans: {
       partCooling: "Вентилятор охолодження моделі",
       auxiliary: "Допоміжний вентилятор",
+      leftAuxiliary: "Лівий допоміжний вентилятор",
+      exhaust: "Витяжка",
       chamber: "Камерний вентилятор",
     },
     // HMS errors
@@ -2319,6 +2321,7 @@ export default {
       connectionFailed: "Помилка підключення",
       testFailed: "Тест не вдалося",
       cameraConnected: "Камера підключена{{resolution}}",
+      cameraConnectedCoalesced: "Камера підключена{{resolution}} (спільно з уже виконуваним захопленням)",
     },
     testConnection: "Тестове підключення",
     catalog: {

frontend/src/i18n/locales/zh-CN.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: '零件冷却风扇',
       auxiliary: '辅助风扇',
+      leftAuxiliary: '左辅助风扇',
+      exhaust: '排气',
       chamber: '腔室风扇',
     },
     // HMS errors
@@ -2301,6 +2303,7 @@ export default {
       connectionFailed: '连接失败',
       testFailed: '测试失败',
       cameraConnected: '摄像头已连接{{resolution}}',
+      cameraConnectedCoalesced: '摄像头已连接{{resolution}}（与正在进行的抓取共享）',
     },
     testConnection: '测试连接',
     catalog: {

frontend/src/i18n/locales/zh-TW.ts

@@ -668,6 +668,8 @@ export default {
     fans: {
       partCooling: '零件冷卻風扇',
       auxiliary: '輔助風扇',
+      leftAuxiliary: '左輔助風扇',
+      exhaust: '排氣',
       chamber: '腔室風扇',
     },
     // HMS errors
@@ -2301,6 +2303,7 @@ export default {
       connectionFailed: '連線失敗',
       testFailed: '測試失敗',
       cameraConnected: '攝影機已連線{{resolution}}',
+      cameraConnectedCoalesced: '攝影機已連線{{resolution}}（與進行中的擷取共用）',
     },
     testConnection: '測試連線',
     catalog: {

frontend/src/pages/PrintersPage.tsx

@@ -1490,6 +1490,17 @@ const MODELS_WITH_CHAMBER_FAN: ReadonlySet<string> = new Set([
   'H2S',
 ]);
 
+// On the P2S/X2D, the enclosure fan (big_fan2 / airduct part id 3) is a
+// dedicated chamber EXHAUST fan: it's its own control and stays the same
+// regardless of cooling/heating mode (unlike the aux fan, id 2, which a flap
+// re-tasks between part-cooling and chamber-filter recirculation). Bambu's own
+// firmware/UI and Bambu Studio (FAN_CHAMBER_0_IDX -> "Exhaust") label it
+// "Exhaust" on these models. Other enclosed models (X1/P1S/H2*) keep "Chamber".
+const MODELS_WITH_EXHAUST_LABEL: ReadonlySet<string> = new Set([
+  'P2S',
+  'X2D',
+]);
+
 // Map SSDP model codes to display names
 function mapModelCode(ssdpModel: string | null): string {
   if (!ssdpModel) return '';
@@ -2449,7 +2460,7 @@ function PrinterCard({
   });
 
   const fanSpeedMutation = useMutation({
-    mutationFn: ({ fan, speed }: { fan: 'part' | 'aux' | 'chamber'; speed: number }) =>
+    mutationFn: ({ fan, speed }: { fan: 'part' | 'aux' | 'aux2' | 'chamber'; speed: number }) =>
       api.setFanSpeed(printer.id, fan, speed),
     onMutate: async ({ fan, speed }) => {
       await queryClient.cancelQueries({ queryKey: ['printerStatus', printer.id] });
@@ -2457,6 +2468,7 @@ function PrinterCard({
       const fanField = {
         part: 'cooling_fan_speed',
         aux: 'big_fan1_speed',
+        aux2: 'left_aux_fan_speed',
         chamber: 'big_fan2_speed',
       }[fan];
       queryClient.setQueryData(['printerStatus', printer.id], (old: PrinterStatus | undefined) =>
@@ -3886,7 +3898,30 @@ function PrinterCard({
               // control that does nothing. Mirrors the enclosure-door badge
               // gate above.
               const hasChamberFan = MODELS_WITH_CHAMBER_FAN.has(printer.model ?? '');
-              const fanItems = [
+              // On P2S/X2D the big_fan2 fan is the dedicated chamber EXHAUST fan
+              // ("Exhaust" in Bambu's naming) and is an add-on kit, not preinstalled:
+              // show it only when the printer actually reports it (airduct part id 3,
+              // surfaced as exhaust_fan_present). Other enclosed models (X1/P1S/H2*)
+              // have a built-in chamber fan that's always present, so they keep the
+              // existing model-list gate and the "Chamber Fan" label.
+              const isExhaustModel = MODELS_WITH_EXHAUST_LABEL.has(printer.model ?? '');
+              const chamberFanLabel = isExhaustModel
+                ? t('printers.fans.exhaust')
+                : t('printers.fans.chamber');
+              // Composed rather than either/or so both lists stay live for
+              // P2S/X2D: the model must have an enclosure fan at all, AND —
+              // where that fan is an add-on kit — actually report it. Written
+              // as `isExhaustModel ? exhaust_fan_present : hasChamberFan` the
+              // P2S/X2D entries in MODELS_WITH_CHAMBER_FAN became unreachable,
+              // which reads as if removing them were safe.
+              const showChamberFan = hasChamberFan && (!isExhaustModel || status.exhaust_fan_present);
+              const fanItems: {
+                key: string;
+                label: string;
+                value: number;
+                Icon: typeof Fan;
+                activeClass: string;
+              }[] = [
                 {
                   key: 'part',
                   label: t('printers.fans.partCooling'),
@@ -3894,6 +3929,22 @@ function PrinterCard({
                   Icon: Fan,
                   activeClass: 'text-cyan-600 dark:text-cyan-400',
                 },
+                // Left auxiliary part cooling fan (optional P2S/X2D accessory).
+                // Only reported (non-null) when the firmware lists airduct part
+                // id 10, i.e. when the fan is physically installed. Placed
+                // before the right-hand auxiliary fan so the two aux badges read
+                // left-to-right in the same order as the physical hardware.
+                ...(status.left_aux_fan_speed != null
+                  ? [
+                      {
+                        key: 'aux2',
+                        label: t('printers.fans.leftAuxiliary'),
+                        value: status.left_aux_fan_speed,
+                        Icon: Wind,
+                        activeClass: 'text-indigo-600 dark:text-indigo-400',
+                      },
+                    ]
+                  : []),
                 {
                   key: 'aux',
                   label: t('printers.fans.auxiliary'),
@@ -3901,11 +3952,11 @@ function PrinterCard({
                   Icon: Wind,
                   activeClass: 'text-blue-600 dark:text-blue-400',
                 },
-                ...(hasChamberFan
+                ...(showChamberFan
                   ? [
                       {
                         key: 'chamber',
-                        label: t('printers.fans.chamber'),
+                        label: chamberFanLabel,
                         value: status.big_fan2_speed ?? 0,
                         Icon: AirVent,
                         activeClass: 'text-green-600 dark:text-green-400',
@@ -4157,7 +4208,7 @@ function PrinterCard({
                               isPending={fanSpeedMutation.isPending}
                               options={buildPresetOptions(fanSpeedPresets, '%')}
                               onClose={() => setStatusControlMenu(null)}
-                              onSubmit={(speed) => fanSpeedMutation.mutate({ fan: key as 'part' | 'aux' | 'chamber', speed })}
+                              onSubmit={(speed) => fanSpeedMutation.mutate({ fan: key as 'part' | 'aux' | 'aux2' | 'chamber', speed })}
                             />
                           )}
                         </div>

frontend/src/pages/SettingsPage.tsx

@@ -1164,7 +1164,15 @@ export function SettingsPage() {
       const result = await api.testExternalCamera(printerId, url, cameraType);
       setExtCameraTestResults(prev => ({ ...prev, [printerId]: result }));
       if (result.success) {
-        showToast(t('settings.toast.cameraConnected', { resolution: result.resolution || '' }), 'success');
+        // A shared capture means the frame is real but was not fetched over a
+        // connection this test opened, so say so rather than implying the
+        // camera was just reached.
+        showToast(
+          result.coalesced
+            ? t('settings.toast.cameraConnectedCoalesced', { resolution: result.resolution || '' })
+            : t('settings.toast.cameraConnected', { resolution: result.resolution || '' }),
+          'success'
+        );
       } else {
         showToast(result.error || t('settings.toast.connectionFailed'), 'error');
       }

static/index.html

@@ -26,8 +26,8 @@
 
     <!-- Splash screens for iOS -->
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
-    <script type="module" crossorigin src="/assets/index-apAuCUp0.js"></script>
-    <link rel="stylesheet" crossorigin href="/assets/index-D4bpNaiw.css">
+    <script type="module" crossorigin src="/assets/index-fmZ_9rRe.js"></script>
+    <link rel="stylesheet" crossorigin href="/assets/index-oReXTzKG.css">
   </head>
   <body>
     <div id="root"></div>