Timelapse: a baseline off an unreadable card no longer attaches an old video

- The print-start baseline records whether the card was read: an empty
  listing is asked again, and only counts when every directory answered
  (a listing or 550). Stored as print_archives.timelapse_baseline_unread
- With an unread baseline the scan leaves several unclaimed videos on
  the printer instead of attaching, deleting and photographing the first
- The card is read again every 2 min while the same print runs; the
  printer writes the video only after the print ends
- After a restart, a stored baseline that was read beats an unread one
  taken in memory

CHANGELOG.md

@@ -205,6 +205,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **A timelapse baseline the printer's card could not give no longer attaches an old video to the print (#3412, reported by @ojimpo)** — When the card could not be read at print start (here during the FTPS handshake cool-off), Bambuddy recorded the empty listing as a card holding no videos. If the card was readable again by the end, every video still on the printer counted as new: the scan attached the first one to the print, deleted it from the printer, and took the archive's finish photo from its last frame. A baseline now records whether the card was actually read, including when a listing failed for another reason. With an unread one, several unclaimed videos are left on the printer for **Scan for Timelapse**, and a single one is still attached. The card is also read again every two minutes while the print runs, because the printer writes the video only after the print ends. Wiki updated.
 - **SpoolBuddy reads Bambu tags reliably on the first scan (#3377, reported and contributed by @hibikipr in #3378)** — A Bambu tag's sector reads sometimes failed on the first try (authentication not answered, or a block read that came back empty), and the scan then went out without its tray UUID. Each step is now retried before giving up, a block is accepted only when two reads agree and the reader flags no error, and a reader fault during the read no longer stops SpoolBuddy from polling: the tag is still reported, matched by its UID.
 - **A large tag catalog no longer pushes the File Manager's file list off the page (#3411, reported by @iammenasco)** — The tag filter showed every tag in the catalog as a chip, so 160 tags wrapped to about a dozen rows and left the file list a strip at the bottom. Active tags now always show first. The others take up to two rows, with **Show all (N)** opening them in a box that scrolls on its own. From 12 tags on, **Search tags** narrows them by name. Filtering works as before. Wiki updated.
 - **A printer whose SD card or USB drive refuses the file no longer eats the queue (#3409, reported by @simonkreuchauff-oss)** — When a printer refused an upload with 553 or 552 (storage missing, full or unusable), or rejected the access code, the job failed. The printer was still idle, so the next job went to it and failed the same way: an H2D with an unusable USB stick put seven jobs in History within an hour.

backend/app/core/database.py

@@ -5896,6 +5896,15 @@ async def _run_migrations(conn):
             {"off": False},
         )
 
+    # Migration: mark a timelapse baseline taken off an unreadable card
+    # (#3412). Off for every existing row, which is what they were treated as.
+    if is_sqlite():
+        await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN timelapse_baseline_unread BOOLEAN DEFAULT 0")
+    else:
+        await _safe_execute(
+            conn, "ALTER TABLE print_archives ADD COLUMN timelapse_baseline_unread BOOLEAN DEFAULT FALSE"
+        )
+
     await _migrate_auth_datetimes_to_timestamp(conn)
     await _migrate_add_missing_set_null_fks(conn)
     await _migrate_wallet_transactions_keep_deleted_users(conn)
@@ -5957,6 +5966,7 @@ _PG_COLUMNS_SET_FROM_MODEL = (
     ("oidc_providers", "is_autologin"),
     ("oidc_providers", "is_env_managed"),
     ("print_archives", "bambuddy_forced_timelapse"),
+    ("print_archives", "timelapse_baseline_unread"),
     ("print_archives", "wallet_charge_skipped"),
     ("print_queue", "bed_levelling"),
     ("print_queue", "cleanup_library_after_dispatch"),

backend/app/main.py

@@ -562,6 +562,9 @@ _hms_forget_when_gone: dict[int, set[str]] = {}
 # Track timelapse file baselines at print start: {printer_id: set of video filenames}
 # Used for snapshot-diff detection at print completion
 _timelapse_baselines: dict[int, set[str]] = {}
+# Printers whose baseline above was taken off a card that could not be read,
+# so it is empty without the card being empty (#3412).
+_timelapse_baselines_unread: set[int] = set()
 
 # Track printers waiting for bed to cool after print completion.
 # Event-driven: fires when bed_temper arrives via MQTT below threshold.
@@ -1263,6 +1266,7 @@ async def forget_printer(printer_id: int) -> None:
         _printer_offline_published,
         _nozzle_count_updated,
         _runtime_seen_connected,
+        _timelapse_baselines_unread,
     ):
         marked.discard(printer_id)
 
@@ -6333,6 +6337,17 @@ async def on_print_start(printer_id: int, data: dict):
 
 
 _TIMELAPSE_VIDEO_EXTENSIONS = (".mp4", ".avi")
+_TIMELAPSE_DIRECTORIES = ("/timelapse", "/timelapse/video", "/record", "/recording")
+
+# A baseline that could not be read at print start is read again on this
+# schedule while the print runs (#3412). The printer writes the video only once
+# the print has ended, so a reading taken mid-print is as good as one taken at
+# the start. The interval is shorter than the FTPS handshake cool-off (300 s),
+# which is what usually makes the card unreadable, so the first reading after
+# it ends is not long behind; while the cool-off runs a round costs nothing, as
+# no connection is opened. The cap bounds a print that never reports an end.
+_TIMELAPSE_BASELINE_REREAD_INTERVAL_SECONDS = 120.0
+_TIMELAPSE_BASELINE_REREAD_MAX_ATTEMPTS = 720
 
 # Poll schedule for the post-print timelapse scan (#2704). Module-level so
 # tests can shrink them without waiting out real delays.
@@ -6391,26 +6406,55 @@ async def _claimed_timelapse_names(db, printer_id: int, exclude_archive_id: int)
     return {Path(p).stem for p in rows.scalars().all() if p}
 
 
-def _timelapse_listing_is_trustworthy(printer) -> bool:
-    """Whether an *empty* timelapse listing for *printer* can be believed.
+async def _read_timelapse_baseline(printer) -> tuple[set[str], bool]:
+    """The video names on *printer*'s card, and whether the card was read.
 
-    ``list_files_async`` answers ``[]`` when its connect fails rather than
-    raising, so a card behind the FTPS handshake cool-off is indistinguishable
-    from one holding no videos. Everywhere that only wants to know "is there a
-    video yet" the difference does not matter — both mean "not yet, retry".
+    ``_list_timelapse_videos`` answers ``[]`` when its connect fails rather
+    than raising, so a card behind the FTPS handshake cool-off, or one whose
+    listing timed out, is indistinguishable from one holding no videos.
+    Everywhere that only wants to know "is there a video yet" the difference
+    does not matter -- both mean "not yet, retry".
 
     It matters where an empty listing is recorded as a *baseline*. Recording
     "the card held nothing" for a card that was never read means every video on
-    it counts as new once the cool-off expires, and the completion scan then
-    attaches a stale video to this print and deletes it from the printer
-    (#2957 follow-up). Those two callers ask this first.
+    it counts as new, and the completion scan then attaches a stale video to
+    this print and deletes it from the printer (#2957 follow-up, #3412). So an
+    empty listing is asked again here in a way that tells the two apart: the
+    card counts as read only when every directory walked answered, with a
+    listing or with "no such directory". One that failed may be the one
+    holding the videos.
     """
-    from backend.app.services.bambu_ftp import ftps_handshake_blocked
+    from backend.app.services.bambu_ftp import ftps_handshake_blocked, list_files_result_async
+
+    files, _ = await _list_timelapse_videos(printer)
+    if files:
+        return {f.get("name", "") for f in files}, True
 
     ip_address = getattr(printer, "ip_address", None)
+    printer_id = getattr(printer, "id", None)
     if not ip_address:
-        return True
-    return not ftps_handshake_blocked(ip_address)
+        return set(), True
+    # No card in the slot: nothing to read, and nothing a video could land on.
+    if printer_id is not None and not external_storage_present(printer_manager.get_status(printer_id)):
+        return set(), True
+
+    for timelapse_path in _TIMELAPSE_DIRECTORIES:
+        if ftps_handshake_blocked(ip_address):
+            # Below any path: every remaining one would fail the same way.
+            return set(), False
+        listing = await list_files_result_async(
+            ip_address, printer.access_code, timelapse_path, printer_model=printer.model
+        )
+        if not (listing.available or listing.missing):
+            return set(), False
+        videos = [
+            f
+            for f in listing.files
+            if not f.get("is_directory") and f.get("name", "").lower().endswith(_TIMELAPSE_VIDEO_EXTENSIONS)
+        ]
+        if videos:
+            return {f.get("name", "") for f in videos}, True
+    return set(), True
 
 
 async def _list_timelapse_videos(printer) -> tuple[list[dict], str | None]:
@@ -6421,8 +6465,8 @@ async def _list_timelapse_videos(printer) -> tuple[list[dict], str | None]:
     and found_path is the directory where they were found, or ([], None).
 
     An empty return does not distinguish "no videos" from "could not read the
-    card" — see :func:`_timelapse_listing_is_trustworthy`, which the two
-    baseline callers consult before believing one.
+    card" — see :func:`_read_timelapse_baseline`, which the two baseline
+    callers use instead.
     """
     from backend.app.services.bambu_ftp import list_files_async
 
@@ -6441,7 +6485,7 @@ async def _list_timelapse_videos(printer) -> tuple[list[dict], str | None]:
         logger.debug("[TIMELAPSE] Skipping the scan for printer %s: it reports no external storage", printer_id)
         return [], None
 
-    for timelapse_path in ["/timelapse", "/timelapse/video", "/record", "/recording"]:
+    for timelapse_path in _TIMELAPSE_DIRECTORIES:
         try:
             found_files = await list_files_async(
                 printer.ip_address, printer.access_code, timelapse_path, printer_model=printer.model
@@ -6484,25 +6528,28 @@ async def _capture_timelapse_baseline_at_start(
     completion already contains the new video and would poison a later scan.
     """
     names: set[str] | None = None
+    unread = False
+    _timelapse_baselines_unread.discard(printer_id)
     try:
-        if not _timelapse_listing_is_trustworthy(printer):
-            # Recorded anyway, deliberately. An empty baseline taken off a card
-            # we could not read is not authoritative, but it is still the right
-            # *default*: Bambuddy deletes each video from the printer once it is
-            # attached, so the usual card holds exactly one video at completion
-            # and an empty baseline resolves it correctly. Persisting NULL
+        names, read = await _read_timelapse_baseline(printer)
+        unread = not read
+        _timelapse_baselines[printer_id] = names
+        if unread:
+            # Recorded anyway, deliberately, and marked. An empty baseline taken
+            # off a card we could not read is not evidence, but it is still the
+            # right *default*: Bambuddy deletes each video from the printer once
+            # it is attached, so the usual card holds exactly one video at
+            # completion and an empty baseline resolves it. Persisting NULL
             # instead would send completion to take its own snapshot, by which
             # point this print's video is on the card and would be swallowed by
-            # it. The ambiguity is handled where it actually bites — see
-            # ``require_unambiguous`` in the scan (#2957 follow-up).
+            # it. The mark keeps the scan from choosing between several videos
+            # (#3412), and the card is read again while the print runs.
+            _timelapse_baselines_unread.add(printer_id)
             logger.warning(
-                "[TIMELAPSE] Baseline for printer %s taken while its file service is in the FTPS "
-                "handshake cool-off, so the card could not be read — treating it as empty",
+                "[TIMELAPSE] Baseline for printer %s taken while its card could not be read — "
+                "treating it as empty for now and reading it again during the print",
                 printer_id,
             )
-        baseline_files, _ = await _list_timelapse_videos(printer)
-        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(names),
@@ -6511,29 +6558,145 @@ async def _capture_timelapse_baseline_at_start(
     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
+    run_started_at: datetime | None = None
+    if archive_id is not None:
+        try:
+            async with async_session() as db:
+                from backend.app.models.archive import PrintArchive
 
-            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)
+                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
+                    archive.timelapse_baseline_unread = unread
+                    run_started_at = archive.started_at
+                    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 unread and names is not None:
+        spawn_background_task(
+            _reread_unread_timelapse_baseline(printer_id, names, archive_id, run_started_at),
+            name=f"reread-timelapse-baseline-{printer_id}",
+        )
+
+
+# Before the print has ended there is no video yet: a long PREPARE (heating,
+# levelling) counts as much as the print itself.
+_BASELINE_REREAD_STATES = _ACTIVE_PRINT_STATES | {"PREPARE", "SLICING"}
+
+
+def _schedule_timelapse_scan(archive_id: int, printer_id: int, *, plate_id: int | None) -> None:
+    """Start the completion scan with the print-start baseline: the names, and
+    whether they were taken off a card that could not be read. Taking them
+    ends any re-read still running for it."""
+    unread = printer_id in _timelapse_baselines_unread
+    _timelapse_baselines_unread.discard(printer_id)
+    baseline = _timelapse_baselines.pop(printer_id, None)
+    spawn_background_task(
+        _scan_for_timelapse_with_retries(archive_id, baseline, plate_id=plate_id, baseline_unread=unread),
+        name=f"scan-timelapse-{archive_id}",
+    )
+
+
+def _unread_baseline_round(printer_id: int, names: set[str]) -> bool | None:
+    """Whether to read an unread baseline again now: True to read, None to
+    skip this round, False once it is over -- the print it was taken for is
+    no longer the one on the printer, has stopped printing, or the baseline
+    has been read.
+
+    ``names`` is the very set object recorded at print start: completion pops
+    it and the next print records a new one, so identity marks the run. A
+    printer that is momentarily disconnected says nothing about its print, so
+    that only skips the round.
+    """
+    if _timelapse_baselines.get(printer_id) is not names or printer_id not in _timelapse_baselines_unread:
+        return False
+    state = printer_manager.get_status(printer_id)
+    if state is None or not state.connected:
+        return None
+    return state.state in _BASELINE_REREAD_STATES
+
+
+async def _reread_unread_timelapse_baseline(
+    printer_id: int, names: set[str], archive_id: int | None, run_started_at: datetime | None
+) -> None:
+    """Read again a print-start baseline the card could not give (#3412).
+
+    The printer writes the video only once the print has ended, so the card
+    read at any point during the print holds exactly what it held at the
+    start. Each round stops once the print is no longer the one the baseline
+    was taken for or has stopped printing, and checks that again after the
+    listing returns, before anything is recorded.
+    """
+    from backend.app.models.archive import PrintArchive
+    from backend.app.models.printer import Printer
+
+    logger = logging.getLogger(__name__)
+    for _ in range(_TIMELAPSE_BASELINE_REREAD_MAX_ATTEMPTS):
+        await asyncio.sleep(_TIMELAPSE_BASELINE_REREAD_INTERVAL_SECONDS)
+        now = _unread_baseline_round(printer_id, names)
+        if now is False:
+            return
+        if now is None:
+            continue
+        try:
+            async with async_session() as db:
+                printer = await db.get(Printer, printer_id)
+            if printer is None:
+                return
+            fresh, read = await _read_timelapse_baseline(printer)
+        except Exception as e:
+            logger.debug("[TIMELAPSE] Re-reading the baseline for printer %s failed: %s", printer_id, e)
+            continue
+        if not read:
+            continue
+        now = _unread_baseline_round(printer_id, names)
+        if now is False:
+            return
+        if now is None:
+            # The printer dropped off while the listing was out, so nothing
+            # says it did not end meanwhile: keep the reading for a round that
+            # can vouch for it.
+            continue
+
+        names.update(fresh)
+        _timelapse_baselines_unread.discard(printer_id)
+        logger.info(
+            "[TIMELAPSE] Baseline for printer %s read during the print: %s video files",
+            printer_id,
+            len(fresh),
+        )
+        if archive_id is None:
+            return
+        try:
+            async with async_session() as db:
+                archive = await db.get(PrintArchive, archive_id)
+                # Only the run it was taken for: by now the row can belong to a
+                # reprint, whose baseline is its own.
+                if archive is not None and archive.started_at == run_started_at and archive.timelapse_baseline_unread:
+                    archive.timelapse_baseline = sorted(names)
+                    archive.timelapse_baseline_unread = False
+                    await db.commit()
+        except Exception as e:
+            # The in-memory baseline is read now, which covers completion
+            # unless Bambuddy restarts first.
+            logger.warning("[TIMELAPSE] Failed to persist the re-read baseline for archive %s: %s", archive_id, e)
+        return
 
 
 async def _scan_for_timelapse_with_retries(
-    archive_id: int, baseline_names: set[str] | None = None, *, plate_id: int | None = None
+    archive_id: int,
+    baseline_names: set[str] | None = None,
+    *,
+    plate_id: int | None = None,
+    baseline_unread: bool = False,
 ):
     """Poll the printer for this print's timelapse and attach it.
 
@@ -6552,11 +6715,14 @@ async def _scan_for_timelapse_with_retries(
 
     On success the video is deleted from the printer, which keeps ``/timelapse``
     down to the unclaimed files and makes the next diff unambiguous.
+
+    ``baseline_unread`` marks the caller's baseline as taken off a card that
+    could not be read; a stored one carries the same mark on the archive.
     """
     logger = logging.getLogger(__name__)
 
-    # Cleared when the baseline had to be taken off a card we could not read, so
-    # the attach step refuses to choose between several candidates (#2957).
+    # Cleared when the baseline was taken off a card we could not read, so the
+    # attach step refuses to choose between several candidates (#2957, #3412).
     baseline_trusted = True
 
     # --- Phase 1: establish the baseline -------------------------------------
@@ -6581,7 +6747,23 @@ async def _scan_for_timelapse_with_retries(
             # video is its own.
             run_started_at = archive.started_at
 
-            if baseline_names is not None:
+            if (
+                baseline_names is not None
+                and baseline_unread
+                and archive.timelapse_baseline is not None
+                and not archive.timelapse_baseline_unread
+            ):
+                # The card was read at print start, and the in-memory copy is
+                # one Bambuddy took after a restart off a card it could not
+                # read: the stored one is this run's and is evidence.
+                baseline_names = set(archive.timelapse_baseline)
+                logger.info(
+                    "[TIMELAPSE] Using stored baseline over an unread one: %s existing video files for archive %s",
+                    len(baseline_names),
+                    archive_id,
+                )
+            elif baseline_names is not None:
+                baseline_trusted = not baseline_unread
                 logger.info(
                     "[TIMELAPSE] Using print-start baseline: %s existing video files for archive %s",
                     len(baseline_names),
@@ -6590,6 +6772,7 @@ async def _scan_for_timelapse_with_retries(
             elif archive.timelapse_baseline is not None:
                 # Persisted at print start — survives a restart mid-print.
                 baseline_names = set(archive.timelapse_baseline)
+                baseline_trusted = not archive.timelapse_baseline_unread
                 logger.info(
                     "[TIMELAPSE] Using stored baseline: %s existing video files for archive %s",
                     len(baseline_names),
@@ -6602,25 +6785,22 @@ async def _scan_for_timelapse_with_retries(
                     logger.warning("[TIMELAPSE] Printer not found for archive %s, aborting", archive_id)
                     return
 
-                if not _timelapse_listing_is_trustworthy(printer):
+                baseline_names, baseline_trusted = await _read_timelapse_baseline(printer)
+                if not baseline_trusted:
                     # The card is unreadable at the one moment a baseline has to
-                    # be taken, so the empty listing below means "we never
-                    # looked", not "these are all new". Carry on with it anyway
-                    # — the usual card holds exactly one video, which resolves
+                    # be taken, so the empty listing means "we never looked",
+                    # not "these are all new". Carry on with it anyway — the
+                    # usual card holds exactly one video, which resolves
                     # correctly — but stop the poll from *choosing* between
                     # several, which is how a stale video got attached to this
                     # print and then deleted off the printer (#2957 follow-up).
-                    baseline_trusted = False
                     logger.warning(
-                        "[TIMELAPSE] Baseline for archive %s taken while printer %s is in the FTPS "
-                        "handshake cool-off. A single new video still resolves; several will not be "
-                        "guessed between — use Scan for Timelapse to pick one by hand",
+                        "[TIMELAPSE] Baseline for archive %s taken while printer %s's card could not be "
+                        "read. A single new video still resolves; several will not be guessed between — "
+                        "use Scan for Timelapse to pick one by hand",
                         archive_id,
                         archive.printer_id,
                     )
-
-                baseline_files, _ = await _list_timelapse_videos(printer)
-                baseline_names = {f.get("name", "") for f in baseline_files}
                 logger.info(
                     "[TIMELAPSE] Baseline snapshot (fallback): %s existing video files for archive %s",
                     len(baseline_names),
@@ -9945,11 +10125,7 @@ async def _close_out_print(printer_id: int, data: dict):
         logger.info("[TIMELAPSE] Timelapse was active during print, scheduling auto-scan for archive %s", archive_id)
         # Schedule timelapse scan as background task with retries
         # The printer needs time to encode the video after print completion
-        baseline = _timelapse_baselines.pop(printer_id, None)
-        spawn_background_task(
-            _scan_for_timelapse_with_retries(archive_id, baseline, plate_id=notify_plate_id),
-            name=f"scan-timelapse-{archive_id}",
-        )
+        _schedule_timelapse_scan(archive_id, printer_id, plate_id=notify_plate_id)
         log_timing("Timelapse scan scheduled")
 
     logger.info("[CALLBACK] on_print_complete finished for printer %s, archive %s", printer_id, archive_id)

backend/app/models/archive.py

@@ -53,6 +53,12 @@ class PrintArchive(Base):
     # 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)
+    # The baseline above was taken off a card that could not be read, so it
+    # says nothing about what was on it. The completion scan then attaches a
+    # video only when exactly one is unclaimed, rather than picking between
+    # several (#3412). Read only alongside a non-NULL baseline, and written by
+    # everything that writes one.
+    timelapse_baseline_unread: Mapped[bool] = mapped_column(Boolean, default=False, server_default=false())
     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

@@ -286,6 +286,9 @@ class FileListResult:
     # The printer did not answer within the wait: slow, not necessarily gone
     # (a large /ipcam on a slow card), so not reported as unavailable (TO-C13).
     timed_out: bool = False
+    # The printer answered that the directory does not exist (550). Still not
+    # ``available``, but an answer about the card, not a failure to reach it.
+    missing: bool = False
 
 
 # Listings still running, by printer and folder: a refresh joins the one in
@@ -2180,6 +2183,8 @@ async def list_files_result_async(
             connected.set()
             try:
                 return FileListResult(files=client.list_files(path, raise_on_error=True), available=True)
+            except ftplib.error_perm as e:
+                return FileListResult(files=[], available=False, missing=str(e).startswith("550"))
             except (OSError, ftplib.Error):
                 return FileListResult(files=[], available=False)
             finally:
@@ -2203,7 +2208,7 @@ async def list_files_result_async(
         logger.warning("FTP list_files timed out after %ss for %s", timeout, path)
         return FileListResult(files=[], available=False, timed_out=connected.is_set())
     # Joined callers share the listing; each gets entries of its own to edit.
-    return FileListResult(files=[dict(f) for f in result.files], available=result.available)
+    return FileListResult(files=[dict(f) for f in result.files], available=result.available, missing=result.missing)
 
 
 async def find_remote_file_async(

backend/tests/unit/services/test_ftp_list_timeout_state.py

@@ -7,6 +7,7 @@ started another one in the shared FTP pool.
 """
 
 import asyncio
+import ftplib
 import threading
 from unittest.mock import patch
 
@@ -95,3 +96,37 @@ async def test_a_printer_that_never_answered_is_unavailable_not_slow():
 
     assert result.available is False
     assert result.timed_out is False
+
+
+class Refuses:
+    """Connects, then refuses the directory with *reply*."""
+
+    reply = ""
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def connect(self):
+        return True
+
+    def list_files(self, path, raise_on_error=False):
+        raise ftplib.error_perm(type(self).reply)
+
+    def disconnect(self):
+        pass
+
+
+@pytest.mark.parametrize(
+    ("reply", "missing"),
+    [("550 Failed to change directory.", True), ("530 Login incorrect.", False)],
+)
+async def test_a_directory_that_does_not_exist_is_an_answer_about_the_card(reply, missing):
+    """550 says the card has no such directory, which the timelapse baseline
+    counts as reading the card (#3412); any other refusal says nothing."""
+    bambu_ftp._inflight_listings.clear()
+    Refuses.reply = reply
+    with patch.object(bambu_ftp, "BambuFTPClient", Refuses):
+        result = await bambu_ftp.list_files_result_async("10.0.0.6", "code", "/record", timeout=2)
+
+    assert result.available is False
+    assert result.missing is missing

backend/tests/unit/test_archive_filtering.py

@@ -11,6 +11,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from backend.app.core.printer_scope import ALL_PRINTERS
+from backend.app.services.bambu_ftp import FileListResult
 
 # Patch paths for lazy imports inside functions
 _FTP_MODULE = "backend.app.services.bambu_ftp"
@@ -320,6 +321,7 @@ class TestScanForTimelapseWithRetries:
         with (
             patch("backend.app.main.async_session", return_value=mock_session),
             patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
+            patch(f"{_FTP_MODULE}.list_files_result_async", AsyncMock(return_value=FileListResult([], True))),
             patch("backend.app.main.ws_manager") as mock_ws,
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
@@ -365,6 +367,7 @@ class TestScanForTimelapseWithRetries:
         with (
             patch("backend.app.main.async_session", return_value=mock_session),
             patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
+            patch(f"{_FTP_MODULE}.list_files_result_async", AsyncMock(return_value=FileListResult([], True))),
             patch("backend.app.main.ws_manager") as mock_ws,
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
@@ -417,6 +420,7 @@ class TestScanForTimelapseWithRetries:
         with (
             patch("backend.app.main.async_session", return_value=mock_session),
             patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
+            patch(f"{_FTP_MODULE}.list_files_result_async", AsyncMock(return_value=FileListResult([], True))),
             patch("backend.app.main.ws_manager") as mock_ws,
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),
@@ -509,6 +513,7 @@ class TestScanForTimelapseWithRetries:
         with (
             patch("backend.app.main.async_session", return_value=mock_session),
             patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
+            patch(f"{_FTP_MODULE}.list_files_result_async", AsyncMock(return_value=FileListResult([], True))),
             patch("backend.app.main.ws_manager") as mock_ws,
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock) as mock_sleep,
             patch("backend.app.main.ArchiveService", return_value=mock_service),
@@ -637,6 +642,7 @@ class TestListTimelapseVideosAvi:
         with (
             patch("backend.app.main.async_session", return_value=mock_session),
             patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_videos),
+            patch(f"{_FTP_MODULE}.list_files_result_async", AsyncMock(return_value=FileListResult([], True))),
             patch("backend.app.main.ws_manager") as mock_ws,
             patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
             patch("backend.app.main.ArchiveService", return_value=mock_service),

backend/tests/unit/test_fallback_timelapse_baseline_2957.py

@@ -236,7 +236,11 @@ class TestTheBaselineCaptureItself:
             printer = await db.get(Printer, printer_id)
 
         try:
-            with _patches(main_module, maker, monkeypatch, tmp_path, lambda: []):
+            with (
+                _patches(main_module, maker, monkeypatch, tmp_path, lambda: []),
+                # The re-read it starts for an unread card is #3412's to test.
+                patch.object(main_module, "spawn_background_task", lambda coro, name=None: coro.close()),
+            ):
                 await main_module._capture_timelapse_baseline_at_start(
                     printer, printer_id, main_module.logging.getLogger(__name__), archive_id=archive_id
                 )
@@ -247,6 +251,7 @@ class TestTheBaselineCaptureItself:
             assert (await db.get(PrintArchive, archive_id)).timelapse_baseline == []
         assert main_module._timelapse_baselines[printer_id] == set()
         main_module._timelapse_baselines.pop(printer_id, None)
+        main_module._timelapse_baselines_unread.discard(printer_id)
 
     async def test_a_readable_card_is_captured_and_persisted(self, test_engine, tmp_path, monkeypatch):
         from backend.app import main as main_module

backend/tests/unit/test_forget_printer.py

@@ -76,6 +76,7 @@ def _entries(printer_id: int) -> dict[str, tuple]:
         "_hms_print_state": (printer_id, "RUNNING"),
         "_hms_forget_when_gone": (printer_id, {"0300_0001"}),
         "_timelapse_baselines": (printer_id, {"video.mp4"}),
+        "_timelapse_baselines_unread": (printer_id, None),
         "_bed_cool_waiters": (printer_id, {"threshold": 35.0, "filename": "cube", "registered_at": 1.0}),
         "_user_stopped_printers": (printer_id, None),
         "_kill_switch_stopped": (printer_id, None),

backend/tests/unit/test_timelapse_unread_baseline_3412.py

@@ -0,0 +1,510 @@
+"""A print-start baseline taken off a card that could not be read (#3412).
+
+The capture at print start knew the card was unreadable -- it logged so -- and
+then recorded the empty listing exactly like a card that held nothing, both in
+memory and on the archive row. The completion scan believed either copy, so
+with two old videos still on the printer it attached the first of them to this
+print, deleted it off the printer and took the finish photo from it. The
+``require_unambiguous`` guard from #2957 only covered a baseline the scan took
+itself at completion.
+
+Now the baseline carries whether the card was read; an unread one keeps the
+scan from choosing between several videos, and is read again while the print
+runs, since the printer writes the video only once the print has ended.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.printer import Printer
+from backend.app.services import bambu_ftp
+from backend.app.services.bambu_ftp import FileListResult
+
+pytestmark = pytest.mark.asyncio
+
+PRINTER_IP = "172.25.12.150"
+# Two videos of earlier prints that nothing claimed, as on the reporter's X2D.
+OLD_A = "video_2026-08-22_20-19-02_4.mp4"
+OLD_B = "video_2026-08-23_09-00-00.mp4"
+NEW_VIDEO = "video_2026-10-24_14-00-00.mp4"
+STARTED = datetime(2026, 10, 24, 11, 5, 1)
+
+
+def _entry(name: str) -> dict:
+    return {"name": name, "size": 1024, "is_directory": False, "path": f"/timelapse/{name}"}
+
+
+@pytest.fixture(autouse=True)
+def _forget_baselines():
+    from backend.app import main
+
+    yield
+    main._timelapse_baselines.clear()
+    main._timelapse_baselines_unread.clear()
+
+
+async def _seed(engine, *, baseline: list[str] | None = None, unread: bool = False):
+    maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    async with maker() as db:
+        printer = Printer(
+            name="X2D", serial_number="20P9AJ5A0700123", ip_address=PRINTER_IP, access_code="12345678", model="X2D"
+        )
+        db.add(printer)
+        await db.commit()
+        await db.refresh(printer)
+        archive = PrintArchive(
+            printer_id=printer.id,
+            filename="Bracket.gcode.3mf",
+            file_path="",
+            file_size=0,
+            print_name="Bracket",
+            status="printing",
+            started_at=STARTED,
+            timelapse_baseline=baseline,
+            timelapse_baseline_unread=unread,
+        )
+        db.add(archive)
+        await db.commit()
+        await db.refresh(archive)
+        return maker, printer, archive.id
+
+
+def _ftp(main_module, maker, monkeypatch, tmp_path, listing, *, state="RUNNING", others=None):
+    """Stand in for the FTP layer and the printer's live status.
+
+    ``listing()`` is what ``/timelapse`` holds at that moment, or None while
+    the card cannot be read -- what both listing calls then answer. The other
+    candidate directories answer "no such directory" unless ``others`` gives
+    them a FileListResult of their own.
+    """
+    from backend.app.core.config import settings as app_config
+    from backend.app.services import bambu_ftp
+
+    monkeypatch.setattr(app_config, "base_dir", tmp_path)
+    monkeypatch.setattr(app_config, "archive_dir", tmp_path / "archive")
+    monkeypatch.setattr(main_module, "_TIMELAPSE_SCAN_FIRST_DELAY_SECONDS", 0.01)
+    monkeypatch.setattr(main_module, "_TIMELAPSE_SCAN_POLL_INTERVAL_SECONDS", 0.01)
+    monkeypatch.setattr(main_module, "_TIMELAPSE_SCAN_TIMEOUT_SECONDS", 0.3)
+    monkeypatch.setattr(main_module, "_TIMELAPSE_BASELINE_REREAD_INTERVAL_SECONDS", 0.01)
+    monkeypatch.setattr(main_module, "_TIMELAPSE_BASELINE_REREAD_MAX_ATTEMPTS", 20)
+
+    async def _list(ip, code, path, printer_model=None):
+        return (listing() or []) if path == "/timelapse" else []
+
+    async def _list_result(ip, code, path, timeout=30.0, socket_timeout=None, printer_model=None):
+        if path != "/timelapse":
+            return (others or {}).get(path, FileListResult(files=[], available=False, missing=True))
+        files = listing()
+        return FileListResult(files=files or [], available=files is not None)
+
+    status = SimpleNamespace(connected=True, state=state, sdcard=True, sdcard_reported=True)
+    monkeypatch.setattr(bambu_ftp, "list_files_async", _list)
+    monkeypatch.setattr(bambu_ftp, "list_files_result_async", _list_result)
+    monkeypatch.setattr(bambu_ftp, "download_file_bytes_async", AsyncMock(return_value=b"x" * 1024))
+    monkeypatch.setattr(bambu_ftp, "remote_file_settled", AsyncMock(return_value=True))
+    monkeypatch.setattr(bambu_ftp, "delete_archived_timelapse", AsyncMock(return_value=True))
+    monkeypatch.setattr(main_module.printer_manager, "get_status", lambda _pid: status)
+    return status, patch.object(main_module, "async_session", maker)
+
+
+async def _complete(main, archive_id, printer_id):
+    """What on_print_complete does: schedule the scan, here run inline."""
+    spawned = []
+    with patch.object(main, "spawn_background_task", lambda coro, name=None: spawned.append(coro)):
+        main._schedule_timelapse_scan(archive_id, printer_id, plate_id=None)
+    assert len(spawned) == 1
+    await spawned[0]
+
+
+async def _attached(maker, archive_id) -> str | None:
+    async with maker() as db:
+        return (await db.get(PrintArchive, archive_id)).timelapse_path
+
+
+class TestTheScanDoesNotGuess:
+    """The reporter's chain, from the stored baseline on: [] at print start, and
+    by completion the card readable again with this print's video beside two
+    old ones."""
+
+    async def test_a_stored_unread_baseline_leaves_several_videos_alone(self, test_engine, tmp_path, monkeypatch):
+        from backend.app import main
+
+        maker, _, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_A), _entry(OLD_B), _entry(NEW_VIDEO)])
+
+        with session:
+            await main._scan_for_timelapse_with_retries(archive_id, None)
+
+        assert await _attached(maker, archive_id) is None, "an unread baseline must not pick between three videos"
+
+    async def test_an_unread_baseline_handed_over_in_memory_leaves_them_alone(self, test_engine, tmp_path, monkeypatch):
+        """The in-process path: on_print_complete hands the in-memory copy over,
+        and the archive row is never consulted."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_A), _entry(OLD_B), _entry(NEW_VIDEO)])
+        main._timelapse_baselines[printer.id] = set()
+        main._timelapse_baselines_unread.add(printer.id)
+
+        with session:
+            await _complete(main, archive_id, printer.id)
+        assert printer.id not in main._timelapse_baselines
+        assert printer.id not in main._timelapse_baselines_unread
+
+        assert await _attached(maker, archive_id) is None
+
+    async def test_a_stored_read_baseline_beats_an_unread_one_taken_after_a_restart(
+        self, test_engine, tmp_path, monkeypatch
+    ):
+        """Read at print start; after a restart the running-observed capture
+        could not read the card. The stored one is this run's."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[OLD_A, OLD_B], unread=False)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_A), _entry(OLD_B), _entry(NEW_VIDEO)])
+        main._timelapse_baselines[printer.id] = set()
+        main._timelapse_baselines_unread.add(printer.id)
+
+        with session:
+            await _complete(main, archive_id, printer.id)
+
+        attached = await _attached(maker, archive_id)
+        assert attached is not None and NEW_VIDEO in attached
+
+    async def test_a_lone_video_still_resolves_off_an_unread_baseline(self, test_engine, tmp_path, monkeypatch):
+        """The usual card: Bambuddy deletes each video once attached, so only
+        this print's is there, and one candidate needs no baseline."""
+        from backend.app import main
+
+        maker, _, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(NEW_VIDEO)])
+
+        with session:
+            await main._scan_for_timelapse_with_retries(archive_id, None)
+
+        attached = await _attached(maker, archive_id)
+        assert attached is not None and NEW_VIDEO in attached
+
+    async def test_a_read_baseline_still_picks_the_new_video(self, test_engine, tmp_path, monkeypatch):
+        from backend.app import main
+
+        maker, _, archive_id = await _seed(test_engine, baseline=[OLD_A, OLD_B], unread=False)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_A), _entry(OLD_B), _entry(NEW_VIDEO)])
+
+        with session:
+            await main._scan_for_timelapse_with_retries(archive_id, None)
+
+        attached = await _attached(maker, archive_id)
+        assert attached is not None and NEW_VIDEO in attached
+
+    async def test_a_baseline_the_scan_takes_itself_off_a_failed_listing(self, test_engine, tmp_path, monkeypatch):
+        """No cool-off running, the listing simply failed (a timeout, a 421 on
+        this very connection): the old check only knew about the cool-off."""
+        from backend.app import main
+
+        maker, _, archive_id = await _seed(test_engine)
+        rounds = iter([None, None])  # both baseline listings fail, the poll's succeed
+        _, session = _ftp(
+            main,
+            maker,
+            monkeypatch,
+            tmp_path,
+            lambda: next(rounds, [_entry(OLD_A), _entry(OLD_B), _entry(NEW_VIDEO)]),
+        )
+
+        with session:
+            await main._scan_for_timelapse_with_retries(archive_id, None)
+
+        assert await _attached(maker, archive_id) is None
+
+
+class TestTheCaptureAtPrintStart:
+    async def _capture(self, main, printer, archive_id):
+        spawned = []
+
+        def _spawn(coro, name=None):
+            spawned.append(name)
+            coro.close()
+
+        with patch.object(main, "spawn_background_task", _spawn):
+            await main._capture_timelapse_baseline_at_start(
+                printer, printer.id, logging.getLogger(__name__), archive_id=archive_id
+            )
+        return spawned
+
+    async def test_an_unreadable_card_is_recorded_as_unread(self, test_engine, tmp_path, monkeypatch):
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: None)
+
+        with session:
+            spawned = await self._capture(main, printer, archive_id)
+
+        async with maker() as db:
+            archive = await db.get(PrintArchive, archive_id)
+            assert (archive.timelapse_baseline, archive.timelapse_baseline_unread) == ([], True)
+        assert main._timelapse_baselines[printer.id] == set()
+        assert printer.id in main._timelapse_baselines_unread
+        assert spawned == [f"reread-timelapse-baseline-{printer.id}"]
+
+    async def test_a_card_in_cooloff_is_recorded_as_unread(self, test_engine, tmp_path, monkeypatch):
+        """The reporter's log line: the cool-off was running at print start."""
+        from backend.app import main
+        from backend.app.services.bambu_ftp import BambuFTPClient
+
+        maker, printer, archive_id = await _seed(test_engine)
+        # A listing would find videos -- it must not be asked while blocked.
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [])
+        BambuFTPClient._handshake_blocked_until[PRINTER_IP] = time.monotonic() + 60
+        try:
+            with session:
+                await self._capture(main, printer, archive_id)
+        finally:
+            BambuFTPClient._handshake_blocked_until.pop(PRINTER_IP, None)
+
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).timelapse_baseline_unread is True
+
+    async def test_an_empty_card_that_answered_is_read(self, test_engine, tmp_path, monkeypatch):
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [])
+
+        with session:
+            spawned = await self._capture(main, printer, archive_id)
+
+        async with maker() as db:
+            archive = await db.get(PrintArchive, archive_id)
+            assert (archive.timelapse_baseline, archive.timelapse_baseline_unread) == ([], False)
+        assert printer.id not in main._timelapse_baselines_unread
+        assert spawned == []
+
+    async def test_a_failed_directory_is_not_outvoted_by_one_that_answered(self, test_engine, tmp_path, monkeypatch):
+        """/timelapse answers empty, /timelapse/video -- where this card keeps
+        its videos -- fails: the card was not read."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine)
+        failed = {"/timelapse/video": FileListResult(files=[], available=False)}
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [], others=failed)
+
+        with session:
+            await self._capture(main, printer, archive_id)
+
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).timelapse_baseline_unread is True
+
+    async def test_a_card_without_any_timelapse_directory_is_read(self, test_engine, tmp_path, monkeypatch):
+        """A fresh card: every directory answers 550. That is an answer, and a
+        re-read could only get the same one."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [])
+        monkeypatch.setattr(
+            bambu_ftp,
+            "list_files_result_async",
+            AsyncMock(return_value=FileListResult(files=[], available=False, missing=True)),
+        )
+
+        with session:
+            spawned = await self._capture(main, printer, archive_id)
+
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).timelapse_baseline_unread is False
+        assert spawned == []
+
+    async def test_an_empty_first_listing_is_replaced_by_the_one_that_answered(
+        self, test_engine, tmp_path, monkeypatch
+    ):
+        """The first listing fails, the second reads the card: its names are the
+        baseline, not the empty set the first one returned."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine)
+        rounds = iter([None])
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: next(rounds, [_entry(OLD_A), _entry(OLD_B)]))
+
+        with session:
+            await self._capture(main, printer, archive_id)
+
+        async with maker() as db:
+            archive = await db.get(PrintArchive, archive_id)
+            assert (archive.timelapse_baseline, archive.timelapse_baseline_unread) == ([OLD_A, OLD_B], False)
+
+
+class TestTheCardIsReadAgainDuringThePrint:
+    async def _start(self, main, printer):
+        names: set[str] = set()
+        main._timelapse_baselines[printer.id] = names
+        main._timelapse_baselines_unread.add(printer.id)
+        return names
+
+    async def test_a_later_reading_becomes_the_baseline(self, test_engine, tmp_path, monkeypatch):
+        """And with it the scan picks this print's video out of three."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        card: list[list[dict] | None] = [None]
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: card[0])
+        names = await self._start(main, printer)
+
+        with session:
+            task = asyncio.create_task(main._reread_unread_timelapse_baseline(printer.id, names, archive_id, STARTED))
+            await asyncio.sleep(0.05)
+            assert printer.id in main._timelapse_baselines_unread, "nothing to record while the card is unreadable"
+            card[0] = [_entry(OLD_A), _entry(OLD_B)]  # the cool-off ends
+            await asyncio.wait_for(task, 2)
+
+            assert main._timelapse_baselines[printer.id] is names
+            assert names == {OLD_A, OLD_B}
+            assert printer.id not in main._timelapse_baselines_unread
+            async with maker() as db:
+                archive = await db.get(PrintArchive, archive_id)
+                assert (archive.timelapse_baseline, archive.timelapse_baseline_unread) == ([OLD_A, OLD_B], False)
+
+            card[0] = [_entry(OLD_A), _entry(OLD_B), _entry(NEW_VIDEO)]  # the print ends
+            await _complete(main, archive_id, printer.id)
+
+        attached = await _attached(maker, archive_id)
+        assert attached is not None and NEW_VIDEO in attached
+
+    async def test_a_disconnect_skips_a_round_and_does_not_end_the_rereading(self, test_engine, tmp_path, monkeypatch):
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        status, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_A)])
+        status.connected = False
+        names = await self._start(main, printer)
+
+        with session:
+            task = asyncio.create_task(main._reread_unread_timelapse_baseline(printer.id, names, archive_id, STARTED))
+            await asyncio.sleep(0.05)
+            assert not task.done(), "a dropped connection says nothing about the print"
+            assert names == set()
+            status.connected = True
+            await asyncio.wait_for(task, 2)
+
+        assert names == {OLD_A}
+
+    async def test_a_reading_that_returns_to_a_disconnected_printer_is_not_kept(
+        self, test_engine, tmp_path, monkeypatch
+    ):
+        """The printer dropped off while the listing was out, so nothing says
+        the print had not ended meanwhile."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        names = await self._start(main, printer)
+        status = None
+        card = [[_entry(OLD_A), _entry(NEW_VIDEO)]]
+
+        def listing():
+            if status.connected:
+                status.connected = False  # drops while this listing is out
+                return card[0]
+            return None
+
+        status, session = _ftp(main, maker, monkeypatch, tmp_path, listing)
+
+        with session:
+            task = asyncio.create_task(main._reread_unread_timelapse_baseline(printer.id, names, archive_id, STARTED))
+            await asyncio.sleep(0.05)
+            task.cancel()
+
+        assert names == set()
+        assert printer.id in main._timelapse_baselines_unread
+
+    async def test_nothing_is_read_once_the_print_has_ended(self, test_engine, tmp_path, monkeypatch):
+        """After the end the printer may already be writing this print's video,
+        which a reading would put into the baseline."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(NEW_VIDEO)], state="FINISH")
+        names = await self._start(main, printer)
+
+        with session:
+            await asyncio.wait_for(main._reread_unread_timelapse_baseline(printer.id, names, archive_id, STARTED), 2)
+
+        assert names == set()
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).timelapse_baseline_unread is True
+
+    async def test_a_reading_that_returns_after_completion_is_dropped(self, test_engine, tmp_path, monkeypatch):
+        """The print ends while the listing is out: completion has taken the
+        baseline by the time it comes back."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        names = await self._start(main, printer)
+        status = None
+
+        def listing():
+            status.state = "FINISH"
+            main._timelapse_baselines.pop(printer.id)
+            main._timelapse_baselines_unread.discard(printer.id)
+            return [_entry(OLD_A), _entry(NEW_VIDEO)]
+
+        status, session = _ftp(main, maker, monkeypatch, tmp_path, listing)
+
+        with session:
+            await asyncio.wait_for(main._reread_unread_timelapse_baseline(printer.id, names, archive_id, STARTED), 2)
+
+        assert names == set()
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).timelapse_baseline == []
+
+    async def test_a_reading_still_in_the_print_but_with_the_baseline_gone_is_dropped(
+        self, test_engine, tmp_path, monkeypatch
+    ):
+        """State alone is not the run: the next print can already be running."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        names = await self._start(main, printer)
+
+        def listing():
+            main._timelapse_baselines[printer.id] = {"the next print's"}
+            return [_entry(OLD_A)]
+
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, listing)
+
+        with session:
+            await asyncio.wait_for(main._reread_unread_timelapse_baseline(printer.id, names, archive_id, STARTED), 2)
+
+        assert names == set()
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).timelapse_baseline_unread is True
+
+    async def test_a_reprinted_row_keeps_its_own_baseline(self, test_engine, tmp_path, monkeypatch):
+        """The row passed to a newer run meanwhile: the reading belongs to the
+        old run, so memory takes it and the row does not."""
+        from backend.app import main
+
+        maker, printer, archive_id = await _seed(test_engine, baseline=[], unread=True)
+        _, session = _ftp(main, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_A)])
+        names = await self._start(main, printer)
+
+        with session:
+            await asyncio.wait_for(
+                main._reread_unread_timelapse_baseline(printer.id, names, archive_id, datetime(2026, 10, 1)), 2
+            )
+
+        assert names == {OLD_A}
+        async with maker() as db:
+            archive = await db.get(PrintArchive, archive_id)
+            assert (archive.timelapse_baseline, archive.timelapse_baseline_unread) == ([], True)