fix(timelapse): poll longer, diff without a clock, delete once archived (#2704)

Timelapse was on, the video never reached the archive, and Scan for Timelapse
found nothing afterwards. Across 247 support bundles this was the norm, not an
edge case: 457 automatic scans scheduled, 262 attached.

The scan looked four times over ~65s. The attempt that found the video was #1
272 times, then 17 / 13 / 13 — flat against the cutoff, not decaying, i.e.
files were still arriving when we stopped. What ran afterwards searched for the
print name inside the filename; Bambu only writes "video_<timestamp>", so it
fired 159 times and matched zero.

The manual Scan had no baseline at all and matched on filename timestamp, FTP
mtime, or "there is only one video" — all reading a clock a LAN-only printer
cannot sync. The reporter's P1S was six and a half days out.

- Poll for minutes instead of ~65s; drop the name-match fallback.
- Persist the print-start baseline on the archive, so the diff survives a
  restart mid-print and the manual Scan runs the same comparison. With a
  baseline present the clock-based strategies are skipped entirely — they can
  only turn an honest "pick one" into a confident wrong answer.
- When several files are new (a previous print's video landing late), exclude
  the ones already attached to another archive instead of ordering the
  candidates. Ordering could only be done on the printer's clock.
- Delete the video from the printer once archived. Keeps /timelapse to
  unclaimed files, which is what makes the diff unambiguous, and stops P1S
  cards filling with AVIs.
- Gate that delete on a verified transfer: download_file now compares against
  the size from the listing. An FTPS connection closing early does not always
  raise, so a partial buffer was being attached as a complete video — which
  would also have been the one case where deleting the source lost data.

Bounded twice on purpose: wall-clock deadline plus a derived round cap, since
the deadline stops bounding the loop as soon as the sleeps are shortened.
Per-round logging only speaks when the listing changed — 31 rounds of full
listings would bury the interesting line in the support bundle.

Migration adds print_archives.timelapse_baseline as JSON, spelled the same on
both dialects so a migrated database matches a fresh one.

-----------

fix(finish-photo): add the timelapse frame to the archive after the notification (#2704)

When a print records a timelapse, its last frame is the better finish photo:
the firmware stops recording with the toolhead parked and before the end
G-code drops the bed, where a live grab at that moment catches a lowered
plate. Bambuddy waited 60s for the video and then gave up, because the
print-complete notification blocks on that photo and holding a notification
for minutes is worse than sending it with the live grab.

P1-series printers write MJPEG AVI rather than H.264 MP4 and serve it slowly.
Measured over 261 attaches in the support bundles: P1S median 33s, p90 167s,
worst 546s, while every other model finished inside 26s. So the printers that
most needed the better framing were the ones that never got it.

Keep the notification on the same bound, and keep waiting off to the side.
_capture_finish_photo_from_timelapse now reports whether it ran out of time or
concluded — a video that landed and failed extraction is not worth retrying,
one that never arrived is. On the first, schedule a background task that waits
up to 15 minutes and inserts the extracted frame at the front of the archive's
photo list, where the gallery opens.

The live grab stays on disk: the notification already links to that exact
file, so removing it would leave a broken image in Discord or Telegram.

The length check proves we received what the listing said, not that the file
was finished. The first look happens ~5s after the print ends, while the
printer may still be writing, so a growing file can be listed short, served
short, and pass. Re-list after the download and only accept the video once its
size has stopped changing — a failed re-list counts as not settled, since
"could not check" must not mean "safe to delete".

backend/app/api/routes/archives.py

@@ -123,6 +123,28 @@ def _match_timelapse_by_timestamp(
     return best_video, best_diff
 
 
+async def _claimed_timelapse_stems(db, printer_id: int | None, exclude_archive_id: int) -> set[str]:
+    """Video filenames already attached to another archive of this printer (#2704).
+
+    Lets the baseline diff drop a previous print's late-landing video from the
+    candidate list without ordering the candidates — ordering could only be done
+    on mtime or the filename timestamp, and both come from a clock the printer
+    can't sync in LAN-only mode. ``attach_timelapse`` stores the video under the
+    printer's own filename and the MP4 conversion keeps the stem, so the stem of
+    ``timelapse_path`` is what was claimed.
+    """
+    if printer_id is None:
+        return set()
+    rows = await db.execute(
+        select(PrintArchive.timelapse_path).where(
+            PrintArchive.printer_id == printer_id,
+            PrintArchive.id != exclude_archive_id,
+            PrintArchive.timelapse_path.is_not(None),
+        )
+    )
+    return {Path(p).stem for p in rows.scalars().all() if p}
+
+
 def _ensure_archive_visible(
     archive: PrintArchive | None,
     user: User | None,
@@ -2270,9 +2292,11 @@ async def scan_timelapse(
     from backend.app.core.database import async_session
     from backend.app.models.printer import Printer
     from backend.app.services.bambu_ftp import (
+        delete_archived_timelapse,
         download_file_bytes_async,
         get_ftp_retry_settings,
         list_files_async,
+        remote_file_settled,
         with_ftp_retry,
     )
 
@@ -2322,18 +2346,48 @@ async def scan_timelapse(
         f for f in files if not f.get("is_directory") and f.get("name", "").lower().endswith((".mp4", ".avi"))
     ]
 
+    # Strategy 0: snapshot diff against the baseline captured at print start
+    # (#2704). This is the same comparison the automatic scan makes, and the
+    # only one here that doesn't depend on the printer's clock — a printer in
+    # LAN-only mode can't reach Bambu's NTP server, so the timestamps in both
+    # the filename and the FTP mtime can be days out. One reporter's P1S was
+    # six and a half days off, which defeats every strategy below.
+    #
+    # When a baseline exists it is authoritative and the clock-based strategies
+    # are skipped entirely: they can only turn an honest "pick one yourself"
+    # into a confident wrong answer. Those strategies stay for archives created
+    # before the baseline was persisted.
+    used_baseline = archive.timelapse_baseline is not None
+    if used_baseline:
+        baseline = set(archive.timelapse_baseline)
+        async with async_session() as db:
+            claimed = await _claimed_timelapse_stems(db, archive.printer_id, archive_id)
+        candidates = [
+            f for f in video_files if f.get("name", "") not in baseline and Path(f.get("name", "")).stem not in claimed
+        ]
+        if len(candidates) == 1:
+            matching_file = candidates[0]
+            logger.info("Matched timelapse by print-start baseline: %s", matching_file.get("name"))
+        elif candidates:
+            # Ambiguous — offer only the plausible files instead of guessing.
+            video_files = candidates
+            logger.info("Baseline left %s unclaimed candidates for archive %s", len(candidates), archive_id)
+        else:
+            logger.info("Baseline shows no unclaimed new video on the printer for archive %s", archive_id)
+
     # Strategy 1: Match by print name in filename
-    for f in video_files:
-        fname = f.get("name", "")
-        if base_name.lower() in fname.lower():
-            matching_file = f
-            break
+    if not used_baseline:
+        for f in video_files:
+            fname = f.get("name", "")
+            if base_name.lower() in fname.lower():
+                matching_file = f
+                break
 
     # Strategy 2: Match by timestamp proximity against print START time.
     # Bambu timelapse filename embeds the print start time in printer-local clock.
     # See _match_timelapse_by_timestamp for the offset-search rationale and why we
     # intentionally don't try to match filename against end time here.
-    if not matching_file and archive.started_at:
+    if not used_baseline and not matching_file and archive.started_at:
         candidate, diff = _match_timelapse_by_timestamp(video_files, archive.started_at)
         if candidate is not None:
             matching_file = candidate
@@ -2341,7 +2395,7 @@ async def scan_timelapse(
 
     # Strategy 3: Use file modification time from FTP listing
     # This handles cases where printer's filename timestamp is wrong but file mtime is correct
-    if not matching_file and (archive.started_at or archive.completed_at or archive.created_at):
+    if not used_baseline and not matching_file and (archive.started_at or archive.completed_at or archive.created_at):
         from datetime import datetime, timedelta
 
         _archive_start = archive.started_at
@@ -2369,7 +2423,7 @@ async def scan_timelapse(
 
     # Strategy 4: If only one timelapse exists and archive was recently completed, use it
     # This handles cases where printer clock is wrong or timezone issues exist
-    if not matching_file and len(video_files) == 1:
+    if not used_baseline and not matching_file and len(video_files) == 1:
         from datetime import datetime, timedelta, timezone
 
         archive_completed = archive.completed_at or archive.created_at
@@ -2419,6 +2473,7 @@ async def scan_timelapse(
             remote_path,
             socket_timeout=ftp_timeout,
             printer_model=printer.model,
+            expected_size=matching_file.get("size"),
             max_retries=ftp_retry_count,
             retry_delay=ftp_retry_delay,
             operation_name=f"Download timelapse {matching_file['name']}",
@@ -2430,11 +2485,24 @@ async def scan_timelapse(
             remote_path,
             socket_timeout=ftp_timeout,
             printer_model=printer.model,
+            expected_size=matching_file.get("size"),
         )
 
     if not timelapse_data:
         raise HTTPException(500, "Failed to download timelapse")
 
+    # Confirm the printer has finished writing before we commit to this file and
+    # delete the original: matching the listing's size proves we got what it
+    # said, not that the file was complete (#2704).
+    if not await remote_file_settled(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        len(timelapse_data),
+        printer_model=printer.model,
+    ):
+        raise HTTPException(409, "The printer is still writing this video — try again in a moment")
+
     # Attach in a fresh short session (the read session was released before FTP).
     async with async_session() as db:
         success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, matching_file["name"])
@@ -2442,6 +2510,17 @@ async def scan_timelapse(
     if not success:
         raise HTTPException(500, "Failed to attach timelapse")
 
+    # Safe now, and only now: the transfer matched the size the listing reported
+    # and the bytes are committed to the archive (#2704).
+    await delete_archived_timelapse(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        verified=matching_file.get("size") is not None,
+        printer_model=printer.model,
+        printer_name=printer.name,
+    )
+
     return {
         "status": "attached",
         "message": f"Timelapse '{matching_file['name']}' attached successfully",
@@ -2459,9 +2538,11 @@ async def select_timelapse(
     from backend.app.core.database import async_session
     from backend.app.models.printer import Printer
     from backend.app.services.bambu_ftp import (
+        delete_archived_timelapse,
         download_file_bytes_async,
         get_ftp_retry_settings,
         list_files_async,
+        remote_file_settled,
         with_ftp_retry,
     )
 
@@ -2484,6 +2565,7 @@ async def select_timelapse(
     # Find the file on the printer
     files = []
     remote_path = None
+    expected_size = None
     for timelapse_dir in ["/timelapse", "/timelapse/video", "/record", "/recording"]:
         try:
             files = await list_files_async(
@@ -2492,6 +2574,7 @@ async def select_timelapse(
             for f in files:
                 if f.get("name") == filename:
                     remote_path = f.get("path") or f"{timelapse_dir}/{filename}"
+                    expected_size = f.get("size")
                     break
             if remote_path:
                 break
@@ -2512,6 +2595,7 @@ async def select_timelapse(
             remote_path,
             socket_timeout=ftp_timeout,
             printer_model=printer.model,
+            expected_size=expected_size,
             max_retries=ftp_retry_count,
             retry_delay=ftp_retry_delay,
             operation_name=f"Download timelapse {filename}",
@@ -2523,17 +2607,41 @@ async def select_timelapse(
             remote_path,
             socket_timeout=ftp_timeout,
             printer_model=printer.model,
+            expected_size=expected_size,
         )
 
     if not timelapse_data:
         raise HTTPException(500, "Failed to download timelapse")
 
+    # Confirm the printer has finished writing before we commit to this file and
+    # delete the original: matching the listing's size proves we got what it
+    # said, not that the file was complete (#2704).
+    if not await remote_file_settled(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        len(timelapse_data),
+        printer_model=printer.model,
+    ):
+        raise HTTPException(409, "The printer is still writing this video — try again in a moment")
+
     # Attach in a fresh short session (the read session was released before FTP).
     async with async_session() as db:
         success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, filename)
     if not success:
         raise HTTPException(500, "Failed to attach timelapse")
 
+    # Safe now, and only now: the transfer matched the size the listing reported
+    # and the bytes are committed to the archive (#2704).
+    await delete_archived_timelapse(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        verified=expected_size is not None,
+        printer_model=printer.model,
+        printer_name=printer.name,
+    )
+
     return {
         "status": "attached",
         "message": f"Timelapse '{filename}' attached successfully",

backend/app/core/database.py

@@ -3798,6 +3798,18 @@ async def run_migrations(conn):
     await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN library_file_id INTEGER")
     await _safe_execute(conn, "ALTER TABLE projects ADD COLUMN target_sets INTEGER")
 
+    # Migration: persist the timelapse snapshot-diff baseline (#2704).
+    # The list of video filenames present on the printer when the print began,
+    # so the diff survives a restart and the manual scan can use it instead of
+    # the clock-based matching that a LAN-only printer defeats. No dialect
+    # branch: SQLAlchemy renders this column as `JSON` on both SQLite and
+    # Postgres for a fresh install (checked with CreateTable against each
+    # dialect), so spelling the ALTER the same way keeps a migrated database
+    # identical to a new one. Matching matters on Postgres in particular —
+    # asyncpg binds the serialised value as json and would reject a TEXT column
+    # (mirrors the `projects.attachments JSON` migration above).
+    await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN timelapse_baseline JSON")
+
     # Migration: plate-clear-required notification opt-in (#2525). Off by
     # default — it fires after every print, at the same moment as the
     # print-complete alert. Postgres rejects `DEFAULT 0` for BOOLEAN.

backend/app/main.py

@@ -2700,6 +2700,11 @@ async def on_print_start(printer_id: int, data: dict):
                 # scanner runs fresh; also unlink the old video file so reprints
                 # don't accumulate orphans in the archive directory. Photos list
                 # is left alone — accumulating one finish photo per run is fine.
+                # The print-start baseline (#2704) is stale for the same reason:
+                # it describes the printer before the previous run. The capture
+                # below overwrites it, but clear it here too so an early failure
+                # can't leave the scan diffing against the wrong snapshot.
+                archive.timelapse_baseline = None
                 stale_timelapse_relpath = archive.timelapse_path
                 if stale_timelapse_relpath:
                     archive.timelapse_path = None
@@ -2838,7 +2843,7 @@ async def on_print_start(printer_id: int, data: dict):
                 # falls into its "take baseline now" fallback, which snapshots
                 # AFTER the new MP4 already exists and never matches a diff
                 # (#1403 follow-up — see pwostran's 2026-05-18 support bundle).
-                await _capture_timelapse_baseline_at_start(printer, printer_id, logger)
+                await _capture_timelapse_baseline_at_start(printer, printer_id, logger, archive_id=archive.id)
 
             return  # Skip creating a new archive
 
@@ -3488,7 +3493,7 @@ async def on_print_start(printer_id: int, data: dict):
                     logger.warning("[SPOOLMAN] Failed to store tracking data: %s", e)
 
                 # Capture timelapse file baseline for snapshot-diff on completion
-                await _capture_timelapse_baseline_at_start(printer, printer_id, logger)
+                await _capture_timelapse_baseline_at_start(printer, printer_id, logger, archive_id=archive.id)
         finally:
             # Keep temp_path around until print completes so the cover endpoint
             # can reuse it (#972). Cache eviction in on_print_complete deletes
@@ -3501,6 +3506,62 @@ async def on_print_start(printer_id: int, data: dict):
 
 _TIMELAPSE_VIDEO_EXTENSIONS = (".mp4", ".avi")
 
+# Poll schedule for the post-print timelapse scan (#2704). Module-level so
+# tests can shrink them without waiting out real delays.
+#
+# This replaced a fixed [5, 10, 20, 30] retry ladder, i.e. roughly 65 s of
+# looking. Across 247 support bundles the attempt that found the video was #1
+# 272 times, then 17 / 13 / 13 — a flat tail against the cutoff rather than a
+# decaying one, which is the signature of a budget that expires while files are
+# still arriving. 457 scans were scheduled and only 262 ever attached. Big
+# prints make big videos and the printer writes them after the print ends, so
+# the poll now runs for minutes and costs one FTP LIST per round.
+_TIMELAPSE_SCAN_FIRST_DELAY_SECONDS: float = 5.0
+_TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS: float = 30.0
+_TIMELAPSE_SCAN_TIMEOUT_SECONDS: float = 900.0
+
+
+def _timelapse_scan_max_attempts() -> int:
+    """Round cap for the poll, derived from the wall-clock budget.
+
+    The deadline alone is not a sufficient bound: it assumes each round really
+    waits, which stops being true the moment ``asyncio.sleep`` is patched out,
+    and an FTP list that fails immediately would otherwise spin against the
+    printer at full speed for the whole window. Whichever bound is reached
+    first ends the poll.
+    """
+    if _TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS <= 0:
+        # A zero interval makes the wall-clock budget meaningless; fall back to
+        # the round count the production interval would have given.
+        return 32
+    return max(1, int(_TIMELAPSE_SCAN_TIMEOUT_SECONDS // _TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS) + 1)
+
+
+async def _claimed_timelapse_names(db, printer_id: int, exclude_archive_id: int) -> set[str]:
+    """Video filenames already attached to some other archive of this printer.
+
+    Used to disambiguate when more than one file is new since the baseline —
+    which happens when a previous print's video landed after this print's
+    baseline was taken. Ordering the candidates would be the obvious fix and is
+    the wrong one: it can only be done on mtime or on the filename timestamp,
+    both of which come from the printer's own clock, and a LAN-only printer
+    can't reach Bambu's NTP server. Exclusion needs no clock at all.
+
+    ``attach_timelapse`` saves the video into the archive directory under the
+    printer's original filename, and the later MP4 conversion keeps the stem,
+    so the stem of ``timelapse_path`` recovers what was claimed.
+    """
+    from backend.app.models.archive import PrintArchive
+
+    rows = await db.execute(
+        select(PrintArchive.timelapse_path).where(
+            PrintArchive.printer_id == printer_id,
+            PrintArchive.id != exclude_archive_id,
+            PrintArchive.timelapse_path.is_not(None),
+        )
+    )
+    return {Path(p).stem for p in rows.scalars().all() if p}
+
 
 async def _list_timelapse_videos(printer) -> tuple[list[dict], str | None]:
     """List video files from printer's timelapse directory.
@@ -3533,7 +3594,9 @@ async def _list_timelapse_videos(printer) -> tuple[list[dict], str | None]:
     return [], None
 
 
-async def _capture_timelapse_baseline_at_start(printer, printer_id: int, logger: logging.Logger) -> None:
+async def _capture_timelapse_baseline_at_start(
+    printer, printer_id: int, logger: logging.Logger, archive_id: int | None = None
+) -> None:
     """Snapshot the printer's timelapse directory at print start so the
     completion-time scan can pick the new file by set-difference.
 
@@ -3546,39 +3609,69 @@ async def _capture_timelapse_baseline_at_start(printer, printer_id: int, logger:
 
     Bambu printers in LAN-only mode don't sync NTP, so mtime ordering is
     unreliable — the snapshot-diff approach sidesteps that entirely.
+
+    When ``archive_id`` is known the baseline is also written to the archive
+    row, so it survives a restart and the manual "Scan for Timelapse" button
+    can run the same diff instead of falling back to clock-based matching
+    (#2704). Only baselines taken at print start are persisted — one taken at
+    completion already contains the new video and would poison a later scan.
     """
+    names: set[str] | None = None
     try:
         baseline_files, _ = await _list_timelapse_videos(printer)
-        _timelapse_baselines[printer_id] = {f.get("name", "") for f in baseline_files}
+        names = {f.get("name", "") for f in baseline_files}
+        _timelapse_baselines[printer_id] = names
         logger.info(
             "[TIMELAPSE] Baseline at print start: %s video files for printer %s",
-            len(_timelapse_baselines[printer_id]),
+            len(names),
             printer_id,
         )
     except Exception as e:
         logger.warning("[TIMELAPSE] Failed to capture baseline at print start: %s", e)
 
+    if archive_id is None:
+        return
+    try:
+        async with async_session() as db:
+            from backend.app.models.archive import PrintArchive
 
-async def _scan_for_timelapse_with_retries(archive_id: int, baseline_names: set[str] | None = None):
-    """
-    Scan for timelapse with retries using a snapshot-diff approach.
-
-    Instead of picking the "most recent by mtime" (unreliable when the printer
-    clock is wrong in LAN-only mode), we snapshot existing MP4 filenames BEFORE
-    waiting, then look for any NEW filename that appears after each delay.
+            archive = await db.get(PrintArchive, archive_id)
+            if archive is not None:
+                # Written even when the listing failed, and then as NULL. A
+                # reprint reuses the archive row, so leaving the previous run's
+                # baseline in place would have the scan diff this print against
+                # the state of the printer before the *last* one — and a stale
+                # baseline reads as authoritative, where NULL correctly falls
+                # back to a fresh snapshot.
+                archive.timelapse_baseline = sorted(names) if names is not None else None
+                await db.commit()
+    except Exception as e:
+        # In-memory baseline still covers the normal completion path.
+        logger.warning("[TIMELAPSE] Failed to persist baseline for archive %s: %s", archive_id, e)
 
-    If baseline_names is provided (captured at print start), it is used directly.
-    Otherwise falls back to taking a baseline at completion time (best-effort
-    for prints started before app restart).
 
-    Falls back to name-matching (print name contained in MP4 filename) if no
-    new file appears after all retries.
+async def _scan_for_timelapse_with_retries(archive_id: int, baseline_names: set[str] | None = None):
+    """Poll the printer for this print's timelapse and attach it.
+
+    Snapshot diff, not timestamp matching: a printer in LAN-only mode cannot
+    reach Bambu's NTP server, so the clock behind both the filename and the FTP
+    mtime is arbitrarily wrong — one reporter's P1S was six and a half days out
+    (#2704). Comparing the current listing against the set of filenames that
+    existed when the print started needs no clock at all, because the printer
+    writes the video only once the print has ended.
+
+    Baseline precedence: the caller's in-memory set, then the one persisted on
+    the archive at print start, then a snapshot taken now. The last of those is
+    a poor substitute — by completion the new video may already be on the card,
+    in which case it lands in the "baseline" and no diff can ever match — but it
+    is all that is available for a print that began before Bambuddy started.
+
+    On success the video is deleted from the printer, which keeps ``/timelapse``
+    down to the unclaimed files and makes the next diff unambiguous.
     """
-    from pathlib import Path
-
     logger = logging.getLogger(__name__)
 
-    # --- Phase 1: Take baseline snapshot of existing timelapse files ---
+    # --- Phase 1: establish the baseline -------------------------------------
     try:
         async with async_session() as db:
             from backend.app.models.printer import Printer
@@ -3597,14 +3690,20 @@ async def _scan_for_timelapse_with_retries(archive_id: int, baseline_names: set[
                 return
 
             if baseline_names is not None:
-                # Use pre-captured baseline from print start (no race condition)
                 logger.info(
                     "[TIMELAPSE] Using print-start baseline: %s existing video files for archive %s",
                     len(baseline_names),
                     archive_id,
                 )
+            elif archive.timelapse_baseline is not None:
+                # Persisted at print start — survives a restart mid-print.
+                baseline_names = set(archive.timelapse_baseline)
+                logger.info(
+                    "[TIMELAPSE] Using stored baseline: %s existing video files for archive %s",
+                    len(baseline_names),
+                    archive_id,
+                )
             else:
-                # Fallback: take baseline now (e.g. app restarted mid-print)
                 result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
                 printer = result.scalar_one_or_none()
                 if not printer:
@@ -3619,144 +3718,208 @@ async def _scan_for_timelapse_with_retries(archive_id: int, baseline_names: set[
                     archive_id,
                 )
 
-            # Derive base_name for name-matching fallback
-            base_name = Path(archive.filename).stem if archive.filename else ""
-            if base_name.endswith(".gcode"):
-                base_name = base_name[:-6]
-
     except Exception as e:
         logger.warning("[TIMELAPSE] Failed to take baseline snapshot for archive %s: %s", archive_id, e)
         return
 
-    # --- Phase 2: Retry loop — look for NEW files that weren't in baseline ---
-    retry_delays = [5, 10, 20, 30]
+    # --- Phase 2: poll for a file that was not there when the print began -----
+    deadline = time.monotonic() + _TIMELAPSE_SCAN_TIMEOUT_SECONDS
+    max_attempts = _timelapse_scan_max_attempts()
+    seen_names: set[str] = set()
+    delay = _TIMELAPSE_SCAN_FIRST_DELAY_SECONDS
+    attempt = 0
 
-    for attempt, delay in enumerate(retry_delays, 1):
-        logger.info(
-            "[TIMELAPSE] Attempt %s/%s: waiting %ss before scanning for archive %s",
-            attempt,
-            len(retry_delays),
-            delay,
-            archive_id,
-        )
+    while True:
         await asyncio.sleep(delay)
+        delay = _TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS
+        attempt += 1
 
         try:
             from backend.app.models.printer import Printer
-            from backend.app.services.bambu_ftp import download_file_bytes_async
 
             # Read phase: fetch archive + printer in a short session and release
             # the pooled connection BEFORE the FTP list/download below. Holding it
             # across the FTP round-trips left one connection idle-in-transaction per
-            # in-flight scan — ×4 retries, per completed print (issue #2572).
+            # in-flight scan (issue #2572).
             async with async_session() as db:
                 service = ArchiveService(db)
                 archive = await service.get_archive(archive_id)
 
                 if not archive:
-                    logger.warning("[TIMELAPSE] Archive %s not found, stopping retries", archive_id)
+                    logger.warning("[TIMELAPSE] Archive %s not found, stopping poll", archive_id)
                     return
                 if archive.timelapse_path:
-                    logger.info("[TIMELAPSE] Archive %s already has timelapse attached, stopping retries", archive_id)
+                    logger.info("[TIMELAPSE] Archive %s already has timelapse attached, stopping poll", archive_id)
                     return
 
                 result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
                 printer = result.scalar_one_or_none()
                 if not printer:
-                    logger.warning("[TIMELAPSE] Printer not found for archive %s, stopping retries", archive_id)
+                    logger.warning("[TIMELAPSE] Printer not found for archive %s, stopping poll", archive_id)
                     return
 
+                claimed = await _claimed_timelapse_names(db, archive.printer_id, archive_id)
+
             # I/O phase (no DB connection held): FTP list + download.
             video_files, found_path = await _list_timelapse_videos(printer)
 
-            if not video_files:
-                logger.info("[TIMELAPSE] Attempt %s: No video files found, will retry", attempt)
-                continue
-
-            logger.info("[TIMELAPSE] Attempt %s: Found %s video files in %s", attempt, len(video_files), found_path)
-            for f in video_files[:5]:
-                logger.info("[TIMELAPSE]   - %s", f.get("name"))
-
-            # Find files that are NEW (not in baseline snapshot)
-            new_files = [f for f in video_files if f.get("name", "") not in baseline_names]
-
-            if new_files:
-                # Pick the first new file (there should typically be exactly one)
-                target = new_files[0]
-                file_name = target.get("name")
-                remote_path = target.get("path") or f"/timelapse/{file_name}"
-                logger.info(
-                    "[TIMELAPSE] Attempt %s: New file detected: %s (downloading for archive %s)",
-                    attempt,
-                    file_name,
-                    archive_id,
+            # The poll can run for dozens of rounds, so only narrate a round
+            # that saw something change. Repeating the whole listing every 30 s
+            # would bury the one interesting line in the support bundle.
+            names_now = {f.get("name", "") for f in video_files}
+            changed = attempt == 1 or names_now != seen_names
+            seen_names = names_now
+            speak = logger.info if changed else logger.debug
+
+            if video_files:
+                speak("[TIMELAPSE] Attempt %s: Found %s video files in %s", attempt, len(video_files), found_path)
+                if changed:
+                    for f in video_files[:5]:
+                        logger.info("[TIMELAPSE]   - %s", f.get("name"))
+
+                attached = await _attach_first_unclaimed_timelapse(
+                    archive_id, printer, video_files, baseline_names, claimed, attempt, logger, quiet=not changed
                 )
-
-                timelapse_data = await download_file_bytes_async(
-                    printer.ip_address, printer.access_code, remote_path, printer_model=printer.model
-                )
-                if timelapse_data:
-                    # Write phase: attach in a fresh short-lived session.
-                    async with async_session() as db:
-                        success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, file_name)
-                    if success:
-                        logger.info("[TIMELAPSE] Successfully attached timelapse to archive %s", archive_id)
-                        await ws_manager.send_archive_updated({"id": archive_id, "timelapse_attached": True})
-                        return
-                    else:
-                        logger.warning("[TIMELAPSE] Failed to attach timelapse to archive %s", archive_id)
-                else:
-                    logger.warning("[TIMELAPSE] Attempt %s: Failed to download new file, will retry", attempt)
+                if attached:
+                    return
             else:
-                logger.info("[TIMELAPSE] Attempt %s: No new files since baseline, will retry", attempt)
+                speak("[TIMELAPSE] Attempt %s: No video files found, will retry", attempt)
 
         except Exception as e:
             logger.warning("[TIMELAPSE] Attempt %s failed with error: %s", attempt, e)
 
-    # --- Phase 3: Fallback — try name matching against all files ---
-    if base_name:
-        logger.info("[TIMELAPSE] Retries exhausted, trying name-match fallback for '%s'", base_name)
-        try:
-            from backend.app.models.printer import Printer
-            from backend.app.services.bambu_ftp import download_file_bytes_async
+        if attempt >= max_attempts or time.monotonic() >= deadline:
+            break
 
-            # Read phase: short session, released before the FTP work (issue #2572).
-            async with async_session() as db:
-                service = ArchiveService(db)
-                archive = await service.get_archive(archive_id)
-                if not archive or archive.timelapse_path:
-                    return
+    # No name-match fallback: it compared the print name against the filename,
+    # and Bambu firmware only ever writes "video_<timestamp>". Across 247 support
+    # bundles it fired 159 times and matched zero times, so all it added was a
+    # misleading log line before giving up.
+    logger.warning(
+        "[TIMELAPSE] No new video appeared for archive %s within %ss, giving up",
+        archive_id,
+        int(_TIMELAPSE_SCAN_TIMEOUT_SECONDS),
+    )
 
-                result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
-                printer = result.scalar_one_or_none()
-                if not printer:
-                    return
 
-            # I/O phase (no DB connection held): FTP list + download.
-            video_files, found_path = await _list_timelapse_videos(printer)
-            for f in video_files:
-                fname = f.get("name", "")
-                if base_name.lower() in fname.lower():
-                    remote_path = f.get("path") or f"/timelapse/{fname}"
-                    logger.info("[TIMELAPSE] Name-match fallback: '%s' matches '%s'", base_name, fname)
-
-                    timelapse_data = await download_file_bytes_async(
-                        printer.ip_address, printer.access_code, remote_path, printer_model=printer.model
-                    )
-                    if timelapse_data:
-                        # Write phase: attach in a fresh short-lived session.
-                        async with async_session() as db:
-                            success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, fname)
-                        if success:
-                            logger.info("[TIMELAPSE] Name-match fallback attached timelapse to archive %s", archive_id)
-                            await ws_manager.send_archive_updated({"id": archive_id, "timelapse_attached": True})
-                            return
-                    break  # Only try the first name match
+async def _attach_first_unclaimed_timelapse(
+    archive_id: int,
+    printer,
+    video_files: list[dict],
+    baseline_names: set[str],
+    claimed: set[str],
+    attempt: int,
+    logger: logging.Logger,
+    *,
+    quiet: bool = False,
+) -> bool:
+    """Download and attach the one video that belongs to this print.
+
+    A candidate is any file absent from the print-start baseline. More than one
+    can qualify when a previous print's video landed late, after this print's
+    baseline was taken — those are filtered out by name, because they are
+    already attached to another archive. Sorting the candidates instead would
+    mean sorting on mtime or on the filename timestamp, both of which come from
+    the printer's unsynced clock.
+
+    Returns True once a video is attached. The printer's copy is deleted only
+    after the attach succeeds on bytes whose length matched the listing.
+
+    ``quiet`` downgrades the "nothing yet" lines to DEBUG when the caller has
+    already seen this exact listing — the poll runs for many rounds and only the
+    rounds where something changed are worth an INFO line.
+    """
+    from backend.app.services.bambu_ftp import (
+        delete_archived_timelapse,
+        download_file_bytes_async,
+        remote_file_settled,
+    )
 
-        except Exception as e:
-            logger.warning("[TIMELAPSE] Name-match fallback failed: %s", e)
+    speak = logger.debug if quiet else logger.info
+
+    new_files = [f for f in video_files if f.get("name", "") not in baseline_names]
+    if not new_files:
+        speak("[TIMELAPSE] Attempt %s: No new files since baseline, will retry", attempt)
+        return False
+
+    candidates = [f for f in new_files if Path(f.get("name", "")).stem not in claimed]
+    if not candidates:
+        speak(
+            "[TIMELAPSE] Attempt %s: %s new file(s), all already attached to other archives, will retry",
+            attempt,
+            len(new_files),
+        )
+        return False
+    if len(candidates) > 1:
+        logger.warning(
+            "[TIMELAPSE] Attempt %s: %s unclaimed new files (%s) — taking the first; "
+            "the rest stay on the printer for manual selection",
+            attempt,
+            len(candidates),
+            ", ".join(str(f.get("name")) for f in candidates),
+        )
+
+    target = candidates[0]
+    file_name = target.get("name")
+    remote_path = target.get("path") or f"/timelapse/{file_name}"
+    logger.info(
+        "[TIMELAPSE] Attempt %s: New file detected: %s (downloading for archive %s)",
+        attempt,
+        file_name,
+        archive_id,
+    )
+
+    # The listing always carries a size (`list_files` skips entries it can't
+    # parse), but read it explicitly: the delete below is destructive and must
+    # depend on a size we actually had, not on one we hoped was there.
+    expected_size = target.get("size")
+
+    timelapse_data = await download_file_bytes_async(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        printer_model=printer.model,
+        expected_size=expected_size,
+    )
+    if not timelapse_data:
+        # Short or failed transfer. The printer keeps its copy, so the next
+        # round can try again — which is exactly why the delete below is
+        # gated on a verified download.
+        logger.warning("[TIMELAPSE] Attempt %s: Failed to download new file, will retry", attempt)
+        return False
 
-    logger.warning("[TIMELAPSE] All attempts exhausted for archive %s, giving up", archive_id)
+    # The length check above proves we got what the listing said, not that the
+    # printer had finished writing. A video still being written can be listed
+    # short, served short, and pass — so confirm it has stopped growing before
+    # committing to it and deleting the original (#2704).
+    if not await remote_file_settled(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        len(timelapse_data),
+        printer_model=printer.model,
+    ):
+        return False
+
+    # Write phase: attach in a fresh short-lived session.
+    async with async_session() as db:
+        success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, file_name)
+    if not success:
+        logger.warning("[TIMELAPSE] Failed to attach timelapse to archive %s", archive_id)
+        return False
+
+    logger.info("[TIMELAPSE] Successfully attached timelapse to archive %s", archive_id)
+    await ws_manager.send_archive_updated({"id": archive_id, "timelapse_attached": True})
+
+    await delete_archived_timelapse(
+        printer.ip_address,
+        printer.access_code,
+        remote_path,
+        verified=expected_size is not None,
+        printer_model=printer.model,
+        printer_name=printer.name,
+    )
+    return True
 
 
 # Defaults for the finish-photo-from-timelapse polling loop (#1397). These are
@@ -3764,11 +3927,25 @@ async def _scan_for_timelapse_with_retries(archive_id: int, baseline_names: set[
 _FINISH_PHOTO_TIMELAPSE_POLL_INTERVAL_SECONDS: float = 3.0
 _FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS: float = 60.0
 
+# How long the *background* upgrade keeps waiting after the notification has
+# already gone out (#2704 follow-up). The short bound above exists so a slow
+# printer can't hold up the print-complete notification; this one exists so the
+# archive still ends up with the better frame afterwards.
+#
+# Measured across 261 attaches in the support bundles, the video lands a median
+# 13s after the print ends — but the P1 series writes MJPEG AVI rather than
+# H.264 MP4 and serves it slowly, so its p90 is 167s and the worst observed case
+# was 546s. Every other model was inside 26s. The long budget is therefore
+# almost entirely for P1-series users; on everything else the short wait already
+# wins and this task never runs.
+_FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS: float = 900.0
+
 
 async def _capture_finish_photo_from_timelapse(
     archive_id: int,
     archive_dir: Path,
-) -> str | None:
+    timeout: float | None = None,
+) -> tuple[str | None, bool]:
     """Wait for the per-print timelapse to land on the archive and extract its
     last frame as the finish photo (#1397).
 
@@ -3779,10 +3956,14 @@ async def _capture_finish_photo_from_timelapse(
 
     ``_scan_for_timelapse_with_retries`` runs in parallel and writes
     ``archive.timelapse_path`` when the file lands. This function polls for
-    that field. Returns the saved photo filename on success, or None if the
-    timelapse never arrives within the timeout / extraction fails / no
-    timelapse path was set — in which case the caller falls back to the
-    existing live-camera capture chain.
+    that field.
+
+    Returns ``(filename, still_pending)``. ``still_pending`` is True only when
+    the wait ran out with no video on the archive yet — i.e. the video may
+    still be coming and a later attempt could succeed. It is False when the
+    video landed (whether or not extraction worked), because in that case
+    waiting longer changes nothing. The caller uses that to decide between
+    falling back permanently and scheduling a background upgrade.
     """
     import uuid
 
@@ -3791,7 +3972,8 @@ async def _capture_finish_photo_from_timelapse(
 
     logger = logging.getLogger(__name__)
 
-    deadline = asyncio.get_event_loop().time() + _FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS
+    budget = _FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS if timeout is None else timeout
+    deadline = asyncio.get_event_loop().time() + budget
     poll_interval = _FINISH_PHOTO_TIMELAPSE_POLL_INTERVAL_SECONDS
 
     while True:
@@ -3814,25 +3996,71 @@ async def _capture_finish_photo_from_timelapse(
                         video_path.name,
                         archive_id,
                     )
-                    return filename
+                    return filename, False
                 logger.warning(
                     "[PHOTO-BG] Timelapse %s landed but last-frame extraction failed for archive %s; falling back",
                     video_path.name,
                     archive_id,
                 )
-                return None
+                return None, False
 
         if asyncio.get_event_loop().time() >= deadline:
             logger.info(
                 "[PHOTO-BG] Timelapse for archive %s didn't land within %.0fs; falling back to live camera",
                 archive_id,
-                _FINISH_PHOTO_TIMELAPSE_POLL_TIMEOUT_SECONDS,
+                budget,
             )
-            return None
+            return None, True
 
         await asyncio.sleep(poll_interval)
 
 
+async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Path) -> None:
+    """Add the timelapse's last frame to an archive after the fact (#2704).
+
+    The print-complete notification waits only ~60s for the video, because
+    holding a notification for minutes is worse than sending it with a live
+    camera grab. On a P1-series printer the video often lands well after that,
+    so the archive used to be stuck with the live grab — which is taken at
+    ``gcode_state=FINISH``, after the end G-code has dropped the bed, and is
+    the worse photo of the two.
+
+    This keeps waiting in the background and, when the video arrives, extracts
+    the frame and puts it *first* in the archive's photo list, so opening the
+    gallery shows it. The live grab is deliberately kept: the notification that
+    already went out links to that exact file, and deleting it would leave a
+    broken image in Discord or Telegram.
+    """
+    logger = logging.getLogger(__name__)
+
+    filename, _ = await _capture_finish_photo_from_timelapse(
+        archive_id, archive_dir, timeout=_FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS
+    )
+    if not filename:
+        logger.info("[PHOTO-UPGRADE] No timelapse frame for archive %s; keeping the live grab", archive_id)
+        return
+
+    try:
+        async with async_session() as db:
+            from backend.app.models.archive import PrintArchive
+
+            archive = await db.get(PrintArchive, archive_id)
+            if archive is None:
+                return
+            photos = list(archive.photos or [])
+            if filename in photos:
+                return
+            # Front of the list: PhotoGalleryModal opens at index 0.
+            archive.photos = [filename, *photos]
+            await db.commit()
+    except Exception as e:
+        logger.warning("[PHOTO-UPGRADE] Failed to attach upgraded photo to archive %s: %s", archive_id, e)
+        return
+
+    logger.info("[PHOTO-UPGRADE] Archive %s now leads with the timelapse frame %s", archive_id, filename)
+    await ws_manager.send_archive_updated({"id": archive_id, "photo_added": filename})
+
+
 async def on_print_running_observed(printer_id: int, data: dict):
     """Restart-recovery: capture a fresh timelapse baseline for a print that
     started before Bambuddy came up.
@@ -5029,8 +5257,9 @@ async def on_print_complete(printer_id: int, data: dict):
                 printer.external_camera_enabled and printer.external_camera_url
             )
 
+            timelapse_still_pending = False
             if prefer_timelapse_source:
-                photo_filename = await _capture_finish_photo_from_timelapse(
+                photo_filename, timelapse_still_pending = await _capture_finish_photo_from_timelapse(
                     archive_id=archive_id,
                     archive_dir=archive_dir,
                 )
@@ -5134,8 +5363,28 @@ async def on_print_complete(printer_id: int, data: dict):
                         arch.photos = photos
                         await db.commit()
                 logger.info("[PHOTO-BG] Saved: %s", photo_filename)
-                return photo_filename
-            return None
+
+            # The short wait above is bounded so a slow printer can't hold up
+            # the print-complete notification, which is what the caller is
+            # blocking on. When it ran out with the video still on its way,
+            # keep waiting off to the side and add the better frame to the
+            # archive once it arrives (#2704 follow-up) — otherwise P1-series
+            # users, whose videos routinely take minutes to transfer, never get
+            # the pre-bed-drop framing this path exists to provide.
+            #
+            # Spawned here rather than at the point the wait gave up: both this
+            # function and the upgrade do a read-modify-write on `photos`, and
+            # the live-camera fallback above can take tens of seconds. Starting
+            # the upgrade before that write means the two can interleave and one
+            # silently drops the other's entry, leaving a JPEG on disk that the
+            # gallery never lists.
+            if timelapse_still_pending:
+                spawn_background_task(
+                    _upgrade_finish_photo_from_timelapse(archive_id, archive_dir),
+                    name=f"finish-photo-upgrade-{archive_id}",
+                )
+
+            return photo_filename
         except Exception as e:
             logger.warning("[PHOTO-BG] Failed: %s", e)
             return None

backend/app/models/archive.py

@@ -32,6 +32,16 @@ class PrintArchive(Base):
     # both locally and on the printer's SD after extraction — the user
     # didn't opt in to a timelapse recording.
     bambuddy_forced_timelapse: Mapped[bool] = mapped_column(Boolean, default=False, server_default="0")
+    # Video filenames present in the printer's /timelapse directory when this
+    # print started (#2704). The printer writes its video only at print end, so
+    # anything not in this list belongs to this print — a comparison that needs
+    # no clock, which matters because a LAN-only printer can't reach Bambu's NTP
+    # server and its filename timestamps are arbitrarily wrong. Persisted (not
+    # just held in memory) so the diff survives a restart and so the manual
+    # "Scan for Timelapse" button can use it instead of guessing from
+    # timestamps. NULL for archives predating this, and for baselines taken at
+    # completion time, which are useless by construction.
+    timelapse_baseline: Mapped[list | None] = mapped_column(JSON, nullable=True)
     source_3mf_path: Mapped[str | None] = mapped_column(String(500))  # Original project 3MF from slicer
     f3d_path: Mapped[str | None] = mapped_column(String(500))  # Fusion 360 design file
 

backend/app/services/bambu_ftp.py

@@ -353,18 +353,43 @@ class BambuFTPClient:
 
         return files
 
-    def download_file(self, remote_path: str) -> bytes | None:
-        """Download a file from the printer."""
+    def download_file(self, remote_path: str, expected_size: int | None = None) -> bytes | None:
+        """Download a file from the printer.
+
+        ``expected_size`` is the byte count the directory listing reported for
+        this file. Pass it whenever a short read must not be mistaken for a
+        successful download: an FTPS data connection that closes early does
+        not always raise, so ``retrbinary`` can hand back a partial buffer that
+        looks like a perfectly good file to everything downstream. That is
+        tolerable when the printer keeps its copy, and not tolerable when the
+        caller goes on to delete the source (#2704).
+
+        A zero-byte result is always treated as a failure, matching
+        :meth:`download_to_file` — no caller has a use for an empty file.
+        """
         if not self._ftp:
             return None
 
         try:
             buffer = BytesIO()
             self._ftp.retrbinary(f"RETR {remote_path}", buffer.write)
-            return buffer.getvalue()
+            data = buffer.getvalue()
         except (OSError, ftplib.Error):
             return None
 
+        if not data:
+            logger.warning("FTP download returned 0 bytes for %s", remote_path)
+            return None
+        if expected_size is not None and len(data) != expected_size:
+            logger.warning(
+                "FTP download of %s is short: got %s bytes, listing reported %s — treating as failed",
+                remote_path,
+                len(data),
+                expected_size,
+            )
+            return None
+        return data
+
     def download_to_file(self, remote_path: str, local_path: Path) -> bool:
         """Download a file from the printer to local filesystem."""
         if not self._ftp:
@@ -1301,6 +1326,7 @@ async def download_file_bytes_async(
     socket_timeout: float | None = None,
     printer_model: str | None = None,
     timeout: float = 300.0,
+    expected_size: int | None = None,
 ) -> bytes | None:
     """Async wrapper for downloading file as bytes.
 
@@ -1313,6 +1339,9 @@ async def download_file_bytes_async(
             video, gcode) which can legitimately take minutes over slow Wi-Fi —
             the cap only guards against a permanently-starved pool, not a
             slow-but-progressing transfer.
+        expected_size: size from the directory listing; a mismatch fails the
+            download instead of returning a truncated file. See
+            :meth:`BambuFTPClient.download_file`.
     """
     loop = asyncio.get_event_loop()
 
@@ -1320,7 +1349,7 @@ async def download_file_bytes_async(
         client = BambuFTPClient(ip_address, access_code, timeout=socket_timeout, printer_model=printer_model)
         if client.connect():
             try:
-                return client.download_file(remote_path)
+                return client.download_file(remote_path, expected_size=expected_size)
             finally:
                 client.disconnect()
         return None
@@ -1332,6 +1361,121 @@ async def download_file_bytes_async(
         return None
 
 
+async def remote_file_settled(
+    ip_address: str,
+    access_code: str,
+    remote_path: str,
+    downloaded_bytes: int,
+    *,
+    printer_model: str | None = None,
+) -> bool:
+    """Confirm the printer has finished writing the file we just downloaded.
+
+    Matching the download against the size from the directory listing proves we
+    received what the listing *said*, not that the file was *finished*. The
+    timelapse scan's first look happens seconds after the print ends, which is
+    exactly when the printer is writing the video — so a file still growing can
+    be listed at a partial size, served at that size, and pass the length check
+    as a complete video (#2704).
+
+    That was survivable while the printer kept its copy. It isn't now that a
+    successful attach deletes the source, so re-list afterwards: if the file has
+    grown, what we hold is a prefix and the caller should discard it and try
+    again on the next round.
+
+    Returns True when the remote file can no longer differ from what we hold —
+    the size still matches, or the file is gone from the listing entirely and
+    so cannot grow any further. Returns False when it has changed size, and on
+    a listing failure, because "we could not check" must not read as "safe to
+    delete".
+    """
+    directory, _, name = remote_path.rpartition("/")
+    files = await list_files_async(ip_address, access_code, directory or "/", printer_model=printer_model)
+    if not files:
+        logger.warning("[TIMELAPSE] Could not re-list %s to confirm %s is complete", directory or "/", name)
+        return False
+
+    for f in files:
+        if f.get("name") == name:
+            size = f.get("size")
+            if size == downloaded_bytes:
+                return True
+            logger.info(
+                "[TIMELAPSE] %s is still being written (%s bytes now, %s when downloaded) — will retry",
+                name,
+                size,
+                downloaded_bytes,
+            )
+            return False
+
+    # Vanished between the download and now. Nothing left that could grow, and
+    # nothing left to delete either.
+    logger.debug("[TIMELAPSE] %s is no longer on the printer after download", name)
+    return True
+
+
+async def delete_archived_timelapse(
+    ip_address: str,
+    access_code: str,
+    remote_path: str,
+    *,
+    verified: bool,
+    printer_model: str | None = None,
+    printer_name: str = "",
+) -> bool:
+    """Remove a timelapse from the printer once it is safely in the archive.
+
+    Call this only after the attach succeeded (#2704). Keeping ``/timelapse``
+    down to just the unclaimed videos is what makes the snapshot diff
+    unambiguous rather than merely usually-right, and it stops P1S cards
+    filling with AVIs.
+
+    ``verified`` must say whether the downloaded byte count was checked against
+    the size the directory listing reported. It is required rather than
+    defaulted because this is the one irreversible step in the flow: an FTPS
+    data connection that closes early does not always raise, so an unverified
+    transfer can be a partial file that looks complete, and deleting the source
+    would then destroy the only good copy. The check lives here rather than at
+    each call site so no future caller can omit it.
+
+    Best-effort otherwise: a printer that refuses the delete keeps its copy, the
+    diff still excludes that filename next time because it is attached to an
+    archive, and nothing else in the flow cares. Returns True only on an actual
+    delete or a 550 (already gone).
+    """
+    if not verified:
+        logger.warning(
+            "[TIMELAPSE] Not deleting %s from printer %s: the download was never size-checked",
+            remote_path,
+            printer_name,
+        )
+        return False
+
+    for attempt in range(1, 4):
+        try:
+            result = await delete_file_async(ip_address, access_code, remote_path, printer_model=printer_model)
+        except Exception as e:
+            result = DeleteResult.FAILED
+            logger.warning("[TIMELAPSE] Delete attempt %d/3 raised for %s: %s", attempt, remote_path, e)
+
+        if result == DeleteResult.DELETED:
+            logger.info("[TIMELAPSE] Deleted %s from printer %s after archiving", remote_path, printer_name)
+            return True
+        if result == DeleteResult.NOT_FOUND:
+            # 550 never recovers by waiting — the printer already cleaned up.
+            logger.debug("[TIMELAPSE] %s already gone from printer %s", remote_path, printer_name)
+            return True
+        if attempt < 3:
+            await asyncio.sleep(2)
+
+    logger.warning(
+        "[TIMELAPSE] Could not delete %s from printer %s (it stays on the card; the archive copy is unaffected)",
+        remote_path,
+        printer_name,
+    )
+    return False
+
+
 async def get_storage_info_async(
     ip_address: str,
     access_code: str,

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/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_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