fix(archives): archives made without their 3MF store #RRGGBB colours (existing ones repaired at startup), find AMS-HT filament and are priced from their spools once the 3MF arrives; nozzle temperature of the printed plate

CHANGELOG.md

@@ -164,6 +164,9 @@ 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
+- **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.

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

@@ -1498,7 +1498,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 +1518,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 +3819,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 +3830,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 +3883,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 +3910,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 +3946,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.
@@ -7765,7 +7790,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)

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

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/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_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()