Merge branch 'dev' into feature/oidc-env-config

backend/tests/integration/test_timelapse_scan_session.py

@@ -100,14 +100,17 @@ async def test_scan_timelapse_attaches_and_persists_via_fresh_session(
 
     # base_name = Path("test_print.gcode.3mf").stem = "test_print.gcode", so this
     # video matches by name (strategy 1). .mp4 → no background conversion task.
+    video_bytes = b"fake-timelapse-video-bytes"
     matched = {
         "name": "test_print.gcode.mp4",
         "path": "/timelapse/test_print.gcode.mp4",
         "is_directory": False,
-        "size": 4096,
+        # Must equal len(video_bytes): the download is checked against the
+        # listing, and the file is re-listed afterwards to confirm the printer
+        # has stopped writing it (#2704).
+        "size": len(video_bytes),
         "mtime": None,
     }
-    video_bytes = b"fake-timelapse-video-bytes"
 
     with (
         patch("backend.app.services.bambu_ftp.list_files_async", AsyncMock(return_value=[matched])),
@@ -119,6 +122,9 @@ async def test_scan_timelapse_attaches_and_persists_via_fresh_session(
             "backend.app.services.bambu_ftp.download_file_bytes_async",
             AsyncMock(return_value=video_bytes),
         ) as mock_download,
+        # A successful attach now removes the printer's copy (#2704); without
+        # this the endpoint would open a real FTP connection to the fixture IP.
+        patch("backend.app.services.bambu_ftp.delete_archived_timelapse", AsyncMock()) as mock_delete,
     ):
         response = await async_client.post(f"/api/v1/archives/{archive.id}/timelapse/scan")
 
@@ -127,6 +133,7 @@ async def test_scan_timelapse_attaches_and_persists_via_fresh_session(
     assert data["status"] == "attached"
     assert data["filename"] == "test_print.gcode.mp4"
     mock_download.assert_awaited_once()
+    mock_delete.assert_awaited_once()
 
     # The write happened in the route's fresh session; confirm it was committed
     # by re-reading the row on the separate test session.

backend/tests/unit/services/test_bambu_mqtt.py

@@ -5,6 +5,7 @@ These tests focus on timelapse tracking during prints.
 """
 
 import json
+import logging
 import time
 
 import pytest
@@ -6653,3 +6654,326 @@ class TestKProfileResponseDoesNotClobberNozzle:
         mqtt_client.state.nozzles[0].nozzle_diameter = "0.8"
         mqtt_client._process_message({"print": {"nozzle_diameter": "0.4"}})
         assert mqtt_client.state.nozzles[0].nozzle_diameter == "0.4"
+
+
+class TestConnectRefusalReporting:
+    """#2698: a refused CONNACK must leave a trace.
+
+    ``_on_connect``'s failure branch used to be a bare ``connected = False``.
+    A printer refusing our access code then looked exactly like one that was
+    powered off: paho reports the follow-up drop as the generic "Unspecified
+    error", so the support bundle from a 30-second reconnect loop carried no
+    hint of the real cause. Bambu speaks MQTT 3.1.1, whose CONNACK return codes
+    4 and 5 paho maps to reason codes 134 / 135.
+    """
+
+    @pytest.fixture
+    def mqtt_client(self):
+        from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+        return BambuMQTTClient(
+            ip_address="192.168.1.100",
+            serial_number="TEST123",
+            access_code="12345678",
+        )
+
+    @staticmethod
+    def _connack(v3_return_code):
+        from paho.mqtt.client import convert_connack_rc_to_reason_code
+
+        return convert_connack_rc_to_reason_code(v3_return_code)
+
+    def test_no_error_recorded_before_any_attempt(self, mqtt_client):
+        assert mqtt_client.last_connect_error is None
+        assert mqtt_client.last_connect_error_name is None
+
+    @pytest.mark.parametrize("v3_rc", [4, 5])
+    def test_credential_refusal_recorded(self, mqtt_client, v3_rc, caplog):
+        with caplog.at_level(logging.WARNING):
+            mqtt_client._on_connect(None, None, None, self._connack(v3_rc))
+
+        assert mqtt_client.state.connected is False
+        assert mqtt_client.last_connect_error == "auth_rejected"
+        assert "refused" in caplog.text.lower()
+        # The remedy has to be in the log — that line is what a maintainer
+        # reads out of a support bundle.
+        assert "access code" in caplog.text.lower()
+        # Never leak the credential itself into a bundle.
+        assert "12345678" not in caplog.text
+
+    def test_non_credential_refusal_recorded_separately(self, mqtt_client, caplog):
+        # CONNACK 3 = server unavailable: a real refusal, but not about creds.
+        with caplog.at_level(logging.WARNING):
+            mqtt_client._on_connect(None, None, None, self._connack(3))
+
+        assert mqtt_client.last_connect_error == "refused"
+        assert "access code" not in caplog.text.lower()
+
+    def test_successful_connect_clears_previous_error(self, mqtt_client):
+        mqtt_client._on_connect(None, None, None, self._connack(5))
+        assert mqtt_client.last_connect_error == "auth_rejected"
+
+        mock_client = type("MockClient", (), {"subscribe": lambda self, topic: (0, 1)})()
+        mqtt_client._on_connect(mock_client, None, None, 0)
+
+        assert mqtt_client.state.connected is True
+        assert mqtt_client.last_connect_error is None
+        assert mqtt_client.last_connect_error_name is None
+
+    def test_disconnect_line_carries_the_refusal(self, mqtt_client, caplog):
+        """The reconnect loop is what fills the log, so it must say why."""
+        mqtt_client._on_connect(None, None, None, self._connack(5))
+        caplog.clear()
+
+        with caplog.at_level(logging.WARNING):
+            mqtt_client._on_disconnect(None, None)
+
+        assert "MQTT disconnected" in caplog.text
+        assert "refused" in caplog.text
+        assert "Not authorized" in caplog.text
+
+    def test_disconnect_line_unchanged_without_a_refusal(self, mqtt_client, caplog):
+        with caplog.at_level(logging.WARNING):
+            mqtt_client._on_disconnect(None, None)
+
+        assert "MQTT disconnected" in caplog.text
+        assert "refused" not in caplog.text
+
+
+class TestEndOfPrintProbe:
+    """Tests for #2547: the end-of-print telemetry probe.
+
+    The probe exists to answer a question no existing support bundle can:
+    what do the stage/action fields do between the last object layer and
+    gcode_state=FINISH? stg_cur=22 was supposed to mark "toolhead parked,
+    before filament unload" (#1721) and fires on no model in the field, and
+    Bambuddy drops every other stage field unread. These tests pin the
+    window's boundaries and the guarantee that instrumentation stays
+    instrumentation — it must never raise into the ingest path.
+    """
+
+    LOGGER = "backend.app.services.bambu_mqtt"
+
+    @pytest.fixture
+    def mqtt_client(self):
+        from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+        client = BambuMQTTClient(
+            ip_address="192.168.1.100",
+            serial_number="TEST123",
+            access_code="12345678",
+        )
+        client._was_running = True
+        client.state.state = "RUNNING"
+        client.state.total_layers = 100
+        client.state.layer_num = 98
+        client.state.progress = 90.0
+        client.state.remaining_time = 12
+        return client
+
+    def test_silent_when_debug_logging_is_off(self, mqtt_client, caplog):
+        """The probe is a debug tool; at INFO it must cost nothing and say
+        nothing, including for a frame that would otherwise open the window."""
+        with caplog.at_level(logging.INFO, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+
+        assert "EOP-PROBE" not in caplog.text
+        assert mqtt_client._eop_probe_open is False
+
+    def test_does_not_open_mid_print(self, mqtt_client, caplog):
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 99, "mc_percent": 91}})
+
+        assert "EOP-PROBE" not in caplog.text
+        assert mqtt_client._eop_probe_open is False
+
+    def test_opens_on_the_last_layer_frame_itself(self, mqtt_client, caplog):
+        """The frame carrying the signal must be captured, not just the ones
+        after it — so the probe has to read the raw frame rather than state,
+        which _update_state only updates further down the same call."""
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100, "stg_cur": 0}})
+
+        assert "EOP-PROBE open" in caplog.text
+        assert "'layer_num': 100" in caplog.text
+        assert mqtt_client._eop_probe_open is True
+
+    def test_opens_on_progress_when_the_last_layer_packet_is_missed(self, mqtt_client, caplog):
+        """The layer_num edge is a single transient packet and is dropped
+        intermittently (the reason #1867 needed a second mechanism). Progress
+        has to be able to open the window on its own."""
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"mc_percent": 99}})
+
+        assert "EOP-PROBE open" in caplog.text
+        assert mqtt_client._eop_probe_open is True
+
+    def test_zero_remaining_does_not_open_before_the_print_progresses(self, mqtt_client, caplog):
+        """mc_remaining_time reads 0 during pre-print calibration too, so it
+        only counts once progress is non-zero."""
+        mqtt_client.state.progress = 0.0
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"mc_remaining_time": 0, "mc_percent": 0}})
+
+        assert "EOP-PROBE" not in caplog.text
+
+    def test_does_not_open_when_the_print_never_ran(self, mqtt_client, caplog):
+        """Bambuddy restarted mid-print, or firmware replayed a stale frame."""
+        mqtt_client._was_running = False
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+
+        assert "EOP-PROBE" not in caplog.text
+
+    def test_logs_only_changed_fields_after_opening(self, mqtt_client, caplog):
+        """Most probed fields are static across the window; logging all of
+        them every frame would bury the transitions we're looking for."""
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100, "stg_cur": 0}})
+            caplog.clear()
+            # Identical frame — nothing moved, so nothing to say.
+            mqtt_client._process_message({"print": {"layer_num": 100, "stg_cur": 0}})
+            assert "EOP-PROBE" not in caplog.text
+
+            mqtt_client._process_message({"print": {"layer_num": 100, "stg_cur": 22}})
+
+        assert "'stg_cur': 22" in caplog.text
+        assert "layer_num" not in caplog.text.split("EOP-PROBE")[-1]
+
+    def test_captures_the_fields_bambuddy_does_not_parse(self, mqtt_client, caplog):
+        """The whole point: mc_stage / mc_action / print_real_action are read
+        by nothing else in the codebase, so only the probe can show them."""
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+            mqtt_client._process_message(
+                {
+                    "print": {
+                        "mc_stage": 3,
+                        "mc_action": 8,
+                        "print_real_action": 2,
+                        "print_gcode_action": 5,
+                        "stg_cd": 1,
+                        "home_flag": 2231371,
+                        "spd_lvl": 0,
+                    }
+                }
+            )
+
+        for field in ("mc_stage", "mc_action", "print_real_action", "print_gcode_action", "stg_cd", "spd_lvl"):
+            assert field in caplog.text
+
+    def test_closes_on_finish_and_does_not_reopen(self, mqtt_client, caplog):
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+            mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
+
+            assert "EOP-PROBE 2 CLOSE" in caplog.text
+            assert mqtt_client._eop_probe_open is False
+            assert mqtt_client._eop_probe_armed is False
+
+            caplog.clear()
+            # Firmware re-sending FINISH, or a stale replay, must not restart it.
+            mqtt_client._process_message({"print": {"layer_num": 100, "mc_percent": 100}})
+
+        assert "EOP-PROBE" not in caplog.text
+
+    def test_closing_frame_is_self_contained_when_nothing_changed(self, mqtt_client, caplog):
+        """A FINISH frame that repeats values already seen still has to log
+        something — otherwise the window has no visible end."""
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"gcode_state": "RUNNING", "mc_percent": 100}})
+            caplog.clear()
+            mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
+            # Same value the probe already recorded on the opening frame.
+            mqtt_client._eop_probe_open = True
+            mqtt_client._eop_probe_armed = True
+            mqtt_client._eop_probe_last = {"gcode_state": "FINISH"}
+            mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
+
+        assert "CLOSE" in caplog.text
+        assert "'gcode_state': 'FINISH'" in caplog.text
+
+    def test_rearms_for_the_next_print(self, mqtt_client, caplog):
+        # A completion callback is what lets _update_state finish the print
+        # (and clear _was_running), which the new-print detection depends on.
+        mqtt_client.on_print_complete = lambda data: None
+        mqtt_client.state.gcode_file = "current.3mf"
+        mqtt_client._previous_gcode_state = "RUNNING"
+
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+            mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
+            assert mqtt_client._eop_probe_armed is False
+
+            # New print: RUNNING again with a file, after the previous print
+            # completed. _update_state rearms the probe alongside the
+            # finish-photo one-shot.
+            mqtt_client._process_message(
+                {"print": {"gcode_state": "RUNNING", "gcode_file": "next.3mf", "subtask_name": "next"}}
+            )
+            assert mqtt_client._eop_probe_armed is True
+
+            caplog.clear()
+            mqtt_client.state.total_layers = 50
+            mqtt_client._process_message({"print": {"layer_num": 50}})
+
+        assert "EOP-PROBE open" in caplog.text
+
+    def test_frame_budget_caps_output_but_still_logs_the_close(self, mqtt_client, caplog):
+        """A long final layer holds the window open at ~1 frame/second; the
+        user still has to be able to upload the resulting log."""
+        from backend.app.services.bambu_mqtt import _END_OF_PRINT_PROBE_MAX_FRAMES
+
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+            for i in range(_END_OF_PRINT_PROBE_MAX_FRAMES + 50):
+                mqtt_client._process_message({"print": {"mc_remaining_time": i}})
+
+            assert "frame budget" in caplog.text
+            caplog.clear()
+            mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
+
+        assert "CLOSE" in caplog.text
+
+    def test_opens_on_numeric_strings(self, mqtt_client, caplog):
+        """Firmware sends these as ints or as numeric strings depending on
+        model and field, so the window checks must coerce rather than compare
+        a str against an int and silently never open."""
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": "100", "mc_percent": "99"}})
+
+        assert "EOP-PROBE open" in caplog.text
+        assert mqtt_client._eop_probe_open is True
+
+    def test_coercion_helper_falls_back_on_junk(self, mqtt_client):
+        """Unit-level, because feeding junk through _process_message would trip
+        the pre-existing parsers before ever reaching the probe. The guarantee
+        under test is only that the probe's own reads can't raise."""
+        assert mqtt_client._probe_number("100") == 100.0
+        assert mqtt_client._probe_number("not-a-number", 7) == 7
+        assert mqtt_client._probe_number(None) is None
+        assert mqtt_client._probe_number({"unexpected": "shape"}, 0) == 0
+
+    def test_probe_failure_cannot_break_ingest(self, mqtt_client, caplog, monkeypatch):
+        """Instrumentation must stay instrumentation: if the probe ever throws,
+        state parsing still has to complete."""
+
+        def boom(_data):
+            raise RuntimeError("probe exploded")
+
+        monkeypatch.setattr(mqtt_client, "_probe_end_of_print", boom)
+
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"gcode_state": "RUNNING", "layer_num": 100}})
+
+        assert mqtt_client.state.layer_num == 100
+        assert "EOP-PROBE failed" in caplog.text
+
+    def test_never_logs_the_access_code(self, mqtt_client, caplog):
+        with caplog.at_level(logging.DEBUG, logger=self.LOGGER):
+            mqtt_client._process_message({"print": {"layer_num": 100}})
+            mqtt_client._process_message({"print": {"gcode_state": "FINISH"}})
+
+        probe_lines = [line for line in caplog.text.splitlines() if "EOP-PROBE" in line]
+        assert probe_lines
+        assert not any("12345678" in line for line in probe_lines)

backend/tests/unit/services/test_printer_diagnostic.py

@@ -51,6 +51,7 @@ class _Env:
         state=None,
         test_connection_success=True,
         report_messages_since_connect: int | None = 5,
+        connect_error: str | None = None,
     ):
         self.ports = ports or _port_probe()
         self.in_docker = in_docker
@@ -61,17 +62,26 @@ class _Env:
         # ``None`` means get_client returns None (e.g. pre-add flow); an int
         # means there's a client with that counter value.
         self.report_messages_since_connect = report_messages_since_connect
+        # CONNACK-refusal slug the live client reports, or None when the last
+        # connection attempt was never refused (#2698).
+        self.connect_error = connect_error
         self._stack = ExitStack()
 
     def __enter__(self):
         manager = MagicMock()
         manager.get_status.return_value = self.state
-        manager.test_connection = AsyncMock(return_value={"success": self.test_connection_success})
+        manager.test_connection = AsyncMock(
+            return_value={
+                "success": self.test_connection_success,
+                "reason": None if self.test_connection_success else self.connect_error,
+            }
+        )
         if self.report_messages_since_connect is None:
             manager.get_client.return_value = None
         else:
             client = MagicMock()
             client.report_messages_since_connect = self.report_messages_since_connect
+            client.last_connect_error = self.connect_error
             manager.get_client.return_value = client
         self._stack.enter_context(patch(f"{MOD}._check_port", new_callable=AsyncMock, side_effect=self.ports))
         self._stack.enter_context(patch(f"{MOD}.is_running_in_docker", return_value=self.in_docker))
@@ -248,6 +258,52 @@ class TestExistingPrinter:
         assert params == {}
 
 
+class TestAuthRejectedReason:
+    """#2698: "not connected" and "credentials refused" are different answers.
+
+    `state.connected == False` only says we have no session — the printer may
+    be rebooting, at its connection limit, or refusing the access code. When
+    the printer actually sent a CONNACK refusal the client records it, and the
+    check surfaces it as a `params.reason` variant so the UI can name the cause
+    instead of making the user guess. Without a recorded refusal the params
+    stay empty and the generic text is used.
+    """
+
+    def _params(self, result):
+        return next(c.params for c in result.checks if c.id == "mqtt_auth")
+
+    async def test_recorded_refusal_surfaces_reason(self):
+        with _Env(state=_state(connected=False), connect_error="auth_rejected"):
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer())
+        assert _statuses(result)["mqtt_auth"] == "fail"
+        assert self._params(result) == {"reason": "auth_rejected"}
+
+    async def test_disconnected_without_refusal_stays_generic(self):
+        with _Env(state=_state(connected=False)):
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer())
+        assert _statuses(result)["mqtt_auth"] == "fail"
+        assert self._params(result) == {}
+
+    async def test_unknown_slug_falls_back_to_generic(self):
+        # `refused` has no dedicated message — degrade to the plain fail text
+        # rather than asking the frontend for a key that doesn't exist.
+        with _Env(state=_state(connected=False), connect_error="refused"):
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer())
+        assert self._params(result) == {}
+
+    async def test_connected_printer_carries_no_reason(self):
+        with _Env(state=_state(connected=True), connect_error="auth_rejected"):
+            result = await run_connection_diagnostic("192.168.1.50", printer=_printer())
+        assert _statuses(result)["mqtt_auth"] == "pass"
+        assert self._params(result) == {}
+
+    async def test_pre_add_probe_surfaces_reason(self):
+        with _Env(test_connection_success=False, connect_error="auth_rejected"):
+            result = await run_connection_diagnostic("192.168.1.50", serial_number="01P", access_code="wrong")
+        assert _statuses(result)["mqtt_auth"] == "fail"
+        assert self._params(result) == {"reason": "auth_rejected"}
+
+
 class TestPreAddFlow:
     async def test_bad_credentials_fail_mqtt_auth(self):
         with _Env(test_connection_success=False):

backend/tests/unit/test_a2l_ams_lite_2619.py

@@ -20,6 +20,7 @@ from backend.app.services.bambu_mqtt import (
     A2L_LITE_PHYSICAL_AMS_ID,
     BambuMQTTClient,
     a2l_lite_wire_ids,
+    apply_tray_exist_bits,
     normalize_am_unit_id,
 )
 
@@ -142,6 +143,71 @@ class TestTrayNowGlobalisation:
         assert client.state.last_loaded_tray == 26
 
 
+class TestTrayExistBitsBitBase:
+    """#2697: ``apply_tray_exist_bits`` is reached with BOTH ids.
+
+    ``_handle_ams_data`` normalises 16 -> 6 before calling it, but the VP
+    bridge parses the raw printer payload itself and still holds the physical
+    16. Reading 16 as ``16 * 4`` lands on bits 64-67, where nothing is ever
+    set, so every A2L slot was wiped in the slicer-facing cache. Both ids must
+    resolve to bit base 24.
+    """
+
+    # Reporter's capture: bits 24, 25, 26 set -> slots 0/1/2 loaded, slot 3 empty.
+    BITS = "7000000"
+
+    def _units(self, ams_id):
+        return [
+            {
+                "id": ams_id,
+                "tray": [
+                    {
+                        "id": str(i),
+                        "state": 3,
+                        "tray_type": "PLA",
+                        "tray_color": "C12E1FFF",
+                        "tray_info_idx": "GFA00",
+                        "remain": 100,
+                    }
+                    for i in range(4)
+                ],
+            }
+        ]
+
+    def test_physical_id_16_uses_bit_base_24(self):
+        units = self._units(A2L_LITE_PHYSICAL_AMS_ID)
+        cleared = apply_tray_exist_bits(units, self.BITS)
+        trays = units[0]["tray"]
+        # Slots 0-2 are loaded and must survive untouched.
+        for slot in range(3):
+            assert trays[slot]["tray_type"] == "PLA", f"slot {slot} wrongly cleared"
+            assert trays[slot]["state"] == 3
+        # Only the genuinely empty slot 3 is cleared.
+        assert cleared == 1
+        assert trays[3]["state"] == 9
+        assert trays[3]["tray_type"] == ""
+
+    def test_normalised_id_6_matches_physical_id_16(self):
+        physical = self._units(A2L_LITE_PHYSICAL_AMS_ID)
+        normalised = self._units(A2L_LITE_NORMALIZED_AMS_ID)
+        apply_tray_exist_bits(physical, self.BITS)
+        apply_tray_exist_bits(normalised, self.BITS)
+        assert physical[0]["tray"] == normalised[0]["tray"]
+
+    def test_exists_annotation_matches_physical_slots(self):
+        units = self._units(A2L_LITE_PHYSICAL_AMS_ID)
+        apply_tray_exist_bits(units, self.BITS, annotate_exists=True)
+        assert [t["exists"] for t in units[0]["tray"]] == [True, True, True, False]
+
+    def test_regular_ams_unchanged(self):
+        # id 0 still reads bits 0-3 — the fold must not touch any other unit.
+        units = self._units(0)
+        apply_tray_exist_bits(units, "e")  # bits 1,2,3
+        trays = units[0]["tray"]
+        assert trays[0]["state"] == 9
+        assert [t["tray_type"] for t in trays] == ["", "PLA", "PLA", "PLA"]
+
+
 class TestOutboundTranslation:
     def test_set_filament_setting_uses_physical_16_local_slot(self):
         client = _wired(_client())

backend/tests/unit/test_archive_filtering.py

@@ -259,6 +259,9 @@ class TestScanForTimelapseWithRetries:
         mock_archive.timelapse_path = timelapse_path
         mock_archive.printer_id = 1
         mock_archive.filename = archive_filename
+        # No persisted print-start baseline (#2704) — these cases exercise the
+        # in-memory / fallback baseline paths.
+        mock_archive.timelapse_baseline = None
 
         mock_printer = MagicMock()
         mock_printer.id = 1
@@ -273,8 +276,13 @@ class TestScanForTimelapseWithRetries:
         mock_session = AsyncMock()
         mock_session.__aenter__ = AsyncMock(return_value=mock_session)
         mock_session.__aexit__ = AsyncMock()
+        # Serves both the printer lookup and the "already claimed by another
+        # archive" query the candidate filter runs (#2704).
         mock_session.execute = AsyncMock(
-            return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
+            return_value=MagicMock(
+                scalar_one_or_none=MagicMock(return_value=mock_printer),
+                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
+            )
         )
         return mock_session
 
@@ -311,9 +319,14 @@ class TestScanForTimelapseWithRetries:
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
             patch(f"{_FTP_MODULE}.download_file_bytes_async", new_callable=AsyncMock) as mock_download,
+            # The attach re-lists the file to confirm the printer has finished
+            # writing it (#2704); without this it opens a real FTP connection.
+            patch(f"{_FTP_MODULE}.remote_file_settled", new_callable=AsyncMock) as mock_settled,
+            patch(f"{_FTP_MODULE}.delete_archived_timelapse", new_callable=AsyncMock),
         ):
+            mock_settled.return_value = True
             mock_ws.send_archive_updated = AsyncMock()
-            mock_download.return_value = b"fake video data"
+            mock_download.return_value = b"x" * 2000  # must match the listed size (#2704)
 
             from backend.app.main import _scan_for_timelapse_with_retries
 
@@ -351,9 +364,14 @@ class TestScanForTimelapseWithRetries:
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
             patch(f"{_FTP_MODULE}.download_file_bytes_async", new_callable=AsyncMock) as mock_download,
+            # The attach re-lists the file to confirm the printer has finished
+            # writing it (#2704); without this it opens a real FTP connection.
+            patch(f"{_FTP_MODULE}.remote_file_settled", new_callable=AsyncMock) as mock_settled,
+            patch(f"{_FTP_MODULE}.delete_archived_timelapse", new_callable=AsyncMock),
         ):
+            mock_settled.return_value = True
             mock_ws.send_archive_updated = AsyncMock()
-            mock_download.return_value = b"fake video data"
+            mock_download.return_value = b"x" * 2000  # must match the listed size (#2704)
 
             from backend.app.main import _scan_for_timelapse_with_retries
 
@@ -363,8 +381,14 @@ class TestScanForTimelapseWithRetries:
         mock_service.attach_timelapse.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_name_match_fallback(self):
-        """When no new file appears, should fall back to name matching."""
+    async def test_no_name_match_rescue(self):
+        """The name-match fallback was removed (#2704).
+
+        It looked for the print name inside the video filename, but Bambu
+        firmware only ever writes "video_<timestamp>" — across 247 support
+        bundles it ran 159 times and matched zero times. A file already present
+        at baseline belongs to an earlier print, and guessing otherwise from its
+        name attaches the wrong video."""
         mock_archive, mock_printer = self._make_mocks()
 
         baseline_files = [
@@ -392,18 +416,22 @@ class TestScanForTimelapseWithRetries:
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
             patch(f"{_FTP_MODULE}.download_file_bytes_async", new_callable=AsyncMock) as mock_download,
+            # The attach re-lists the file to confirm the printer has finished
+            # writing it (#2704); without this it opens a real FTP connection.
+            patch(f"{_FTP_MODULE}.remote_file_settled", new_callable=AsyncMock) as mock_settled,
+            patch(f"{_FTP_MODULE}.delete_archived_timelapse", new_callable=AsyncMock),
         ):
+            mock_settled.return_value = True
             mock_ws.send_archive_updated = AsyncMock()
-            mock_download.return_value = b"fake video data"
+            mock_download.return_value = b"x" * 2000  # must match the listed size (#2704)
 
             from backend.app.main import _scan_for_timelapse_with_retries
 
             await _scan_for_timelapse_with_retries(1)
 
-        # Name-match fallback: "benchy" is in "benchy_20240101.mp4"
-        mock_service.attach_timelapse.assert_called_once()
-        attached_filename = mock_service.attach_timelapse.call_args[0][2]
-        assert attached_filename == "benchy_20240101.mp4"
+        # "benchy" is in "benchy_20240101.mp4", but that file was there before
+        # the print started, so it is not this print's video.
+        mock_service.attach_timelapse.assert_not_called()
 
     @pytest.mark.asyncio
     async def test_stops_when_archive_already_has_timelapse(self):
@@ -455,8 +483,14 @@ class TestScanForTimelapseWithRetries:
         mock_sleep.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_retries_four_times(self):
-        """Should retry with delays [5, 10, 20, 30]."""
+    async def test_polls_until_the_budget_runs_out(self):
+        """The fixed [5, 10, 20, 30] ladder gave up after ~65s (#2704).
+
+        Support bundles showed the attempt that found the video was #1 272
+        times and then 17 / 13 / 13 — flat against the cutoff, i.e. files were
+        still arriving when the old budget expired. It is now a poll: one short
+        first look, then a steady interval until the wall-clock budget or the
+        derived round cap is reached, whichever comes first."""
         mock_archive, mock_printer = self._make_mocks(archive_filename="test.gcode.3mf")
 
         # Never find any files
@@ -480,10 +514,18 @@ class TestScanForTimelapseWithRetries:
 
             await _scan_for_timelapse_with_retries(1)
 
-        # Should have slept 4 times with delays [5, 10, 20, 30]
-        assert mock_sleep.call_count == 4
+        from backend.app.main import (
+            _TIMELAPSE_SCAN_FIRST_DELAY_SECONDS,
+            _TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS,
+            _timelapse_scan_max_attempts,
+        )
+
         sleep_args = [call.args[0] for call in mock_sleep.call_args_list]
-        assert sleep_args == [5, 10, 20, 30]
+        assert len(sleep_args) == _timelapse_scan_max_attempts()
+        assert sleep_args[0] == _TIMELAPSE_SCAN_FIRST_DELAY_SECONDS
+        assert set(sleep_args[1:]) == {_TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS}
+        # Substantially longer than the ladder it replaced.
+        assert sum(sleep_args) > 300
 
 
 class TestListTimelapseVideosAvi:
@@ -546,6 +588,7 @@ class TestListTimelapseVideosAvi:
         mock_archive.timelapse_path = None
         mock_archive.printer_id = 1
         mock_archive.filename = "benchy.gcode.3mf"
+        mock_archive.timelapse_baseline = None
 
         mock_printer = MagicMock()
         mock_printer.id = 1
@@ -580,7 +623,10 @@ class TestListTimelapseVideosAvi:
         mock_session.__aenter__ = AsyncMock(return_value=mock_session)
         mock_session.__aexit__ = AsyncMock()
         mock_session.execute = AsyncMock(
-            return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
+            return_value=MagicMock(
+                scalar_one_or_none=MagicMock(return_value=mock_printer),
+                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
+            )
         )
 
         with (
@@ -590,9 +636,14 @@ class TestListTimelapseVideosAvi:
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
             patch(f"{_FTP_MODULE}.download_file_bytes_async", new_callable=AsyncMock) as mock_download,
+            # The attach re-lists the file to confirm the printer has finished
+            # writing it (#2704); without this it opens a real FTP connection.
+            patch(f"{_FTP_MODULE}.remote_file_settled", new_callable=AsyncMock) as mock_settled,
+            patch(f"{_FTP_MODULE}.delete_archived_timelapse", new_callable=AsyncMock),
         ):
+            mock_settled.return_value = True
             mock_ws.send_archive_updated = AsyncMock()
-            mock_download.return_value = b"fake avi data"
+            mock_download.return_value = b"x" * 50000  # must match the listed size (#2704)
 
             from backend.app.main import _scan_for_timelapse_with_retries
 

backend/tests/unit/test_finish_photo_from_timelapse.py

@@ -62,11 +62,14 @@ def patched_session(fake_archive, monkeypatch):
 async def test_returns_none_when_timelapse_never_lands(tmp_path: Path, patched_session):
     """Print finished without a timelapse — bail after timeout so the caller
     falls back to the live-camera grab."""
-    result = await _capture_finish_photo_from_timelapse(
+    result, pending = await _capture_finish_photo_from_timelapse(
         archive_id=42,
         archive_dir=tmp_path,
     )
     assert result is None
+    # Ran out of time rather than concluded: the video may still be on its way,
+    # which is what tells the caller to schedule a background upgrade (#2704).
+    assert pending is True
 
 
 async def test_extracts_frame_when_timelapse_lands(tmp_path: Path, patched_session, monkeypatch):
@@ -96,7 +99,7 @@ async def test_extracts_frame_when_timelapse_lands(tmp_path: Path, patched_sessi
         "backend.app.services.camera.extract_video_last_frame",
         new=fake_extract,
     ):
-        result = await _capture_finish_photo_from_timelapse(
+        result, pending = await _capture_finish_photo_from_timelapse(
             archive_id=42,
             archive_dir=tmp_path / "archive_dir",
         )
@@ -105,6 +108,7 @@ async def test_extracts_frame_when_timelapse_lands(tmp_path: Path, patched_sessi
     assert result.startswith("finish_")
     assert result.endswith(".jpg")
     assert (tmp_path / "archive_dir" / "photos" / result).exists()
+    assert pending is False
 
 
 async def test_returns_none_when_extraction_fails(tmp_path: Path, patched_session, monkeypatch):
@@ -125,12 +129,15 @@ async def test_returns_none_when_extraction_fails(tmp_path: Path, patched_sessio
         "backend.app.services.camera.extract_video_last_frame",
         new=fake_extract_fails,
     ):
-        result = await _capture_finish_photo_from_timelapse(
+        result, pending = await _capture_finish_photo_from_timelapse(
             archive_id=42,
             archive_dir=tmp_path / "archive_dir",
         )
 
     assert result is None
+    # The video arrived and ffmpeg refused it — waiting longer cannot help, so
+    # this must NOT ask for a background retry.
+    assert pending is False
 
 
 async def test_polls_until_file_appears(tmp_path: Path, patched_session, monkeypatch):
@@ -163,7 +170,7 @@ async def test_polls_until_file_appears(tmp_path: Path, patched_session, monkeyp
             "backend.app.services.camera.extract_video_last_frame",
             new=fake_extract,
         ):
-            result = await _capture_finish_photo_from_timelapse(
+            result, pending = await _capture_finish_photo_from_timelapse(
                 archive_id=42,
                 archive_dir=tmp_path / "archive_dir",
             )

backend/tests/unit/test_log_credential_redaction.py

@@ -0,0 +1,112 @@
+"""Credentials must never reach bambuddy.log.
+
+Subprocesses echo their input URL back at us: ffmpeg prints the RTSP input in
+its ``Input #0`` line, so logging its stderr verbatim published the printer
+access code (or an external camera's password) into the log file — which users
+routinely attach to public GitHub issues.
+
+These cover the shared helper plus the two funnels that carry subprocess output
+into the log.
+"""
+
+import asyncio
+
+from backend.app.api.routes.camera import _read_ffmpeg_stderr, _summarize_ffmpeg_stderr
+from backend.app.core.logging_filters import redact_url_credentials
+from backend.app.services.log_reader import sanitize_log_content
+
+# What ffmpeg actually prints for the camera's local TLS-proxy input. The
+# access code sits in the userinfo of the URL it quotes back.
+FFMPEG_INPUT_LINE = "Input #0, rtsp, from 'rtsp://bblp:38A4KQ2P@127.0.0.1:48521/streaming/live/1':"
+
+
+class TestRedactUrlCredentials:
+    def test_masks_the_printer_access_code(self):
+        result = redact_url_credentials(FFMPEG_INPUT_LINE)
+        assert "38A4KQ2P" not in result
+        assert result == "Input #0, rtsp, from 'rtsp://bblp:[REDACTED]@127.0.0.1:48521/streaming/live/1':"
+
+    def test_keeps_everything_that_is_not_the_secret(self):
+        """Host, port, path and username stay — the line has to remain diagnosable."""
+        result = redact_url_credentials("rtsp://admin:hunter2@192.168.1.50:554/stream1")
+        assert result == "rtsp://admin:[REDACTED]@192.168.1.50:554/stream1"
+
+    def test_masks_every_scheme_not_just_the_ones_we_use_today(self):
+        for url, expected in (
+            ("http://user:pw@cam.local/snapshot", "http://user:[REDACTED]@cam.local/snapshot"),
+            ("https://user:pw@cam.local/snapshot", "https://user:[REDACTED]@cam.local/snapshot"),
+            ("rtsps://bblp:code@printer:322/streaming/live/1", "rtsps://bblp:[REDACTED]@printer:322/streaming/live/1"),
+            ("ftp://bblp:code@printer:990/", "ftp://bblp:[REDACTED]@printer:990/"),
+        ):
+            assert redact_url_credentials(url) == expected
+
+    def test_masks_a_password_containing_an_at_sign(self):
+        """The userinfo ends at the LAST @ before the path — no tail may survive."""
+        result = redact_url_credentials("rtsp://admin:p@ssw0rd@192.168.1.50/stream")
+        assert result == "rtsp://admin:[REDACTED]@192.168.1.50/stream"
+        assert "ssw0rd" not in result
+
+    def test_masks_several_urls_in_one_blob(self):
+        text = "first rtsp://bblp:AAAAAAAA@10.0.0.1/live then rtsp://bblp:BBBBBBBB@10.0.0.2/live"
+        result = redact_url_credentials(text)
+        assert "AAAAAAAA" not in result
+        assert "BBBBBBBB" not in result
+        assert result.count("[REDACTED]") == 2
+
+    def test_never_runs_past_the_authority_into_the_path(self):
+        """A later @ in the path must not drag the host into the mask."""
+        result = redact_url_credentials("rtsp://bblp:code@10.0.0.1/live/user@example")
+        assert result == "rtsp://bblp:[REDACTED]@10.0.0.1/live/user@example"
+
+    def test_leaves_credential_free_text_alone(self):
+        for untouched in (
+            "Connection refused",
+            "rtsp://10.0.0.1:554/stream1",
+            "mailto and user@example.com in prose",
+            "Starting USB camera stream from /dev/video0 at 10 fps",
+        ):
+            assert redact_url_credentials(untouched) == untouched
+
+    def test_tolerates_empty_and_none(self):
+        assert redact_url_credentials("") == ""
+        assert redact_url_credentials(None) is None
+
+
+class TestFfmpegStderrFunnel:
+    """`_summarize_ffmpeg_stderr` is the one funnel every stderr log in the
+    camera route passes through, so redaction lands there."""
+
+    def test_summary_strips_the_access_code(self):
+        stderr = f"{FFMPEG_INPUT_LINE}\n[rtsp @ 0x5] Could not find codec parameters\n"
+        result = _summarize_ffmpeg_stderr(stderr)
+        assert "38A4KQ2P" not in result
+        assert "[REDACTED]" in result
+        # The actionable error is untouched.
+        assert "Could not find codec parameters" in result
+
+    def test_incremental_reader_strips_the_access_code(self):
+        async def run():
+            reader = asyncio.StreamReader()
+            reader.feed_data(f"{FFMPEG_INPUT_LINE}\nError opening input: Connection refused\n".encode())
+            reader.feed_eof()
+
+            class _FakeProcess:
+                stderr = reader
+
+            return await _read_ffmpeg_stderr(_FakeProcess())
+
+        result = asyncio.run(run())
+        assert result is not None
+        assert "38A4KQ2P" not in result
+        assert "Connection refused" in result
+
+
+class TestSupportBundleSanitizerUnchanged:
+    """The bundle sanitizer shares the pattern but keeps its own, stricter
+    replacement — it drops the username too. Guard against drift."""
+
+    def test_bundle_still_drops_the_whole_userinfo(self):
+        result = sanitize_log_content("rtsp://bblp:38A4KQ2P@10.0.0.1/live")
+        assert "38A4KQ2P" not in result
+        assert "bblp" not in result
+        assert "[CREDENTIALS]@" in result

backend/tests/unit/test_timelapse_scan_2704.py

@@ -0,0 +1,767 @@
+"""Timelapse scan reliability (#2704).
+
+A Bambu printer in LAN-only mode never reaches Bambu's NTP server, so the clock
+behind both the timelapse filename and the FTP mtime drifts freely — the P1S in
+the report was six and a half days out. That is why the automatic scan works by
+diffing the printer's ``/timelapse`` listing against a snapshot taken when the
+print started, and why nothing in that path may fall back to comparing times.
+
+These tests pin the parts that make the diff dependable:
+
+* the candidate is chosen by exclusion, never by ordering (ordering could only
+  be done on the printer's clock);
+* a download that comes up short never attaches and never triggers a delete —
+  deleting the printer's copy is only safe because the transfer was verified;
+* the baseline persisted at print start is what the manual Scan button uses,
+  instead of the clock-based strategies that cannot work on a drifted printer.
+"""
+
+import logging
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+logger = logging.getLogger(__name__)
+
+
+def _printer():
+    p = MagicMock()
+    p.id = 1
+    p.name = "TestP1S"
+    p.ip_address = "192.168.1.100"
+    p.access_code = "12345678"
+    p.model = "P1S"
+    return p
+
+
+def _video(name: str, size: int = 1000):
+    return {"name": name, "is_directory": False, "path": f"/timelapse/{name}", "size": size}
+
+
+def _session(archive=None):
+    session = AsyncMock()
+    session.__aenter__ = AsyncMock(return_value=session)
+    session.__aexit__ = AsyncMock()
+    if archive is not None:
+        session.get = AsyncMock(return_value=archive)
+    return session
+
+
+class TestCandidateSelection:
+    """Which of the printer's videos belongs to this print."""
+
+    @pytest.fixture
+    def attach(self):
+        from backend.app.main import _attach_first_unclaimed_timelapse
+
+        return _attach_first_unclaimed_timelapse
+
+    @pytest.mark.asyncio
+    async def test_nothing_new_since_baseline_is_not_an_attach(self, attach):
+        result = await attach(
+            42,
+            _printer(),
+            [_video("video_2026-07-21_09-17-37.avi")],
+            {"video_2026-07-21_09-17-37.avi"},
+            set(),
+            1,
+            logger,
+        )
+        assert result is False
+
+    @pytest.mark.asyncio
+    async def test_attaches_the_one_new_file_and_deletes_it_from_the_printer(self, attach):
+        download = AsyncMock(return_value=b"x" * 1000)
+        delete = AsyncMock(return_value=True)
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", download),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", AsyncMock(return_value=True)),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", delete),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+            patch("backend.app.main.ws_manager", MagicMock(send_archive_updated=AsyncMock())),
+        ):
+            result = await attach(
+                42,
+                _printer(),
+                [_video("old.avi"), _video("video_2026-07-22_06-18-39.avi")],
+                {"old.avi"},
+                set(),
+                1,
+                logger,
+            )
+
+        assert result is True
+        service.attach_timelapse.assert_awaited_once()
+        assert service.attach_timelapse.await_args.args[2] == "video_2026-07-22_06-18-39.avi"
+        delete.assert_awaited_once()
+        assert delete.await_args.args[2] == "/timelapse/video_2026-07-22_06-18-39.avi"
+
+    @pytest.mark.asyncio
+    async def test_skips_a_previous_prints_late_landing_video(self, attach):
+        """Two files are new since the baseline because the previous print's
+        video only landed after this print started. It is already attached to
+        another archive, so it is excluded by name — no timestamps involved."""
+        download = AsyncMock(return_value=b"y" * 1000)
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", download),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", AsyncMock(return_value=True)),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", AsyncMock()),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+            patch("backend.app.main.ws_manager", MagicMock(send_archive_updated=AsyncMock())),
+        ):
+            result = await attach(
+                42,
+                _printer(),
+                # Listing order puts the previous print's video first, so a
+                # naive "take the first new one" would grab the wrong video.
+                [_video("previous_print.avi"), _video("this_print.avi")],
+                set(),
+                {"previous_print"},
+                1,
+                logger,
+            )
+
+        assert result is True
+        assert service.attach_timelapse.await_args.args[2] == "this_print.avi"
+
+    @pytest.mark.asyncio
+    async def test_claimed_match_survives_the_mp4_conversion(self, attach):
+        """Attached AVIs are converted to MP4 afterwards, which keeps the stem
+        but changes the extension — so exclusion has to compare stems."""
+        result = await attach(
+            42,
+            _printer(),
+            [_video("video_2026-07-22_06-18-39.avi")],
+            set(),
+            {"video_2026-07-22_06-18-39"},  # stored as .mp4 on the archive
+            1,
+            logger,
+        )
+        assert result is False
+
+    @pytest.mark.asyncio
+    async def test_all_new_files_claimed_keeps_polling(self, attach):
+        result = await attach(42, _printer(), [_video("a.avi"), _video("b.avi")], set(), {"a", "b"}, 1, logger)
+        assert result is False
+
+
+class TestDownloadVerificationGatesTheDelete:
+    """The printer's copy is the only other copy — it goes only after the
+    transfer is verified against the size the listing reported."""
+
+    @pytest.fixture
+    def attach(self):
+        from backend.app.main import _attach_first_unclaimed_timelapse
+
+        return _attach_first_unclaimed_timelapse
+
+    @pytest.mark.asyncio
+    async def test_passes_the_listed_size_to_the_downloader(self, attach):
+        download = AsyncMock(return_value=b"z" * 4096)
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", download),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", AsyncMock(return_value=True)),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", AsyncMock()),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+            patch("backend.app.main.ws_manager", MagicMock(send_archive_updated=AsyncMock())),
+        ):
+            await attach(42, _printer(), [_video("new.avi", size=4096)], set(), set(), 1, logger)
+
+        assert download.await_args.kwargs["expected_size"] == 4096
+
+    @pytest.mark.asyncio
+    async def test_short_download_does_not_attach_or_delete(self, attach):
+        """download_file_bytes_async returns None on a size mismatch. The
+        printer must keep its copy so the next poll round can retry."""
+        delete = AsyncMock()
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", AsyncMock(return_value=None)),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", AsyncMock(return_value=True)),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", delete),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+        ):
+            result = await attach(42, _printer(), [_video("new.avi")], set(), set(), 1, logger)
+
+        assert result is False
+        service.attach_timelapse.assert_not_awaited()
+        delete.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_failed_attach_does_not_delete(self, attach):
+        delete = AsyncMock()
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=False)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", AsyncMock(return_value=b"x" * 1000)),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", AsyncMock(return_value=True)),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", delete),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+        ):
+            result = await attach(42, _printer(), [_video("new.avi")], set(), set(), 1, logger)
+
+        assert result is False
+        delete.assert_not_awaited()
+
+
+class TestFtpDownloadSizeCheck:
+    """`download_file` is where a truncated FTPS transfer used to pass for a
+    complete one — a partial buffer is non-empty, so every caller downstream
+    treated it as a good file."""
+
+    def _client(self, payload: bytes):
+        from backend.app.services.bambu_ftp import BambuFTPClient
+
+        client = BambuFTPClient("192.168.1.100", "12345678")
+        ftp = MagicMock()
+        ftp.retrbinary = MagicMock(side_effect=lambda cmd, cb: cb(payload))
+        client._ftp = ftp
+        return client
+
+    def test_exact_size_passes(self):
+        assert self._client(b"a" * 500).download_file("/timelapse/v.avi", expected_size=500) == b"a" * 500
+
+    def test_short_read_is_a_failure(self):
+        assert self._client(b"a" * 499).download_file("/timelapse/v.avi", expected_size=500) is None
+
+    def test_long_read_is_a_failure(self):
+        """Not expected in practice, but a mismatch either way means we don't
+        know what we have, and we're about to delete the original."""
+        assert self._client(b"a" * 501).download_file("/timelapse/v.avi", expected_size=500) is None
+
+    def test_zero_bytes_is_a_failure_even_without_an_expected_size(self):
+        assert self._client(b"").download_file("/cache/whatever.3mf") is None
+
+    def test_unverified_download_still_works_for_callers_that_do_not_pass_a_size(self):
+        assert self._client(b"abc").download_file("/cache/whatever.3mf") == b"abc"
+
+
+class TestDeleteIsBestEffort:
+    """A printer that refuses the delete must not break the flow — the video
+    is already in the archive, and the diff excludes it by name from then on."""
+
+    @pytest.mark.asyncio
+    async def test_reports_success_on_delete(self):
+        from backend.app.services.bambu_ftp import DeleteResult, delete_archived_timelapse
+
+        with patch(
+            "backend.app.services.bambu_ftp.delete_file_async", AsyncMock(return_value=DeleteResult.DELETED)
+        ) as d:
+            assert await delete_archived_timelapse("1.2.3.4", "code", "/timelapse/v.avi", verified=True) is True
+        assert d.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_not_found_is_success_and_is_not_retried(self):
+        """550 means the printer already cleaned up; waiting cannot change it."""
+        from backend.app.services.bambu_ftp import DeleteResult, delete_archived_timelapse
+
+        with patch(
+            "backend.app.services.bambu_ftp.delete_file_async", AsyncMock(return_value=DeleteResult.NOT_FOUND)
+        ) as d:
+            assert await delete_archived_timelapse("1.2.3.4", "code", "/timelapse/v.avi", verified=True) is True
+        assert d.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_failure_retries_then_gives_up_without_raising(self):
+        from backend.app.services.bambu_ftp import DeleteResult, delete_archived_timelapse
+
+        with (
+            patch("backend.app.services.bambu_ftp.delete_file_async", AsyncMock(return_value=DeleteResult.FAILED)),
+            patch("backend.app.services.bambu_ftp.asyncio.sleep", AsyncMock()),
+        ):
+            assert await delete_archived_timelapse("1.2.3.4", "code", "/timelapse/v.avi", verified=True) is False
+
+    @pytest.mark.asyncio
+    async def test_raising_transport_does_not_propagate(self):
+        from backend.app.services.bambu_ftp import delete_archived_timelapse
+
+        with (
+            patch("backend.app.services.bambu_ftp.delete_file_async", AsyncMock(side_effect=OSError("boom"))),
+            patch("backend.app.services.bambu_ftp.asyncio.sleep", AsyncMock()),
+        ):
+            assert await delete_archived_timelapse("1.2.3.4", "code", "/timelapse/v.avi", verified=True) is False
+
+
+class TestBaselineIsPersisted:
+    """The baseline has to outlive the process: a restart mid-print used to
+    lose it, and the manual scan never had access to it at all."""
+
+    @pytest.mark.asyncio
+    async def test_written_to_the_archive_row_at_print_start(self):
+        from backend.app.main import _capture_timelapse_baseline_at_start
+
+        archive = MagicMock()
+        archive.timelapse_baseline = None
+        session = _session(archive)
+
+        with (
+            patch("backend.app.main.async_session", return_value=session),
+            patch(
+                "backend.app.main._list_timelapse_videos",
+                new=AsyncMock(return_value=([_video("a.avi"), _video("b.avi")], "/timelapse")),
+            ),
+        ):
+            await _capture_timelapse_baseline_at_start(_printer(), 1, logger, archive_id=7)
+
+        assert archive.timelapse_baseline == ["a.avi", "b.avi"]
+        session.commit.assert_awaited()
+
+    @pytest.mark.asyncio
+    async def test_no_archive_id_keeps_it_in_memory_only(self):
+        from backend.app.main import _capture_timelapse_baseline_at_start, _timelapse_baselines
+
+        _timelapse_baselines.pop(1, None)
+        session = _session(MagicMock())
+
+        with (
+            patch("backend.app.main.async_session", return_value=session),
+            patch(
+                "backend.app.main._list_timelapse_videos",
+                new=AsyncMock(return_value=([_video("a.avi")], "/timelapse")),
+            ),
+        ):
+            await _capture_timelapse_baseline_at_start(_printer(), 1, logger)
+
+        assert _timelapse_baselines[1] == {"a.avi"}
+        session.commit.assert_not_awaited()
+        _timelapse_baselines.pop(1, None)
+
+    @pytest.mark.asyncio
+    async def test_listing_failure_stores_null_not_an_empty_baseline(self):
+        """An empty list would make every video on the printer look new; NULL
+        correctly means "no baseline" and falls back to a fresh snapshot."""
+        from backend.app.main import _capture_timelapse_baseline_at_start
+
+        archive = MagicMock()
+        session = _session(archive)
+
+        with (
+            patch("backend.app.main.async_session", return_value=session),
+            patch("backend.app.main._list_timelapse_videos", new=AsyncMock(side_effect=OSError("ftp down"))),
+        ):
+            await _capture_timelapse_baseline_at_start(_printer(), 1, logger, archive_id=7)
+
+        assert archive.timelapse_baseline is None
+
+
+class TestManualScanUsesTheBaseline:
+    """The reporter's second symptom: pressing "Scan for Timelapse" found
+    nothing. Every strategy the endpoint had was clock-based, and their
+    printer's clock was days out, so it could not match on any of them."""
+
+    def _archive(self, baseline):
+        from datetime import datetime, timezone
+
+        a = MagicMock()
+        a.id = 64
+        a.printer_id = 1
+        a.filename = "mops.3mf"
+        a.timelapse_path = None
+        a.timelapse_baseline = baseline
+        a.started_at = datetime(2026, 7, 28, 20, 30, tzinfo=timezone.utc)
+        a.completed_at = datetime(2026, 7, 28, 21, 19, tzinfo=timezone.utc)
+        a.created_at = a.completed_at
+        return a
+
+    async def _scan(self, archive, listing, download=None, delete=None):
+        from backend.app.api.routes import archives as archives_mod
+
+        service = MagicMock()
+        service.get_archive = AsyncMock(return_value=archive)
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        session = AsyncMock()
+        session.__aenter__ = AsyncMock(return_value=session)
+        session.__aexit__ = AsyncMock()
+        session.execute = AsyncMock(
+            return_value=MagicMock(
+                scalar_one_or_none=MagicMock(return_value=_printer()),
+                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
+            )
+        )
+
+        with (
+            patch("backend.app.core.database.async_session", return_value=session),
+            patch("backend.app.api.routes.archives.ArchiveService", return_value=service),
+            patch("backend.app.services.bambu_ftp.list_files_async", AsyncMock(return_value=listing)),
+            patch(
+                "backend.app.services.bambu_ftp.get_ftp_retry_settings",
+                AsyncMock(return_value=(False, 3, 2, 30)),
+            ),
+            patch(
+                "backend.app.services.bambu_ftp.download_file_bytes_async",
+                download or AsyncMock(return_value=b"x" * 1000),
+            ),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", delete or AsyncMock()),
+        ):
+            return await archives_mod.scan_timelapse(archive.id, None)
+
+    @pytest.mark.asyncio
+    async def test_attaches_the_single_unclaimed_new_file(self):
+        """The printer's clock is six days out here — exactly the reporter's
+        case. Nothing in this path looks at a timestamp."""
+        archive = self._archive(["video_2026-07-21_22-49-47.avi"])
+        listing = [
+            _video("video_2026-07-21_22-49-47.avi"),
+            _video("video_2026-07-22_06-18-39.avi"),
+        ]
+
+        result = await self._scan(archive, listing)
+
+        assert result["status"] == "attached"
+        assert result["filename"] == "video_2026-07-22_06-18-39.avi"
+
+    @pytest.mark.asyncio
+    async def test_deletes_from_the_printer_after_attaching(self):
+        delete = AsyncMock()
+        archive = self._archive(["old.avi"])
+
+        await self._scan(archive, [_video("old.avi"), _video("new.avi")], delete=delete)
+
+        delete.assert_awaited_once()
+        assert delete.await_args.args[2] == "/timelapse/new.avi"
+
+    @pytest.mark.asyncio
+    async def test_baseline_showing_nothing_new_does_not_guess(self):
+        """With a baseline saying no new video exists, the clock strategies
+        must not run — otherwise a coincidental timestamp match attaches
+        someone else's video and calls it this print's."""
+        archive = self._archive(["video_2026-07-28_20-30-00.avi"])
+        # This file's embedded time is minutes from started_at, so the old
+        # timestamp strategy would have matched it confidently.
+        listing = [_video("video_2026-07-28_20-30-00.avi")]
+
+        result = await self._scan(archive, listing)
+
+        assert result["status"] == "not_found"
+
+    @pytest.mark.asyncio
+    async def test_ambiguous_baseline_offers_only_the_plausible_files(self):
+        archive = self._archive(["old.avi"])
+        listing = [_video("old.avi"), _video("candidate_a.avi"), _video("candidate_b.avi")]
+
+        result = await self._scan(archive, listing)
+
+        assert result["status"] == "not_found"
+        assert {f["name"] for f in result["available_files"]} == {"candidate_a.avi", "candidate_b.avi"}
+
+    @pytest.mark.asyncio
+    async def test_archives_without_a_baseline_keep_the_old_strategies(self):
+        """Rows predating the persisted baseline still get the best guess the
+        endpoint can make, rather than nothing at all."""
+        archive = self._archive(None)
+        listing = [_video("mops_something.avi")]  # matches by print name
+
+        result = await self._scan(archive, listing)
+
+        assert result["status"] == "attached"
+        assert result["filename"] == "mops_something.avi"
+
+
+class TestPollBounds:
+    """The poll is bounded twice on purpose."""
+
+    def test_round_cap_tracks_the_wall_clock_budget(self):
+        from backend.app.main import (
+            _TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS,
+            _TIMELAPSE_SCAN_TIMEOUT_SECONDS,
+            _timelapse_scan_max_attempts,
+        )
+
+        assert (
+            _timelapse_scan_max_attempts()
+            == int(_TIMELAPSE_SCAN_TIMEOUT_SECONDS // _TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS) + 1
+        )
+
+    def test_zero_interval_does_not_divide_by_zero(self, monkeypatch):
+        """The deadline alone can't bound the loop once sleeps are shortened to
+        nothing, which is exactly what a test or a future tweak would do."""
+        import backend.app.main as main_mod
+
+        monkeypatch.setattr(main_mod, "_TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS", 0)
+        assert main_mod._timelapse_scan_max_attempts() > 1
+
+    def test_budget_is_much_longer_than_the_ladder_it_replaced(self):
+        """The old [5, 10, 20, 30] ladder gave up after ~65 seconds, while the
+        support bundles showed videos still arriving at the cutoff."""
+        from backend.app.main import _TIMELAPSE_SCAN_TIMEOUT_SECONDS
+
+        assert _TIMELAPSE_SCAN_TIMEOUT_SECONDS >= 300
+
+
+class TestFinishPhotoUpgrade:
+    """The print-complete notification waits ~60s for the timelapse, because
+    holding it for minutes is worse than sending a live grab. On a P1S the
+    video routinely lands later than that (p90 167s, worst observed 546s), so
+    the archive kept the live grab — taken after the end G-code dropped the
+    bed, which is the worse of the two photos. The upgrade runs afterwards."""
+
+    @pytest.mark.asyncio
+    async def test_puts_the_timelapse_frame_first_and_keeps_the_live_grab(self):
+        """First, because the gallery opens at index 0. Kept, because the
+        notification that already went out links to that exact file."""
+        from backend.app.main import _upgrade_finish_photo_from_timelapse
+
+        archive = MagicMock()
+        archive.photos = ["finish_live_grab.jpg"]
+        session = _session(archive)
+
+        with (
+            patch(
+                "backend.app.main._capture_finish_photo_from_timelapse",
+                AsyncMock(return_value=("finish_from_timelapse.jpg", False)),
+            ),
+            patch("backend.app.main.async_session", return_value=session),
+            patch("backend.app.main.ws_manager", MagicMock(send_archive_updated=AsyncMock())) as ws,
+        ):
+            await _upgrade_finish_photo_from_timelapse(7, MagicMock())
+
+        assert archive.photos == ["finish_from_timelapse.jpg", "finish_live_grab.jpg"]
+        session.commit.assert_awaited()
+        ws.send_archive_updated.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_waits_far_longer_than_the_notification_can(self):
+        from backend.app.main import (
+            _FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS,
+            _FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS,
+            _upgrade_finish_photo_from_timelapse,
+        )
+
+        capture = AsyncMock(return_value=(None, True))
+        with patch("backend.app.main._capture_finish_photo_from_timelapse", capture):
+            await _upgrade_finish_photo_from_timelapse(7, MagicMock())
+
+        assert capture.await_args.kwargs["timeout"] == _FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS
+        assert _FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS > _FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS
+        # Covers the 546s worst case seen in the support bundles.
+        assert _FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS >= 600
+
+    @pytest.mark.asyncio
+    async def test_video_never_arrives_leaves_the_archive_alone(self):
+        from backend.app.main import _upgrade_finish_photo_from_timelapse
+
+        session = _session(MagicMock())
+        with (
+            patch("backend.app.main._capture_finish_photo_from_timelapse", AsyncMock(return_value=(None, True))),
+            patch("backend.app.main.async_session", return_value=session),
+        ):
+            await _upgrade_finish_photo_from_timelapse(7, MagicMock())
+
+        session.commit.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_is_idempotent(self):
+        """A second run must not list the same photo twice."""
+        from backend.app.main import _upgrade_finish_photo_from_timelapse
+
+        archive = MagicMock()
+        archive.photos = ["finish_from_timelapse.jpg", "finish_live_grab.jpg"]
+        session = _session(archive)
+
+        with (
+            patch(
+                "backend.app.main._capture_finish_photo_from_timelapse",
+                AsyncMock(return_value=("finish_from_timelapse.jpg", False)),
+            ),
+            patch("backend.app.main.async_session", return_value=session),
+        ):
+            await _upgrade_finish_photo_from_timelapse(7, MagicMock())
+
+        assert archive.photos == ["finish_from_timelapse.jpg", "finish_live_grab.jpg"]
+        session.commit.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_missing_archive_does_not_raise(self):
+        from backend.app.main import _upgrade_finish_photo_from_timelapse
+
+        session = AsyncMock()
+        session.__aenter__ = AsyncMock(return_value=session)
+        session.__aexit__ = AsyncMock()
+        session.get = AsyncMock(return_value=None)
+
+        with (
+            patch("backend.app.main._capture_finish_photo_from_timelapse", AsyncMock(return_value=("f.jpg", False))),
+            patch("backend.app.main.async_session", return_value=session),
+        ):
+            await _upgrade_finish_photo_from_timelapse(7, MagicMock())
+
+        session.commit.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_refuses_to_delete_an_unverified_download(self):
+        """The safety rule lives with the destructive call, not at the call
+        sites — an unverified transfer may be a truncated file, and deleting
+        the source would destroy the only complete copy."""
+        from backend.app.services.bambu_ftp import delete_archived_timelapse
+
+        with patch("backend.app.services.bambu_ftp.delete_file_async", AsyncMock()) as d:
+            assert await delete_archived_timelapse("1.2.3.4", "code", "/timelapse/v.avi", verified=False) is False
+        d.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_verified_is_required_not_defaulted(self):
+        """A future call site must not be able to silently skip the check."""
+        import inspect
+
+        from backend.app.services.bambu_ftp import delete_archived_timelapse
+
+        param = inspect.signature(delete_archived_timelapse).parameters["verified"]
+        assert param.default is inspect.Parameter.empty
+        assert param.kind is inspect.Parameter.KEYWORD_ONLY
+
+
+class TestStaleBaselineCannotSurvive:
+    """A reprint reuses the archive row, so a baseline left over from the
+    previous run would have the scan diff this print against the printer's
+    state before the *last* one — and unlike NULL, a stale list reads as
+    authoritative and suppresses the fresh-snapshot fallback."""
+
+    @pytest.mark.asyncio
+    async def test_failed_capture_clears_rather_than_leaves_the_old_value(self):
+        from backend.app.main import _capture_timelapse_baseline_at_start
+
+        archive = MagicMock()
+        archive.timelapse_baseline = ["from_the_previous_run.avi"]
+        session = _session(archive)
+
+        with (
+            patch("backend.app.main.async_session", return_value=session),
+            patch("backend.app.main._list_timelapse_videos", new=AsyncMock(side_effect=OSError("ftp down"))),
+        ):
+            await _capture_timelapse_baseline_at_start(_printer(), 1, logger, archive_id=7)
+
+        assert archive.timelapse_baseline is None
+        session.commit.assert_awaited()
+
+    @pytest.mark.asyncio
+    async def test_successful_capture_overwrites_the_old_value(self):
+        from backend.app.main import _capture_timelapse_baseline_at_start
+
+        archive = MagicMock()
+        archive.timelapse_baseline = ["from_the_previous_run.avi"]
+        session = _session(archive)
+
+        with (
+            patch("backend.app.main.async_session", return_value=session),
+            patch(
+                "backend.app.main._list_timelapse_videos",
+                new=AsyncMock(return_value=([_video("now_on_the_printer.avi")], "/timelapse")),
+            ),
+        ):
+            await _capture_timelapse_baseline_at_start(_printer(), 1, logger, archive_id=7)
+
+        assert archive.timelapse_baseline == ["now_on_the_printer.avi"]
+
+
+class TestFileMustHaveStoppedGrowing:
+    """Matching the listing's size proves we received what it said, not that
+    the printer had finished writing. The scan's first look lands seconds after
+    the print ends — exactly when the video is being written — so a growing
+    file can be listed short, served short, and pass the length check. That was
+    survivable while the printer kept its copy; it isn't now that a successful
+    attach deletes it."""
+
+    @pytest.mark.asyncio
+    async def test_same_size_afterwards_is_settled(self):
+        from backend.app.services.bambu_ftp import remote_file_settled
+
+        with patch(
+            "backend.app.services.bambu_ftp.list_files_async",
+            AsyncMock(return_value=[_video("v.avi", size=4096)]),
+        ):
+            assert await remote_file_settled("1.2.3.4", "code", "/timelapse/v.avi", 4096) is True
+
+    @pytest.mark.asyncio
+    async def test_grown_since_download_is_not_settled(self):
+        """We hold a prefix of the video, not the video."""
+        from backend.app.services.bambu_ftp import remote_file_settled
+
+        with patch(
+            "backend.app.services.bambu_ftp.list_files_async",
+            AsyncMock(return_value=[_video("v.avi", size=9000)]),
+        ):
+            assert await remote_file_settled("1.2.3.4", "code", "/timelapse/v.avi", 4096) is False
+
+    @pytest.mark.asyncio
+    async def test_vanished_counts_as_settled(self):
+        """Nothing left that can grow, and nothing left to delete either."""
+        from backend.app.services.bambu_ftp import remote_file_settled
+
+        with patch(
+            "backend.app.services.bambu_ftp.list_files_async",
+            AsyncMock(return_value=[_video("something_else.avi")]),
+        ):
+            assert await remote_file_settled("1.2.3.4", "code", "/timelapse/v.avi", 4096) is True
+
+    @pytest.mark.asyncio
+    async def test_listing_failure_is_not_settled(self):
+        """ "Could not check" must not read as "safe to delete"."""
+        from backend.app.services.bambu_ftp import remote_file_settled
+
+        with patch("backend.app.services.bambu_ftp.list_files_async", AsyncMock(return_value=[])):
+            assert await remote_file_settled("1.2.3.4", "code", "/timelapse/v.avi", 4096) is False
+
+    @pytest.mark.asyncio
+    async def test_scan_discards_a_still_growing_video_without_deleting(self):
+        from backend.app.main import _attach_first_unclaimed_timelapse
+
+        delete = AsyncMock()
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", AsyncMock(return_value=b"x" * 1000)),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", AsyncMock(return_value=False)),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", delete),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+        ):
+            result = await _attach_first_unclaimed_timelapse(
+                42, _printer(), [_video("new.avi", size=1000)], set(), set(), 1, logger
+            )
+
+        assert result is False
+        service.attach_timelapse.assert_not_awaited()
+        delete.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_scan_attaches_once_the_video_has_settled(self):
+        from backend.app.main import _attach_first_unclaimed_timelapse
+
+        settled = AsyncMock(return_value=True)
+        service = MagicMock()
+        service.attach_timelapse = AsyncMock(return_value=True)
+
+        with (
+            patch("backend.app.services.bambu_ftp.download_file_bytes_async", AsyncMock(return_value=b"x" * 1000)),
+            patch("backend.app.services.bambu_ftp.remote_file_settled", settled),
+            patch("backend.app.services.bambu_ftp.delete_archived_timelapse", AsyncMock()),
+            patch("backend.app.main.async_session", return_value=_session()),
+            patch("backend.app.main.ArchiveService", return_value=service),
+            patch("backend.app.main.ws_manager", MagicMock(send_archive_updated=AsyncMock())),
+        ):
+            result = await _attach_first_unclaimed_timelapse(
+                42, _printer(), [_video("new.avi", size=1000)], set(), set(), 1, logger
+            )
+
+        assert result is True
+        # Checked against what we actually received, not against the listing.
+        assert settled.await_args.args[3] == 1000

backend/tests/unit/test_vp_mqtt_bridge.py

@@ -783,6 +783,77 @@ class TestPushStatusCache:
 
         await bridge.stop()
 
+    @pytest.mark.asyncio
+    async def test_a2l_ams_lite_slots_survive_in_slicer_cache(self):
+        """#2697 (reported by @qoatzelcoat): every A2L slot rendered as "?" in
+        BambuStudio through the VP, while Bambuddy's own AMS card was correct.
+
+        The A2L reports its AMS Lite as physical unit id 16 but packs the
+        presence bits at base 24. Bambuddy's internal path normalises 16 -> 6
+        before the cleanup runs, so it read the right bits; the bridge parses
+        the raw printer payload itself and still held 16, so the cleanup read
+        bits 64-67 — never set — and wiped all four slots in the cache the
+        slicer reads. A slicer-side filament pick reverted on the next 1 Hz
+        push for the same reason.
+
+        The cached units must keep the physical id 16: BambuStudio addresses
+        the Lite as 16 (it sends `ams_get_rfid {ams_id: 16}` through the VP).
+        """
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        # Reporter's capture: tray_exist_bits 0x7000000 = bits 24/25/26 →
+        # slots 0, 1, 2 loaded, slot 3 empty.
+        bridge._on_printer_raw(
+            f"device/{H2D_SERIAL}/report",
+            json.dumps(
+                {
+                    "print": {
+                        "command": "push_status",
+                        "ams": {
+                            "ams": [
+                                {
+                                    "id": "16",
+                                    "tray": [
+                                        {
+                                            "id": "0",
+                                            "state": 3,
+                                            "tray_type": "PLA",
+                                            "tray_sub_brands": "PLA Basic",
+                                            "tray_color": "C12E1FFF",
+                                            "tray_info_idx": "GFA00",
+                                            "remain": 100,
+                                        },
+                                        {"id": "1", "state": 3, "tray_type": "PETG", "tray_info_idx": "GFG00"},
+                                        {"id": "2", "state": 3, "tray_type": "ABS", "tray_info_idx": "GFB00"},
+                                        {"id": "3", "state": 3, "tray_type": "TPU", "tray_info_idx": "GFU00"},
+                                    ],
+                                }
+                            ],
+                            "tray_exist_bits": "7000000",
+                        },
+                    }
+                }
+            ).encode(),
+        )
+        await asyncio.sleep(0.01)
+
+        cached = bridge.get_latest_print_state()
+        unit = cached["ams"]["ams"][0]
+        # The slicer-facing cache keeps the PHYSICAL id — BambuStudio speaks 16.
+        assert unit["id"] == "16"
+        trays = unit["tray"]
+        assert trays[0]["tray_type"] == "PLA", "loaded slot wrongly cleared (bit base 64 regression)"
+        assert trays[1]["tray_type"] == "PETG"
+        assert trays[2]["tray_type"] == "ABS"
+        assert trays[0]["tray_info_idx"] == "GFA00"
+        # Slot 3 is genuinely empty and still gets the normal cleanup.
+        assert trays[3]["state"] == 9
+        assert trays[3]["tray_type"] == ""
+
+        await bridge.stop()
+
     @pytest.mark.asyncio
     async def test_tray_exist_bits_shutdown_guard_preserves_cache(self):
         """#765 shutdown guard mirrored at the bridge: when the printer