Archives: find each printer's running print by its run (archive redesign P1b)

- Plate, AMS mapping and cost centre registered at dispatch are kept
  per printer and archive: plates of one archive on two printers each
  keep their own trays, plate and cost centre
- A print's end unbinds only its own printer, and after a restart finds
  its archive by the run of the job ending (never another job's run,
  nor one of job "0")
- The running-print lookup, the kill switch and restart resume find the
  run by its job id first; the archive row holds the latest start's
- An end without its archive no longer drops the printer's open runs

.basedpyright/baseline.json

@@ -1795,22 +1795,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 73,
-                    "endColumn": 75,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 61,
-                    "endColumn": 63,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {
@@ -1819,22 +1803,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 69,
-                    "endColumn": 71,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 57,
-                    "endColumn": 59,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {

backend/app/main.py

@@ -511,19 +511,21 @@ _expected_prints: dict[tuple[int, str], int] = {}
 # the printer's screen included, so it is no proof of ownership on its own.
 _owned_print_archive_ids: set[int] = set()
 
-# Track AMS mapping for prints: {archive_id: [global_tray_id_per_slot]}
+# What a dispatch registered for its print, keyed (printer_id, archive_id):
+# plates of one archive print on two printers at once, each with its own.
+# Track AMS mapping for prints: {(printer_id, archive_id): [global_tray_id_per_slot]}
 # Used by usage tracker to map 3MF slots to physical AMS trays
-_print_ams_mappings: dict[int, list[int]] = {}
+_print_ams_mappings: dict[tuple[int, int], list[int]] = {}
 
-# Track cost center selection for the current print run: {archive_id: cost_center_id}
-_print_cost_center_ids: dict[int, int] = {}
+# Track cost center selection for the current print run: {(printer_id, archive_id): cost_center_id}
+_print_cost_center_ids: dict[tuple[int, int], int] = {}
 
-# Track plate_id for prints from multi-plate 3MFs: {archive_id: plate_id}
+# Track plate_id for prints from multi-plate 3MFs: {(printer_id, archive_id): plate_id}
 # Used by usage tracker to scope 3MF parsing to the dispatched plate (#1697).
 # Populated by direct-Print and queue dispatch paths; queue prints also have a
 # redundant queue-item lookup in on_print_start so this dict isn't load-bearing
 # for the queue path. Cleared on print completion or TTL eviction.
-_print_plate_ids: dict[int, int] = {}
+_print_plate_ids: dict[tuple[int, int], int] = {}
 
 # Track progress milestones for notifications: {printer_id: last_milestone_notified}
 # Milestones are 25, 50, 75. Value of 0 means no milestone notified yet for current print.
@@ -938,9 +940,19 @@ async def running_print_archive(db, printer_id: int, state: PrinterState, *, pri
     never a deleted one.
     """
     from backend.app.models.archive import PrintArchive
+    from backend.app.services.print_log import open_run_on_printer
 
     subtask_id = str(getattr(state, "subtask_id", "") or "").strip()
     if subtask_id not in ("", "0"):
+        # The run open on the printer for this job: the archive row is shared
+        # by every plate of it, its printer and job id the latest start's.
+        run = await open_run_on_printer(db, printer_id, subtask_id=subtask_id)
+        if run is not None and run.archive_id is not None:
+            archive = await db.scalar(
+                select(PrintArchive).where(PrintArchive.id == run.archive_id, PrintArchive.deleted_at.is_(None))
+            )
+            if archive is not None:
+                return archive
         query = select(PrintArchive).where(
             PrintArchive.printer_id == printer_id,
             PrintArchive.subtask_id == subtask_id,
@@ -986,17 +998,23 @@ def _bind_active_print(
             _active_prints[(printer_id, name)] = archive_id
 
 
-async def _discard_open_runs_of(printer_id: int) -> None:
-    """Drop the printer's open runs when its print ended with no archive to
-    end them on. Its own session; never in the way of the completion."""
+async def _archive_of_open_run(printer_id: int, subtask_id: str | None = None) -> int | None:
+    """The archive of the run open on the printer, if it is still there.
+    With ``subtask_id``, only a run of that job."""
     try:
-        from backend.app.services.print_log import discard_open_runs
+        from backend.app.models.archive import PrintArchive
+        from backend.app.services.print_log import open_run_on_printer
 
         async with async_session() as db:
-            await discard_open_runs(db, printer_id)
-            await db.commit()
+            run = await open_run_on_printer(db, printer_id, subtask_id=subtask_id)
+            if run is None or run.archive_id is None:
+                return None
+            return await db.scalar(
+                select(PrintArchive.id).where(PrintArchive.id == run.archive_id, PrintArchive.deleted_at.is_(None))
+            )
     except Exception as e:
-        logging.getLogger(__name__).warning("[RUN] Could not drop the open runs of printer %s: %s", printer_id, e)
+        logging.getLogger(__name__).warning("[RUN] Could not read the open run of printer %s: %s", printer_id, e)
+        return None
 
 
 async def _open_run_for(archive_id: int, printer_id: int) -> None:
@@ -1015,7 +1033,11 @@ async def _open_run_for(archive_id: int, printer_id: int) -> None:
                 return
             printer_name = await db.scalar(select(Printer.name).where(Printer.id == printer_id))
             await open_run(
-                db, archive, printer_id=printer_id, printer_name=printer_name, plate_id=_print_plate_ids.get(archive_id)
+                db,
+                archive,
+                printer_id=printer_id,
+                printer_name=printer_name,
+                plate_id=_print_plate_ids.get((printer_id, archive_id)),
             )
             await db.commit()
     except Exception as e:
@@ -1097,17 +1119,25 @@ async def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState, db
             return None
         archive = await _printing_archive_without_job_id(db, printer_id, reported, dispatched_only=True)
     else:
-        result = await db.execute(
-            select(PrintArchive)
-            .where(
-                PrintArchive.printer_id == printer_id,
-                PrintArchive.status == "printing",
-                PrintArchive.subtask_id == subtask_id,
+        from backend.app.services.print_log import open_run_on_printer
+
+        # This job's run open on the printer: the archive is shared by every
+        # plate of it, and its status, printer and job id are the latest
+        # start's, which may be another printer's.
+        run = await open_run_on_printer(db, printer_id, subtask_id=subtask_id)
+        archive = await db.get(PrintArchive, run.archive_id) if run is not None and run.archive_id else None
+        if archive is None:
+            result = await db.execute(
+                select(PrintArchive)
+                .where(
+                    PrintArchive.printer_id == printer_id,
+                    PrintArchive.status == "printing",
+                    PrintArchive.subtask_id == subtask_id,
+                )
+                .order_by(PrintArchive.created_at.desc())
+                .limit(1)
             )
-            .order_by(PrintArchive.created_at.desc())
-            .limit(1)
-        )
-        archive = result.scalar_one_or_none()
+            archive = result.scalar_one_or_none()
 
     # An archive row on its own proves nothing: `on_print_start` archives every
     # print it observes, including ones started from Bambu Studio or Handy, and
@@ -1292,9 +1322,9 @@ async def forget_printer(printer_id: int) -> None:
             archive_ids.add(tracked.pop(key))
     for archive_id in archive_ids:
         _owned_print_archive_ids.discard(archive_id)
-        _print_ams_mappings.pop(archive_id, None)
-        _print_cost_center_ids.pop(archive_id, None)
-        _print_plate_ids.pop(archive_id, None)
+    for registered in (_print_ams_mappings, _print_cost_center_ids, _print_plate_ids):
+        for key in [key for key in registered if key[0] == printer_id]:
+            del registered[key]
     for tracked in (_expected_print_creators, _expected_print_registered_at):
         for key in [key for key in tracked if key[0] == printer_id]:
             del tracked[key]
@@ -1379,19 +1409,20 @@ def register_expected_print(
     # the printed plate (#1697). They describe this dispatch: one that has none
     # drops what an earlier dispatch of the archive left (sent, never started),
     # unless a plate of the archive is printing now and still needs them.
-    archive_printing = archive_id in set(_active_prints.values())
+    dispatch = (printer_id, archive_id)
+    archive_printing = _printing_here(printer_id, archive_id)
     if ams_mapping is not None:
-        _print_ams_mappings[archive_id] = ams_mapping
+        _print_ams_mappings[dispatch] = ams_mapping
     elif not archive_printing:
-        _print_ams_mappings.pop(archive_id, None)
+        _print_ams_mappings.pop(dispatch, None)
     if cost_center_id is not None:
-        _print_cost_center_ids[archive_id] = cost_center_id
+        _print_cost_center_ids[dispatch] = cost_center_id
     elif not archive_printing:
-        _print_cost_center_ids.pop(archive_id, None)
+        _print_cost_center_ids.pop(dispatch, None)
     if plate_id is not None:
-        _print_plate_ids[archive_id] = plate_id
+        _print_plate_ids[dispatch] = plate_id
     elif not archive_printing:
-        _print_plate_ids.pop(archive_id, None)
+        _print_plate_ids.pop(dispatch, None)
     # Store created_by_id so the user start email can be sent even when the archive
     # itself has no created_by_id (e.g. library-file-based queue prints)
     if created_by_id is not None:
@@ -1428,8 +1459,8 @@ def unregister_expected_print(printer_id: int, filename: str, archive_id: int) -
 
     Mirrors the sweep's rules, including the one that is easy to get wrong:
     ``_print_ams_mappings`` / ``_print_plate_ids`` / ``_print_cost_center_ids``
-    are keyed by archive, not by file, so they may only be dropped once no
-    live key still points at that archive.
+    are keyed by printer and archive, not by file, so they may only be dropped
+    once no live key of the printer still points at that archive.
     """
     keys = [(printer_id, filename)]
     if filename.endswith(".3mf"):
@@ -1444,12 +1475,10 @@ def unregister_expected_print(printer_id: int, filename: str, archive_id: int) -
         _expected_print_creators.pop(key, None)
         _expected_print_registered_at.pop(key, None)
 
-    # Still needed by another expected print of the archive, or a plate of it
-    # that is printing, as in the sweep.
-    if archive_id not in set(_expected_prints.values()) | set(_active_prints.values()):
-        _print_ams_mappings.pop(archive_id, None)
-        _print_cost_center_ids.pop(archive_id, None)
-        _print_plate_ids.pop(archive_id, None)
+    # Still needed by another expected print of the archive on the printer,
+    # or a plate of it printing there, as in the sweep.
+    if not _expected_or_printing_here(printer_id, archive_id):
+        _drop_dispatch(printer_id, archive_id)
 
     if removed:
         logging.getLogger(__name__).info(
@@ -1492,15 +1521,35 @@ def _compute_run_filament_grams(
     return None
 
 
-def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | None:
+def _printing_here(printer_id: int, archive_id: int) -> bool:
+    """Whether a plate of the archive is printing on the printer."""
+    return any(key[0] == printer_id and bound == archive_id for key, bound in _active_prints.items())
+
+
+def _expected_or_printing_here(printer_id: int, archive_id: int) -> bool:
+    """Whether a print of the archive is expected or printing on the printer."""
+    return _printing_here(printer_id, archive_id) or any(
+        key[0] == printer_id and expected == archive_id for key, expected in _expected_prints.items()
+    )
+
+
+def _drop_dispatch(printer_id: int, archive_id: int) -> None:
+    """Forget what the dispatch of the archive to the printer registered."""
+    dispatch = (printer_id, archive_id)
+    _print_ams_mappings.pop(dispatch, None)
+    _print_cost_center_ids.pop(dispatch, None)
+    _print_plate_ids.pop(dispatch, None)
+
+
+def _get_start_ams_mapping(data: dict, printer_id: int, archive_id: int | None) -> list[int] | None:
     """Resolve AMS mapping for print start without consuming stored queue/reprint state."""
     stored_ams_mapping = data.get("ams_mapping")
     if not stored_ams_mapping and archive_id:
-        stored_ams_mapping = _print_ams_mappings.get(archive_id)
+        stored_ams_mapping = _print_ams_mappings.get((printer_id, archive_id))
     return stored_ams_mapping
 
 
-def _get_start_plate_id(archive_id: int | None) -> int | None:
+def _get_start_plate_id(printer_id: int, archive_id: int | None) -> int | None:
     """Resolve plate_id for print start without consuming stored direct-Print state.
 
     Direct-Print of a single plate from a multi-plate 3MF registers plate_id in
@@ -1510,7 +1559,7 @@ def _get_start_plate_id(archive_id: int | None) -> int | None:
     """
     if archive_id is None:
         return None
-    return _print_plate_ids.get(archive_id)
+    return _print_plate_ids.get((printer_id, archive_id))
 
 
 def _partial_progress_scale(progress: int | float | None) -> float:
@@ -4153,7 +4202,7 @@ async def _recover_fallback_archive_locked(
                 recovered.file_path,
                 db,
                 printer_manager,
-                plate_id=_get_start_plate_id(recovered.id) or recovered.plate_id,
+                plate_id=_get_start_plate_id(printer_id, recovered.id) or recovered.plate_id,
                 may_start=not price or (recovered.status == "printing" and not is_completing(printer_id)),
             )
         except Exception:
@@ -4172,8 +4221,8 @@ async def _recover_fallback_archive_locked(
                 printer_id,
                 recovered.id,
                 printer_manager,
-                ams_mapping=_get_start_ams_mapping(print_data, recovered.id),
-                plate_id=_get_start_plate_id(recovered.id) or recovered.plate_id,
+                ams_mapping=_get_start_ams_mapping(print_data, printer_id, recovered.id),
+                plate_id=_get_start_plate_id(printer_id, recovered.id) or recovered.plate_id,
                 session_factory=async_session,
             )
         # `archive_updated`, not `archive_created` — the row was already on the
@@ -5120,7 +5169,7 @@ async def on_print_start(printer_id: int, data: dict):
                 archive.timelapse_baseline = None
                 stale_timelapse_relpath = archive.timelapse_path
                 stale_timelapse_plate_id = archive.timelapse_plate_id
-                starting_plate_id = _print_plate_ids.get(expected_archive_id)
+                starting_plate_id = _print_plate_ids.get((printer_id, expected_archive_id))
                 if stale_timelapse_relpath:
                     archive.timelapse_path = None
                     archive.timelapse_plate_id = None
@@ -5248,8 +5297,8 @@ async def on_print_start(printer_id: int, data: dict):
                 # Inject ams_mapping into usage tracker session — the session was created
                 # before expected-print promotion, so it may have ams_mapping=None when
                 # the MQTT request topic subscription failed (common on P1S/A1).
-                _stored_map = _print_ams_mappings.get(expected_archive_id)
-                _stored_plate_id = _print_plate_ids.get(expected_archive_id)
+                _stored_map = _print_ams_mappings.get((printer_id, expected_archive_id))
+                _stored_plate_id = _print_plate_ids.get((printer_id, expected_archive_id))
                 if _stored_map or _stored_plate_id is not None:
                     try:
                         from backend.app.services.usage_tracker import _active_sessions
@@ -5301,8 +5350,8 @@ async def on_print_start(printer_id: int, data: dict):
                         archive.file_path,
                         db,
                         printer_manager,
-                        ams_mapping=_get_start_ams_mapping(data, archive.id),
-                        plate_id=_get_start_plate_id(archive.id),
+                        ams_mapping=_get_start_ams_mapping(data, printer_id, archive.id),
+                        plate_id=_get_start_plate_id(printer_id, archive.id),
                     )
                 except Exception as e:
                     logger.warning("[SPOOLMAN] Failed to store tracking data: %s", e)
@@ -5312,8 +5361,8 @@ async def on_print_start(printer_id: int, data: dict):
                     printer_id,
                     archive.id,
                     printer_manager,
-                    ams_mapping=_get_start_ams_mapping(data, archive.id),
-                    plate_id=_get_start_plate_id(archive.id),
+                    ams_mapping=_get_start_ams_mapping(data, printer_id, archive.id),
+                    plate_id=_get_start_plate_id(printer_id, archive.id),
                     session_factory=async_session,
                 )
 
@@ -5345,13 +5394,28 @@ async def on_print_start(printer_id: int, data: dict):
         from backend.app.models.archive import PrintArchive
 
         existing_archive: PrintArchive | None = None
+        resumed_by_run = False
+
+        # Best match: this job's run, open on the printer since the print
+        # started. The archive row is shared by every plate of it, and its
+        # printer, job id and status are the latest start's, which may be
+        # another printer's.
+        if subtask_id:
+            from backend.app.services.print_log import open_run_on_printer
+
+            run = await open_run_on_printer(db, printer_id, subtask_id=subtask_id)
+            if run is not None and run.archive_id is not None:
+                existing_archive = await db.scalar(
+                    select(PrintArchive).where(PrintArchive.id == run.archive_id, PrintArchive.deleted_at.is_(None))
+                )
+                resumed_by_run = existing_archive is not None
 
         # Preferred match: subtask_id equality. MQTT reports the same subtask_id
         # across a backend restart for the same print, so this is the most
         # reliable way to reattach. Only a row still printing: a cancelled one
         # has been through completion (usage, history, notifications) already,
         # and reopening it would count the print twice.
-        if subtask_id:
+        if subtask_id and existing_archive is None:
             by_id = await db.execute(
                 select(PrintArchive)
                 .where(PrintArchive.printer_id == printer_id)
@@ -5390,7 +5454,7 @@ async def on_print_start(printer_id: int, data: dict):
         if existing_archive:
             # subtask_id match → always resume, regardless of age. Same print,
             # just a backend restart.
-            subtask_match = bool(subtask_id and existing_archive.subtask_id == subtask_id)
+            subtask_match = resumed_by_run or bool(subtask_id and existing_archive.subtask_id == subtask_id)
 
             if subtask_match:
                 logger.info("Resuming archive %s on subtask_id match (%s)", existing_archive.id, subtask_id)
@@ -6101,7 +6165,9 @@ async def on_print_start(printer_id: int, data: dict):
                 # state at print start was sitting right there in `data`.
                 # The slicer's ams_mapping (when present) narrows the result
                 # to slots actually used by the print (#1533).
-                mqtt_filament_meta = _extract_filament_data_from_mqtt(data, _get_start_ams_mapping(data, None))
+                mqtt_filament_meta = _extract_filament_data_from_mqtt(
+                    data, _get_start_ams_mapping(data, printer_id, None)
+                )
 
                 # Create minimal archive entry
                 fallback_archive = PrintArchive(
@@ -6184,8 +6250,8 @@ async def on_print_start(printer_id: int, data: dict):
                         fallback_archive.file_path,
                         db,
                         printer_manager,
-                        ams_mapping=_get_start_ams_mapping(data, fallback_archive.id),
-                        plate_id=_get_start_plate_id(fallback_archive.id),
+                        ams_mapping=_get_start_ams_mapping(data, printer_id, fallback_archive.id),
+                        plate_id=_get_start_plate_id(printer_id, fallback_archive.id),
                     )
                 except Exception as e:
                     logger.debug("[SPOOLMAN] Could not store tracking for fallback archive: %s", e)
@@ -6383,8 +6449,8 @@ async def on_print_start(printer_id: int, data: dict):
                         archive.file_path,
                         db,
                         printer_manager,
-                        ams_mapping=_get_start_ams_mapping(data, archive.id),
-                        plate_id=_get_start_plate_id(archive.id),
+                        ams_mapping=_get_start_ams_mapping(data, printer_id, archive_id),
+                        plate_id=_get_start_plate_id(printer_id, archive_id),
                     )
                 except Exception as e:
                     logger.warning("[SPOOLMAN] Failed to store tracking data: %s", e)
@@ -6394,8 +6460,8 @@ async def on_print_start(printer_id: int, data: dict):
                     printer_id,
                     archive.id,
                     printer_manager,
-                    ams_mapping=_get_start_ams_mapping(data, archive.id),
-                    plate_id=_get_start_plate_id(archive.id),
+                    ams_mapping=_get_start_ams_mapping(data, printer_id, archive_id),
+                    plate_id=_get_start_plate_id(printer_id, archive_id),
                     session_factory=async_session,
                 )
 
@@ -7911,7 +7977,7 @@ def _bound_print(printer_id: int, data: dict) -> tuple[int | None, int | None]:
         archive_id = next((_active_prints[key] for key in keys if key in _active_prints), None)
         if archive_id is None:
             return None, None
-        return archive_id, _print_plate_ids.get(archive_id)
+        return archive_id, _print_plate_ids.get((printer_id, archive_id))
     except Exception:
         logging.getLogger(__name__).debug("Could not read the print binding for printer %s", printer_id, exc_info=True)
         return None, None
@@ -8804,8 +8870,9 @@ async def _close_out_print(printer_id: int, data: dict):
         if archive_id:
             _owned_print_archive_ids.discard(archive_id)
             logger.info("Found archive %s with key %s", archive_id, key)
-            # Also clean up any other keys pointing to this archive
-            keys_to_remove = [k for k, v in _active_prints.items() if v == archive_id]
+            # Also clean up the printer's other keys pointing to this archive;
+            # another printer's are a plate of it still printing there.
+            keys_to_remove = [k for k, v in _active_prints.items() if v == archive_id and k[0] == printer_id]
             for k in keys_to_remove:
                 _active_prints.pop(k, None)
             break
@@ -8814,6 +8881,16 @@ async def _close_out_print(printer_id: int, data: dict):
     # the search matches names by substring.
     bound_archive_id = archive_id
 
+    ending_job = str(data.get("subtask_id") or "").strip()
+    if not archive_id and ending_job not in ("", "0"):
+        # Not bound (a restart since the start): this job's run, open on the
+        # printer. Only by its job id: a run open there can be of the print
+        # running now, while the end is of an earlier one closed late (the
+        # restart's reconcile of a missed end).
+        archive_id = await _archive_of_open_run(printer_id, subtask_id=ending_job)
+        if archive_id:
+            logger.info("Found archive %s by the run of job %s open on printer %s", archive_id, ending_job, printer_id)
+
     if not archive_id:
         # Try to find by filename or subtask_name if not tracked (for prints started before app)
         async with async_session() as db:
@@ -9068,7 +9145,9 @@ async def _close_out_print(printer_id: int, data: dict):
                     billing_planned_grams, billing_base_cost = _plate_scoped_run_estimate(
                         billing_archive,
                         billing_path,
-                        billing_plate_id if billing_plate_id is not None else _get_start_plate_id(archive_id),
+                        billing_plate_id
+                        if billing_plate_id is not None
+                        else _get_start_plate_id(printer_id, archive_id),
                     )
         except Exception as e:
             logger.warning("[FINANCE] Failed to capture planned usage for archive %s: %s", archive_id, e)
@@ -9081,7 +9160,7 @@ async def _close_out_print(printer_id: int, data: dict):
     # Fallback to _print_ams_mappings for queue/reprint (set before print starts).
     # Drained either way, like the plate below: left over, it would be read by
     # the archive's next run, dispatched without a mapping.
-    registered_ams_mapping = _print_ams_mappings.pop(archive_id, None) if archive_id else None
+    registered_ams_mapping = _print_ams_mappings.pop((printer_id, archive_id), None) if archive_id else None
     if not stored_ams_mapping:
         stored_ams_mapping = registered_ams_mapping
 
@@ -9093,7 +9172,7 @@ async def _close_out_print(printer_id: int, data: dict):
     # down to the single plate that was actually printed (#1785).
     notify_plate_id: int | None = None
     if archive_id:
-        notify_plate_id = _print_plate_ids.pop(archive_id, None)
+        notify_plate_id = _print_plate_ids.pop((printer_id, archive_id), None)
 
     # The start estimate must not outlive the real cost written below (COST-4).
     if archive_id:
@@ -9284,7 +9363,6 @@ async def _close_out_print(printer_id: int, data: dict):
                 logger.warning("[NOTIFY-BG] Failed to send notification without archive: %s", e, exc_info=True)
 
         spawn_background_task(_notify_no_archive(), name="notify-no-archive")
-        await _discard_open_runs_of(printer_id)
         return
 
     log_timing("Archive lookup")
@@ -9385,7 +9463,7 @@ async def _close_out_print(printer_id: int, data: dict):
                     usage_results,
                 )
                 filament_usage = (actual_run_grams, billing_planned_grams) if run_status != "completed" else None
-                in_memory_cost_center_id = _print_cost_center_ids.pop(archive_id, None)
+                in_memory_cost_center_id = _print_cost_center_ids.pop((printer_id, archive_id), None)
                 charged = await apply_print_charge_for_archive(
                     db,
                     archive_id,
@@ -11221,11 +11299,11 @@ def _evict_stale_expected_prints() -> None:
     if not stale_keys:
         return
 
-    evicted_archive_ids: set[int] = set()
+    evicted: set[tuple[int, int]] = set()
     for key in stale_keys:
         archive_id = _expected_prints.pop(key, None)
         if archive_id is not None:
-            evicted_archive_ids.add(archive_id)
+            evicted.add((key[0], archive_id))
         _expected_print_creators.pop(key, None)
         _expected_print_registered_at.pop(key, None)
 
@@ -11236,12 +11314,9 @@ def _evict_stale_expected_prints() -> None:
     # outlives it; without the printing check its eviction two hours after
     # dispatch threw away a running print's plate, AMS mapping and cost
     # centre, which completion still reads (#3275).
-    live_archive_ids = set(_expected_prints.values()) | set(_active_prints.values())
-    for archive_id in evicted_archive_ids:
-        if archive_id not in live_archive_ids:
-            _print_ams_mappings.pop(archive_id, None)
-            _print_cost_center_ids.pop(archive_id, None)
-            _print_plate_ids.pop(archive_id, None)
+    for printer_id, archive_id in evicted:
+        if not _expected_or_printing_here(printer_id, archive_id):
+            _drop_dispatch(printer_id, archive_id)
 
     logging.getLogger(__name__).info(
         "Evicted %d stale expected-print entries (TTL=%ds)", len(stale_keys), _EXPECTED_PRINT_TTL_SECONDS

backend/app/services/print_log.py

@@ -262,6 +262,20 @@ async def _open_run_on(db: AsyncSession, archive_id: int, printer_id: int) -> Pr
     )
 
 
+async def open_run_on_printer(
+    db: AsyncSession, printer_id: int, *, subtask_id: str | None = None
+) -> PrintLogEntry | None:
+    """The run open on the printer: the print it is running now, which
+    Bambuddy saw start. One at most (``open_run``). With ``subtask_id``, only
+    a run of that job."""
+    query = select(PrintLogEntry).where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN)
+    if subtask_id is not None:
+        query = query.where(PrintLogEntry.subtask_id == subtask_id)
+    return await db.scalar(
+        query.order_by(PrintLogEntry.id.desc()).limit(1).execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+
+
 async def discard_open_runs(db: AsyncSession, printer_id: int) -> None:
     """Drop the runs left open on the printer by prints that ended without
     their end reaching a run: cancelled at the next start as stale, or whose

backend/tests/integration/test_print_complete_end_to_end.py

@@ -391,13 +391,13 @@ async def test_the_layer_timelapse_is_attached_for_the_run_that_ended(test_engin
 async def test_the_registered_mapping_is_drained_when_the_printer_reported_one(test_engine, db_session, queued_print):
     """PAM/I3: left behind, the archive's next run (sent without a mapping)
     would charge its usage through this run's trays."""
-    main_module._print_ams_mappings[queued_print["archive"]] = [4]
+    main_module._print_ams_mappings[(queued_print["printer"], queued_print["archive"])] = [4]
     try:
         await _complete(test_engine, queued_print["printer"], "completed", extra={"ams_mapping": [0]})
 
-        assert queued_print["archive"] not in main_module._print_ams_mappings
+        assert (queued_print["printer"], queued_print["archive"]) not in main_module._print_ams_mappings
     finally:
-        main_module._print_ams_mappings.pop(queued_print["archive"], None)
+        main_module._print_ams_mappings.pop((queued_print["printer"], queued_print["archive"]), None)
 
 
 async def test_a_hung_sd_cleanup_does_not_hold_up_the_print_s_end(test_engine, db_session, queued_print, monkeypatch):

backend/tests/integration/test_run_lifecycle.py

@@ -37,7 +37,7 @@ pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
 FILE = f"{SUBTASK}.gcode.3mf"
 
 
-async def _complete(test_engine, printer_id: int, status: str, name: str):
+async def _complete(test_engine, printer_id: int, status: str, name: str, extra: dict | None = None):
     """``on_print_complete`` with its background work run one task at a time.
 
     The SQLite test database is one shared connection, so tasks writing at
@@ -61,7 +61,7 @@ async def _complete(test_engine, printer_id: int, status: str, name: str):
         return spawn_background_task(coro, name=name, **kwargs)
 
     with patch("backend.app.main.spawn_background_task", side_effect=spawn):
-        return await _complete_concurrently(test_engine, printer_id, status, name=name)
+        return await _complete_concurrently(test_engine, printer_id, status, name=name, extra=extra)
 
 
 @pytest.fixture(autouse=True)
@@ -465,36 +465,36 @@ async def test_a_downloaded_print_opens_its_run(test_engine, db_session, printer
     assert [r.status for r in await _all_runs(db_session, archives[0].id)] == ["printing"]
 
 
-async def test_two_plates_on_two_printers_at_once_leave_no_open_run(test_engine, db_session, farm, printer_factory):
-    """The second start moves the archive's start to its own; the first
-    printer's end still fills its own run, with its own start.
-
-    The second printer's end finds the archive no longer printing and logs
-    nothing, as before runs were opened at the start: the archive holds one
-    print's state. Its open run goes with that end. Moving the print's state
-    onto the run (P1b) is what gives it its own record."""
+async def test_two_plates_on_two_printers_at_once_each_end_their_own_run(
+    test_engine, db_session, farm, printer_factory
+):
+    """The second start moves the archive's start, printer and job id to its
+    own. Each printer's end still finds its print, fills its own run with its
+    own start and plate, and leaves nothing open."""
     other = (await printer_factory(name="Q")).id
-    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1)
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1, cost_center_id=farm.center)
     await _drive_print_start(test_engine, SimpleNamespace(id=farm.printer), download_ok=False)
     [first] = await _all_runs(db_session, farm.archive)
     first_start = first.started_at
     await asyncio.sleep(1.1)  # the second start is a later second
-    register_expected_print(other, FILE, farm.archive, plate_id=2)
+    register_expected_print(other, FILE, farm.archive, plate_id=2, cost_center_id=farm.center)
     await _drive_print_start(test_engine, SimpleNamespace(id=other), download_ok=False)
 
     await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+    assert main_module._print_cost_center_ids == {(other, farm.archive): farm.center}
+    # The other printer's print is still bound, for its own end.
+    assert main_module._active_prints.get((other, FILE)) == farm.archive
+    assert main_module._print_plate_ids == {(other, farm.archive): 2}, "the first end drains only its own"
     await _complete(test_engine, other, "completed", name=SUBTASK)
+    assert main_module._print_plate_ids == {}
 
     runs = await _all_runs(db_session, farm.archive)
-    assert [(r.id, r.printer_id, r.status, r.started_at) for r in runs] == [
-        (first.id, farm.printer, "completed", first_start)
+    assert [(r.printer_id, r.status, r.plate_id) for r in runs] == [
+        (farm.printer, "completed", 1),
+        (other, "completed", 2),
     ]
-    open_left = (
-        await db_session.scalars(
-            select(PrintLogEntry).where(PrintLogEntry.status == "printing").execution_options(include_open_runs=True)
-        )
-    ).all()
-    assert open_left == []
+    assert runs[0].id == first.id and runs[0].started_at == first_start
+    assert runs[1].started_at > first_start
 
 
 async def test_a_run_is_dated_by_its_end(test_engine, db_session, farm):
@@ -604,3 +604,149 @@ async def test_a_reprint_after_an_unseen_end_opens_a_run_of_its_own(test_engine,
     archive = await db_session.get(PrintArchive, farm.archive)
     assert run.started_at != datetime(2026, 10, 1, 8)
     assert (run.started_at, run.plate_id) == (archive.started_at, 2)
+
+
+async def _two_printers_printing(test_engine, db_session, farm, printer_factory) -> int:
+    """Plate 1 on the farm's printer, then plate 2 of the archive on another,
+    both running. Returns the other printer's id."""
+    other = (await printer_factory(name="Q")).id
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1)
+    await _drive_print_start(test_engine, SimpleNamespace(id=farm.printer), download_ok=False)
+    await asyncio.sleep(1.1)
+    register_expected_print(other, FILE, farm.archive, plate_id=2)
+    await _drive_print_start(test_engine, SimpleNamespace(id=other), download_ok=False)
+    return other
+
+
+async def test_after_a_restart_each_printers_end_finds_its_run(test_engine, db_session, farm, printer_factory):
+    """A restart forgets which print runs where; the run open on the printer
+    still says, though the archive now names the other printer."""
+    from sqlalchemy import update
+
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    await db_session.execute(
+        update(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == farm.printer)
+        .values(subtask_id="job-p")
+        .execution_options(include_open_runs=True)
+    )
+    await db_session.commit()
+    main_module._active_prints.clear()  # the restart
+    archive = await db_session.get(PrintArchive, farm.archive)
+    assert archive.printer_id == other
+
+    await _complete(test_engine, farm.printer, "failed", name=SUBTASK, extra={"subtask_id": "job-p"})
+
+    runs = {r.printer_id: r.status for r in await _all_runs(db_session, farm.archive)}
+    assert runs == {farm.printer: "failed", other: "printing"}
+
+
+async def test_the_running_print_on_each_printer_is_its_own(test_engine, db_session, farm, printer_factory):
+    """The skip-objects, plate label and pause lookups ask which archive runs
+    on a printer, by its job id; the archive row has the latest start's."""
+    from unittest.mock import MagicMock
+
+    from sqlalchemy import update
+
+    from backend.app.main import running_print_archive
+
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    await db_session.execute(
+        update(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == farm.printer)
+        .values(subtask_id="job-p")
+        .execution_options(include_open_runs=True)
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.subtask_id = "job-q"
+    await db_session.commit()
+    main_module._active_prints.clear()
+
+    found = await running_print_archive(
+        db_session, farm.printer, MagicMock(subtask_id="job-p"), printing_only=True, by_name=False
+    )
+
+    assert found is not None and found.id == farm.archive
+    assert archive.printer_id == other
+
+
+async def test_the_kill_switch_knows_its_own_print_on_each_printer(test_engine, db_session, farm, printer_factory):
+    """Bambuddy's print on the first printer is authorized by its own run,
+    though the archive's job id and printer are the second start's."""
+    from unittest.mock import MagicMock
+
+    from sqlalchemy import update
+
+    from backend.app.main import _is_bambuddy_authorized_print
+
+    await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    await db_session.execute(
+        update(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == farm.printer)
+        .values(subtask_id="job-p")
+        .execution_options(include_open_runs=True)
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.subtask_id = "job-q"
+    await db_session.commit()
+    main_module._active_prints.clear()
+    main_module._owned_print_archive_ids.clear()
+
+    state = MagicMock(subtask_id="job-p", subtask_name=SUBTASK, gcode_file="")
+    assert await _is_bambuddy_authorized_print(farm.printer, state, db_session) is True
+
+
+async def test_a_restart_resumes_each_printers_print_on_its_archive(test_engine, db_session, farm, printer_factory):
+    """Bambuddy comes back while both plates print: the first printer's print
+    is resumed on the archive by its run, not archived anew, though the
+    archive names the other printer and its job."""
+    from sqlalchemy import update
+
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    await db_session.execute(
+        update(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == farm.printer)
+        .values(subtask_id="job-p")
+        .execution_options(include_open_runs=True)
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.subtask_id = "job-q"
+    await db_session.commit()
+    main_module._active_prints.clear()  # the restart
+
+    await _drive_print_start(
+        test_engine,
+        SimpleNamespace(id=farm.printer),
+        download_ok=False,
+        state_fields={"progress": 40.0},
+        raw_data={"subtask_id": "job-p"},
+    )
+
+    assert await db_session.scalar(select(func.count()).select_from(PrintArchive)) == 1, "resumed, not archived anew"
+    assert main_module._active_prints.get((farm.printer, FILE)) == farm.archive
+    runs = {r.printer_id: r.status for r in await _all_runs(db_session, farm.archive)}
+    assert runs == {farm.printer: "printing", other: "printing"}
+
+
+@pytest.mark.parametrize("jobs", [("job-z", "job-y"), ("0", "0")], ids=["other-job", "no-job-ids"])
+async def test_a_late_end_of_an_earlier_job_leaves_the_running_print_alone(test_engine, db_session, farm, jobs):
+    """The restart's reconcile closes a print that ended unseen, while the
+    printer runs another one now: that one's open run is not the end's. A
+    print started outside the cloud reports job "0", which tells nothing."""
+    running_job, ending_job = jobs
+    from sqlalchemy import update
+
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-z")
+    await db_session.execute(
+        update(PrintLogEntry).values(subtask_id=running_job).execution_options(include_open_runs=True)
+    )
+    await db_session.commit()
+    main_module._active_prints.clear()  # the restart
+
+    await _complete(
+        test_engine, farm.printer, "aborted", name="Other_part", extra={"subtask_id": ending_job, "_reconciled": True}
+    )
+
+    assert [(r.status, r.subtask_id) for r in await _all_runs(db_session, farm.archive)] == [("printing", running_job)]
+    archive = await db_session.get(PrintArchive, farm.archive)
+    assert archive.status == "printing"

backend/tests/unit/test_expected_print_rollback.py

@@ -92,8 +92,8 @@ def test_archive_keyed_tables_survive_while_another_file_still_points_at_them(
 ):
     """Mirrors the TTL sweep's rule, which is the easy thing to get wrong.
 
-    ``_print_ams_mappings`` and ``_print_plate_ids`` are keyed by archive, not
-    by file. Two files can be registered against one archive, so dropping them
+    ``_print_ams_mappings`` and ``_print_plate_ids`` are keyed by printer and
+    archive, not by file. Two files can be registered against one archive, so dropping them
     on the first unregister would strip usage-tracking data from a print that is
     still expected.
     """
@@ -104,8 +104,8 @@ def test_archive_keyed_tables_survive_while_another_file_still_points_at_them(
 
     main.unregister_expected_print(1, "plate1.3mf", 298)
 
-    assert main._print_ams_mappings.get(298) == [3]
-    assert 298 in main._print_plate_ids
+    assert main._print_ams_mappings.get((1, 298)) == [3]
+    assert (1, 298) in main._print_plate_ids
 
 
 def test_unregistering_an_unknown_print_is_a_no_op(expected_print_tables):
@@ -150,7 +150,7 @@ def test_scheduler_rollback_is_a_no_op_after_a_confirmed_send(expected_print_tab
     sched._rollback_unconfirmed_expected_print(597)
 
     assert main._expected_prints[(1, "widget.3mf")] == 298
-    assert main._print_ams_mappings[298] == [3, 6]
+    assert main._print_ams_mappings[(1, 298)] == [3, 6]
 
 
 def test_scheduler_rollback_never_raises(expected_print_tables, monkeypatch):
@@ -248,7 +248,7 @@ async def test_dispatch_keeps_the_expectation_when_the_print_was_sent(expected_p
         await sched._dispatch_one(597)
 
     assert main._expected_prints[(1, "widget.3mf")] == 298
-    assert main._print_ams_mappings[298] == [3, 6]
+    assert main._print_ams_mappings[(1, 298)] == [3, 6]
 
 
 def test_rollback_entries_are_per_item(expected_print_tables):
@@ -269,7 +269,7 @@ def test_rollback_entries_are_per_item(expected_print_tables):
 
 
 # ---------------------------------------------------------------------------
-# The archive-keyed tables follow the newest dispatch of the archive
+# The dispatch tables follow the newest dispatch of the archive to the printer
 # ---------------------------------------------------------------------------
 
 
@@ -281,9 +281,9 @@ def test_a_fresh_dispatch_without_a_plate_drops_the_previous_plate(expected_prin
     main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3], plate_id=2, cost_center_id=5)
     main.register_expected_print(1, "widget.3mf", 298)
 
-    assert 298 not in main._print_plate_ids
-    assert 298 not in main._print_ams_mappings
-    assert 298 not in main._print_cost_center_ids
+    assert (1, 298) not in main._print_plate_ids
+    assert (1, 298) not in main._print_ams_mappings
+    assert (1, 298) not in main._print_cost_center_ids
 
 
 def test_but_not_while_a_plate_of_the_archive_is_printing(expected_print_tables):
@@ -292,10 +292,10 @@ def test_but_not_while_a_plate_of_the_archive_is_printing(expected_print_tables)
 
     main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3], plate_id=1)
     main._active_prints[(1, "widget.3mf")] = 298
-    main.register_expected_print(2, "widget.3mf", 298)
+    main.register_expected_print(1, "widget.3mf", 298)
 
-    assert main._print_plate_ids.get(298) == 1
-    assert main._print_ams_mappings.get(298) == [3]
+    assert main._print_plate_ids.get((1, 298)) == 1
+    assert main._print_ams_mappings.get((1, 298)) == [3]
 
 
 def test_a_failed_send_leaves_a_running_plate_s_entries(expected_print_tables):
@@ -311,6 +311,22 @@ def test_a_failed_send_leaves_a_running_plate_s_entries(expected_print_tables):
 
     main.unregister_expected_print(2, "plate2.3mf", 298)
 
-    assert main._print_ams_mappings.get(298) == [3]
-    assert main._print_plate_ids.get(298) == 1
-    assert main._print_cost_center_ids.get(298) == 5
+    assert main._print_ams_mappings.get((1, 298)) == [3]
+    assert main._print_plate_ids.get((1, 298)) == 1
+    assert main._print_cost_center_ids.get((1, 298)) == 5
+    assert (2, 298) not in main._print_plate_ids
+
+
+def test_plates_of_one_archive_on_two_printers_keep_their_own(expected_print_tables):
+    """Each printer's print of the archive is charged with its own trays,
+    plate and cost centre: the second dispatch does not take the first's."""
+    main = expected_print_tables
+
+    main.register_expected_print(1, "widget.3mf", 298, ams_mapping=[3], plate_id=1, cost_center_id=5)
+    main._active_prints[(1, "widget.3mf")] = 298
+    main.register_expected_print(2, "widget.3mf", 298, ams_mapping=[7], plate_id=2, cost_center_id=6)
+
+    assert main._print_ams_mappings == {(1, 298): [3], (2, 298): [7]}
+    assert main._print_plate_ids == {(1, 298): 1, (2, 298): 2}
+    assert main._print_cost_center_ids == {(1, 298): 5, (2, 298): 6}
+    assert main._get_start_plate_id(1, 298) == 1 and main._get_start_plate_id(2, 298) == 2

backend/tests/unit/test_fallback_archive_recovery_2957.py

@@ -601,9 +601,11 @@ class TestARecoveredArchiveIsPricedAsAtStart:
             patch.object(main_module, "async_session", maker),
             patch.object(main_module, "_schedule_archive_cost_estimate", estimate),
             patch.dict(main_module._active_prints, {(printer_id, DISPATCH_FILENAME): archive_id}, clear=True),
-            patch.dict(main_module._print_ams_mappings, {archive_id: [2]}, clear=True),
+            patch.dict(main_module._print_ams_mappings, {(printer_id, archive_id): [2]}, clear=True),
             patch.dict(
-                main_module._print_plate_ids, {archive_id: 1} if start_plates is None else start_plates, clear=True
+                main_module._print_plate_ids,
+                {(printer_id, archive_id): 1} if start_plates is None else start_plates,
+                clear=True,
             ),
         ):
             assert await main_module.try_recover_fallback_archive(printer_id, DISPATCH_FILENAME, source) is True

backend/tests/unit/test_finish_photo_moment_sync.py

@@ -666,7 +666,7 @@ class TestPlateRestoreWiring:
 
         monkeypatch.setattr("backend.app.api.routes.settings.get_setting", _yielding_get_setting)
         main_module._active_prints[(patched_env.id, "job.3mf")] = 5
-        main_module._print_plate_ids[5] = 3
+        main_module._print_plate_ids[(patched_env.id, 5)] = 3
 
         async def _on_print_complete_pops():
             main_module._active_prints.clear()
@@ -984,7 +984,7 @@ class TestBoundPrint:
     def test_finds_the_archive_under_the_keys_on_print_complete_tries(self):
         main_module._active_prints[(7, "Cube v2.gcode.3mf")] = 169
         main_module._active_prints[(7, "Cube_v2.3mf")] = 169
-        main_module._print_plate_ids[169] = 12
+        main_module._print_plate_ids[(7, 169)] = 12
 
         bound = main_module._bound_print(7, {"filename": "Cube_v2.3mf", "subtask_name": "Cube_v2"})
 
@@ -993,12 +993,12 @@ class TestBoundPrint:
     def test_reads_without_consuming(self):
         """on_print_complete still has to find the binding after we looked."""
         main_module._active_prints[(7, "job.3mf")] = 5
-        main_module._print_plate_ids[5] = 2
+        main_module._print_plate_ids[(7, 5)] = 2
 
         main_module._bound_print(7, {"filename": "job.3mf", "subtask_name": "job"})
 
         assert main_module._active_prints == {(7, "job.3mf"): 5}
-        assert main_module._print_plate_ids == {5: 2}
+        assert main_module._print_plate_ids == {(7, 5): 2}
 
     def test_unbound_print_and_other_printers(self):
         main_module._active_prints[(8, "job.3mf")] = 5

backend/tests/unit/test_forget_printer.py

@@ -22,9 +22,9 @@ GONE, KEPT = 5, 6
 # Module-level maps in main.py that are not keyed by a printer, or are kept on
 # purpose. Anything else main.py adds must be forgotten with the printer.
 NOT_FORGOTTEN = {
-    "_print_ams_mappings": "keyed by archive (dropped with the printer's expected/active prints)",
-    "_print_cost_center_ids": "keyed by archive (dropped with the printer's expected/active prints)",
-    "_print_plate_ids": "keyed by archive (dropped with the printer's expected/active prints)",
+    "_print_ams_mappings": "keyed by (printer, archive); checked below",
+    "_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",
@@ -120,9 +120,11 @@ async def test_forget_printer_clears_main_and_keeps_other_printers():
     for name in gone:
         _put(name, *gone[name])
         _put(name, *kept[name])
-    main._print_ams_mappings[100 + GONE] = [0]
-    main._print_cost_center_ids[200 + GONE] = 1
-    main._print_plate_ids[200 + GONE] = 2
+    main._print_ams_mappings[(GONE, 100)] = [0]
+    main._print_cost_center_ids[(GONE, 200)] = 1
+    main._print_plate_ids[(GONE, 200)] = 2
+    # Another printer's print of the same archive keeps what it registered.
+    main._print_plate_ids[(KEPT, 200)] = 3
     pending = {name: gone[name][1] for name in ("_printer_offline_notify_tasks", "_fallback_3mf_retry_tasks")}
     try:
         await main.forget_printer(GONE)
@@ -131,9 +133,9 @@ async def test_forget_printer_clears_main_and_keeps_other_printers():
         assert left == []
         lost = [name for name in kept if not _has(name, kept[name][0])]
         assert lost == []
-        assert 100 + GONE not in main._print_ams_mappings
-        assert 200 + GONE not in main._print_cost_center_ids
-        assert 200 + GONE not in main._print_plate_ids
+        assert (GONE, 100) not in main._print_ams_mappings
+        assert (GONE, 200) not in main._print_cost_center_ids
+        assert main._print_plate_ids == {(KEPT, 200): 3}
         await asyncio.sleep(0)
         assert all(task.cancelled() for task in pending.values())
     finally:

backend/tests/unit/test_print_start_expected_promotion.py

@@ -55,11 +55,11 @@ class TestRegisterExpectedPrint:
 
     def test_stores_ams_mapping(self):
         register_expected_print(1, "test.3mf", archive_id=10, ams_mapping=[2, -1, 3])
-        assert _print_ams_mappings[10] == [2, -1, 3]
+        assert _print_ams_mappings[(1, 10)] == [2, -1, 3]
 
     def test_no_ams_mapping_when_none(self):
         register_expected_print(1, "test.3mf", archive_id=10, ams_mapping=None)
-        assert 10 not in _print_ams_mappings
+        assert (1, 10) not in _print_ams_mappings
 
     def test_stores_creator(self):
         register_expected_print(1, "test.3mf", archive_id=10, created_by_id=5)
@@ -76,20 +76,20 @@ class TestRegisterExpectedPrint:
     def test_stores_plate_id(self):
         """plate_id is registered so usage tracking can scope multi-plate 3MFs (#1697)."""
         register_expected_print(1, "test.3mf", archive_id=10, plate_id=2)
-        assert _print_plate_ids[10] == 2
+        assert _print_plate_ids[(1, 10)] == 2
 
     def test_no_plate_id_when_none(self):
         """Direct-Print of a single-plate file passes plate_id=None; nothing stored."""
         register_expected_print(1, "test.3mf", archive_id=10, plate_id=None)
-        assert 10 not in _print_plate_ids
+        assert (1, 10) not in _print_plate_ids
 
     def test_get_start_plate_id_reads_back(self):
         register_expected_print(1, "test.3mf", archive_id=10, plate_id=3)
-        assert _get_start_plate_id(10) == 3
+        assert _get_start_plate_id(1, 10) == 3
 
     def test_get_start_plate_id_returns_none_for_unregistered(self):
-        assert _get_start_plate_id(10) is None
-        assert _get_start_plate_id(None) is None
+        assert _get_start_plate_id(1, 10) is None
+        assert _get_start_plate_id(1, None) is None
 
 
 class TestExpectedPrintDetection:
@@ -249,7 +249,7 @@ class TestExpectedPrintPromotion:
         )
 
         # ams_mapping should still be available for on_print_complete
-        assert _print_ams_mappings[54] == [1]
+        assert _print_ams_mappings[(1, 54)] == [1]
 
     def test_completion_lookup_finds_promoted_archive(self):
         """Simulate on_print_complete finding the archive in _active_prints."""
@@ -276,7 +276,7 @@ class TestExpectedPrintPromotion:
 
         assert found_id == 54
         # And ams_mapping is retrievable
-        assert _print_ams_mappings.pop(54, None) == [1]
+        assert _print_ams_mappings.pop((1, 54), None) == [1]
 
     def test_no_promotion_for_external_print(self):
         """When no expected print exists, nothing is promoted."""
@@ -318,7 +318,7 @@ class TestAMSMappingInjection:
         register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
 
         # Simulate the injection logic from on_print_start
-        _stored_map = _print_ams_mappings.get(54)
+        _stored_map = _print_ams_mappings.get((1, 54))
         assert _stored_map == [1]
 
         ut_session = _active_sessions.get(1)
@@ -351,7 +351,7 @@ class TestAMSMappingInjection:
 
         register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
 
-        _stored_map = _print_ams_mappings.get(54)
+        _stored_map = _print_ams_mappings.get((1, 54))
         ut_session = _active_sessions.get(1)
 
         # Guard: don't overwrite if session already has a mapping
@@ -392,7 +392,7 @@ class TestPlateIdInjection:
         register_expected_print(1, "Box.3mf", archive_id=54, plate_id=2)
 
         # Mirror the injection branch from main.py.
-        _stored_plate_id = _print_plate_ids.get(54)
+        _stored_plate_id = _print_plate_ids.get((1, 54))
         assert _stored_plate_id == 2
 
         ut_session = _active_sessions.get(1)
@@ -427,7 +427,7 @@ class TestPlateIdInjection:
 
         register_expected_print(1, "Box.3mf", archive_id=54, plate_id=2)
 
-        _stored_plate_id = _print_plate_ids.get(54)
+        _stored_plate_id = _print_plate_ids.get((1, 54))
         ut_session = _active_sessions.get(1)
 
         # Guard: don't overwrite if session already has a plate_id

backend/tests/unit/test_printer_kill_switch.py

@@ -348,7 +348,8 @@ async def test_persisted_print_is_authorized_after_restart(monkeypatch, printer_
         created_by_id=None,
     )
     query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
-    db = SimpleNamespace(execute=AsyncMock(return_value=query_result))
+    # No open run for it: a print started before runs were opened at start.
+    db = SimpleNamespace(execute=AsyncMock(return_value=query_result), scalar=AsyncMock(return_value=None))
 
     class FakeSessionContext:
         async def __aenter__(self):
@@ -442,12 +443,14 @@ async def test_kill_switch_defers_when_restart_identity_is_not_available(monkeyp
 
 
 def _authorization_db(archive, dispatched_queue_item_id=None):
-    """Fake session answering the two lookups `_is_bambuddy_authorized_print` makes."""
+    """Fake session answering the lookups `_is_bambuddy_authorized_print` makes:
+    the job's open run (none here: a print started before runs were opened at
+    start), the archive, then the queue."""
 
     query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
     return SimpleNamespace(
         execute=AsyncMock(return_value=query_result),
-        scalar=AsyncMock(return_value=dispatched_queue_item_id),
+        scalar=AsyncMock(side_effect=[None, dispatched_queue_item_id]),
     )
 
 
@@ -497,8 +500,9 @@ async def test_either_dispatch_marker_authorizes_after_a_restart(monkeypatch, ma
 
     assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is True
     assert main_module._active_prints[(9, "some_job.gcode.3mf")] == 77
-    # The fast path is rehydrated, so the queue is never consulted.
-    db.scalar.assert_not_awaited()
+    # The fast path is rehydrated, so the queue is never consulted: the one
+    # scalar read is the open-run lookup.
+    assert db.scalar.await_count == 1
 
 
 @pytest.mark.asyncio

backend/tests/unit/test_run_estimate_plate.py

@@ -89,7 +89,7 @@ async def test_a_direct_print_is_estimated_for_its_plate(test_engine, db_session
         printer.id, filename="multi.gcode.3mf", plate_id=1, status="printing", with_run=False
     )
     main_module._active_prints[(printer.id, "multi.gcode.3mf")] = archive.id
-    main_module._print_plate_ids[archive.id] = 3
+    main_module._print_plate_ids[(printer.id, archive.id)] = 3
 
     assert await _complete(test_engine, printer.id) == [3, 3]
 
@@ -117,7 +117,7 @@ async def test_a_direct_print_logs_its_plate(test_engine, db_session, printer_fa
         printer.id, filename="multi.gcode.3mf", plate_id=1, status="printing", with_run=False
     )
     main_module._active_prints[(printer.id, "multi.gcode.3mf")] = archive.id
-    main_module._print_plate_ids[archive.id] = 3
+    main_module._print_plate_ids[(printer.id, archive.id)] = 3
 
     logged: list = []
     await _complete(test_engine, printer.id, logged)

backend/tests/unit/test_send_all_timelapse_per_plate_3275.py

@@ -230,13 +230,13 @@ class TestEviction:
         m._evict_stale_expected_prints()
 
         assert (1, "Plates.gcode") not in m._expected_prints
-        assert m._print_plate_ids[42] == 2
-        assert m._print_ams_mappings[42] == [0]
+        assert m._print_plate_ids[(1, 42)] == 2
+        assert m._print_ams_mappings[(1, 42)] == [0]
 
     def test_a_print_that_never_started_still_releases_it(self):
         m = self._register_and_start()
 
         m._evict_stale_expected_prints()
 
-        assert 42 not in m._print_plate_ids
-        assert 42 not in m._print_ams_mappings
+        assert (1, 42) not in m._print_plate_ids
+        assert (1, 42) not in m._print_ams_mappings