fix(timelapse): a video goes to the run it belongs to (scan stops on a newer run, claimed videos left out of matching, camera video replaces the printer's without orphaning it); every timelapse change refreshes open Archives pages

.basedpyright/baseline.json

@@ -2019,14 +2019,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 88,
-                    "endColumn": 97,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportArgumentType",
                 "range": {

CHANGELOG.md

@@ -164,6 +164,11 @@ 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 goes to the print it belongs to** — Several cases gave an archive the wrong video, or none:
+  - **The next plate or a reprint started meanwhile:** the printer's timelapse is looked for for some minutes after a print ends. When the same archive started printing again in that time, the previous print's video went to the new print, and the new print's own video was then never imported. The search now stops once the archive is printing again, and nothing is attached or deleted from the printer.
+  - **Scan for Timelapse on an older archive** (from before the video list was recorded at print start) could pick a video another archive already has, for instance by a short name like "1" that every video's timestamp contains. Those are now left out of the automatic match, and are still offered for picking by hand.
+  - **Both the external camera's and the printer's timelapse on:** the archive keeps the external camera's video, as before, but the printer's video it replaces is now deleted instead of staying in the archive folder unused. A camera video stitched for a print whose archive is already printing again (the next plate, a reprint) is no longer attached to the new print.
+- **Archives pages show a timelapse as soon as it changes** — Only the automatic scan told other open pages; a video picked or uploaded by hand, the external camera's, the conversion to MP4 and deleting one showed only after a reload.
 - **An archive cost set through the API, or by Recalculate costs, reaches the statistics** — Setting an archive's cost (`PATCH /api/v1/archives/{id}`) or **Recalculate costs** changed the archive only, while statistics, project totals and batch sums read the cost from the print log, so they kept the old figure. A completed print that had the archive's cost now gets the new one too; a print priced from its own spools, and a failed or cancelled one, keep theirs. **Recalculate costs** still recalculates a cost that was set by hand, as before. A negative cost, or one over 1,000,000, is refused.
 - **A request with `Infinity` or `NaN` in a field that fails validation answers 422** — Such a request ended in a server error, because the error response repeats the value sent and JSON can't hold it.
 - **Editing an archive while it is being reprinted leaves the previous print alone** — Its outcome, failure reason, verdict and filament weight went onto the previous print's log entry, rewriting that print's result in the print log and failure statistics. The edit now changes the archive only.

backend/app/api/routes/archives.py

@@ -2958,6 +2958,9 @@ async def delete_timelapse(
     archive.timelapse_plate_id = None
     await db.commit()
 
+    from backend.app.services.archive import _announce_timelapse
+
+    await _announce_timelapse(archive_id, attached=False)
     return {"status": "deleted"}
 
 
@@ -3053,19 +3056,23 @@ async def scan_timelapse(
     # into a confident wrong answer. Those strategies stay for archives created
     # before the baseline was persisted.
     used_baseline = archive.timelapse_baseline is not None
+    # A video another archive of this printer holds is that print's, whichever
+    # strategy below would pick it (a short name like "1" is in every video's
+    # timestamp). Picking by hand still offers it: that is how a wrong match
+    # is corrected.
+    offered = video_files
+    async with async_session() as db:
+        claimed = await _claimed_timelapse_stems(db, archive.printer_id, archive_id)
+    video_files = [f for f in video_files if Path(f.get("name", "")).stem not in claimed]
     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
-        ]
+        candidates = [f for f in video_files if f.get("name", "") not in baseline]
         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
+            offered = 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)
@@ -3144,7 +3151,7 @@ async def scan_timelapse(
                 "size": f.get("size"),
                 "mtime": f.get("mtime").isoformat() if f.get("mtime") else None,
             }
-            for f in video_files
+            for f in offered
         ]
         # Sort by mtime descending (most recent first)
         available_files.sort(key=lambda x: x.get("mtime") or "", reverse=True)

backend/app/main.py

@@ -6281,6 +6281,10 @@ async def _scan_for_timelapse_with_retries(
             if not archive.printer_id:
                 logger.warning("[TIMELAPSE] Archive %s has no printer, aborting", archive_id)
                 return
+            # The run this scan is for. The archive row passes to a newer run
+            # when it prints again (the next plate, a reprint), and that run's
+            # video is its own.
+            run_started_at = archive.started_at
 
             if baseline_names is not None:
                 logger.info(
@@ -6361,6 +6365,11 @@ async def _scan_for_timelapse_with_retries(
                 if archive.timelapse_path:
                     logger.info("[TIMELAPSE] Archive %s already has timelapse attached, stopping poll", archive_id)
                     return
+                if archive.started_at != run_started_at:
+                    logger.info(
+                        "[TIMELAPSE] Archive %s is printing again, stopping the previous run's poll", archive_id
+                    )
+                    return
 
                 result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
                 printer = result.scalar_one_or_none()
@@ -6398,6 +6407,7 @@ async def _scan_for_timelapse_with_retries(
                     quiet=not changed,
                     require_unambiguous=not baseline_trusted,
                     plate_id=plate_id,
+                    run_started_at=run_started_at,
                 )
                 if attached:
                     return
@@ -6421,6 +6431,55 @@ async def _scan_for_timelapse_with_retries(
     )
 
 
+async def _attach_layer_timelapse(
+    archive_id: int, timelapse_path: Path, plate_id: int | None, run_started_at: datetime | None
+) -> None:
+    """Attach the external camera's stitched timelapse to the archive.
+
+    The camera's video is the one the user set up in Bambuddy, so it wins
+    over the printer's own when both are on: it replaces the printer's video
+    (usually there first; stitching takes minutes) and that file is deleted,
+    leaving nothing in the archive folder unreferenced. ``run_started_at`` is
+    the run the frames were taken for: when the archive is printing again by
+    the time the stitch is done, the video is not the new run's and is
+    dropped. The stitched file is a temporary either way.
+    """
+    from backend.app.utils.safe_path import PathTraversalError, safe_join_under
+
+    logger = logging.getLogger(__name__)
+    try:
+        async with async_session() as db:
+            service = ArchiveService(db)
+            archive = await service.get_archive(archive_id)
+            if archive is None:
+                return
+            if archive.started_at != run_started_at:
+                logger.info(
+                    "[LAYER-TL] Archive %s is printing again; dropping the previous run's layer timelapse", archive_id
+                )
+                return
+            replaced = archive.timelapse_path
+            logger.info("[LAYER-TL] Attaching timelapse %s to archive %s", timelapse_path, archive_id)
+            timelapse_data = await asyncio.to_thread(timelapse_path.read_bytes)
+            if not await service.attach_timelapse(archive_id, timelapse_data, "layer_timelapse.mp4", plate_id=plate_id):
+                return
+            attached = archive.timelapse_path
+        logger.info("[LAYER-TL] Layer timelapse attached successfully")
+        if replaced and replaced != attached:
+            try:
+                old = safe_join_under(app_settings.base_dir, replaced, http=False)
+            except PathTraversalError:
+                return
+            await asyncio.to_thread(old.unlink, missing_ok=True)
+            logger.info("[LAYER-TL] Replaced the printer's timelapse %s on archive %s", replaced, archive_id)
+    finally:
+        await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
+
+
+# ``run_started_at`` not given: attach whichever run the archive is on.
+_ANY_RUN = object()
+
+
 async def _attach_first_unclaimed_timelapse(
     archive_id: int,
     printer,
@@ -6433,6 +6492,7 @@ async def _attach_first_unclaimed_timelapse(
     quiet: bool = False,
     require_unambiguous: bool = False,
     plate_id: int | None = None,
+    run_started_at: datetime | None | object = _ANY_RUN,
 ) -> bool:
     """Download and attach the one video that belongs to this print.
 
@@ -6495,7 +6555,7 @@ async def _attach_first_unclaimed_timelapse(
         )
 
     target = candidates[0]
-    file_name = target.get("name")
+    file_name = str(target.get("name") or "")  # every listed entry has one
     remote_path = target.get("path") or f"/timelapse/{file_name}"
     logger.info(
         "[TIMELAPSE] Attempt %s: New file detected: %s (downloading for archive %s)",
@@ -6540,15 +6600,28 @@ async def _attach_first_unclaimed_timelapse(
     ):
         return False
 
-    # Write phase: attach in a fresh short-lived session.
+    # Write phase: attach in a fresh short-lived session. The download took a
+    # while: by now the archive may have its video, or be printing again for a
+    # newer run, whose video this is not. Then nothing is attached or deleted,
+    # and the poll ends (True).
     async with async_session() as db:
-        success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, file_name, plate_id=plate_id)
+        service = ArchiveService(db)
+        if run_started_at is not _ANY_RUN:
+            current = await service.get_archive(archive_id)
+            if current is None or current.timelapse_path or current.started_at != run_started_at:
+                logger.info(
+                    "[TIMELAPSE] Archive %s moved on during the download (video attached or printing again); "
+                    "%s stays on the printer",
+                    archive_id,
+                    file_name,
+                )
+                return True
+        success = await service.attach_timelapse(archive_id, timelapse_data, file_name, plate_id=plate_id)
     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,
@@ -9353,18 +9426,16 @@ async def on_print_complete(printer_id: int, data: dict):
         try:
             if print_status == "completed":
                 logger.info("[LAYER-TL] Stitching layer timelapse for printer %s", printer_id)
+                # The run the frames belong to, read before the stitch: it can
+                # take long enough for the archive to start printing again.
+                run_started_at = None
+                if archive_id:
+                    async with async_session() as db:
+                        run_archive = await ArchiveService(db).get_archive(archive_id)
+                        run_started_at = run_archive.started_at if run_archive else None
                 timelapse_path = await tl_complete(printer_id)
                 if timelapse_path and archive_id:
-                    logger.info("[LAYER-TL] Attaching timelapse %s to archive %s", timelapse_path, archive_id)
-                    async with async_session() as db:
-                        service = ArchiveService(db)
-                        timelapse_data = await asyncio.to_thread(timelapse_path.read_bytes)
-                        await service.attach_timelapse(
-                            archive_id, timelapse_data, "layer_timelapse.mp4", plate_id=notify_plate_id
-                        )
-                        # Clean up the temp file
-                        await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
-                        logger.info("[LAYER-TL] Layer timelapse attached successfully")
+                    await _attach_layer_timelapse(archive_id, timelapse_path, notify_plate_id, run_started_at)
                 elif timelapse_path:
                     # Timelapse created but no archive - just clean up
                     await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)

backend/app/services/archive.py

@@ -1981,6 +1981,7 @@ class ArchiveService:
         archive.timelapse_path = str(timelapse_file.relative_to(settings.base_dir))
         archive.timelapse_plate_id = plate_id
         await self.db.commit()
+        await _announce_timelapse(archive_id, attached=True)
 
         # For non-MP4 videos (e.g. AVI from P1S), kick off background conversion
         if not filename.lower().endswith(".mp4"):
@@ -2017,6 +2018,16 @@ def _free_timelapse_name(target: Path, current_relpath: str | None, base_dir: Pa
         n += 1
 
 
+async def _announce_timelapse(archive_id: int, *, attached: bool) -> None:
+    """Tell open Archives pages that this archive's timelapse changed."""
+    from backend.app.core.websocket import ws_manager
+
+    try:
+        await ws_manager.send_archive_updated({"id": archive_id, "timelapse_attached": attached})
+    except Exception:  # noqa: BLE001 -- a refresh hint must not fail the change it reports
+        logger.debug("Could not announce the timelapse of archive %s", archive_id, exc_info=True)
+
+
 async def _convert_timelapse_to_mp4(archive_id: int, source_path: Path) -> None:
     """Background task: convert non-MP4 timelapse (e.g. AVI from P1S) to MP4.
 
@@ -2107,6 +2118,7 @@ async def _convert_timelapse_to_mp4(archive_id: int, source_path: Path) -> None:
                 return
             archive.timelapse_path = str(mp4_path.relative_to(settings.base_dir))
             await db.commit()
+        await _announce_timelapse(archive_id, attached=True)
 
         # Remove original non-MP4 file
         if source_path.exists():

backend/tests/integration/test_print_complete_end_to_end.py

@@ -307,3 +307,21 @@ class TestACompletionNoArchiveWasBoundTo:
 
         db_session.expire_all()
         assert (await db_session.get(PrintArchive, own_id)).status == "completed"
+
+
+async def test_the_layer_timelapse_is_attached_for_the_run_that_ended(test_engine, db_session, queued_print, tmp_path):
+    """The run is read before the stitch, which can take long enough for the
+    archive to start printing again (TL-3)."""
+    stitched = tmp_path / "stitched.mp4"
+    stitched.write_bytes(b"layer video")
+    attach = AsyncMock()
+
+    with (
+        patch("backend.app.services.layer_timelapse.on_print_complete", AsyncMock(return_value=stitched)),
+        patch("backend.app.main._attach_layer_timelapse", attach),
+    ):
+        await _complete(test_engine, queued_print["printer"], "completed")
+
+    attach.assert_awaited_once()
+    archive_id, path, _plate, run_started_at = attach.await_args.args
+    assert (archive_id, path, run_started_at) == (queued_print["archive"], stitched, datetime(2026, 10, 19, 8, 0))

backend/tests/integration/test_timelapse_changes_announced.py

@@ -0,0 +1,43 @@
+"""Every change to an archive's timelapse tells open Archives pages (TL-5).
+
+Only the automatic scan sent ``archive_updated``. A video attached any other
+way (picked or uploaded by hand, the external camera's, the MP4 conversion)
+or deleted showed on other open pages only once they refetched.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def test_an_attach_is_announced(db_session, archive_factory, printer_factory, tmp_path, monkeypatch):
+    from backend.app.core.config import settings as app_config
+    from backend.app.services.archive import ArchiveService
+
+    monkeypatch.setattr(app_config, "base_dir", tmp_path)
+    monkeypatch.setattr(app_config, "archive_dir", tmp_path / "archive")
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id)
+    announce = AsyncMock()
+
+    with patch("backend.app.services.archive._announce_timelapse", announce):
+        assert await ArchiveService(db_session).attach_timelapse(archive.id, b"video", "clip.mp4")
+
+    announce.assert_awaited_once_with(archive.id, attached=True)
+
+
+async def test_a_delete_is_announced(async_client, db_session, archive_factory, printer_factory, tmp_path, monkeypatch):
+    from backend.app.core.config import settings as app_config
+
+    monkeypatch.setattr(app_config, "base_dir", tmp_path)
+    printer = await printer_factory()
+    archive = await archive_factory(printer.id, timelapse_path="archive/x/clip.mp4")
+    announce = AsyncMock()
+
+    with patch("backend.app.services.archive._announce_timelapse", announce):
+        response = await async_client.delete(f"/api/v1/archives/{archive.id}/timelapse")
+
+    assert response.status_code == 200, response.text
+    announce.assert_awaited_once_with(archive.id, attached=False)

backend/tests/unit/test_archive_filtering.py

@@ -690,6 +690,7 @@ class TestConvertTimelapseToMp4:
             patch("backend.app.services.archive.settings") as mock_settings,
             patch("backend.app.utils.archive_paths.settings") as mock_paths_settings,
             patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
+            patch("backend.app.services.archive._announce_timelapse", AsyncMock()) as announce,
         ):
             mock_settings.base_dir = tmp_path
             mock_paths_settings.base_dir = tmp_path
@@ -702,8 +703,22 @@ class TestConvertTimelapseToMp4:
 
             await _convert_timelapse_to_mp4(42, source)
 
+        mock_session.announce = announce
         return mock_exec, mock_session, mock_archive, mp4_path
 
+    @pytest.mark.asyncio
+    async def test_the_converted_video_is_announced(self, tmp_path):
+        """Open Archives pages refresh to the MP4 (TL-5)."""
+        _, session, _, _ = await self._convert(tmp_path, shown="video.avi")
+
+        session.announce.assert_awaited_once_with(42, attached=True)
+
+    @pytest.mark.asyncio
+    async def test_a_discarded_conversion_is_not_announced(self, tmp_path):
+        _, session, _, _ = await self._convert(tmp_path, shown="other.avi")
+
+        session.announce.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_converts_avi_to_mp4(self, tmp_path):
         """Should call FFmpeg to convert and update the DB path."""
@@ -794,6 +809,8 @@ class TestAttachTimelapseBackgroundConversion:
             patch("backend.app.services.archive.settings") as mock_settings,
             patch("backend.app.utils.archive_paths.settings") as mock_paths_settings,
             patch("asyncio.create_task") as mock_create_task,
+            # Its broadcast opens a session task of its own
+            patch("backend.app.services.archive._announce_timelapse", AsyncMock()),
         ):
             mock_settings.base_dir = tmp_path
             mock_paths_settings.base_dir = tmp_path
@@ -823,6 +840,8 @@ class TestAttachTimelapseBackgroundConversion:
             patch("backend.app.services.archive.settings") as mock_settings,
             patch("backend.app.utils.archive_paths.settings") as mock_paths_settings,
             patch("asyncio.create_task") as mock_create_task,
+            # Its broadcast opens a session task of its own
+            patch("backend.app.services.archive._announce_timelapse", AsyncMock()),
         ):
             mock_settings.base_dir = tmp_path
             mock_paths_settings.base_dir = tmp_path

backend/tests/unit/test_layer_timelapse_attach.py

@@ -0,0 +1,82 @@
+"""The external camera's timelapse and the printer's own (TL-3).
+
+With both on, both videos arrive for one archive. The camera's, the one set
+up in Bambuddy, wins: it replaces the printer's (usually there first, since
+stitching takes minutes), and the replaced file is deleted instead of being
+left in the archive folder with nothing pointing at it. Frames taken for one
+run never go to a newer run of the same archive.
+"""
+
+from datetime import datetime
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from backend.app.models.archive import PrintArchive
+from backend.tests.unit.test_fallback_timelapse_baseline_2957 import _seed
+
+pytestmark = pytest.mark.asyncio
+
+RUN = datetime(2026, 10, 19, 8, 0)
+
+
+async def _attach(test_engine, tmp_path, monkeypatch, *, existing: str | None, run_started_at=RUN):
+    from backend.app import main as main_module
+    from backend.app.core.config import settings as app_config
+
+    monkeypatch.setattr(app_config, "base_dir", tmp_path)
+    monkeypatch.setattr(app_config, "archive_dir", tmp_path / "archive")
+    maker, _printer_id, archive_id = await _seed(test_engine)
+    async with maker() as db:
+        archive = await db.get(PrintArchive, archive_id)
+        archive.started_at = RUN
+        archive.timelapse_path = existing
+        await db.commit()
+    if existing:
+        (tmp_path / existing).parent.mkdir(parents=True, exist_ok=True)
+        (tmp_path / existing).write_bytes(b"printer video")
+    stitched = tmp_path / "stitched.mp4"
+    stitched.write_bytes(b"layer video")
+    announce = AsyncMock()
+
+    with (
+        patch.object(main_module, "async_session", maker),
+        patch("backend.app.services.archive._announce_timelapse", announce),
+    ):
+        await main_module._attach_layer_timelapse(archive_id, stitched, None, run_started_at)
+
+    async with maker() as db:
+        attached = (await db.get(PrintArchive, archive_id)).timelapse_path
+    return attached, stitched, announce
+
+
+async def test_it_replaces_the_printer_s_video_and_deletes_that_file(test_engine, tmp_path, monkeypatch):
+    attached, stitched, announce = await _attach(test_engine, tmp_path, monkeypatch, existing="archive/1/video_a.mp4")
+
+    assert attached and attached.endswith("layer_timelapse.mp4")
+    assert not (tmp_path / "archive/1/video_a.mp4").exists(), "the replaced video would be left orphaned"
+    assert not stitched.exists(), "the stitched temporary is removed either way"
+    assert announce.await_args.kwargs == {"attached": True}
+
+
+async def test_without_one_it_is_attached(test_engine, tmp_path, monkeypatch):
+    attached, stitched, _announce = await _attach(test_engine, tmp_path, monkeypatch, existing=None)
+
+    assert attached and attached.endswith("layer_timelapse.mp4")
+    assert not stitched.exists()
+
+
+async def test_a_newer_run_of_the_archive_does_not_get_it(test_engine, tmp_path, monkeypatch):
+    """The stitch took long enough for the archive to start printing again."""
+    attached, stitched, announce = await _attach(
+        test_engine,
+        tmp_path,
+        monkeypatch,
+        existing="archive/1/video_a.mp4",
+        run_started_at=datetime(2026, 10, 19, 7, 0),
+    )
+
+    assert attached == "archive/1/video_a.mp4"
+    assert (tmp_path / "archive/1/video_a.mp4").exists()
+    assert not stitched.exists()
+    announce.assert_not_awaited()

backend/tests/unit/test_timelapse_scan_2704.py

@@ -382,7 +382,7 @@ class TestManualScanUsesTheBaseline:
         a.created_at = a.completed_at
         return a
 
-    async def _scan(self, archive, listing, download=None, delete=None):
+    async def _scan(self, archive, listing, download=None, delete=None, claimed=()):
         from backend.app.api.routes import archives as archives_mod
 
         service = MagicMock()
@@ -395,7 +395,8 @@ class TestManualScanUsesTheBaseline:
         session.execute = AsyncMock(
             return_value=MagicMock(
                 scalar_one_or_none=MagicMock(return_value=_printer()),
-                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
+                # Videos other archives of this printer hold
+                scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=list(claimed)))),
             )
         )
 
@@ -767,3 +768,35 @@ class TestFileMustHaveStoppedGrowing:
         assert result is True
         # Checked against what we actually received, not against the listing.
         assert settled.await_args.args[3] == 1000
+
+
+class TestAScanWithoutABaselineSkipsClaimedVideos:
+    """An archive from before the baseline was stored is matched on name and
+    clock, which must not hand it a video another archive already holds
+    (SN-C7)."""
+
+    _archive = TestManualScanUsesTheBaseline._archive
+    _scan = TestManualScanUsesTheBaseline._scan
+
+    @pytest.mark.asyncio
+    async def test_a_short_name_does_not_take_another_archive_s_video(self):
+        archive = self._archive(None)
+        archive.filename = "1.3mf"  # "1" is in every video's timestamp
+        listing = [_video("video_2026-07-28_21-19-10.avi"), _video("video_2026-07-28_21-19-40.avi")]
+
+        result = await self._scan(archive, listing, claimed=["archive/1/other/video_2026-07-28_21-19-10.mp4"])
+
+        assert result.get("filename") != "video_2026-07-28_21-19-10.avi"
+
+    @pytest.mark.asyncio
+    async def test_a_claimed_video_is_still_offered_for_picking_by_hand(self):
+        """Picking by hand is how a wrong match is put right."""
+        archive = self._archive(None)
+        archive.filename = "mops.3mf"
+        archive.started_at = archive.completed_at = archive.created_at = None
+        listing = [_video("video_a.avi"), _video("video_b.avi")]
+
+        result = await self._scan(archive, listing, claimed=["archive/1/other/video_a.mp4"])
+
+        assert result["status"] == "not_found"
+        assert {f["name"] for f in result["available_files"]} == {"video_a.avi", "video_b.avi"}

backend/tests/unit/test_timelapse_scan_newer_run.py

@@ -0,0 +1,115 @@
+"""A timelapse scan attaches only to the run it was started for (TL-2).
+
+The scan polls the printer for minutes after a print ends. When the same
+archive starts printing again meanwhile (the next plate of a shared archive, a
+reprint), the archive row belongs to the new run: attaching then gave the new
+run the old run's video, and the new run's own video was never imported
+because the archive already had one.
+"""
+
+from datetime import datetime
+from unittest.mock import AsyncMock
+
+import pytest
+
+from backend.app.models.archive import PrintArchive
+from backend.tests.unit.test_fallback_timelapse_baseline_2957 import (
+    NEW_VIDEO,
+    OLD_VIDEO,
+    _entry,
+    _patches,
+    _seed,
+)
+
+pytestmark = pytest.mark.asyncio
+
+FIRST_RUN = datetime(2026, 10, 19, 8, 0)
+NEXT_RUN = datetime(2026, 10, 19, 9, 30)
+
+
+async def _first_run_ended(maker, archive_id):
+    async with maker() as db:
+        archive = await db.get(PrintArchive, archive_id)
+        archive.started_at = FIRST_RUN
+        archive.timelapse_baseline = [OLD_VIDEO]
+        await db.commit()
+
+
+async def _next_run_starts(maker, archive_id):
+    async with maker() as db:
+        archive = await db.get(PrintArchive, archive_id)
+        archive.started_at = NEXT_RUN
+        archive.status = "printing"
+        await db.commit()
+
+
+async def test_a_run_that_started_during_the_download_does_not_get_the_video(test_engine, tmp_path, monkeypatch):
+    from backend.app import main as main_module
+    from backend.app.services import bambu_ftp
+
+    maker, _printer_id, archive_id = await _seed(test_engine)
+    await _first_run_ended(maker, archive_id)
+
+    with _patches(main_module, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_VIDEO), _entry(NEW_VIDEO)]):
+
+        async def download(*_args, **_kwargs):
+            await _next_run_starts(maker, archive_id)
+            return b"x" * 1024
+
+        monkeypatch.setattr(bambu_ftp, "download_file_bytes_async", AsyncMock(side_effect=download))
+        await main_module._scan_for_timelapse_with_retries(archive_id, None)
+
+    async with maker() as db:
+        assert (await db.get(PrintArchive, archive_id)).timelapse_path is None
+    bambu_ftp.delete_archived_timelapse.assert_not_awaited()
+
+
+async def test_the_poll_stops_once_a_newer_run_has_started(test_engine, tmp_path, monkeypatch):
+    from backend.app import main as main_module
+    from backend.app.services import bambu_ftp
+
+    maker, _printer_id, archive_id = await _seed(test_engine)
+    await _first_run_ended(maker, archive_id)
+    rounds = []
+
+    with _patches(main_module, maker, monkeypatch, tmp_path, lambda: []):
+
+        async def listing(_ip, _code, path, printer_model=None):
+            if path != "/timelapse":
+                return []
+            rounds.append(path)
+            if len(rounds) == 1:
+                await _next_run_starts(maker, archive_id)
+                return [_entry(OLD_VIDEO)]
+            return [_entry(OLD_VIDEO), _entry(NEW_VIDEO)]
+
+        monkeypatch.setattr(bambu_ftp, "list_files_async", listing)
+        await main_module._scan_for_timelapse_with_retries(archive_id, None)
+
+    assert len(rounds) == 1, "the poll went on for a run that is not its own"
+    bambu_ftp.download_file_bytes_async.assert_not_awaited()
+
+
+async def test_a_video_attached_during_the_download_is_kept(test_engine, tmp_path, monkeypatch):
+    """Picked by hand (Scan for Timelapse) while the automatic scan downloaded."""
+    from backend.app import main as main_module
+    from backend.app.services import bambu_ftp
+
+    maker, _printer_id, archive_id = await _seed(test_engine)
+    await _first_run_ended(maker, archive_id)
+
+    with _patches(main_module, maker, monkeypatch, tmp_path, lambda: [_entry(OLD_VIDEO), _entry(NEW_VIDEO)]):
+
+        async def download(*_args, **_kwargs):
+            async with maker() as db:
+                archive = await db.get(PrintArchive, archive_id)
+                archive.timelapse_path = "archive/1/picked_by_hand.mp4"
+                await db.commit()
+            return b"x" * 1024
+
+        monkeypatch.setattr(bambu_ftp, "download_file_bytes_async", AsyncMock(side_effect=download))
+        await main_module._scan_for_timelapse_with_retries(archive_id, None)
+
+    async with maker() as db:
+        assert (await db.get(PrintArchive, archive_id)).timelapse_path == "archive/1/picked_by_hand.mp4"
+    bambu_ftp.delete_archived_timelapse.assert_not_awaited()