CHANGELOG.md

@@ -164,6 +164,17 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **An archive cost set through the API, or by Recalculate costs, reaches the statistics** — Setting an archive's cost (`PATCH /api/v1/archives/{id}`) or **Recalculate costs** changed the archive only, while statistics, project totals and batch sums read the cost from the print log, so they kept the old figure. A completed print that had the archive's cost now gets the new one too; a print priced from its own spools, and a failed or cancelled one, keep theirs. **Recalculate costs** still recalculates a cost that was set by hand, as before. A negative cost, or one over 1,000,000, is refused.
+- **A request with `Infinity` or `NaN` in a field that fails validation answers 422** — Such a request ended in a server error, because the error response repeats the value sent and JSON can't hold it.
+- **Editing an archive while it is being reprinted leaves the previous print alone** — Its outcome, failure reason, verdict and filament weight went onto the previous print's log entry, rewriting that print's result in the print log and failure statistics. The edit now changes the archive only.
+- **Archives made without their 3MF show their colours and AMS-HT filament** — When a print's 3MF couldn't be downloaded at start, the archive took its filament from the printer's AMS report. Colours were stored as the printer reports them (`FF8800FF`), so their swatches on the Archives page stayed blank and the colour filter listed them apart from the same colour of other archives; they are now stored as `#FF8800`, like an archive with its 3MF, and existing archives and their print log entries are repaired on startup. Filament the print took from an AMS-HT was not found, so a print using only an AMS-HT got no filament.
+- **An archive whose 3MF arrives later is priced from its spools** — Such an archive started with the default filament price, having no 3MF to estimate from at print start, and kept it when the 3MF was found during the print. It is now estimated from its spools then, as at print start; a 3MF found only as the print ends leaves the pricing to the print's end, as before.
+- **An archive shows the nozzle temperature of the filament its plate prints** — The temperature was taken from the project's first filament, so a plate printing only PETG showed the PLA temperature of filament 1. It is now that of the plate's first used filament.
+- **A print stopped while its printer was offline counts in the print log, and its real end replaces the stop** — When **Stop** couldn't reach the printer, the queue item and archive were closed as cancelled, but no print log entry was written, so the run was missing from the print log and failure statistics, and the Archives page kept showing it as printing until reloaded. The stop now writes the cancelled run and updates the Archives page. Should the printer report the print's real end on reconnect, that outcome replaces the cancelled entry, and the archive no longer shows **Completed** with the reason "Cancelled by user".
+  - **One entry per run:** a printer reporting the same end twice (after a reconnect) logged the run twice, doubling it in the statistics. The repeat is now left out.
+  - **A reprint's reason:** an archive whose earlier run failed kept that failure reason when a reprint completed. Each run's end now sets the archive's reason, or clears it.
+- **A print Bambuddy didn't start is matched to its own archive** — At its end, such a print was matched to an archive whose name merely contained its name, so finishing "Clip" could complete (and bill) a "Cable Clip" still marked as printing; `_` in a name also matched any character. It now needs the same name, allowing for the printer writing spaces as underscores, cutting long names short, or reporting the file name for a print started on its screen.
+- **Print start picks the print's own 3MF from the printer's folders** — When the 3MF couldn't be downloaded directly, Bambuddy searched the printer's folders and took the first file containing the print's name, so "Cube" could be archived with "Cube_large.3mf". A file named for the print is now taken, in whichever folder, and of several files that only contain the name, none.
 - **The A1, A1 Mini, P1 and A2L built-in calibration is not archived as a print** — These printers report their calibration run as a bare `auto_cali_for_user_param.gcode`, not under the system path other models use, so Bambuddy archived each run as a print (without a 3MF) and announced it. It is now recognised as the printer's own job. The printer kill switch still stops it on these models, since nothing tells it apart from a file given that name: switch the kill switch off to calibrate them.
 - **A notification channel limited to one printer gets that printer's events only** — A job added for **Any <model>** or not yet assigned, and any job waiting (for filament, a plate, a printer), were sent without a printer, so no printer filter applied: every channel limited to a printer received all of them, other printers' included. A job that waits on a printer now names it, and reaches that printer's channels; one with no printer yet goes to channels set to all printers, and a channel limited to one printer hears of it once it is assigned to or started there. A power alert from a plug linked to no printer likewise goes to channels set to all printers. Events about the whole setup (the queue finishing, stock, app messages) still reach every channel.
 - **Queue drying pre-dries in blocking mode** — With **Wait for drying to complete** on, a new drying cycle was refused on any printer with a pending job, scheduled ones included, so queue drying never dried anything ahead of a scheduled print. Only a job that could start now keeps a new cycle off; a printer whose job is scheduled for later is dried ahead of it, and the job still waits if the cycle has not finished by then.

backend/app/api/routes/archives.py

@@ -1939,6 +1939,8 @@ async def update_archive(
     # Read before the writes below: the mirror needs to know whether the run's
     # figure was inherited from this archive or measured on its own (#1820).
     previous_filament_grams = archive.filament_used_grams
+    previous_cost = archive.cost
+    previous_status = archive.status
 
     update_payload = update_data.model_dump(exclude_unset=True)
     # Later runs of the archive are charged from its cost and weight: a caller
@@ -2006,9 +2008,11 @@ async def update_archive(
     # ENTRY's grams, not the archive's, so correcting only the archive would fix
     # the card and leave every aggregate reading the old figure -- or, for a
     # print that archived without its 3MF, no figure at all.
+    # While the archive is being reprinted its newest entry is the previous
+    # run's, whose outcome this edit is not about.
     mirror_fields = {"failure_reason", "status", "filament_used_grams", "user_verdict"}
     to_mirror = {k: v for k, v in update_payload.items() if k in mirror_fields}
-    if to_mirror:
+    if to_mirror and previous_status != "printing":
         from backend.app.models.print_log import PrintLogEntry
 
         latest_entry = await db.scalar(
@@ -2036,6 +2040,13 @@ async def update_archive(
             for field, value in to_mirror.items():
                 setattr(latest_entry, field, value)
 
+    # A cost set by hand reaches the run that had the archive's (statistics,
+    # project totals and batch sums read the run's cost).
+    if "cost" in update_payload and previous_status != "printing":
+        from backend.app.services.print_log import mirror_archive_cost
+
+        await mirror_archive_cost(db, archive, previous_cost)
+
     await db.commit()
 
     # Re-fetch with relationships loaded after commit
@@ -2218,6 +2229,7 @@ async def recalculate_all_costs(
     printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Recalculate costs for all archives based on filament usage and prices."""
+    from backend.app.services.print_log import mirror_archive_cost
 
     # Only the archives of the caller's printers, as the archive list shows them.
     query = select(PrintArchive)
@@ -2297,7 +2309,10 @@ async def recalculate_all_costs(
             else:
                 new_cost = None
         if new_cost is not None and archive.cost != new_cost:
+            previous_cost = archive.cost
             archive.cost = new_cost
+            # Its run had the old figure: statistics read the run's cost.
+            await mirror_archive_cost(db, archive, previous_cost)
             updated += 1
 
     await db.commit()

backend/app/api/routes/print_queue.py

@@ -28,6 +28,7 @@ from backend.app.core.config import settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
 from backend.app.core.printer_scope import PrinterScope, ensure_model_target_allowed
+from backend.app.core.websocket import ws_manager
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_batch import PrintBatch, PrintBatchPlate
@@ -2562,6 +2563,7 @@ async def stop_queue_item(
     # archive 436). Close it out here, mirroring what the MQTT path would have
     # done. Only touch a still-"printing" archive so we never overwrite a real
     # completion that raced in.
+    closed_archive_id: int | None = None
     if not stop_sent and item.archive_id:
         archive = await db.get(PrintArchive, item.archive_id)
         if archive and archive.status == "printing":
@@ -2570,9 +2572,34 @@ async def stop_queue_item(
             # The vocabulary key every user cancel stores (#2974); that the
             # printer was offline is on the queue row's error_message.
             archive.failure_reason = "userCancelled"
+            # The run's log entry, so the stop counts in the print log and
+            # failure statistics. Without word from the printer, how far it got
+            # is unknown: no grams, cost or wear are entered. Should the printer
+            # report the real end after all, that entry is replaced.
+            from backend.app.services.print_log import write_log_entry
+
+            await write_log_entry(
+                db,
+                status="cancelled",
+                archive_id=archive.id,
+                queue_item_id=item.id,
+                print_name=archive.print_name,
+                printer_name=await db.scalar(select(Printer.name).where(Printer.id == printer_id)),
+                printer_id=printer_id,
+                started_at=archive.started_at,
+                completed_at=archive.completed_at,
+                filament_type=archive.filament_type,
+                filament_color=archive.filament_color,
+                failure_reason=archive.failure_reason,
+                thumbnail_path=archive.thumbnail_path,
+                created_by_id=archive.created_by_id,
+            )
+            closed_archive_id = archive.id
 
     await db.commit()
     await refresh_batches_after([item.id])
+    if closed_archive_id is not None:
+        await ws_manager.send_archive_updated({"id": closed_archive_id, "status": "cancelled"})
 
     logger.info("Stopped printing queue item %s (stop command sent: %s)", item_id, stop_sent)
 

backend/app/core/database.py

@@ -1348,6 +1348,55 @@ async def _reclassify_sliced_3mf_library_files(conn) -> None:
         )
 
 
+def _hash_hex_colors(value: str) -> str:
+    """``FF8800FF,#00AE42`` -> ``#FF8800,#00AE42``: bare 6- or 8-digit hex
+    tokens get the ``#`` and lose the alpha byte; everything else is kept."""
+    tokens = []
+    for token in value.split(","):
+        bare = token.strip()
+        if re.fullmatch(r"[0-9A-Fa-f]{6}(?:[0-9A-Fa-f]{2})?", bare):
+            token = f"#{bare[:6].upper()}"
+        tokens.append(token)
+    return ",".join(tokens)
+
+
+async def _repair_bare_hex_filament_colors(conn) -> None:
+    """Give the colours of archives made without their 3MF a ``#``.
+
+    Such archives took their colours from the printer's AMS report and stored
+    them as it sends them (``FF8800FF``), so their swatches stayed blank and the
+    colour filter listed them apart from the same colour elsewhere. New ones are
+    written as ``#FF8800``; this brings the existing archives, and the print log
+    entries copied from them, in line.
+
+    Repeatable rather than flagged, like the failure-reason repair: the query
+    only picks values with a token lacking ``#``, and a repaired value has
+    none, so a second run changes nothing, while a restored older database is
+    still repaired.
+    """
+    from sqlalchemy import text
+
+    async with conn.begin_nested():
+        # nosec B608 -- ``table`` comes from the literal tuple below.
+        for table in ("print_archives", "print_log_entries"):
+            rows = (
+                await conn.execute(
+                    text(
+                        f"SELECT id, filament_color FROM {table} "  # nosec B608
+                        "WHERE filament_color IS NOT NULL AND filament_color != '' "
+                        "AND (filament_color NOT LIKE '#%' OR REPLACE(filament_color, ',#', '') LIKE '%,%')"
+                    )
+                )
+            ).fetchall()
+            for row_id, color in rows:
+                repaired = _hash_hex_colors(color)
+                if repaired != color:
+                    await conn.execute(
+                        text(f"UPDATE {table} SET filament_color = :c WHERE id = :i"),  # nosec B608
+                        {"c": repaired, "i": row_id},
+                    )
+
+
 async def _backfill_archive_bed_temperature(conn) -> None:
     """Fill in ``print_archives.bed_temperature`` for archives written before #2989.
 
@@ -5404,6 +5453,10 @@ async def _run_migrations(conn):
     # already on disk. One-shot; see the function for why it is gated.
     await _backfill_archive_bed_temperature(conn)
 
+    # Repair: archives made without their 3MF stored bare hex colours
+    # (FF8800FF); repeatable, see the function.
+    await _repair_bare_hex_filament_colors(conn)
+
     # Backfill: library rows typed from the filename alone kept a sliced 3MF
     # named `Foo.3mf` filed as a source-only project (#2993). Re-reads the zip
     # already on disk. One-shot, internal rows only; see the function.

backend/app/main.py

@@ -14,7 +14,9 @@ from pathlib import Path, PurePosixPath
 from urllib.parse import urlparse
 
 from fastapi import FastAPI
-from fastapi.responses import FileResponse
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from sqlalchemy import delete, inspect as sa_inspect, or_, select, text, update
 from sqlalchemy.exc import IntegrityError
@@ -1498,7 +1500,10 @@ def _extract_filament_data_from_mqtt(data: dict, ams_mapping: list[int] | None =
     if not isinstance(ams_units, list) or not ams_units:
         return result
 
-    # Map global tray id (unit * 4 + tray) → (type, color).
+    from backend.app.services.print_scheduler import _global_tray_id
+
+    # Map global tray id → (type, color), addressed as the print's ams_mapping
+    # is: unit * 4 + tray, or the bare unit id for an AMS-HT (from 128).
     loaded: dict[int, tuple[str, str]] = {}
     for unit in ams_units:
         if not isinstance(unit, dict):
@@ -1515,10 +1520,13 @@ def _extract_filament_data_from_mqtt(data: dict, ams_mapping: list[int] | None =
             except (TypeError, ValueError):
                 continue
             ttype = (tray.get("tray_type") or "").strip()
-            tcolor = (tray.get("tray_color") or "").strip().upper()
+            # RRGGBBAA from the printer, stored as the 3MF path stores it:
+            # "#RRGGBB" (the Archives page's swatches and colour filter).
+            hex_color = (tray.get("tray_color") or "").strip().upper()[:6]
+            tcolor = f"#{hex_color}" if re.fullmatch(r"[0-9A-F]{6}", hex_color) else ""
             if not ttype:
                 continue  # Empty / unloaded slot.
-            loaded[unit_id * 4 + tray_id] = (ttype, tcolor)
+            loaded[_global_tray_id(unit_id, tray_id)] = (ttype, tcolor)
 
     if not loaded:
         return result
@@ -3813,7 +3821,7 @@ _fallback_3mf_retry_tasks: dict[int, asyncio.Task] = {}
 _fallback_recovery_locks: dict[int, asyncio.Lock] = {}
 
 
-async def _recover_fallback_archive(archive_id: int, source_3mf: Path, printer_id: int) -> bool:
+async def _recover_fallback_archive(archive_id: int, source_3mf: Path, printer_id: int, *, price: bool = True) -> bool:
     """Fill in a no-3MF archive from a 3MF that turned up later.
 
     Returns True when the row was upgraded. Safe to call speculatively: it
@@ -3824,11 +3832,17 @@ async def _recover_fallback_archive(archive_id: int, source_3mf: Path, printer_i
     """
     lock = _fallback_recovery_locks.setdefault(printer_id, asyncio.Lock())
     async with lock:
-        return await _recover_fallback_archive_locked(archive_id, source_3mf, printer_id)
+        return await _recover_fallback_archive_locked(archive_id, source_3mf, printer_id, price=price)
+
 
+async def _recover_fallback_archive_locked(
+    archive_id: int, source_3mf: Path, printer_id: int, *, price: bool = True
+) -> bool:
+    """The body of :func:`_recover_fallback_archive`, under its per-printer lock.
 
-async def _recover_fallback_archive_locked(archive_id: int, source_3mf: Path, printer_id: int) -> bool:
-    """The body of :func:`_recover_fallback_archive`, under its per-printer lock."""
+    ``price`` is False when the print's end is being handled: the archive
+    still reads "printing" then, and its end prices it.
+    """
     import zipfile
 
     from backend.app.models.archive import PrintArchive
@@ -3871,6 +3885,19 @@ async def _recover_fallback_archive_locked(archive_id: int, source_3mf: Path, pr
             source_3mf,
             recovered.file_size,
         )
+        if price and recovered.status == "printing":
+            # Print start priced it from the default rate, having no 3MF to
+            # estimate from; now it has one. An ended print is left alone: its
+            # end prices it. A print Bambuddy didn't dispatch has no start
+            # plate; the one the archive just read from the 3MF stands in.
+            _schedule_archive_cost_estimate(
+                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,
+                session_factory=async_session,
+            )
         # `archive_updated`, not `archive_created` — the row was already on the
         # Archives page as an empty card and is now filled in, not new.
         await ws_manager.send_archive_updated(
@@ -3885,7 +3912,7 @@ async def _recover_fallback_archive_locked(archive_id: int, source_3mf: Path, pr
         return True
 
 
-async def try_recover_fallback_archive(printer_id: int, name: str, path: Path) -> bool:
+async def try_recover_fallback_archive(printer_id: int, name: str, path: Path, *, price: bool = True) -> bool:
     """Offer a freshly-downloaded 3MF to this printer's running fallback archive.
 
     Called from the paths that pull a 3MF for a print that is already under way
@@ -3921,7 +3948,7 @@ async def try_recover_fallback_archive(printer_id: int, name: str, path: Path) -
             return False
 
     try:
-        return await _recover_fallback_archive(archive_id, path, printer_id)
+        return await _recover_fallback_archive(archive_id, path, printer_id, price=price)
     except Exception as e:
         # Recovery is opportunistic. A failure here must never take down the
         # caller, which is usually just trying to render a thumbnail.
@@ -5465,65 +5492,68 @@ async def on_print_start(printer_id: int, data: dict):
             search_term = (subtask_name or filename).lower().replace(".gcode", "").replace(".3mf", "")
             logger.info("Direct FTP download failed, searching directories for '%s'", search_term)
             search_dirs = ["/cache", "/model", "/data", "/data/Metadata", "/"]
+            # Every folder is listed before one file is picked, so a file named
+            # for the print in a later folder wins over one that merely
+            # contains the name in an earlier folder.
+            listed_paths: list[str] = []
             for search_dir in search_dirs:
-                if downloaded_filename:
-                    break
                 try:
                     dir_files = await list_files_async(
                         printer.ip_address, printer.access_code, search_dir, printer_model=printer.model
                     )
-                    threemf_files = [f.get("name") for f in dir_files if f.get("name", "").endswith(".3mf")]
-                    if threemf_files:
-                        logger.info(
-                            f"Found {len(threemf_files)} 3MF files in {search_dir}: {threemf_files[:5]}{'...' if len(threemf_files) > 5 else ''}"
-                        )
-                    for f in dir_files:
-                        if f.get("is_directory"):
-                            continue
-                        fname = f.get("name", "")
-                        # Normalize both for comparison (spaces and underscores are equivalent)
-                        fname_normalized = fname.lower().replace(" ", "_")
-                        search_normalized = search_term.replace(" ", "_")
-                        if fname.endswith(".3mf") and search_normalized in fname_normalized:
-                            logger.info("Found matching file in %s: %s", search_dir, fname)
-                            temp_path = app_settings.archive_dir / "temp" / fname
-                            temp_path.parent.mkdir(parents=True, exist_ok=True)
-                            remote_full_path = posixpath.join(search_dir, fname)
-                            if ftp_retry_enabled:
-                                downloaded = await with_ftp_retry(
-                                    download_file_async,
-                                    printer.ip_address,
-                                    printer.access_code,
-                                    remote_full_path,
-                                    temp_path,
-                                    timeout=ftp_timeout,
-                                    socket_timeout=ftp_timeout,
-                                    printer_model=printer.model,
-                                    max_retries=ftp_retry_count,
-                                    retry_delay=ftp_retry_delay,
-                                    operation_name=f"Download 3MF from {remote_full_path}",
-                                    cooloff_ip=printer.ip_address,
-                                )
-                            else:
-                                downloaded = await download_file_async(
-                                    printer.ip_address,
-                                    printer.access_code,
-                                    remote_full_path,
-                                    temp_path,
-                                    timeout=ftp_timeout,
-                                    socket_timeout=ftp_timeout,
-                                    printer_model=printer.model,
-                                )
-                            if downloaded:
-                                downloaded_filename = fname
-                                logger.info("Found and downloaded from %s: %s", search_dir, fname)
-                                cache_3mf_download(printer_id, fname, temp_path)
-                                break
-                            # The listing named the file, so it is on the card;
-                            # only the transfer failed (#3063).
-                            ftp_transfer_failed = True
                 except Exception as e:
                     logger.debug("Failed to list %s: %s", search_dir, e)
+                    continue
+                threemf_files = [f.get("name") for f in dir_files if f.get("name", "").endswith(".3mf")]
+                if threemf_files:
+                    logger.info(
+                        f"Found {len(threemf_files)} 3MF files in {search_dir}: {threemf_files[:5]}{'...' if len(threemf_files) > 5 else ''}"
+                    )
+                listed_paths += [
+                    posixpath.join(search_dir, f.get("name", "")) for f in dir_files if not f.get("is_directory")
+                ]
+            for remote_full_path in _3mf_files_for_print(listed_paths, search_term):
+                fname = posixpath.basename(remote_full_path)
+                logger.info("Found matching file: %s", remote_full_path)
+                temp_path = app_settings.archive_dir / "temp" / fname
+                temp_path.parent.mkdir(parents=True, exist_ok=True)
+                try:
+                    if ftp_retry_enabled:
+                        downloaded = await with_ftp_retry(
+                            download_file_async,
+                            printer.ip_address,
+                            printer.access_code,
+                            remote_full_path,
+                            temp_path,
+                            timeout=ftp_timeout,
+                            socket_timeout=ftp_timeout,
+                            printer_model=printer.model,
+                            max_retries=ftp_retry_count,
+                            retry_delay=ftp_retry_delay,
+                            operation_name=f"Download 3MF from {remote_full_path}",
+                            cooloff_ip=printer.ip_address,
+                        )
+                    else:
+                        downloaded = await download_file_async(
+                            printer.ip_address,
+                            printer.access_code,
+                            remote_full_path,
+                            temp_path,
+                            timeout=ftp_timeout,
+                            socket_timeout=ftp_timeout,
+                            printer_model=printer.model,
+                        )
+                except Exception as e:
+                    logger.debug("Failed to download %s: %s", remote_full_path, e)
+                    continue
+                if downloaded:
+                    downloaded_filename = fname
+                    logger.info("Found and downloaded: %s", remote_full_path)
+                    cache_3mf_download(printer_id, fname, temp_path)
+                    break
+                # The listing named the file, so it is on the card;
+                # only the transfer failed (#3063).
+                ftp_transfer_failed = True
 
         # Validate the downloaded 3MF actually matches the plate that's running
         # (#1204): subtask_name lags across consecutive plates of the same model,
@@ -7636,6 +7666,28 @@ def _normalise_subtask_name(name: str) -> str:
     return name.strip().replace(" ", "_").strip("_").casefold()
 
 
+def _3mf_files_for_print(paths: list[str], search_term: str) -> list[str]:
+    """The 3MF files of the printer's folder listings that are this print's.
+
+    A file of the print's own name wins, in whichever folder. Only without one
+    is a file that merely contains the name taken, and then only when it is
+    the one such file: "Cube_large.3mf" is not "Cube"'s file, and of several
+    candidates none can be told apart. Paths keep the listings' order.
+    """
+    threemf = [path for path in paths if path.lower().endswith(".3mf")]
+    exact = [path for path in threemf if _subtask_names_match(_subtask_name_from_filename(path), search_term)]
+    if exact:
+        return exact
+    term = _normalise_subtask_name(search_term)
+    partial = [path for path in threemf if term in _normalise_subtask_name(PurePosixPath(path).name)]
+    if len(partial) > 1:
+        logging.getLogger(__name__).info(
+            "Several 3MF files contain %r and none is named for it: %s", search_term, partial
+        )
+        return []
+    return partial
+
+
 def _subtask_names_match(expected: str, observed: str) -> bool:
     """Whether two subtask names describe the same print.
 
@@ -7740,7 +7792,9 @@ async def _recover_fallback_from_cache_before_eviction(printer_id: int, data: di
     for name in names:
         try:
             cached = get_cached_3mf(printer_id, name)
-            if cached and await try_recover_fallback_archive(printer_id, name, cached):
+            # The print has ended: its completion prices the archive, so no
+            # print-start estimate is scheduled to race it.
+            if cached and await try_recover_fallback_archive(printer_id, name, cached, price=False):
                 return
         except Exception as e:
             logger.debug("[RECOVER] Pre-eviction recovery for %s failed: %s", name, e)
@@ -8003,22 +8057,35 @@ async def on_print_complete(printer_id: int, data: dict):
         async with async_session() as db:
             from backend.app.models.archive import PrintArchive
 
-            # Try matching by subtask_name (stored as print_name) first
+            # Try matching by subtask_name (stored as print_name) first. The
+            # same name rule as the queue's completion check: a name inside
+            # another ("Clip" in "Cable Clip") is a different print, and the
+            # printer's underscores for spaces and truncation still match.
             if subtask_name:
-                result = await db.execute(
+                printing = await db.scalars(
                     select(PrintArchive)
                     .where(PrintArchive.printer_id == printer_id)
                     .where(PrintArchive.status == "printing")
-                    .where(
-                        or_(
-                            PrintArchive.print_name.ilike(f"%{subtask_name}%"),
-                            PrintArchive.filename.ilike(f"%{subtask_name}%"),
-                        )
-                    )
                     .order_by(PrintArchive.created_at.desc())
-                    .limit(1)
                 )
-                archive = result.scalar_one_or_none()
+                # A print started from the printer's screen reports its file
+                # name, extensions included, as the subtask (#3009).
+                reported = {subtask_name, _subtask_name_from_filename(subtask_name)}
+                archive = next(
+                    (
+                        candidate
+                        for candidate in printing
+                        if any(
+                            (candidate.print_name and _subtask_names_match(candidate.print_name, name))
+                            or (
+                                candidate.filename
+                                and _subtask_names_match(_subtask_name_from_filename(candidate.filename), name)
+                            )
+                            for name in reported
+                        )
+                    ),
+                    None,
+                )
                 if archive:
                     archive_id = archive.id
                     logger.info("Found archive %s by subtask_name match: %s", archive_id, subtask_name)
@@ -11036,6 +11103,28 @@ app = FastAPI(
 app.add_middleware(RefusedHeaderResponseMask)
 
 
+def _json_safe(value):
+    """The validation errors with every non-finite number written as text."""
+    if isinstance(value, float) and not math.isfinite(value):
+        return str(value)
+    if isinstance(value, dict):
+        return {key: _json_safe(item) for key, item in value.items()}
+    if isinstance(value, list | tuple):
+        return [_json_safe(item) for item in value]
+    return value
+
+
+@app.exception_handler(RequestValidationError)
+async def _validation_error(_request, exc: RequestValidationError) -> JSONResponse:
+    """FastAPI's own 422, except that it survives a non-finite number.
+
+    A body may carry ``Infinity`` or ``NaN`` (Python's JSON reader accepts
+    them), and the default handler echoes the input back in its errors,
+    which JSON cannot hold: the request ended in an error instead of a 422.
+    """
+    return JSONResponse(status_code=422, content={"detail": _json_safe(jsonable_encoder(exc.errors()))})
+
+
 # =============================================================================
 # Authentication Middleware - Secures ALL API routes by default
 # =============================================================================

backend/app/schemas/archive.py

@@ -17,7 +17,8 @@ class ArchiveBase(BaseModel):
     is_favorite: bool | None = None
     tags: str | None = None
     notes: str | None = None
-    cost: float | None = None
+    # Added to every total: a negative (or infinite) cost would wreck them
+    cost: Annotated[float | None, Field(ge=0, le=1_000_000)] = None
     failure_reason: str | None = None
     # Number of items printed. 0 is a legal answer -- a plate that jammed and
     # came off ruined produced nothing, and the project's completed-items count

backend/app/services/archive.py

@@ -277,6 +277,36 @@ class ThreeMFParser:
             )
         return self.metadata
 
+    def _first_used_filament_index(self, plates: list) -> int:
+        """Position, in the project's per-filament settings, of the first
+        filament the printed plate uses: 0 when that can't be told.
+
+        ``project_settings`` lists one value per filament of the whole
+        project, so the printed plate's own filament picks which applies
+        (a plate printing only filament 3 has filament 3's nozzle
+        temperature). Without a plate number, the only plate decides.
+        """
+        if self.plate_number:
+            printed = [
+                plate
+                for plate in plates
+                if any(
+                    meta.get("key") == "index" and meta.get("value") == str(self.plate_number)
+                    for meta in plate.findall("metadata")
+                )
+            ]
+        else:
+            printed = plates if len(plates) == 1 else []
+        used: list[int] = []
+        for plate in printed[:1]:
+            for filament in plate.findall("filament"):
+                try:
+                    if float(filament.get("used_g", "0")) > 0:
+                        used.append(int(filament.get("id", "")))
+                except (TypeError, ValueError):
+                    continue
+        return min(used) - 1 if used and min(used) > 0 else 0
+
     def _parse_slice_info(self, zf: zipfile.ZipFile):
         """Parse slice_info.config for print settings and printable objects."""
         try:
@@ -365,6 +395,8 @@ class ThreeMFParser:
                         if printable_objects:
                             self.metadata["printable_objects"] = printable_objects
 
+                self._plate_filament_index = self._first_used_filament_index(plates)
+
                 if any_time_seen:
                     self.metadata["print_time_seconds"] = summed_time
                 if any_grams_seen:
@@ -588,12 +620,14 @@ class ThreeMFParser:
             if bed_temperature is not None:
                 self.metadata["bed_temperature"] = bed_temperature
 
-            # Nozzle temperature
+            # Nozzle temperature: one per filament, so the printed plate's own
+            # (nozzle diameter above is one per nozzle, not per filament).
             for key in ["nozzle_temperature_initial_layer", "nozzle_temperature"]:
                 if key in data:
                     val = data[key]
                     if isinstance(val, list) and val:
-                        self.metadata["nozzle_temperature"] = int(float(val[0]))
+                        index = getattr(self, "_plate_filament_index", 0)
+                        self.metadata["nozzle_temperature"] = int(float(val[index if index < len(val) else 0]))
                     elif isinstance(val, (int, float, str)):
                         self.metadata["nozzle_temperature"] = int(float(val))
                     break
@@ -1628,8 +1662,10 @@ class ArchiveService:
         archive.status = status
         if completed_at:
             archive.completed_at = completed_at
-        if failure_reason:
-            archive.failure_reason = failure_reason
+        # This run's reason, or none: a reprint that completes, or a print
+        # that finished after all once it was stopped offline, keeps no reason
+        # of an earlier outcome.
+        archive.failure_reason = failure_reason
 
         await self.db.commit()
         return True

backend/app/services/print_log.py

@@ -4,9 +4,9 @@ Log entries are written to a separate table and never touch archives or queue it
 """
 
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.models.print_log import PrintLogEntry
@@ -66,37 +66,125 @@ async def write_log_entry(
     if reconciled:
         duration: int | None = 0
     elif started_at and completed_at:
-        duration = int((completed_at - started_at).total_seconds())
+        duration = int((_utc(completed_at) - _utc(started_at)).total_seconds())
     else:
         duration = None
 
-    entry = PrintLogEntry(
-        archive_id=archive_id,
-        queue_item_id=queue_item_id,
-        print_name=print_name,
-        printer_name=printer_name,
-        printer_id=printer_id,
-        status=status,
-        started_at=started_at,
-        completed_at=completed_at,
-        duration_seconds=duration,
-        filament_type=filament_type,
-        filament_color=filament_color,
-        filament_used_grams=filament_used_grams,
-        cost=cost,
-        energy_kwh=energy_kwh,
-        energy_cost=energy_cost,
-        wear_cost=wear_cost_for_run(duration, wear_cost_per_hour),
-        failure_reason=failure_reason,
-        thumbnail_path=thumbnail_path,
-        created_by_id=created_by_id,
-        created_by_username=created_by_username,
-    )
-    db.add(entry)
+    fields = {
+        "archive_id": archive_id,
+        "queue_item_id": queue_item_id,
+        "print_name": print_name,
+        "printer_name": printer_name,
+        "printer_id": printer_id,
+        "status": status,
+        "started_at": started_at,
+        "completed_at": completed_at,
+        "duration_seconds": duration,
+        "filament_type": filament_type,
+        "filament_color": filament_color,
+        "filament_used_grams": filament_used_grams,
+        "cost": cost,
+        "energy_kwh": energy_kwh,
+        "energy_cost": energy_cost,
+        "wear_cost": wear_cost_for_run(duration, wear_cost_per_hour),
+        "failure_reason": failure_reason,
+        "thumbnail_path": thumbnail_path,
+        "created_by_id": created_by_id,
+        "created_by_username": created_by_username,
+    }
+    entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
+    if entry is None:
+        entry = PrintLogEntry(**fields)
+        db.add(entry)
+    elif entry.status != status:
+        # The run ended differently from how it was first logged: a print
+        # stopped while its printer was offline that finished after all. The
+        # real outcome replaces the stop rather than counting the run twice.
+        for name, value in fields.items():
+            if value is None and name in _KEPT_WHEN_NOT_REPORTED:
+                continue
+            setattr(entry, name, value)
+    # The same outcome reported again (a repeated completion after a reconnect)
+    # changes nothing: the repeat no longer has the run's tracked usage, and the
+    # entry may have been edited or given its energy since.
     await db.flush()
     return entry
 
 
+async def mirror_archive_cost(db: AsyncSession, archive, previous_cost: float | None) -> None:
+    """Give the archive's newest run the archive's new cost, if it had the old one.
+
+    Statistics, project totals and batch sums read a run's cost. A completed
+    run without its own spool price took the archive's (or none), so a new
+    archive cost belongs to it too. A run its spools priced keeps that price,
+    and a failed or cancelled run keeps its own: it printed only part of it.
+    """
+    entry = await db.scalar(
+        select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id).order_by(PrintLogEntry.id.desc()).limit(1)
+    )
+    if entry is not None and entry.status == "completed" and entry.cost in (None, previous_cost):
+        entry.cost = archive.cost
+
+
+async def other_runs_of(db: AsyncSession, archive) -> int:
+    """How many log entries *archive* has from runs other than its current one.
+
+    The archive card keeps its first run's figures; a run whose own entry is
+    already written (its stop, logged while the printer was offline) is still
+    the first run.
+    """
+    # Without a start time or printer on the archive, ``!= None`` reads as
+    # ``IS NOT NULL``: every entry is another run's.
+    query = select(func.count(PrintLogEntry.id)).where(
+        PrintLogEntry.archive_id == archive.id,
+        or_(
+            PrintLogEntry.started_at.is_(None),
+            PrintLogEntry.started_at != archive.started_at,
+            PrintLogEntry.printer_id.is_(None),
+            PrintLogEntry.printer_id != archive.printer_id,
+        ),
+    )
+    return await db.scalar(query) or 0
+
+
+# Who and what the run was, which a repeated report may no longer know: a
+# completion that finds the queue item already closed reports none. Its
+# outcome (status, reason, grams, cost) is replaced as reported.
+_KEPT_WHEN_NOT_REPORTED = frozenset(
+    {"queue_item_id", "print_name", "printer_name", "thumbnail_path", "created_by_id", "created_by_username"}
+)
+
+
+def _utc(moment: datetime) -> datetime:
+    """Naive UTC, as the columns store it; an aware time is converted first."""
+    return moment.astimezone(timezone.utc).replace(tzinfo=None) if moment.tzinfo else moment
+
+
+async def _entry_of_this_run(
+    db: AsyncSession,
+    archive_id: int | None,
+    printer_id: int | None,
+    started_at: datetime | None,
+) -> PrintLogEntry | None:
+    """The entry already written for this run, if any.
+
+    A run is its archive, printer and start time together: plates of one
+    archive printing on two printers share the archive row (and can share its
+    start time), and a reprint starts at a new time. The queue item is not
+    part of it: a repeated completion finds the item already closed and
+    reports none. Without an archive, printer or start time a run can't be
+    told apart, so nothing is matched.
+    """
+    if archive_id is None or printer_id is None or started_at is None:
+        return None
+    query = select(PrintLogEntry).where(
+        PrintLogEntry.archive_id == archive_id,
+        PrintLogEntry.printer_id == printer_id,
+        PrintLogEntry.started_at == started_at,
+    )
+    return (await db.scalars(query.order_by(PrintLogEntry.id.desc()).limit(1))).first()
+
+
 async def record_archive_wear(db: AsyncSession, archive, entry: PrintLogEntry) -> None:
     """Copy a run's wear cost onto its archive when it is the archive's first run.
 

backend/app/services/spoolman_tracking.py

@@ -1894,11 +1894,8 @@ async def _apply_spool_cost_to_archive(db, archive_id: int, print_cost: _PrintCo
             )
         return
 
-    from sqlalchemy import func
-
     from backend.app.api.routes.settings import get_setting
     from backend.app.models.archive import PrintArchive
-    from backend.app.models.print_log import PrintLogEntry
 
     archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
     if archive is None:
@@ -1921,10 +1918,9 @@ async def _apply_spool_cost_to_archive(db, archive_id: int, print_cost: _PrintCo
     if total <= 0:
         return
 
-    existing_runs = (
-        await db.execute(select(func.count(PrintLogEntry.id)).where(PrintLogEntry.archive_id == archive_id))
-    ).scalar()
-    if existing_runs:
+    from backend.app.services.print_log import other_runs_of
+
+    if await other_runs_of(db, archive):
         return
 
     new_cost = round(total, 2)

backend/app/services/usage_tracker.py

@@ -989,10 +989,9 @@ async def on_print_complete(
     # so the overwrite must reconstruct the whole-print cost.
 
     if archive_id and results:
-        from sqlalchemy import func, select
+        from sqlalchemy import select
 
         from backend.app.models.archive import PrintArchive
-        from backend.app.models.print_log import PrintLogEntry
 
         archive_result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
         archive = archive_result.scalar_one_or_none()
@@ -1008,11 +1007,9 @@ async def on_print_complete(
                 # live in PrintLogEntry; the archive card keeps the first run's
                 # cost so a failed reprint doesn't visually clobber a successful
                 # 100 g/$X print with a 10 g/$X/10 partial (#1378).
-                _existing_runs_result = await db.execute(
-                    select(func.count(PrintLogEntry.id)).where(PrintLogEntry.archive_id == archive_id)
-                )
-                _existing_runs = _existing_runs_result.scalar()
-                if not _existing_runs:
+                from backend.app.services.print_log import other_runs_of
+
+                if not await other_runs_of(db, archive):
                     archive.cost = round(total_cost, 2)
                     await db.commit()
 

backend/tests/conftest.py

@@ -446,49 +446,11 @@ async def test_engine():
     """Create a test database engine."""
     engine = create_async_engine(TEST_DATABASE_URL, echo=False)
 
-    # Import all models to register them
-    from backend.app.models import (
-        active_print_session,  # noqa: F401
-        ams_history,
-        ams_label,
-        api_key,
-        archive,
-        auth_ephemeral,
-        color_catalog,
-        external_link,
-        filament,
-        group,
-        kprofile_note,
-        maintenance,
-        notification,
-        notification_template,
-        oidc_provider,
-        print_log,
-        print_queue,
-        printer,
-        project,
-        project_bom,
-        scheduled_drying,
-        settings,
-        slot_preset,
-        smart_plug,
-        smart_plug_energy_snapshot,  # noqa: F401
-        sponsor_toast_state,  # noqa: F401
-        spool,
-        spool_assignment,
-        spool_catalog,
-        spool_filament_preset,
-        spool_k_profile,
-        spool_usage_history,
-        spoolbuddy_device,
-        spoolman_k_profile,
-        spoolman_slot_assignment,
-        user,
-        user_email_pref,
-        user_otp_code,
-        user_totp,
-        virtual_printer,
-    )
+    # Every model, as the app's schema has them: a hand-kept list here fell
+    # behind and left four tables out of the SQLite test database.
+    from backend.app.core.database import load_all_models
+
+    load_all_models()
 
     if TEST_PG_URL is not None:
         # The same engine setup as the app's (UTC session, aware datetimes

backend/tests/integration/test_api_key_queue_acts_as_owner.py

@@ -311,6 +311,62 @@ class TestBatches:
         assert batch.name == "theirs"
 
 
+class TestAnOwnOnlyOwner:
+    """An owner who may change only their own queue items.
+
+    Their key is refused at the permission gate (it needs the ``*_all``
+    permission of its owner), so it changes no other user's item, and an
+    owner's deactivation stops it.
+    """
+
+    async def _items(self, db_session, world) -> tuple[PrintQueueItem, PrintQueueItem]:
+        await _set(db_session, "billing_enabled", "false")
+        mine = await _archive(db_session, world["owner"].id, 1)
+        theirs = await _archive(db_session, world["other"].id, 2)
+        own = PrintQueueItem(
+            printer_id=world["printer"].id, archive_id=mine.id, status="pending", created_by_id=world["owner"].id
+        )
+        other = PrintQueueItem(
+            printer_id=world["printer"].id, archive_id=theirs.id, status="pending", created_by_id=world["other"].id
+        )
+        db_session.add_all([own, other])
+        await db_session.commit()
+        return own, other
+
+    async def test_another_users_item_is_left_alone(self, async_client, db_session, world):
+        own, other = await self._items(db_session, world)
+        key = await _key(db_session, world["owner"].id)
+
+        patched = await async_client.patch(f"/api/v1/queue/{other.id}", json={"manual_start": True}, headers=key)
+        bulk = await async_client.patch(
+            "/api/v1/queue/bulk", json={"item_ids": [own.id, other.id], "manual_start": True}, headers=key
+        )
+        deleted = await async_client.delete(f"/api/v1/queue/{other.id}", headers=key)
+
+        assert patched.status_code in (403, 404)
+        assert bulk.status_code in (200, 403), bulk.text
+        assert deleted.status_code in (403, 404)
+        await db_session.refresh(other)
+        assert other.manual_start is False
+        assert await db_session.get(PrintQueueItem, other.id) is not None
+
+    async def test_a_deactivated_owners_key_changes_nothing(self, async_client, db_session, world):
+        """An admin's key, so the deactivation alone is what refuses it."""
+        own, _other = await self._items(db_session, world)
+        key = await _key(db_session, world["admin"].id)
+        world["admin"].is_active = False
+        await db_session.commit()
+
+        patched = await async_client.patch(f"/api/v1/queue/{own.id}", json={"manual_start": True}, headers=key)
+        deleted = await async_client.delete(f"/api/v1/queue/{own.id}", headers=key)
+
+        assert patched.status_code in (401, 403)
+        assert deleted.status_code in (401, 403)
+        await db_session.refresh(own)
+        assert own.manual_start is False
+        assert await db_session.get(PrintQueueItem, own.id) is not None
+
+
 @pytest.fixture
 async def library_world(db_session, world):
     """The same owner, who may also upload, read stats and pipelines, and a

backend/tests/integration/test_archive_edit_cost_and_reprint.py

@@ -0,0 +1,123 @@
+"""Editing an archive: its cost reaches the run, a reprint's run is not touched.
+
+* A hand-set cost was saved on the archive only. Statistics, project totals
+  and batch sums read the run's log entry, so they kept the old figure (COST-5).
+  Like the grams, it reaches the run that inherited it, never a run that
+  measured its own.
+* While the archive is being reprinted, its newest log entry is the previous
+  run's; an edit then rewrote that run's outcome (FG/I9).
+* A negative cost was accepted and subtracted from every total (COST-6).
+"""
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.print_log import PrintLogEntry
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def _entry(db_session, archive_id):
+    query = select(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id)
+    return (await db_session.scalars(query.execution_options(populate_existing=True))).one()
+
+
+class TestAHandSetCost:
+    async def test_reaches_the_run_that_inherited_it(self, async_client, archive_factory, printer_factory, db_session):
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, cost=1.50)
+
+        response = await async_client.patch(f"/api/v1/archives/{archive.id}", json={"cost": 5.00})
+
+        assert response.status_code == 200, response.text
+        assert (await _entry(db_session, archive.id)).cost == 5.00
+
+    async def test_leaves_a_run_that_measured_its_own(self, async_client, archive_factory, printer_factory, db_session):
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, cost=1.50)
+        entry = await _entry(db_session, archive.id)
+        entry.cost = 1.23  # the tracked spools' price for this run
+        await db_session.commit()
+
+        await async_client.patch(f"/api/v1/archives/{archive.id}", json={"cost": 5.00})
+
+        assert (await _entry(db_session, archive.id)).cost == 1.23
+
+    async def test_a_negative_cost_is_refused(self, async_client, archive_factory, printer_factory, db_session):
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, cost=1.50)
+
+        response = await async_client.patch(f"/api/v1/archives/{archive.id}", json={"cost": -100})
+
+        assert response.status_code == 422
+        assert (await _entry(db_session, archive.id)).cost == 1.50
+
+
+class TestAnEditDuringAReprint:
+    async def test_leaves_the_previous_run_alone(self, async_client, archive_factory, printer_factory, db_session):
+        from backend.app.models.archive import PrintArchive
+
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, status="completed", filament_used_grams=10.0, cost=1.50)
+        # The reprint is running: the archive row is its, the entry the last run's
+        row = await db_session.get(PrintArchive, archive.id)
+        row.status = "printing"
+        await db_session.commit()
+
+        response = await async_client.patch(
+            f"/api/v1/archives/{archive.id}",
+            json={"failure_reason": "spaghetti", "user_verdict": "reject", "filament_used_grams": 12.0, "cost": 9.0},
+        )
+
+        assert response.status_code == 200, response.text
+        entry = await _entry(db_session, archive.id)
+        assert (entry.status, entry.failure_reason, entry.user_verdict, entry.filament_used_grams, entry.cost) == (
+            "completed",
+            None,
+            None,
+            10.0,
+            1.50,
+        )
+
+
+class TestWhichRunsTakeIt:
+    async def test_not_a_failed_run_that_printed_only_part(
+        self, async_client, archive_factory, printer_factory, db_session
+    ):
+        """A failed run without tracked spools has no cost on purpose."""
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, cost=1.50, status="failed")
+        entry = await _entry(db_session, archive.id)
+        entry.cost = None
+        await db_session.commit()
+
+        await async_client.patch(f"/api/v1/archives/{archive.id}", json={"cost": 5.00})
+
+        assert (await _entry(db_session, archive.id)).cost is None
+
+    async def test_after_recalculate_costs_too(self, async_client, archive_factory, printer_factory, db_session):
+        """Recalculate gives the run the new figure as well, so a later hand
+        edit still finds the run carrying the archive's cost."""
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, cost=1.50, filament_used_grams=100.0, filament_type="PLA")
+
+        recalculated = await async_client.post("/api/v1/archives/recalculate-costs")
+        assert recalculated.json()["updated"] == 1
+        recalculated_cost = (await _entry(db_session, archive.id)).cost
+        assert recalculated_cost not in (None, 1.50)
+
+        await async_client.patch(f"/api/v1/archives/{archive.id}", json={"cost": 2.00})
+
+        assert (await _entry(db_session, archive.id)).cost == 2.00
+
+    async def test_an_infinite_cost_is_refused(self, async_client, archive_factory, printer_factory):
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, cost=1.50)
+
+        response = await async_client.patch(
+            f"/api/v1/archives/{archive.id}",
+            content=b'{"cost": Infinity}',
+            headers={"Content-Type": "application/json"},
+        )
+
+        assert response.status_code == 422

backend/tests/integration/test_auth_apikey_rbac.py

@@ -141,6 +141,10 @@ class TestApiKeyDenylistIntegrity:
             Permission.GITHUB_BACKUP,
             Permission.GITHUB_RESTORE,
             Permission.FIRMWARE_UPDATE,
+            # Destructive: drop prints from the stats, or scan the network
+            Permission.LIBRARY_PURGE,
+            Permission.ARCHIVES_PURGE,
+            Permission.DISCOVERY_SCAN,
         }
         missing = expected_denied - _APIKEY_DENIED_PERMISSIONS
         assert not missing, (
@@ -452,13 +456,18 @@ class TestCheckApiKeyPermissionsMatrix:
 
         from backend.app.core.auth import _check_apikey_permissions
         from backend.app.core.permissions import Permission
+        from backend.app.models.api_key import APIKey
 
         perm = Permission[perm_name].value
-        all_flags = _FakeApiKey(can_read_status=True, can_queue=True, can_control_printer=True, can_manage_library=True)
+        # Every scope flag the model has, so no flag can be what refuses it
+        all_flags = _FakeApiKey()
+        for column in APIKey.__table__.columns:
+            if column.name.startswith("can_"):
+                setattr(all_flags, column.name, True)
         with pytest.raises(HTTPException) as exc:
             _check_apikey_permissions(all_flags, [perm])
         assert exc.value.status_code == 403
-        assert "administrative" in exc.value.detail.lower() or "does not have" in exc.value.detail.lower()
+        assert "administrative" in exc.value.detail.lower()
 
     def test_unknown_permission_string_is_admin_denied(self):
         """An unrecognised permission string must fail closed, not silently pass."""

backend/tests/integration/test_bare_hex_filament_color_repair.py

@@ -0,0 +1,73 @@
+"""Archives made without their 3MF get ``#`` colours on startup.
+
+They stored the printer's AMS colours as sent (``FF8800FF``), so their
+swatches stayed blank; new ones are written as ``#FF8800``. The repair brings
+existing archives and the print log entries copied from them in line.
+"""
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.core.database import _repair_bare_hex_filament_colors
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_log import PrintLogEntry
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+CASES = {
+    "FF8800FF": "#FF8800",
+    "ff8800,00AE42FF": "#FF8800,#00AE42",
+    "#00AE42,FF0000FF": "#00AE42,#FF0000",
+    "#00AE42,#FF0000": "#00AE42,#FF0000",
+    "Red": "Red",
+}
+
+
+async def _repair(db_session):
+    await _repair_bare_hex_filament_colors(await db_session.connection())
+    await db_session.commit()
+    db_session.expire_all()
+
+
+async def test_bare_hex_gets_its_hash_on_archives_and_log_entries(db_session, printer_factory):
+    printer = await printer_factory()
+    for stored in CASES:
+        archive = PrintArchive(
+            printer_id=printer.id,
+            filename="a.3mf",
+            file_path="",
+            file_size=0,
+            status="completed",
+            filament_color=stored,
+        )
+        db_session.add(archive)
+        await db_session.flush()
+        db_session.add(PrintLogEntry(archive_id=archive.id, status="completed", filament_color=stored))
+    await db_session.commit()
+
+    await _repair(db_session)
+
+    archives = [a.filament_color for a in (await db_session.scalars(select(PrintArchive).order_by(PrintArchive.id)))]
+    logs = [e.filament_color for e in (await db_session.scalars(select(PrintLogEntry).order_by(PrintLogEntry.id)))]
+    assert archives == list(CASES.values())
+    assert logs == list(CASES.values())
+
+
+async def test_running_twice_changes_nothing(db_session, printer_factory):
+    printer = await printer_factory()
+    db_session.add(
+        PrintArchive(
+            printer_id=printer.id,
+            filename="a.3mf",
+            file_path="",
+            file_size=0,
+            status="completed",
+            filament_color="FF8800FF",
+        )
+    )
+    await db_session.commit()
+
+    await _repair(db_session)
+    await _repair(db_session)
+
+    assert await db_session.scalar(select(PrintArchive.filament_color)) == "#FF8800"

backend/tests/integration/test_ended_session_gates.py

@@ -0,0 +1,179 @@
+"""An ended session's JWT is refused by every kind of route gate.
+
+Each gate resolves the bearer token itself: the permission gate, the
+ownership gate (``*_own`` / ``*_all``), the any-of gate and the admin gate. A
+session ends at logout, at a password change (own or by an admin) and when
+the user is deactivated; before that the same token passes every gate.
+
+The API gateway (``auth_middleware``) refuses such a token before any route
+runs, and most routes stack more than one gate, so a broken gate would not show
+on a real route. Each gate is therefore also tried alone, on a route of its
+own that the gateway lets through.
+
+The session under test began 30 s ago: a token issued in the same second as a
+password change passes on purpose (JWT ``iat`` is whole seconds).
+"""
+
+from datetime import timedelta
+
+import jwt
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import update
+
+from backend.app.core import auth
+from backend.app.models.user import User
+from backend.app.utils.local_time import utcnow_naive
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+PASSWORD = "Gatepass1!"
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _login(client: AsyncClient, username: str, password: str) -> str:
+    response = await client.post("/api/v1/auth/login", json={"username": username, "password": password})
+    assert response.status_code == 200, response.text
+    return response.json()["access_token"]
+
+
+def _session_from_30_s_ago(username: str) -> str:
+    issued = utcnow_naive() - timedelta(seconds=30)
+    return jwt.encode(
+        {"sub": username, "iat": issued, "exp": issued + timedelta(days=1), "jti": f"gates-{username}"},
+        auth.SECRET_KEY,
+        algorithm=auth.ALGORITHM,
+    )
+
+
+@pytest.fixture
+async def users(async_client: AsyncClient, db_session) -> dict:
+    """The first admin, and a second admin whose session the tests end."""
+    await async_client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": "gateadmin", "admin_password": PASSWORD},
+    )
+    admin = await _login(async_client, "gateadmin", PASSWORD)
+    created = await async_client.post(
+        "/api/v1/users/",
+        headers=_auth(admin),
+        json={"username": "gateuser", "password": PASSWORD, "role": "admin"},
+    )
+    assert created.status_code == 201, created.text
+    await db_session.execute(
+        update(User)
+        .where(User.username == "gateuser")
+        .values(password_changed_at=utcnow_naive() - timedelta(minutes=1))
+    )
+    await db_session.commit()
+    return {"admin": admin, "user_id": created.json()["id"], "jwt": _session_from_30_s_ago("gateuser")}
+
+
+async def _gates(client: AsyncClient, token: str, name: str) -> dict[str, int]:
+    headers = _auth(token)
+    return {
+        "permission": (await client.get("/api/v1/printers/", headers=headers)).status_code,
+        "ownership": (await client.get("/api/v1/queue/", headers=headers)).status_code,
+        "any-of": (await client.get("/api/v1/inventory/spools/by-tag?tag_uid=00", headers=headers)).status_code,
+        "admin": (
+            await client.post("/api/v1/groups/", headers=headers, json={"name": name, "permissions": []})
+        ).status_code,
+    }
+
+
+async def _logout(client: AsyncClient, users: dict) -> None:
+    assert (await client.post("/api/v1/auth/logout", headers=_auth(users["jwt"]))).status_code == 200
+
+
+async def _own_password_change(client: AsyncClient, users: dict) -> None:
+    response = await client.post(
+        "/api/v1/users/me/change-password",
+        headers=_auth(users["jwt"]),
+        json={"current_password": PASSWORD, "new_password": "Changed1!pass"},
+    )
+    assert response.status_code == 200, response.text
+
+
+async def _admin_password_change(client: AsyncClient, users: dict) -> None:
+    response = await client.patch(
+        f"/api/v1/users/{users['user_id']}", headers=_auth(users["admin"]), json={"password": "Changed1!pass"}
+    )
+    assert response.status_code == 200, response.text
+
+
+async def _deactivation(client: AsyncClient, users: dict) -> None:
+    response = await client.patch(
+        f"/api/v1/users/{users['user_id']}", headers=_auth(users["admin"]), json={"is_active": False}
+    )
+    assert response.status_code == 200, response.text
+
+
+@pytest.mark.parametrize("end_session", [_logout, _own_password_change, _admin_password_change, _deactivation])
+async def test_every_gate_refuses_an_ended_session(async_client: AsyncClient, users, end_session):
+    before = await _gates(async_client, users["jwt"], "before")
+    assert all(code != 401 for code in before.values()), before
+
+    await end_session(async_client, users)
+    after = await _gates(async_client, users["jwt"], "after")
+
+    assert after == dict.fromkeys(after, 401)
+
+
+def _lone_gates():
+    from fastapi import Depends
+
+    from backend.app.core import auth as a
+    from backend.app.core.permissions import Permission
+
+    return {
+        "gateway": None,
+        "permission": a.RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
+        "any-of": a.RequireAnyPermissionIfAuthEnabled(Permission.INVENTORY_READ, Permission.INVENTORY_UPDATE),
+        "ownership": Depends(a.require_ownership_permission(Permission.QUEUE_READ_ALL, Permission.QUEUE_READ_OWN)),
+        "admin": a.RequireAdminIfAuthEnabled(),
+        "user": Depends(a.get_current_user),
+        "optional user": Depends(a.get_current_user_optional),
+        "auth if enabled": Depends(a.require_auth_if_enabled),
+        "key scope": Depends(
+            a.require_api_key_scope("can_update_energy_cost", "update_energy_cost", Permission.SETTINGS_UPDATE)
+        ),
+    }
+
+
+LONE = "/api/v1/lone-gate-test/"
+
+
+@pytest.fixture
+def lone_gate_routes(monkeypatch):
+    """One route per gate. All but "gateway" are let through by the gateway."""
+    from backend.app import main
+
+    routes = []
+    for name, gate in _lone_gates().items():
+        path = LONE + name.replace(" ", "-")
+        main.app.add_api_route(path, lambda: {"ok": True}, dependencies=[gate] if gate else [])
+        # Ahead of the SPA's catch-all route
+        routes.append(main.app.router.routes.pop())
+        main.app.router.routes.insert(0, routes[-1])
+    monkeypatch.setattr(main, "PUBLIC_API_ROUTES", {*main.PUBLIC_API_ROUTES, *(r.path for r in routes[1:])})
+    yield {name: route.path for name, route in zip(_lone_gates(), routes, strict=True)}
+    for route in routes:
+        main.app.router.routes.remove(route)
+
+
+@pytest.mark.parametrize("end_session", [_logout, _own_password_change, _admin_password_change, _deactivation])
+@pytest.mark.parametrize("gate", list(_lone_gates()))
+async def test_each_gate_alone_refuses_an_ended_session(
+    async_client: AsyncClient, users, lone_gate_routes, end_session, gate
+):
+    path = lone_gate_routes[gate]
+    headers = _auth(users["jwt"])
+    assert (await async_client.get(path, headers=headers)).status_code == 200
+
+    await end_session(async_client, users)
+
+    # get_current_user answers a deactivated account with 403 "disabled"
+    assert (await async_client.get(path, headers=headers)).status_code in (401, 403)

backend/tests/integration/test_print_complete_end_to_end.py

@@ -0,0 +1,309 @@
+"""A queued print's completion, end to end on the database.
+
+The archive row, the print log entry, the queue item and the wallet charge are
+what the rest of Bambuddy reads afterwards (archives page, statistics, project
+totals, the cost centre's budget). Each was only pinned through mocks of the
+function that writes it; here ``on_print_complete`` runs against real rows and
+only the outside world (websocket, MQTT relay, notifications, plugs, the
+printer's SD card) is mocked; the printer manager knows no printer.
+"""
+
+import asyncio
+from datetime import datetime
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app import main as main_module
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import CostCenter, WalletTransaction
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.app.services.bambu_ftp import DeleteResult
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+RUN = "run-end-to-end"
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    """The completion marks the printer as awaiting a plate clear on the real
+    printer manager; the next test on this worker must not inherit that."""
+    from backend.app.services.printer_manager import printer_manager
+
+    main_module._active_prints.clear()
+    yield
+    main_module._active_prints.clear()
+    printer_manager._awaiting_plate_clear.clear()
+
+
+@pytest.fixture(autouse=True)
+async def _no_finish_photo(db_session):
+    """The finish photo would open the printer's camera."""
+    db_session.add(Settings(key="capture_finish_photo", value="false"))
+    await db_session.commit()
+
+
+@pytest.fixture
+async def queued_print(db_session, printer_factory):
+    printer = await printer_factory(name="P")
+    db_session.add(Settings(key="billing_enabled", value="true"))
+    user = User(username="payer", password_hash="x", is_active=True)
+    center = CostCenter(name="Lab", is_active=True, is_private=False)
+    db_session.add_all([user, center])
+    await db_session.flush()
+    archive = PrintArchive(
+        printer_id=printer.id,
+        filename="part.3mf",
+        print_name="part",
+        file_path="archives/test/part.3mf",
+        file_size=1,
+        status="printing",
+        started_at=datetime(2026, 10, 19, 8, 0),
+        cost=5.0,
+        created_by_id=user.id,
+        billing_run_id=RUN,
+    )
+    db_session.add(archive)
+    await db_session.flush()
+    item = PrintQueueItem(
+        printer_id=printer.id,
+        archive_id=archive.id,
+        status="printing",
+        position=1,
+        created_by_id=user.id,
+        cost_center_id=center.id,
+        estimated_cost=5.0,
+        billing_run_id=RUN,
+    )
+    db_session.add(item)
+    await db_session.commit()
+    main_module._active_prints[(printer.id, "part.3mf")] = archive.id
+    return {"printer": printer.id, "user": user.id, "center": center.id, "archive": archive.id, "item": item.id}
+
+
+async def _complete(test_engine, printer_id: int, status: str, name: str = "part") -> None:
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    tasks_before = set(asyncio.all_tasks())
+    with (
+        patch("backend.app.main.async_session", maker),
+        patch("backend.app.core.database.async_session", maker),
+        patch("backend.app.main.ws_manager") as ws,
+        patch("backend.app.main.mqtt_relay") as relay,
+        patch("backend.app.main.notification_service") as notif,
+        patch("backend.app.main.smart_plug_manager") as plug,
+        patch("backend.app.services.spoolman_tracking.async_session", maker),
+        # The SD-card cleanup of the uploaded file
+        patch("backend.app.services.bambu_ftp.delete_file_async", AsyncMock(return_value=DeleteResult.DELETED)),
+        patch("backend.app.services.bambu_ftp.list_files_async", AsyncMock(return_value=[])),
+    ):
+        ws.send_print_complete = AsyncMock()
+        ws.send_archive_updated = AsyncMock()
+        ws.broadcast = AsyncMock()
+        relay.on_print_complete = AsyncMock()
+        notif.on_print_complete = AsyncMock()
+        plug.on_print_complete = AsyncMock()
+        await main_module.on_print_complete(
+            printer_id,
+            {"status": status, "filename": f"{name}.gcode", "subtask_name": name, "timelapse_was_active": False},
+        )
+        spawned = asyncio.all_tasks() - tasks_before
+        if spawned:
+            _done, pending = await asyncio.wait(spawned, timeout=10)
+            # Left running, they would carry on into the next test unpatched
+            assert not pending, f"background work still running: {pending}"
+
+
+async def _rows(db_session, archive_id: int, item_id: int):
+    db_session.expire_all()
+    archive = await db_session.get(PrintArchive, archive_id)
+    item = await db_session.get(PrintQueueItem, item_id)
+    logs = (await db_session.scalars(select(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id))).all()
+    charges = (
+        await db_session.scalars(select(WalletTransaction).where(WalletTransaction.transaction_type == "print_charge"))
+    ).all()
+    return archive, item, logs, charges
+
+
+async def test_a_completed_print(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "completed")
+
+    archive, item, logs, charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert archive.status == "completed"
+    assert item.status == "completed"
+    assert [log.status for log in logs] == ["completed"]
+    assert len(charges) == 1
+    charge = charges[0]
+    assert (charge.user_id, charge.cost_center_id, charge.print_queue_id, charge.print_run_id) == (
+        queued_print["user"],
+        queued_print["center"],
+        queued_print["item"],
+        RUN,
+    )
+    assert abs(charge.amount) == pytest.approx(5.0)
+
+
+async def test_a_failed_print(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "failed")
+
+    archive, item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert archive.status == "failed"
+    assert item.status == "failed"
+    assert [log.status for log in logs] == ["failed"]
+
+
+async def test_a_second_completion_charges_nothing_more(test_engine, db_session, queued_print):
+    """A duplicate completion callback (reconnect replay) is idempotent."""
+    await _complete(test_engine, queued_print["printer"], "completed")
+    main_module._active_prints[(queued_print["printer"], "part.3mf")] = queued_print["archive"]
+    await _complete(test_engine, queued_print["printer"], "completed")
+
+    _archive, _item, _logs, charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert len(charges) == 1
+
+
+async def test_a_duplicate_completion_logs_the_run_once(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "completed")
+    main_module._active_prints[(queued_print["printer"], "part.3mf")] = queued_print["archive"]
+    await _complete(test_engine, queued_print["printer"], "completed")
+
+    _archive, _item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert [(log.status, log.queue_item_id) for log in logs] == [("completed", queued_print["item"])]
+
+
+async def test_a_completion_takes_its_own_reason(test_engine, db_session, queued_print):
+    """A reprint that completes carries no reason of an earlier run."""
+    archive = await db_session.get(PrintArchive, queued_print["archive"])
+    archive.failure_reason = "spaghetti"
+    await db_session.commit()
+
+    await _complete(test_engine, queued_print["printer"], "completed")
+
+    archive, _item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert (archive.status, archive.failure_reason) == ("completed", None)
+    assert [(log.status, log.failure_reason) for log in logs] == [("completed", None)]
+
+
+class TestAStopWhileThePrinterIsOffline:
+    """The stop could not reach the printer, so no completion may ever come (#2603)."""
+
+    async def _stop(self, async_client, queued_print):
+        with (
+            patch("backend.app.services.print_control.stop_print_by_user", return_value=False),
+            patch("backend.app.api.routes.print_queue.ws_manager") as ws,
+        ):
+            ws.send_archive_updated = AsyncMock()
+            response = await async_client.post(f"/api/v1/queue/{queued_print['item']}/stop")
+        assert response.status_code == 200, response.text
+        return ws
+
+    async def test_logs_the_run_as_cancelled_and_tells_the_archives_page(self, async_client, db_session, queued_print):
+        ws = await self._stop(async_client, queued_print)
+
+        archive, item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+        assert (archive.status, archive.failure_reason) == ("cancelled", "userCancelled")
+        assert item.status == "cancelled"
+        assert [(log.status, log.failure_reason, log.queue_item_id) for log in logs] == [
+            ("cancelled", "userCancelled", queued_print["item"])
+        ]
+        assert ws.send_archive_updated.await_args.args[0] == {"id": queued_print["archive"], "status": "cancelled"}
+
+    async def test_a_completion_that_arrives_after_all_replaces_it(
+        self, async_client, test_engine, db_session, queued_print
+    ):
+        """The printer finished the print offline and reports it on reconnect."""
+        await self._stop(async_client, queued_print)
+
+        await _complete(test_engine, queued_print["printer"], "completed")
+
+        archive, item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+        assert (archive.status, archive.failure_reason) == ("completed", None)
+        assert item.status == "cancelled"
+        assert [(log.status, log.failure_reason, log.queue_item_id) for log in logs] == [
+            ("completed", None, queued_print["item"])
+        ]
+
+
+class TestACompletionNoArchiveWasBoundTo:
+    """A print Bambuddy did not start is matched to its archive by name (L-W3)."""
+
+    async def _printing(self, db_session, printer_id: int, name: str) -> int:
+        archive = PrintArchive(
+            printer_id=printer_id,
+            filename=f"{name}.3mf",
+            print_name=name,
+            file_path=f"archives/test/{name}.3mf",
+            file_size=1,
+            status="printing",
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        return archive.id
+
+    @pytest.mark.parametrize(
+        ("printed", "other"),
+        [
+            ("Clip", "Cable Clip"),  # a name inside another
+            ("Clip_A", "ClipXA"),  # "_" is a LIKE wildcard
+        ],
+    )
+    async def test_another_archive_whose_name_contains_it_is_left_alone(
+        self, test_engine, db_session, printer_factory, printed, other
+    ):
+        printer = await printer_factory(name="P")
+        other_id = await self._printing(db_session, printer.id, other)
+
+        await _complete(test_engine, printer.id, "completed", name=printed)
+
+        db_session.expire_all()
+        assert (await db_session.get(PrintArchive, other_id)).status == "printing"
+
+    async def test_its_own_archive_is_completed(self, test_engine, db_session, printer_factory):
+        printer = await printer_factory(name="P")
+        other_id = await self._printing(db_session, printer.id, "Cable Clip")
+        own_id = await self._printing(db_session, printer.id, "Clip")
+
+        await _complete(test_engine, printer.id, "completed", name="Clip")
+
+        db_session.expire_all()
+        assert (await db_session.get(PrintArchive, own_id)).status == "completed"
+        assert (await db_session.get(PrintArchive, other_id)).status == "printing"
+
+    async def test_the_printers_underscores_for_spaces_still_match(self, test_engine, db_session, printer_factory):
+        printer = await printer_factory(name="P")
+        own_id = await self._printing(db_session, printer.id, "Cable Clip")
+
+        await _complete(test_engine, printer.id, "completed", name="Cable_Clip")
+
+        db_session.expire_all()
+        assert (await db_session.get(PrintArchive, own_id)).status == "completed"
+
+    async def test_a_renamed_archive_is_found_by_its_file_name(self, test_engine, db_session, printer_factory):
+        printer = await printer_factory(name="P")
+        own_id = await self._printing(db_session, printer.id, "Clip")
+        own = await db_session.get(PrintArchive, own_id)
+        own.print_name = "Clip for the desk"
+        await db_session.commit()
+
+        await _complete(test_engine, printer.id, "completed", name="Clip")
+
+        db_session.expire_all()
+        assert (await db_session.get(PrintArchive, own_id)).status == "completed"
+
+    async def test_a_print_started_on_the_printer_reports_its_file_name(self, test_engine, db_session, printer_factory):
+        """The subtask then carries the extensions (#3009)."""
+        printer = await printer_factory(name="P")
+        own_id = await self._printing(db_session, printer.id, "Clip")
+        own = await db_session.get(PrintArchive, own_id)
+        own.filename = "Clip.gcode.3mf"
+        await db_session.commit()
+
+        await _complete(test_engine, printer.id, "completed", name="Clip.gcode.3mf")
+
+        db_session.expire_all()
+        assert (await db_session.get(PrintArchive, own_id)).status == "completed"

backend/tests/integration/test_print_log_one_entry_per_run.py

@@ -0,0 +1,135 @@
+"""One print log entry per run, however often the run is reported.
+
+A run is its archive, printer and start time. A reprint starts at a new time
+and a plate printing on a second printer is a run of its own, so both get
+their own entries; the same run reported again replaces its entry.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.services.print_log import write_log_entry
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+START = datetime(2026, 10, 19, 8, 0)
+
+
+@pytest.fixture
+async def rows(db_session, printer_factory) -> dict:
+    """Two printers and an archive, so the foreign keys hold (PostgreSQL)."""
+    from backend.app.models.archive import PrintArchive
+
+    first, second = await printer_factory(name="A"), await printer_factory(name="B")
+    archive = PrintArchive(
+        printer_id=first.id, filename="a.3mf", file_path="a.3mf", file_size=1, status="printing", started_at=START
+    )
+    db_session.add(archive)
+    await db_session.commit()
+    return {"printer": first.id, "other_printer": second.id, "archive": archive}
+
+
+async def _write(db, rows, *, printer_id=None, started_at=START, status="completed", queue_item_id=None, cost=None):
+    await write_log_entry(
+        db,
+        status=status,
+        archive_id=rows["archive"].id,
+        printer_id=printer_id or rows["printer"],
+        started_at=started_at,
+        completed_at=started_at + timedelta(hours=1),
+        queue_item_id=queue_item_id,
+        cost=cost,
+    )
+    await db.commit()
+
+
+async def _entries(db) -> list[tuple]:
+    rows = (await db.scalars(select(PrintLogEntry).order_by(PrintLogEntry.id))).all()
+    return [(r.printer_id, r.started_at, r.status, r.queue_item_id) for r in rows]
+
+
+async def test_the_same_run_reported_twice_is_one_entry(db_session, rows):
+    await _write(db_session, rows, status="cancelled")
+    await _write(db_session, rows, status="completed")
+
+    assert await _entries(db_session) == [(rows["printer"], START, "completed", None)]
+
+
+async def test_a_reprint_is_an_entry_of_its_own(db_session, rows):
+    await _write(db_session, rows)
+    await _write(db_session, rows, started_at=START + timedelta(days=1))
+
+    assert len(await _entries(db_session)) == 2
+
+
+async def test_a_plate_on_another_printer_is_an_entry_of_its_own(db_session, rows):
+    await _write(db_session, rows)
+    await _write(db_session, rows, printer_id=rows["other_printer"])
+
+    assert len(await _entries(db_session)) == 2
+
+
+async def test_an_aware_start_time_is_the_same_run(db_session, rows):
+    """The stop route holds aware times, the columns store naive UTC."""
+    await _write(db_session, rows, status="cancelled")
+    await _write(db_session, rows, started_at=START.replace(tzinfo=timezone.utc))
+
+    assert [status for _p, _s, status, _q in await _entries(db_session)] == ["completed"]
+
+
+async def test_the_same_outcome_reported_again_changes_nothing(db_session, rows):
+    """A repeat lacks the run's tracked usage; the entry may have been edited
+    or given its energy since."""
+    await _write(db_session, rows, cost=4.20)
+    entry = await db_session.scalar(select(PrintLogEntry))
+    entry.energy_kwh = 0.5
+    entry.failure_reason = "edited"
+    await db_session.commit()
+
+    await _write(db_session, rows, cost=5.00)
+
+    db_session.expire_all()
+    entry = await db_session.scalar(select(PrintLogEntry))
+    assert (entry.cost, entry.energy_kwh, entry.failure_reason) == (4.20, 0.5, "edited")
+
+
+class TestOtherRunsOf:
+    """Whether an archive's card still shows its first run (#1378)."""
+
+    async def _entry(self, db, rows, started_at, printer_id=None):
+        db.add(
+            PrintLogEntry(
+                archive_id=rows["archive"].id,
+                printer_id=printer_id or rows["printer"],
+                status="cancelled",
+                started_at=started_at,
+            )
+        )
+        await db.commit()
+
+    async def test_the_current_run_s_own_entry_is_not_another_run(self, db_session, rows):
+        from backend.app.services.print_log import other_runs_of
+
+        await self._entry(db_session, rows, START)
+
+        assert await other_runs_of(db_session, rows["archive"]) == 0
+
+    async def test_an_earlier_run_or_another_printer_is(self, db_session, rows):
+        from backend.app.services.print_log import other_runs_of
+
+        await self._entry(db_session, rows, START - timedelta(days=1))
+        await self._entry(db_session, rows, START, printer_id=rows["other_printer"])
+
+        assert await other_runs_of(db_session, rows["archive"]) == 2
+
+    async def test_without_a_start_time_every_entry_is(self, db_session, rows):
+        from backend.app.services.print_log import other_runs_of
+
+        await self._entry(db_session, rows, START)
+        rows["archive"].started_at = None
+        await db_session.commit()
+
+        assert await other_runs_of(db_session, rows["archive"]) == 1

backend/tests/integration/test_printer_scope_1727.py

@@ -296,6 +296,40 @@ class TestTokens:
         assert wall.status_code == 200, wall.text
         assert [p["id"] for p in wall.json()] == [a.id]
 
+    async def _long_lived(self, async_client, jwt: str, scope: str) -> str:
+        created = await async_client.post(
+            "/api/v1/auth/tokens", headers=_auth(jwt), json={"name": scope, "expires_in_days": 30, "scope": scope}
+        )
+        assert created.status_code in (200, 201), created.text
+        return created.json()["token"]
+
+    async def test_long_lived_overlay_token_keeps_to_its_owners_printers(
+        self, async_client, printer_factory, mock_printer_manager
+    ):
+        a, b = await printer_factory(name="A"), await printer_factory(name="B")
+        admin = await _admin_token(async_client)
+        jwt, _ = await _team_member(async_client, admin, "member", [a.id])
+        token = await self._long_lived(async_client, jwt, "overlay")
+
+        hidden = await async_client.get(f"/api/v1/printers/{b.id}/overlay-status?token={token}")
+        own = await async_client.get(f"/api/v1/printers/{a.id}/overlay-status?token={token}")
+
+        assert hidden.status_code == 404
+        assert own.status_code == 200, own.text
+
+    async def test_long_lived_camera_token_keeps_to_its_owners_printers(self, async_client, printer_factory):
+        from backend.app.core.auth import verify_camera_stream_token
+
+        a, b = await printer_factory(name="A"), await printer_factory(name="B")
+        admin = await _admin_token(async_client)
+        jwt, _ = await _team_member(async_client, admin, "member", [a.id])
+        token = await self._long_lived(async_client, jwt, "camera_stream")
+
+        scope = await verify_camera_stream_token(token)
+        assert scope is not None and scope.printer_ids == frozenset({a.id})
+        snapshot = await async_client.get(f"/api/v1/printers/{b.id}/camera/snapshot?token={token}")
+        assert snapshot.status_code == 404
+
     async def test_websocket_token_carries_its_minters_scope(self, async_client, printer_factory):
         from backend.app.core.auth import principal_printer_scope, verify_websocket_token_principal
         from backend.app.core.database import async_session
@@ -427,6 +461,120 @@ class TestQueueAndHistory:
         assert on_b.id not in ids
 
 
+class TestModelTargetsOfALimitedKey:
+    """A limited key may not reach "any printer of a model" by any route."""
+
+    async def _key(self, async_client, admin: str, printer_id: int) -> dict[str, str]:
+        created = await async_client.post(
+            "/api/v1/api-keys/",
+            headers=_auth(admin),
+            json={"name": "k", "can_queue": True, "can_manage_library": True, "printer_ids": [printer_id]},
+        )
+        assert created.status_code == 200, created.text
+        return {"X-API-Key": created.json()["key"]}
+
+    async def test_editing_a_job_onto_a_model(self, async_client, printer_factory, archive_factory, db_session):
+        from backend.app.models.print_queue import PrintQueueItem
+
+        a = await printer_factory(name="A", model="X1C")
+        await printer_factory(name="B", model="X1C")
+        archive = await archive_factory(a.id)
+        admin = await _admin_token(async_client)
+        key = await self._key(async_client, admin, a.id)
+        item = await async_client.post(
+            "/api/v1/queue/", headers=key, json={"archive_id": archive.id, "printer_id": a.id}
+        )
+        assert item.status_code == 200, item.text
+
+        moved = await async_client.patch(
+            f"/api/v1/queue/{item.json()['id']}", headers=key, json={"printer_id": None, "target_model": "X1C"}
+        )
+
+        assert moved.status_code == 400, moved.text
+        assert "specific printer" in moved.json()["detail"]
+        row = await db_session.get(PrintQueueItem, item.json()["id"])
+        await db_session.refresh(row)
+        assert (row.printer_id, row.target_model) == (a.id, None)
+
+    async def test_dispatching_a_batch_of_model_jobs(self, async_client, printer_factory, archive_factory):
+        a = await printer_factory(name="A", model="X1C")
+        await printer_factory(name="B", model="X1C")
+        archive = await archive_factory(a.id)
+        admin = await _admin_token(async_client)
+        batch = await async_client.post(
+            "/api/v1/queue/batches",
+            headers=_auth(admin),
+            json={"name": "o", "archive_id": archive.id, "plates": [{"plate_id": None, "quantity_target": 3}]},
+        )
+        assert batch.status_code in (200, 201), batch.text
+        batch_id = batch.json()["id"]
+        first = await async_client.post(
+            "/api/v1/queue/",
+            headers=_auth(admin),
+            json={"archive_id": archive.id, "target_model": "X1C", "batch_id": batch_id},
+        )
+        assert first.status_code == 200, first.text
+        key = await self._key(async_client, admin, a.id)
+
+        dispatched = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", headers=key, json={})
+
+        assert dispatched.status_code == 400, dispatched.text
+        assert "specific printer" in dispatched.json()["detail"]
+        listed = await async_client.get("/api/v1/queue/", headers=_auth(admin))
+        assert len(listed.json()) == 1
+
+    async def test_running_a_pipeline_aimed_at_a_model(self, async_client, printer_factory, db_session, request):
+        import uuid
+        from pathlib import Path
+
+        from backend.app.core.config import settings as app_settings
+        from backend.app.models.library import LibraryFile
+
+        a = await printer_factory(name="A", model="X1C")
+        admin = await _admin_token(async_client)
+        pipeline = await async_client.post(
+            "/api/v1/slicer-pipelines/",
+            headers=_auth(admin),
+            json={
+                "name": "p",
+                "printer_preset": {"source": "local", "id": "1"},
+                "process_preset": {"source": "local", "id": "2"},
+                "filament_presets": [{"source": "local", "id": "3"}],
+            },
+        )
+        assert pipeline.status_code == 201, pipeline.text
+        aimed = await async_client.put(
+            f"/api/v1/slicer-pipelines/{pipeline.json()['id']}",
+            headers=_auth(admin),
+            json={"target_kind": "printer_class", "target_model_class": "X1C"},
+        )
+        assert aimed.status_code == 200, aimed.text
+        name = f"scope_pipeline_{uuid.uuid4().hex}.3mf"
+        source_path = Path(app_settings.base_dir) / name
+        source_path.write_bytes(b"")
+        request.addfinalizer(lambda: source_path.unlink(missing_ok=True))
+        source = LibraryFile(
+            filename="cube.3mf",
+            file_path=name,
+            file_type="3mf",
+            file_size=0,
+            file_hash="scope-pipeline",
+            source_type="uploaded",
+        )
+        db_session.add(source)
+        await db_session.commit()
+        key = await self._key(async_client, admin, a.id)
+
+        run = await async_client.post(
+            f"/api/v1/slicer-pipelines/{pipeline.json()['id']}/run",
+            headers=key,
+            json={"source_library_file_id": source.id},
+        )
+
+        assert run.status_code == 400, run.text
+        assert "specific printer" in run.json()["detail"]
+
+
 class TestScheduler:
     async def test_model_matching_stays_within_the_scope(self, db_session, printer_factory):
         from backend.app.core.printer_scope import PrinterScope

backend/tests/integration/test_webhook_api_key_check.py

@@ -130,3 +130,20 @@ async def test_a_wrong_key_and_no_key_are_401s(async_client: AsyncClient, key):
     assert wrong.status_code == 401
     assert wrong.json()["detail"] == "Invalid API key"
     assert missing.status_code == 401
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.parametrize("route", [ROUTE, "/api/v1/printers/"])
+@pytest.mark.parametrize("header", ["X-API-Key", "Authorization"])
+async def test_a_disabled_key_is_a_401(async_client: AsyncClient, key, db_session, route, header):
+    from backend.app.models.settings import Settings
+
+    db_session.add(Settings(key="auth_enabled", value="true"))
+    await db_session.execute(update(APIKey).values(enabled=False).where(APIKey.name == "webhook-key"))
+    await db_session.commit()
+    value = key if header == "X-API-Key" else f"Bearer {key}"
+
+    response = await async_client.get(route, headers={header: value})
+
+    assert response.status_code == 401, response.text

backend/tests/unit/services/test_spoolman_print_cost_2591.py

@@ -144,8 +144,6 @@ class TestReportUsagePricesTheArchive:
         def _next_row(*_args, **_kwargs):
             result = MagicMock()
             result.scalar_one_or_none.return_value = next(rows, archive)
-            # The first-run guard counts PrintLogEntry rows.
-            result.scalar.return_value = existing_runs
             return result
 
         db = AsyncMock()
@@ -182,6 +180,8 @@ class TestReportUsagePricesTheArchive:
                     AsyncMock(return_value=None),
                 ),
                 patch("backend.app.services.printer_manager.printer_manager", pm),
+                # The first-run guard: entries of runs other than this one.
+                patch("backend.app.services.print_log.other_runs_of", AsyncMock(return_value=existing_runs)),
             ):
                 await report_usage(printer_id=1, archive_id=7)
 

backend/tests/unit/test_3mf_files_for_print.py

@@ -0,0 +1,71 @@
+"""Which file of a printer directory listing is the starting print's 3MF (SN-C4).
+
+When no direct download path works, print start lists the printer's
+directories and takes the file named for the print. A file that only contains
+the name is another print's ("Cube_large" is not "Cube").
+"""
+
+import pytest
+
+from backend.app.main import _3mf_files_for_print
+
+
+@pytest.mark.parametrize(
+    ("listed", "term", "picked"),
+    [
+        (["Cube_large.gcode.3mf", "Cube.gcode.3mf"], "cube", ["Cube.gcode.3mf"]),
+        (["Cube_large.gcode.3mf", "Cube.3mf"], "cube", ["Cube.3mf"]),
+        # The printer's underscores for spaces
+        (["Cable Clip.gcode.3mf"], "cable_clip", ["Cable Clip.gcode.3mf"]),
+        # No file of its own name: one that contains it is taken only alone
+        (["Cube_v2.gcode.3mf"], "cube", ["Cube_v2.gcode.3mf"]),
+        (["Cube_large.gcode.3mf", "Cube_small.gcode.3mf"], "cube", []),
+        (["Cube.gcode", "notes.txt"], "cube", []),
+        # Across folders: the print's own file in a later folder wins
+        (["/cache/Cube_large.3mf", "/Cube.3mf"], "cube", ["/Cube.3mf"]),
+        (["/cache/Cube.3mf", "/model/Cube.gcode.3mf"], "cube", ["/cache/Cube.3mf", "/model/Cube.gcode.3mf"]),
+        # A folder named like the print does not make its files the print's
+        (["/cube/other.3mf"], "cube", []),
+    ],
+)
+def test_the_print_s_own_file(listed, term, picked):
+    assert _3mf_files_for_print(listed, term) == picked
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_print_start_downloads_only_the_print_s_own_file(test_engine, printer_factory):
+    """Direct paths fail, so print start searches the printer's directories."""
+    from unittest.mock import AsyncMock, patch
+
+    from backend.tests.integration.test_external_print_confirmation_1898 import SUBTASK, _drive_print_start
+
+    printer = await printer_factory()
+    listing = [{"name": f"{SUBTASK}_large.gcode.3mf"}, {"name": f"{SUBTASK}.gcode.3mf"}]
+    listed = patch("backend.app.services.bambu_ftp.list_files_async", new=AsyncMock(return_value=listing))
+
+    mocks = await _drive_print_start(test_engine, printer, download_ok=False, extra_patches=[listed])
+
+    requested = [c.args[2] for c in mocks["download_file_async"].await_args_list]
+    assert "/cache/" + listing[1]["name"] in requested, "the directory search did not try the print's file"
+    assert not [path for path in requested if "_large" in path], requested
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_the_print_s_own_file_in_a_later_folder_wins(test_engine, printer_factory):
+    from unittest.mock import AsyncMock, patch
+
+    from backend.tests.integration.test_external_print_confirmation_1898 import SUBTASK, _drive_print_start
+
+    printer = await printer_factory()
+
+    async def listing(_ip, _code, folder, **_kwargs):
+        return {"/cache": [{"name": f"{SUBTASK}_large.gcode.3mf"}], "/": [{"name": f"{SUBTASK}.3mf"}]}.get(folder, [])
+
+    listed = patch("backend.app.services.bambu_ftp.list_files_async", new=AsyncMock(side_effect=listing))
+
+    mocks = await _drive_print_start(test_engine, printer, download_ok=False, extra_patches=[listed])
+
+    requested = [c.args[2] for c in mocks["download_file_async"].await_args_list]
+    assert not [path for path in requested if "_large" in path], requested

backend/tests/unit/test_archive_nozzle_temperature_of_the_plate.py

@@ -0,0 +1,54 @@
+"""An archive's nozzle temperature is that of the filament its plate prints (PL-C11).
+
+``project_settings.config`` lists one nozzle temperature per filament of the
+whole project. Reading the first entry gave a plate that prints only
+filament 3 (PETG at 250 °C) the temperature of filament 1 (PLA at 220 °C).
+"""
+
+import json
+import zipfile
+from pathlib import Path
+
+from backend.app.services.archive import ThreeMFParser
+
+
+def _3mf(path: Path, plates: dict[int, list[int]], unused: tuple[int, ...] = ()) -> Path:
+    """A 3MF whose plate N prints the filaments listed for it (1-based ids);
+    ``unused`` ones are listed on every plate with 0 g."""
+    plate_xml = "".join(
+        "<plate>"
+        f"<metadata key='index' value='{index}'/>"
+        + "".join(f"<filament id='{fid}' type='PLA' color='#FFFFFF' used_g='0'/>" for fid in unused)
+        + "".join(f"<filament id='{fid}' type='PLA' color='#FFFFFF' used_g='5'/>" for fid in filaments)
+        + "</plate>"
+        for index, filaments in plates.items()
+    )
+    settings = {"nozzle_temperature": ["220", "230", "250"], "nozzle_diameter": ["0.4"]}
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr("Metadata/slice_info.config", f"<?xml version='1.0'?><config>{plate_xml}</config>")
+        zf.writestr("Metadata/project_settings.config", json.dumps(settings))
+        zf.writestr("3D/3dmodel.model", "<model/>")
+    return path
+
+
+def _nozzle_temperature(path: Path, plate_number: int | None = None) -> int | None:
+    return ThreeMFParser(path, plate_number=plate_number).parse().get("nozzle_temperature")
+
+
+def test_a_plate_printing_only_filament_3_has_its_temperature(tmp_path):
+    assert _nozzle_temperature(_3mf(tmp_path / "a.3mf", {1: [3]})) == 250
+
+
+def test_the_printed_plate_of_several_decides(tmp_path):
+    path = _3mf(tmp_path / "a.3mf", {1: [1], 2: [2]})
+
+    assert _nozzle_temperature(path, plate_number=2) == 230
+    assert _nozzle_temperature(path, plate_number=1) == 220
+
+
+def test_a_plate_of_several_filaments_takes_the_first_it_uses(tmp_path):
+    assert _nozzle_temperature(_3mf(tmp_path / "a.3mf", {1: [3, 2]})) == 230
+
+
+def test_a_filament_listed_but_not_used_does_not_count(tmp_path):
+    assert _nozzle_temperature(_3mf(tmp_path / "a.3mf", {1: [3]}, unused=(1,))) == 250

backend/tests/unit/test_cost_tracking.py

@@ -443,6 +443,65 @@ class TestCostCalculation:
         # Cost = 20g / 1000 * 20.0 = 0.40
         assert results[0]["cost"] == 0.40
 
+    async def _failed_after_firmware_reset(self, *, last_progress=0.0, last_layer_num=0, layer_usage=None):
+        """A failed print whose printer already reset progress and layer to 0."""
+        spool = _make_spool(spool_id=1, label_weight=1000, cost_per_kg=20.0)
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="Test",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 80},
+            tray_now_at_start=0,
+        )
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]}, progress=0, layer_num=0, tray_now=0
+        )
+        db = _mock_db_sequential([_make_archive(archive_id=10), None, _make_assignment(spool_id=1), spool])
+        cumulative = MagicMock(return_value={0: 1000.0})  # filament 0 = slot 1
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch("backend.app.api.routes.settings.get_setting", return_value="15.0"),
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=[{"slot_id": 1, "used_g": 40.0, "type": "PLA", "color": "#FF0000"}],
+            ),
+            patch("backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf", return_value=layer_usage),
+            patch("backend.app.utils.threemf_tools.get_cumulative_usage_at_layer", cumulative),
+            patch("backend.app.utils.threemf_tools.extract_filament_properties_from_3mf", return_value={}),
+            patch("backend.app.utils.threemf_tools.mm_to_grams", return_value=7.0),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+            results = await on_print_complete(
+                printer_id=1,
+                data={"status": "failed", "last_progress": last_progress, "last_layer_num": last_layer_num},
+                printer_manager=printer_manager,
+                db=db,
+                archive_id=10,
+            )
+        return results, cumulative
+
+    @pytest.mark.asyncio
+    async def test_a_progress_reset_to_zero_falls_back_to_the_last_progress(self):
+        """Firmware zeroes progress on cancel; 0 % would charge nothing."""
+        results, _ = await self._failed_after_firmware_reset(last_progress=40.0)
+
+        # 40 % of 40 g
+        assert results[0]["weight_used"] == 16.0
+
+    @pytest.mark.asyncio
+    async def test_a_layer_reset_to_zero_falls_back_to_the_last_layer(self):
+        results, cumulative = await self._failed_after_firmware_reset(
+            last_progress=40.0, last_layer_num=12, layer_usage={1: {12: 1000.0}}
+        )
+
+        assert cumulative.call_args.args[1] == 12
+        # The per-layer grams, not the 40 % estimate
+        assert results[0]["weight_used"] == 7.0
+
     @pytest.mark.asyncio
     async def test_cost_with_ams_fallback_tracking(self):
         """AMS fallback tracking also calculates cost correctly."""
@@ -730,6 +789,8 @@ class TestCostAggregation:
         filament_usage = [{"slot_id": 1, "used_g": 20.0, "type": "PLA", "color": "#FF0000"}]
 
         with (
+            # The first run of this archive
+            patch("backend.app.services.print_log.other_runs_of", AsyncMock(return_value=0)),
             patch("backend.app.core.config.settings") as mock_settings,
             patch("backend.app.api.routes.settings.get_setting", return_value="15.0"),
             patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
@@ -813,6 +874,8 @@ class TestCostAggregation:
         filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": "#FF0000"}]
 
         with (
+            # The first run of this archive
+            patch("backend.app.services.print_log.other_runs_of", AsyncMock(return_value=0)),
             patch("backend.app.core.config.settings") as mock_settings,
             patch("backend.app.api.routes.settings.get_setting", return_value="10.0"),
             patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
@@ -896,6 +959,8 @@ class TestCostAggregation:
         filament_usage = [{"slot_id": 1, "used_g": 20.0, "type": "PLA", "color": "#FF0000"}]
 
         with (
+            # The first run of this archive
+            patch("backend.app.services.print_log.other_runs_of", AsyncMock(return_value=0)),
             patch("backend.app.core.config.settings") as mock_settings,
             patch("backend.app.api.routes.settings.get_setting", return_value="15.0"),
             patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),

backend/tests/unit/test_fallback_archive_mqtt_filament.py

@@ -65,7 +65,7 @@ class TestExtractFilamentDataFromMqtt:
         }
         result = _extract_filament_data_from_mqtt(data)
         # Order is by ascending global tray id, colors uppercased.
-        assert result == {"filament_type": "PLA,PETG,ABS", "filament_color": "FF0000,00FF00,0000FF"}
+        assert result == {"filament_type": "PLA,PETG,ABS", "filament_color": "#FF0000,#00FF00,#0000FF"}
 
     def test_ams_mapping_narrows_to_used_slots(self):
         """The slicer's slot-per-print-filament mapping wins — only used
@@ -88,7 +88,7 @@ class TestExtractFilamentDataFromMqtt:
         }
         # Print uses slots 3 then 0 then 1 (slot 2 untouched, no entry).
         result = _extract_filament_data_from_mqtt(data, ams_mapping=[3, 0, 1])
-        assert result == {"filament_type": "TPU,PLA,PETG", "filament_color": "FFFF00,FF0000,00FF00"}
+        assert result == {"filament_type": "TPU,PLA,PETG", "filament_color": "#FFFF00,#FF0000,#00FF00"}
 
     def test_ams_mapping_with_vt_tray_sentinels_filtered_out(self):
         """ams_mapping entries equal to -1 represent the VT tray (external
@@ -108,7 +108,7 @@ class TestExtractFilamentDataFromMqtt:
             }
         }
         result = _extract_filament_data_from_mqtt(data, ams_mapping=[-1, 0, 1])
-        assert result == {"filament_type": "PLA,PETG", "filament_color": "FF0000,00FF00"}
+        assert result == {"filament_type": "PLA,PETG", "filament_color": "#FF0000,#00FF00"}
 
     def test_dual_ams_global_ids_use_unit4_offset(self):
         """A dual-AMS rig has unit 0 → trays 0-3, unit 1 → trays 4-7.
@@ -122,7 +122,7 @@ class TestExtractFilamentDataFromMqtt:
             }
         }
         result = _extract_filament_data_from_mqtt(data, ams_mapping=[4, 0])
-        assert result == {"filament_type": "PETG-CF,PLA", "filament_color": "112233,FF0000"}
+        assert result == {"filament_type": "PETG-CF,PLA", "filament_color": "#112233,#FF0000"}
 
     def test_mapping_pointing_at_unknown_slot_falls_through_to_known_only(self):
         data = {
@@ -134,7 +134,7 @@ class TestExtractFilamentDataFromMqtt:
         }
         # Slot 7 isn't in our AMS — entry skipped, only slot 0 remains.
         result = _extract_filament_data_from_mqtt(data, ams_mapping=[7, 0])
-        assert result == {"filament_type": "PLA", "filament_color": "FF0000"}
+        assert result == {"filament_type": "PLA", "filament_color": "#FF0000"}
 
     def test_mapping_entirely_unknown_returns_empty(self):
         """If every mapped slot is unknown the helper returns {} rather
@@ -244,7 +244,7 @@ class TestOnPrintStartCallbackShape:
         # The 6-char rgba `FFFFFF` is what the AMS reports (with `FF` alpha
         # tail trimmed by the catalog) — the helper preserves whatever the
         # firmware sends.
-        assert result == {"filament_type": "PETG", "filament_color": "FFFFFFFF"}
+        assert result == {"filament_type": "PETG", "filament_color": "#FFFFFF"}
 
     def test_wrapper_with_no_ams_mapping_falls_back_to_all_loaded(self):
         """Wrapper shape without an ams_mapping behaves the same as the
@@ -258,7 +258,7 @@ class TestOnPrintStartCallbackShape:
         }
         wrapper = {"raw_data": inner}
         result = _extract_filament_data_from_mqtt(wrapper)
-        assert result == {"filament_type": "PLA,PETG", "filament_color": "FF0000,00FF00"}
+        assert result == {"filament_type": "PLA,PETG", "filament_color": "#FF0000,#00FF00"}
 
     def test_inner_shape_still_supported_after_wrapper_lookup(self):
         """Existing callers that pass the inner shape directly (e.g. the
@@ -270,7 +270,7 @@ class TestOnPrintStartCallbackShape:
         }
         assert _extract_filament_data_from_mqtt(inner) == {
             "filament_type": "ASA",
-            "filament_color": "112233",
+            "filament_color": "#112233",
         }
 
     def test_wrapper_with_missing_raw_data_returns_empty(self):
@@ -289,5 +289,32 @@ class TestOnPrintStartCallbackShape:
         }
         assert _extract_filament_data_from_mqtt(wrapper) == {
             "filament_type": "PLA",
-            "filament_color": "FF0000",
+            "filament_color": "#FF0000",
         }
+
+
+class TestTheFallbackMatchesThe3mfFormat:
+    """Colours as the 3MF path stores them, and AMS-HT trays by their id."""
+
+    def test_a_colour_is_hash_and_six_hex_digits(self):
+        data = {"ams": {"ams": [_ams_unit(0, [_tray(0, "PLA", "ff8800ff"), _tray(1, "PETG", "nonsense")])]}}
+
+        result = _extract_filament_data_from_mqtt(data)
+
+        # The alpha byte is dropped; something that is no colour is left out
+        assert result == {"filament_type": "PLA,PETG", "filament_color": "#FF8800"}
+
+    def test_an_ams_ht_tray_is_addressed_by_its_unit_id(self):
+        """AMS-HT units have ids from 128 and one tray, mapped as the bare id."""
+        data = {
+            "ams": {
+                "ams": [
+                    _ams_unit(0, [_tray(0, "PLA", "FF0000FF")]),
+                    _ams_unit(128, [_tray(0, "PA-CF", "222222FF")]),
+                ]
+            }
+        }
+
+        result = _extract_filament_data_from_mqtt(data, ams_mapping=[128, 0])
+
+        assert result == {"filament_type": "PA-CF,PLA", "filament_color": "#222222,#FF0000"}

backend/tests/unit/test_fallback_archive_recovery_2957.py

@@ -574,3 +574,82 @@ class TestPhotosSurviveRecovery:
         (stale / "snap.jpg").write_bytes(b"old")
 
         assert find_archive_photo(archive, "snap.jpg") == current / "snap.jpg"
+
+
+class TestARecoveredArchiveIsPricedAsAtStart:
+    """The precise cost estimate print start runs needs the 3MF, which a
+    fallback archive did not have then (COST-8)."""
+
+    async def _recover(self, test_engine, tmp_path, status: str, *, start_plates=None, gcode_path=None):
+        from unittest.mock import MagicMock
+
+        from backend.app import main as main_module
+
+        maker, printer_id, archive_id = await _seed(test_engine, tmp_path)
+        async with maker() as db:
+            archive = await db.get(PrintArchive, archive_id)
+            archive.status = status
+            if gcode_path:
+                archive.extra_data = {**archive.extra_data, "_print_data": {"filename": gcode_path}}
+            await db.commit()
+        source = _write_3mf(tmp_path / "temp" / DISPATCH_FILENAME)
+        with zipfile.ZipFile(source, "a") as zf:
+            zf.writestr("Metadata/plate_1.gcode", "; sliced plate 1")
+        estimate = MagicMock()
+
+        with (
+            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_plate_ids, {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
+        return estimate, printer_id, archive_id
+
+    async def test_while_it_prints(self, test_engine, tmp_path):
+        estimate, printer_id, archive_id = await self._recover(test_engine, tmp_path, "printing")
+
+        estimate.assert_called_once()
+        assert estimate.call_args.args[:2] == (printer_id, archive_id)
+        assert estimate.call_args.kwargs["ams_mapping"] == [2]
+        assert estimate.call_args.kwargs["plate_id"] == 1
+
+    async def test_not_once_the_print_has_ended(self, test_engine, tmp_path):
+        """Its end has priced it already; an estimate would undo that."""
+        estimate, _printer_id, _archive_id = await self._recover(test_engine, tmp_path, "failed")
+
+        estimate.assert_not_called()
+
+    async def test_a_print_bambuddy_did_not_dispatch_takes_the_plate_its_archive_read(self, test_engine, tmp_path):
+        """No start plate is known; the gcode path named plate 1."""
+        estimate, _printer_id, _archive_id = await self._recover(
+            test_engine, tmp_path, "printing", start_plates={}, gcode_path="/data/Metadata/plate_1.gcode"
+        )
+
+        assert estimate.call_args.kwargs["plate_id"] == 1
+
+    async def test_not_from_the_print_s_own_completion(self, test_engine, tmp_path):
+        """The archive still reads "printing" while its completion runs; the
+        completion prices it, and an estimate would race that."""
+        from unittest.mock import MagicMock
+
+        from backend.app import main as main_module
+
+        maker, printer_id, archive_id = await _seed(test_engine, tmp_path)
+        source = _write_3mf(tmp_path / "temp" / DISPATCH_FILENAME)
+        estimate = MagicMock()
+
+        with (
+            patch.object(main_module, "async_session", maker),
+            patch.object(main_module, "_schedule_archive_cost_estimate", estimate),
+            patch.object(main_module, "get_cached_3mf", lambda _printer_id, _name: source),
+            patch.dict(main_module._active_prints, {(printer_id, DISPATCH_FILENAME): archive_id}, clear=True),
+        ):
+            await main_module._recover_fallback_from_cache_before_eviction(printer_id, {"filename": DISPATCH_FILENAME})
+
+        async with maker() as db:
+            assert (await db.get(PrintArchive, archive_id)).file_path, "the archive was not recovered"
+        estimate.assert_not_called()

backend/tests/unit/test_phantom_print_hardening.py

@@ -211,138 +211,5 @@ class TestMultiplePrintingQueueItemsWarning:
         assert "printer 7" in bug_warnings[0].message
 
 
-class TestBusyPrinterSeedingFromPrintingItems:
-    """Regression for the duplicate-dispatch bug observed with quantity>1 batches.
-
-    The old scheduler seeded ``busy_printers`` with an empty set and relied on
-    ``_is_printer_idle()`` to gate dispatch. On H2D / P1 series the MQTT state
-    lags several seconds behind the print command, so the next ``check_queue``
-    tick saw IDLE and dispatched a second queue item onto the same printer —
-    both items ended up in 'printing' status. The fix seeds ``busy_printers``
-    up-front with every printer that already has an item in 'printing' status.
-    """
-
-    @pytest.mark.asyncio
-    async def test_seed_query_returns_printers_with_printing_items(self):
-        """The seeding query must return every printer_id that has a 'printing' item."""
-        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-        import backend.app.models  # noqa: F401
-        from backend.app.core.database import Base
-        from backend.app.models.print_queue import PrintQueueItem
-
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-        async with session_maker() as db:
-            db.add_all(
-                [
-                    PrintQueueItem(printer_id=1, status="printing", position=1, archive_id=10),
-                    PrintQueueItem(printer_id=1, status="pending", position=2, archive_id=10),
-                    PrintQueueItem(printer_id=2, status="printing", position=1, archive_id=11),
-                    PrintQueueItem(printer_id=3, status="pending", position=1, archive_id=12),
-                    PrintQueueItem(printer_id=None, status="pending", position=1, archive_id=13),
-                ]
-            )
-            await db.commit()
-
-            result = await db.execute(
-                select(PrintQueueItem.printer_id)
-                .where(PrintQueueItem.status == "printing")
-                .where(PrintQueueItem.printer_id.is_not(None))
-            )
-            busy_printers = {pid for (pid,) in result.all() if pid is not None}
-
-        assert busy_printers == {1, 2}
-
-        await engine.dispose()
-
-    @pytest.mark.asyncio
-    async def test_seed_query_empty_when_no_printing_items(self):
-        """With only pending items, no printer is considered busy by the query."""
-        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-        import backend.app.models  # noqa: F401
-        from backend.app.core.database import Base
-        from backend.app.models.print_queue import PrintQueueItem
-
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-        async with session_maker() as db:
-            db.add_all(
-                [
-                    PrintQueueItem(printer_id=1, status="pending", position=1, archive_id=10),
-                    PrintQueueItem(printer_id=2, status="completed", position=1, archive_id=11),
-                    PrintQueueItem(printer_id=3, status="failed", position=1, archive_id=12),
-                    PrintQueueItem(printer_id=4, status="cancelled", position=1, archive_id=13),
-                ]
-            )
-            await db.commit()
-
-            result = await db.execute(
-                select(PrintQueueItem.printer_id)
-                .where(PrintQueueItem.status == "printing")
-                .where(PrintQueueItem.printer_id.is_not(None))
-            )
-            busy_printers = {pid for (pid,) in result.all() if pid is not None}
-
-        assert busy_printers == set()
-
-        await engine.dispose()
-
-    @pytest.mark.asyncio
-    async def test_check_queue_skips_printer_with_existing_printing_item(self, caplog):
-        """Simulate the exact observed bug: a pending item targets a printer that already
-        has another queue item in 'printing' status. The scheduler must NOT dispatch the
-        pending item even if the live MQTT state reports IDLE.
-        """
-        from unittest.mock import AsyncMock, patch
-
-        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-        import backend.app.models  # noqa: F401
-        from backend.app.core.database import Base
-        from backend.app.models.print_queue import PrintQueueItem
-        from backend.app.services.print_scheduler import PrintScheduler
-
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-        async with session_maker() as db:
-            db.add_all(
-                [
-                    PrintQueueItem(printer_id=1, status="printing", position=1, archive_id=84),
-                    PrintQueueItem(printer_id=1, status="pending", position=2, archive_id=84),
-                ]
-            )
-            await db.commit()
-
-        scheduler = PrintScheduler()
-        start_print_mock = AsyncMock()
-
-        with (
-            patch("backend.app.services.print_scheduler.async_session", session_maker),
-            patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=False)),
-            patch.object(scheduler, "_is_printer_idle", return_value=True),
-            patch.object(scheduler, "_check_auto_drying", AsyncMock()),
-            patch.object(scheduler, "_start_print", start_print_mock),
-            patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
-        ):
-            mock_pm.is_connected.return_value = True
-            await scheduler.check_queue()
-
-        start_print_mock.assert_not_called()
-
-        async with session_maker() as db:
-            rows = (await db.execute(select(PrintQueueItem).order_by(PrintQueueItem.position))).scalars().all()
-            statuses = [r.status for r in rows]
-        assert statuses == ["printing", "pending"]
-
-        await engine.dispose()
+# The busy-printer seed (an item already printing blocks a second dispatch) is
+# pinned on the real check_queue in test_scheduler_check_queue_holds.py.

backend/tests/unit/test_route_auth_coverage.py

@@ -36,8 +36,6 @@ from backend.app.main import app
 _AUTH_QUALNAME_PATTERNS: tuple[str, ...] = (
     "require_",  # require_permission_if_auth_enabled, require_admin_*, require_auth_*, require_any_*, require_ownership_*, require_camera_stream_token_*, require_energy_cost_update
     "cloud_caller",  # cloud.py route-level dep
-    "_cloud_api_key_gate",  # cloud.py router-level dep
-    "resolve_api_key_cloud_owner",  # used by slicer routes that need the API key's owner
     "get_current_user",  # JWT identity resolution
     "get_current_active_user",  # JWT identity resolution
     "get_api_key",  # webhook routes use this directly
@@ -80,6 +78,14 @@ _PUBLIC_ROUTES: frozenset[tuple[str, str]] = frozenset(
         # Forgot-password (anonymous request) + confirm (signed token in the URL).
         ("POST", "/api/v1/auth/forgot-password"),
         ("POST", "/api/v1/auth/forgot-password/confirm"),
+        # Own password change — resolves the caller with get_current_user_optional
+        # and answers 401 inline when there is none; the current password in the
+        # body is checked too.
+        ("POST", "/api/v1/users/me/change-password"),
+        # Who am I — checks the JWT or API key inline (revocation, freshness,
+        # active user) and answers 401 without one; the auth middleware also
+        # guards it.
+        ("GET", "/api/v1/auth/me"),
         # ---- HTTP API: signed-URL routes (token in path is the auth) ----
         # Signed download URLs — token in path validated by the handler.
         ("GET", "/api/v1/archives/{archive_id}/dl/{token}/{filename}"),
@@ -158,9 +164,33 @@ def _walk_dependant_qualnames(dependant) -> list[str]:
     return names
 
 
+def _is_auth_qualname(qualname: str) -> bool:
+    """An auth-bearing callable. An ``*_optional`` resolver is not one: it
+    lets an anonymous caller through (``get_current_user_optional`` answers
+    None), so a route that relies on it checks inline and is listed."""
+    if "_optional" in qualname:
+        return False
+    return any(p in qualname for p in _AUTH_QUALNAME_PATTERNS)
+
+
 def _has_auth_dep(dependant) -> bool:
-    """True if any callable in the dependant tree matches an auth pattern."""
-    return any(any(p in qn for p in _AUTH_QUALNAME_PATTERNS) for qn in _walk_dependant_qualnames(dependant))
+    """True if any dependency in the tree matches an auth pattern. The route's
+    own handler is not one, whatever it is called."""
+    return any(_is_auth_qualname(qn) for sub in dependant.dependencies for qn in _walk_dependant_qualnames(sub))
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    ("qualname", "is_auth"),
+    [
+        ("require_permission_if_auth_enabled.<locals>.permission_checker", True),
+        ("get_current_user", True),
+        ("get_current_user_optional", False),
+        ("some_helper", False),
+    ],
+)
+def test_the_matcher_does_not_count_an_optional_resolver(qualname, is_auth) -> None:
+    assert _is_auth_qualname(qualname) is is_auth
 
 
 def _ws_endpoint_does_inline_token_check(route: APIWebSocketRoute) -> bool:

backend/tests/unit/test_scheduler_check_queue_holds.py

@@ -0,0 +1,97 @@
+"""What check_queue holds back, run on a real database with the dispatch drained.
+
+* An item whose creator may no longer use its printer (#1727) is held with a
+  reason, not dispatched and not failed.
+* A printer that already has an item printing takes no second one, even while
+  it reports IDLE (the phantom-print seed).
+"""
+
+import pytest
+from sqlalchemy import insert, update
+
+from backend.app.models.group import Group, group_printers
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.tests.unit.test_scheduler_busy_reasons_3018 import _scheduler, one_printer  # noqa: F401
+
+
+async def _restricted_creator(session_maker, allowed_printer_id: int) -> int:
+    async with session_maker() as db:
+        db.add(Settings(key="auth_enabled", value="true"))
+        group = Group(name="team", permissions=["queue:create"], restrict_printers=True)
+        user = User(username="member", password_hash="x", is_active=True, groups=[group])
+        db.add_all([group, user])
+        await db.flush()
+        await db.execute(insert(group_printers).values(group_id=group.id, printer_id=allowed_printer_id))
+        await db.commit()
+        return user.id
+
+
+async def _other_printer(session_maker) -> int:
+    async with session_maker() as db:
+        printer = Printer(name="Other", serial_number="OTHER", ip_address="10.0.0.2", access_code="code", model="X1C")
+        db.add(printer)
+        await db.commit()
+        return printer.id
+
+
+class TestACreatorWhoLostThePrinter:
+    @pytest.mark.asyncio
+    async def test_the_item_is_held_with_a_reason(self, one_printer):  # noqa: F811
+        allowed = await _other_printer(one_printer.session_maker)
+        creator = await _restricted_creator(one_printer.session_maker, allowed)
+        item_id = await one_printer.add_item()
+        async with one_printer.session_maker() as db:
+            await db.execute(update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(created_by_id=creator))
+            await db.commit()
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+        async with one_printer.session_maker() as db:
+            item = await db.get(PrintQueueItem, item_id)
+        assert item.status == "pending"
+        assert "no longer has access" in (item.waiting_reason or "")
+
+    @pytest.mark.asyncio
+    async def test_a_creator_who_still_has_it_is_dispatched(self, one_printer):  # noqa: F811
+        creator = await _restricted_creator(one_printer.session_maker, one_printer.printer_id)
+        item_id = await one_printer.add_item()
+        async with one_printer.session_maker() as db:
+            await db.execute(update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(created_by_id=creator))
+            await db.commit()
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+
+        async with one_printer.session_maker() as db:
+            assert (await db.get(PrintQueueItem, item_id)).status != "pending"
+
+
+class TestAPrinterWithAnItemPrinting:
+    @pytest.mark.asyncio
+    async def test_takes_no_second_item_while_it_reports_idle(self, one_printer):  # noqa: F811
+        printing_id = await one_printer.add_item(status="printing", position=0)
+        pending_id = await one_printer.add_item(position=1)
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+        async with one_printer.session_maker() as db:
+            assert (await db.get(PrintQueueItem, printing_id)).status == "printing"
+            assert (await db.get(PrintQueueItem, pending_id)).status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_without_it_the_same_item_goes_out(self, one_printer):  # noqa: F811
+        """The control: nothing else holds this item back."""
+        pending_id = await one_printer.add_item(position=1)
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+
+        async with one_printer.session_maker() as db:
+            assert (await db.get(PrintQueueItem, pending_id)).status != "pending"

backend/tests/unit/test_scheduler_dispatch_failures.py

@@ -0,0 +1,107 @@
+"""How a dispatch that cannot go ahead ends.
+
+Before the upload: an exhausted cost-centre budget, a printer row that is gone
+and a disconnected printer each fail the item with their own message, upload
+nothing and power the printer off if it was powered on for the job. After the
+upload: a start command the printer refused while it stayed idle fails the
+item, deletes the uploaded file again and sends the failure notification. A
+refusal because the printer turned busy defers instead (#2598, see
+test_scheduler_busy_defer_2598).
+"""
+
+from contextlib import ExitStack
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.services.print_scheduler import PrintScheduler
+from backend.tests.unit.test_scheduler_busy_defer_2598 import _base_patches, _final_item, dispatch_case  # noqa: F401
+
+IDLE = SimpleNamespace(state="IDLE", subtask_id=None, gcode_file=None)
+
+
+async def _dispatch(ctx, *, start_print=None, extra=(), before=None):
+    scheduler = PrintScheduler()
+    upload = AsyncMock(return_value=True)
+    start = start_print or MagicMock(return_value=True)
+    failed = AsyncMock()
+    async with ctx.session_maker() as db:
+        item = await db.get(PrintQueueItem, ctx.ids.item_id)
+        if before is not None:
+            await before(db, item)
+        with ExitStack() as stack:
+            for p in _base_patches(scheduler, ctx, upload, start, MagicMock(return_value=IDLE)):
+                stack.enter_context(p)
+            for p in extra:
+                stack.enter_context(p)
+            stack.enter_context(
+                patch(
+                    "backend.app.services.notification_service.notification_service.on_queue_job_failed",
+                    failed,
+                )
+            )
+            delete = stack.enter_context(
+                patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True))
+            )
+            power_off = scheduler._power_off_if_needed
+            await scheduler._start_print(db, item)
+    return SimpleNamespace(
+        power_off=power_off, upload=upload, start=start, delete=delete, failed=failed, item=await _final_item(ctx)
+    )
+
+
+class TestBeforeTheUpload:
+    @pytest.mark.asyncio
+    async def test_an_exhausted_budget(self, dispatch_case):  # noqa: F811
+        refused = AsyncMock(side_effect=HTTPException(400, "Cost center budget exceeded"))
+
+        result = await _dispatch(
+            dispatch_case, extra=[patch("backend.app.services.print_scheduler.validate_print_budget", refused)]
+        )
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Cost center budget exceeded")
+        result.upload.assert_not_awaited()
+        result.start.assert_not_called()
+        result.power_off.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_a_printer_that_is_gone(self, dispatch_case):  # noqa: F811
+        async def gone(db, item):
+            item.printer_id = 9999
+            await db.commit()
+
+        result = await _dispatch(dispatch_case, before=gone)
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Printer not found")
+        result.upload.assert_not_awaited()
+        result.power_off.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_a_disconnected_printer(self, dispatch_case):  # noqa: F811
+        offline = patch(
+            "backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=False)
+        )
+
+        result = await _dispatch(dispatch_case, extra=[offline])
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Printer not connected")
+        result.upload.assert_not_awaited()
+        result.start.assert_not_called()
+        result.power_off.assert_awaited_once()
+
+
+class TestARefusedStartOnAnIdlePrinter:
+    @pytest.mark.asyncio
+    async def test_fails_the_item_and_removes_the_upload(self, dispatch_case):  # noqa: F811
+        result = await _dispatch(dispatch_case, start_print=MagicMock(return_value=False))
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Failed to send print command to printer")
+        result.upload.assert_awaited_once()
+        uploaded_to = result.upload.await_args.args[3]
+        # One clears the way before the upload, one removes the upload again
+        assert [c.args[2] for c in result.delete.await_args_list] == [uploaded_to, uploaded_to]
+        result.failed.assert_awaited_once()
+        assert result.failed.await_args.kwargs["printer_id"] == dispatch_case.ids.printer_id

backend/tests/unit/test_spoolman_partial_usage_layer_reset.py

@@ -0,0 +1,49 @@
+"""Spoolman partial usage of an aborted print whose printer reset its layer to 0.
+
+The firmware zeroes ``layer_num`` on cancel. The layer captured during the
+print then decides how far the print got; without it the abort reports no
+usage at all (the progress estimate needs a total layer count, which an
+aborted print may not have either).
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services import spoolman_tracking
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("last_layer_num", "charged_at"), [(12, 12), (None, None)])
+async def test_the_captured_layer_stands_in_for_the_reset_one(last_layer_num, charged_at):
+    session = MagicMock()
+    session.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+    session.return_value.__aexit__ = AsyncMock(return_value=False)
+    tracking = SimpleNamespace(
+        layer_usage={"12": {"0": 500.0}},
+        filament_properties={},
+        filament_usage=[{"slot_id": 1, "used_g": 10.0}],
+        ams_trays={},
+        slot_to_tray=[0],
+        tray_remain_start={},
+        archive_id=1,
+        tray_now_at_start=0,
+    )
+    cumulative = MagicMock(return_value={})
+    status = SimpleNamespace(layer_num=0, total_layers=None, raw_data={}, tray_change_log=[])
+
+    with (
+        patch.object(spoolman_tracking, "async_session", session),
+        patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+        patch("backend.app.services.printer_manager.printer_manager.get_status", MagicMock(return_value=status)),
+        patch.object(spoolman_tracking, "_get_printer_serial", AsyncMock(return_value="SERIAL")),
+        patch.object(spoolman_tracking, "_get_spoolman_client_with_fallback", AsyncMock(return_value=MagicMock())),
+        patch("backend.app.utils.threemf_tools.get_cumulative_usage_at_layer", cumulative),
+    ):
+        await spoolman_tracking._report_partial_usage(1, tracking, last_layer_num=last_layer_num)
+
+    if charged_at is None:
+        cumulative.assert_not_called()
+    else:
+        assert cumulative.call_args.args[1] == charged_at

backend/tests/unit/test_vp_mqtt_server.py

@@ -5,7 +5,7 @@ import asyncio
 import inspect
 import json
 from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -761,3 +761,85 @@ class TestSlicerCommandsTheLogKeepsWhole:
         )
 
         assert '"tray_id": 26' not in text
+
+
+def _client_writer(peer: str = "1.2.3.4") -> MagicMock:
+    writer = MagicMock()
+    writer.write = MagicMock()
+    writer.drain = AsyncMock()
+    writer.close = MagicMock()
+    writer.wait_closed = AsyncMock()
+    writer.get_extra_info = MagicMock(side_effect=lambda name: (peer, 12345) if name == "peername" else None)
+    return writer
+
+
+class TestNothingBeforeConnect:
+    """PUBLISH and SUBSCRIBE before a successful CONNECT reach nothing (V-U4)."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("packet", "handler"),
+        [
+            (bytes([0x30, 0x05, 0x00, 0x01, ord("t"), ord("{"), ord("}")]), "_handle_publish"),
+            (bytes([0x82, 0x06, 0x00, 0x01, 0x00, 0x01, ord("t"), 0x00]), "_handle_subscribe"),
+        ],
+    )
+    async def test_a_packet_before_connect_is_dropped(self, packet, handler):
+        server = _make_server()
+        server._running = True
+        reader = asyncio.StreamReader()
+        connect = _build_connect_payload(keep_alive=60)
+        # The packet first, then a valid CONNECT and the same packet again:
+        # only the one after the CONNECT may reach the handler.
+        reader.feed_data(packet + bytes([0x10, len(connect)]) + connect + packet + bytes([0xE0, 0x00]))
+        reader.feed_eof()
+        server._send_status_report = AsyncMock()
+        handled = AsyncMock()
+
+        with patch.object(server, handler, handled):
+            await server._handle_client(reader, _client_writer())
+
+        assert handled.await_count == 1
+
+
+class TestARateLimitedAddress:
+    """The rate limit refuses the right access code too, through the real handler (V-W2)."""
+
+    @pytest.mark.asyncio
+    async def test_a_correct_connect_is_refused(self):
+        from backend.app.services.virtual_printer.mqtt_server import _AUTH_RATE_LIMIT_MAX_ATTEMPTS
+
+        server = _make_server()
+        server._running = True
+        for _ in range(_AUTH_RATE_LIMIT_MAX_ATTEMPTS):
+            server._record_auth_failure("1.2.3.4")
+        reader = asyncio.StreamReader()
+        connect = _build_connect_payload(keep_alive=60)
+        reader.feed_data(bytes([0x10, len(connect)]) + connect)
+        reader.feed_eof()
+        writer = _client_writer()
+        server._handle_connect = AsyncMock(return_value=(True, 60))
+
+        await server._handle_client(reader, writer)
+
+        server._handle_connect.assert_not_awaited()
+        writer.write.assert_called_once_with(bytes([0x20, 0x02, 0x00, 0x05]))
+        assert not server._clients
+
+    @pytest.mark.asyncio
+    async def test_another_address_still_connects(self):
+        from backend.app.services.virtual_printer.mqtt_server import _AUTH_RATE_LIMIT_MAX_ATTEMPTS
+
+        server = _make_server()
+        server._running = True
+        for _ in range(_AUTH_RATE_LIMIT_MAX_ATTEMPTS):
+            server._record_auth_failure("1.2.3.4")
+        reader = asyncio.StreamReader()
+        connect = _build_connect_payload(keep_alive=60)
+        reader.feed_data(bytes([0x10, len(connect)]) + connect)
+        reader.feed_eof()
+        server._handle_connect = AsyncMock(return_value=(True, 60))
+
+        await server._handle_client(reader, _client_writer(peer="5.6.7.8"))
+
+        server._handle_connect.assert_awaited_once()