Archives: a file printing on several printers at once is one archive (archive redesign P1b)

- A library or project dispatch adds its run to the file's archive also
  while the file prints, is queued or finishes elsewhere; the idle-only
  claim and the run-finishing marks are gone
- While another print of the archive runs, the dispatch leaves the
  archive's plate and cost centre to that print
- A queued print's cost centre is its queue item's, none included, when
  its run opens and when it is charged
- A plate keeps its latest print's video: an earlier print's scan goes
  on until a later print of the plate has its own video, whose arrival
  removes the earlier ones

backend/app/main.py

@@ -568,35 +568,6 @@ _timelapse_baselines: dict[int, set[str]] = {}
 # so it is empty without the card being empty (#3412).
 _timelapse_baselines_unread: set[int] = set()
 
-# Archives whose last run still has work after it completed: the timelapse
-# scan and the layer timelapse stitch. Both tell their run from a later one by
-# the archive's start time, so a queue dispatch of the same file must not start
-# a new run on the archive until they end (_claim_reusable_archive in the
-# scheduler). archive_id -> number of such tasks.
-_archive_runs_finishing: dict[int, int] = {}
-
-
-async def _while_run_finishes(archive_id: int, work):
-    """Run ``work`` with the archive marked as still finishing its last run.
-    Marked by the caller's spawn already: see ``_mark_run_finishing``."""
-    try:
-        return await work
-    finally:
-        left = _archive_runs_finishing.get(archive_id, 0) - 1
-        if left > 0:
-            _archive_runs_finishing[archive_id] = left
-        else:
-            _archive_runs_finishing.pop(archive_id, None)
-
-
-def _mark_run_finishing(archive_id: int, work):
-    """``work`` wrapped to keep the archive marked until it ends. Marked now,
-    before the task starts, so no dispatch between the spawn and its first
-    step can take the archive."""
-    _archive_runs_finishing[archive_id] = _archive_runs_finishing.get(archive_id, 0) + 1
-    return _while_run_finishes(archive_id, work)
-
-
 # Track printers waiting for bed to cool after print completion.
 # Event-driven: fires when bed_temper arrives via MQTT below threshold.
 # {printer_id: {"threshold": float, "filename": str, "registered_at": float}}
@@ -6797,11 +6768,8 @@ def _schedule_timelapse_scan(
     _timelapse_baselines_unread.discard(printer_id)
     baseline = _timelapse_baselines.pop(printer_id, None)
     spawn_background_task(
-        _mark_run_finishing(
-            archive_id,
-            _scan_for_timelapse_with_retries(
-                archive_id, baseline, plate_id=plate_id, baseline_unread=unread, run_id=run_id
-            ),
+        _scan_for_timelapse_with_retries(
+            archive_id, baseline, plate_id=plate_id, baseline_unread=unread, run_id=run_id
         ),
         name=f"scan-timelapse-{archive_id}",
     )
@@ -10518,10 +10486,7 @@ async def _close_out_print(printer_id: int, data: dict):
                 except Exception:
                     pass  # Best-effort timelapse session cancellation on error
 
-    layer_timelapse = _background_layer_timelapse()
-    if archive_id and print_status == "completed":
-        layer_timelapse = _mark_run_finishing(archive_id, layer_timelapse)
-    spawn_background_task(layer_timelapse, name="background-layer-timelapse")
+    spawn_background_task(_background_layer_timelapse(), name="background-layer-timelapse")
 
     log_timing("All background tasks scheduled")
 

backend/app/services/archive.py

@@ -1916,7 +1916,7 @@ class ArchiveService:
         """
         import asyncio
 
-        from backend.app.services.print_log import is_newest_run, newest_run, run_of
+        from backend.app.services.print_log import drop_earlier_plate_videos, is_newest_run, newest_run, run_of
 
         archive = await self.get_archive(archive_id)
         if not archive:
@@ -1974,10 +1974,20 @@ class ArchiveService:
         relpath = str(timelapse_file.relative_to(settings.base_dir))
         if run is not None:
             run.timelapse_path = relpath
+        # A plate keeps its latest print's video: earlier prints of it that
+        # got theirs meanwhile (printing on other printers at once) lose them.
+        replaced = await drop_earlier_plate_videos(self.db, run) if run_id is not None and run is not None else []
         if run_id is None or (run is not None and await is_newest_run(self.db, run)):
             archive.timelapse_path = relpath
             archive.timelapse_plate_id = plate_id
         await self.db.commit()
+        for old_relpath in replaced:
+            try:
+                old_file = safe_join_under(settings.base_dir, old_relpath, http=False)
+            except PathTraversalError:
+                continue
+            await asyncio.to_thread(old_file.unlink, missing_ok=True)
+            logger.info("Replaced plate %s's earlier timelapse %s on archive %s", plate_id, old_relpath, archive_id)
         await _announce_timelapse(archive_id, attached=True)
 
         # For non-MP4 videos (e.g. AVI from P1S), kick off background conversion

backend/app/services/finance_billing.py

@@ -224,7 +224,10 @@ async def apply_print_charge_for_archive(
             actual_user_id = archive.created_by_id
         if cost_center_id is not None:
             actual_cost_center_id = cost_center_id
-        elif run is not None and run.cost_center_id is not None:
+        elif run is not None and (run.cost_center_id is not None or run.queue_item_id is not None):
+            # A queued print's is its queue item's, none included: the
+            # archive's is another print's when the same file prints on
+            # several printers at once (archive redesign).
             actual_cost_center_id = run.cost_center_id
         else:
             actual_cost_center_id = archive.cost_center_id

backend/app/services/print_log.py

@@ -10,7 +10,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from types import SimpleNamespace
 
-from sqlalchemy import delete, func, or_, select
+from sqlalchemy import and_, delete, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry
@@ -447,7 +447,9 @@ async def open_run(
         # Not the archive's: that is the latest dispatch's, or the id an
         # earlier print of the archive was charged under.
         billing_run_id=(item.billing_run_id if item is not None else None) or str(uuid.uuid4()),
-        cost_center_id=(item.cost_center_id if item is not None else None) or archive.cost_center_id,
+        # The queue item's, none included: the archive's is another print's
+        # when the same file prints on several printers at once.
+        cost_center_id=item.cost_center_id if item is not None else archive.cost_center_id,
         confirm_requested=archive.confirm_requested,
         quantity=archive.quantity,
         # Its own finish photos from here on (``owns_media``).
@@ -625,16 +627,17 @@ def owns_media(run: PrintLogEntry) -> bool:
 
 
 async def superseded(db: AsyncSession, run: PrintLogEntry) -> bool:
-    """Whether a later print of the run's archive has started on its printer,
-    or of its plate: a video landing now is not the run's to take. The one on
-    its printer is that print's own, and a plate keeps one video, the latest
-    print's (a reprint replaces it). Another plate on another printer leaves
-    the run's video to it."""
+    """Whether a later print of the run's archive took over its video: one
+    started on its printer, whose video a video landing now may be, or one of
+    its plate that has its video already: a plate keeps one video, its latest
+    print's (a reprint replaces it, ``drop_earlier_plate_videos``). A later
+    print of the plate still running on another printer leaves the run its
+    video until it has one of its own, which it may never get."""
     if run.archive_id is None or run.started_at is None:
         return False
     same = [PrintLogEntry.printer_id == run.printer_id] if run.printer_id is not None else []
     if run.plate_id is not None:
-        same.append(PrintLogEntry.plate_id == run.plate_id)
+        same.append(and_(PrintLogEntry.plate_id == run.plate_id, PrintLogEntry.timelapse_path.is_not(None)))
     if not same:
         return False
     later = await db.scalar(
@@ -651,6 +654,43 @@ async def superseded(db: AsyncSession, run: PrintLogEntry) -> bool:
     return later is not None
 
 
+async def drop_earlier_plate_videos(db: AsyncSession, run: PrintLogEntry) -> list[str]:
+    """Take the videos of the archive's earlier prints of ``run``'s plate off
+    their runs, now that ``run`` has its own: a plate keeps one video, its
+    latest print's. The one the archive shows too, when it is one of them.
+    Returns their paths, for the caller to delete once committed; a path
+    another run (of another plate) still holds is left. Does not commit."""
+    from backend.app.models.archive import PrintArchive
+
+    if run.archive_id is None or run.plate_id is None or run.started_at is None:
+        return []
+    runs = list(
+        await db.scalars(
+            select(PrintLogEntry)
+            .where(PrintLogEntry.archive_id == run.archive_id, PrintLogEntry.timelapse_path.is_not(None))
+            .execution_options(**{INCLUDE_OPEN_RUNS: True})
+        )
+    )
+    earlier = [
+        r
+        for r in runs
+        if r.id != run.id
+        and r.plate_id == run.plate_id
+        and r.started_at is not None
+        and r.started_at < run.started_at
+        and r.timelapse_path != run.timelapse_path
+    ]
+    kept = {r.timelapse_path for r in runs if r not in earlier}
+    paths = {r.timelapse_path for r in earlier} - kept
+    for r in earlier:
+        r.timelapse_path = None
+    archive = await db.get(PrintArchive, run.archive_id)
+    if archive is not None and archive.timelapse_path in paths:
+        archive.timelapse_path = None
+        archive.timelapse_plate_id = None
+    return sorted(p for p in paths if p)
+
+
 def add_run_photo(run: PrintLogEntry | None, filename: str, *, first: bool = False) -> None:
     """Put a finish photo on its run's list too: last, or ``first``."""
     if run is None:

backend/app/services/print_scheduler.py

@@ -1378,12 +1378,6 @@ class PrintScheduler:
         # event-loop thread, so this dict needs no lock.
         # item_id -> (task, printer_id)
         self._inflight: dict[int, tuple[asyncio.Task, int | None]] = {}
-        # Archives a library dispatch still in flight is adding its run to:
-        # archive_id -> item_id. Uploads run in parallel, and an archive holds
-        # one active run, so a second dispatch of the same file must not take
-        # it before the first one's print shows as running. Released by
-        # `_dispatch_one` when the dispatch ends; same threading as above.
-        self._archive_claims: dict[int, int] = {}
         # Expected prints registered by `_start_print` that have not yet had a
         # print command sent. Populated at registration, dropped once
         # `start_print()` succeeds, and rolled back by `_dispatch_one` on every
@@ -2853,9 +2847,6 @@ class PrintScheduler:
                 # Likewise the print user it set: the printer's next print is
                 # not this item's.
                 self._rollback_unconfirmed_print_user(item_id)
-                # A sent print now shows as running on its archive; an unsent
-                # one leaves it free again.
-                self._release_archive_claims(item_id)
                 # Mirror the pre-#1625 background-dispatch lifecycle: a
                 # reservation survives only after start_print() accepted the
                 # command. Failure, cancellation, deferral, and exceptions all
@@ -8063,24 +8054,33 @@ class PrintScheduler:
 
         return prev_item.status in ("completed", "cancelled")
 
-    def _release_archive_claims(self, item_id: int) -> None:
-        for archive_id in [a for a, owner in self._archive_claims.items() if owner == item_id]:
-            del self._archive_claims[archive_id]
+    @staticmethod
+    async def _printing_elsewhere(db: AsyncSession, archive_id: int) -> bool:
+        """Whether a print of the archive is running (its run open)."""
+        from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry
+
+        return (
+            await db.scalar(
+                select(PrintLogEntry.id)
+                .where(PrintLogEntry.archive_id == archive_id, PrintLogEntry.status == RUN_OPEN)
+                .limit(1)
+                .execution_options(**{INCLUDE_OPEN_RUNS: True})
+            )
+            is not None
+        )
 
     async def _claim_reusable_archive(
         self, db: AsyncSession, item: PrintQueueItem, file_path: Path
     ) -> PrintArchive | None:
-        """The archive a library dispatch adds its run to, claimed for it, or
-        None for a new one.
+        """The archive a library dispatch adds its run to, or None for a new one.
 
         The same file of the same owner, in the same project, archived from
         this library file or from none (one since deleted, or a print started
-        elsewhere), with its copy of the file still on disk. Only an idle
-        archive: it holds one active run, so while one of its prints is
-        queued, being sent, running, or still finishing (its timelapse),
-        another print of the file gets an archive of its own, as before.
+        elsewhere), with its copy of the file still on disk. Also while it is
+        printing, queued or finishing elsewhere: each print of an archive is
+        its own run, with its own start, end, charge, energy, outcome,
+        timelapse and photos (archive redesign), as for the copies of a batch.
         """
-        from backend.app.main import _active_prints, _archive_runs_finishing
         from backend.app.services.archive import ArchiveService
 
         # Off the file as it is now: the stored hash is not refreshed when the
@@ -8103,42 +8103,13 @@ class PrintScheduler:
                     project,
                     or_(PrintArchive.library_file_id == item.library_file_id, PrintArchive.library_file_id.is_(None)),
                     PrintArchive.deleted_at.is_(None),
-                    PrintArchive.status != "printing",
                 )
                 .order_by(PrintArchive.id.desc())
             )
         ).all()
-        if not candidates:
-            return None
-        # A dispatch sets the item's archive while the item is still pending
-        # and keeps it through printing, so this also covers a print being
-        # sent or waiting to start. (The expected-print table can't say that:
-        # a print start leaves one of its names there for two hours.)
-        queued = set(
-            (
-                await db.scalars(
-                    select(PrintQueueItem.archive_id).where(
-                        PrintQueueItem.archive_id.in_([c.id for c in candidates]),
-                        PrintQueueItem.status.in_(("pending", "printing")),
-                        PrintQueueItem.id != item.id,
-                    )
-                )
-            ).all()
-        )
         for candidate in candidates:
-            if not candidate.file_path or not (settings.base_dir / candidate.file_path).is_file():
-                continue
-            # No await from here to the claim: two parallel dispatches of the
-            # file cannot both take it.
-            if (
-                candidate.id in queued
-                or candidate.id in self._archive_claims
-                or candidate.id in _active_prints.values()
-                or candidate.id in _archive_runs_finishing
-            ):
-                continue
-            self._archive_claims[candidate.id] = item.id
-            return candidate
+            if candidate.file_path and (settings.base_dir / candidate.file_path).is_file():
+                return candidate
         return None
 
     async def _repoint_siblings_at_archive(
@@ -9099,11 +9070,16 @@ class PrintScheduler:
                     if archive.library_file_id is None:
                         archive.library_file_id = item.library_file_id
                     # This run's, as a new archive would have them: the
-                    # archive shows its latest run's plate, and billing falls
-                    # back to the archive's cost centre.
-                    if item.plate_id is not None:
-                        archive.plate_id = item.plate_id
-                    archive.cost_center_id = item.cost_center_id
+                    # archive shows its latest run's plate, and a print
+                    # without a queue item falls back to the archive's plate
+                    # and cost centre. Not while another print of it runs: they
+                    # are that print's (this one's run has its own, from the
+                    # dispatch), and the archive takes the latest run's plate
+                    # once none prints (print_log.mirror_latest_run).
+                    if not await self._printing_elsewhere(db, archive.id):
+                        if item.plate_id is not None:
+                            archive.plate_id = item.plate_id
+                        archive.cost_center_id = item.cost_center_id
                 else:
                     archive_service = ArchiveService(db)
                     archive = await archive_service.archive_print(

backend/tests/integration/test_run_lifecycle.py

@@ -12,6 +12,7 @@ made here on purpose.
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime
 from types import SimpleNamespace
 
 import pytest
@@ -69,10 +70,8 @@ def _clean():
     from backend.app.services.printer_manager import printer_manager
 
     main_module._print_plate_ids.clear()
-    main_module._archive_runs_finishing.clear()
     yield
     main_module._print_plate_ids.clear()
-    main_module._archive_runs_finishing.clear()
     printer_manager._awaiting_plate_clear.clear()
 
 
@@ -1525,3 +1524,53 @@ async def test_a_restart_keeps_the_first_printers_run_while_another_plate_prints
     [after] = [r for r in await _all_runs(db_session, farm.archive) if r.printer_id == farm.printer]
     assert (after.id, after.started_at, after.billing_run_id) == kept
     assert after.energy_start_kwh == 10.0
+
+
+async def test_a_queued_prints_cost_centre_is_its_queue_items_none_included(db_session, farm):
+    """The same file printing on another printer under a cost centre left it
+    on the archive: a queued print without one is not charged to it."""
+    from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.cost_center_id = farm.center
+    archive.extra_data = {"filament_grams_total": 100.0}
+    item = PrintQueueItem(printer_id=farm.printer, archive_id=farm.archive, status="printing", position=1)
+    db_session.add(item)
+    await db_session.flush()
+    run = PrintLogEntry(
+        archive_id=farm.archive,
+        printer_id=farm.printer,
+        status="printing",
+        billing_run_id="run-1",
+        queue_item_id=item.id,
+        cost_center_id=None,
+    )
+    db_session.add(run)
+    await db_session.commit()
+
+    assert await apply_print_charge_for_archive(
+        db_session,
+        farm.archive,
+        charged_user_id=farm.user,
+        run_entry_id=run.id,
+        run_status="completed",
+        filament_usage=(10.0, 100.0),
+    )
+    await db_session.commit()
+
+    [charge] = await _charges(db_session)
+    assert charge.cost_center_id is None
+
+
+async def test_a_run_opened_for_a_queue_item_takes_its_cost_centre_none_included(db_session, farm):
+    from backend.app.services.print_log import open_run
+
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.cost_center_id = farm.center
+    archive.started_at = datetime(2026, 10, 31, 9)
+    db_session.add(PrintQueueItem(printer_id=farm.printer, archive_id=farm.archive, status="printing", position=1))
+    await db_session.commit()
+
+    run = await open_run(db_session, archive, printer_id=farm.printer)
+
+    assert run is not None and run.cost_center_id is None

backend/tests/integration/test_timelapse_per_run.py

@@ -162,6 +162,46 @@ class TestAttach:
         first = await _row(db_session, PrintLogEntry, two_plates.first)
         assert first.timelapse_path == f"{ARCHIVE_DIR}/layer_timelapse_2.mp4"
 
+    async def test_a_plates_video_replaces_its_earlier_prints(self, db_session, two_plates):
+        """Plate 1 printed again on B while A's copy ran: once B's video is
+        in, A's goes (a plate keeps its latest print's)."""
+        (two_plates.files / ARCHIVE_DIR / "video_1.mp4").write_bytes(b"one")
+        await _set(db_session, PrintLogEntry, two_plates.first, timelapse_path=f"{ARCHIVE_DIR}/video_1.mp4")
+        later = PrintLogEntry(
+            archive_id=two_plates.archive,
+            printer_id=two_plates.b,
+            plate_id=1,
+            status="completed",
+            started_at=T2 + timedelta(hours=1),
+            photos=[],
+        )
+        db_session.add(later)
+        await db_session.commit()
+        later_id = later.id
+
+        assert await _attach(two_plates.archive, b"again", "video_3.mp4", plate_id=1, run_id=later_id)
+
+        assert (await _row(db_session, PrintLogEntry, two_plates.first)).timelapse_path is None
+        assert not (two_plates.files / ARCHIVE_DIR / "video_1.mp4").exists()
+        assert (await _row(db_session, PrintLogEntry, later_id)).timelapse_path == f"{ARCHIVE_DIR}/video_3.mp4"
+
+    async def test_an_earlier_prints_late_video_leaves_the_later_ones(self, db_session, two_plates):
+        later = PrintLogEntry(
+            archive_id=two_plates.archive,
+            printer_id=two_plates.b,
+            plate_id=1,
+            status="completed",
+            started_at=T2 + timedelta(hours=1),
+            timelapse_path=f"{ARCHIVE_DIR}/video_3.mp4",
+        )
+        (two_plates.files / ARCHIVE_DIR / "video_3.mp4").write_bytes(b"again")
+        db_session.add(later)
+        await db_session.commit()
+
+        assert await _attach(two_plates.archive, b"one", "video_1.mp4", plate_id=1, run_id=two_plates.first)
+
+        assert (two_plates.files / ARCHIVE_DIR / "video_3.mp4").exists()
+
 
 class TestConversion:
     async def _convert(self, archive_id, source):
@@ -297,26 +337,38 @@ class TestScan:
 
         assert (await _row(db_session, PrintLogEntry, two_plates.first)).timelapse_path == f"{ARCHIVE_DIR}/new.mp4"
 
-    async def _later_print(self, db_session, plates, *, printer_id, plate_id):
+    async def _later_print(self, db_session, plates, *, printer_id, plate_id, video=None):
         db_session.add(
             PrintLogEntry(
                 archive_id=plates.archive,
                 printer_id=printer_id,
                 plate_id=plate_id,
-                status=RUN_OPEN,
+                status=RUN_OPEN if video is None else "completed",
                 started_at=T2 + timedelta(hours=1),
+                timelapse_path=video,
             )
         )
         await db_session.commit()
 
-    async def test_a_reprint_of_its_plate_ends_it(self, db_session, two_plates):
+    async def test_a_reprint_of_its_plate_with_its_video_ends_it(self, db_session, two_plates):
         """The reprint's video is the plate's one (it replaces this one)."""
-        await self._later_print(db_session, two_plates, printer_id=two_plates.b, plate_id=1)
+        await self._later_print(
+            db_session, two_plates, printer_id=two_plates.b, plate_id=1, video=f"{ARCHIVE_DIR}/later.mp4"
+        )
 
         await self._scan(two_plates, [[_video("new.mp4")]], baseline=set())
 
         assert (await _row(db_session, PrintLogEntry, two_plates.first)).timelapse_path is None
 
+    async def test_a_reprint_of_its_plate_still_printing_elsewhere_does_not_end_it(self, db_session, two_plates):
+        """The same plate printing on two printers at once: until the later
+        print has a video, which it may never get, this one keeps its own."""
+        await self._later_print(db_session, two_plates, printer_id=two_plates.b, plate_id=1)
+
+        await self._scan(two_plates, [[_video("new.mp4")]], baseline=set())
+
+        assert (await _row(db_session, PrintLogEntry, two_plates.first)).timelapse_path == f"{ARCHIVE_DIR}/new.mp4"
+
     async def test_the_next_print_on_its_printer_ends_it(self, db_session, two_plates):
         """A video landing now can be that print's."""
         await self._later_print(db_session, two_plates, printer_id=two_plates.a, plate_id=3)
@@ -406,7 +458,7 @@ class TestLayerTimelapse:
         assert not stitched.exists()
         assert (await _row(db_session, PrintArchive, two_plates.archive)).timelapse_path is None
 
-    async def test_a_reprint_of_its_plate_drops_it(self, db_session, two_plates):
+    async def test_a_reprint_of_its_plate_with_its_video_drops_it(self, db_session, two_plates):
         from backend.app.main import _attach_layer_timelapse
 
         db_session.add(
@@ -414,8 +466,9 @@ class TestLayerTimelapse:
                 archive_id=two_plates.archive,
                 printer_id=two_plates.b,
                 plate_id=1,
-                status=RUN_OPEN,
+                status="completed",
                 started_at=T2 + timedelta(hours=1),
+                timelapse_path=f"{ARCHIVE_DIR}/later.mp4",
             )
         )
         await db_session.commit()

backend/tests/unit/test_archive_run_finishing.py

@@ -1,86 +0,0 @@
-"""An archive stays marked while its last run's timelapse work goes on.
-
-The timelapse scan and the layer timelapse stitch run after the print
-completed and tell their run from a later one by the archive's start time. A
-queue dispatch of the same file adds its run to the archive only once that
-work is over (``_claim_reusable_archive``), or the work would stop short and
-the video never be attached.
-"""
-
-import asyncio
-from unittest.mock import patch
-
-import pytest
-
-from backend.app import main as main_module
-from backend.tests.unit.test_run_estimate_plate import _complete
-
-
-@pytest.fixture(autouse=True)
-def _clean_state():
-    main_module._archive_runs_finishing.clear()
-    main_module._active_prints.clear()
-    main_module._print_plate_ids.clear()
-    yield
-    main_module._archive_runs_finishing.clear()
-    main_module._active_prints.clear()
-    main_module._print_plate_ids.clear()
-
-
-@pytest.mark.asyncio
-async def test_marked_from_the_spawn_until_the_work_ends():
-    release = asyncio.Event()
-
-    async def work():
-        await release.wait()
-
-    wrapped = main_module._mark_run_finishing(7, work())
-    assert main_module._archive_runs_finishing == {7: 1}
-    second = main_module._mark_run_finishing(7, asyncio.sleep(0))
-    await second
-    assert main_module._archive_runs_finishing == {7: 1}, "the other task is still running"
-
-    task = asyncio.create_task(wrapped)
-    release.set()
-    await task
-    assert main_module._archive_runs_finishing == {}
-
-
-@pytest.mark.asyncio
-async def test_a_failing_task_unmarks_too():
-    async def work():
-        raise RuntimeError("scan failed")
-
-    with pytest.raises(RuntimeError):
-        await main_module._mark_run_finishing(7, work())
-    assert main_module._archive_runs_finishing == {}
-
-
-def test_the_timelapse_scan_marks_its_archive():
-    spawned = []
-    with patch("backend.app.main.spawn_background_task", side_effect=lambda coro, name: spawned.append(coro)):
-        main_module._schedule_timelapse_scan(12, 3, plate_id=None)
-    assert main_module._archive_runs_finishing == {12: 1}
-    for coro in spawned:
-        coro.close()
-
-
-@pytest.mark.asyncio
-async def test_the_layer_timelapse_of_a_completed_print_marks_its_archive(
-    test_engine, db_session, printer_factory, archive_factory
-):
-    printer = await printer_factory()
-    archive = await archive_factory(printer.id, filename="multi.gcode.3mf", status="printing", with_run=False)
-    main_module._active_prints[(printer.id, "multi.gcode.3mf")] = archive.id
-    spawned: list[str] = []
-    real_spawn = main_module.spawn_background_task
-
-    def spawn(coro, name=None, **kwargs):
-        if name == "background-layer-timelapse":
-            spawned.append(main_module._archive_runs_finishing.get(archive.id, 0))
-        return real_spawn(coro, name=name, **kwargs)
-
-    with patch("backend.app.main.spawn_background_task", side_effect=spawn):
-        await _complete(test_engine, printer.id)
-
-    assert spawned == [1]

backend/tests/unit/test_forget_printer.py

@@ -26,7 +26,6 @@ NOT_FORGOTTEN = {
     "_print_cost_center_ids": "keyed by (printer, archive); checked below",
     "_print_plate_ids": "keyed by (printer, archive); checked below",
     "_owned_print_archive_ids": "keyed by archive (dropped with the printer's expected/active prints)",
-    "_archive_runs_finishing": "keyed by archive, cleared by the task it counts when that task ends",
     "_ams_assignment_locks": "a lock carries nothing over; popping one a task holds would let a second be made",
     "_fallback_recovery_locks": "a lock carries nothing over (see its comment on popping)",
     "_HMS_FAILURE_REASONS": "constant",

backend/tests/unit/test_library_dispatch_reuses_archive.py

@@ -6,10 +6,10 @@ same file, flagged as duplicates of each other. The dispatch now adds its run
 to the archive the file already has: same file, same owner, same project, from
 this library file (or one since deleted).
 
-An archive holds one active run, and uploads run in parallel, so it is only
-taken while idle: not printing, not sent or waiting to start, not queued, not
-claimed by another dispatch still uploading. A print of the file running at
-the same time gets an archive of its own, as before.
+Also while the file prints, is queued or still finishing elsewhere: each
+print of an archive is its own run, with its own start, end, charge, energy,
+outcome, timelapse and photos (archive redesign), as the copies of a batch
+always were.
 """
 
 import hashlib
@@ -41,11 +41,9 @@ HASH = hashlib.sha256(CONTENT).hexdigest()
 def _clean_print_state():
     main_module._active_prints.clear()
     main_module._expected_prints.clear()
-    main_module._archive_runs_finishing.clear()
     yield
     main_module._active_prints.clear()
     main_module._expected_prints.clear()
-    main_module._archive_runs_finishing.clear()
 
 
 @pytest.fixture
@@ -153,7 +151,7 @@ async def test_an_archive_whose_library_file_was_deleted_is_reused(case):
 
 
 @pytest.mark.asyncio
-async def test_the_newest_idle_archive_is_taken(case):
+async def test_the_newest_archive_is_taken(case):
     await case.archive()
     newest = await case.archive(tags="newest")
     assert await _claim(case, PrintScheduler(), await case.item()) == newest
@@ -169,7 +167,6 @@ async def test_the_newest_idle_archive_is_taken(case):
         pytest.param({}, {"project_id": 3}, id="project-print-of-a-loose-archive"),
         pytest.param({"library_file_id": 999}, {}, id="another-library-file"),
         pytest.param({"deleted_at": datetime(2026, 10, 1)}, {}, id="deleted"),
-        pytest.param({"status": "printing"}, {}, id="printing"),
         pytest.param({"on_disk": False}, {}, id="its-copy-is-gone"),
     ],
 )
@@ -180,23 +177,17 @@ async def test_no_archive_that_is_not_this_files_or_not_usable(case, archive_ove
 
 
 @pytest.mark.asyncio
-async def test_not_while_another_queue_item_prints_or_waits_on_it(case):
+async def test_also_while_another_queue_item_prints_or_waits_on_it(case):
     existing = await case.archive()
     await case.item(archive_id=existing, library_file_id=None, status="pending")
-    assert await _claim(case, PrintScheduler(), await case.item()) is None
+    assert await _claim(case, PrintScheduler(), await case.item()) == existing
 
 
 @pytest.mark.asyncio
-async def test_not_while_one_of_its_prints_runs_or_finishes(case):
-    existing = await case.archive()
+async def test_also_while_one_of_its_prints_runs(case):
+    existing = await case.archive(status="printing")
     main_module._active_prints[(case.printer_id, "planter.3mf")] = existing
-    assert await _claim(case, PrintScheduler(), await case.item()) is None
-
-    # Completed, but its timelapse is still being looked for: that scan tells
-    # its run by the archive's start time, which a new run would change.
-    main_module._active_prints.clear()
-    main_module._archive_runs_finishing[existing] = 1
-    assert await _claim(case, PrintScheduler(), await case.item()) is None
+    assert await _claim(case, PrintScheduler(), await case.item()) == existing
 
 
 @pytest.mark.asyncio
@@ -209,16 +200,13 @@ async def test_a_name_left_from_an_earlier_dispatch_does_not_block_it(case):
 
 
 @pytest.mark.asyncio
-async def test_two_dispatches_in_flight_never_share_it(case):
+async def test_two_dispatches_in_flight_share_it(case):
+    """The same file sent to two printers at once: one archive, two runs."""
     existing = await case.archive()
     scheduler = PrintScheduler()
     first, second = await case.item(), await case.item()
 
     assert await _claim(case, scheduler, first) == existing
-    assert await _claim(case, scheduler, second) is None
-
-    # The first dispatch ended: the archive is free for the next print.
-    scheduler._release_archive_claims(first)
     assert await _claim(case, scheduler, second) == existing
 
 
@@ -304,31 +292,42 @@ async def test_a_reprint_of_another_plate_shows_that_plate(case):
     assert archive.plate_id == 3
 
 
-@pytest.mark.parametrize("fails", [False, True], ids=["sent", "failed"])
 @pytest.mark.asyncio
-async def test_the_dispatch_gives_its_claim_back_when_it_ends(fails):
-    """Sent, its print shows as running; failed, the archive is free again.
-    Another dispatch's claim is left alone."""
-    scheduler = PrintScheduler()
+async def test_the_file_sent_to_two_printers_at_once_has_one_archive(case):
+    """The first copy is printing when the second is sent: both are the
+    archive's, each printer's print its own run."""
+    existing = await case.archive()
+    async with case.session_maker() as db:
+        other = Printer(name="P1S", serial_number="SERIAL-2", ip_address="127.0.0.2", access_code="code", model="P1S")
+        db.add(other)
+        await db.commit()
+        other_id = other.id
+    archive_print = AsyncMock()
 
-    async def start_print(db, item):
-        scheduler._archive_claims[298] = 597
-        if fails:
-            raise RuntimeError("upload failed")
-
-    scheduler._archive_claims[300] = 600
-    db = MagicMock()
-    db.get = AsyncMock(return_value=MagicMock(id=597))
-    session = MagicMock()
-    session.__aenter__ = AsyncMock(return_value=db)
-    session.__aexit__ = AsyncMock(return_value=False)
-    with ExitStack() as stack:
-        stack.enter_context(patch("backend.app.services.print_scheduler.async_session", return_value=session))
-        stack.enter_context(patch.object(scheduler, "_claim_for_dispatch", AsyncMock(return_value=True)))
-        stack.enter_context(patch.object(scheduler, "_start_print", side_effect=start_print))
-        stack.enter_context(patch.object(scheduler, "_clear_dispatch_claim", AsyncMock()))
-        if fails:
-            stack.enter_context(pytest.raises(RuntimeError))
-        await scheduler._dispatch_one(597)
-
-    assert scheduler._archive_claims == {300: 600}
+    first, _ = await _start(case, await case.item(), archive_print)
+    async with case.session_maker() as db:
+        (await db.get(PrintArchive, existing)).status = "printing"
+        (await db.get(PrintQueueItem, first.id)).status = "printing"
+        await db.commit()
+    main_module._active_prints[(case.printer_id, "planter.3mf")] = existing
+    second, _ = await _start(case, await case.item(printer_id=other_id), archive_print)
+
+    archive_print.assert_not_called()
+    assert (first.archive_id, second.archive_id) == (existing, existing)
+
+
+@pytest.mark.asyncio
+async def test_while_it_prints_elsewhere_its_plate_and_cost_centre_stay_that_prints(case):
+    """A print without a queue item falls back to the archive's plate and cost
+    centre at its end; the new dispatch's run has its own."""
+    from backend.app.models.print_log import RUN_OPEN, PrintLogEntry
+
+    existing = await case.archive(cost_center_id=5, plate_id=1, status="printing")
+    async with case.session_maker() as db:
+        db.add(PrintLogEntry(archive_id=existing, printer_id=case.printer_id, status=RUN_OPEN, plate_id=1))
+        await db.commit()
+
+    item, archive = await _start(case, await case.item(plate_id=2), AsyncMock())
+
+    assert item.archive_id == existing
+    assert (archive.plate_id, archive.cost_center_id) == (1, 5)