Repair no-3MF archives' photos and their silent filament writes (#1820)

Two faults behind the same kind of print: one that arrives without a
retrievable 3MF, which on an H2S is any job started from the printer's
own internal library.

Such an archive has no file_path, and Path("").parent is Path("."), so
every site that derived the archive's folder from it landed on the data
directory itself. The finish-photo capture spotted that and wrote to
<archive_dir>/<id>/photos instead. Nothing else did. The photo was
written in one place and looked for in another: reads 404'd, deletes
dropped the name and left the file, and the notification attachment
never found the image. Hand-uploaded photos worked only because upload
and read agreed with each other rather than with the capture. Give the
question one owner in utils/archive_paths and have all four sites ask
it. Lookups check the old shared location too, so photos already
uploaded there stay reachable; uploads now go where captures go.

Separately, the remain%-delta fallback that stands in for a missing 3MF
can charge nothing for several reasons, and did so without a word. The
AMS reading is coarse and, on the reporter's printer, noisy: it rises
mid-print, swings five points over a job, sits at 100% through a
36-minute print on a fresh spool, and goes negative on a nearly empty
one -- which the start-of-print gate rejects, dropping the only slot
that was printing. Two of their prints went uncounted for two different
reasons and both read as "no spools updated", which is also what a print
with nothing to charge prints. Name the slot and the two readings in
each case, on the Spoolman path and on the internal-inventory path,
which has carried the same gates since #1119.

The Spoolman path also had no notion of which slots the print used, so a
spool swapped into an idle slot mid-print reads as consumption and is
billed to whoever that slot is assigned to -- the fault #1269 fixed for
the internal tracker, still open here, and likeliest on exactly the
prints this fallback serves, where nothing else narrows the field. Use
the same three pieces of evidence it does: the print's mapping, its
mid-print tray changes, and the tray it started on. The last needs
storing, because the internal tracker's row is deleted before this runs
and a screen-started print has no mapping to fall back on -- hence a new
nullable column, and no backfill, since a row from before it existed has
nothing to say. Where no evidence exists at all, every slot is still
considered.

Both paths also treated tray_now == 255 as naming a slot. It does not:
it is the field's initial value, the fallback for an unparseable
reading, and what it reports with nothing loaded. Mapped as a tray id it
becomes (255, 1), so as the only evidence it excluded every real slot
and charged nothing at all -- this issue's own bug, arriving by a new
route. On the internal path that is live today; on the Spoolman path it
would have shipped with the guard above. The external holder reports 254
when it is genuinely in use.

The arithmetic is untouched: at one percent per step this cannot resolve
a small print, and pretending otherwise would be worse than saying so.

backend/app/api/routes/archives.py

@@ -32,8 +32,8 @@ from backend.app.services.archive import ArchiveService
 from backend.app.services.design_settings import overrides_from_config
 from backend.app.services.filament_requirements import annotate_rack_groups
 from backend.app.services.print_storage import REASON_INTERNAL_STORAGE, REASON_NO_EXTERNAL_STORAGE
+from backend.app.utils.archive_paths import archive_photos_dir, find_archive_photo
 from backend.app.utils.http import build_content_disposition
-from backend.app.utils.safe_path import safe_join_under
 from backend.app.utils.threemf_tools import (
     default_plate_gcode_name,
     expand_to_project_slots,
@@ -2932,10 +2932,10 @@ async def upload_photo(
     if not file.filename or not file.filename.lower().endswith((".jpg", ".jpeg", ".png", ".webp")):
         raise HTTPException(400, "File must be an image (.jpg, .jpeg, .png, .webp)")
 
-    # Get archive directory
-    archive_dir = settings.base_dir / Path(archive.file_path).parent
-    photos_dir = archive_dir / "photos"
-    photos_dir.mkdir(exist_ok=True)
+    # Get archive directory. parents=True because an archive with no 3MF owns
+    # <archive_dir>/<id>/, which nothing else has necessarily created yet.
+    photos_dir = archive_photos_dir(archive)
+    photos_dir.mkdir(parents=True, exist_ok=True)
 
     # Generate unique filename
     import uuid
@@ -2982,15 +2982,14 @@ async def get_photo(
     if not archive.photos or filename not in archive.photos:
         raise HTTPException(404, "Photo not found")
 
-    archive_dir = settings.base_dir / Path(archive.file_path).parent
-    photos_dir = archive_dir / "photos"
     # Defence-in-depth: even though the membership check above already
-    # constrains `filename` to UUID-generated names from upload, the
-    # resolve + containment check guards against future code paths that
-    # might populate `archive.photos` from a less-trusted source.
-    photo_path = safe_join_under(photos_dir, filename)
+    # constrains `filename` to UUID-generated names from upload,
+    # find_archive_photo resolves and containment-checks each candidate,
+    # guarding against future code paths that might populate
+    # `archive.photos` from a less-trusted source.
+    photo_path = find_archive_photo(archive, filename)
 
-    if not photo_path.exists():
+    if photo_path is None:
         raise HTTPException(404, "Photo not found")
 
     # Determine media type
@@ -3026,11 +3025,11 @@ async def delete_photo(
     if not archive.photos or filename not in archive.photos:
         raise HTTPException(404, "Photo not found")
 
-    # Delete file — same defence-in-depth as get_photo above.
-    archive_dir = settings.base_dir / Path(archive.file_path).parent
-    photos_dir = archive_dir / "photos"
-    photo_path = safe_join_under(photos_dir, filename)
-    if photo_path.exists():
+    # Delete file — same lookup as get_photo above, so a photo that is
+    # readable is also deletable. Removing the name while leaving the file is
+    # how a no-3MF archive accumulated photos nobody could see or remove.
+    photo_path = find_archive_photo(archive, filename)
+    if photo_path is not None:
         photo_path.unlink()
 
     # Update archive photos list

backend/app/core/database.py

@@ -2374,6 +2374,7 @@ async def run_migrations(conn):
             layer_usage TEXT,
             filament_properties TEXT,
             tray_remain_start TEXT,
+            tray_now_at_start INTEGER,
             UNIQUE(printer_id, archive_id)
         )
         """
@@ -2389,6 +2390,7 @@ async def run_migrations(conn):
             layer_usage TEXT,
             filament_properties TEXT,
             tray_remain_start TEXT,
+            tray_now_at_start INTEGER,
             UNIQUE(printer_id, archive_id)
         )
         """,
@@ -2397,6 +2399,18 @@ async def run_migrations(conn):
     # the original schema: add tray_remain_start, and relax filament_usage's
     # NOT NULL so the no-3MF branch can persist a remain-only tracking row.
     await _safe_execute(conn, "ALTER TABLE active_print_spoolman ADD COLUMN tray_remain_start TEXT")
+    # Which slot the print was drawing from at the start, so the remain%-delta
+    # fallback can tell a slot this print used from one it never touched
+    # (#1820). Nullable, because a row written mid-upgrade has no answer to
+    # give. INTEGER is spelled the same either way; the branch is only for
+    # IF NOT EXISTS, which SQLite's ALTER TABLE does not accept.
+    if is_sqlite():
+        await _safe_execute(conn, "ALTER TABLE active_print_spoolman ADD COLUMN tray_now_at_start INTEGER")
+    else:
+        await _safe_execute(
+            conn,
+            "ALTER TABLE active_print_spoolman ADD COLUMN IF NOT EXISTS tray_now_at_start INTEGER",
+        )
     if is_sqlite():
         # SQLite can't ALTER COLUMN; patch sqlite_master directly. Mirrors the
         # users.password_hash NULL-relaxation a few hundred lines below — see

backend/app/main.py

@@ -6327,13 +6327,12 @@ async def on_print_complete(printer_id: int, data: dict):
 
             import uuid
             from datetime import datetime
-            from pathlib import Path
 
-            if archive.file_path:
-                archive_dir = app_settings.base_dir / Path(archive.file_path).parent
-            else:
+            from backend.app.utils.archive_paths import archive_dir as resolve_archive_dir
+
+            if not archive.file_path:
                 logger.warning("[PHOTO-BG] Archive %s has no file_path, using fallback dir", archive_id)
-                archive_dir = app_settings.archive_dir / str(archive.id)
+            archive_dir = resolve_archive_dir(archive)
             photo_filename = None
 
             # Prefer the timelapse last-frame source when a timelapse was
@@ -6653,15 +6652,10 @@ async def on_print_complete(printer_id: int, data: dict):
 
                             # Read finish photo bytes for image attachment (e.g. Pushover)
                             try:
-                                from pathlib import Path
+                                from backend.app.utils.archive_paths import find_archive_photo
 
-                                photo_path = (
-                                    app_settings.base_dir
-                                    / Path(archive.file_path).parent
-                                    / "photos"
-                                    / finish_photo_filename
-                                )
-                                if photo_path.exists():
+                                photo_path = find_archive_photo(archive, finish_photo_filename)
+                                if photo_path is not None:
                                     photo_bytes = await asyncio.to_thread(photo_path.read_bytes)
                                     if len(photo_bytes) <= 2_500_000:
                                         archive_data["image_data"] = photo_bytes

backend/app/models/active_print_spoolman.py

@@ -51,3 +51,14 @@ class ActivePrintSpoolman(Base):
     # ``tray_remain_start`` snapshot at usage_tracker.py:301.
     # Format: {"<ams_id>-<tray_id>": {"remain": int, "tray_uuid": str}, ...}
     tray_remain_start: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
+    # Global tray id the printer was drawing from when the print started.
+    # Evidence of which slot the print actually used, for the remain%-delta
+    # fallback (#1269 on the internal side, #1820 here): without it, a spool
+    # swapped in an untouched slot mid-print reads as consumption and is
+    # charged to whatever spool that slot was assigned. Often the only
+    # evidence there is — a print started from the printer's own screen
+    # carries no ams_mapping and may change tray never. Nullable: rows written
+    # before this column existed, and printers that report no tray_now, simply
+    # provide no evidence and are handled as before.
+    tray_now_at_start: Mapped[int | None] = mapped_column(nullable=True)

backend/app/services/spoolman_tracking.py

@@ -26,6 +26,19 @@ logger = logging.getLogger(__name__)
 _ZERO_UUID = "00000000000000000000000000000000"
 _ZERO_TAG_UID = "0000000000000000"
 
+# Highest global tray id that names a real slot. 255 does not: it is
+# ``PrinterState.tray_now``'s initial value, what an unparseable reading falls
+# back to, and what the field reads while nothing is loaded. The external spool
+# reports 254 when it is actually in use, and ``bambu_mqtt`` applies the same
+# cut-off when it seeds the tray-change log. Treating 255 as a slot would put
+# ``(255, 1)`` into the "slots this print used" evidence and exclude every real
+# one -- silently disabling the very fallback this guard protects (#1820).
+#
+# Applied to ``tray_now`` only. A 255 in the print's mapping or its tray-change
+# log was written there by a print and is evidence, however odd; a 255 in
+# ``tray_now`` is the field at rest, which is the absence of evidence.
+_MAX_REAL_TRAY_ID = 254
+
 
 def _is_non_zero_identifier(value: str) -> bool:
     """Return True when identifier is non-empty and not all zeros."""
@@ -238,7 +251,7 @@ def build_ams_tray_lookup(raw_data: dict) -> dict[int, dict]:
     return lookup
 
 
-def _snapshot_tray_remain(raw_data: dict) -> dict[str, dict]:
+def _snapshot_tray_remain(raw_data: dict, skipped_out: list[str] | None = None) -> dict[str, dict]:
     """Capture per-slot ``remain%`` + ``tray_uuid`` at print start so the
     completion path can compute a remain-delta when 3MF data doesn't cover
     the slot (or there's no 3MF at all — #1820).
@@ -248,6 +261,12 @@ def _snapshot_tray_remain(raw_data: dict) -> dict[str, dict]:
     values mean the AMS hasn't read the spool yet and a delta would be
     meaningless. Mirrors the gate in
     ``usage_tracker.on_print_start:309``.
+
+    A rejected slot is appended to *skipped_out* when one is supplied, so the
+    caller can say which slots this print will not be able to charge. That is
+    not hypothetical: an AMS reports a negative ``remain`` on a nearly empty
+    spool, so the gate can drop the one slot that is about to do the printing
+    (#1820).
     """
     snapshot: dict[str, dict] = {}
     ams_raw = raw_data.get("ams", [])
@@ -266,6 +285,8 @@ def _snapshot_tray_remain(raw_data: dict) -> dict[str, dict]:
                     "remain": remain,
                     "tray_uuid": tray.get("tray_uuid", "") or "",
                 }
+            elif skipped_out is not None:
+                skipped_out.append(f"AMS{ams_id}-T{tray_id}(remain={remain})")
     vt_tray_raw = raw_data.get("vt_tray") or []
     if isinstance(vt_tray_raw, dict):
         vt_tray_raw = [vt_tray_raw]
@@ -281,6 +302,8 @@ def _snapshot_tray_remain(raw_data: dict) -> dict[str, dict]:
                 "remain": remain,
                 "tray_uuid": vt.get("tray_uuid", "") or "",
             }
+        elif skipped_out is not None:
+            skipped_out.append(f"VT{vt_id}(remain={remain})")
     return snapshot
 
 
@@ -329,7 +352,17 @@ async def store_print_data(
     tray_remain_start: dict[str, dict] = {}
     if state and state.raw_data:
         ams_trays = build_ams_tray_lookup(state.raw_data)
-        tray_remain_start = _snapshot_tray_remain(state.raw_data)
+        skipped_slots: list[str] = []
+        tray_remain_start = _snapshot_tray_remain(state.raw_data, skipped_slots)
+        if skipped_slots:
+            # Matches what usage_tracker.on_print_start reports for the
+            # internal inventory, so both backends name the slots that this
+            # print will not be able to charge at AMS granularity.
+            logger.info(
+                "[SPOOLMAN] Printer %s: slots with no usable remain%% at print start: %s",
+                printer_id,
+                ", ".join(skipped_slots),
+            )
 
     # Try to read per-slot filament estimates from the 3MF. Two paths can
     # leave ``filament_usage`` empty: (1) fallback archive (no .gcode.3mf
@@ -409,6 +442,11 @@ async def store_print_data(
         layer_usage=layer_usage_json,
         filament_properties=filament_properties,
         tray_remain_start=tray_remain_start or None,
+        # Which slot the printer was drawing from when this print began. For a
+        # print with no ams_mapping -- one started from the printer's own
+        # screen, which is the case this whole fallback exists for -- it is the
+        # only evidence of which slot the print used (#1820).
+        tray_now_at_start=getattr(state, "tray_now", None) if state else None,
     )
     db.add(tracking)
     await db.commit()
@@ -881,6 +919,7 @@ async def _report_partial_usage(
             current_lookup=current_lookup,
             handled_global_tray_ids=set(),
             archive_id=getattr(tracking, "archive_id", -1),
+            print_used_keys=_print_used_tray_keys(slot_to_tray, getattr(tracking, "tray_now_at_start", None), state),
         )
         return
 
@@ -1026,6 +1065,7 @@ async def report_usage(printer_id: int, archive_id: int):
         # on read.
         layer_usage_raw = getattr(tracking, "layer_usage", None) or {}
         filament_properties = getattr(tracking, "filament_properties", None) or {}
+        tray_now_at_start = getattr(tracking, "tray_now_at_start", None)
         printer_serial = await _get_printer_serial(printer_id)
 
         # Delete tracking row (we're done with it)
@@ -1177,6 +1217,7 @@ async def report_usage(printer_id: int, archive_id: int):
                 current_lookup=current_lookup,
                 handled_global_tray_ids=handled_global_tray_ids,
                 archive_id=archive_id,
+                print_used_keys=_print_used_tray_keys(slot_to_tray, tray_now_at_start, current),
                 slot_colors_out=slot_colors,
                 slot_materials_out=slot_materials,
             )
@@ -1197,6 +1238,49 @@ async def report_usage(printer_id: int, archive_id: int):
         await _apply_spool_types_to_archive(db, archive_id, filament_usage, slot_materials)
 
 
+def _print_used_tray_keys(
+    slot_to_tray: list | None,
+    tray_now_at_start: int | None,
+    state,
+) -> set[tuple[int, int]]:
+    """Which AMS slots this print actually drew from, as far as we can tell.
+
+    Mirrors the guard the internal tracker has carried since #1269. Without
+    it, swapping a spool in a slot the print never touched drops that slot's
+    ``remain%``, and the remain-delta path reads the drop as consumption and
+    charges it to whoever the slot is assigned to. That is a phantom write to
+    an uninvolved spool, and it is likeliest on exactly the prints this
+    fallback serves -- ones with no 3MF, where nothing else limits which slots
+    are considered.
+
+    Three sources, matching the internal tracker's:
+
+    - the print's ``ams_mapping``, stored here as ``slot_to_tray``;
+    - every tray the printer switched to mid-print;
+    - the tray it was drawing from at the start.
+
+    An empty result means no evidence, not "no slots" -- callers must then
+    consider every slot, as before, or a printer that reports none of the
+    three would silently stop being tracked at all.
+
+    Takes the two stored values rather than the tracking row: the caller
+    deletes that row before it gets this far, and everything read off it is
+    read into locals beforehand.
+    """
+    keys: set[tuple[int, int]] = set()
+    for global_tray_id in list(slot_to_tray or []):
+        if isinstance(global_tray_id, int) and global_tray_id >= 0:
+            keys.add(_global_tray_id_to_ams_slot(global_tray_id))
+    for change in getattr(state, "tray_change_log", None) or []:
+        if isinstance(change, (tuple, list)) and change:
+            global_tray_id = change[0]
+            if isinstance(global_tray_id, int) and global_tray_id >= 0:
+                keys.add(_global_tray_id_to_ams_slot(global_tray_id))
+    if isinstance(tray_now_at_start, int) and 0 <= tray_now_at_start <= _MAX_REAL_TRAY_ID:
+        keys.add(_global_tray_id_to_ams_slot(tray_now_at_start))
+    return keys
+
+
 async def _report_remain_delta_for_slots(
     client,
     *,
@@ -1205,6 +1289,7 @@ async def _report_remain_delta_for_slots(
     current_lookup: dict[str, dict],
     handled_global_tray_ids: set[int],
     archive_id: int,
+    print_used_keys: set[tuple[int, int]] | None = None,
     slot_colors_out: dict[int, str] | None = None,
     slot_materials_out: dict[int, str] | None = None,
 ) -> int:
@@ -1217,6 +1302,7 @@ async def _report_remain_delta_for_slots(
     unreliable ``tray_weight`` (which is the failure mode #1119 documented).
     """
     spools_updated = 0
+    not_in_print: list[str] = []
     for slot_key, start in tray_remain_start.items():
         try:
             ams_id_str, tray_id_str = slot_key.split("-", 1)
@@ -1236,9 +1322,24 @@ async def _report_remain_delta_for_slots(
         if global_tray_id in handled_global_tray_ids:
             continue
 
+        # Slots the print never touched (#1269's guard, see _print_used_tray_keys).
+        # Only enforced when there is evidence of which slots it did use.
+        # Collected rather than logged per slot: on a four-AMS farm a
+        # single-colour print leaves fifteen of these, and they are the
+        # expected case, unlike the "consumed but charged nothing" lines below.
+        if print_used_keys and (ams_id, tray_id) not in print_used_keys:
+            not_in_print.append(f"AMS{ams_id}-T{tray_id}")
+            continue
+
         current = current_lookup.get(slot_key)
         if not current:
-            logger.debug("[SPOOLMAN] AMS%d-T%d: no current remain%% at completion, skipping fallback", ams_id, tray_id)
+            # Reported at info, like the internal tracker's equivalent: on a
+            # near-empty spool the AMS reports a negative remain%, which the
+            # snapshot gate rejects, and the slot that was actually printing
+            # disappears from this path entirely (#1820).
+            logger.info(
+                "[SPOOLMAN] AMS%d-T%d: no valid remain%% at completion, nothing charged for this slot", ams_id, tray_id
+            )
             continue
 
         # Spool swap mid-print — tray_uuid changed. We don't know how much
@@ -1253,11 +1354,28 @@ async def _report_remain_delta_for_slots(
 
         delta_pct = start["remain"] - current["remain"]
         if delta_pct <= 0:
+            # A fresh spool reads 100% for the first tens of grams and the AMS
+            # estimate drifts upward on its own, so this covers a real print
+            # that simply left no trace at AMS granularity -- not only a refill.
+            # Said out loud so it can be told apart from having nothing to
+            # charge, which is what "no spools updated" alone looked like.
+            logger.info(
+                "[SPOOLMAN] AMS%d-T%d: remain%% did not fall over the print (%d%% -> %d%%), nothing charged",
+                ams_id,
+                tray_id,
+                start["remain"],
+                current["remain"],
+            )
             continue  # No consumption captured at AMS granularity, or refilled
 
         spool_id = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id)
         if spool_id is None:
-            logger.debug("[SPOOLMAN] AMS%d-T%d: no Spoolman slot assignment, skipping fallback", ams_id, tray_id)
+            logger.info(
+                "[SPOOLMAN] AMS%d-T%d: consumed %d%% but has no Spoolman slot assignment, nothing charged",
+                ams_id,
+                tray_id,
+                delta_pct,
+            )
             continue
 
         # Look up the spool's filament reference weight. Use a fresh GET so
@@ -1314,6 +1432,12 @@ async def _report_remain_delta_for_slots(
             ref_weight,
             spool_id,
         )
+    if not_in_print:
+        logger.info(
+            "[SPOOLMAN] Archive %s: slots not part of this print, left alone: %s",
+            archive_id,
+            ", ".join(not_in_print),
+        )
     return spools_updated
 
 

backend/app/services/usage_tracker.py

@@ -791,7 +791,13 @@ async def on_print_complete(
                     gid = change[0]
                     if isinstance(gid, int) and gid >= 0:
                         print_used_keys.add(_global_to_ams_key(gid))
-            if session.tray_now_at_start is not None and session.tray_now_at_start >= 0:
+            # 255 is not a slot: it is what ``tray_now`` reads at rest, before
+            # the printer has reported one and while nothing is loaded, and an
+            # unparseable reading falls back to it too. Mapped as a tray id it
+            # becomes (255, 1), and if it were the only evidence every real
+            # slot would be excluded and the fallback would charge nothing at
+            # all (#1820). The external spool reports 254 when in use.
+            if session.tray_now_at_start is not None and 0 <= session.tray_now_at_start <= 254:
                 print_used_keys.add(_global_to_ams_key(session.tray_now_at_start))
 
             # Collect all trays to check: AMS trays + VT (external) trays
@@ -851,7 +857,23 @@ async def on_print_complete(
                 delta_pct = start_remain - current_remain
 
                 if delta_pct <= 0:
-                    continue  # No consumption or tray was refilled
+                    # Not necessarily "nothing was printed". A fresh spool sits
+                    # at 100% for the first tens of grams, and the AMS estimate
+                    # drifts upward on its own, so a real print can end with the
+                    # same or a higher reading than it started with. Said out
+                    # loud because the alternative -- charging nothing, silently
+                    # -- is indistinguishable from having nothing to charge, and
+                    # the operator has no other way to find the prints that went
+                    # uncounted (#1820).
+                    logger.info(
+                        "[UsageTracker] %s: remain%% did not fall over the print (%d%% -> %d%%), "
+                        "nothing charged for printer %d",
+                        tray_label,
+                        start_remain,
+                        current_remain,
+                        printer_id,
+                    )
+                    continue
 
                 spool_id = await _resolve_spool_id_for_tray(
                     printer_id=printer_id,

backend/app/utils/archive_paths.py

@@ -0,0 +1,79 @@
+"""Where an archive's files live on disk (#1820).
+
+An archive normally owns a directory, derived from its ``file_path``:
+``<base_dir>/<dirname of file_path>/``. An archive created without a 3MF has
+``file_path == ""``, and ``Path("").parent`` is ``Path(".")`` -- so every site
+that derived the directory that way silently resolved to ``base_dir`` itself,
+and all such archives shared one pile.
+
+The finish-photo capture path spotted that and used ``<archive_dir>/<id>/``
+instead. Nothing else did, so a captured photo was written to one directory and
+then looked for in another: the read 404'd, the delete removed the name and
+left the file, and the notification attachment never found the image. Four
+sites deriving the same directory four times is what let them drift, so they
+now all ask here.
+
+Photos written before this are still where they were put, which is why lookups
+check both locations rather than only the current one.
+
+Scope note: a *source 3MF* uploaded onto a no-3MF archive has its own layout,
+``archive/no_source/<id>/``, chosen separately and stored in its own column.
+This module does not model that -- do not reach for ``archive_dir`` to find one.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from backend.app.core.config import settings
+from backend.app.utils.safe_path import PathTraversalError, safe_join_under
+
+
+def archive_dir(archive: object) -> Path:
+    """The directory belonging to *archive*.
+
+    Falls back to ``<archive_dir>/<id>`` for an archive with no 3MF, matching
+    what the finish-photo capture has always written.
+    """
+    file_path = getattr(archive, "file_path", "") or ""
+    if file_path:
+        return settings.base_dir / Path(file_path).parent
+    return settings.archive_dir / str(archive.id)  # SEC-PATH-OK: archive.id is an int primary key
+
+
+def archive_photos_dir(archive: object) -> Path:
+    """Where photos for *archive* are written."""
+    return archive_dir(archive) / "photos"  # SEC-PATH-OK: constant subdirectory
+
+
+def _legacy_shared_photos_dir(archive: object) -> Path | None:
+    """Where a no-3MF archive's photos used to be read from, and uploaded to.
+
+    ``<base_dir>/photos``, shared by every no-3MF archive at once. Only ever
+    consulted for an archive that has no ``file_path``; one with a real path
+    always resolved correctly and has no second location to check.
+    """
+    if getattr(archive, "file_path", "") or "":
+        return None
+    return settings.base_dir / "photos"  # SEC-PATH-OK: constant subdirectory
+
+
+def find_archive_photo(archive: object, filename: str) -> Path | None:
+    """Locate an existing photo, or None if it is in neither location.
+
+    *filename* must already have been checked for membership in
+    ``archive.photos``; it is joined containment-checked regardless. A name
+    that fails that check is treated as not found rather than raised on --
+    one caller is a background notification task, where an HTTP error would
+    have nowhere to go.
+    """
+    for directory in (archive_photos_dir(archive), _legacy_shared_photos_dir(archive)):
+        if directory is None:
+            continue
+        try:
+            candidate = safe_join_under(directory, filename, http=False)
+        except PathTraversalError:
+            return None
+        if candidate.exists():
+            return candidate
+    return None

backend/tests/integration/test_fallback_archive_photos_1820.py

@@ -0,0 +1,179 @@
+"""Photos on an archive that has no 3MF (#1820).
+
+The finish-photo capture writes to ``<archive_dir>/<id>/photos/`` when the
+archive has no ``file_path``. Every reader derived the directory from
+``file_path`` instead, and ``Path("").parent`` is ``Path(".")`` -- so they all
+resolved to ``<base_dir>/photos``. The photo was written to one place and
+looked for in another: reads 404'd, deletes removed the name and left the file,
+and the notification attachment never found the image.
+
+The reporter hit the read. These cover all of it, plus the photos already
+written to the old location, which must not become unreachable in the fix.
+"""
+
+import pytest
+from httpx import AsyncClient
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+PHOTO = "deadbeef.jpg"
+JPEG = b"\xff\xd8\xff\xe0" + b"0" * 64
+
+
+@pytest.fixture
+def base_dir(monkeypatch, tmp_path):
+    """Point both roots at a tmp dir, keeping their real relationship."""
+    from backend.app.core.config import settings
+
+    monkeypatch.setattr(settings, "base_dir", tmp_path)
+    monkeypatch.setattr(settings, "archive_dir", tmp_path / "archive")
+    return tmp_path
+
+
+async def _fallback_archive(archive_factory, printer_factory, **kwargs):
+    printer = await printer_factory()
+    return await archive_factory(
+        printer.id,
+        print_name="Started From The Printer",
+        filename="Started From The Printer.3mf",
+        file_path="",
+        **kwargs,
+    )
+
+
+def _write(directory, name=PHOTO, content=JPEG):
+    directory.mkdir(parents=True, exist_ok=True)
+    (directory / name).write_bytes(content)
+
+
+class TestReadingACapturedPhoto:
+    async def test_a_captured_finish_photo_is_served(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        """The reporter's 404: written by the capture, unreadable forever."""
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+        _write(base_dir / "archive" / str(archive.id) / "photos")
+
+        response = await async_client.get(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 200
+        assert response.content == JPEG
+
+    async def test_a_photo_in_the_old_shared_location_is_still_served(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        """Manual uploads landed in <base_dir>/photos, where reads also looked,
+        so those worked. Moving the lookup must not orphan them."""
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+        _write(base_dir / "photos")
+
+        response = await async_client.get(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 200
+        assert response.content == JPEG
+
+    async def test_a_photo_that_is_in_neither_place_is_404(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+
+        response = await async_client.get(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 404
+
+    async def test_a_name_not_on_the_archive_is_404(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        """The membership check comes first and still does."""
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+        _write(base_dir / "archive" / str(archive.id) / "photos", name="someone_elses.jpg")
+
+        response = await async_client.get(f"/api/v1/archives/{archive.id}/photos/someone_elses.jpg")
+
+        assert response.status_code == 404
+
+
+class TestANormalArchiveIsUnaffected:
+    async def test_it_reads_from_its_own_directory(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, file_path="archives/test/print.gcode.3mf", photos=[PHOTO])
+        _write(base_dir / "archives" / "test" / "photos")
+
+        response = await async_client.get(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 200
+
+    async def test_it_does_not_borrow_from_the_shared_location(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        """An archive with a real path has one location and only one. Looking
+        in the shared pile as well would serve another archive's photo when
+        the names ever collided."""
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, file_path="archives/test/print.gcode.3mf", photos=[PHOTO])
+        _write(base_dir / "photos")
+
+        response = await async_client.get(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 404
+
+
+class TestDeleting:
+    async def test_a_captured_photo_is_removed_from_disk(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        """It used to drop the name and leave the file, so the photo became
+        both invisible and unremovable."""
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+        photos_dir = base_dir / "archive" / str(archive.id) / "photos"
+        _write(photos_dir)
+
+        response = await async_client.delete(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 200
+        assert not (photos_dir / PHOTO).exists()
+
+    async def test_a_photo_in_the_old_location_is_removed_too(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+        _write(base_dir / "photos")
+
+        response = await async_client.delete(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 200
+        assert not (base_dir / "photos" / PHOTO).exists()
+
+    async def test_a_missing_file_still_clears_the_name(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        archive = await _fallback_archive(archive_factory, printer_factory, photos=[PHOTO])
+
+        response = await async_client.delete(f"/api/v1/archives/{archive.id}/photos/{PHOTO}")
+
+        assert response.status_code == 200
+        assert response.json()["photos"] is None
+
+
+class TestUploading:
+    async def test_an_uploaded_photo_can_be_read_back(
+        self, async_client: AsyncClient, archive_factory, printer_factory, base_dir
+    ):
+        """Upload and read now agree on the location for these archives, which
+        also means the directory has to be created with its parents."""
+        archive = await _fallback_archive(archive_factory, printer_factory)
+
+        upload = await async_client.post(
+            f"/api/v1/archives/{archive.id}/photos",
+            files={"file": ("shot.jpg", JPEG, "image/jpeg")},
+        )
+        assert upload.status_code == 200, upload.text
+        filename = upload.json()["filename"]
+
+        assert (base_dir / "archive" / str(archive.id) / "photos" / filename).is_file()
+
+        read_back = await async_client.get(f"/api/v1/archives/{archive.id}/photos/{filename}")
+        assert read_back.status_code == 200
+        assert read_back.content == JPEG

backend/tests/unit/services/test_usage_tracker.py

@@ -186,6 +186,63 @@ class TestOnPrintCompleteAMSDelta:
         assert results == []
         db.commit.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_an_unloaded_tray_at_start_does_not_exclude_every_slot(self):
+        """tray_now reads 255 at rest -- its initial value, and what an
+        unparseable reading falls back to. Mapped as a tray id that is (255, 1),
+        so taking it as evidence of which slots the print used would exclude
+        every real one and charge nothing (#1820)."""
+        _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=255,
+        )
+        ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
+        pm = _make_printer_manager(_make_printer_state(ams_data))
+
+        spool = _make_spool(label_weight=1000, weight_used=0)
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(),  # _find_3mf_by_filename: library search
+                MagicMock(),  # _find_3mf_by_filename: archive search
+                MagicMock(scalar_one_or_none=MagicMock(return_value=_make_assignment())),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        results = await on_print_complete(1, {"status": "completed"}, pm, db)
+
+        assert len(results) == 1
+        assert results[0]["weight_used"] == 100.0
+
+    @pytest.mark.asyncio
+    async def test_a_delta_that_charges_nothing_says_so(self, caplog):
+        """Charging nothing has to be distinguishable from having nothing to
+        charge (#1820). A fresh spool reads 100% for its first tens of grams
+        and the AMS estimate drifts upward on its own, so this fires on real
+        prints, not only on refills -- and used to fire in complete silence."""
+        import logging
+
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="test",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 100},
+        )
+        ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 100}]}]
+        pm = _make_printer_manager(_make_printer_state(ams_data))
+        db = AsyncMock()
+
+        with caplog.at_level(logging.INFO, logger="backend.app.services.usage_tracker"):
+            results = await on_print_complete(1, {"status": "completed"}, pm, db)
+
+        assert results == []
+        assert "did not fall" in caplog.text
+        assert "100% -> 100%" in caplog.text
+
     @pytest.mark.asyncio
     async def test_no_session_falls_through_to_3mf(self):
         """When no session exists, AMS delta path skipped (3MF may still run)."""

backend/tests/unit/test_remain_delta_silence_1820.py

@@ -0,0 +1,282 @@
+"""The remain%-delta fallback must say when it charges nothing (#1820).
+
+The fallback exists so a print with no 3MF still moves the spool weight. On an
+H2S the AMS ``remain%`` it reads is too coarse and too noisy to carry that: the
+reporter measured it rising mid-print, swinging +/-5 points over one job,
+saturating at 100 on a fresh spool, and going negative near the end of one.
+
+Two of their prints wrote nothing, each for a different one of those reasons,
+and both looked identical from the outside -- ``no spools updated``, which is
+also what a print with genuinely nothing to charge prints. The arithmetic is a
+separate question; this is about not failing silently, so an operator can tell
+which prints need correcting by hand.
+"""
+
+import logging
+import types
+
+import pytest
+
+from backend.app.services.spoolman_tracking import (
+    _print_used_tray_keys,
+    _report_remain_delta_for_slots,
+    _snapshot_tray_remain,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _raw(remain, tray_uuid="uuid-a"):
+    return {"ams": [{"id": 0, "tray": [{"id": 0, "remain": remain, "tray_uuid": tray_uuid}]}]}
+
+
+def _slot(remain, tray_uuid="uuid-a"):
+    return {"0-0": {"remain": remain, "tray_uuid": tray_uuid}}
+
+
+class _Client:
+    """Records anything the fallback tries to write."""
+
+    def __init__(self):
+        self.used = []
+
+    async def get_spool(self, spool_id):
+        return {"filament": {"weight": 1000}}
+
+    async def use_spool(self, spool_id, grams):
+        self.used.append((spool_id, grams))
+
+
+async def _run(caplog, **kwargs):
+    client = _Client()
+    with caplog.at_level(logging.INFO, logger="backend.app.services.spoolman_tracking"):
+        written = await _report_remain_delta_for_slots(
+            client,
+            printer_id=1,
+            handled_global_tray_ids=set(),
+            archive_id=7,
+            **kwargs,
+        )
+    return client, written, caplog.text
+
+
+class TestTheSnapshotGate:
+    """A negative remain% -- what the AMS reports on a nearly empty spool --
+    keeps the slot out of the snapshot entirely. That is how the reporter's
+    second print lost the only slot that was printing."""
+
+    def test_a_negative_remain_is_reported_as_skipped(self):
+        skipped = []
+
+        snapshot = _snapshot_tray_remain(_raw(-3), skipped)
+
+        assert snapshot == {}
+        assert skipped == ["AMS0-T0(remain=-3)"]
+
+    def test_a_valid_remain_is_not_reported(self):
+        skipped = []
+
+        snapshot = _snapshot_tray_remain(_raw(42), skipped)
+
+        assert snapshot == {"0-0": {"remain": 42, "tray_uuid": "uuid-a"}}
+        assert skipped == []
+
+    def test_the_external_spool_holder_is_reported_too(self):
+        skipped = []
+
+        _snapshot_tray_remain({"vt_tray": {"id": 254, "remain": -1}}, skipped)
+
+        assert skipped == ["VT254(remain=-1)"]
+
+    def test_the_collector_is_optional(self):
+        """Two of the three call sites pass nothing; they must still work."""
+        assert _snapshot_tray_remain(_raw(-3)) == {}
+
+
+@pytest.mark.asyncio
+class TestNothingCharged:
+    async def test_a_spool_still_reading_full_is_reported(self, caplog):
+        """The reporter's first print: 36 minutes on a fresh spool, 100% at
+        both ends, so the delta was zero and the slot was skipped in silence."""
+        client, written, text = await _run(caplog, tray_remain_start=_slot(100), current_lookup=_slot(100))
+
+        assert written == 0
+        assert client.used == []
+        assert "did not fall" in text
+        assert "100% -> 100%" in text
+
+    async def test_a_reading_that_rose_is_reported(self, caplog):
+        """remain% moving upward mid-print is noise, not a refill, but either
+        way nothing is charged and the operator should hear about it."""
+        _, written, text = await _run(caplog, tray_remain_start=_slot(12), current_lookup=_slot(17))
+
+        assert written == 0
+        assert "12% -> 17%" in text
+
+    async def test_a_slot_missing_at_completion_is_reported(self, caplog):
+        """The completion-side twin of the snapshot gate."""
+        _, written, text = await _run(caplog, tray_remain_start=_slot(50), current_lookup={})
+
+        assert written == 0
+        assert "no valid remain" in text
+
+    async def test_an_unassigned_slot_names_what_was_lost(self, caplog, monkeypatch):
+        """It consumed something real and there is nowhere to put it, which is
+        worth more than the debug line it used to get."""
+        monkeypatch.setattr(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+            _fake_resolver(None),
+        )
+
+        _, written, text = await _run(caplog, tray_remain_start=_slot(60), current_lookup=_slot(50))
+
+        assert written == 0
+        assert "no Spoolman slot assignment" in text
+        assert "consumed 10%" in text
+
+
+@pytest.mark.asyncio
+class TestItStillWritesWhenItCan:
+    async def test_a_real_drop_is_charged(self, caplog, monkeypatch):
+        monkeypatch.setattr(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+            _fake_resolver(42),
+        )
+
+        client, written, text = await _run(caplog, tray_remain_start=_slot(60), current_lookup=_slot(50))
+
+        assert written == 1
+        assert client.used == [(42, 100.0)]  # 10% of a 1000 g reference weight
+        assert "did not fall" not in text
+
+    async def test_a_spool_swap_is_still_refused(self, caplog, monkeypatch):
+        monkeypatch.setattr(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+            _fake_resolver(42),
+        )
+
+        client, written, text = await _run(
+            caplog,
+            tray_remain_start=_slot(60, "uuid-a"),
+            current_lookup=_slot(10, "uuid-b"),
+        )
+
+        assert written == 0
+        assert client.used == []
+        assert "swapped mid-print" in text
+
+
+def _fake_resolver(spool_id):
+    async def _resolve(*_args, **_kwargs):
+        return spool_id
+
+    return _resolve
+
+
+class TestWhichSlotsThePrintUsed:
+    """The guard the internal tracker has had since #1269, now on this path
+    too. Without it a spool swapped into an idle slot mid-print reads as
+    consumption and is charged to whatever that slot is assigned to."""
+
+    def test_the_mapping_names_the_slots(self):
+        """Global tray ids: 0-3 are AMS 0, 4-7 are AMS 1."""
+        assert _print_used_tray_keys([0, 5], None, None) == {(0, 0), (1, 1)}
+
+    def test_a_slicer_slot_routed_to_the_external_spool_is_ignored(self):
+        """-1 means "external spool" in the flat mapping and names no AMS slot;
+        the external holder arrives as 254/255 when it is really used."""
+        assert _print_used_tray_keys([-1], None, None) == set()
+        assert _print_used_tray_keys([254], None, None) == {(255, 0)}
+
+    def test_an_ams_ht_keeps_its_own_id(self):
+        assert _print_used_tray_keys([128], None, None) == {(128, 0)}
+
+    def test_a_mid_print_tray_change_counts(self):
+        """Filament backup switches trays mid-print; the substitute fed part of
+        the job and has to be chargeable."""
+        state = types.SimpleNamespace(tray_change_log=[[0, 0], [5, 120]])
+
+        assert _print_used_tray_keys(None, None, state) == {(0, 0), (1, 1)}
+
+    def test_the_tray_in_use_at_the_start_counts(self):
+        """Often the only evidence: a print started from the printer's screen
+        carries no mapping and may never change tray."""
+        assert _print_used_tray_keys(None, 2, None) == {(0, 2)}
+
+    def test_an_unloaded_printer_is_not_read_as_a_slot(self):
+        """255 is what tray_now reads at rest -- its initial value, the
+        unparseable-reading fallback, and "nothing loaded". Mapped as a tray id
+        it becomes (255, 1), and as the only evidence it would exclude every
+        real slot and charge nothing at all, which is this issue's own bug."""
+        assert _print_used_tray_keys(None, 255, None) == set()
+
+    def test_the_external_spool_in_use_is_a_slot(self):
+        """It reports 254 when actually in use, which is a real slot."""
+        assert _print_used_tray_keys(None, 254, None) == {(255, 0)}
+
+    def test_no_evidence_at_all_yields_nothing(self):
+        """Which callers must read as "consider every slot", not "no slots" --
+        otherwise a printer reporting none of the three stops being tracked."""
+        assert _print_used_tray_keys(None, None, None) == set()
+        assert _print_used_tray_keys([], -1, types.SimpleNamespace(tray_change_log=[])) == set()
+
+    def test_a_row_written_before_the_column_existed(self):
+        """tray_now_at_start is nullable for exactly this reason."""
+        assert _print_used_tray_keys([4], None, None) == {(1, 0)}
+
+
+@pytest.mark.asyncio
+class TestSlotsThePrintNeverTouched:
+    async def test_an_untouched_slot_is_not_charged(self, caplog, monkeypatch):
+        """A spool swapped into an idle slot drops that slot's remain%. Reading
+        that as consumption is a phantom write to an uninvolved spool."""
+        monkeypatch.setattr(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+            _fake_resolver(42),
+        )
+
+        client, written, text = await _run(
+            caplog,
+            tray_remain_start=_slot(60),
+            current_lookup=_slot(10),
+            print_used_keys={(1, 3)},  # this print used AMS1-T3, not AMS0-T0
+        )
+
+        assert written == 0
+        assert client.used == []
+        assert "slots not part of this print" in text
+        assert "AMS0-T0" in text
+
+    async def test_the_slot_the_print_used_is_still_charged(self, caplog, monkeypatch):
+        monkeypatch.setattr(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+            _fake_resolver(42),
+        )
+
+        client, written, _ = await _run(
+            caplog,
+            tray_remain_start=_slot(60),
+            current_lookup=_slot(50),
+            print_used_keys={(0, 0)},
+        )
+
+        assert written == 1
+        assert client.used == [(42, 100.0)]
+
+    async def test_without_evidence_every_slot_is_still_considered(self, caplog, monkeypatch):
+        """The reporter's own prints have no mapping and no tray changes. The
+        guard must not turn "we don't know" into "charge nothing"."""
+        monkeypatch.setattr(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+            _fake_resolver(42),
+        )
+
+        client, written, _ = await _run(
+            caplog,
+            tray_remain_start=_slot(60),
+            current_lookup=_slot(50),
+            print_used_keys=set(),
+        )
+
+        assert written == 1
+        assert client.used == [(42, 100.0)]

backend/tests/unit/test_spoolman_no3mf_remain_fallback.py

@@ -126,6 +126,75 @@ class TestStorePrintDataNo3mf:
             "0-1": {"remain": 20, "tray_uuid": "BBBB"},
         }
 
+    @pytest.mark.asyncio
+    async def test_records_the_tray_the_print_started_on(self):
+        """A print with no 3MF also has no ams_mapping, so the tray in use at
+        the start is the only evidence of which slot it drew from — and without
+        that, the completion path would charge every slot whose remain% moved,
+        including one a spool was merely swapped into (#1269's fault, on this
+        path)."""
+        db = AsyncMock()
+        db.execute = AsyncMock(side_effect=[MagicMock()])
+        db.add = MagicMock()
+        db.commit = AsyncMock()
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            tray_now=5,
+            raw_data={"ams": [{"id": 1, "tray": [{"id": 1, "tray_uuid": "AAAA", "remain": 80}]}]},
+        )
+
+        mock_settings = MagicMock()
+        mock_path = MagicMock()
+        mock_path.exists.return_value = False
+        mock_settings.base_dir.__truediv__.return_value = mock_path
+
+        with (
+            patch("backend.app.services.spoolman_tracking.app_settings", mock_settings),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+        ):
+            await store_print_data(
+                printer_id=1,
+                archive_id=42,
+                file_path="",
+                db=db,
+                printer_manager=printer_manager,
+            )
+
+        assert db.add.call_args.args[0].tray_now_at_start == 5
+
+    @pytest.mark.asyncio
+    async def test_a_printer_that_reports_no_tray_records_none(self):
+        """Nullable on purpose: no answer is not the same as slot 0."""
+        db = AsyncMock()
+        db.execute = AsyncMock(side_effect=[MagicMock()])
+        db.add = MagicMock()
+        db.commit = AsyncMock()
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "remain": 80}]}]},
+        )
+
+        mock_settings = MagicMock()
+        mock_path = MagicMock()
+        mock_path.exists.return_value = False
+        mock_settings.base_dir.__truediv__.return_value = mock_path
+
+        with (
+            patch("backend.app.services.spoolman_tracking.app_settings", mock_settings),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+        ):
+            await store_print_data(
+                printer_id=1,
+                archive_id=42,
+                file_path="",
+                db=db,
+                printer_manager=printer_manager,
+            )
+
+        assert db.add.call_args.args[0].tray_now_at_start is None
+
     @pytest.mark.asyncio
     async def test_no_row_when_no_3mf_and_no_remain_data(self):
         """If the AMS has no slot with valid remain either (e.g. printer

backend/tests/unit/test_spoolman_tray_now_migration_1820.py

@@ -0,0 +1,199 @@
+"""Migration for active_print_spoolman.tray_now_at_start (#1820).
+
+The remain%-delta fallback needs to know which slot a print actually drew from,
+or a spool swapped into an idle slot mid-print is charged for consumption it
+never had. For a print with no 3MF -- the case this fallback exists for -- the
+tray in use at the start is often the only evidence, so it is captured at print
+start and has to survive on an upgraded database.
+
+Nullable, not backfilled: a row written before the column existed has no answer
+to give, and inventing 0 would name a real slot.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from backend.app.core.database import run_migrations
+
+LEGACY_TABLE = """
+CREATE TABLE active_print_spoolman (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    printer_id INTEGER NOT NULL,
+    archive_id INTEGER NOT NULL,
+    filament_usage TEXT,
+    ams_trays TEXT NOT NULL,
+    slot_to_tray TEXT,
+    layer_usage TEXT,
+    filament_properties TEXT,
+    tray_remain_start TEXT,
+    UNIQUE(printer_id, archive_id)
+)
+"""
+
+
+@pytest.fixture(autouse=True)
+def force_sqlite_dialect(monkeypatch):
+    """settings.database_url may point at Postgres in dev configs; the test
+    engine is SQLite, so force the dialect where run_migrations reads it."""
+    from backend.app.core import database as database_module, db_dialect
+
+    monkeypatch.setattr(db_dialect, "is_sqlite", lambda: True)
+    monkeypatch.setattr(db_dialect, "is_postgres", lambda: False)
+    monkeypatch.setattr(database_module, "is_sqlite", lambda: True)
+
+
+def _register_every_model() -> None:
+    """Put every table on ``Base.metadata``.
+
+    ``backend.app.models``'s ``__init__`` re-exports only some of them, and
+    ``run_migrations`` walks the whole schema -- a table that was never
+    imported is missing, and its ALTER fails the run before reaching ours.
+    Walking the package keeps this from rotting as models are added.
+    """
+    import importlib
+    import pkgutil
+
+    import backend.app.models as models_pkg
+
+    for module in pkgutil.iter_modules(models_pkg.__path__):
+        importlib.import_module(f"backend.app.models.{module.name}")
+
+
+@pytest.fixture
+async def legacy_engine():
+    """A modern schema whose tracking table predates the column, mid-print."""
+    from backend.app.core.database import Base
+
+    _register_every_model()
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+        await conn.execute(text("DROP TABLE active_print_spoolman"))
+        await conn.execute(text(LEGACY_TABLE))
+        await conn.execute(
+            text(
+                "INSERT INTO active_print_spoolman (id, printer_id, archive_id, ams_trays, tray_remain_start) "
+                'VALUES (1, 1, 42, \'{}\', \'{"0-0": {"remain": 80, "tray_uuid": "AAAA"}}\')'
+            )
+        )
+    yield engine
+    await engine.dispose()
+
+
+async def test_column_missing_before_migration(legacy_engine):
+    """Sanity check, so the assertion below cannot pass by accident."""
+    async with legacy_engine.begin() as conn:
+        columns = {row[1] for row in await conn.execute(text("PRAGMA table_info(active_print_spoolman)"))}
+
+    assert "tray_now_at_start" not in columns
+
+
+async def test_the_column_is_added_and_the_row_survives(legacy_engine):
+    """A print running across the upgrade keeps its remain snapshot; it simply
+    has no tray evidence, which the fallback reads as "consider every slot"."""
+    async with legacy_engine.begin() as conn:
+        await run_migrations(conn)
+
+    async with legacy_engine.begin() as conn:
+        row = (
+            await conn.execute(
+                text("SELECT tray_now_at_start, tray_remain_start FROM active_print_spoolman WHERE id = 1")
+            )
+        ).one()
+
+    assert row[0] is None
+    assert "AAAA" in row[1]
+
+
+async def test_migration_is_idempotent(legacy_engine):
+    """Second boot must not fail on the already-present column."""
+    async with legacy_engine.begin() as conn:
+        await run_migrations(conn)
+    async with legacy_engine.begin() as conn:
+        await run_migrations(conn)
+
+    async with legacy_engine.begin() as conn:
+        columns = {row[1] for row in await conn.execute(text("PRAGMA table_info(active_print_spoolman)"))}
+
+    assert "tray_now_at_start" in columns
+
+
+async def test_a_fresh_database_has_the_column(legacy_engine):
+    """The CREATE TABLE carries it too, so a new install never runs the ALTER."""
+    from backend.app.core.database import Base
+
+    _register_every_model()
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+            columns = {row[1] for row in await conn.execute(text("PRAGMA table_info(active_print_spoolman)"))}
+    finally:
+        await engine.dispose()
+
+    assert "tray_now_at_start" in columns
+
+
+class TestPostgresBranch:
+    """CI runs on SQLite, so the Postgres side of the dialect switch would be
+    dead code without this. Captures the SQL run_migrations would emit,
+    mirroring test_smart_plug_power_flag_migration_2629.
+    """
+
+    @staticmethod
+    async def _capture_sql(is_sqlite_value: bool) -> list[str]:
+        from unittest.mock import AsyncMock, MagicMock, patch
+
+        from backend.app.core import database as db_module
+
+        class _AsyncCtxStub:
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *_exc):
+                return False
+
+        executed_sql: list[str] = []
+
+        async def fake_safe_execute(_conn, sql: str) -> None:
+            executed_sql.append(sql)
+
+        fake_conn = MagicMock()
+        fake_conn.begin_nested = lambda: _AsyncCtxStub()
+        fake_conn.execute = AsyncMock(return_value=MagicMock(fetchone=MagicMock(return_value=None)))
+
+        with (
+            patch("backend.app.core.database.is_sqlite", return_value=is_sqlite_value),
+            patch("backend.app.core.database._safe_execute", side_effect=fake_safe_execute),
+            patch("backend.app.core.database._migrate_update_auto_link_constraint", AsyncMock()),
+            patch("backend.app.core.database._migrate_widen_spoolman_slot_ams_id_range", AsyncMock()),
+        ):
+            await db_module.run_migrations(fake_conn)
+
+        return executed_sql
+
+    @staticmethod
+    def _alter_statements(executed: list[str]) -> list[str]:
+        return [s for s in executed if "tray_now_at_start" in s and "ALTER" in s.upper()]
+
+    @pytest.mark.asyncio
+    async def test_pg_branch_is_idempotent_on_its_own(self):
+        """Postgres has no _safe_execute retry semantics to lean on."""
+        stmts = self._alter_statements(await self._capture_sql(is_sqlite_value=False))
+
+        assert len(stmts) == 1, f"expected exactly one ALTER, got: {stmts!r}"
+        assert "IF NOT EXISTS" in stmts[0]
+        assert "INTEGER" in stmts[0]
+
+    @pytest.mark.asyncio
+    async def test_sqlite_branch_omits_if_not_exists(self):
+        """SQLite's ALTER TABLE has no IF NOT EXISTS; _safe_execute swallows the
+        duplicate-column error instead."""
+        stmts = self._alter_statements(await self._capture_sql(is_sqlite_value=True))
+
+        assert len(stmts) == 1
+        assert "IF NOT EXISTS" not in stmts[0]
+        assert "INTEGER" in stmts[0]